1MCE::Core(3) User Contributed Perl Documentation MCE::Core(3)
2
6 MCE::Core - Documentation describing the core MCE API
7
9 This document describes MCE::Core version 1.889
10
12 This is a simplistic use case of MCE running with 5 workers.
13 # Construction using the Core API
15 use MCE;
17 my $mce = MCE->new(
19 max_workers => 5,
20 user_func => sub {
21 my ($mce) = @_;
22 $mce->say("Hello from " . $mce->wid);
23 }
24 );
25 $mce->run;
27 # Construction using a MCE model
29 use MCE::Flow max_workers => 5;
31 mce_flow sub {
33 my ($mce) = @_;
34 MCE->say("Hello from " . MCE->wid);
35 };
36 -- Output
38 Hello from 2
40 Hello from 4
41 Hello from 5
42 Hello from 1
43 Hello from 3
44 MCE->new ( [ options ] )
46 Below, a new instance is configured with all available options.
47 use MCE;
49 my $mce = MCE->new(
51 max_workers => 8, # Default 1
53 # Number of workers to spawn.
55 # MCE sets an upper-limit of 8 for 'auto'. MCE 1.521+.
57 # max_workers => 'auto', # # of lcores, 8 maximum
58 # max_workers => 'auto-1', # 7 on HW with 16 lcores
59 # max_workers => 'auto-1', # 3 on HW with 4 lcores
60 # Specify a percentage. MCE 1.875+.
62 # max_workers => '25%', # 4 on HW with 16 lcores
63 # max_workers => '50%', # 8 on HW with 16 lcores
64 # Run on all logical cores.
66 # max_workers => MCE::Util::get_ncpu(),
67 chunk_size => 2000, # Default 1
69 # Can also take a suffix; k (kibiBytes) or m (mebiBytes).
71 # The default is 1 when using the Core API and 'auto' for
72 # MCE Models. For arrays or queues, chunk_size means the
73 # number of records per chunk. For iterators, MCE will not
74 # use chunk_size, though the iterator may use it to determine
75 # how much to return per iteration. For files, smaller than or
76 # equal to 8192 is the number of records. Greater than 8192
77 # is the number of bytes. MCE reads until the end of record
78 # before calling user_func.
79 # chunk_size => 1, # Consists of 1 record
81 # chunk_size => 1000, # Consists of 1000 records
82 # chunk_size => '16k', # Approximate 16 kibiBytes (KiB)
83 # chunk_size => '20m', # Approximate 20 mebiBytes (MiB)
84 tmp_dir => $tmp_dir, # Default $MCE::Signal::tmp_dir
86 # Default is $MCE::Signal::tmp_dir which points to
88 # $ENV{TEMP} if defined. Otherwise, tmp_dir points
89 # to a location under /tmp.
90 freeze => \&encode_sereal, # Default \&Storable::freeze
92 thaw => \&decode_sereal, # Default \&Storable::thaw
93 # Release 1.412 allows freeze and thaw to be overridden.
95 # Simply include a serialization module prior to loading
96 # MCE. Configure freeze/thaw options.
97 # use Sereal qw( encode_sereal decode_sereal );
99 # use CBOR::XS qw( encode_cbor decode_cbor );
100 # use JSON::XS qw( encode_json decode_json );
101 #
102 # use MCE;
103 gather => \@a, # Default undef
105 # Release 1.5 allows for gathering of data to an array or
107 # hash reference, a MCE::Queue/Thread::Queue object, or code
108 # reference. One invokes gathering by calling the gather
109 # method as often as needed.
110 # gather => \@array,
112 # gather => \%hash,
113 # gather => $queue,
114 # gather => \&order,
115 init_relay => 0, # Default undef
117 # For specifying the initial relay value. Allowed values
119 # are array_ref, hash_ref, or scalar. The MCE::Relay module
120 # is loaded automatically when specified.
121 # init_relay => \@array,
123 # init_relay => \%hash,
124 # init_relay => scalar,
125 input_data => $input_file, # Default undef
127 RS => "\n>", # Default undef
128 # input_data => '/path/to/file' # Process file
130 # input_data => \@array # Process array
131 # input_data => \*FILE_HNDL # Process file handle
132 # input_data => $io # Process IO::All { File, Pipe, STDIO }
133 # input_data => \$scalar # Treated like a file
134 # input_data => \&iterator # User specified iterator
135 # The RS option (for input record separator) applies to files
137 # and file handles.
138 # MCE applies additional logic when RS begins with a newline
140 # character; e.g. RS => "\n>". It trims away characters after
141 # the newline and prepends them to the next record.
142 #
143 # Typically, the left side is what happens for $/ = "\n>".
144 # The right side is what user_func receives.
145 #
146 # All records begin with > and end with \n
147 # Record 1: >seq1 ... \n> (to) >seq1 ... \n
148 # Record 2: seq2 ... \n> >seq2 ... \n
149 # Record 3: seq3 ... \n> >seq3 ... \n
150 # Last Rec: seqN ... \n >seqN ... \n
151 loop_timeout => 20, # Default 0
153 # Added in 1.7, enables the manager process to timeout of a read
155 # operation on channel 0 (UNIX platforms only). The manager process
156 # decrements the total workers running for any worker which have
157 # died in an uncontrollable manner. Specify this option if on
158 # occassion a worker dies unexpectedly (i.e. from an XS module).
159 # Option works with init_relay on UNIX platforms since MCE 1.844.
161 # A number smaller than 5 is silently increased to 5.
162 max_retries => 2, # Default 0
164 # This option, added in 1.7, causes MCE to retry a failed
166 # chunk from a worker dying while processing input data or
167 # sequence of numbers.
168 parallel_io => 1, # Default 0
170 posix_exit => 1, # Default 0
171 use_slurpio => 1, # Default 0
172 # The parallel_io option enables parallel reads during large
174 # slurpio, useful when reading from fast storage. Do not enable
175 # parallel_io when running MCE on many nodes with input coming
176 # from shared storage.
177 # Set posix_exit to avoid all END and destructor processing.
179 # Constructing MCE inside a thread implies 1 or if present CGI,
180 # FCGI, Coro, Curses, Gearman::Util, Gearman::XS, LWP::UserAgent,
181 # Mojo::IOLoop, STFL, Tk, Wx, or Win32::GUI.
182 # Enable slurpio to pass the raw chunk (scalar ref) to the user
184 # function when reading input files.
185 use_threads => 1, # Auto 0 or 1
187 # By default MCE spawns child processes on UNIX platforms and
189 # threads on Windows (i.e. $^O eq 'MSWin32').
190 # MCE supports threads via two threading libraries if threads
192 # is preferred over child processes. The use of threads requires
193 # a thread library prior to loading MCE, causing the use_threads
194 # option to default to 1. Specify 0 for child processes.
195 #
196 # use threads; use forks;
197 # use threads::shared; use forks::shared;
198 # use MCE (or) use MCE; (or) use MCE;
199 spawn_delay => 0.045, # Default undef
201 submit_delay => 0.015, # Default undef
202 job_delay => 0.060, # Default undef
203 # Time to wait in fractional seconds after spawning a worker,
205 # after submitting parameters to worker (MCE->run, MCE->process),
206 # and worker running (one time staggered delay).
207 # Specify job_delay to stagger workers connecting to a database.
209 on_post_exit => \&on_post_exit, # Default undef
211 on_post_run => \&on_post_run, # Default undef
212 # Execute the code block after a worker exits or dies.
214 # (i.e. MCE->exit, exit, die)
215 # Execute the code block after running.
217 # (i.e. MCE->process, MCE->run)
218 progress => sub { ... }, # Default undef
220 # A code block for receiving info on the progress made.
222 # See section labeled "MCE PROGRESS DEMONSTRATIONS" at the
223 # end of this document.
224 user_args => { env => 'test' }, # Default undef
226 # MCE release 1.4 added a new parameter to allow one to
228 # specify arbitrary arguments such as a string, an ARRAY
229 # or HASH reference. Workers can access this directly.
230 # (i.e. my $args = $mce->{user_args} or MCE->user_args)
231 user_begin => \&user_begin, # Default undef
233 user_func => \&user_func, # Default undef
234 user_end => \&user_end, # Default undef
235 # Think of user_begin, user_func, and user_end as in
237 # the awk scripting language:
238 # awk 'BEGIN { begin } { func } { func } ... END { end }'
239 # MCE workers call user_begin once at the start of a job,
241 # then user_func repeatedly until no chunks remain.
242 # Afterwards, user_end is called.
243 user_error => \&user_error, # Default undef
245 user_output => \&user_output, # Default undef
246 # MCE will forward data to user_error/user_output,
248 # when defined, for the following methods.
249 # MCE->sendto(\*STDERR, "sent to user_error\n");
251 # MCE->printf(\*STDERR, "%s\n", "sent to user_error");
252 # MCE->print(\*STDERR, "sent to user_error\n");
253 # MCE->say(\*STDERR, "sent to user_error");
254 # MCE->sendto(\*STDOUT, "sent to user_output\n");
256 # MCE->printf("%s\n", "sent to user_output");
257 # MCE->print("sent to user_output\n");
258 # MCE->say("sent to user_output");
259 stderr_file => 'err_file', # Default STDERR
261 stdout_file => 'out_file', # Default STDOUT
262 # Or to file; user_error and user_output take precedence.
264 flush_file => 0, # Default 1
266 flush_stderr => 0, # Default 1
267 flush_stdout => 0, # Default 1
268 # Flush sendto file, standard error, or standard output.
270 interval => {
272 delay => 0.007 [, max_nodes => 4, node_id => 1 ]
273 },
274 # For use with the yield method introduced in MCE 1.5.
276 # Both max_nodes & node_id are optional and default to 1.
277 # Delay is the amount of time between intervals.
278 # interval => 0.007 # Shorter; MCE 1.506+
280 sequence => { # Default undef
282 begin => -1, end => 1 [, step => 0.1 [, format => "%4.1f" ] ]
283 },
284 bounds_only => 1, # Default undef
286 # For looping through a sequence of numbers in parallel.
288 # STEP, if omitted, defaults to 1 if BEGIN is smaller than
289 # END or -1 if BEGIN is greater than END. The FORMAT string
290 # is passed to sprintf behind the scene (% may be omitted).
291 # e.g. $seq_n_formatted = sprintf("%4.1f", $seq_n);
292 # Do not specify both options; input_data and sequence.
294 # Release 1.4 allows one to specify an array reference.
295 # e.g. sequence => [ -1, 1, 0.1, "%4.1f" ]
296 # The bounds_only => 1 option will compute the 'begin' and
298 # 'end' items only for the chunk and not the items in between
299 # (hence boundaries only). This option has no effect when
300 # sequence is not specified or chunk_size equals 1.
301 # my $begin = $chunk_ref->[0]; my $end = $chunk_ref->[1];
303 task_end => \&task_end, # Default undef
305 # This is called by the manager process after the task
307 # has completed processing. MCE 1.5 allows this option
308 # to be specified at the top level.
309 task_name => 'string', # Default 'MCE'
311 # Added in MCE 1.5 and mainly beneficial for user_tasks.
313 # One may specify a unique name per each sub-task.
314 # The string is passed as the 3rd arg to task_end.
315 user_tasks => [ # Default undef
317 { ... }, # Options for task 0
318 { ... }, # Options for task 1
319 { ... }, # Options for task 2
320 ],
321 # Takes a list of hash references, each allowing up to 17
323 # options. All other MCE options are ignored. The init_relay,
324 # input_data, RS, and use_slurpio options are applicable to
325 # the first task only.
326 # max_workers, chunk_size, input_data, interval, sequence,
328 # bounds_only, user_args, user_begin, user_end, user_func,
329 # gather, task_end, task_name, use_slurpio, use_threads,
330 # init_relay, RS
331 # Options not specified here will default to same option
333 # specified at the top level.
334 );
335 EXPORT_CONST, CONST
337 There are 3 constants which are exportable. Using the constants in lieu
338 of 0,1,2 makes it more legible when accessing the user_func arguments
339 directly.
340 SELF CHUNK CID - MCE CONSTANTS
342 Exports SELF => 0, CHUNK => 1, and CID => 2.
344 use MCE export_const => 1;
346 use MCE const => 1; # Shorter; MCE 1.415+
347 user_func => sub {
349 # my ($mce, $chunk_ref, $chunk_id) = @_;
350 print "Hello from ", $_[SELF]->wid, "\n";
351 }
352 MCE 1.5 allows all public method to be called directly.
354 use MCE;
356 user_func => sub {
358 # my ($mce, $chunk_ref, $chunk_id) = @_;
359 print "Hello from ", MCE->wid, "\n";
360 }
361 OVERRIDING DEFAULTS
363 The following list options which may be overridden when loading the
364 module.
365 use Sereal qw( encode_sereal decode_sereal );
367 use CBOR::XS qw( encode_cbor decode_cbor );
368 use JSON::XS qw( encode_json decode_json );
369 use MCE
371 max_workers => 4, # Default 1
372 chunk_size => 100, # Default 1
373 tmp_dir => "/path/to/app/tmp", # $MCE::Signal::tmp_dir
374 freeze => \&encode_sereal, # \&Storable::freeze
375 thaw => \&decode_sereal, # \&Storable::thaw
376 init_relay => 0, # Default undef; MCE 1.882+
377 use_threads => 0, # Default undef; MCE 1.882+
378 ;
379 my $mce = MCE->new( ... );
381 From MCE 1.8 onwards, Sereal 3.015+ is loaded automatically if
383 available. Specify "Sereal => 0" to use Storable instead.
384 use MCE Sereal => 0;
386 RUNNING
388 Run calls spawn, submits the job; workers call user_begin, user_func,
389 and user_end. Run shuts down workers afterwards. Call spawn whenever
390 the need arises for large data structures prior to running.
391 $mce->spawn; # Call early if desired
393 $mce->run; # Call run or process below
395 # Acquire data arrays and/or input_files. Workers persist after
397 # processing.
398 $mce->process(\@input_data_1); # Process array
400 $mce->process(\@input_data_2);
401 $mce->process(\@input_data_n);
402 $mce->process(\%input_hash_1); # Process hash, current API
404 $mce->process(\%input_hash_2); # available since 1.828
405 $mce->process(\%input_hash_n);
406 $mce->process('input_file_1'); # Process file
408 $mce->process('input_file_2');
409 $mce->process('input_file_n');
410 $mce->shutdown; # Shutdown workers
412 SYNTAX for ON_POST_EXIT
414 Often times, one may want to capture the exit status. The on_post_exit
415 option, if defined, is executed immediately by the manager process
416 after a worker exits via exit (children only), MCE->exit (children and
417 threads), or die.
418 The format of $e->{pid} is PID_123 for children and THR_123 for
420 threads.
421 my $restart_flag = 1;
423 sub on_post_exit {
425 my ($mce, $e) = @_;
426 # Display all possible hash elements.
428 print "$e->{wid}: $e->{pid}: $e->{status}: $e->{msg}: $e->{id}\n";
429 # Restart this worker if desired.
431 if ($restart_flag && $e->{wid} == 2) {
432 $mce->restart_worker;
433 $restart_flag = 0;
434 }
435 }
436 sub user_func {
438 my ($mce) = @_;
439 MCE->exit(0, 'msg_foo', 1000 + MCE->wid); # Args, not necessary
440 }
441 my $mce = MCE->new(
443 on_post_exit => \&on_post_exit,
444 user_func => \&user_func,
445 max_workers => 3
446 );
447 $mce->run;
449 -- Output (child processes)
451 2: PID_33223: 0: msg_foo: 1002
453 1: PID_33222: 0: msg_foo: 1001
454 3: PID_33224: 0: msg_foo: 1003
455 2: PID_33225: 0: msg_foo: 1002
456 -- Output (running with threads)
458 3: TID_3: 0: msg_foo: 1003
460 2: TID_2: 0: msg_foo: 1002
461 1: TID_1: 0: msg_foo: 1001
462 2: TID_4: 0: msg_foo: 1002
463 SYNTAX for ON_POST_RUN
465 The on_post_run option, if defined, is executed immediately by the
466 manager process after running MCE->process or MCE->run. This option
467 receives an array reference of hashes.
468 The difference between on_post_exit and on_post_run is that the former
470 is called immediately whereas the latter is called after all workers
471 have completed running.
472 sub on_post_run {
474 my ($mce, $status_ref) = @_;
475 foreach my $e ( @{ $status_ref } ) {
476 # Display all possible hash elements.
477 print "$e->{wid}: $e->{pid}: $e->{status}: $e->{msg}: $e->{id}\n";
478 }
479 }
480 sub user_func {
482 my ($mce) = @_;
483 MCE->exit(0, 'msg_foo', 1000 + MCE->wid); # Args, not necessary
484 }
485 my $mce = MCE->new(
487 on_post_run => \&on_post_run,
488 user_func => \&user_func,
489 max_workers => 3
490 );
491 $mce->run;
493 -- Output (child processes)
495 3: PID_33174: 0: msg_foo: 1003
497 1: PID_33172: 0: msg_foo: 1001
498 2: PID_33173: 0: msg_foo: 1002
499 -- Output (running with threads)
501 2: TID_2: 0: msg_foo: 1002
503 3: TID_3: 0: msg_foo: 1003
504 1: TID_1: 0: msg_foo: 1001
505 SYNTAX for INPUT_DATA
507 MCE supports many ways to specify input_data. Support for iterators was
508 added in MCE 1.505. The RS option allows one to specify the record
509 separator when processing files.
510 MCE is a chunking engine. Therefore, chunk_size is applicable to
512 input_data. Specifying 1 for use_slurpio causes user_func to receive a
513 scalar reference containing the raw data (applicable to files only)
514 instead of an array reference.
515 "IO::All" { File, Pipe, STDIO } is supported since MCE 1.845.
517 input_data => '/path/to/file', # process file
519 input_data => \@array, # process array
520 input_data => \%hash, # process hash, API since 1.828
521 input_data => \*FILE_HNDL, # process file handle
522 input_data => $fh, # open $fh, "<", "file"
523 input_data => $fh, # IO::File "file", "r"
524 input_data => $fh, # IO::Uncompress::Gunzip "file.gz"
525 input_data => $io, # IO::All { File, Pipe, STDIO }
526 input_data => \$scalar, # treated like a file
527 input_data => \&iterator, # user specified iterator
528 chunk_size => 1, # >1 means looping inside user_func
530 use_slurpio => 1, # $chunk_ref is a scalar ref
531 RS => "\n>", # input record separator
532 The chunk_size value determines the chunking mode to use when
534 processing files. Otherwise, chunk_size is the number of elements for
535 arrays. For files, a chunk size value of <= 8192 is how many records to
536 read. Greater than 8192 is how many bytes to read. MCE appends (the
537 rest) up to the next record separator.
538 chunk_size => 8192, # Consists of 8192 records
540 chunk_size => 8193, # Approximate 8193 bytes for files
541 chunk_size => 1, # Consists of 1 record or element
543 chunk_size => 1000, # Consists of 1000 records
544 chunk_size => '16k', # Approximate 16 kibiBytes (KiB)
545 chunk_size => '20m', # Approximate 20 mebiBytes (MiB)
546 The construction for user_func when chunk_size > 1 and assuming
548 use_slurpio equals 0.
549 user_func => sub {
551 my ($mce, $chunk_ref, $chunk_id) = @_;
552 # $_ is $chunk_ref->[0] when chunk_size equals 1
554 # $_ is $chunk_ref otherwise; $_ can be used below
555 for my $record ( @{ $chunk_ref } ) {
557 print "$chunk_id: $record\n";
558 }
559 }
560 # input_data => \%hash
562 # current API available since 1.828
563 user_func => sub {
565 my ($mce, $chunk_ref, $chunk_id) = @_;
566 # $_ points to $chunk_ref regardless of chunk_size
568 for my $key ( keys %{ $chunk_ref } ) {
570 print "$key: ", $chunk_ref->{$key}, "\n";
571 }
572 }
573 Specifying a value for input_data is straight forward for arrays and
575 files. The next several examples specify an iterator reference for
576 input_data.
577 use MCE;
579 # A factory function which creates a closure (the iterator itself)
581 # for generating a sequence of numbers. The external variables
582 # ($n, $max, $step) are used for keeping state across successive
583 # calls to the closure. The iterator simply returns when $n > max.
584 sub input_iterator {
586 my ($n, $max, $step) = @_;
587 return sub {
589 return if $n > $max;
590 my $current = $n;
592 $n += $step;
593 return $current;
595 };
596 }
597 # Run user_func in parallel. Input data can be specified during
599 # the construction or as an argument to the process method.
600 my $mce = MCE->new(
602 # input_data => input_iterator(10, 30, 2),
604 chunk_size => 1, max_workers => 4,
605 user_func => sub {
607 my ($mce, $chunk_ref, $chunk_id) = @_;
608 MCE->print("$_: ", $_ * 2, "\n");
609 }
610 )->spawn;
612 $mce->process( input_iterator(10, 30, 2) );
614 -- Output Note that output order is not guaranteed
616 Take a look at iterator.pl for ordered output
617 10: 20
619 12: 24
620 16: 32
621 20: 40
622 14: 28
623 22: 44
624 18: 36
625 24: 48
626 26: 52
627 28: 56
628 30: 60
629 The following example queries the DB for the next 1000 rows. Notice the
631 use of fetchall_arrayref. The iterator function itself receives one
632 argument which is chunk_size (added in MCE 1.510) to determine how much
633 to return per iteration. The default is 1 for the Core API and MCE
634 Models.
635 use DBI;
637 use MCE;
638 sub db_iter {
640 my $dsn = "DBI:Oracle:host=db_server;port=db_port;sid=db_name";
642 my $dbh = DBI->connect($dsn, 'db_user', 'db_passwd') ||
644 die "Could not connect to database: $DBI::errstr";
645 my $sth = $dbh->prepare('select color, desc from table');
647 $sth->execute;
649 return sub {
651 my ($chunk_size) = @_;
652 if (my $aref = $sth->fetchall_arrayref(undef, $chunk_size)) {
654 return @{ $aref };
655 }
656 return;
658 };
659 }
660 # Let's enumerate column indexes for easy column retrieval.
662 my ($i_color, $i_desc) = (0 .. 1);
663 my $mce = MCE->new(
665 max_workers => 3, chunk_size => 1000,
666 input_data => db_iter(),
667 user_func => sub {
669 my ($mce, $chunk_ref, $chunk_id) = @_;
670 my $ret = '';
671 foreach my $row (@{ $chunk_ref }) {
673 $ret .= $row->[$i_color] .": ". $row->[$i_desc] ."\n";
674 }
675 MCE->print($ret);
677 }
678 );
679 $mce->run;
681 There are many modules on CPAN which return an iterator reference.
683 Showing one such example below. The demonstration ensures MCE workers
684 are spawned before obtaining the iterator. Note the worker_id value
685 (left column) in the output.
686 use Path::Iterator::Rule;
688 use MCE;
689 my $start_dir = shift
691 or die "Please specify a starting directory";
692 -d $start_dir
694 or die "Cannot open ($start_dir): No such file or directory";
695 my $mce = MCE->new(
697 max_workers => 'auto',
698 user_func => sub { MCE->say( MCE->wid . ": $_" ) }
699 )->spawn;
700 my $rule = Path::Iterator::Rule->new->file->name( qr/[.](pm)$/ );
702 my $iterator = $rule->iter(
704 $start_dir, { follow_symlinks => 0, depthfirst => 1 }
705 );
706 $mce->process( $iterator );
708 -- Output
710 8: lib/MCE/Core/Input/Generator.pm
712 5: lib/MCE/Core/Input/Handle.pm
713 6: lib/MCE/Core/Input/Iterator.pm
714 2: lib/MCE/Core/Input/Request.pm
715 3: lib/MCE/Core/Manager.pm
716 4: lib/MCE/Core/Input/Sequence.pm
717 7: lib/MCE/Core/Validation.pm
718 1: lib/MCE/Core/Worker.pm
719 8: lib/MCE/Flow.pm
720 5: lib/MCE/Grep.pm
721 6: lib/MCE/Loop.pm
722 2: lib/MCE/Map.pm
723 3: lib/MCE/Queue.pm
724 4: lib/MCE/Signal.pm
725 7: lib/MCE/Stream.pm
726 1: lib/MCE/Subs.pm
727 8: lib/MCE/Util.pm
728 5: lib/MCE.pm
729 Although MCE supports arrays, extra measures are needed to use a "lazy"
731 array as input data. The reason for this is that MCE needs the size of
732 the array before processing which may be unknown for lazy arrays.
733 Therefore, closures provides an excellent mechanism for this.
734 The code block belonging to the lazy array must return undef after
736 exhausting its input data. Otherwise, the process will never end.
737 use Tie::Array::Lazy;
739 use MCE;
740 tie my @a, 'Tie::Array::Lazy', [], sub {
742 my $i = $_[0]->index;
743 return ($i < 10) ? $i : undef;
745 };
746 sub make_iterator {
748 my $i = 0; my $a_ref = shift;
749 return sub {
751 return $a_ref->[$i++];
752 };
753 }
754 my $mce = MCE->new(
756 max_workers => 4, input_data => make_iterator(\@a),
757 user_func => sub {
759 my ($mce, $chunk_ref, $chunk_id) = @_;
760 MCE->say($_);
761 }
762 )->run;
764 -- Output
766 0
768 1
769 2
770 3
771 4
772 6
773 7
774 8
775 5
776 9
777 The following demonstrates how to retrieve a chunk from the lazy array
779 per each successive call. Here, undef is sent by the iterator block
780 when $i is greater than $max. Iterators may optionally use chunk_size
781 to determine how much to return per iteration.
782 use Tie::Array::Lazy;
784 use MCE;
785 tie my @a, 'Tie::Array::Lazy', [], sub {
787 $_[0]->index;
788 };
789 sub make_iterator {
791 my $j = 0; my ($a_ref, $max) = @_;
792 return sub {
794 my ($chunk_size) = @_;
795 my $i = $j; $j += $chunk_size;
796 return if $i > $max;
798 return $j <= $max ? @$a_ref[$i .. $j - 1] : @$a_ref[$i .. $max];
799 };
800 }
801 my $mce = MCE->new(
803 chunk_size => 15, max_workers => 4,
804 input_data => make_iterator(\@a, 100),
805 user_func => sub {
807 my ($mce, $chunk_ref, $chunk_id) = @_;
808 MCE->say("$chunk_id: " . join(' ', @{ $chunk_ref }));
809 }
810 )->run;
812 -- Output
814 1: 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14
816 2: 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29
817 3: 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44
818 4: 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59
819 5: 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74
820 6: 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89
821 7: 90 91 92 93 94 95 96 97 98 99 100
822 SYNTAX for SEQUENCE
824 The 1.3 release and above allows workers to loop through a sequence of
825 numbers computed mathematically without the overhead of an array. The
826 sequence can be specified separately per each user_task entry unlike
827 input_data which is applicable to the first task only.
828 See the seq_demo.pl example, included with this distribution, on
830 applying sequences with the user_tasks option.
831 Sequence can be defined using an array or a hash reference.
833 use MCE;
835 my $mce = MCE->new(
837 max_workers => 3,
838 # sequence => [ 10, 19, 0.7, "%4.1f" ], # up to 4 options
840 sequence => {
842 begin => 10, end => 19, step => 0.7, format => "%4.1f"
843 },
844 user_func => sub {
846 my ($mce, $n, $chunk_id) = @_;
847 print $n, " from ", MCE->wid, " id ", $chunk_id, "\n";
848 }
849 );
850 $mce->run;
852 -- Output (sorted afterwards, notice wid and chunk_id in output)
854 10.0 from 1 id 1
856 10.7 from 2 id 2
857 11.4 from 3 id 3
858 12.1 from 1 id 4
859 12.8 from 2 id 5
860 13.5 from 3 id 6
861 14.2 from 1 id 7
862 14.9 from 2 id 8
863 15.6 from 3 id 9
864 16.3 from 1 id 10
865 17.0 from 2 id 11
866 17.7 from 3 id 12
867 18.4 from 1 id 13
868 The 1.5 release includes a new option (bounds_only). This option tells
870 the sequence engine to compute 'begin' and 'end' items only, for the
871 chunk, and not the items in between (hence boundaries only). This
872 option applies to sequence only and has no effect when chunk_size
873 equals 1.
874 The time to run is 0.006s below. This becomes 0.827s without the
876 bounds_only option due to computing all items in between, thus creating
877 a very large array. Basically, specify bounds_only => 1 when boundaries
878 is all you need for looping inside the block; e.g. Monte Carlo
879 simulations.
880 Time was measured using 1 worker to emphasize the difference.
882 use MCE;
884 my $mce = MCE->new(
886 max_workers => 1, chunk_size => 1_250_000,
887 sequence => { begin => 1, end => 10_000_000 },
889 bounds_only => 1,
890 # For sequence, the input scalar $_ points to $chunk_ref
892 # when chunk_size > 1, otherwise $chunk_ref->[0].
893 #
894 # user_func => sub {
895 # my $begin = $_->[0]; my $end = $_->[-1];
896 #
897 # for ($begin .. $end) {
898 # ...
899 # }
900 # },
901 user_func => sub {
903 my ($mce, $chunk_ref, $chunk_id) = @_;
904 # $chunk_ref contains 2 items, not 1_250_000
905 my $begin = $chunk_ref->[ 0];
907 my $end = $chunk_ref->[-1]; # or $chunk_ref->[1]
908 MCE->printf("%7d .. %8d\n", $begin, $end);
910 }
911 );
912 $mce->run;
914 -- Output
916 1 .. 1250000
918 1250001 .. 2500000
919 2500001 .. 3750000
920 3750001 .. 5000000
921 5000001 .. 6250000
922 6250001 .. 7500000
923 7500001 .. 8750000
924 8750001 .. 10000000
925 SYNTAX for MAX_RETRIES
927 The max_retries option, added in 1.7, allows MCE to retry a failed
928 chunk from a worker dying while processing input data or a sequence of
929 numbers.
930 When max_retries is set, MCE configures the on_post_exit option
932 automatically using the following code before running. Specify
933 on_post_exit explicitly for any further tailoring. The restart_worker
934 line is necessary, obviously.
935 on_post_exit => sub {
937 my ( $mce, $e, $retry_cnt ) = @_;
938 if ( $e->{id} ) {
940 my $cnt = $retry_cnt + 1;
941 my $msg = "Error: chunk $e->{id} failed";
942 if ( defined $mce->{init_relay} ) {
944 print {*STDERR} "$msg, retrying chunk attempt # $cnt\n"
945 if ( $retry_cnt < $mce->{max_retries} );
946 }
947 else {
948 ( $retry_cnt < $mce->{max_retries} )
949 ? print {*STDERR} "$msg, retrying chunk attempt # $cnt\n"
950 : print {*STDERR} "$msg\n";
951 }
952 $mce->restart_worker;
954 }
955 }
956 We let MCE handle on_post_exit automatically below, which is
958 essentially the same code shown above. For max_retries to work, the
959 worker must die, abnormally included, or call MCE->exit. Notice that we
960 pass the chunk_id value for the 3rd argument to MCE->exit (defaults to
961 chunk_id if omitted since MCE 1.844).
962 # max_retries demonstration
964 use strict;
966 use warnings;
967 use MCE;
969 sub user_func {
971 my ( $mce, $chunk_ref, $chunk_id ) = @_;
972 # die "Died : chunk_id = 3\n" if $chunk_id == 3;
974 MCE->exit(1, undef, $chunk_id) if $chunk_id == 3;
975 print "$chunk_id\n";
977 }
978 my $mce = MCE->new(
980 max_workers => 1,
981 max_retries => 2,
982 user_func => \&user_func,
983 )->spawn;
984 my $input_data = [ 0..7 ];
986 $mce->process( { chunk_size => 1 }, $input_data );
988 $mce->shutdown;
989 -- Output
991 1
993 2
994 Error: chunk 3 failed, retrying chunk attempt # 1
995 Error: chunk 3 failed, retrying chunk attempt # 2
996 Error: chunk 3 failed
997 4
998 5
999 6
1000 7
1001 8
1002 Orderly output with max_retries is possible since MCE 1.844. Below,
1004 chunk 3 succeeds whereas chunk 5 fails due to exceeding the number of
1005 retries. Be sure to call MCE::relay inside "user_func" and near the end
1006 of the block.
1007 # max_retries demonstration with init_relay
1009 use strict;
1011 use warnings;
1012 use MCE;
1014 use MCE::Shared;
1015 tie my $retries1, 'MCE::Shared', 0;
1017 tie my $retries2, 'MCE::Shared', 0;
1018 MCE->new(
1020 max_workers => 4,
1021 input_data => [ 1..7 ],
1022 chunk_size => 1,
1023 max_retries => 2,
1025 init_relay => 0,
1026 user_func => sub {
1028 if ( MCE->chunk_id == 3 ) {
1029 MCE->exit if ++$retries1 <= 2;
1030 }
1031 if ( MCE->chunk_id == 5 ) {
1032 MCE->exit if ++$retries2 <= 3;
1033 }
1034 MCE::relay {
1035 $_ += 1;
1036 print MCE->chunk_id, "\n";
1037 };
1038 }
1039 )->run;
1040 print "final: ", MCE::relay_final(), "\n";
1042 -- Output
1044 1
1046 2
1047 Error: chunk 3 failed, retrying chunk attempt # 1
1048 Error: chunk 5 failed, retrying chunk attempt # 1
1049 Error: chunk 3 failed, retrying chunk attempt # 2
1050 Error: chunk 5 failed, retrying chunk attempt # 2
1051 3
1052 4
1053 Error: chunk 5 failed
1054 6
1055 7
1056 final: 6
1057 SYNTAX for USER_BEGIN and USER_END
1059 The user_begin and user_end options, if specified, behave similarly to
1060 awk 'BEGIN { begin } { func } { func } ... END { end }'. These are
1061 called once per worker during each run.
1062 MCE 1.510 passes 2 additional parameters ($task_id and $task_name).
1064 sub user_begin { # Called once at the beginning
1066 my ($mce, $task_id, $task_name) = @_;
1067 $mce->{wk_total_rows} = 0;
1068 }
1069 sub user_func { # Called while processing
1071 my $mce = shift;
1072 $mce->{wk_total_rows} += 1;
1073 }
1074 sub user_end { # Called once at the end
1076 my ($mce, $task_id, $task_name) = @_;
1077 printf "## %d: Processed %d rows\n",
1078 MCE->wid, $mce->{wk_total_rows};
1079 }
1080 my $mce = MCE->new(
1082 user_begin => \&user_begin,
1083 user_func => \&user_func,
1084 user_end => \&user_end
1085 );
1086 $mce->run;
1088 SYNTAX for USER_FUNC with USE_SLURPIO => 0
1090 When processing input data, MCE can pass an array of rows or a slurped
1091 chunk. Below, a reference to an array containing the chunk data is
1092 processed.
1093 e.g. $chunk_ref = [ record1, record2, record3, ... ]
1095 sub user_func {
1097 my ($mce, $chunk_ref, $chunk_id) = @_;
1099 foreach my $row ( @{ $chunk_ref } ) {
1101 $mce->{wk_total_rows} += 1;
1102 print $row;
1103 }
1104 }
1105 my $mce = MCE->new(
1107 chunk_size => 100,
1108 input_data => "/path/to/file",
1109 user_func => \&user_func,
1110 use_slurpio => 0
1111 );
1112 $mce->run;
1114 SYNTAX for USER_FUNC with USE_SLURPIO => 1
1116 Here, a reference to a scalar containing the raw chunk data is
1117 processed.
1118 sub user_func {
1120 my ($mce, $chunk_ref, $chunk_id) = @_;
1122 my $count = () = $$chunk_ref =~ /abc/;
1124 }
1125 my $mce = MCE->new(
1127 chunk_size => 16000,
1128 input_data => "/path/to/file",
1129 user_func => \&user_func,
1130 use_slurpio => 1
1131 );
1132 $mce->run;
1134 SYNTAX for USER_ERROR and USER_OUTPUT
1136 Output from MCE->sendto('STDERR/STDOUT', ...), MCE->printf, MCE->print,
1137 and MCE->say can be intercepted by specifying the user_error and
1138 user_output options. MCE on receiving output will forward to user_error
1139 or user_output in a serialized fashion.
1140 Handy when wanting to filter, modify, and/or direct the output
1142 elsewhere.
1143 sub user_error { # Redirect STDERR to STDOUT
1145 my $error = shift;
1146 print {*STDOUT} $error;
1147 }
1148 sub user_output { # Redirect STDOUT to STDERR
1150 my $output = shift;
1151 print {*STDERR} $output;
1152 }
1153 sub user_func {
1155 my ($mce, $chunk_ref, $chunk_id) = @_;
1156 my $count = 0;
1157 foreach my $row ( @{ $chunk_ref } ) {
1159 MCE->print($row);
1160 $count += 1;
1161 }
1162 MCE->print(\*STDERR, "$chunk_id: processed $count rows\n");
1164 }
1165 my $mce = MCE->new(
1167 chunk_size => 1000,
1168 input_data => "/path/to/file",
1169 user_error => \&user_error,
1170 user_output => \&user_output,
1171 user_func => \&user_func
1172 );
1173 $mce->run;
1175 SYNTAX for USER_TASKS and TASK_END
1177 This option takes an array of tasks. Each task allows up to 17 options.
1178 The init_relay, input_data, RS, and use_slurpio options may be defined
1179 inside the first task or at the top level, otherwise ignored under
1180 other sub-tasks.
1181 max_workers, chunk_size, input_data, interval, sequence,
1183 bounds_only, user_args, user_begin, user_end, user_func,
1184 gather, task_end, task_name, use_slurpio, use_threads,
1185 init_relay, RS
1186 Sequence and chunk_size were added in 1.3. User_args was introduced in
1188 1.4. Name and input_data are new options allowed in 1.5. In addition,
1189 one can specify task_end at the top level. Task_end also receives 2
1190 additional arguments $task_id and $task_name (shown below).
1191 Options not specified here will default to the same option specified at
1193 the top level. The task_end option is called by the manager process
1194 when all workers for that sub-task have completed processing.
1195 Forking and threading can be intermixed among tasks unless running
1197 Cygwin. The run method will continue running until all workers have
1198 completed processing.
1199 use threads;
1201 use threads::shared;
1202 use MCE;
1204 sub parallel_task1 { sleep 2; }
1206 sub parallel_task2 { sleep 1; }
1207 my $mce = MCE->new(
1209 task_end => sub {
1211 my ($mce, $task_id, $task_name) = @_;
1212 print "Task [$task_id -- $task_name] completed processing\n";
1213 },
1214 user_tasks => [{
1216 task_name => 'foo',
1217 max_workers => 2,
1218 user_func => \¶llel_task1,
1219 use_threads => 0 # Not using threads
1220 },{
1222 task_name => 'bar',
1223 max_workers => 4,
1224 user_func => \¶llel_task2,
1225 use_threads => 1 # Yes, threads
1226 }]
1228 );
1229 $mce->run;
1231 -- Output
1233 Task [1 -- bar] completed processing
1235 Task [0 -- foo] completed processing
1236
1238 Beginning with MCE 1.5, the input scalar $_ is localized prior to
1239 calling user_func for input_data and sequence of numbers. The following
1240 applies.
1241 use_slurpio => 1
1243 $_ is a reference to the buffer e.g. $_ = \$_buffer;
1244 $_ is a reference regardless of whether chunk_size is 1 or greater
1245 user_func => sub {
1247 # my ($mce, $chunk_ref, $chunk_id) = @_;
1248 print ${ $_ }; # $_ is same as $chunk_ref
1249 }
1250 chunk_size is greater than 1, use_slurpio => 0
1252 $_ is a reference to an array. $_ = \@_records; $_ = \@_seq_n;
1253 $_ is same as $chunk_ref or $_[CHUNK]
1254 user_func => sub {
1256 # my ($mce, $chunk_ref, $chunk_id) = @_;
1257 for my $row ( @{ $_ } ) {
1258 print $row, "\n";
1259 }
1260 }
1261 use MCE const => 1;
1263 user_func => sub {
1265 # my ($mce, $chunk_ref, $chunk_id) = @_;
1266 for my $row ( @{ $_[CHUNK] } ) {
1267 print $row, "\n";
1268 }
1269 }
1270 chunk_size equals 1, use_slurpio => 0
1272 $_ contains the actual value. $_ = $_buffer; $_ = $seq_n;
1273 # Note that $_ and $chunk_ref are not the same below.
1275 # $chunk_ref is a reference to an array.
1276 user_func => sub {
1278 # my ($mce, $chunk_ref, $chunk_id) = @_;
1279 print $_, "\n; # Same as $chunk_ref->[0];
1280 }
1281 $mce->foreach("/path/to/file", sub {
1283 # my ($mce, $chunk_ref, $chunk_id) = @_;
1284 print $_; # Same as $chunk_ref->[0];
1285 });
1286 # However, that is not the case for the forseq method.
1288 # Both $_ and $n_seq are the same when chunk_size => 1.
1289 $mce->forseq([ 1, 9 ], sub {
1291 # my ($mce, $n_seq, $chunk_id) = @_;
1292 print $_, "\n"; # Same as $n_seq
1293 });
1294 Sequence can also be specified using an array reference. The below
1296 is the same as the example afterwards.
1297 $mce->forseq( { begin => 10, end => 40, step => 2 }, ... );
1299 The code block receives an array containing the next 5 sequences.
1301 Chunk 1 (chunk_id 1) contains 10,12,14,16,18. $n_seq is a reference
1302 to an array, same as $_, due to chunk_size being greater than 1.
1303 $mce->forseq( [ 10, 40000, 2 ], { chunk_size => 5 }, sub {
1305 # my ($mce, $n_seq, $chunk_id) = @_;
1306 my @result;
1307 for my $n ( @{ $_ } ) {
1308 ... do work, append to result for 5
1309 }
1310 ... do something with result afterwards
1311 });
1312
1314 The methods listed below are callable by the main process and workers.
1315 MCE->abort ( void )
1317 $mce->abort ( void )
1318 The 'abort' method is applicable when processing input_data only. This
1319 causes all workers to abort after processing the current chunk.
1320 Workers write the next offset position to the queue socket for the next
1322 available worker. In essence, the 'abort' method writes the last offset
1323 position. Workers, on requesting the next offset position, will think
1324 the end of input_data has been reached and leave the chunking loop.
1325 MCE->abort;
1327 $mce->abort;
1328 MCE->chunk_id ( void )
1330 $mce->chunk_id ( void )
1331 Returns the chunk_id for the current chunk. The value starts at 1.
1332 Chunking applies to input_data or sequence. The value is 0 for the
1333 manager process.
1334 my $chunk_id = MCE->chunk_id;
1336 my $chunk_id = $mce->chunk_id;
1337 MCE->chunk_size ( void )
1339 $mce->chunk_size ( void )
1340 Getter method for chunk_size used by MCE.
1341 my $chunk_size = MCE->chunk_size;
1343 my $chunk_size = $mce->chunk_size;
1344 MCE->do ( 'callback_func' [, $arg1, ... ] )
1346 $mce->do ( 'callback_func' [, $arg1, ... ] )
1347 MCE serializes data transfers from a worker process via helper
1348 functions do & sendto to the manager process. The callback function can
1349 optionally return a reply. Support for calling by the manager process
1350 was enabled in MCE 1.839.
1351 [ $reply = ] MCE->do('callback' [, $arg1, ... ]);
1353 Passing args to a callback function using references & scalar.
1355 sub callback {
1357 my ($array_ref, $hash_ref, $scalar_ref, $scalar) = @_;
1358 ...
1359 }
1360 MCE->do('main::callback', \@a, \%h, \$s, 'foo');
1362 MCE->do('callback', \@a, \%h, \$s, 'foo');
1363 MCE knows if wanting a void, list, hash, or a scalar return value.
1365 MCE->do('callback' [, $arg1, ... ]);
1367 my @array = MCE->do('callback' [, $arg1, ... ]);
1369 my %hash = MCE->do('callback' [, $arg1, ... ]);
1370 my $scalar = MCE->do('callback' [, $arg1, ... ]);
1371 MCE->freeze ( $object_ref )
1373 $mce->freeze ( $object_ref )
1374 Calls the internal freeze method to serialize an object. The default
1375 serialization routines are handled by Sereal if available or Storable.
1376 my $frozen = MCE->freeze([ 0, 2, 4 ]);
1378 my $frozen = $mce->freeze([ 0, 2, 4 ]);
1379 MCE->max_retries ( void )
1381 $mce->max_retries ( void )
1382 Getter method for max_retries used by MCE.
1383 my $max_retries = MCE->max_retries;
1385 my $max_retries = $mce->max_retries;
1386 MCE->max_workers ( void )
1388 $mce->max_workers ( void )
1389 Getter method for max_workers used by MCE.
1390 my $max_workers = MCE->max_workers;
1392 my $max_workers = $mce->max_workers;
1393 MCE->pid ( void )
1395 $mce->pid ( void )
1396 Returns the Process ID. Threads have thread ID attached to the value.
1397 my $pid = MCE->pid; # 16180 (pid) ; 16180.2 (pid.tid)
1399 my $pid = $mce->pid;
1400 MCE->printf ( $format, $list [, ... ] )
1402 MCE->print ( $list [, ... ] )
1403 MCE->say ( $list [, ... ] )
1404 $mce->printf ( $format, $list [, ... ] )
1405 $mce->print ( $list [, ... ] )
1406 $mce->say ( $list [, ... ] )
1407 Use the printf, print, and say methods when wanting to serialize output
1408 among workers and the manager process. These are sugar syntax for the
1409 sendto method. These behave similar to the native subroutines in Perl
1410 with the exception that barewords must be passed as a reference and
1411 require the comma after it including file handles.
1412 Say is like print, but implicitly appends a newline.
1414 MCE->printf(\*STDOUT, "%s: %d\n", $name, $age);
1416 MCE->printf($fh, "%s: %d\n", $name, $age);
1417 MCE->printf("%s: %d\n", $name, $age);
1418 MCE->print(\*STDERR, "$error_msg\n");
1420 MCE->print($fh, $log_msg."\n");
1421 MCE->print("$output_msg\n");
1422 MCE->say(\*STDERR, $error_msg);
1424 MCE->say($fh, $log_msg);
1425 MCE->say($output_msg);
1426 Caveat: Use the following syntax when passing a reference not a glob or
1428 file handle. Otherwise, MCE will error indicating the first argument is
1429 not a glob reference.
1430 MCE->print(\*STDOUT, \@array, "\n");
1432 MCE->print("", \@array, "\n"); # ok
1433 Sending to "IO::All" { File, Pipe, STDIO } is supported since MCE
1435 1.845.
1436 use IO::All;
1438 my $out = io->stdout;
1440 my $err = io->stderr;
1441 MCE->printf($out, "%s\n", "sent to stdout");
1443 MCE->printf($err, "%s\n", "sent to stderr");
1444 MCE->print($out, "sent to stdout\n");
1446 MCE->print($err, "sent to stderr\n");
1447 MCE->say($out, "sent to stdout");
1449 MCE->say($err, "sent to stderr");
1450 MCE->sess_dir ( void )
1452 $mce->sess_dir ( void )
1453 Returns the session directory used by the MCE instance. This is defined
1454 during spawning and removed during shutdown.
1455 my $sess_dir = MCE->sess_dir;
1457 my $sess_dir = $mce->sess_dir;
1458 MCE->task_id ( void )
1460 $mce->task_id ( void )
1461 Returns the task ID. This applies to the user_tasks option (starts at
1462 0).
1463 my $task_id = MCE->task_id;
1465 my $task_id = $mce->task_id;
1466 MCE->task_name ( void )
1468 $mce->task_name ( void )
1469 Returns the task_name value specified via the task_name option when
1470 configuring MCE.
1471 my $task_name = MCE->task_name;
1473 my $task_name = $mce->task_name;
1474 MCE->task_wid ( void )
1476 $mce->task_wid ( void )
1477 Returns the task worker ID (applies to user_tasks). The value starts at
1478 1 per each task configured within user_tasks. The value is 0 for the
1479 manager process.
1480 my $task_wid = MCE->task_wid;
1482 my $task_wid = $mce->task_wid;
1483 MCE->thaw ( $frozen )
1485 $mce->thaw ( $frozen )
1486 Calls the internal thaw method to un-serialize the frozen object.
1487 my $object_ref = MCE->thaw($frozen);
1489 my $object_ref = $mce->thaw($frozen);
1490 MCE->tmp_dir ( void )
1492 $mce->tmp_dir ( void )
1493 Returns the temporary directory used by MCE.
1494 my $tmp_dir = MCE->tmp_dir;
1496 my $tmp_dir = $mce->tmp_dir;
1497 MCE->user_args ( void )
1499 $mce->user_args ( void )
1500 Returns the arguments specified via the user_args option.
1501 my ($arg1, $arg2, $arg3) = MCE->user_args;
1503 my ($arg1, $arg2, $arg3) = $mce->user_args;
1504 MCE->wid ( void )
1506 $mce->wid ( void )
1507 Returns the MCE worker ID. Starts at 1 per each MCE instance. The value
1508 is 0 for the manager process.
1509 my $wid = MCE->wid;
1511 my $wid = $mce->wid;
1512
1514 Methods listed below are callable by the main process only.
1515 MCE->forchunk ( $input_data [, { options } ], sub { ... } )
1517 MCE->foreach ( $input_data [, { options } ], sub { ... } )
1518 MCE->forseq ( $sequence_spec [, { options } ], sub { ... } )
1519 $mce->forchunk ( $input_data [, { options } ], sub { ... } )
1520 $mce->foreach ( $input_data [, { options } ], sub { ... } )
1521 $mce->forseq ( $sequence_spec [, { options } ], sub { ... } )
1522 Forchunk, foreach, and forseq are sugar methods and described in
1523 MCE::Candy. Stubs exist in MCE which load MCE::Candy automatically.
1524 MCE->process ( $input_data [, { options } ] )
1526 $mce->process ( $input_data [, { options } ] )
1527 The process method will spawn workers automatically if not already
1528 spawned. It will set input_data => $input_data. It calls run(0) to not
1529 auto-shutdown workers. Specifying options is optional.
1530 Allowable options { key => value, ... } are:
1532 chunk_size input_data job_delay spawn_delay submit_delay
1534 flush_file flush_stderr flush_stdout stderr_file stdout_file
1535 on_post_exit on_post_run sequence user_args user_begin user_end
1536 user_func user_error user_output use_slurpio RS
1537 Options remain persistent going forward unless changed. Setting
1539 user_begin, user_end, or user_func will cause already spawned workers
1540 to shut down and re-spawn automatically. Therefore, define these during
1541 instantiation.
1542 The below will cause workers to re-spawn after running.
1544 my $mce = MCE->new( max_workers => 'auto' );
1546 $mce->process( {
1548 user_begin => sub { # connect to DB },
1549 user_func => sub { # process each row },
1550 user_end => sub { # close handle to DB },
1551 }, \@input_data );
1552 $mce->process( {
1554 user_begin => sub { # connect to DB },
1555 user_func => sub { # process each file },
1556 user_end => sub { # close handle to DB },
1557 }, "/list/of/files" );
1558 Do the following if wanting workers to persist between jobs.
1560 use MCE max_workers => 'auto';
1562 my $mce = MCE->new(
1564 user_begin => sub { # connect to DB },
1565 user_func => sub { # process each chunk or row or host },
1566 user_end => sub { # close handle to DB },
1567 );
1568 $mce->spawn; # Spawn early if desired
1570 $mce->process("/one/very_big_file/_mce_/will_chunk_in_parallel");
1572 $mce->process(\@array_of_files_to_grep);
1573 $mce->process("/path/to/host/list");
1574 $mce->process($array_ref);
1576 $mce->process($array_ref, { stdout_file => $output_file });
1577 # This was not allowed before. Fixed in 1.415.
1579 $mce->process({ sequence => { begin => 10, end => 90, step 2 } });
1580 $mce->process({ sequence => [ 10, 90, 2 ] });
1581 $mce->shutdown;
1583 MCE->relay_final ( void )
1585 $mce->relay_final ( void )
1586 The relay methods are described in MCE::Relay. Relay capabilities are
1587 enabled by specifying the "init_relay" MCE option.
1588 MCE->restart_worker ( void )
1590 $mce->restart_worker ( void )
1591 One can restart a worker who has died or exited. The job never ends
1592 below due to restarting each time. Recommended is to call MCE->exit or
1593 $mce->exit instead of the native exit function for better handling,
1594 especially under the Windows environment.
1595 The $e->{wid} argument is no longer necessary starting with the 1.5
1597 release.
1598 Press [ctrl-c] to terminate the script.
1600 my $mce = MCE->new(
1602 on_post_exit => sub {
1604 my ($mce, $e) = @_;
1605 print "$e->{wid}: $e->{pid}: status $e->{status}: $e->{msg}";
1606 # $mce->restart_worker($e->{wid}); # MCE-1.415 and below
1607 $mce->restart_worker; # MCE-1.500 and above
1608 },
1609 user_begin => sub {
1611 my ($mce, $task_id, $task_name) = @_;
1612 # Not interested in die messages going to STDERR,
1613 # because the die handler calls MCE->exit(255, $_[0]).
1614 close STDERR;
1615 },
1616 user_tasks => [{
1618 max_workers => 5,
1619 user_func => sub {
1620 my ($mce) = @_; sleep MCE->wid;
1621 MCE->exit(3, "exited from " . MCE->wid . "\n");
1622 }
1623 },{
1624 max_workers => 4,
1625 user_func => sub {
1626 my ($mce) = @_; sleep MCE->wid;
1627 die("died from " . MCE->wid . "\n");
1628 }
1629 }]
1630 );
1631 $mce->run;
1633 -- Output
1635 1: PID_85388: status 3: exited from 1
1637 2: PID_85389: status 3: exited from 2
1638 1: PID_85397: status 3: exited from 1
1639 3: PID_85390: status 3: exited from 3
1640 1: PID_85399: status 3: exited from 1
1641 4: PID_85391: status 3: exited from 4
1642 2: PID_85398: status 3: exited from 2
1643 1: PID_85401: status 3: exited from 1
1644 5: PID_85392: status 3: exited from 5
1645 1: PID_85404: status 3: exited from 1
1646 6: PID_85393: status 255: died from 6
1647 3: PID_85400: status 3: exited from 3
1648 2: PID_85403: status 3: exited from 2
1649 1: PID_85406: status 3: exited from 1
1650 7: PID_85394: status 255: died from 7
1651 1: PID_85410: status 3: exited from 1
1652 8: PID_85395: status 255: died from 8
1653 4: PID_85402: status 3: exited from 4
1654 2: PID_85409: status 3: exited from 2
1655 1: PID_85412: status 3: exited from 1
1656 9: PID_85396: status 255: died from 9
1657 3: PID_85408: status 3: exited from 3
1658 1: PID_85416: status 3: exited from 1
1659 ...
1661 MCE->run ( [ $auto_shutdown [, { options } ] ] )
1663 $mce->run ( [ $auto_shutdown [, { options } ] ] )
1664 The run method, by default, spawns workers, processes once, and shuts
1665 down afterwards. Specify 0 for $auto_shutdown when wanting workers to
1666 persist after running (default 1).
1667 Specifying options is optional. Valid options are the same as for the
1669 process method.
1670 my $mce = MCE->new( ... );
1672 # Disables auto-shutdown
1674 $mce->run(0);
1675 MCE->send ( $data_ref )
1677 $mce->send ( $data_ref )
1678 The 'send' method is useful when wanting to spawn workers early to
1679 minimize memory consumption and afterwards send data individually to
1680 each worker. One cannot send more than the total workers spawned.
1681 Workers store the received data as $mce->{user_data}.
1682 The data which can be sent is restricted to an ARRAY, HASH, or PDL
1684 reference. Workers begin processing immediately after receiving data.
1685 Workers set $mce->{user_data} to undef after processing. One cannot
1686 specify input_data, sequence, or user_tasks when using the "send"
1687 method.
1688 Passing any options e.g. run(0, { options }) is ignored due to workers
1690 running immediately after receiving user data. There is no guarantee to
1691 which worker will receive data first. It depends on which worker is
1692 available awaiting data.
1693 use MCE;
1695 my $mce = MCE->new(
1697 max_workers => 5,
1698 user_func => sub {
1700 my ($mce) = @_;
1701 my $data = $mce->{user_data};
1702 my $first_name = $data->{first_name};
1703 print MCE->wid, ": Hello from $first_name\n";
1704 }
1705 );
1706 $mce->spawn; # Optional, send will spawn if necessary.
1708 $mce->send( { first_name => "Theresa" } );
1710 $mce->send( { first_name => "Francis" } );
1711 $mce->send( { first_name => "Padre" } );
1712 $mce->send( { first_name => "Anthony" } );
1713 $mce->run; # Wait for workers to complete processing.
1715 -- Output
1717 2: Hello from Theresa
1719 5: Hello from Anthony
1720 3: Hello from Francis
1721 4: Hello from Padre
1722 MCE->shutdown ( void )
1724 $mce->shutdown ( void )
1725 The run method will automatically spawn workers, run once, and shutdown
1726 workers automatically. Workers persist after running below. Shutdown
1727 may be called as needed or prior to exiting.
1728 my $mce = MCE->new( ... );
1730 $mce->spawn;
1732 $mce->process(\@input_data_1); # Processing multiple arrays
1734 $mce->process(\@input_data_2);
1735 $mce->process(\@input_data_n);
1736 $mce->shutdown;
1738 $mce->process('input_file_1'); # Processing multiple files
1740 $mce->process('input_file_2');
1741 $mce->process('input_file_n');
1742 $mce->shutdown;
1744 MCE->spawn ( void )
1746 $mce->spawn ( void )
1747 Workers are normally spawned automatically. The spawn method allows one
1748 to spawn workers early if so desired.
1749 my $mce = MCE->new( ... );
1751 $mce->spawn;
1753 MCE->status ( void )
1755 $mce->status ( void )
1756 The greatest exit status is saved among workers while running. Look at
1757 the on_post_exit or on_post_run options for callback support.
1758 my $mce = MCE->new( ... );
1760 $mce->run;
1762 my $exit_status = $mce->status;
1764
1766 Methods listed below are callable by workers only.
1767 MCE->exit ( [ $status [, $message [, $id ] ] ] )
1769 $mce->exit ( [ $status [, $message [, $id ] ] ] )
1770 A worker exits from MCE entirely. $id (optional) can be used for
1771 passing the primary key or a string along with the message. Look at the
1772 on_post_exit or on_post_run options for callback support.
1773 MCE->exit; # default 0
1775 MCE->exit(1);
1776 MCE->exit(2, 'chunk failed', $chunk_id);
1777 MCE->exit(0, 'msg_foo', 'id_1000');
1778 MCE->gather ( $arg1, [, $arg2, ... ] )
1780 $mce->gather ( $arg1, [, $arg2, ... ] )
1781 A worker can submit data to the location specified via the gather
1782 option by calling this method. See MCE::Flow and MCE::Loop for
1783 additional use-case.
1784 use MCE;
1786 my @hosts = qw(
1788 hosta hostb hostc hostd hoste
1789 );
1790 my $mce = MCE->new(
1792 chunk_size => 1, max_workers => 3,
1793 user_func => sub {
1795 # my ($mce, $chunk_ref, $chunk_id) = @_;
1796 my ($output, $error, $status); my $host = $_;
1797 # Do something with $host;
1799 $output = "Worker ". MCE->wid .": Hello from $host";
1800 if (MCE->chunk_id % 3 == 0) {
1802 # Simulating an error condition
1803 local $? = 1; $status = $?;
1804 $error = "Error from $host"
1805 }
1806 else {
1807 $status = 0;
1808 }
1809 # Ensure unique keys (key, value) when gathering to a
1811 # hash.
1812 MCE->gather("$host.out", $output, "$host.sta", $status);
1813 MCE->gather("$host.err", $error) if (defined $error);
1814 }
1815 );
1816 my %h; $mce->process(\@hosts, { gather => \%h });
1818 foreach my $host (@hosts) {
1820 print $h{"$host.out"}, "\n";
1821 print $h{"$host.err"}, "\n" if (exists $h{"$host.err"});
1822 print "Exit status: ", $h{"$host.sta"}, "\n\n";
1823 }
1824 -- Output
1826 Worker 2: Hello from hosta
1828 Exit status: 0
1829 Worker 1: Hello from hostb
1831 Exit status: 0
1832 Worker 3: Hello from hostc
1834 Error from hostc
1835 Exit status: 1
1836 Worker 2: Hello from hostd
1838 Exit status: 0
1839 Worker 1: Hello from hoste
1841 Exit status: 0
1842 MCE->last ( void )
1844 $mce->last ( void )
1845 Worker leaves the chunking loop or user_func block immediately.
1846 Callable from inside foreach, forchunk, forseq, and user_func.
1847 use MCE;
1849 my $mce = MCE->new(
1851 max_workers => 5
1852 );
1853 my @list = (1 .. 80);
1855 $mce->forchunk(\@list, { chunk_size => 2 }, sub {
1857 my ($mce, $chunk_ref, $chunk_id) = @_;
1859 MCE->last if ($chunk_id > 4);
1860 my @output = ();
1862 foreach my $rec ( @{ $chunk_ref } ) {
1864 push @output, $rec, "\n";
1865 }
1866 MCE->print(@output);
1868 });
1869 -- Output (each chunk above consists of 2 elements)
1871 3
1873 4
1874 1
1875 2
1876 7
1877 8
1878 5
1879 6
1880 MCE->next ( void )
1882 $mce->next ( void )
1883 Worker starts the next iteration of the chunking loop. Callable from
1884 inside foreach, forchunk, forseq, and user_func.
1885 use MCE;
1887 my $mce = MCE->new(
1889 max_workers => 5
1890 );
1891 my @list = (1 .. 80);
1893 $mce->forchunk(\@list, { chunk_size => 4 }, sub {
1895 my ($mce, $chunk_ref, $chunk_id) = @_;
1897 MCE->next if ($chunk_id < 20);
1898 my @output = ();
1900 foreach my $rec ( @{ $chunk_ref } ) {
1902 push @output, $rec, "\n";
1903 }
1904 MCE->print(@output);
1906 });
1907 -- Output (each chunk above consists of 4 elements)
1909 77
1911 78
1912 79
1913 80
1914 MCE::relay { code }
1916 MCE->relay ( sub { code } )
1917 MCE->relay_recv ( void )
1918 $mce->relay ( sub { code } )
1919 $mce->relay_recv ( void )
1920 The relay methods are described in MCE::Relay. Relay capabilities are
1921 enabled by specifying the "init_relay" MCE option.
1922 MCE->sendto ( $to, $arg1, ... )
1924 $mce->sendto ( $to, $arg1, ... )
1925 The sendto method is called by workers for serializing data to standard
1926 output, standard error, or end of file. The action is done by the
1927 manager process.
1928 Release 1.00x supported 1 data argument, not more.
1930 MCE->sendto('file', \@array, '/path/to/file');
1932 MCE->sendto('file', \$scalar, '/path/to/file');
1933 MCE->sendto('file', $scalar, '/path/to/file');
1934 MCE->sendto('STDERR', \@array);
1936 MCE->sendto('STDERR', \$scalar);
1937 MCE->sendto('STDERR', $scalar);
1938 MCE->sendto('STDOUT', \@array);
1940 MCE->sendto('STDOUT', \$scalar);
1941 MCE->sendto('STDOUT', $scalar);
1942 Release 1.100 added the ability to pass multiple arguments. Notice the
1944 syntax change for sending to a file. Passing a reference to an array is
1945 no longer necessary.
1946 MCE->sendto('file:/path/to/file', $arg1 [, $arg2, ... ]);
1948 MCE->sendto('STDERR', $arg1 [, $arg2, ... ]);
1949 MCE->sendto('STDOUT', $arg1 [, $arg2, ... ]);
1950 MCE->sendto('STDOUT', @a, "\n", %h, "\n", $s, "\n");
1952 To retain 1.00x compatibility, sendto outputs the content when a single
1954 data reference is specified. Otherwise, the reference for \@array or
1955 \$scalar is shown in 1.500, not the content.
1956 MCE->sendto('STDERR', \@array); # 1.00x behavior, content
1958 MCE->sendto('STDOUT', \$scalar);
1959 MCE->sendto('file:/path/to/file', \@array);
1960 # Output matches the print statement
1962 MCE->sendto(\*STDERR, \@array); # 1.500 behavior, reference
1964 MCE->sendto(\*STDOUT, \$scalar);
1965 MCE->sendto($fh, \@array);
1966 MCE->sendto('STDOUT', \@array, "\n", \$scalar, "\n");
1968 print {*STDOUT} \@array, "\n", \$scalar, "\n";
1969 MCE 1.500 added support for sending to a glob reference, file
1971 descriptor, and file handle.
1972 MCE->sendto(\*STDERR, "foo\n", \@array, \$scalar, "\n");
1974 MCE->sendto('fd:2', "foo\n", \@array, \$scalar, "\n");
1975 MCE->sendto($fh, "foo\n", \@array, \$scalar, "\n");
1976 MCE->sync ( void )
1978 $mce->sync ( void )
1979 A barrier sync operation means any worker must stop at this point until
1980 all workers reach this barrier. Barrier syncing is useful for many
1981 computer algorithms.
1982 Barrier synchronization is supported for task 0 only or omitting
1984 user_tasks. All workers assigned task_id 0 must call sync whenever
1985 barrier syncing.
1986 use MCE;
1988 sub user_func {
1990 my ($mce) = @_;
1992 my $wid = MCE->wid;
1993 MCE->sendto("STDOUT", "a: $wid\n"); # MCE 1.0+
1995 MCE->sync;
1996 MCE->sendto(\*STDOUT, "b: $wid\n"); # MCE 1.5+
1998 MCE->sync;
1999 MCE->print("c: $wid\n"); # MCE 1.5+
2001 MCE->sync;
2002 return;
2004 }
2005 my $mce = MCE->new(
2007 max_workers => 4, user_func => \&user_func
2008 )->run;
2009 -- Output (without barrier synchronization)
2011 a: 1
2013 a: 2
2014 b: 1
2015 b: 2
2016 c: 1
2017 c: 2
2018 a: 3
2019 b: 3
2020 c: 3
2021 a: 4
2022 b: 4
2023 c: 4
2024 -- Output (with barrier synchronization)
2026 a: 1
2028 a: 2
2029 a: 4
2030 a: 3
2031 b: 2
2032 b: 1
2033 b: 3
2034 b: 4
2035 c: 1
2036 c: 4
2037 c: 2
2038 c: 3
2039 Consider the following example. The MCE->sync operation is done inside
2041 a loop along with MCE->do. A stall may occur for workers calling sync
2042 the 2nd or 3rd time while other workers are sending results via MCE->do
2043 or MCE->sendto.
2044 It requires another semaphore lock in MCE to solve this which was not
2046 done in order to keep resources low. Therefore, please keep this in
2047 mind when mixing MCE->sync with MCE->do or output serialization methods
2048 inside a loop.
2049 sub user_func {
2051 my ($mce) = @_;
2053 my @result;
2054 for (1 .. 3) {
2056 ... compute algorithm ...
2057 MCE->sync;
2059 ... compute algorithm ...
2061 MCE->sync;
2063 MCE->do('aggregate_result', \@result); # or MCE->sendto
2065 MCE->sync; # The sync operation is also needed here to
2067 # prevent MCE from stalling.
2068 }
2069 }
2070 MCE->yield ( void )
2072 $mce->yield ( void )
2073 There may be on occasion when the MCE driven app is too fast. The
2074 interval option combined with the yield method, both introduced with
2075 MCE 1.5, allows one to throttle the app. It adds a "grace" factor to
2076 the design.
2077 A use case is an app configured with 100 workers running on a 24
2079 logical way box. Data is polled from a database containing over 2.5
2080 million rows. Workers chunk away at 300 rows per chunk performing SNMP
2081 gets (300 sockets per worker) polling 25 metrics from each device. With
2082 this scenario, the load on the box may rise beyond 90+. In addition,
2083 IP_Tables may reach its contention point causing the entire application
2084 to fail.
2085 The scenario above is solved by simply having workers yield among
2087 themselves in a synchronized fashion. A delay of 0.007 seconds between
2088 intervals is all that's needed. The load on the box will hover between
2089 23 ~ 27 for the duration of the run. Polling completes in under 17
2090 minutes time. This is quite fast considering the app polls 62.5 million
2091 metrics combined. The math equates to 3,676,470 per minute or rather
2092 61,275 per second from a single box.
2093 # Both max_nodes and node_id are optional (default 1).
2095 interval => {
2097 delay => 0.007, max_nodes => $max_nodes, node_id => $node_id
2098 }
2099 A 4 node setup can poll 10 million devices without the additional
2101 overhead of a distribution agent. The difference between the 4 nodes
2102 are simply node_id and the where clause used to query the database. The
2103 mac addresses are random such that the data divides equally to any
2104 power of 2. The distribution key lies in the mac address itself. In
2105 fact, the 2nd character from the right is sufficient for maximizing on
2106 the power of randomness for equal distribution.
2107 Query NodeID 1: ... AND substr(MAC, -2, 1) IN ('0', '1', '2', '3')
2109 Query NodeID 2: ... AND substr(MAC, -2, 1) IN ('4', '5', '6', '7')
2110 Query NodeID 3: ... AND substr(MAC, -2, 1) IN ('8', '9', 'A', 'B')
2111 Query NodeID 4: ... AND substr(MAC, -2, 1) IN ('C', 'D', 'E', 'F')
2112 Below, the user_tasks is configured to simulate 4 nodes. This
2114 demonstration uses 2 workers to minimize the output size. Input is from
2115 the sequence option.
2116 use Time::HiRes qw(time);
2118 use MCE;
2119 my $d = shift || 0.1;
2121 local $| = 1;
2123 sub create_task {
2125 my ($node_id) = @_;
2127 my $seq_size = 6;
2129 my $seq_start = ($node_id - 1) * $seq_size + 1;
2130 my $seq_end = $seq_start + $seq_size - 1;
2131 return {
2133 max_workers => 2, sequence => [ $seq_start, $seq_end ],
2134 interval => { delay => $d, max_nodes => 4, node_id => $node_id }
2135 };
2136 }
2137 sub user_begin {
2139 my ($mce, $task_id, $task_name) = @_;
2141 # The yield method causes this worker to wait for its next time
2143 # interval slot before running. Yield has no effect without the
2144 # 'interval' option.
2145 # Yielding is beneficial inside a user_begin block. A use case
2147 # is staggering database connections among workers in order
2148 # to not impact the DB server.
2149 MCE->yield;
2151 MCE->printf(
2153 "Node %2d: %0.5f -- Worker %2d: %12s -- Started\n",
2154 MCE->task_id + 1, time, MCE->task_wid, ''
2155 );
2156 return;
2158 }
2159 {
2161 my $prev_time = time;
2162 sub user_func {
2164 my ($mce, $seq_n, $chunk_id) = @_;
2166 # Yield simply waits for the next time interval.
2168 MCE->yield;
2169 # Calculate how long this worker has waited.
2171 my $curr_time = time;
2172 my $time_waited = $curr_time - $prev_time;
2173 $prev_time = $curr_time;
2175 MCE->printf(
2177 "Node %2d: %0.5f -- Worker %2d: %12.5f -- Seq_N %3d\n",
2178 MCE->task_id + 1, time, MCE->task_wid, $time_waited, $seq_n
2179 );
2180 return;
2182 }
2183 }
2184 # Simulate a 4 node environment passing node_id to create_task.
2186 print "Node_ID Current_Time Worker_ID Time_Waited Comment\n";
2188 MCE->new(
2190 user_begin => \&user_begin,
2191 user_func => \&user_func,
2192 user_tasks => [
2194 create_task(1),
2195 create_task(2),
2196 create_task(3),
2197 create_task(4)
2198 ]
2199 )->run;
2201 -- Output (notice Current_Time below, stays 0.10 apart)
2203 Node_ID Current_Time Worker_ID Time_Waited Comment
2205 Node 1: 1374807976.74634 -- Worker 1: -- Started
2206 Node 2: 1374807976.84634 -- Worker 1: -- Started
2207 Node 3: 1374807976.94638 -- Worker 1: -- Started
2208 Node 4: 1374807977.04639 -- Worker 1: -- Started
2209 Node 1: 1374807977.14634 -- Worker 2: -- Started
2210 Node 2: 1374807977.24640 -- Worker 2: -- Started
2211 Node 3: 1374807977.34649 -- Worker 2: -- Started
2212 Node 4: 1374807977.44657 -- Worker 2: -- Started
2213 Node 1: 1374807977.54636 -- Worker 1: 0.90037 -- Seq_N 1
2214 Node 2: 1374807977.64638 -- Worker 1: 1.00040 -- Seq_N 7
2215 Node 3: 1374807977.74642 -- Worker 1: 1.10043 -- Seq_N 13
2216 Node 4: 1374807977.84643 -- Worker 1: 1.20045 -- Seq_N 19
2217 Node 1: 1374807977.94636 -- Worker 2: 1.30037 -- Seq_N 2
2218 Node 2: 1374807978.04638 -- Worker 2: 1.40040 -- Seq_N 8
2219 Node 3: 1374807978.14641 -- Worker 2: 1.50042 -- Seq_N 14
2220 Node 4: 1374807978.24644 -- Worker 2: 1.60045 -- Seq_N 20
2221 Node 1: 1374807978.34628 -- Worker 1: 0.79996 -- Seq_N 3
2222 Node 2: 1374807978.44631 -- Worker 1: 0.79996 -- Seq_N 9
2223 Node 3: 1374807978.54634 -- Worker 1: 0.79996 -- Seq_N 15
2224 Node 4: 1374807978.64636 -- Worker 1: 0.79997 -- Seq_N 21
2225 Node 1: 1374807978.74628 -- Worker 2: 0.79996 -- Seq_N 4
2226 Node 2: 1374807978.84632 -- Worker 2: 0.79997 -- Seq_N 10
2227 Node 3: 1374807978.94634 -- Worker 2: 0.79996 -- Seq_N 16
2228 Node 4: 1374807979.04636 -- Worker 2: 0.79996 -- Seq_N 22
2229 Node 1: 1374807979.14628 -- Worker 1: 0.80001 -- Seq_N 5
2230 Node 2: 1374807979.24631 -- Worker 1: 0.80000 -- Seq_N 11
2231 Node 3: 1374807979.34634 -- Worker 1: 0.80001 -- Seq_N 17
2232 Node 4: 1374807979.44636 -- Worker 1: 0.80000 -- Seq_N 23
2233 Node 1: 1374807979.54628 -- Worker 2: 0.80000 -- Seq_N 6
2234 Node 2: 1374807979.64631 -- Worker 2: 0.80000 -- Seq_N 12
2235 Node 3: 1374807979.74633 -- Worker 2: 0.80000 -- Seq_N 18
2236 Node 4: 1374807979.84636 -- Worker 2: 0.80000 -- Seq_N 24
2237 The interval.pl example above is included with MCE.
2239
2241 The "progress" option takes a code block for receiving info on the
2242 progress made while processing input data; e.g. "input_data" or
2243 "sequence". To make this work, one provides the "progress" option a
2244 closure block like so, passing along the size of the input_data; e.g
2245 "scalar @array" or "-s /path/to/file".
2246 Current API available since 1.813.
2248 A worker, upon completing processing its chunk, notifies the manager-
2250 process with the size of the chunk. That could be the number of rows or
2251 literally the size of the chunk when processing an input file. The
2252 manager-process accumulates the size before calling the code block
2253 associated with the "progress" option.
2254 When running many tasks simultaneously, via "user_tasks", the call is
2256 initiated by workers at level 0 only or rather the first task, not
2257 shown here.
2258 use Time::HiRes 'sleep';
2260 use MCE;
2261 sub make_progress {
2263 my ($total_size) = @_;
2264 return sub {
2265 my ($completed_size) = @_;
2266 printf "%0.1f%%\n", $completed_size / $total_size * 100;
2267 };
2268 }
2269 my @input = (1..150);
2271 MCE->new(
2273 chunk_size => 10,
2274 max_workers => 4,
2275 input_data => \@input,
2276 progress => make_progress( scalar @input ),
2277 user_func => sub { sleep 1.5 }
2278 )->run();
2279 -- Output
2281 6.7%
2283 13.3%
2284 20.0%
2285 26.7%
2286 33.3%
2287 40.0%
2288 46.7%
2289 53.3%
2290 60.0%
2291 66.7%
2292 73.3%
2293 80.0%
2294 86.7%
2295 93.3%
2296 100.0%
2297 Next is the code using MCE::Flow and ProgressBar::Stack to do the same
2299 thing, practically.
2300 use Time::HiRes 'sleep';
2302 use ProgressBar::Stack;
2303 use MCE::Flow;
2304 sub make_progress {
2306 my ($total_size) = @_;
2307 init_progress();
2308 return sub {
2309 my ($completed_size) = @_;
2310 update_progress sprintf("%0.1f", $completed_size / $total_size * 100);
2311 };
2312 }
2313 my @input = (1..150);
2315 MCE::Flow->init(
2317 chunk_size => 10,
2318 max_workers => 4,
2319 progress => make_progress( scalar @input )
2320 );
2321 MCE::Flow->run( sub { sleep 1.5 }, \@input );
2323 MCE::Flow->finish();
2324 print "\n";
2326 -- Output
2328 [################ ] 80.0% ETA: 0:01
2330 For sequence of numbers, using the "sequence" option, one must account
2332 for "step_size", typically set to 1 automatically.
2333 use Time::HiRes 'sleep';
2335 use MCE;
2336 sub make_progress {
2338 my ($total_size) = @_;
2339 return sub {
2340 my ($completed_size) = @_;
2341 printf "%0.1f%%\n", $completed_size / $total_size * 100;
2342 };
2343 }
2344 MCE->new(
2346 chunk_size => 10,
2347 max_workers => 4,
2348 sequence => [ 1, 100, 2 ],
2349 progress => make_progress( int( 100 / 2 + 0.5 ) ),
2350 user_func => sub { sleep 1.5 }
2351 )->run();
2352 -- Output
2354 20.0%
2356 40.0%
2357 60.0%
2358 80.0%
2359 100.0%
2360 Changing "chunk_size" to 1 means workers notify the manager process
2362 more often, thus increasing granularity. Take a look at the output.
2363 2.0%
2365 4.0%
2366 6.0%
2367 8.0%
2368 10.0%
2369 ...
2370 92.0%
2371 94.0%
2372 96.0%
2373 98.0%
2374 100.0%
2375 Here is the same thing using MCE::Flow together with
2377 ProgressBar::Stack.
2378 use Time::HiRes 'sleep';
2380 use ProgressBar::Stack;
2381 use MCE::Flow;
2382 sub make_progress {
2384 my ($total_size) = @_;
2385 init_progress();
2386 return sub {
2387 my ($completed_size) = @_;
2388 update_progress sprintf("%0.1f", $completed_size / $total_size * 100);
2389 };
2390 }
2391 MCE::Flow->init(
2393 chunk_size => 1,
2394 max_workers => 4,
2395 progress => make_progress( int( 100 / 2 + 0.5 ) )
2396 );
2397 MCE::Flow->run_seq( sub { sleep 0.5 }, 1, 100, 2 );
2399 MCE::Flow->finish();
2400 print "\n";
2402 -- Output
2404 [######### ] 48.0% ETA: 0:03
2406 For files and file handles, workers send the actual size of the data
2408 read versus counting rows.
2409 use Time::HiRes 'sleep';
2411 use MCE;
2412 sub make_progress {
2414 my ($total_size) = @_;
2415 return sub {
2416 my ($completed_size) = @_;
2417 printf "%0.1f%%\n", $completed_size / $total_size * 100;
2418 };
2419 }
2420 my $input_file = "/path/to/input_file.txt";
2422 MCE->new(
2424 chunk_size => 5,
2425 max_workers => 4,
2426 input_data => $input_file,
2427 progress => make_progress( -s $input_file ),
2428 user_func => sub { sleep 0.03 }
2429 )->run();
2430 For consistency, here is the example using MCE::Flow, again with
2432 ProgressBar::Stack.
2433 use Time::HiRes 'sleep';
2435 use ProgressBar::Stack;
2436 use MCE::Flow;
2437 sub make_progress {
2439 my ($total_size) = @_;
2440 init_progress();
2441 return sub {
2442 my ($completed_size) = @_;
2443 update_progress sprintf("%0.1f", $completed_size / $total_size * 100);
2444 };
2445 }
2446 my $input_file = "/path/to/input_file.txt";
2448 MCE::Flow->init(
2450 chunk_size => 5,
2451 max_workers => 4,
2452 progress => make_progress( -s $input_file )
2453 );
2454 MCE::Flow->run_file( sub { sleep 0.03 }, $input_file );
2456 MCE::Flow->finish();
2457 The next demonstration processes three arrays consecutively. For this
2459 one, MCE workers persist after running. This needs MCE 1.814 or later
2460 to run. Otherwise, the progress output is not shown in MCE 1.813.
2461 use Time::HiRes 'sleep';
2463 use ProgressBar::Stack;
2464 use MCE;
2465 sub make_progress {
2467 my ($total_size, $message) = @_;
2468 init_progress();
2469 return sub {
2470 my ($completed_size) = @_;
2471 update_progress(
2472 sprintf("%0.1f", $completed_size / $total_size * 100),
2473 $message
2474 );
2475 };
2476 }
2477 my $mce = MCE->new(
2479 chunk_size => 10,
2480 max_workers => 4,
2481 user_func => sub { sleep 0.5 }
2482 )->spawn();
2483 my @a1 = ( 1 .. 200 );
2485 my @a2 = ( 1 .. 500 );
2486 my @a3 = ( 1 .. 300 );
2487 $mce->process({ progress => make_progress(scalar(@a1), "array 1") }, \@a1);
2489 print "\n";
2491 $mce->process({ progress => make_progress(scalar(@a2), "array 2") }, \@a2);
2493 print "\n";
2495 $mce->process({ progress => make_progress(scalar(@a3), "array 3") }, \@a3);
2497 print "\n";
2499 $mce->shutdown;
2501 -- Output
2503 [####################] 100.0% ETA: 0:00 array 1
2505 [####################] 100.0% ETA: 0:00 array 2
2506 [####################] 100.0% ETA: 0:00 array 3
2507 When size is not know, such as reading from "STDIN", the only thing one
2509 can do is report the size completed thus far.
2510 # 1 kibibyte equals 1024 bytes
2512 progress => sub {
2514 my ($completed_size) = @_;
2515 printf "%0.1f kibibytes\n", $completed_size / 1024;
2516 }
2517
2519 • MCE::Examples
2520
2522 MCE
2523
2525 Mario E. Roy, <marioeroy AT gmail DOT com>
2526perl v5.38.0 2023-09-14 MCE::Core(3)