MCE::Queue->new([queue=>\@array,await=>1,fast=>1])
This creates a new queue. Available options are queue, porder, type, await, and gather. Note: The barrier
and fast options are silentently ignored (no-op) if specified; starting with 1.867.
use MCE;
use MCE::Queue;
my $q1 = MCE::Queue->new();
my $q2 = MCE::Queue->new( queue => [ 0, 1, 2 ] );
my $q3 = MCE::Queue->new( porder => $MCE::Queue::HIGHEST );
my $q4 = MCE::Queue->new( porder => $MCE::Queue::LOWEST );
my $q5 = MCE::Queue->new( type => $MCE::Queue::FIFO );
my $q6 = MCE::Queue->new( type => $MCE::Queue::LIFO );
my $q7 = MCE::Queue->new( await => 1, barrier => 0 );
my $q8 = MCE::Queue->new( fast => 1 );
The "await" option, when enabled, allows workers to block (semaphore-like) until the number of items
pending is equal to or less than a threshold value. The $q->await method is described below.
Obsolete: On Unix platforms, "barrier" mode (enabled by default) prevents many workers from dequeuing
simultaneously to lessen overhead for the OS kernel. Specify 0 to disable barrier mode and not allocate
sockets. The barrier option has no effect if constructing the queue inside a thread or enabling "fast".
Obsolete: The "fast" option speeds up dequeues and is not enabled by default. It is beneficial for
queues not calling (->dequeue_nb) and not altering the count value while running; e.g. ->dequeue($count).
The "gather" option is mainly for running with MCE and wanting to pass item(s) to a callback function for
appending to the queue. Multiple queues may point to the same callback function. The callback receives
the queue object as the first argument and items after it.
sub _append {
my ($q, @items) = @_;
$q->enqueue(@items);
}
my $q7 = MCE::Queue->new( gather => \&_append );
my $q8 = MCE::Queue->new( gather => \&_append );
## Items are diverted to the callback function, not the queue.
$q7->enqueue( 'apple', 'orange' );
Specifying the "gather" option allows one to store items temporarily while ensuring output order.
Although a queue object is not required, this is simply a demonstration of the gather option in the
context of a queue.
use MCE;
use MCE::Queue;
sub preserve_order {
my %tmp; my $order_id = 1;
return sub {
my ($q, $chunk_id, $data) = @_;
$tmp{$chunk_id} = $data;
while (1) {
last unless exists $tmp{$order_id};
$q->enqueue( delete $tmp{$order_id++} );
}
return;
};
}
my @squares; my $q = MCE::Queue->new(
queue => \@squares, gather => preserve_order
);
my $mce = MCE->new(
chunk_size => 1, input_data => [ 1 .. 100 ],
user_func => sub {
$q->enqueue( MCE->chunk_id, $_ * $_ );
}
);
$mce->run;
print "@squares\n";
$q->await($pending_threshold)
The await method is beneficial when wanting to throttle worker(s) appending to the queue. Perhaps,
consumers are running a bit behind and wanting to keep tabs on memory consumption. Below, the number of
items pending will never go above 20.
use Time::HiRes qw( sleep );
use MCE::Flow;
use MCE::Queue;
my $q = MCE::Queue->new( await => 1, fast => 1 );
my ( $producers, $consumers ) = ( 1, 8 );
mce_flow {
task_name => [ 'producer', 'consumer' ],
max_workers => [ $producers, $consumers ],
},
sub {
## producer
for my $item ( 1 .. 100 ) {
$q->enqueue($item);
## blocks until the # of items pending reaches <= 10
if ($item % 10 == 0) {
MCE->say( 'pending: '.$q->pending() );
$q->await(10);
}
}
## notify consumers no more work
$q->end();
},
sub {
## consumers
while (defined (my $next = $q->dequeue())) {
MCE->say( MCE->task_wid().': '.$next );
sleep 0.100;
}
};
$q->clear(void)
Clears the queue of any items. This has the effect of nulling the queue and the socket used for blocking.
my @a; my $q = MCE::Queue->new( queue => \@a );
@a = (); ## bad, the blocking socket may become out of sync
$q->clear; ## ok
$q->end(void)
Stops the queue from receiving more items. Any worker blocking on "dequeue" will be unblocked
automatically. Subsequent calls to "dequeue" will behave like "dequeue_nb". Current API available since
MCE 1.818.
$q->end();
MCE Models (e.g. MCE::Flow) may persist between runs. In that case, one might want to enqueue "undef"'s
versus calling "end". The number of "undef"'s depends on how many items workers dequeue at a time.
$q->enqueue((undef) x ($N_workers * 1)); # $q->dequeue() 1 item
$q->enqueue((undef) x ($N_workers * 2)); # $q->dequeue(2) 2 items
$q->enqueue((undef) x ($N_workers * N)); # $q->dequeue(N) N items
$q->enqueue($item[,$item,...])
Appends a list of items onto the end of the normal queue.
$q->enqueue( 'foo' );
$q->enqueue( 'bar', 'baz' );
$q->enqueuep($p,$item[,$item,...])
Appends a list of items onto the end of the priority queue with priority.
$q->enqueue( $priority, 'foo' );
$q->enqueue( $priority, 'bar', 'baz' );
$q->dequeue([$count])
Returns the requested number of items (default 1) from the queue. Priority data will always dequeue first
before any data from the normal queue.
$q->dequeue;
$q->dequeue( 2 );
The method will block if the queue contains zero items. If the queue contains fewer than the requested
number of items, the method will not block, but return whatever items there are on the queue.
The $count, used for requesting the number of items, is beneficial when workers are passing parameters
through the queue. For this reason, always remember to dequeue using the same multiple for the count.
This is unlike Thread::Queue which will block until the requested number of items are available.
# MCE::Queue 1.820 and prior releases
while ( my @items = $q->dequeue(2) ) {
last unless ( defined $items[0] );
...
}
# MCE::Queue 1.821 and later
while ( my @items = $q->dequeue(2) ) {
...
}
$q->dequeue_nb([$count])
Returns the requested number of items (default 1) from the queue. Like with dequeue, priority data will
always dequeue first. This method is non-blocking and returns "undef" in the absence of data.
$q->dequeue_nb;
$q->dequeue_nb( 2 );
$q->dequeue_timed(timeout[,$count])
Returns the requested number of items (default 1) from the queue. Like with dequeue, priority data will
always dequeue first. This method is blocking until the timeout is reached and returns "undef" in the
absence of data. Current API available since MCE 1.886.
$q->dequeue_timed( 300 ); # timeout after 5 minutes
$q->dequeue_timed( 300, 2 );
The timeout may be specified as fractional seconds. If timeout is missing, undef, less than or equal to
0, or called by the manager process, then this call behaves like dequeue_nb.
$q->insert($index,$item[,$item,...])
Adds the list of items to the queue at the specified index position (0 is the head of the list). The head
of the queue is that item which would be removed by a call to dequeue.
$q = MCE::Queue->new( type => $MCE::Queue::FIFO );
$q->enqueue(1, 2, 3, 4);
$q->insert(1, 'foo', 'bar');
# Queue now contains: 1, foo, bar, 2, 3, 4
$q = MCE::Queue->new( type => $MCE::Queue::LIFO );
$q->enqueue(1, 2, 3, 4);
$q->insert(1, 'foo', 'bar');
# Queue now contains: 1, 2, 3, 'foo', 'bar', 4
$q->insertp($p,$index,$item[,$item,...])
Adds the list of items to the queue at the specified index position with priority. The behavior is
similarly to "$q->insert" otherwise.
$q->pending(void)
Returns the number of items in the queue. The count includes both normal and priority data. Returns
"undef" if the queue has been ended, and there are no more items in the queue.
$q = MCE::Queue->new();
$q->enqueuep(5, 'foo', 'bar');
$q->enqueue('sunny', 'day');
print $q->pending(), "\n";
# Output: 4
$q->peek([$index])
Returns an item from the normal queue, at the specified index, without dequeuing anything. It defaults to
the head of the queue if index is not specified. The head of the queue is that item which would be
removed by a call to dequeue. Negative index values are supported, similarly to arrays.
$q = MCE::Queue->new( type => $MCE::Queue::FIFO );
$q->enqueue(1, 2, 3, 4, 5);
print $q->peek(1), ' ', $q->peek(-2), "\n";
# Output: 2 4
$q = MCE::Queue->new( type => $MCE::Queue::LIFO );
$q->enqueue(1, 2, 3, 4, 5);
print $q->peek(1), ' ', $q->peek(-2), "\n";
# Output: 4 2
$q->peekp($p[,$index])
Returns an item from the queue with priority, at the specified index, without dequeuing anything. It
defaults to the head of the queue if index is not specified. The behavior is similarly to "$q->peek"
otherwise.
$q->peekh([$index])
Returns an item from the head of the heap or at the specified index.
$q = MCE::Queue->new( porder => $MCE::Queue::HIGHEST );
$q->enqueuep(5, 'foo');
$q->enqueuep(6, 'bar');
$q->enqueuep(4, 'sun');
print $q->peekh(0), "\n";
# Output: 6
$q = MCE::Queue->new( porder => $MCE::Queue::LOWEST );
$q->enqueuep(5, 'foo');
$q->enqueuep(6, 'bar');
$q->enqueuep(4, 'sun');
print $q->peekh(0), "\n";
# Output: 4
$q->heap(void)
Returns an array containing the heap data. Heap data consists of priority numbers, not the data.
@h = $q->heap; # $MCE::Queue::HIGHEST
# Heap contains: 6, 5, 4
@h = $q->heap; # $MCE::Queue::LOWEST
# Heap contains: 4, 5, 6
Install Now
PNG Icons
