MCE - Many-Core Engine for Perl. Provides parallel processing capabilities.
This document describes MCE version 1.106
Many-core Engine (MCE) for Perl helps enable a new level of performance by maximizing all available cores. MCE spawns a pool of workers and therefore does not fork a new process per each element of data. Instead, MCE follows a bank queuing model. Imagine the line being the data and bank-tellers the parallel workers. MCE enhances that model by adding the ability to chunk the next n elements from the input stream to the next available worker.
Both chunking and input are optional in MCE. One can simply use MCE to have many workers run in parallel.
use MCE;
## A new instance shown with all available options. my $mce = MCE->new( tmp_dir => $tmp_dir, ## Default is $MCE::Signal::tmp_dir which points to ## $ENV{TEMP} if defined. Otherwise, tmp_dir points ## to /tmp. input_data => $input_file, ## Default is undef ## input_data => '/path/to/file' for input file ## input_data => \@array for input array ## input_data => \*FILE_HNDL for file handle ## input_data => \$scalar to treat like file chunk_size => 2000, ## Default is 500 ## Less than or equal to 8192 is number of records. ## Greater than 8192 is number of bytes. MCE reads ## till the end of record before calling user_func. ## chunk_size => 1, ## Consists of 1 record ## chunk_size => 1000, ## Consists of 1000 records ## chunk_size => 16384, ## Approximate 16384 bytes ## chunk_size => 50000, ## Approximate 50000 bytes max_workers => 8, ## Default is 2 use_slurpio => 1, ## Default is 0 use_threads => 1, ## Default is 0 or 1 ## Number of workers to spawn, whether or not to enable ## slurpio when reading files (passes raw chunk to user ## function), and whether or not to use threads. ## By default MCE does forking (spawns child processes). ## MCE also supports threads via 2 threading libraries. ## ## The use of threads in MCE requires that you include ## threads support prior to loading MCE. ## ## use threads; use forks; ## use threads::shared; (or) use forks::shared; ## ## use MCE use MCE; job_delay => 0.035, ## Default is undef spawn_delay => 0.150, ## Default is undef submit_delay => 0.001, ## Default is undef ## Time to wait, in fractional seconds, before processing ## job, spawning workers, and parameters submission to ## workers. Use submit_delay if wanting to stagger many ## workers connecting to a database. user_begin => \&user_begin, ## Default is undef user_func => \&user_func, ## Default is undef user_end => \&user_end, ## Default is undef ## Think of user_begin, user_func, user_end like the awk ## scripting language: ## awk 'BEGIN { ... } { ... } END { ... }' ## MCE workers calls user_begin once per job, then ## calls user_func repeatedly until no chunks remain. ## Afterwards, user_end is called. user_error => \&user_error, ## Default is undef user_output => \&user_output, ## Default is undef ## When workers call the following functions, MCE will ## pass the data to user_error/user_output if defined. ## $self->sendto('stderr', 'Sending to STDERR'); ## $self->sendto('stdout', 'Sending to STDOUT'); stderr_file => 'err_file', ## Default is STDERR stdout_file => 'out_file', ## Default is STDOUT ## Or to file. User_error/user_output take precedence. flush_file => 1, ## Default is 0 flush_stderr => 1, ## Default is 0 flush_stdout => 1, ## Default is 0 ## Flush sendto file, standard error, or standard output. );
## Run calls spawn, kicks off job, workers call user_begin, ## user_func, user_end. Run shuts down workers afterwards. $mce->run(); ## OR, spawn workers early. $mce->spawn(); ## Acquire data arrays and/or input_files. The same pool of ## workers are used. $mce->process(\@input_data_1); ## Process arrays $mce->process(\@input_data_2); $mce->process(\@input_data_n); $mce->process('input_file_1'); ## Process files $mce->process('input_file_2'); $mce->process('input_file_n'); ## Shutdown workers afterwards. $mce->shutdown();
## Both user_begin and user_end functions, if specified, behave ## similarly to awk 'BEGIN { ... } { ... } END { ... }'. ## Each worker calls this once prior to processing. sub user_begin { ## Optional via user_begin option my $self = shift; ## Prefix variables with wk_ $self->{wk_total_rows} = 0; } ## And once after completion. sub user_end { ## Optional via user_end option my $self = shift; printf "## %d: Processed %d rows\n", $self->wid(), $self->{wk_total_rows}; }
## MCE passes a reference to an array containing the chunk data. sub user_func { my ($self, $chunk_ref, $chunk_id) = @_; foreach my $row ( @{ $chunk_ref } ) { print $row; $self->{wk_total_rows} += 1; } }
## MCE passes a reference to a scalar containing the raw chunk data. sub user_func { my ($self, $chunk_ref, $chunk_id) = @_; my $count = () = $$chunk_ref =~ /abc/; }
## MCE will direct $self->sendto('stderr/out', ...) calls to these ## functions in a serialized fashion. This is handy if one wants to ## filter, modify, and/or send the data elsewhere. sub user_error { ## Optional via user_error option my $error = shift; print LOGERR $error; } sub user_output { ## Optional via user_output option my $output = shift; print LOGOUT $output; }
Methods listed below are callable by the main process and workers.
## Notifies workers to abort after processing the current chunk. ## The abort method is only meaningful when processing input_data. $self->abort( void );
## Returns the worker ID of worker. my $wid = $self->wid();
Methods listed below are callable by the main process only.
## Both forchunk & foreach are sugar methods in MCE. Workers are ## automatically spawned, the code block is executed in parallel, ## and workers are shut down afterwards. Do not call these methods ## if wanting workers to remain up after processing. ## ## Specifying options is optional. Valid options are the same as ## for the process method. my $mce = MCE->new( chunk_size => 20, max_workers => $max_workers ); ## Arguments inside code block are the same as for user_func. $mce->forchunk(\@input_data, sub { my ($self, $chunk_ref, $chunk_id) = @_; foreach ( @{ $chunk_ref } ) { $self->sendto("stdout", "$chunk_id: $_\n"); } }); ## Passing chunk_size as an option. $mce->forchunk(\@input_data, { chunk_size => 30 }, sub { ... });
## Foreach always implies chunk_size => 1 (cannot be overwritten). my $mce = MCE->new( max_workers => $max_workers ); ## Arguments inside code block are the same as for user_func. ## This holds true even if chunk_size is set to 1. MCE is both ## a chunking engine plus parallel engine all in one. Arguments ## within the block are the same whether chunking is 1 or > 1. $mce->foreach(\@input_data, sub { my ($self, $chunk_ref, $chunk_id) = @_; my $row = $chunk_ref->[0]; $self->sendto("stdout", "$chunk_id: $row\n"); }); ## Passing an anonymous array as input data. For example, ## wanting to parallelize a serial for loop with MCE. for (my $i = 0; $i < $max; $i++) { ... ## Runs serially } for my $i (0 .. $max - 1) { ... ## Runs serially } $mce->foreach([ (0 .. $max - 1) ], sub { my ($self, $chunk_ref, $chunk_id) = @_; my $i = $chunk_ref->[0]; (OR) my $i = $chunk_id - 1; ... ## Runs in parallel });
## The process method will spawn workers automatically if not already ## spawned. It will set input_data => $input_data. It calls run(0) to ## not auto-shutdown workers. Specifying options is optional. ## ## Allowable options { key => value, ... } are: ## chunk_size input_data job_delay spawn_delay submit_delay use_slurpio ## flush_file flush_stderr flush_stdout stderr_file stdout_file ## user_begin user_end user_func user_error user_output ## ## Options remain persistent going forward unless changed. Setting ## user_begin, user_end, or user_func will cause already spawned ## workers to shutdown and re-spawn automatically. Therefore, define ## these during instantiation if possible. my $mce = MCE->new( ... ); $mce->spawn(); $mce->process($array_ref); $mce->process($array_ref, { stdout_file => $output_file }); $mce->shutdown();
## The run method, by default, spawns workers, processes once, ## and shuts down workers. Set $auto_shutdown to 0 if not wanting ## to auto-shutdown workers after processing (default is 1). ## ## Specifying options is optional. Valid options are the same as ## for the process method. my $mce = MCE->new( ... ); $mce->run(0); ## Disables auto-shutdown
## The run method will automatically spawn workers, run once, and ## shutdown workers automatically. The process method leaves workers ## waiting for the next job after processing. Call shutdown after ## processing all jobs. my $mce = MCE->new( ... ); $mce->process(\@input_data_1); ## Processing multiple arrays $mce->process(\@input_data_2); $mce->process(\@input_data_n); $mce->process('input_file_1'); ## Processing multiple files $mce->process('input_file_2'); $mce->process('input_file_n'); $mce->shutdown();
## Workers are normally spawned automatically. The spawn method is ## beneficial when wanting to spawn workers early. my $mce = MCE->new( ... ); $mce->spawn();
Methods listed below are callable by workers only.
## MCE can serialized data transfers from worker processes via ## helper functions do & sendto. The main MCE thread will process ## these in a serial fashion. This utilizes the Storable Perl module ## for passing data from a worker process to the main MCE thread. ## The callback function can optionally return a reply. [ $reply = ] $self->do('callback' [, $arg1, ... ]); ## Passing arguments to a callback function using references & scalar. sub callback { my ($array_ref, $hash_ref, $scalar_ref, $scalar) = @_; ... } $self->do('main::callback', \@a, \%h, \$s, 'hello'); $self->do('callback', \@a, \%h, \$s, 'hello'); ## MCE knows if wanting a void, list, hash, or a scalar return value. $self->do('callback' [, $arg1, ... ]); my @array = $self->do('callback' [, $arg1, ... ]); my %hash = $self->do('callback' [, $arg1, ... ]); my $scalar = $self->do('callback' [, $arg1, ... ]);
## The worker exits the current job. $self->exit();
## Worker immediately exits the chunking loop or user func. ## Call this inside foreach, forchunk, and user_func. my @list = (1 .. 80); $mce->forchunk(\@list, { chunk_size => 2 }, sub { my ($self, $chunk_ref, $chunk_id) = @_; $self->last if ($chunk_id > 4); my @output = (); foreach my $rec ( @{ $chunk_ref } ) { push @output, $rec, "\n"; } $self->sendto('stdout', @output); }); -- Output (each chunk above consists of 2 elements) 1 2 3 4 5 6 7 8
## Worker starts the next iteration of the chunking loop. ## Call this inside foreach, forchunk, and user_func. my @list = (1 .. 80); $mce->forchunk(\@list, { chunk_size => 4 }, sub { my ($self, $chunk_ref, $chunk_id) = @_; $self->next if ($chunk_id < 20); my @output = (); foreach my $rec ( @{ $chunk_ref } ) { push @output, $rec, "\n"; } $self->sendto('stdout', @output); }); -- Output (each chunk above consists of 4 elements) 77 78 79 80
The sendto method is called by workers to serialize data to standard output, standard error, or to end of file. The action is done by the main process or thread.
Release 1.100 adds the ability to pass multiple arguments.
## Release 1.00x supported only 1 data argument. ## /path/to/file is the 3rd argument for 'file'. $self->sendto('stdout', \@array); $self->sendto('stdout', \$scalar); $self->sendto('stdout', $scalar); $self->sendto('stderr', \@array); $self->sendto('stderr', \$scalar); $self->sendto('stderr', $scalar); $self->sendto('file', \@array, '/path/to/file'); $self->sendto('file', \$scalar, '/path/to/file'); $self->sendto('file', $scalar, '/path/to/file');
## Notice the syntax change for appending to a file. $self->sendto('stdout', $arg1 [, $arg2, ... ]); $self->sendto('stderr', $arg1 [, $arg2, ... ]); $self->sendto('file:/path/to/file', $arg1 [, $arg2, ... ]); ## Passing a reference is no longer necessary beginning with 1.100. $self->sendto("stdout", @a, "\n", %h, "\n", $s, "\n"); ## To retain 1.00x compatibility, sendto outputs the content when a ## a single data argument is specified and is a reference. $self->sendto('stdout', \@array); $self->sendto('stderr', \$scalar); $self->sendto('file:/path/to/file', \@array); ## Otherwise, the reference for \@array and \$scalar is shown, ## not the content. Basically, output matches the print statement. ## Ex. print STDOUT "hello\n", \@array, \$scalar, "\n"; $self->sendto('stdout', "hello\n", \@array, \$scalar, "\n");
MCE comes with various examples showing real-world use case scenarios on parallelizing something as small as cat (try with -n) to greping for patterns and word count aggregation.
cat.pl Concatenation script, similar to the cat binary. egrep.pl Egrep script, similar to the egrep binary. wc.pl Word count script, similar to the wc binary. findnull.pl A parallel driven script to report lines containing null fields. It's many times faster than the binary egrep command. Try against a large file containing very long lines. scaling_pings.pl Perform ping test and report back failing IPs to standard output. tbray/wf_mce1.pl, wf_mce2.pl, wf_mce3.pl An implementation of wide finder utilizing MCE. As fast as MMAP IO when file resides in OS FS cache. 2x ~ 3x faster when reading directly from disk. foreach.pl forchunk.pl These take the same sqrt example from Parallel::Loops and measures the overhead of the engine. The number indicates the size of @input which can be submitted and results displayed in 1 second. Parallel::Loops: 600 Forking each @input is expensive MCE foreach....: 18,000 Sends result after each @input MCE forchunk...: 385,000 Chunking reduces overhead
## Imagine a long running process and wanting to parallelize an array ## against a pool of workers. my @input_data = (0 .. 18000 - 1); my $max_workers = 3; my $order_id = 1; my %result; ## Callback function for displaying results. The logic below shows how ## one can display results immediately while still preserving output ## order. The %result hash is a temporary cache to store results ## for out-of-order replies. sub display_result { my ($wk_result, $chunk_id) = @_; $result{$chunk_id} = $wk_result; while (1) { last unless (exists $result{$order_id}); printf "i: %d sqrt(i): %f\n", $input_data[$order_id - 1], $result{$order_id}; delete $result{$order_id}; $order_id++; } } ## Compute via MCE. my $mce = MCE->new( input_data => \@input_data, max_workers => $max_workers, chunk_size => 1, user_func => sub { my ($self, $chunk_ref, $chunk_id) = @_; my $wk_result = sqrt($chunk_ref->[0]); $self->do('display_result', $wk_result, $chunk_id); } ); $mce->run();
## Compute via MCE. Foreach implies chunk_size => 1. my $mce = MCE->new( max_workers => $max_workers ); ## Worker calls code block passing a reference to an array containing ## one item. Use $chunk_ref->[0] to retrieve the single element. $mce->foreach(\@input_data, sub { my ($self, $chunk_ref, $chunk_id) = @_; my $wk_result = sqrt($chunk_ref->[0]); $self->do('display_result', $wk_result, $chunk_id); });
## Chunking reduces overhead many folds. Instead of passing a single ## item from @input_data, a chunk of $chunk_size is sent instead to ## the next available worker. my @input_data = (0 .. 385000 - 1); my $max_workers = 3; my $chunk_size = 500; my $order_id = 1; my %result; ## Callback function for displaying results. sub display_result { my ($wk_result, $chunk_id) = @_; $result{$chunk_id} = $wk_result; while (1) { last unless (exists $result{$order_id}); my $i = ($order_id - 1) * $chunk_size; foreach ( @{ $result{$order_id} } ) { printf "i: %d sqrt(i): %f\n", $input_data[$i++], $_; } delete $result{$order_id}; $order_id++; } } ## Compute via MCE. my $mce = MCE->new( input_data => \@input_data, max_workers => $max_workers, chunk_size => $chunk_size, user_func => sub { my ($self, $chunk_ref, $chunk_id) = @_; my @wk_result; foreach ( @{ $chunk_ref } ) { push @wk_result, sqrt($_); } $self->do('display_result', \@wk_result, $chunk_id); } ); $mce->run();
## Compute via MCE. my $mce = MCE->new( max_workers => $max_workers, chunk_size => $chunk_size ); ## Below, $chunk_ref is a reference to an array containing the next ## $chunk_size items from @input_data. $mce->forchunk(\@input_data, sub { my ($self, $chunk_ref, $chunk_id) = @_; my @wk_result; foreach ( @{ $chunk_ref } ) { push @wk_result, sqrt($_); } $self->do('display_result', \@wk_result, $chunk_id); });
The input_data option is optional. One can simply use MCE to parallelize multiple workers. The "do" & "sendto" methods are used to pass data back to the main process or thread. One doesn't have to wait until the worker has completed processing to pass data back. Both "do" & "sendto" methods are processed serially by the main process on a first come, first serve basis. All 4 workers run in parallel for the demonstration below.
use MCE; sub report_stats { my ($wid, $msg, $hash_ref) = @_; print "Worker $wid says $msg: ", $hash_ref->{'counter'}, "\n"; } my $mce = MCE->new( max_workers => 4, user_func => sub { my ($self) = @_; my $wid = $self->wid(); if ($wid == 1) { my %hash = ('counter' => 0); while (1) { $hash{'counter'} += 1; $self->do('report_stats', $wid, 'Hello there', \%hash); last if ($hash{'counter'} == 4); sleep 2; } } else { my %hash = ('counter' => 0); while (1) { $hash{'counter'} += 1; $self->do('report_stats', $wid, 'Welcome ...', \%hash); last if ($hash{'counter'} == 2); sleep 4; } } $self->sendto('stdout', "Worker $wid is exiting\n"); } ); $mce->run;
Worker 2 gets there first in 2nd output below.
$ ./demo.pl Worker 1 says Hello there: 1 Worker 2 says Welcome ...: 1 Worker 3 says Welcome ...: 1 Worker 4 says Welcome ...: 1 Worker 1 says Hello there: 2 Worker 2 says Welcome ...: 2 Worker 3 says Welcome ...: 2 Worker 1 says Hello there: 3 Worker 2 is exiting Worker 3 is exiting Worker 4 says Welcome ...: 2 Worker 4 is exiting Worker 1 says Hello there: 4 Worker 1 is exiting $ ./demo.pl Worker 2 says Welcome ...: 1 Worker 1 says Hello there: 1 Worker 4 says Welcome ...: 1 Worker 3 says Welcome ...: 1 Worker 1 says Hello there: 2 Worker 2 says Welcome ...: 2 Worker 4 says Welcome ...: 2 Worker 3 says Welcome ...: 2 Worker 2 is exiting Worker 4 is exiting Worker 1 says Hello there: 3 Worker 3 is exiting Worker 1 says Hello there: 4 Worker 1 is exiting
Perl 5.8.0 or later
MCE::Signal
The source is hosted at: http://code.google.com/p/many-core-engine-perl/
Mario E. Roy, <marioeroy AT gmail DOT com>
Copyright (C) 2012 by Mario E. Roy
MCE is free software; you can redistribute it and/or modify it under the same terms as Perl itself http://dev.perl.org/licenses/.
To install MCE, copy and paste the appropriate command in to your terminal.
cpanm
cpanm MCE
CPAN shell
perl -MCPAN -e shell install MCE
For more information on module installation, please visit the detailed CPAN module installation guide.