MCE::Channel(3)

✨ MCE::Channel ✨

okay bestie, welcome to the MCE::Channel(3) page — yes, that's from the User Contributed Perl Documentation gang, and honestly? it's giving main character. it's a channel. it's queue-like. it's two-way. it's everything. let's go 💅

---

🌸 NAME 🌸

MCE::Channel — Queue-like and two-way communication capability

🎀 VERSION 🎀

This document describes MCE::Channel version 1.902

🦋 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 💖

so. a MCE::Channel object is literally just a container for sending and receiving data, and it does that using socketpair handles. slay. the (de)serialization situation is handled by Sereal if it's available, and defaults to Storable if it's not. and — this is the important bit — excluding the "Simple" implementation, both ends of the channel support many workers concurrently (with mp => 1). no cap.

🧚 new ( impl => STRING, mp => BOOLEAN )

this creates a new channel, bestie. there are three implementations provided — "Mutex", "Threads", and "Simple" — and they tell you which locking mechanism is in play: "MCE::Mutex", "threads::shared", and no locking at all, respectively. pick your fighter.

$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. iykyk.

$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'

now set the "mp" option — that's (m)any (p)roducers, cute right — to a true value if there are going to be two or more workers calling enqueue, send, recv2, or recv2_nb on the left end of the channel. this matters!! don't skip it!! otherwise you get a race condition and nobody wants that energy.

$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'

and then there's the "Simple" implementation, which is optimized for one producer and one consumer, max. it omits locking entirely for maximum performance. this one is preferred for parent-to-child communication that isn't shared by another worker. she's lowkey the fastest and she knows it. 💅

$chnl = MCE::Channel->new( impl => 'Simple' );

---

🌈 QUEUE-LIKE BEHAVIOR 🌈

🍓 enqueue ( ITEM1 [, ITEM2, ... ] )

appends a list of items onto the left end of the channel. heads up: this will block once the internal socket buffer becomes full (i.e. it's awaiting workers to dequeue on the other end). that's a feature, not a bug — it stops producer(s) from running faster than consumer(s). the vibes stay balanced. ✨

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). patience is a virtue bestie 🫶

$item  = $chnl->dequeue();       # item1
@items = $chnl->dequeue(2);      # ( item2, item3 )

🍓 dequeue_nb

🍓 dequeue_nb ( COUNT )

same deal — removes the requested number of items (default 1) from the right end of the channel. but this one? if the channel contains fewer than the requested number of items, it will return what it was able to retrieve and return immediately. and if the channel is empty, it returns "an empty list" in list context or "undef" in scalar context. no waiting, no drama.

$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. periodt.

⚠️ calling "end" by multiple producers is not supported. don't do it. i'm serious. 💀

$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. she's organized like that. ✨

$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 });

the fast channel implementations, introduced in MCE 1.877, support one item for "send". so if you want to pass multiple arguments, simply join the arguments into a string. that means the receiver will need to split the string. yes, it's a little DIY but it's giving efficiency 👑

$chnl = MCE::Channel->new(impl => "SimpleFast");

$chnl->send(join(" ", qw/item1 item2 item3/);
my ($item1, $item2, $item3) = split " ", $chnl->recv();

🎀 recv

🎀 recv_nb

these are the blocking and non-blocking fetch methods from the right end of the channel. for the latter, when the channel is empty, it 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. the symmetry? iconic. 🫶

$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 });

the fast channel implementations, introduced in MCE 1.877, support one item for "send2". if you want to pass multiple arguments, simply join the arguments into a string. and not to forget, the receiver must split the string as well. (same routine as before, bestie, just the other direction 💅)

$chnl = MCE::Channel->new(impl => "MutexFast");

$chnl->send2(join(" ", qw/item1 item2 item3/);
my ($item1, $item2, $item3) = split " ", $chnl->recv();

🌸 recv2

🌸 recv2_nb

the blocking and non-blocking fetch methods from the left end of the channel. for the latter, when the channel is empty, it 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 🌸

Bestie. Sit down. We're about to go through five whole entire demos and by the end you're gonna be the moment. 💅

---

✨ Example 1 - threads ✨

MCE::Channel was literally made to work efficiently with threads. The reason comes from using threads::shared for locking versus MCE::Mutex. iykyk. 🧚

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 🦋

This one is similarly threads-like but for Perl that's out here lacking threads support, no cap. Instead it spawns processes, so it requires the "Mutex" channel implementation — which is lowkey the default if you just... omit it. 🙈

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 🎀

Same energy as the last one, but now the manager process awaits a notification from the consumer before inserting into the queue. The glow-up here? The producer gets to end the channel early (i.e. exit the loop). Main character behaviour, honestly. 👑

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 means you gotta set the "mp" option. Internally that enables locking support for the left end of the channel. And heads up — the "mp" option applies to "Mutex" and "Threads" channel implementations only. Don't try it elsewhere, bestie. 💀

Here, it's 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 🌈

Okay this one ate. It 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. But — consumers don't incur reading and what's written is very small (i.e. atomic write is guaranteed by the OS). So, 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 🔥

• <https://github.com/marioroy/mce-examples/tree/master/chameneos> 💖
• threads::lite

---

👑 AUTHOR 👑

Mario E. Roy, <marioeroy AT gmail DOT com>

---

💖 COPYRIGHT AND LICENSE 💖

Copyright (C) 2019-2024 by Mario E. Roy

MCE::Channel is released under the same license as Perl.

See <https://dev.perl.org/licenses/> for more information.

---

And that's the whole page, bestie!! 🫶 This one shipped as perl v5.42.0 on 2025-09-10, and honestly? Iconic. Now go enqueue something. 🦋✨