.\" Automatically generated by Pod::Man 4.14 (Pod::Simple 3.43)
.\"
.\" Standard preamble:
.\" ========================================================================
.de Sp \" Vertical space (when we can't use .PP)
.if t .sp .5v
.if n .sp
..
.de Vb \" Begin verbatim text
.ft CW
.nf
.ne \\$1
..
.de Ve \" End verbatim text
.ft R
.fi
..
.\" Set up some character translations and predefined strings.  \*(-- will
.\" give an unbreakable dash, \*(PI will give pi, \*(L" will give a left
.\" double quote, and \*(R" will give a right double quote.  \*(C+ will
.\" give a nicer C++.  Capital omega is used to do unbreakable dashes and
.\" therefore won't be available.  \*(C` and \*(C' expand to `' in nroff,
.\" nothing in troff, for use with C<>.
.tr \(*W-
.ds C+ C\v'-.1v'\h'-1p'\s-2+\h'-1p'+\s0\v'.1v'\h'-1p'
.ie n \{\
.    ds -- \(*W-
.    ds PI pi
.    if (\n(.H=4u)&(1m=24u) .ds -- \(*W\h'-12u'\(*W\h'-12u'-\" diablo 10 pitch
.    if (\n(.H=4u)&(1m=20u) .ds -- \(*W\h'-12u'\(*W\h'-8u'-\"  diablo 12 pitch
.    ds L" ""
.    ds R" ""
.    ds C` ""
.    ds C' ""
'br\}
.el\{\
.    ds -- \|\(em\|
.    ds PI \(*p
.    ds L" ``
.    ds R" ''
.    ds C`
.    ds C'
'br\}
.\"
.\" Escape single quotes in literal strings from groff's Unicode transform.
.ie \n(.g .ds Aq \(aq
.el       .ds Aq '
.\"
.\" If the F register is >0, we'll generate index entries on stderr for
.\" titles (.TH), headers (.SH), subsections (.SS), items (.Ip), and index
.\" entries marked with X<> in POD.  Of course, you'll have to process the
.\" output yourself in some meaningful fashion.
.\"
.\" Avoid warning from groff about undefined register 'F'.
.de IX
..
.nr rF 0
.if \n(.g .if rF .nr rF 1
.if (\n(rF:(\n(.g==0)) \{\
.    if \nF \{\
.        de IX
.        tm Index:\\$1\t\\n%\t"\\$2"
..
.        if !\nF==2 \{\
.            nr % 0
.            nr F 2
.        \}
.    \}
.\}
.rr rF
.\" ========================================================================
.\"
.IX Title "MCE::Channel 3pm"
.TH MCE::Channel 3pm "2023-09-29" "perl v5.36.0" "User Contributed Perl Documentation"
.\" For nroff, turn off justification.  Always turn off hyphenation; it makes
.\" way too many mistakes in technical documents.
.if n .ad l
.nh
.SH "NAME"
MCE::Channel \- Queue\-like and two\-way communication capability
.SH "VERSION"
.IX Header "VERSION"
This document describes MCE::Channel version 1.889
.SH "SYNOPSIS"
.IX Header "SYNOPSIS"
.Vb 1
\& use MCE::Channel;
\&
\& ########################
\& # Construction
\& ########################
\&
\& # A single producer and many consumers supporting processes and threads
\&
\& my $c1 = MCE::Channel\->new( impl => \*(AqMutex\*(Aq );    # default implementation
\& my $c2 = MCE::Channel\->new( impl => \*(AqThreads\*(Aq );  # 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 => \*(AqMutex\*(Aq, mp => 1 );
\& my $c4 = MCE::Channel\->new( impl => \*(AqThreads\*(Aq, mp => 1 );
\&
\& # Tuned for one producer and one consumer, no locking
\&
\& my $c5 = MCE::Channel\->new( impl => \*(AqSimple\*(Aq );
\&
\& ########################
\& # Queue\-like behavior
\& ########################
\&
\& # Send data to consumers
\& $c1\->enqueue(\*(Aqitem\*(Aq);
\& $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(\*(Aqmessage\*(Aq);
\& $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(\*(Aqmessage\*(Aq);
\& $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)
.Ve
.SH "DESCRIPTION"
.IX Header "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 \f(CW\*(C`Simple\*(C'\fR implementation,
both ends of the \f(CW\*(C`channel\*(C'\fR support many workers concurrently (with mp => 1).
.SS "new ( impl => \s-1STRING,\s0 mp => \s-1BOOLEAN\s0 )"
.IX Subsection "new ( impl => STRING, mp => BOOLEAN )"
This creates a new channel. Three implementations are provided \f(CW\*(C`Mutex\*(C'\fR,
\&\f(CW\*(C`Threads\*(C'\fR, and \f(CW\*(C`Simple\*(C'\fR indicating the locking mechanism to use
\&\f(CW\*(C`MCE::Mutex\*(C'\fR, \f(CW\*(C`threads::shared\*(C'\fR, and no locking respectively.
.PP
.Vb 2
\& $chnl = MCE::Channel\->new();     # default: impl => \*(AqMutex\*(Aq, mp => 0
\&                                  # default: impl => \*(AqThreads\*(Aq on Windows
.Ve
.PP
The \f(CW\*(C`Mutex\*(C'\fR implementation supports processes and threads whereas the
\&\f(CW\*(C`Threads\*(C'\fR implementation is suited for Windows and threads only.
.PP
.Vb 2
\& $chnl = MCE::Channel\->new( impl => \*(AqMutex\*(Aq );    # MCE::Mutex locking
\& $chnl = MCE::Channel\->new( impl => \*(AqThreads\*(Aq );  # threads::shared locking
\&
\& # on Windows, silently becomes impl => \*(AqThreads\*(Aq when specifying \*(AqMutex\*(Aq
.Ve
.PP
Set the \f(CW\*(C`mp\*(C'\fR (m)any (p)roducers option to a true value if there will be two
or more workers calling \f(CW\*(C`enqueue\*(C'\fR, <send>, \f(CW\*(C`recv2\*(C'\fR, or \f(CW\*(C`recv2_nb\*(C'\fR on the
left end of the channel. This is important to not incur a race condition.
.PP
.Vb 2
\& $chnl = MCE::Channel\->new( impl => \*(AqMutex\*(Aq, mp => 1 );
\& $chnl = MCE::Channel\->new( impl => \*(AqThreads\*(Aq, mp => 1 );
\&
\& # on Windows, silently becomes impl => \*(AqThreads\*(Aq when specifying \*(AqMutex\*(Aq
.Ve
.PP
The \f(CW\*(C`Simple\*(C'\fR 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.
.PP
.Vb 1
\& $chnl = MCE::Channel\->new( impl => \*(AqSimple\*(Aq );
.Ve
.SH "QUEUE-LIKE BEHAVIOR"
.IX Header "QUEUE-LIKE BEHAVIOR"
.SS "enqueue ( \s-1ITEM1\s0 [, \s-1ITEM2, ...\s0 ] )"
.IX Subsection "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).
.PP
Object (de)serialization is handled automatically using Sereal if available
or defaults to Storable otherwise.
.PP
.Vb 2
\& $chnl\->enqueue(\*(Aqitem1\*(Aq);
\& $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 }, ...);
.Ve
.SS "dequeue"
.IX Subsection "dequeue"
.SS "dequeue ( \s-1COUNT\s0 )"
.IX Subsection "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).
.PP
.Vb 2
\& $item  = $chnl\->dequeue();       # item1
\& @items = $chnl\->dequeue(2);      # ( item2, item3 )
.Ve
.SS "dequeue_nb"
.IX Subsection "dequeue_nb"
.SS "dequeue_nb ( \s-1COUNT\s0 )"
.IX Subsection "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 \f(CW\*(C`an empty list\*(C'\fR in list context or
\&\f(CW\*(C`undef\*(C'\fR in scalar context.
.PP
.Vb 2
\& $item  = $chnl\->dequeue_nb();    # array_ref1
\& @items = $chnl\->dequeue_nb(2);   # ( array_ref2, array_ref3 )
.Ve
.SS "end"
.IX Subsection "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 \f(CW\*(C`end\*(C'\fR by
multiple producers is not supported.
.PP
.Vb 1
\& $chnl\->end;
.Ve
.SH "TWO-WAY IPC \- PRODUCER TO CONSUMER"
.IX Header "TWO-WAY IPC - PRODUCER TO CONSUMER"
.SS "send ( \s-1ARG1\s0 [, \s-1ARG2, ...\s0 ] )"
.IX Subsection "send ( ARG1 [, ARG2, ... ] )"
Append data onto the left end of the channel. Unlike \f(CW\*(C`enqueue\*(C'\fR, the values
are kept together for the receiving consumer, similarly to calling a method.
Object (de)serialization is handled automatically.
.PP
.Vb 3
\& $chnl\->send(\*(Aqitem\*(Aq);
\& $chnl\->send([ list_ref ]);
\& $chnl\->send([ hash_ref ]);
\&
\& $chnl\->send(qw/item1 item2 .../);
\& $chnl\->send($id, [ list_ref ]);
\& $chnl\->send($id, { hash_ref });
.Ve
.PP
The fast channel implementations, introduced in \s-1MCE 1.877,\s0 support one item
for \f(CW\*(C`send\*(C'\fR. If you want to pass multiple arguments, simply join the arguments
into a string. That means the receiver will need to split the string.
.PP
.Vb 1
\& $chnl = MCE::Channel\->new(impl => "SimpleFast");
\&
\& $chnl\->send(join(" ", qw/item1 item2 item3/);
\& my ($item1, $item2, $item3) = split " ", $chnl\->recv();
.Ve
.SS "recv"
.IX Subsection "recv"
.SS "recv_nb"
.IX Subsection "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 \f(CW\*(C`an empty list\*(C'\fR in
list context or \f(CW\*(C`undef\*(C'\fR in scalar context.
.PP
.Vb 3
\& $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();
.Ve
.SH "TWO-WAY IPC \- CONSUMER TO PRODUCER"
.IX Header "TWO-WAY IPC - CONSUMER TO PRODUCER"
.SS "send2 ( \s-1ARG1\s0 [, \s-1ARG2, ...\s0 ] )"
.IX Subsection "send2 ( ARG1 [, ARG2, ... ] )"
Append data onto the right end of the channel. Unlike \f(CW\*(C`enqueue\*(C'\fR, the values
are kept together for the receiving producer, similarly to calling a method.
Object (de)serialization is handled automatically.
.PP
.Vb 3
\& $chnl\->send2(\*(Aqitem\*(Aq);
\& $chnl\->send2([ list_ref ]);
\& $chnl\->send2([ hash_ref ]);
\&
\& $chnl\->send2(qw/item1 item2 .../);
\& $chnl\->send2($id, [ list_ref ]);
\& $chnl\->send2($id, { hash_ref });
.Ve
.PP
The fast channel implementations, introduced in \s-1MCE 1.877,\s0 support one item
for \f(CW\*(C`send2\*(C'\fR. If you want to pass multiple arguments, simply join the arguments
into a string. Not to forget, the receiver must split the string as well.
.PP
.Vb 1
\& $chnl = MCE::Channel\->new(impl => "MutexFast");
\&
\& $chnl\->send2(join(" ", qw/item1 item2 item3/);
\& my ($item1, $item2, $item3) = split " ", $chnl\->recv();
.Ve
.SS "recv2"
.IX Subsection "recv2"
.SS "recv2_nb"
.IX Subsection "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 \f(CW\*(C`an empty list\*(C'\fR in
list context or \f(CW\*(C`undef\*(C'\fR in scalar context.
.PP
.Vb 3
\& $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();
.Ve
.SH "DEMONSTRATIONS"
.IX Header "DEMONSTRATIONS"
.SS "Example 1 \- threads"
.IX Subsection "Example 1 - threads"
\&\f(CW\*(C`MCE::Channel\*(C'\fR was made to work efficiently with threads. The reason
comes from using threads::shared for locking versus MCE::Mutex.
.PP
.Vb 2
\& use strict;
\& use warnings;
\&
\& use threads;
\& use MCE::Channel;
\&
\& my $queue = MCE::Channel\->new( impl => \*(AqThreads\*(Aq );
\& 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(\*(Aqconsumer\*(Aq) 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{$_}, "\en" for keys %results;
\& print "$total total\en\en";
\&
\& _\|_END_\|_
\&
\& # output
\&
\& 8034
\& 8008
\& 8036
\& 8058
\& 7990
\& 7948
\& 8068
\& 7966
\& 7960
\& 7932
\& 80000 total
.Ve
.SS "Example 2 \- MCE::Child"
.IX Subsection "Example 2 - MCE::Child"
The following is similarly threads-like for Perl lacking threads support.
It spawns processes instead, thus requires the \f(CW\*(C`Mutex\*(C'\fR channel implementation
which is the default if omitted.
.PP
.Vb 2
\& use strict;
\& use warnings;
\&
\& use MCE::Child;
\& use MCE::Channel;
\&
\& my $queue = MCE::Channel\->new( impl => \*(AqMutex\*(Aq );
\& 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(\*(Aqconsumer\*(Aq) 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{$_}, "\en" for keys %results;
\& print "$total total\en\en";
.Ve
.SS "Example 3 \- Consumer requests item"
.IX Subsection "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).
.PP
.Vb 2
\& use strict;
\& use warnings;
\&
\& use MCE::Child;
\& use MCE::Channel;
\&
\& my $queue = MCE::Channel\->new( impl => \*(AqMutex\*(Aq );
\& 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(\*(Aqconsumer\*(Aq) 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{$_}, "\en" for keys %results;
\& print "$total total\en\en";
.Ve
.SS "Example 4 \- Many producers"
.IX Subsection "Example 4 - Many producers"
Running with 2 or more producers requires setting the \f(CW\*(C`mp\*(C'\fR option. Internally,
this enables locking support for the left end of the channel. The \f(CW\*(C`mp\*(C'\fR option
applies to \f(CW\*(C`Mutex\*(C'\fR and \f(CW\*(C`Threads\*(C'\fR channel implementations only.
.PP
Here, using the \s-1MCE\s0 facility for gathering the final count.
.PP
.Vb 2
\& use strict;
\& use warnings;
\&
\& use MCE::Flow;
\& use MCE::Channel;
\&
\& my $queue = MCE::Channel\->new( impl => \*(AqMutex\*(Aq, 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   => [ \*(Aqproducer\*(Aq, \*(Aqconsumer\*(Aq ],
\&    task_end    => sub {
\&       my ($mce, $task_id, $task_name) = @_;
\&       if ( $task_name eq \*(Aqproducer\*(Aq ) {
\&          $queue\->end;
\&       }
\&    }
\& );
\&
\& # consumers call gather above (i.e. send a key\-value pair),
\& # have MCE append to a hash
\&
\& my %results = mce_flow \e&producer, \e&consumer;
\&
\& MCE::Flow\->finish;
\&
\& my $total = 0;
\&
\& for ( keys %results ) {
\&    $total += $results{$_};
\&    print $results{$_}, "\en";
\& }
\&
\& print "$total total\en\en";
.Ve
.SS "Example 5 \- Many channels"
.IX Subsection "Example 5 - Many channels"
This demonstration configures a channel per consumer. Plus, a common channel
for consumers to request the next input item. The \f(CW\*(C`Simple\*(C'\fR implementation is
specified for the individual channels whereas locking may be necessary for the
\&\f(CW$ready\fR channel. However, consumers do not incur reading and what is written
is very small (i.e. atomic write is guaranteed by the \s-1OS\s0). Thus, am safely
choosing the \f(CW\*(C`Simple\*(C'\fR implementation versus \f(CW\*(C`Mutex\*(C'\fR.
.PP
.Vb 2
\& use strict;
\& use warnings;
\&
\& use MCE::Flow;
\& use MCE::Channel;
\&
\& my $prog_name  = $0; $prog_name =~ s{^.*[\e\e/]}{}g;
\& my $input_size = shift || 3000;
\&
\& unless ($input_size =~ /\eA\ed+\ez/) {
\&    print {*STDERR} "usage: $prog_name [ size ]\en";
\&    exit 1;
\& }
\&
\& my $consumers = 4;
\&
\& my @chnls = map { MCE::Channel\->new( impl => \*(AqSimple\*(Aq ) } 1 .. $consumers;
\&
\& my $ready =       MCE::Channel\->new( impl => \*(AqSimple\*(Aq );
\&
\& 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 \*(Aqordered\*(Aq output iterator for MCE\*(Aqs 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\en", $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( \e&producer, \e&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
\& ...
.Ve
.SH "SEE ALSO"
.IX Header "SEE ALSO"
.IP "\(bu" 3
<https://github.com/marioroy/mce\-examples/tree/master/chameneos>
.IP "\(bu" 3
threads::lite
.SH "AUTHOR"
.IX Header "AUTHOR"
Mario E. Roy, <marioeroy AT gmail DOT com>
.SH "COPYRIGHT AND LICENSE"
.IX Header "COPYRIGHT AND LICENSE"
Copyright (C) 2019\-2023 by Mario E. Roy
.PP
MCE::Channel is released under the same license as Perl.
.PP
See <https://dev.perl.org/licenses/> for more information.