Provided by: libmce-perl_1.608-1_all 
NAME
MCE::Queue - Hybrid (normal and priority) queues for Many-Core Engine
VERSION
This document describes MCE::Queue version 1.608
SYNOPSIS
use MCE;
use MCE::Queue;
my $F = MCE::Queue->new(fast => 1);
my $consumers = 8;
my $mce = MCE->new(
task_end => sub {
my ($mce, $task_id, $task_name) = @_;
$F->enqueue((undef) x $consumers)
if $task_name eq 'dir';
},
user_tasks => [{
max_workers => 1, task_name => 'dir',
user_func => sub {
## Create a "standalone queue" only accessible to this worker.
## See included examples for running with multiple workers.
my $D = MCE::Queue->new(queue => [ MCE->user_args->[0] ]);
while (defined (my $dir = $D->dequeue_nb)) {
my (@files, @dirs); foreach (glob("$dir/*")) {
if (-d $_) { push @dirs, $_; next; }
push @files, $_;
}
$D->enqueue(@dirs ) if scalar @dirs;
$F->enqueue(@files) if scalar @files;
}
}
},{
max_workers => $consumers, task_name => 'file',
user_func => sub {
while (defined (my $file = $F->dequeue)) {
MCE->say($file);
}
}
}]
)->run({ user_args => [ $ARGV[0] || '.' ] });
__END__
Results from files_mce.pl and files_thr.pl; included with MCE.
Usage:
time ./files_mce.pl /usr 0 | wc -l
time ./files_mce.pl /usr 1 | wc -l
time ./files_thr.pl /usr | wc -l
Darwin (OS) /usr: 216,271 files
MCE::Queue, fast => 0 : 4.17s
MCE::Queue, fast => 1 : 2.62s
Thread::Queue : 4.14s
Linux (VM) /usr: 186,154 files
MCE::Queue, fast => 0 : 12.57s
MCE::Queue, fast => 1 : 3.36s
Thread::Queue : 5.91s
Solaris (VM) /usr: 603,051 files
MCE::Queue, fast => 0 : 39.04s
MCE::Queue, fast => 1 : 18.08s
Thread::Queue * Perl not built to support threads
DESCRIPTION
This module provides a queue interface supporting normal and priority queues and utilizing
the IPC engine behind MCE. Data resides under the manager process. MCE::Queue also allows
for a worker to create any number of queues locally not available to other workers
including the manager process. Think of a CPU having L3 (shared) and L1 (local) cache.
IMPORT
Three options are available for overriding the default value for new queues. The porder
option applies to priority queues only.
use MCE::Queue porder => $MCE::Queue::HIGHEST,
type => $MCE::Queue::FIFO,
fast => 0;
use MCE::Queue; # Same as above
## Possible values
porder => $MCE::Queue::HIGHEST # Highest priority items dequeue first
$MCE::Queue::LOWEST # Lowest priority items dequeue first
type => $MCE::Queue::FIFO # First in, first out
$MCE::Queue::LIFO # Last in, first out
$MCE::Queue::LILO # (Synonym for FIFO)
$MCE::Queue::FILO # (Synonym for LIFO)
THREE RUN MODES
MCE::Queue can be utilized under the following conditions:
A) use MCE; B) use MCE::Queue; C) use MCE::Queue;
use MCE::Queue; use MCE;
A) MCE is included prior to inclusion of MCE::Queue
The dequeue method blocks for the manager process including workers. All data resides
under the manager process. Workers send/request data via IPC.
Creating a queue from within the worker process will cause the queue to run in local
mode (C). The data resides under the worker process and not available to other workers
including the manager process.
B) MCE::Queue is included prior to inclusion of MCE
Queues behave as if running in local mode for the manager and worker processes for the
duration of the script. I cannot think of a use-case for this, but mentioning the
behavior in the event MCE::Queue is included before MCE.
C) MCE::Queue without inclusion of MCE
The dequeue method is non-blocking. Queues behave similarly to local queuing. This
mode is efficient due to minimum overhead and zero IPC behind the scene. Hence, MCE is
not required to use MCE::Queue.
API DOCUMENTATION
MCE::Queue->new ( [ queue => \@array, fast => 1 ] )
This creates a new queue. Available options are queue, porder, type, fast, and gather. The
gather option is mainly for running with MCE and wanting to pass item(s) to a callback
function for appending to the queue.
The 'fast' option speeds up ->dequeue ops and not enabled by default. It is beneficial for
queues not calling ->clear or ->dequeue_nb and not altering the optional count value while
running; e.g. ->dequeue($count). Basically, do not enable 'fast' if varying $count
dynamically.
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( fast => 1 );
Multiple queues may point to the same callback function. The first argument for the
callback is the queue object.
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' );
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->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->enqueue ( $item [, $item, ... ] )
Appends a list of items onto the end of the normal queue.
$q->enqueuep ( $p, $item [, $item, ... ] )
Appends a list of items onto the end of the priority queue with priority.
$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.
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 the remaining items
and undef for up to the count requested.
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.
$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 will return undef
in the absence of data from the queue.
$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 insert otherwise.
$q->pending ( void )
Returns the number of items in the queue. The count includes both normal and priority
data.
$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 peek otherwise.
$q->peekh ( [ $index ] )
Returns an item from the heap, 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
ACKNOWLEDGEMENTS
POE::Queue::Array
Two if statements were adopted for checking if the item belongs at the end or head of
the queue.
List::BinarySearch
The bsearch_num_pos method was helpful for accommodating the highest and lowest order
in MCE::Queue.
List::Priority
MCE::Queue supports both normal and priority queues.
Thread::Queue
Thread::Queue is used as a template for identifying and documenting the methods.
MCE::Queue is not fully compatible due to supporting normal and priority queues
simultaneously; e.g.
$q->enqueuep( $p, $item [, $item, ... ] ); ## Priority queue
$q->enqueue( $item [, $item, ... ] ); ## Normal queue
$q->dequeue( [ $count ] ); ## Priority data dequeues first
$q->dequeue_nb( [ $count ] ); ## Behavior is not the same
$q->pending(); ## Counts both normal/priority data
## in the queue
Parallel::DataPipe
The recursion example, in the sysopsis above, was largely adopted from this module.
INDEX
MCE
AUTHOR
Mario E. Roy, <marioeroy AT gmail DOT com>