MCE::Channel(3) User Contributed Perl Documentation MCE::Channel(3)
NAME
MCE::Channel - Queue-like and two-way communication capability
VERSION
This document describes MCE::Channel version 1.865
SYNOPSIS
use MCE::Channel;
########################
# Construction
########################
# A single producer and many consumers supporting processes and threads
my $c1 = MCE::Channel->new( impl => 'Mutex' ); # default implementation
my $c2 = MCE::Channel->new( impl => 'Threads' ); # threads::shared locking
# Set the mp flag if two or more workers (many producers) will be calling
# enqueue/send or recv2/recv2_nb on the left end of the channel
my $c3 = MCE::Channel->new( impl => 'Mutex', mp => 1 );
my $c4 = MCE::Channel->new( impl => 'Threads', mp => 1 );
# Tuned for one producer and one consumer, no locking
my $c5 = MCE::Channel->new( impl => 'Simple' );
########################
# Queue-like behavior
########################
# Send data to consumers
$c1->enqueue('item');
$c1->enqueue(qw/item1 item2 item3 itemN/);
# Receive data
my $item = $c1->dequeue(); # item
my @items = $c1->dequeue(2); # (item1, item2)
# Receive, non-blocking
my $item = $c1->dequeue_nb(); # item
my @items = $c1->dequeue_nb(2); # (item1, item2)
# Signal that there is no more work to be sent
$c1->end();
########################
# Two-way communication
########################
# Producer(s) sending data
$c3->send('message');
$c3->send(qw/arg1 arg2 arg3/);
# Consumer(s) receiving data
my $mesg = $c3->recv(); # message
my @args = $c3->recv(); # (arg1, arg2, arg3)
# Alternatively, non-blocking
my $mesg = $c3->recv_nb(); # message
my @args = $c3->recv_nb(); # (arg1, arg2, arg3)
# A producer signaling no more work to be sent
$c3->end();
# Consumers(s) sending data
$c3->send2('message');
$c3->send2(qw/arg1 arg2 arg3/);
# Producer(s) receiving data
my $mesg = $c3->recv2(); # message
my @args = $c3->recv2(); # (arg1, arg2, arg3)
# Alternatively, non-blocking
my $mesg = $c3->recv2_nb(); # message
my @args = $c3->recv2_nb(); # (arg1, arg2, arg3)
DESCRIPTION
A MCE::Channel object is a container for sending and receiving data
using socketpair handles. Serialization is provided by Sereal if
available. Defaults to Storable otherwise. Excluding the "Simple"
implementation, both ends of the "channel" support many workers
concurrently (with mp => 1).
new ( impl => STRING, mp => BOOLEAN )
This creates a new channel. Three implementations are provided "Mutex",
"Threads", and "Simple" indicating the locking mechanism to use
"MCE::Mutex", "threads::shared", and no locking respectively.
$chnl = MCE::Channel->new(); # default: impl => 'Mutex', mp => 0
# default: impl => 'Threads' on Windows
The "Mutex" implementation supports processes and threads whereas the
"Threads" implementation is suited for Windows and threads only.
$chnl = MCE::Channel->new( impl => 'Mutex' ); # MCE::Mutex locking
$chnl = MCE::Channel->new( impl => 'Threads' ); # threads::shared locking
# on Windows, silently becomes impl => 'Threads' when specifying 'Mutex'
Set the "mp" (m)any (p)roducers option to a true value if there will be
two or more workers calling "enqueue", <send>, "recv2", or "recv2_nb"
on the left end of the channel. This is important to not incur a race
condition.
$chnl = MCE::Channel->new( impl => 'Mutex', mp => 1 );
$chnl = MCE::Channel->new( impl => 'Threads', mp => 1 );
# on Windows, silently becomes impl => 'Threads' when specifying 'Mutex'
The "Simple" implementation is optimized for one producer and one
consumer max. It omits locking for maximum performance. This
implementation is preferred for parent to child communication not
shared by another worker.
$chnl = MCE::Channel->new( impl => 'Simple' );
QUEUE-LIKE BEHAVIOR
enqueue ( ITEM1 [, ITEM2, ... ] )
Appends a list of items onto the left end of the channel. This will
block once the internal socket buffer becomes full (i.e. awaiting
workers to dequeue on the other end). This prevents producer(s) from
running faster than consumer(s).
Object (de)serialization is handled automatically using Sereal if
available or defaults to Storable otherwise.
$chnl->enqueue('item1');
$chnl->enqueue(qw/item2 item3 .../);
$chnl->enqueue([ array_ref1 ]);
$chnl->enqueue([ array_ref2 ], [ array_ref3 ], ...);
$chnl->enqueue({ hash_ref1 });
$chnl->enqueue({ hash_ref2 }, { hash_ref3 }, ...);
dequeue
dequeue ( COUNT )
Removes the requested number of items (default 1) from the right end of
the channel. If the channel contains fewer than the requested number of
items, the method will block (i.e. until other producer(s) enqueue more
items).
$item = $chnl->dequeue(); # item1
@items = $chnl->dequeue(2); # ( item2, item3 )
dequeue_nb
dequeue_nb ( COUNT )
Removes the requested number of items (default 1) from the right end of
the channel. If the channel contains fewer than the requested number of
items, the method will return what it was able to retrieve and return
immediately. If the channel is empty, then returns "an empty list" in
list context or "undef" in scalar context.
$item = $chnl->dequeue_nb(); # array_ref1
@items = $chnl->dequeue_nb(2); # ( array_ref2, array_ref3 )
end
This is called by a producer to signal that there is no more work to be
sent. Once ended, no more items may be sent by the producer. Calling
"end" by multiple producers is not supported.
$chnl->end;
TWO-WAY IPC - PRODUCER TO CONSUMER
send ( ARG1 [, ARG2, ... ] )
Append data onto the left end of the channel. Unlike "enqueue", the
values are kept together for the receiving consumer, similarly to
calling a method. Object (de)serialization is handled automatically.
$chnl->send('item');
$chnl->send([ list_ref ]);
$chnl->send([ hash_ref ]);
$chnl->send(qw/item1 item2 .../);
$chnl->send($id, [ list_ref ]);
$chnl->send($id, { hash_ref });
recv
recv_nb
Blocking and non-blocking fetch methods from the right end of the
channel. For the latter and when the channel is empty, returns "an
empty list" in list context or "undef" in scalar context.
$item = $chnl->recv();
$array_ref = $chnl->recv();
$hash_ref = $chnl->recv();
($item1, $item2) = $chnl->recv_nb();
($id, $array_ref) = $chnl->recv_nb();
($id, $hash_ref) = $chnl->recv_nb();
TWO-WAY IPC - CONSUMER TO PRODUCER
send2 ( ARG1 [, ARG2, ... ] )
Append data onto the right end of the channel. Unlike "enqueue", the
values are kept together for the receiving producer, similarly to
calling a method. Object (de)serialization is handled automatically.
$chnl->send2('item');
$chnl->send2([ list_ref ]);
$chnl->send2([ hash_ref ]);
$chnl->send2(qw/item1 item2 .../);
$chnl->send2($id, [ list_ref ]);
$chnl->send2($id, { hash_ref });
recv2
recv2_nb
Blocking and non-blocking fetch methods from the left end of the
channel. For the latter and when the channel is empty, returns "an
empty list" in list context or "undef" in scalar context.
$item = $chnl->recv2();
$array_ref = $chnl->recv2();
$hash_ref = $chnl->recv2();
($item1, $item2) = $chnl->recv2_nb();
($id, $array_ref) = $chnl->recv2_nb();
($id, $hash_ref) = $chnl->recv2_nb();
DEMONSTRATIONS
Example 1 - threads
"MCE::Channel" was made to work efficiently with threads. The reason is
from using threads::shared for locking versus MCE::Mutex.
use strict;
use warnings;
use threads;
use MCE::Channel;
my $queue = MCE::Channel->new( impl => 'Threads' );
my $num_consumers = 10;
sub consumer {
my $count = 0;
# receive items
while ( my ($item1, $item2) = $queue->dequeue(2) ) {
$count += 2;
}
# send result
$queue->send2( threads->tid => $count );
}
threads->create('consumer') for 1 .. $num_consumers;
## producer
$queue->enqueue($_, $_ * 2) for 1 .. 40000;
$queue->end;
my %results;
my $total = 0;
for ( 1 .. $num_consumers ) {
my ($id, $count) = $queue->recv2;
$results{$id} = $count;
$total += $count;
}
$_->join for threads->list;
print $results{$_}, "\n" for keys %results;
print "$total total\n\n";
__END__
# output
8034
8008
8036
8058
7990
7948
8068
7966
7960
7932
80000 total
Example 2 - MCE::Child
The following is similarly threads-like for Perl lacking threads
support. It spawns processes instead, thus requires the "Mutex"
channel implementation which is the default if omitted.
use strict;
use warnings;
use MCE::Child;
use MCE::Channel;
my $queue = MCE::Channel->new( impl => 'Mutex' );
my $num_consumers = 10;
sub consumer {
my $count = 0;
# receive items
while ( my ($item1, $item2) = $queue->dequeue(2) ) {
$count += 2;
}
# send result
$queue->send2( MCE::Child->pid => $count );
}
MCE::Child->create('consumer') for 1 .. $num_consumers;
## producer
$queue->enqueue($_, $_ * 2) for 1 .. 40000;
$queue->end;
my %results;
my $total = 0;
for ( 1 .. $num_consumers ) {
my ($id, $count) = $queue->recv2;
$results{$id} = $count;
$total += $count;
}
$_->join for MCE::Child->list;
print $results{$_}, "\n" for keys %results;
print "$total total\n\n";
Example 3 - Consumer requests item
Like the previous example, but have the manager process await a
notification from the consumer before inserting into the queue. This
allows the producer to end the channel early (i.e. exit loop).
use strict;
use warnings;
use MCE::Child;
use MCE::Channel;
my $queue = MCE::Channel->new( impl => 'Mutex' );
my $num_consumers = 10;
sub consumer {
# receive items
my $count = 0;
while () {
# Notify the manager process to send items. This allows the
# manager process to enqueue only when requested. The benefit
# is being able to end the channel immediately.
$queue->send2( MCE::Child->pid ); # channel is bi-directional
my ($item1, $item2) = $queue->dequeue(2);
last unless ( defined $item1 ); # channel ended
$count += 2;
}
# result
return ( MCE::Child->pid => $count );
}
MCE::Child->create('consumer') for 1 .. $num_consumers;
## producer
for my $num (1 .. 40000) {
# Await worker notification before inserting (blocking).
my $consumer_pid = $queue->recv2;
$queue->enqueue($num, $num * 2);
}
$queue->end;
my %results;
my $total = 0;
for my $child ( MCE::Child->list ) {
my ($id, $count) = $child->join;
$results{$id} = $count;
$total += $count;
}
print $results{$_}, "\n" for keys %results;
print "$total total\n\n";
Example 4 - Many producers
Running with 2 or more producers requires setting the "mp" option.
Internally, this enables locking support for the left end of the
channel. The "mp" option applies to "Mutex" and "Threads" channel
implementations only.
Here, using the MCE facility for gathering the final count.
use strict;
use warnings;
use MCE::Flow;
use MCE::Channel;
my $queue = MCE::Channel->new( impl => 'Mutex', mp => 1 );
my $num_consumers = 10;
sub consumer {
# receive items
my $count = 0;
while ( my ( $item1, $item2 ) = $queue->dequeue(2) ) {
$count += 2;
}
# send result
MCE->gather( MCE->wid => $count );
}
sub producer {
$queue->enqueue($_, $_ * 2) for 1 .. 20000;
}
## run 2 producers and many consumers
MCE::Flow::init(
max_workers => [ 2, $num_consumers ],
task_name => [ 'producer', 'consumer' ],
task_end => sub {
my ($mce, $task_id, $task_name) = @_;
if ( $task_name eq 'producer' ) {
$queue->end;
}
}
);
# consumers call gather above (i.e. send a key-value pair),
# have MCE append to a hash
my %results = mce_flow \&producer, \&consumer;
MCE::Flow::finish;
my $total = 0;
for ( keys %results ) {
$total += $results{$_};
print $results{$_}, "\n";
}
print "$total total\n\n";
Example 5 - Many channels
This demonstration configures a channel per consumer. Plus, a common
channel for consumers to request the next input item. The "Simple"
implementation is specified for the individual channels whereas locking
may be necessary for the $ready channel. However, consumers do not
incur reading and what is written is very small (i.e. atomic write is
guaranteed by the OS). Thus, am safely choosing the "Simple"
implementation versus "Mutex".
use strict;
use warnings;
use MCE::Flow;
use MCE::Channel;
my $prog_name = $0; $prog_name =~ s{^.*[\\/]}{}g;
my $input_size = shift || 3000;
unless ($input_size =~ /\A\d+\z/) {
print {*STDERR} "usage: $prog_name [ size ]\n";
exit 1;
}
my $consumers = 4;
my @chnls = map { MCE::Channel->new( impl => 'Simple' ) } 1 .. $consumers;
my $ready = MCE::Channel->new( impl => 'Simple' );
sub producer {
my $id = 0;
# send the next input item upon request
for ( 0 .. $input_size - 1 ) {
my $chnl_num = $ready->recv2;
$chnls[ $chnl_num ]->send( ++$id, $_ );
}
# signal no more work
$_->send( 0, undef ) for @chnls;
}
sub consumer {
my $chnl_num = MCE->task_wid - 1;
while () {
# notify the producer ready for input
$ready->send2( $chnl_num );
# retrieve input data
my ( $id, $item ) = $chnls[ $chnl_num ]->recv;
# leave loop if no more work
last unless $id;
# compute and send the result to the manager process
# ordered output requires an id (must be 1st argument)
MCE->gather( $id, [ $item, sqrt($item) ] );
}
}
# A custom 'ordered' output iterator for MCE's gather facility.
# It returns a closure block, expecting an ID for 1st argument.
sub output_iterator {
my %tmp; my $order_id = 1;
return sub {
my ( $id, $result ) = @_;
$tmp{ $id } = $result;
while () {
last unless exists $tmp{ $order_id };
$result = delete $tmp{ $order_id };
printf "n: %d sqrt(n): %f\n", $result->[0], $result->[1];
$order_id++;
}
};
}
# Run one producer and many consumers.
# Output to be sent orderly to STDOUT.
MCE::Flow->init(
gather => output_iterator(),
max_workers => [ 1, $consumers ],
);
MCE::Flow->run( \&producer, \&consumer );
MCE::Flow->finish;
__END__
# Output
n: 0 sqrt(n): 0.000000
n: 1 sqrt(n): 1.000000
n: 2 sqrt(n): 1.414214
n: 3 sqrt(n): 1.732051
n: 4 sqrt(n): 2.000000
n: 5 sqrt(n): 2.236068
n: 6 sqrt(n): 2.449490
n: 7 sqrt(n): 2.645751
n: 8 sqrt(n): 2.828427
n: 9 sqrt(n): 3.000000
...
SEE ALSO
o <https://github.com/marioroy/mce-examples/tree/master/chameneos>
o threads::lite(3)
AUTHOR
Mario E. Roy, <marioeroyA ATA gmailA DOTA com>
COPYRIGHT AND LICENSE
Copyright (C) 2019 by Mario E. Roy
MCE::Channel is released under the same license as Perl.
See <http://dev.perl.org/licenses/http://dev.perl.org/licenses/> for more information.
perl v5.28.2 2019-12-26 MCE::Channel(3)
mce 1.865.0 - Generated Sat Jan 4 06:47:12 CST 2020