Forking multiple children in Perl and using pipes for bidirectional communication

Question

I am trying to create a small Perl program with several processing capabilities. Since there are slight changes in my requirements, I cannot find any such scripts anywhere.

I need to read a large log file from STDIN and give first the first number N (large number) of lines for the first child process, and then the next N number of lines to the second child process, etc. I also have a specific constant which is the maximum number of child processes allowed to run at the same time. As soon as the maximum number of children reaches, the parent will wait until the child finishes his work and provides him with another N lines.

The parent process also collects multi-line (5-10 lines) output returned by each child process when they end, and stores it in an array. Then the parent continues to process the contents of this array and finally display the results.

Is there a better sample script that I can modify and use, or can someone help me by sharing one here? I prefer to use only pipes for interaction between processes and to simplify things as much as possible.

Edit: Can someone show an example of how this can only be done using pipes from the IO :: Handle module?

Answer 1

Use Forks::Super , which simplifies the throttling of the number of simultaneous processes and the processing of interprocess communication. For instance,

 use Forks::Super MAX_PROC => 10, # allow 10 simultaneous processes ON_BUSY => 'queue'; # don't block when >=10 jobs are active @loglines = <>; # set up all the background jobs while (@loglines > 0) { $pid = fork { args => [ splice @loglines, 0, $N ], # to pass to sub, below child_fh => "out", # make child STDOUT readable by parent sub => sub { my @loglines = @_; my @result = ... do something with loglines ... print @results; # use $pid->read_stdout() to read in child } }; } # get the results while ($pid = waitpid -1, 0) { last if $pid == -1; my @results_from_job = $pid->read_stdout(); push @results, @results_from_job; } 

Answer 2

I found threads much easier for this kind of process. You need threads and threads :: Queue modules. The process is to set up a queue for submitting workflows, and another to return their results. A workflow is just a function for reading a record, processing and sending the result. I just put this code and did not test it, so it may be a mistake, but I think it shows a general idea:

 use threads (); use Thread::Queue; # # # Set limit on number of workers # my $MAX_THREADS = 5; my $EOD = "\n\n"; # # # Need a queue to feed the workers # and one for them to return results # my $Qresult = Thread::Queue->new(); my $rec; my $n; # # # load STDIN into the input queue # my $Qin = Thread::Queue->new(<>); # # # start worker threads # for($n = 0; $n < $MAX_THREADS; ++$n) { async{ProcessRecord($n);}; $Qin->enqueue($EOD); # need terminator for each } # # # # Wait for the results to come in # $n = 0; while($n < $MAX_THREADS) { $rec = $q->dequeue(); if($rec eq $EOD) { ++$n; next; } : : : #-- process result --# : : : threads->yield(); # let other threads get a chance sleep 1; } exit; ###################################### # # # Worker threads draw from the queue # when a "terminator" is read, quit; # sub ProcessRecord { my $rec; my $result; while(1) { $rec = $Qin->dequeue(); last if $rec eq $EOD; : : : #-- process record --# : : : $Qresult->enqueue($result); threads->yield(); # let other threads get a chance } threads->exit(); }