Parallelization of heterogenous (runs itself Fortran executables) code

[Jochen]

Hi @all
I'm trying to parallelize some Perl-code which serves as glue for some numerically intensive Fortran-executables. I've gone through the description of the threads-module about how to start threads and how to wait until they end. But my problem seems to be untouched by that despite it's a problem which (i presume) many others may have experienced. I want to use a boss/worker-type Model in which 4 workers (in which another script, in which the Fortran-executables are called, is called) should be running all the time. My problem is that i don't see how i can check if ANY of the 4 workers has finished. I have tried to set a flag to zero when the thread starts and to 1 when it has accomplished its task. But that seems to me like a rather dirty (and possible thread-unsafe) Method. Than i thought about using something like @array = threads->list and than reading out #$array in intervals of a second, but that somehow did not work. The last Method i tried is in the code below (having a variable count) that did not work too. Is there nothing better out there (possibly "out of the box")?
Thanks

use File::Find;
use File::Copy;
use File::Path;
use threads;
use Thread::Queue;

my $FilenameQueue = Thread::Queue->new;
my $FilenamewithpathQueue = Thread::Queue->new;
my $DirectorynameQueue = Thread::Queue->new;
my $ThreadnumberQueue = Thread::Queue->new;
my $threadnumber = 0;
my $thread;
my $maxnumberofworkers = 3;
if ($#ARGV != 0) {die "Usage: Perl runall.pl [storagedirectoryname]\n"
+;}
my $storagedirectory = $ARGV[0];
my $workbasedirectory = $storagebasedirectory . "/" . $storagedirector
+y;
mkdir($workbasedirectory);
print "Starting recursive directory search!\n";
find(\&docalc, $basedirectory);

sub docalc {
  /\.txt$/ or return;
  while ($threadnumber >= $maxnumberofworkers){
    sleep(1);}
  $FilenameQueue->enqueue($_);
  $FilenamewithpathQueue->enqueue($File::Find::name);
  $DirectorynameQueue->enqueue($File::Find::dir);
  $thread = threads->new(\&worker);
  $threadnumber++;
  $thread->detach;
  print "New thread created!\n";}

sub worker {
  my @filename = split(/\./,$FilenameQueue->dequeue);
  print "Processing $filename[0]!\n";
  my $workdirectory = $workbasedirectory . substr($DirectorynameQueue-
+>dequeue,length($basedirectory)) . "/" . $filename[0];
  $workdirectory=~ s/\n//gi;
  print "Creating directory" . $workdirectory . "\n";
  mkpath($workdirectory);
  copyfiles($workdirectory, $FilenamewithpathQueue->dequeue, $filename
+[0]);
  print "Processing Calculations...\n";
  open(RUNSCRIPT,">runscript.bash");
  print RUNSCRIPT "cd $workdirectory\nperl scriptthatincludesfortran.p
+l >& $workbasedirectory/$filename[0].log";
  close(RUNSCRIPT);
   if (system("bash runscript.bash") != 0) {
     die "failed!\n";}
   else {
     print "done!\n";}
   print "bash $workdirectory/runscript.bash\n";
  print "\n";
  $threadnumber--;}

[salva]

You can use Parallel::ForkManager or Proc::Queue for that:

# untested
use Proc::Queue size => 4, ignore_children => 1, qw(run_back);

sub docalc {
    run_back {
        my $filename = $_;
        my $filenamewithpath = $File::Find::name;
        my $directoryname = $File::Find::dir;
        my $workdirectory = ...;
        mkpath($workdirectory);
        copyfiles(...);
        chdir $workingdirectory;
        open STDOUT, '>', "$workbasedirectory/$filename.log";
        do "/.../scriptthatincludesfortran.pl"
    } or warn "unable to fork new process to handle '$_': $!"
}

[Jochen]

Thank you very much! This approach seems to suit my problem best! I'm still having some issues but i think i can solve them.

[Corion]

As I had just a similar problem to solve, here is the solution I use, stolen from Dominus - runN, a small, self-contained Perl program that runs up to N instances of a program in parallel. If your command line requirements go beyond what runN provides, it's not too hard to change it, but then I'd likely fall back to using Parallel::ForkManager, which is not much more code either but well-used and more documented.

[Dominus]

You didn't steal it; it was a gift.

It has evolved a little since I posted it. Here is the current version:

#!/usr/bin/perl

use Getopt::Std;
my %opt = (n => 1);
getopts('r:n:v', \%opt) or usage();

my $cmd = shift;
@ARGV = shuffle(@ARGV) if $opt{r};

my %pid;
while (@ARGV) {
  if (keys(%pid) < $opt{n}) {
    $pid{spawn($cmd, split /\s+/, shift @ARGV)} = 1;
  } else {
    delete $pid{wait()};
  }
}

1 while wait() >= 0;

sub spawn {
  my $pid = fork;
  die "fork: $!" unless defined $pid;
  return $pid if $pid;
  warn "@_\n" if $opt{v};
  exec @_;
  die "exec: $!";
}

sub usage {
  print STDERR "Usage: $0 [-n N] [-r] [-v] command arg1 arg2...
Run command arg1, command arg2, etc., concurrently.
Run no more than N processes simultaneously (default 1)
    -r: run commands in random order instead of specified order (unimp
+l.)
    -v: verbose mode
";
  exit 1;
}

The major missing feature is that at present there's no way to get it to run cmd -x arg1, cmd -x arg2...; there's no way to get the constant -x in there.

I hereby put this program in the public domain.

Share and enjoy!

[codeacrobat]

Here is a version using Text::ParseWords and a modified spawn subroutine, which uses exec in scalar context. This allows parameterized command execution.

#!/usr/bin/perl

use Getopt::Std;
use Text::ParseWords;
my %opt = (n => 1);
getopts('r:n:v', \%opt) or usage();

my $cmd = shift;
@ARGV = shuffle(@ARGV) if $opt{r};

my %pid;
while (@ARGV) {
  if (keys(%pid) < $opt{n}) {
    $pid{spawn($cmd, map { shellwords($_) } shift @ARGV)} = 1;
  } else {
    delete $pid{wait()};
  }
}

1 while wait() >= 0;

sub spawn {
  my $pid = fork;
  die "fork: $!" unless defined $pid;
  return $pid if $pid;
  warn "@_\n" if $opt{v};
  exec qq(@_);
  die "exec: $!";
}

sub usage {
  print STDERR "Usage: $0 [-n N] [-r] [-v] command arg1 arg2...
Run command arg1, command arg2, etc., concurrently.
Run no more than N processes simultaneously (default 1)
    -r: run commands in random order instead of specified order (unimp
+l.)
    -v: verbose mode
";
  exit 1;
}

The following snippet will ping google.com and amazon.com once in the first chunk and yahoo.com in the second.

runN -v -n 2 'ping -n 1' google.com amazon.com yahoo.com
ping -n 1 google.com
ping -n 1 amazon.com

Ping google.com [72.14.207.99] mit 32 Bytes Daten:

Antwort von 72.14.207.99: Bytes=32 Zeit=116ms TTL=242

Ping-Statistik für 72.14.207.99:
    Pakete: Gesendet = 1, Empfangen = 1, Verloren = 0 (0% Verlust),
Ca. Zeitangaben in Millisek.:
    Minimum = 116ms, Maximum = 116ms, Mittelwert = 116ms

Ping amazon.com [72.21.210.11] mit 32 Bytes Daten:

ping -n 1 yahoo.com

Ping yahoo.com [66.94.234.13] mit 32 Bytes Daten:

Antwort von 66.94.234.13: Bytes=32 Zeit=185ms TTL=54

Ping-Statistik für 66.94.234.13:
    Pakete: Gesendet = 1, Empfangen = 1, Verloren = 0 (0% Verlust),
Ca. Zeitangaben in Millisek.:
    Minimum = 185ms, Maximum = 185ms, Mittelwert = 185ms
Zeitüberschreitung der Anforderung.

Ping-Statistik für 72.21.210.11:
    Pakete: Gesendet = 1, Empfangen = 0, Verloren = 1 (100% Verlust)

[blazar]

Ok, you are mjd so I shouldn't probably "dare" to comment, but...

use Getopt::Std;

I personally believe that since we recommend newbies and more expert programmers altogether to always use strict and warnings, and this is a nice little utility likely to be picked up as an example, it would be a good thing if it had

use strict;
use warnings;

at the top. So to build a better future for our children...

Also,

use List::Util 'shuffle';

would implement the -r switch straight ahead.

getopts('r:n:v', \%opt) or usage();

The docs do not say anything about getopts() return value, and indeed experimental evidence is that it can't be relied upon for failure checking. Suitable hooks are provided instead, although admittedly I don't like the interface. (Suitably named subs in main::)

sub usage {
  print STDERR "Usage: $0 [-n N] [-r] [-v] command arg1 arg2...
Run command arg1, command arg2, etc., concurrently.
Run no more than N processes simultaneously (default 1)
    -r: run commands in random order instead of specified order (unimp
+l.)
    -v: verbose mode
";
  exit 1;

Any good reason for basically reimplementing die? Incidentally, I would have used a here-doc instead. Personally, I like to implement a USAGE sub like thus:

sub USAGE () {
    my $name=basename $0;  # File::Basename's
    <<".EOT";
Usage: $name [args]
[actual usage here]
.EOT
}

So if the user explicitly asks for help, then I print to STDOUT and exit regularly, for in that case I wouldn't consider the program termination to be "abnormal". Else, I regularly die USAGE.

[Dominus]

The major missing feature is that at present there's no way to get it to run cmd -x arg1, cmd -x arg2...; there's no way to get the constant -x in there.

I tried fixing it by changing:

my $cmd = shift;

to:

my @cmd = split /\s+/, shift;

and:

$pid{spawn($cmd, split /\s+/, shift @ARGV)} = 1;

to:

$pid{spawn(@cmd, shift @ARGV)} = 1;

(I also decided that splitting the remaining arguments was a fairly dumb mistake, so got rid of that.)

Now, to get it to run cmd -x arg1, cmd -x arg2...; you just runN "cmd -x" arg1 arg2... .

I think this has worked out pretty well in practice so far, but only time will tell.

Complete code is now:

#!/usr/bin/perl

use Getopt::Std;
my %opt = (n => 1);
getopts('r:n:v', \%opt) or usage();

my @cmd = split /\s+/, shift;
@ARGV = shuffle(@ARGV) if $opt{r};

my %pid;
while (@ARGV) {
  if (keys(%pid) < $opt{n}) {
    $pid{spawn(@cmd, shift @ARGV)} = 1;
  } else {
    delete $pid{wait()};
  }
}

1 while wait() >= 0;

sub spawn {
  my $pid = fork;
  die "fork: $!" unless defined $pid;
  return $pid if $pid;
  warn "@_\n" if $opt{v};
  exec @_;
  die "exec: $!";
}

sub usage {
  print STDERR "Usage: $0 [-n N] [-r] [-v] command arg1 arg2...
Run command arg1, command arg2, etc., concurrently.
Run no more than N processes simultaneously (default 1)
    -r: run commands in random order instead of specified order (unimp
+l.)
    -v: verbose mode
";
  exit 1;
}

[salva]

GNU xargs has also an option (-P) to run several processes in parallel. Something like...

find /starting/dir -type f -print0 | xargs -P 4 -0 /.../perlscriptcall
+ingfortran.pl

... should work for the OP.

[erroneousBollock]

Update: I, like the following commenters, don't think that threads are appropriate in this case, but I'll answer the question in case someone comes looking for an answer where threads would be appropriate.

My problem is that i don't see how i can check if ANY of the 4 workers has finished.

Have a single control queue shared between the worker threads and the main thread.

Create a Thread::Queue object, which we'll call the "control queue". When you create the workers, pass that queue to each of the workers you create. When a worker is done it should enqueue a "worker N done" message into the "control queue".

In your main thread, once all the worker threads are started, do a blocking dequeue from the "control queue" in while a loop, where the loops ends when you've received the appropriate number of "worker N done" messages. (ie: You could receive other useful messages on the same queue).

In the body of the while loop, you can deal with what should happen when ANY of the workers finish (use a condition variable if you're only interested in doing so for the first).

After the loop, you can deal with what should happen when ALL of the workers have finished.

re: threads and system():

Also, you seem to be doing system() (which is fork() and exec()) inside the worker thread. Are you sure you understand the implications of mixing fork() with Perl threads?

-David

[Cefu]

David,

Could you please elaborate on the implications of mixing fork() with Perl threads?

I'd like to know what the pitfalls are. While it's clear that threads are redundant (system()/fork() already allows the processes to run in parallel) Jochen's approach still seems natural to me. Starting four threads from the main proccess allows you to limit how many processes you are running at once and to use four identical workers to handle the input/output to each fortran executable.

If it's best to avoid mixing fork() (from the system() calls) and Perl threads, what would you suggest for Jochen's situation? Using only system() calls to fork from the main process while the main loop maintains a count to be sure no more than four run at once?

Cefu

[erroneousBollock]

Could you please elaborate on the implications of mixing fork() with Perl threads? ... I'd like to know what the pitfalls are.

The implications stem from how Perl threads are implemented.

Threads.pm clones all non-shared data in the 'parent' thread to the 'child' thread. If you have a lot of data (or code) loaded in the spawning thread, that data is cloned (copied) into the spawned thread.

Users of Unix fork() have come to expect Copy-on-Write semantics; only copying data when it is written to. While that expectation is not necessarily warranted for perl's fork() (perl modifies bit flags on its variables routinely), it certainly doesn't work that way for threads... if you don't understand the behaviour, you'll end up using lots more memory than you intended.

What I don't know (I'm sure BrowserUK does) is whether all the running threads of the (forking) process are then present in the newly spawned (child) process; if they are, it could get messy ;) Tested: only active thread is forked().

If it's best to avoid mixing fork() (from the system() calls) and Perl threads, what would you suggest for Jochen's situation?

I'd recommend he use Parallel::ForkManager to manage worker processes... that's exactly what it was written for.

Also, remember that fork() is emulated using Threads on Win32, so that wouldn't necessarily be the best way to go on that platform.

-David

[jbert]

Given your CPU-intensive processes are already outside your perl process, you don't need threads to make this work.

Just start the fortran child processes all at once and wait for the child processes to exit. You can do this yourself, but a module like Parallel::ForkManager or IPC::Run is probably going to be of some help here.

[moritz]

If you could write your control program in terms of a Makefile, you could use GNU make's -j $count option to handle parallelization.

[perlfan]

It sounds like you should be using MPI, that way you could do everything /in/ Fortran. Additionally, you'll be able to manage /many/ processes across many hosts (using a machinefile definition). MPI allows for heterogenous environments to be utilized as well.

While I agree that it is neat to use Perl to manage this, you are artificially limiting yourself, especially if one day you wish to run this on a set of distinct host machines.

[Anonymous Monk]

The Fortran executables I use are from a colleague and commercial, and i have to use them as they are provided. But thanks for the reply anyway.

[perlfan]

Does this mean no source? If so, then I guess there's not much you can do.

[BrowserUk]

If your target OS is one with a native fork then that or one of the CPAN wrappers is possibly the way to go, but if you wanted a threads solution, your attempt was making a mountain out of a molehill.

There is certainly no logic to using 3 queues to pass the path, the filename, and the path+filename. Pass it once and split it up there if need be.

And there is no need to count the threads or detect when one has finished. Just start the number your want to use (say 4), and have them loop over the queue. As soon as they finish one piece of work they'll pick up the next automatically. It is also far more efficient than starting a new thread for each file.

When you've finished pushing all the filenames, push 4 undefs and the threads will self terminate when there is nothing left to do. Simplicity itself.

#! perl -slw
use strict;
use threads;
use Thread::Queue;
use File::Find;
use File::Copy;
use File::Path;

my $THREADS = 4;

my $Q = new Thread::Queue;

sub worker {
    while( my $path = $Q->dequeue ) {
        print "Creating directory" . $workdirectory . "\n";
        mkpath($workdirectory);
        copyfiles(
            $workdirectory, 
            $FilenamewithpathQueue->dequeue, $filename[0]
        );

        print "Processing Calculations...\n";
        open(RUNSCRIPT,">runscript.bash");

        print RUNSCRIPT <<EOD
        cd $workdirectory
        perl scriptthatincludesfortran.pl >& $workbasedirectory/$filen
+ame[0].log
EOD

        close(RUNSCRIPT);
        if (system("bash runscript.bash") != 0) {
            die "failed!\n";}
        else {
            print "done!\n";
        }

        print "bash $workdirectory/runscript.bash\n";
        print "\n";

    }
}

my @threads = map{ threads->create( \&worker ) } 1 .. $THREADS;

find( sub{ $Q->enqueue( $File::Find::name ) }, $basedirectory );

$Q->enqueue( (undef) x $THREADS );
$_->join for @threads;

The code is untested because I couldn't make much sense of your OP code, as half the variables in it are never declared and never initialised.

[Jochen]

Yes, i know. This is because ist is just the part of my script that I thought would be relevant. Splitting up afterwards would also be my solution. Up to now I was just too lazy to do it. Well, I guess polishing kann be done after it works...
Thanks anyway. Jochen

[sundialsvc4]

Dictum Ne Agas: “ Do not do a thing already done.

If you're doing a CPU-intensive task, then “threads” won't help you: threads divide the computer's time; they do not multiply it. Only a cluster of actual discrete CPUs will do the trick.

And since “that's tricky,” ... “It's a bird! No, it's a plane! No, it's CPAN To The Rescue!” There are numerous tested-and-debugged modules available to do what you have in mind. (There are also full-fledged batch scheduling systems.) So... enter the mindset that you are not setting out to “build” your solution, but merely to “find” it. Rocket-scientists all over the globe have addressed exactly the same problem, solved it thoroughly, and given away their solution.

[Jochen]

"Threads divide the computer's time; they do not multiply it" - AFAIK not if its an smp-machine that would otherwise not take advantage of the extra cores.

[perlfan]

It is all moot if one (or more) of the threads is hogging I/O.

Update 1: while I make a point not to comment on downmodded posts of mine, I have to say that i/o contention among threads is a SERIOUS issue - even on an SMP machine that has multiple cores/cpus. So put that in your pipe and smoke it.

Update 2: I've been asked by BrowserUk to clarify, so here we go. As it was pointed out, threading simply divides time among related processes.

On a single CPU system, the issue of a thread hogging i/o is indistinguishable from a simple blocking thread - sibling threads simply won't get a chance to execute until the thread that is executing finishes with its reads/writes.

On shared memory (multi-core, SMP), the situation changes. Multiple threads may be executing, but if some small subset of the threads are engaging in heavy i/o, things will come screeching to a crawl. My point is that there is an additional contention for i/o bandwidth, and unless there is a high performance disk subsystem available to your threads, simply thowing more CPUs at it won't make a lick of difference - and could in fact make your performance a lot worse than just running the code serially.

We don't know what the OP's code is doing. If it is reading and writing a lot of data, this could have a significant impact on the performance of the threads.

[absolut.todd]

Have you had a read of Randal Schwartz column on his forking parallel link checker?

A forking parallel link checker

I have found it as a good solution for the problem that im trying to achieve running several processes in parallel and managing them.

[TGI]

This sounds like a job for POE and POE::Wheel::Run.

Take a look at this example in the POE Cookbook.

[Jochen]

Thank you for all the suggestions!!!
I ended up using Parallel::Forkmanager which now does exactly what i wanted.