1MCE::Channel(3) User Contributed Perl Documentation MCE::Channel(3)
2
6 MCE::Channel - Queue-like and two-way communication capability
7
9 This document describes MCE::Channel version 1.874
10
12 use MCE::Channel;
13 ########################
15 # Construction
16 ########################
17 # A single producer and many consumers supporting processes and threads
19 my $c1 = MCE::Channel->new( impl => 'Mutex' ); # default implementation
21 my $c2 = MCE::Channel->new( impl => 'Threads' ); # threads::shared locking
22 # Set the mp flag if two or more workers (many producers) will be calling
24 # enqueue/send or recv2/recv2_nb on the left end of the channel
25 my $c3 = MCE::Channel->new( impl => 'Mutex', mp => 1 );
27 my $c4 = MCE::Channel->new( impl => 'Threads', mp => 1 );
28 # Tuned for one producer and one consumer, no locking
30 my $c5 = MCE::Channel->new( impl => 'Simple' );
32 ########################
34 # Queue-like behavior
35 ########################
36 # Send data to consumers
38 $c1->enqueue('item');
39 $c1->enqueue(qw/item1 item2 item3 itemN/);
40 # Receive data
42 my $item = $c1->dequeue(); # item
43 my @items = $c1->dequeue(2); # (item1, item2)
44 # Receive, non-blocking
46 my $item = $c1->dequeue_nb(); # item
47 my @items = $c1->dequeue_nb(2); # (item1, item2)
48 # Signal that there is no more work to be sent
50 $c1->end();
51 ########################
53 # Two-way communication
54 ########################
55 # Producer(s) sending data
57 $c3->send('message');
58 $c3->send(qw/arg1 arg2 arg3/);
59 # Consumer(s) receiving data
61 my $mesg = $c3->recv(); # message
62 my @args = $c3->recv(); # (arg1, arg2, arg3)
63 # Alternatively, non-blocking
65 my $mesg = $c3->recv_nb(); # message
66 my @args = $c3->recv_nb(); # (arg1, arg2, arg3)
67 # A producer signaling no more work to be sent
69 $c3->end();
70 # Consumers(s) sending data
72 $c3->send2('message');
73 $c3->send2(qw/arg1 arg2 arg3/);
74 # Producer(s) receiving data
76 my $mesg = $c3->recv2(); # message
77 my @args = $c3->recv2(); # (arg1, arg2, arg3)
78 # Alternatively, non-blocking
80 my $mesg = $c3->recv2_nb(); # message
81 my @args = $c3->recv2_nb(); # (arg1, arg2, arg3)
82
84 A MCE::Channel object is a container for sending and receiving data
85 using socketpair handles. Serialization is provided by Sereal if
86 available. Defaults to Storable otherwise. Excluding the "Simple"
87 implementation, both ends of the "channel" support many workers
88 concurrently (with mp => 1).
89 new ( impl => STRING, mp => BOOLEAN )
91 This creates a new channel. Three implementations are provided "Mutex",
92 "Threads", and "Simple" indicating the locking mechanism to use
93 "MCE::Mutex", "threads::shared", and no locking respectively.
94 $chnl = MCE::Channel->new(); # default: impl => 'Mutex', mp => 0
96 # default: impl => 'Threads' on Windows
97 The "Mutex" implementation supports processes and threads whereas the
99 "Threads" implementation is suited for Windows and threads only.
100 $chnl = MCE::Channel->new( impl => 'Mutex' ); # MCE::Mutex locking
102 $chnl = MCE::Channel->new( impl => 'Threads' ); # threads::shared locking
103 # on Windows, silently becomes impl => 'Threads' when specifying 'Mutex'
105 Set the "mp" (m)any (p)roducers option to a true value if there will be
107 two or more workers calling "enqueue", <send>, "recv2", or "recv2_nb"
108 on the left end of the channel. This is important to not incur a race
109 condition.
110 $chnl = MCE::Channel->new( impl => 'Mutex', mp => 1 );
112 $chnl = MCE::Channel->new( impl => 'Threads', mp => 1 );
113 # on Windows, silently becomes impl => 'Threads' when specifying 'Mutex'
115 The "Simple" implementation is optimized for one producer and one
117 consumer max. It omits locking for maximum performance. This
118 implementation is preferred for parent to child communication not
119 shared by another worker.
120 $chnl = MCE::Channel->new( impl => 'Simple' );
122
124 enqueue ( ITEM1 [, ITEM2, ... ] )
125 Appends a list of items onto the left end of the channel. This will
126 block once the internal socket buffer becomes full (i.e. awaiting
127 workers to dequeue on the other end). This prevents producer(s) from
128 running faster than consumer(s).
129 Object (de)serialization is handled automatically using Sereal if
131 available or defaults to Storable otherwise.
132 $chnl->enqueue('item1');
134 $chnl->enqueue(qw/item2 item3 .../);
135 $chnl->enqueue([ array_ref1 ]);
137 $chnl->enqueue([ array_ref2 ], [ array_ref3 ], ...);
138 $chnl->enqueue({ hash_ref1 });
140 $chnl->enqueue({ hash_ref2 }, { hash_ref3 }, ...);
141 dequeue
143 dequeue ( COUNT )
144 Removes the requested number of items (default 1) from the right end of
145 the channel. If the channel contains fewer than the requested number of
146 items, the method will block (i.e. until other producer(s) enqueue more
147 items).
148 $item = $chnl->dequeue(); # item1
150 @items = $chnl->dequeue(2); # ( item2, item3 )
151 dequeue_nb
153 dequeue_nb ( COUNT )
154 Removes the requested number of items (default 1) from the right end of
155 the channel. If the channel contains fewer than the requested number of
156 items, the method will return what it was able to retrieve and return
157 immediately. If the channel is empty, then returns "an empty list" in
158 list context or "undef" in scalar context.
159 $item = $chnl->dequeue_nb(); # array_ref1
161 @items = $chnl->dequeue_nb(2); # ( array_ref2, array_ref3 )
162 end
164 This is called by a producer to signal that there is no more work to be
165 sent. Once ended, no more items may be sent by the producer. Calling
166 "end" by multiple producers is not supported.
167 $chnl->end;
169
171 send ( ARG1 [, ARG2, ... ] )
172 Append data onto the left end of the channel. Unlike "enqueue", the
173 values are kept together for the receiving consumer, similarly to
174 calling a method. Object (de)serialization is handled automatically.
175 $chnl->send('item');
177 $chnl->send([ list_ref ]);
178 $chnl->send([ hash_ref ]);
179 $chnl->send(qw/item1 item2 .../);
181 $chnl->send($id, [ list_ref ]);
182 $chnl->send($id, { hash_ref });
183 recv
185 recv_nb
186 Blocking and non-blocking fetch methods from the right end of the
187 channel. For the latter and when the channel is empty, returns "an
188 empty list" in list context or "undef" in scalar context.
189 $item = $chnl->recv();
191 $array_ref = $chnl->recv();
192 $hash_ref = $chnl->recv();
193 ($item1, $item2) = $chnl->recv_nb();
195 ($id, $array_ref) = $chnl->recv_nb();
196 ($id, $hash_ref) = $chnl->recv_nb();
197
199 send2 ( ARG1 [, ARG2, ... ] )
200 Append data onto the right end of the channel. Unlike "enqueue", the
201 values are kept together for the receiving producer, similarly to
202 calling a method. Object (de)serialization is handled automatically.
203 $chnl->send2('item');
205 $chnl->send2([ list_ref ]);
206 $chnl->send2([ hash_ref ]);
207 $chnl->send2(qw/item1 item2 .../);
209 $chnl->send2($id, [ list_ref ]);
210 $chnl->send2($id, { hash_ref });
211 recv2
213 recv2_nb
214 Blocking and non-blocking fetch methods from the left end of the
215 channel. For the latter and when the channel is empty, returns "an
216 empty list" in list context or "undef" in scalar context.
217 $item = $chnl->recv2();
219 $array_ref = $chnl->recv2();
220 $hash_ref = $chnl->recv2();
221 ($item1, $item2) = $chnl->recv2_nb();
223 ($id, $array_ref) = $chnl->recv2_nb();
224 ($id, $hash_ref) = $chnl->recv2_nb();
225
227 Example 1 - threads
228 "MCE::Channel" was made to work efficiently with threads. The reason is
229 from using threads::shared for locking versus MCE::Mutex.
230 use strict;
232 use warnings;
233 use threads;
235 use MCE::Channel;
236 my $queue = MCE::Channel->new( impl => 'Threads' );
238 my $num_consumers = 10;
239 sub consumer {
241 my $count = 0;
242 # receive items
244 while ( my ($item1, $item2) = $queue->dequeue(2) ) {
245 $count += 2;
246 }
247 # send result
249 $queue->send2( threads->tid => $count );
250 }
251 threads->create('consumer') for 1 .. $num_consumers;
253 ## producer
255 $queue->enqueue($_, $_ * 2) for 1 .. 40000;
257 $queue->end;
258 my %results;
260 my $total = 0;
261 for ( 1 .. $num_consumers ) {
263 my ($id, $count) = $queue->recv2;
264 $results{$id} = $count;
265 $total += $count;
266 }
267 $_->join for threads->list;
269 print $results{$_}, "\n" for keys %results;
271 print "$total total\n\n";
272 __END__
274 # output
276 8034
278 8008
279 8036
280 8058
281 7990
282 7948
283 8068
284 7966
285 7960
286 7932
287 80000 total
288 Example 2 - MCE::Child
290 The following is similarly threads-like for Perl lacking threads
291 support. It spawns processes instead, thus requires the "Mutex"
292 channel implementation which is the default if omitted.
293 use strict;
295 use warnings;
296 use MCE::Child;
298 use MCE::Channel;
299 my $queue = MCE::Channel->new( impl => 'Mutex' );
301 my $num_consumers = 10;
302 sub consumer {
304 my $count = 0;
305 # receive items
307 while ( my ($item1, $item2) = $queue->dequeue(2) ) {
308 $count += 2;
309 }
310 # send result
312 $queue->send2( MCE::Child->pid => $count );
313 }
314 MCE::Child->create('consumer') for 1 .. $num_consumers;
316 ## producer
318 $queue->enqueue($_, $_ * 2) for 1 .. 40000;
320 $queue->end;
321 my %results;
323 my $total = 0;
324 for ( 1 .. $num_consumers ) {
326 my ($id, $count) = $queue->recv2;
327 $results{$id} = $count;
328 $total += $count;
329 }
330 $_->join for MCE::Child->list;
332 print $results{$_}, "\n" for keys %results;
334 print "$total total\n\n";
335 Example 3 - Consumer requests item
337 Like the previous example, but have the manager process await a
338 notification from the consumer before inserting into the queue. This
339 allows the producer to end the channel early (i.e. exit loop).
340 use strict;
342 use warnings;
343 use MCE::Child;
345 use MCE::Channel;
346 my $queue = MCE::Channel->new( impl => 'Mutex' );
348 my $num_consumers = 10;
349 sub consumer {
351 # receive items
352 my $count = 0;
353 while () {
355 # Notify the manager process to send items. This allows the
356 # manager process to enqueue only when requested. The benefit
357 # is being able to end the channel immediately.
358 $queue->send2( MCE::Child->pid ); # channel is bi-directional
360 my ($item1, $item2) = $queue->dequeue(2);
362 last unless ( defined $item1 ); # channel ended
363 $count += 2;
365 }
366 # result
368 return ( MCE::Child->pid => $count );
369 }
370 MCE::Child->create('consumer') for 1 .. $num_consumers;
372 ## producer
374 for my $num (1 .. 40000) {
376 # Await worker notification before inserting (blocking).
377 my $consumer_pid = $queue->recv2;
378 $queue->enqueue($num, $num * 2);
379 }
380 $queue->end;
382 my %results;
384 my $total = 0;
385 for my $child ( MCE::Child->list ) {
387 my ($id, $count) = $child->join;
388 $results{$id} = $count;
389 $total += $count;
390 }
391 print $results{$_}, "\n" for keys %results;
393 print "$total total\n\n";
394 Example 4 - Many producers
396 Running with 2 or more producers requires setting the "mp" option.
397 Internally, this enables locking support for the left end of the
398 channel. The "mp" option applies to "Mutex" and "Threads" channel
399 implementations only.
400 Here, using the MCE facility for gathering the final count.
402 use strict;
404 use warnings;
405 use MCE::Flow;
407 use MCE::Channel;
408 my $queue = MCE::Channel->new( impl => 'Mutex', mp => 1 );
410 my $num_consumers = 10;
411 sub consumer {
413 # receive items
414 my $count = 0;
415 while ( my ( $item1, $item2 ) = $queue->dequeue(2) ) {
416 $count += 2;
417 }
418 # send result
419 MCE->gather( MCE->wid => $count );
420 }
421 sub producer {
423 $queue->enqueue($_, $_ * 2) for 1 .. 20000;
424 }
425 ## run 2 producers and many consumers
427 MCE::Flow->init(
429 max_workers => [ 2, $num_consumers ],
430 task_name => [ 'producer', 'consumer' ],
431 task_end => sub {
432 my ($mce, $task_id, $task_name) = @_;
433 if ( $task_name eq 'producer' ) {
434 $queue->end;
435 }
436 }
437 );
438 # consumers call gather above (i.e. send a key-value pair),
440 # have MCE append to a hash
441 my %results = mce_flow \&producer, \&consumer;
443 MCE::Flow->finish;
445 my $total = 0;
447 for ( keys %results ) {
449 $total += $results{$_};
450 print $results{$_}, "\n";
451 }
452 print "$total total\n\n";
454 Example 5 - Many channels
456 This demonstration configures a channel per consumer. Plus, a common
457 channel for consumers to request the next input item. The "Simple"
458 implementation is specified for the individual channels whereas locking
459 may be necessary for the $ready channel. However, consumers do not
460 incur reading and what is written is very small (i.e. atomic write is
461 guaranteed by the OS). Thus, am safely choosing the "Simple"
462 implementation versus "Mutex".
463 use strict;
465 use warnings;
466 use MCE::Flow;
468 use MCE::Channel;
469 my $prog_name = $0; $prog_name =~ s{^.*[\\/]}{}g;
471 my $input_size = shift || 3000;
472 unless ($input_size =~ /\A\d+\z/) {
474 print {*STDERR} "usage: $prog_name [ size ]\n";
475 exit 1;
476 }
477 my $consumers = 4;
479 my @chnls = map { MCE::Channel->new( impl => 'Simple' ) } 1 .. $consumers;
481 my $ready = MCE::Channel->new( impl => 'Simple' );
483 sub producer {
485 my $id = 0;
486 # send the next input item upon request
488 for ( 0 .. $input_size - 1 ) {
489 my $chnl_num = $ready->recv2;
490 $chnls[ $chnl_num ]->send( ++$id, $_ );
491 }
492 # signal no more work
494 $_->send( 0, undef ) for @chnls;
495 }
496 sub consumer {
498 my $chnl_num = MCE->task_wid - 1;
499 while () {
501 # notify the producer ready for input
502 $ready->send2( $chnl_num );
503 # retrieve input data
505 my ( $id, $item ) = $chnls[ $chnl_num ]->recv;
506 # leave loop if no more work
508 last unless $id;
509 # compute and send the result to the manager process
511 # ordered output requires an id (must be 1st argument)
512 MCE->gather( $id, [ $item, sqrt($item) ] );
513 }
514 }
515 # A custom 'ordered' output iterator for MCE's gather facility.
517 # It returns a closure block, expecting an ID for 1st argument.
518 sub output_iterator {
520 my %tmp; my $order_id = 1;
521 return sub {
523 my ( $id, $result ) = @_;
524 $tmp{ $id } = $result;
525 while () {
527 last unless exists $tmp{ $order_id };
528 $result = delete $tmp{ $order_id };
529 printf "n: %d sqrt(n): %f\n", $result->[0], $result->[1];
530 $order_id++;
531 }
532 };
533 }
534 # Run one producer and many consumers.
536 # Output to be sent orderly to STDOUT.
537 MCE::Flow->init(
539 gather => output_iterator(),
540 max_workers => [ 1, $consumers ],
541 );
542 MCE::Flow->run( \&producer, \&consumer );
544 MCE::Flow->finish;
545 __END__
547 # Output
549 n: 0 sqrt(n): 0.000000
551 n: 1 sqrt(n): 1.000000
552 n: 2 sqrt(n): 1.414214
553 n: 3 sqrt(n): 1.732051
554 n: 4 sqrt(n): 2.000000
555 n: 5 sqrt(n): 2.236068
556 n: 6 sqrt(n): 2.449490
557 n: 7 sqrt(n): 2.645751
558 n: 8 sqrt(n): 2.828427
559 n: 9 sqrt(n): 3.000000
560 ...
561
563 · <https://github.com/marioroy/mce-examples/tree/master/chameneos>
564 · threads::lite
566
568 Mario E. Roy, <marioeroy AT gmail DOT com>
569
571 Copyright (C) 2019-2020 by Mario E. Roy
572 MCE::Channel is released under the same license as Perl.
574 See <http://dev.perl.org/licenses/> for more information.
576perl v5.32.0 2020-08-19 MCE::Channel(3)