1MCE::Flow(3) User Contributed Perl Documentation MCE::Flow(3)
2
6 MCE::Flow - Parallel flow model for building creative applications
7
9 This document describes MCE::Flow version 1.866
10
12 MCE::Flow is great for writing custom apps to maximize on all available
13 cores. This module was created to help one harness user_tasks within
14 MCE.
15 It is trivial to parallelize with mce_stream shown below.
17 ## Native map function
19 my @a = map { $_ * 4 } map { $_ * 3 } map { $_ * 2 } 1..10000;
20 ## Same as with MCE::Stream (processing from right to left)
22 @a = mce_stream
23 sub { $_ * 4 }, sub { $_ * 3 }, sub { $_ * 2 }, 1..10000;
24 ## Pass an array reference to have writes occur simultaneously
26 mce_stream \@a,
27 sub { $_ * 4 }, sub { $_ * 3 }, sub { $_ * 2 }, 1..10000;
28 However, let's have MCE::Flow compute the same in parallel. MCE::Queue
30 will be used for data flow among the sub-tasks.
31 use MCE::Flow;
33 use MCE::Queue;
34 This calls for preserving output order.
36 sub preserve_order {
38 my %tmp; my $order_id = 1; my $gather_ref = $_[0];
39 @{ $gather_ref } = (); ## clear the array (optional)
40 return sub {
42 my ($data_ref, $chunk_id) = @_;
43 $tmp{$chunk_id} = $data_ref;
44 while (1) {
46 last unless exists $tmp{$order_id};
47 push @{ $gather_ref }, @{ delete $tmp{$order_id++} };
48 }
49 return;
51 };
52 }
53 Two queues are needed for data flow between the 3 sub-tasks. Notice
55 task_end and how the value from $task_name is used for determining
56 which task has ended.
57 my $b = MCE::Queue->new;
59 my $c = MCE::Queue->new;
60 sub task_end {
62 my ($mce, $task_id, $task_name) = @_;
63 if (defined $mce->{user_tasks}->[$task_id + 1]) {
65 my $n_workers = $mce->{user_tasks}->[$task_id + 1]->{max_workers};
66 if ($task_name eq 'a') {
68 $b->enqueue((undef) x $n_workers);
69 }
70 elsif ($task_name eq 'b') {
71 $c->enqueue((undef) x $n_workers);
72 }
73 }
74 return;
76 }
77 Next are the 3 sub-tasks. The first one reads input and begins the
79 flow. The 2nd task dequeues, performs the calculation, and enqueues
80 into the next. Finally, the last task calls the gather method.
81 Although serialization is done for you automatically, it is done here
83 to save from double serialization. This is the fastest approach for
84 passing data between sub-tasks. Thus, the least overhead.
85 sub task_a {
87 my @ans; my ($mce, $chunk_ref, $chunk_id) = @_;
88 push @ans, map { $_ * 2 } @{ $chunk_ref };
90 $b->enqueue(MCE->freeze([ \@ans, $chunk_id ]));
91 return;
93 }
94 sub task_b {
96 my ($mce) = @_;
97 while (1) {
99 my @ans; my $chunk = $b->dequeue;
100 last unless defined $chunk;
101 $chunk = MCE->thaw($chunk);
103 push @ans, map { $_ * 3 } @{ $chunk->[0] };
104 $c->enqueue(MCE->freeze([ \@ans, $chunk->[1] ]));
105 }
106 return;
108 }
109 sub task_c {
111 my ($mce) = @_;
112 while (1) {
114 my @ans; my $chunk = $c->dequeue;
115 last unless defined $chunk;
116 $chunk = MCE->thaw($chunk);
118 push @ans, map { $_ * 4 } @{ $chunk->[0] };
119 MCE->gather(\@ans, $chunk->[1]);
120 }
121 return;
123 }
124 In summary, MCE::Flow builds out a MCE instance behind the scene and
126 starts running. The task_name (shown), max_workers, and use_threads
127 options can take an anonymous array for specifying the values uniquely
128 per each sub-task.
129 my @a;
131 mce_flow {
133 task_name => [ 'a', 'b', 'c' ], task_end => \&task_end,
134 gather => preserve_order(\@a)
135 }, \&task_a, \&task_b, \&task_c, 1..10000;
137 print "@a\n";
139 If speed is not a concern and wanting to rid of all the MCE->freeze and
141 MCE->thaw statements, simply enqueue and dequeue 2 items at a time. Or
142 better yet, see MCE::Step introduced in MCE 1.506.
143 First, task_end must be updated. The number of undef(s) must match the
145 number of workers times the dequeue count. Otherwise, the script will
146 stall.
147 sub task_end {
149 ...
150 if ($task_name eq 'a') {
151 # $b->enqueue((undef) x $n_workers);
152 $b->enqueue((undef) x ($n_workers * 2));
153 }
154 elsif ($task_name eq 'b') {
155 # $c->enqueue((undef) x $n_workers);
156 $c->enqueue((undef) x ($n_workers * 2));
157 }
158 ...
159 }
160 Next, the 3 sub-tasks enqueuing and dequeuing 2 elements at a time.
162 sub task_a {
164 my @ans; my ($mce, $chunk_ref, $chunk_id) = @_;
165 push @ans, map { $_ * 2 } @{ $chunk_ref };
167 $b->enqueue(\@ans, $chunk_id);
168 return;
170 }
171 sub task_b {
173 my ($mce) = @_;
174 while (1) {
176 my @ans; my ($chunk_ref, $chunk_id) = $b->dequeue(2);
177 last unless defined $chunk_ref;
178 push @ans, map { $_ * 3 } @{ $chunk_ref };
180 $c->enqueue(\@ans, $chunk_id);
181 }
182 return;
184 }
185 sub task_c {
187 my ($mce) = @_;
188 while (1) {
190 my @ans; my ($chunk_ref, $chunk_id) = $c->dequeue(2);
191 last unless defined $chunk_ref;
192 push @ans, map { $_ * 4 } @{ $chunk_ref };
194 MCE->gather(\@ans, $chunk_id);
195 }
196 return;
198 }
199 Finally, run as usual.
201 my @a;
203 mce_flow {
205 task_name => [ 'a', 'b', 'c' ], task_end => \&task_end,
206 gather => preserve_order(\@a)
207 }, \&task_a, \&task_b, \&task_c, 1..10000;
209 print "@a\n";
211
213 Although MCE::Loop may be preferred for running using a single code
214 block, the text below also applies to this module, particularly for the
215 first block.
216 All models in MCE default to 'auto' for chunk_size. The arguments for
218 the block are the same as writing a user_func block using the Core API.
219 Beginning with MCE 1.5, the next input item is placed into the input
221 scalar variable $_ when chunk_size equals 1. Otherwise, $_ points to
222 $chunk_ref containing many items. Basically, line 2 below may be
223 omitted from your code when using $_. One can call MCE->chunk_id to
224 obtain the current chunk id.
225 line 1: user_func => sub {
227 line 2: my ($mce, $chunk_ref, $chunk_id) = @_;
228 line 3:
229 line 4: $_ points to $chunk_ref->[0]
230 line 5: in MCE 1.5 when chunk_size == 1
231 line 6:
232 line 7: $_ points to $chunk_ref
233 line 8: in MCE 1.5 when chunk_size > 1
234 line 9: }
235 Follow this synopsis when chunk_size equals one. Looping is not
237 required from inside the first block. Hence, the block is called once
238 per each item.
239 ## Exports mce_flow, mce_flow_f, and mce_flow_s
241 use MCE::Flow;
242 MCE::Flow::init {
244 chunk_size => 1
245 };
246 ## Array or array_ref
248 mce_flow sub { do_work($_) }, 1..10000;
249 mce_flow sub { do_work($_) }, \@list;
250 ## Important; pass an array_ref for deeply input data
252 mce_flow sub { do_work($_) }, [ [ 0, 1 ], [ 0, 2 ], ... ];
253 mce_flow sub { do_work($_) }, \@deeply_list;
254 ## File path, glob ref, IO::All::{ File, Pipe, STDIO } obj, or scalar ref
256 ## Workers read directly and not involve the manager process
257 mce_flow_f sub { chomp; do_work($_) }, "/path/to/file"; # efficient
258 ## Involves the manager process, therefore slower
260 mce_flow_f sub { chomp; do_work($_) }, $file_handle;
261 mce_flow_f sub { chomp; do_work($_) }, $io;
262 mce_flow_f sub { chomp; do_work($_) }, \$scalar;
263 ## Sequence of numbers (begin, end [, step, format])
265 mce_flow_s sub { do_work($_) }, 1, 10000, 5;
266 mce_flow_s sub { do_work($_) }, [ 1, 10000, 5 ];
267 mce_flow_s sub { do_work($_) }, {
269 begin => 1, end => 10000, step => 5, format => undef
270 };
271
273 Follow this synopsis when chunk_size equals 'auto' or greater than 1.
274 This means having to loop through the chunk from inside the first
275 block.
276 use MCE::Flow;
278 MCE::Flow::init { ## Chunk_size defaults to 'auto' when
280 chunk_size => 'auto' ## not specified. Therefore, the init
281 }; ## function may be omitted.
282 ## Syntax is shown for mce_flow for demonstration purposes.
284 ## Looping inside the block is the same for mce_flow_f and
285 ## mce_flow_s.
286 ## Array or array_ref
288 mce_flow sub { do_work($_) for (@{ $_ }) }, 1..10000;
289 mce_flow sub { do_work($_) for (@{ $_ }) }, \@list;
290 ## Important; pass an array_ref for deeply input data
292 mce_flow sub { do_work($_) for (@{ $_ }) }, [ [ 0, 1 ], [ 0, 2 ], ... ];
293 mce_flow sub { do_work($_) for (@{ $_ }) }, \@deeply_list;
294 ## Resembles code using the core MCE API
296 mce_flow sub {
297 my ($mce, $chunk_ref, $chunk_id) = @_;
298 for (@{ $chunk_ref }) {
300 do_work($_);
301 }
302 }, 1..10000;
304 Chunking reduces the number of IPC calls behind the scene. Think in
306 terms of chunks whenever processing a large amount of data. For
307 relatively small data, choosing 1 for chunk_size is fine.
308
310 The following list options which may be overridden when loading the
311 module.
312 use Sereal qw( encode_sereal decode_sereal );
314 use CBOR::XS qw( encode_cbor decode_cbor );
315 use JSON::XS qw( encode_json decode_json );
316 use MCE::Flow
318 max_workers => 8, # Default 'auto'
319 chunk_size => 500, # Default 'auto'
320 tmp_dir => "/path/to/app/tmp", # $MCE::Signal::tmp_dir
321 freeze => \&encode_sereal, # \&Storable::freeze
322 thaw => \&decode_sereal # \&Storable::thaw
323 ;
324 From MCE 1.8 onwards, Sereal 3.015+ is loaded automatically if
326 available. Specify "Sereal => 0" to use Storable instead.
327 use MCE::Flow Sereal => 0;
329
331 MCE::Flow->init ( options )
332 MCE::Flow::init { options }
333 The init function accepts a hash of MCE options. Unlike with
335 MCE::Stream, both gather and bounds_only options may be specified when
336 calling init (not shown below).
337 use MCE::Flow;
339 MCE::Flow::init {
341 chunk_size => 1, max_workers => 4,
342 user_begin => sub {
344 print "## ", MCE->wid, " started\n";
345 },
346 user_end => sub {
348 print "## ", MCE->wid, " completed\n";
349 }
350 };
351 my %a = mce_flow sub { MCE->gather($_, $_ * $_) }, 1..100;
353 print "\n", "@a{1..100}", "\n";
355 -- Output
357 ## 3 started
359 ## 2 started
360 ## 4 started
361 ## 1 started
362 ## 2 completed
363 ## 4 completed
364 ## 3 completed
365 ## 1 completed
366 1 4 9 16 25 36 49 64 81 100 121 144 169 196 225 256 289 324 361
368 400 441 484 529 576 625 676 729 784 841 900 961 1024 1089 1156
369 1225 1296 1369 1444 1521 1600 1681 1764 1849 1936 2025 2116 2209
370 2304 2401 2500 2601 2704 2809 2916 3025 3136 3249 3364 3481 3600
371 3721 3844 3969 4096 4225 4356 4489 4624 4761 4900 5041 5184 5329
372 5476 5625 5776 5929 6084 6241 6400 6561 6724 6889 7056 7225 7396
373 7569 7744 7921 8100 8281 8464 8649 8836 9025 9216 9409 9604 9801
374 10000
375 Like with MCE::Flow::init above, MCE options may be specified using an
377 anonymous hash for the first argument. Notice how task_name,
378 max_workers, and use_threads can take an anonymous array for setting
379 uniquely per each code block.
380 Unlike MCE::Stream which processes from right-to-left, MCE::Flow begins
382 with the first code block, thus processing from left-to-right.
383 use threads;
385 use MCE::Flow;
386 my @a = mce_flow {
388 task_name => [ 'a', 'b', 'c' ],
389 max_workers => [ 3, 4, 2, ],
390 use_threads => [ 1, 0, 0, ],
391 user_end => sub {
393 my ($mce, $task_id, $task_name) = @_;
394 MCE->print("$task_id - $task_name completed\n");
395 },
396 task_end => sub {
398 my ($mce, $task_id, $task_name) = @_;
399 MCE->print("$task_id - $task_name ended\n");
400 }
401 },
402 sub { sleep 1; }, ## 3 workers, named a
403 sub { sleep 2; }, ## 4 workers, named b
404 sub { sleep 3; }; ## 2 workers, named c
405 -- Output
407 0 - a completed
409 0 - a completed
410 0 - a completed
411 0 - a ended
412 1 - b completed
413 1 - b completed
414 1 - b completed
415 1 - b completed
416 1 - b ended
417 2 - c completed
418 2 - c completed
419 2 - c ended
420
422 Although input data is optional for MCE::Flow, the following assumes
423 chunk_size equals 1 in order to demonstrate all the possibilities for
424 providing input data.
425 MCE::Flow->run ( sub { code }, list )
427 mce_flow sub { code }, list
428 Input data may be defined using a list, an array ref, or a hash ref.
430 Unlike MCE::Loop, Map, and Grep which take a block as "{ ... }", Flow
432 takes a "sub { ... }" or a code reference. The other difference is that
433 the comma is needed after the block.
434 # $_ contains the item when chunk_size => 1
436 mce_flow sub { do_work($_) }, 1..1000;
438 mce_flow sub { do_work($_) }, \@list;
439 # Important; pass an array_ref for deeply input data
441 mce_flow sub { do_work($_) }, [ [ 0, 1 ], [ 0, 2 ], ... ];
443 mce_flow sub { do_work($_) }, \@deeply_list;
444 # Chunking; any chunk_size => 1 or greater
446 my %res = mce_flow sub {
448 my ($mce, $chunk_ref, $chunk_id) = @_;
449 my %ret;
450 for my $item (@{ $chunk_ref }) {
451 $ret{$item} = $item * 2;
452 }
453 MCE->gather(%ret);
454 },
455 \@list;
456 # Input hash; current API available since 1.828
458 my %res = mce_flow sub {
460 my ($mce, $chunk_ref, $chunk_id) = @_;
461 my %ret;
462 for my $key (keys %{ $chunk_ref }) {
463 $ret{$key} = $chunk_ref->{$key} * 2;
464 }
465 MCE->gather(%ret);
466 },
467 \%hash;
468 # Unlike MCE::Loop, MCE::Flow doesn't need input to run
470 mce_flow { max_workers => 4 }, sub {
472 MCE->say( MCE->wid );
473 };
474 # ... and can run multiple tasks
476 mce_flow {
478 max_workers => [ 1, 3 ],
479 task_name => [ 'p', 'c' ]
480 },
481 sub {
482 # 1 producer
483 MCE->say( "producer: ", MCE->wid );
484 },
485 sub {
486 # 3 consumers
487 MCE->say( "consumer: ", MCE->wid );
488 };
489 # Here, options are specified via init
491 MCE::Flow::init {
493 max_workers => [ 1, 3 ],
494 task_name => [ 'p', 'c' ]
495 };
496 mce_flow \&producer, \&consumers;
498 MCE::Flow->run_file ( sub { code }, file )
500 mce_flow_f sub { code }, file
501 The fastest of these is the /path/to/file. Workers communicate the next
503 offset position among themselves with zero interaction by the manager
504 process.
505 "IO::All" { File, Pipe, STDIO } is supported since MCE 1.845.
507 # $_ contains the line when chunk_size => 1
509 mce_flow_f sub { $_ }, "/path/to/file"; # faster
511 mce_flow_f sub { $_ }, $file_handle;
512 mce_flow_f sub { $_ }, $io; # IO::All
513 mce_flow_f sub { $_ }, \$scalar;
514 # chunking, any chunk_size => 1 or greater
516 my %res = mce_flow_f sub {
518 my ($mce, $chunk_ref, $chunk_id) = @_;
519 my $buf = '';
520 for my $line (@{ $chunk_ref }) {
521 $buf .= $line;
522 }
523 MCE->gather($chunk_id, $buf);
524 },
525 "/path/to/file";
526 MCE::Flow->run_seq ( sub { code }, $beg, $end [, $step, $fmt ] )
528 mce_flow_s sub { code }, $beg, $end [, $step, $fmt ]
529 Sequence may be defined as a list, an array reference, or a hash
531 reference. The functions require both begin and end values to run.
532 Step and format are optional. The format is passed to sprintf (% may be
533 omitted below).
534 my ($beg, $end, $step, $fmt) = (10, 20, 0.1, "%4.1f");
536 # $_ contains the sequence number when chunk_size => 1
538 mce_flow_s sub { $_ }, $beg, $end, $step, $fmt;
540 mce_flow_s sub { $_ }, [ $beg, $end, $step, $fmt ];
541 mce_flow_s sub { $_ }, {
543 begin => $beg, end => $end,
544 step => $step, format => $fmt
545 };
546 # chunking, any chunk_size => 1 or greater
548 my %res = mce_flow_s sub {
550 my ($mce, $chunk_ref, $chunk_id) = @_;
551 my $buf = '';
552 for my $seq (@{ $chunk_ref }) {
553 $buf .= "$seq\n";
554 }
555 MCE->gather($chunk_id, $buf);
556 },
557 [ $beg, $end ];
558 The sequence engine can compute 'begin' and 'end' items only, for the
560 chunk, and not the items in between (hence boundaries only). This
561 option applies to sequence only and has no effect when chunk_size
562 equals 1.
563 The time to run is 0.006s below. This becomes 0.827s without the
565 bounds_only option due to computing all items in between, thus creating
566 a very large array. Basically, specify bounds_only => 1 when boundaries
567 is all you need for looping inside the block; e.g. Monte Carlo
568 simulations.
569 Time was measured using 1 worker to emphasize the difference.
571 use MCE::Flow;
573 MCE::Flow::init {
575 max_workers => 1, chunk_size => 1_250_000,
576 bounds_only => 1
577 };
578 # Typically, the input scalar $_ contains the sequence number
580 # when chunk_size => 1, unless the bounds_only option is set
581 # which is the case here. Thus, $_ points to $chunk_ref.
582 mce_flow_s sub {
584 my ($mce, $chunk_ref, $chunk_id) = @_;
585 # $chunk_ref contains 2 items, not 1_250_000
587 # my ( $begin, $end ) = ( $_->[0], $_->[1] );
588 my $begin = $chunk_ref->[0];
590 my $end = $chunk_ref->[1];
591 # for my $seq ( $begin .. $end ) {
593 # ...
594 # }
595 MCE->printf("%7d .. %8d\n", $begin, $end);
597 },
598 [ 1, 10_000_000 ];
599 -- Output
601 1 .. 1250000
603 1250001 .. 2500000
604 2500001 .. 3750000
605 3750001 .. 5000000
606 5000001 .. 6250000
607 6250001 .. 7500000
608 7500001 .. 8750000
609 8750001 .. 10000000
610 MCE::Flow->run ( { input_data => iterator }, sub { code } )
612 mce_flow { input_data => iterator }, sub { code }
613 An iterator reference may be specified for input_data. The only other
615 way is to specify input_data via MCE::Flow::init. This prevents
616 MCE::Flow from configuring the iterator reference as another user task
617 which will not work.
618 Iterators are described under section "SYNTAX for INPUT_DATA" at
620 MCE::Core.
621 MCE::Flow::init {
623 input_data => iterator
624 };
625 mce_flow sub { $_ };
627
629 Unlike MCE::Map where gather and output order are done for you
630 automatically, the gather method is used to have results sent back to
631 the manager process.
632 use MCE::Flow chunk_size => 1;
634 ## Output order is not guaranteed.
636 my @a1 = mce_flow sub { MCE->gather($_ * 2) }, 1..100;
637 print "@a1\n\n";
638 ## Outputs to a hash instead (key, value).
640 my %h1 = mce_flow sub { MCE->gather($_, $_ * 2) }, 1..100;
641 print "@h1{1..100}\n\n";
642 ## This does the same thing due to chunk_id starting at one.
644 my %h2 = mce_flow sub { MCE->gather(MCE->chunk_id, $_ * 2) }, 1..100;
645 print "@h2{1..100}\n\n";
646 The gather method may be called multiple times within the block unlike
648 return which would leave the block. Therefore, think of gather as
649 yielding results immediately to the manager process without actually
650 leaving the block.
651 use MCE::Flow chunk_size => 1, max_workers => 3;
653 my @hosts = qw(
655 hosta hostb hostc hostd hoste
656 );
657 my %h3 = mce_flow sub {
659 my ($output, $error, $status); my $host = $_;
660 ## Do something with $host;
662 $output = "Worker ". MCE->wid .": Hello from $host";
663 if (MCE->chunk_id % 3 == 0) {
665 ## Simulating an error condition
666 local $? = 1; $status = $?;
667 $error = "Error from $host"
668 }
669 else {
670 $status = 0;
671 }
672 ## Ensure unique keys (key, value) when gathering to
674 ## a hash.
675 MCE->gather("$host.out", $output);
676 MCE->gather("$host.err", $error) if (defined $error);
677 MCE->gather("$host.sta", $status);
678 }, @hosts;
680 foreach my $host (@hosts) {
682 print $h3{"$host.out"}, "\n";
683 print $h3{"$host.err"}, "\n" if (exists $h3{"$host.err"});
684 print "Exit status: ", $h3{"$host.sta"}, "\n\n";
685 }
686 -- Output
688 Worker 3: Hello from hosta
690 Exit status: 0
691 Worker 2: Hello from hostb
693 Exit status: 0
694 Worker 1: Hello from hostc
696 Error from hostc
697 Exit status: 1
698 Worker 3: Hello from hostd
700 Exit status: 0
701 Worker 2: Hello from hoste
703 Exit status: 0
704 The following uses an anonymous array containing 3 elements when
706 gathering data. Serialization is automatic behind the scene.
707 my %h3 = mce_flow sub {
709 ...
710 MCE->gather($host, [$output, $error, $status]);
712 }, @hosts;
714 foreach my $host (@hosts) {
716 print $h3{$host}->[0], "\n";
717 print $h3{$host}->[1], "\n" if (defined $h3{$host}->[1]);
718 print "Exit status: ", $h3{$host}->[2], "\n\n";
719 }
720 Although MCE::Map comes to mind, one may want additional control when
722 gathering data such as retaining output order.
723 use MCE::Flow;
725 sub preserve_order {
727 my %tmp; my $order_id = 1; my $gather_ref = $_[0];
728 return sub {
730 $tmp{ (shift) } = \@_;
731 while (1) {
733 last unless exists $tmp{$order_id};
734 push @{ $gather_ref }, @{ delete $tmp{$order_id++} };
735 }
736 return;
738 };
739 }
740 ## Workers persist for the most part after running. Though, not always
742 ## the case and depends on Perl. Pass a reference to a subroutine if
743 ## workers must persist; e.g. mce_flow { ... }, \&foo, 1..100000.
744 MCE::Flow::init {
746 chunk_size => 'auto', max_workers => 'auto'
747 };
748 for (1..2) {
750 my @m2;
751 mce_flow {
753 gather => preserve_order(\@m2)
754 },
755 sub {
756 my @a; my ($mce, $chunk_ref, $chunk_id) = @_;
757 ## Compute the entire chunk data at once.
759 push @a, map { $_ * 2 } @{ $chunk_ref };
760 ## Afterwards, invoke the gather feature, which
762 ## will direct the data to the callback function.
763 MCE->gather(MCE->chunk_id, @a);
764 }, 1..100000;
766 print scalar @m2, "\n";
768 }
769 MCE::Flow::finish;
771 All 6 models support 'auto' for chunk_size unlike the Core API. Think
773 of the models as the basis for providing JIT for MCE. They create the
774 instance, tune max_workers, and tune chunk_size automatically
775 regardless of the hardware.
776 The following does the same thing using the Core API. Workers persist
778 after running.
779 use MCE;
781 sub preserve_order {
783 ...
784 }
785 my $mce = MCE->new(
787 max_workers => 'auto', chunk_size => 8000,
788 user_func => sub {
790 my @a; my ($mce, $chunk_ref, $chunk_id) = @_;
791 ## Compute the entire chunk data at once.
793 push @a, map { $_ * 2 } @{ $chunk_ref };
794 ## Afterwards, invoke the gather feature, which
796 ## will direct the data to the callback function.
797 MCE->gather(MCE->chunk_id, @a);
798 }
799 );
800 for (1..2) {
802 my @m2;
803 $mce->process({ gather => preserve_order(\@m2) }, [1..100000]);
805 print scalar @m2, "\n";
807 }
808 $mce->shutdown;
810
812 MCE::Flow->finish
813 MCE::Flow::finish
814 Workers remain persistent as much as possible after running. Shutdown
816 occurs automatically when the script terminates. Call finish when
817 workers are no longer needed.
818 use MCE::Flow;
820 MCE::Flow::init {
822 chunk_size => 20, max_workers => 'auto'
823 };
824 mce_flow sub { ... }, 1..100;
826 MCE::Flow::finish;
828
830 MCE, MCE::Core
831
833 Mario E. Roy, <marioeroy AT gmail DOT com>
834perl v5.30.1 2020-02-09 MCE::Flow(3)