1MCE::Child(3) User Contributed Perl Documentation MCE::Child(3)
2
6 MCE::Child - A threads-like parallelization module compatible with Perl
7 5.8
8
10 This document describes MCE::Child version 1.866
11
13 use MCE::Child;
14 MCE::Child->init(
16 max_workers => 'auto', # default undef, unlimited
17 child_timeout => 20, # default undef, no timeout
18 posix_exit => 1, # default undef, CORE::exit
19 void_context => 1, # default undef
20 on_start => sub {
21 my ( $pid, $ident ) = @_;
22 ...
23 },
24 on_finish => sub {
25 my ( $pid, $exit, $ident, $signal, $error, @ret ) = @_;
26 ...
27 }
28 );
29 MCE::Child->create( sub { print "Hello from child\n" } )->join();
31 sub parallel {
33 my ($arg1) = @_;
34 print "Hello again, $arg1\n" if defined($arg1);
35 print "Hello again, $_\n"; # same thing
36 }
37 MCE::Child->create( \¶llel, $_ ) for 1 .. 3;
39 my @procs = MCE::Child->list();
41 my @pids = MCE::Child->list_pids();
42 my @running = MCE::Child->list_running();
43 my @joinable = MCE::Child->list_joinable();
44 my @count = MCE::Child->pending();
45 # Joining is orderly, e.g. child1 is joined first, child2, child3.
47 $_->join() for @procs; # (or)
48 $_->join() for @joinable;
49 # Joining occurs immediately as child processes complete execution.
51 1 while MCE::Child->wait_one();
52 my $child = mce_child { foreach (@files) { ... } };
54 $child->join();
56 if ( my $err = $child->error() ) {
58 warn "Child error: $err\n";
59 }
60 # Get a child's object
62 $child = MCE::Child->self();
63 # Get a child's ID
65 $pid = MCE::Child->pid(); # $$
66 $pid = $child->pid();
67 $pid = MCE::Child->tid(); # tid is an alias for pid
68 $pid = $child->tid();
69 # Test child objects
71 if ( $child1 == $child2 ) {
72 ...
73 }
74 # Give other workers a chance to run
76 MCE::Child->yield();
77 MCE::Child->yield(0.05);
78 # Return context, wantarray aware
80 my ($value1, $value2) = $child->join();
81 my $value = $child->join();
82 # Check child's state
84 if ( $child->is_running() ) {
85 sleep 1;
86 }
87 if ( $child->is_joinable() ) {
88 $child->join();
89 }
90 # Send a signal to a child
92 $child->kill('SIGUSR1');
93 # Exit a child
95 MCE::Child->exit(0);
96 MCE::Child->exit(0, @ret);
97
99 MCE::Child is a fork of MCE::Hobo for compatibility with Perl 5.8.
100 A child is a migratory worker inside the machine that carries the
102 asynchronous gene. Child processes are equipped with "threads"-like
103 capability for running code asynchronously. Unlike threads, each child
104 is a unique process to the underlying OS. The IPC is handled via
105 "MCE::Channel", which runs on all the major platforms including Cygwin
106 and Strawberry Perl.
107 "MCE::Child" may be used as a standalone or together with "MCE"
109 including running alongside "threads".
110 use MCE::Child;
112 use MCE::Shared;
113 # synopsis: head -20 file.txt | perl script.pl
115 my $ifh = MCE::Shared->handle( "<", \*STDIN ); # shared
117 my $ofh = MCE::Shared->handle( ">", \*STDOUT );
118 my $ary = MCE::Shared->array();
119 sub parallel_task {
121 my ( $id ) = @_;
122 while ( <$ifh> ) {
123 printf {$ofh} "[ %4d ] %s", $., $_;
124 # $ary->[ $. - 1 ] = "[ ID $id ] read line $.\n" ); # dereferencing
125 $ary->set( $. - 1, "[ ID $id ] read line $.\n" ); # faster via OO
126 }
127 }
128 my $child1 = MCE::Child->new( "parallel_task", 1 );
130 my $child2 = MCE::Child->new( \¶llel_task, 2 );
131 my $child3 = MCE::Child->new( sub { parallel_task(3) } );
132 $_->join for MCE::Child->list(); # ditto: MCE::Child->wait_all();
134 # search array (total one round-trip via IPC)
136 my @vals = $ary->vals( "val =~ / ID 2 /" );
137 print {*STDERR} join("", @vals);
139
141 $child = MCE::Child->create( FUNCTION, ARGS )
142 $child = MCE::Child->new( FUNCTION, ARGS )
143 This will create a new child process that will begin execution with
144 function as the entry point, and optionally ARGS for list of
145 parameters. It will return the corresponding MCE::Child object, or
146 undef if child creation failed.
147 FUNCTION may either be the name of a function, an anonymous
149 subroutine, or a code ref.
150 my $child = MCE::Child->create( "func_name", ... );
152 # or
153 my $child = MCE::Child->create( sub { ... }, ... );
154 # or
155 my $child = MCE::Child->create( \&func, ... );
156 $child = MCE::Child->create( { options }, FUNCTION, ARGS )
158 $child = MCE::Child->create( IDENT, FUNCTION, ARGS )
159 Options, excluding "ident", may be specified globally via the "init"
160 function. Otherwise, "ident", "child_timeout", "posix_exit", and
161 "void_context" may be set uniquely.
162 The "ident" option is used by callback functions "on_start" and
164 "on_finish" for identifying the started and finished child process
165 respectively.
166 my $child1 = MCE::Child->create( { posix_exit => 1 }, sub {
168 ...
169 } );
170 $child1->join;
172 my $child2 = MCE::Child->create( { child_timeout => 3 }, sub {
174 sleep 1 for ( 1 .. 9 );
175 } );
176 $child2->join;
178 if ( $child2->error() eq "Child timed out\n" ) {
180 ...
181 }
182 The "new()" method is an alias for "create()".
184 mce_child { BLOCK } ARGS;
186 mce_child { BLOCK };
187 "mce_child" runs the block asynchronously similarly to
188 "MCE::Child->create()". It returns the child object, or undef if
189 child creation failed.
190 my $child = mce_child { foreach (@files) { ... } };
192 $child->join();
194 if ( my $err = $child->error() ) {
196 warn("Child error: $err\n");
197 }
198 $child->join()
200 This will wait for the corresponding child process to complete its
201 execution. In non-voided context, "join()" will return the value(s)
202 of the entry point function.
203 The context (void, scalar or list) for the return value(s) for
205 "join" is determined at the time of joining and mostly "wantarray"
206 aware.
207 my $child1 = MCE::Child->create( sub {
209 my @res = qw(foo bar baz);
210 return (@res);
211 });
212 my @res1 = $child1->join(); # ( foo, bar, baz )
214 my $res1 = $child1->join(); # baz
215 my $child2 = MCE::Child->create( sub {
217 return 'foo';
218 });
219 my @res2 = $child2->join(); # ( foo )
221 my $res2 = $child2->join(); # foo
222 $child1->equal( $child2 )
224 Tests if two child objects are the same child or not. Child
225 comparison is based on process IDs. This is overloaded to the more
226 natural forms.
227 if ( $child1 == $child2 ) {
229 print("Child objects are the same\n");
230 }
231 # or
232 if ( $child1 != $child2 ) {
233 print("Child objects differ\n");
234 }
235 $child->error()
237 Child processes are executed in an "eval" context. This method will
238 return "undef" if the child terminates normally. Otherwise, it
239 returns the value of $@ associated with the child's execution status
240 in its "eval" context.
241 $child->exit()
243 This sends 'SIGQUIT' to the child process, notifying the child to
244 exit. It returns the child object to allow for method chaining. It
245 is important to join later if not immediately to not leave a zombie
246 or defunct process.
247 $child->exit()->join();
249 ...
250 $child->join(); # later
252 MCE::Child->exit( 0 )
254 MCE::Child->exit( 0, @ret )
255 A child can exit at any time by calling "MCE::Child->exit()".
256 Otherwise, the behavior is the same as "exit(status)" when called
257 from the main process. The child process may optionally return data,
258 to be sent via IPC.
259 MCE::Child->finish()
261 This class method is called automatically by "END", but may be
262 called explicitly. An error is emitted via croak if there are active
263 child processes not yet joined.
264 MCE::Child->create( 'task1', $_ ) for 1 .. 4;
266 $_->join for MCE::Child->list();
267 MCE::Child->create( 'task2', $_ ) for 1 .. 4;
269 $_->join for MCE::Child->list();
270 MCE::Child->create( 'task3', $_ ) for 1 .. 4;
272 $_->join for MCE::Child->list();
273 MCE::Child->finish();
275 MCE::Child->init( options )
277 The init function accepts a list of MCE::Child options.
278 MCE::Child->init(
280 max_workers => 'auto', # default undef, unlimited
281 child_timeout => 20, # default undef, no timeout
282 posix_exit => 1, # default undef, CORE::exit
283 void_context => 1, # default undef
284 on_start => sub {
285 my ( $pid, $ident ) = @_;
286 ...
287 },
288 on_finish => sub {
289 my ( $pid, $exit, $ident, $signal, $error, @ret ) = @_;
290 ...
291 }
292 );
293 # Identification given as an option or the 1st argument.
295 for my $key ( 'aa' .. 'zz' ) {
297 MCE::Child->create( { ident => $key }, sub { ... } );
298 MCE::Child->create( $key, sub { ... } );
299 }
300 MCE::Child->wait_all;
302 Set "max_workers" if you want to limit the number of workers by
304 waiting automatically for an available slot. Specify "auto" to
305 obtain the number of logical cores via "MCE::Util::get_ncpu()".
306 Set "child_timeout", in number of seconds, if you want the child
308 process to terminate after some time. The default is 0 for no
309 timeout.
310 Set "posix_exit" to avoid all END and destructor processing.
312 Constructing MCE::Child inside a thread implies 1 or if present CGI,
313 FCGI, Coro, Curses, Gearman::Util, Gearman::XS, LWP::UserAgent,
314 Mojo::IOLoop, STFL, Tk, Wx, or Win32::GUI.
315 Set "void_context" to create the child process in void context for
317 the return value. Otherwise, the return context is wantarray-aware
318 for "join()" and "result()" and determined when retrieving the data.
319 The callback options "on_start" and "on_finish" are called in the
321 parent process after starting the worker and later when terminated.
322 The arguments for the subroutines were inspired by
323 Parallel::ForkManager.
324 The parameters for "on_start" are the following:
326 - pid of the child process
328 - identification (ident option or 1st arg to create)
329 The parameters for "on_finish" are the following:
331 - pid of the child process
333 - program exit code
334 - identification (ident option or 1st arg to create)
335 - exit signal id
336 - error message from eval inside MCE::Child
337 - returned data
338 $child->is_running()
340 Returns true if a child is still running.
341 $child->is_joinable()
343 Returns true if the child has finished running and not yet joined.
344 $child->kill( 'SIG...' )
346 Sends the specified signal to the child. Returns the child object to
347 allow for method chaining. As with "exit", it is important to join
348 eventually if not immediately to not leave a zombie or defunct
349 process.
350 $child->kill('SIG...')->join();
352 The following is a parallel demonstration comparing "MCE::Shared"
354 against "Redis" and "Redis::Fast" on a Fedora 23 VM. Joining begins
355 after all workers have been notified to quit.
356 use Time::HiRes qw(time);
358 use Redis;
360 use Redis::Fast;
361 use MCE::Child;
363 use MCE::Shared;
364 my $redis = Redis->new();
366 my $rfast = Redis::Fast->new();
367 my $array = MCE::Shared->array();
368 sub parallel_redis {
370 my ($_redis) = @_;
371 my ($count, $quit, $len) = (0, 0);
372 # instead, use a flag to exit loop
374 $SIG{'QUIT'} = sub { $quit = 1 };
375 while () {
377 $len = $_redis->rpush('list', $count++);
378 last if $quit;
379 }
380 $count;
382 }
383 sub parallel_array {
385 my ($count, $quit, $len) = (0, 0);
386 # do not exit from inside handler
388 $SIG{'QUIT'} = sub { $quit = 1 };
389 while () {
391 $len = $array->push($count++);
392 last if $quit;
393 }
394 $count;
396 }
397 sub benchmark_this {
399 my ($desc, $num_procs, $timeout, $code, @args) = @_;
400 my ($start, $total) = (time(), 0);
401 MCE::Child->new($code, @args) for 1..$num_procs;
403 sleep $timeout;
404 # joining is not immediate; ok
406 $_->kill('QUIT') for MCE::Child->list();
407 # joining later; ok
409 $total += $_->join() for MCE::Child->list();
410 printf "$desc <> duration: %0.03f secs, count: $total\n",
412 time() - $start;
413 sleep 0.2;
415 }
416 benchmark_this('Redis ', 8, 5.0, \¶llel_redis, $redis);
418 benchmark_this('Redis::Fast', 8, 5.0, \¶llel_redis, $rfast);
419 benchmark_this('MCE::Shared', 8, 5.0, \¶llel_array);
420 MCE::Child->list()
422 Returns a list of all child objects not yet joined.
423 @procs = MCE::Child->list();
425 MCE::Child->list_pids()
427 Returns a list of all child pids not yet joined (available since
428 1.849).
429 @pids = MCE::Child->list_pids();
431 $SIG{INT} = $SIG{HUP} = $SIG{TERM} = sub {
433 # Signal workers all at once
434 CORE::kill('KILL', MCE::Child->list_pids());
435 exec('reset');
436 };
437 MCE::Child->list_running()
439 Returns a list of all child objects that are still running.
440 @procs = MCE::Child->list_running();
442 MCE::Child->list_joinable()
444 Returns a list of all child objects that have completed running.
445 Thus, ready to be joined without blocking.
446 @procs = MCE::Child->list_joinable();
448 MCE::Child->max_workers([ N ])
450 Getter and setter for max_workers. Specify a number or 'auto' to
451 acquire the total number of cores via MCE::Util::get_ncpu. Specify a
452 false value to set back to no limit.
453 MCE::Child->pending()
455 Returns a count of all child objects not yet joined.
456 $count = MCE::Child->pending();
458 $child->result()
460 Returns the result obtained by "join", "wait_one", or "wait_all". If
461 the process has not yet exited, waits for the corresponding child to
462 complete its execution.
463 use MCE::Child;
465 use Time::HiRes qw(sleep);
466 sub task {
468 my ($id) = @_;
469 sleep $id * 0.333;
470 return $id;
471 }
472 MCE::Child->create('task', $_) for ( reverse 1 .. 3 );
474 # 1 while MCE::Child->wait_one();
476 while ( my $child = MCE::Child->wait_one() ) {
478 my $err = $child->error() || 'no error';
479 my $res = $child->result();
480 my $pid = $child->pid();
481 print "[$pid] $err : $res\n";
483 }
484 Like "join" described above, the context (void, scalar or list) for
486 the return value(s) is determined at the time "result" is called and
487 mostly "wantarray" aware.
488 my $child1 = MCE::Child->create( sub {
490 my @res = qw(foo bar baz);
491 return (@res);
492 });
493 my @res1 = $child1->result(); # ( foo, bar, baz )
495 my $res1 = $child1->result(); # baz
496 my $child2 = MCE::Child->create( sub {
498 return 'foo';
499 });
500 my @res2 = $child2->result(); # ( foo )
502 my $res2 = $child2->result(); # foo
503 MCE::Child->self()
505 Class method that allows a child to obtain it's own MCE::Child
506 object.
507 $child->pid()
509 $child->tid()
510 Returns the ID of the child.
511 pid: $$ process id
513 tid: $$ alias for pid
514 MCE::Child->pid()
516 MCE::Child->tid()
517 Class methods that allows a child to obtain its own ID.
518 pid: $$ process id
520 tid: $$ alias for pid
521 MCE::Child->wait_one()
523 MCE::Child->waitone()
524 MCE::Child->wait_all()
525 MCE::Child->waitall()
526 Meaningful for the manager process only, waits for one or all child
527 processes to complete execution. Afterwards, returns the
528 corresponding child objects. If a child doesn't exist, returns the
529 "undef" value or an empty list for "wait_one" and "wait_all"
530 respectively.
531 The "waitone" and "waitall" methods are aliases for compatibility
533 with "MCE::Hobo".
534 use MCE::Child;
536 use Time::HiRes qw(sleep);
537 sub task {
539 my $id = shift;
540 sleep $id * 0.333;
541 return $id;
542 }
543 MCE::Child->create('task', $_) for ( reverse 1 .. 3 );
545 # join, traditional use case
547 $_->join() for MCE::Child->list();
548 # wait_one, simplistic use case
550 1 while MCE::Child->wait_one();
551 # wait_one
553 while ( my $child = MCE::Child->wait_one() ) {
554 my $err = $child->error() || 'no error';
555 my $res = $child->result();
556 my $pid = $child->pid();
557 print "[$pid] $err : $res\n";
559 }
560 # wait_all
562 my @procs = MCE::Child->wait_all();
563 for ( @procs ) {
565 my $err = $_->error() || 'no error';
566 my $res = $_->result();
567 my $pid = $_->pid();
568 print "[$pid] $err : $res\n";
570 }
571 MCE::Child->yield( [ floating_seconds ] )
573 Give other workers a chance to run, optionally for given time. Yield
574 behaves similarly to MCE's interval option. It throttles workers
575 from running too fast. A demonstration is provided in the next
576 section for fetching URLs in parallel.
577 The default "floating_seconds" is 0.008 and 0.015 on UNIX and
579 Windows, respectively. Pass 0 if simply wanting to give other
580 workers a chance to run.
581 # total run time: 1.00 second
583 MCE::Child->create( sub { MCE::Child->yield(0.25) } ) for 1 .. 4;
585 MCE::Child->wait_all();
586
588 MCE::Child behaves similarly to threads for the most part. It also
589 provides Parallel::ForkManager-like capabilities. The
590 "Parallel::ForkManager" example is shown first followed by a version
591 using "MCE::Child".
592 Parallel::ForkManager
594 use strict;
595 use warnings;
596 use Parallel::ForkManager;
598 use Time::HiRes 'time';
599 my $start = time;
601 my $pm = Parallel::ForkManager->new(10);
603 $pm->set_waitpid_blocking_sleep(0);
604 $pm->run_on_finish( sub {
606 my ($pid, $exit_code, $ident, $exit_signal, $core_dumped, $resp) = @_;
607 print "child $pid completed: $ident => ", $resp->[0], "\n";
608 });
609 DATA_LOOP:
611 foreach my $data ( 1..2000 ) {
612 # forks and returns the pid for the child
613 my $pid = $pm->start($data) and next DATA_LOOP;
614 my $ret = [ $data * 2 ];
615 $pm->finish(0, $ret);
617 }
618 $pm->wait_all_children;
620 printf STDERR "duration: %0.03f seconds\n", time - $start;
622 MCE::Child
624 use strict;
625 use warnings;
626 use MCE::Child 1.843;
628 use Time::HiRes 'time';
629 my $start = time;
631 MCE::Child->init(
633 max_workers => 10,
634 on_finish => sub {
635 my ($pid, $exit_code, $ident, $exit_signal, $error, $resp) = @_;
636 print "child $pid completed: $ident => ", $resp->[0], "\n";
637 }
638 );
639 foreach my $data ( 1..2000 ) {
641 MCE::Child->create( $data, sub {
642 [ $data * 2 ];
643 });
644 }
645 MCE::Child->wait_all;
647 printf STDERR "duration: %0.03f seconds\n", time - $start;
649 Time to spin 2,000 workers and obtain results (in seconds).
651 Results were obtained on a Macbook Pro (2.6 GHz ~ 3.6 GHz with Turbo
652 Boost). Parallel::ForkManager 2.02 uses Moo. Therefore, I ran again
653 with Moo loaded at the top of the script.
654 MCE::Hobo uses MCE::Shared to retrieve data during reaping.
656 MCE::Child uses MCE::Channel, no shared-manager.
657 Version Cygwin Windows Linux macOS FreeBSD
659 MCE::Child 1.843 19.099s 17.091s 0.965s 1.534s 1.229s
661 MCE::Hobo 1.843 20.514s 19.594s 1.246s 1.629s 1.613s
662 P::FM 1.20 19.703s 19.235s 0.875s 1.445s 1.346s
663 MCE::Child 1.843 20.426s 18.417s 1.116s 1.632s 1.338s Moo loaded
665 MCE::Hobo 1.843 21.809s 20.810s 1.407s 1.759s 1.722s Moo loaded
666 P::FM 2.02 21.668s 25.927s 1.882s 2.612s 2.483s Moo used
667 Set posix_exit to avoid all END and destructor processing.
669 This is helpful for reducing overhead when workers exit. Ditto if
670 using a Perl module not parallel safe. The option is ignored on
671 Windows "$^O eq 'MSWin32'".
672 MCE::Child->init( posix_exit => 1, ... );
674 MCE::Hobo->init( posix_exit => 1, ... );
675 Version Cygwin Windows Linux macOS FreeBSD
677 MCE::Child 1.843 19.815s ignored 0.824s 1.284s 1.245s Moo loaded
679 MCE::Hobo 1.843 21.029s ignored 0.953s 1.335s 1.439s Moo loaded
680
682 This demonstration constructs two queues, two handles, starts the
683 shared-manager process if needed, and spawns four workers. For this
684 demonstration, am chunking 64 URLs per job. In reality, one may run
685 with 200 workers and chunk 300 URLs on a 24-way box.
686 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
688 # perl demo.pl -- all output
689 # perl demo.pl >/dev/null -- mngr/child output
690 # perl demo.pl 2>/dev/null -- show results only
691 #
692 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
693 use strict;
695 use warnings;
696 use AnyEvent;
698 use AnyEvent::HTTP;
699 use Time::HiRes qw( time );
700 use MCE::Child;
702 use MCE::Shared;
703 # Construct two queues, input and return.
705 my $que = MCE::Shared->queue();
707 my $ret = MCE::Shared->queue();
708 # Construct shared handles for serializing output from many workers
710 # writing simultaneously. This prevents garbled output.
711 mce_open my $OUT, ">>", \*STDOUT or die "open error: $!";
713 mce_open my $ERR, ">>", \*STDERR or die "open error: $!";
714 # Spawn workers early for minimum memory consumption.
716 MCE::Child->create({ posix_exit => 1 }, 'task', $_) for 1 .. 4;
718 # Obtain or generate input data for workers to process.
720 my ( $count, @urls ) = ( 0 );
722 push @urls, map { "http://127.0.0.$_/" } 1..254;
724 push @urls, map { "http://192.168.0.$_/" } 1..254; # 508 URLs total
725 while ( @urls ) {
727 my @chunk = splice(@urls, 0, 64);
728 $que->enqueue( { ID => ++$count, INPUT => \@chunk } );
729 }
730 # So that workers leave the loop after consuming the queue.
732 $que->end();
734 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
736 # Loop for the manager process. The manager may do other work if
737 # need be and periodically check $ret->pending() not shown here.
738 #
739 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
740 my $start = time;
742 printf {$ERR} "Mngr - entering loop\n";
744 while ( $count ) {
746 my ( $result, $failed ) = $ret->dequeue( 2 );
747 # Remove ID from result, so not treated as a URL item.
749 printf {$ERR} "Mngr - received job %s\n", delete $result->{ID};
751 # Display the URL and the size captured.
753 foreach my $url ( keys %{ $result } ) {
755 printf {$OUT} "%s: %d\n", $url, length($result->{$url})
756 if $result->{$url}; # url has content
757 }
758 # Display URLs could not reach.
760 if ( @{ $failed } ) {
762 foreach my $url ( @{ $failed } ) {
763 print {$OUT} "Failed: $url\n";
764 }
765 }
766 # Decrement the count.
768 $count--;
770 }
771 MCE::Child->wait_all();
773 printf {$ERR} "Mngr - exiting loop\n\n";
775 printf {$ERR} "Duration: %0.3f seconds\n\n", time - $start;
776 exit;
778 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
780 # Child processes enqueue two items ( $result and $failed ) per each
781 # job for the manager process. Likewise, the manager process dequeues
782 # two items above. Optionally, child processes may include the ID in
783 # the result.
784 #
785 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
786 sub task {
788 my ( $id ) = @_;
789 printf {$ERR} "Child $id entering loop\n";
790 while ( my $job = $que->dequeue() ) {
792 my ( $result, $failed ) = ( { ID => $job->{ID} }, [ ] );
793 # Walk URLs, provide a hash and array refs for data.
795 printf {$ERR} "Child $id running job $job->{ID}\n";
797 walk( $job, $result, $failed );
798 # Send results to the manager process.
800 $ret->enqueue( $result, $failed );
802 }
803 printf {$ERR} "Child $id exiting loop\n";
805 }
806 sub walk {
808 my ( $job, $result, $failed ) = @_;
809 # Yielding is critical when running an event loop in parallel.
811 # Not doing so means that the app may reach contention points
812 # with the firewall and likely impose unnecessary hardship at
813 # the OS level. The idea here is not to have multiple workers
814 # initiate HTTP requests to a batch of URLs at the same time.
815 # Yielding behaves similarly like scatter to have the child
816 # process run solo for a fraction of time.
817 MCE::Child->yield( 0.03 );
819 my $cv = AnyEvent->condvar();
821 # Populate the hash ref for the URLs it could reach.
823 # Do not mix AnyEvent timeout with child timeout.
824 # Therefore, choose event timeout when available.
825 foreach my $url ( @{ $job->{INPUT} } ) {
827 $cv->begin();
828 http_get $url, timeout => 2, sub {
829 my ( $data, $headers ) = @_;
830 $result->{$url} = $data;
831 $cv->end();
832 };
833 }
834 $cv->recv();
836 # Populate the array ref for URLs it could not reach.
838 foreach my $url ( @{ $job->{INPUT} } ) {
840 push @{ $failed }, $url unless (exists $result->{ $url });
841 }
842 return;
844 }
845 __END__
847 $ perl demo.pl
849 Child 1 entering loop
851 Child 2 entering loop
852 Child 3 entering loop
853 Mngr - entering loop
854 Child 2 running job 2
855 Child 3 running job 3
856 Child 1 running job 1
857 Child 4 entering loop
858 Child 4 running job 4
859 Child 2 running job 5
860 Mngr - received job 2
861 Child 3 running job 6
862 Mngr - received job 3
863 Child 1 running job 7
864 Mngr - received job 1
865 Child 4 running job 8
866 Mngr - received job 4
867 http://192.168.0.1/: 3729
868 Child 2 exiting loop
869 Mngr - received job 5
870 Child 3 exiting loop
871 Mngr - received job 6
872 Child 1 exiting loop
873 Mngr - received job 7
874 Child 4 exiting loop
875 Mngr - received job 8
876 Mngr - exiting loop
877 Duration: 4.131 seconds
879
881 Making an executable is possible with the PAR::Packer module. On the
882 Windows platform, threads, threads::shared, and exiting via threads are
883 necessary for the binary to exit successfully.
884 # https://metacpan.org/pod/PAR::Packer
886 # https://metacpan.org/pod/pp
887 #
888 # pp -o demo.exe demo.pl
889 # ./demo.exe
890 use strict;
892 use warnings;
893 use if $^O eq "MSWin32", "threads";
895 use if $^O eq "MSWin32", "threads::shared";
896 # Include minimum dependencies for MCE::Child.
898 # Add other modules required by your application here.
899 use Storable ();
901 use Time::HiRes ();
902 # use IO::FDPass (); # optional: for condvar, handle, queue
904 # use Sereal (); # optional: for faster serialization
905 use MCE::Child;
907 use MCE::Shared;
908 # For PAR to work on the Windows platform, one must include manually
910 # any shared modules used by the application.
911 # use MCE::Shared::Array; # if using MCE::Shared->array
913 # use MCE::Shared::Cache; # if using MCE::Shared->cache
914 # use MCE::Shared::Condvar; # if using MCE::Shared->condvar
915 # use MCE::Shared::Handle; # if using MCE::Shared->handle, mce_open
916 # use MCE::Shared::Hash; # if using MCE::Shared->hash
917 # use MCE::Shared::Minidb; # if using MCE::Shared->minidb
918 # use MCE::Shared::Ordhash; # if using MCE::Shared->ordhash
919 # use MCE::Shared::Queue; # if using MCE::Shared->queue
920 # use MCE::Shared::Scalar; # if using MCE::Shared->scalar
921 # Et cetera. Only load modules needed for your application.
923 use MCE::Shared::Sequence; # if using MCE::Shared->sequence
925 my $seq = MCE::Shared->sequence( 1, 9 );
927 sub task {
929 my ( $id ) = @_;
930 while ( defined ( my $num = $seq->next() ) ) {
931 print "$id: $num\n";
932 sleep 1;
933 }
934 }
935 sub main {
937 MCE::Child->new( \&task, $_ ) for 1 .. 3;
938 MCE::Child->wait_all();
939 }
940 # Main must run inside a thread on the Windows platform or workers
942 # will fail duing exiting, causing the exe to crash. The reason is
943 # that PAR or a dependency isn't multi-process safe.
944 ( $^O eq "MSWin32" ) ? threads->create(\&main)->join() : main();
946 threads->exit(0) if $INC{"threads.pm"};
948
950 MCE::Child emits an error when "is_joinable", "is_running", and "join"
951 isn't called by the managed process, where the child was spawned. This
952 is a limitation in MCE::Child only due to not involving a shared-
953 manager process for IPC.
954 This use-case is not typical.
956
958 The inspiration for "MCE::Child" comes from wanting "threads"-like
959 behavior for processes compatible with Perl 5.8. Both can run side-by-
960 side including safe-use by MCE workers. Likewise, the documentation
961 resembles "threads".
962 The inspiration for "wait_all" and "wait_one" comes from the
964 "Parallel::WorkUnit" module.
965
967 · forks
968 · forks::BerkeleyDB
970 · MCE::Hobo
972 · Parallel::ForkManager
974 · Parallel::Loops
976 · Parallel::Prefork
978 · Parallel::WorkUnit
980 · Proc::Fork
982 · Thread::Tie
984 · threads
986
988 MCE, MCE::Channel, MCE::Shared
989
991 Mario E. Roy, <marioeroy AT gmail DOT com>
992perl v5.30.1 2020-02-09 MCE::Child(3)