1MCE::Shared::Queue(3) User Contributed Perl DocumentationMCE::Shared::Queue(3)
2
6 MCE::Shared::Queue - Hybrid-queue helper class
7
9 This document describes MCE::Shared::Queue version 1.877
10
12 A queue helper class for use as a standalone or managed by MCE::Shared.
13 This module is mostly compatible with MCE::Queue except for the
15 "gather" option which is not supported in this context. It provides a
16 queue interface supporting normal and priority queues. Data from shared
17 queues reside under the shared-manager process, otherwise locally.
18
20 # non-shared or local construction for use by a single process
21 use MCE::Shared::Queue;
23 my $qu = MCE::Shared::Queue->new( await => 1, queue => [ "." ] );
25 # construction for sharing with other threads and processes
27 use MCE::Shared;
29 use MCE::Shared::Queue;
30 my $qu = MCE::Shared->queue(
32 porder => $MCE::Shared::Queue::HIGHEST,
33 type => $MCE::Shared::Queue::FIFO,
34 );
35 # possible values for "porder" and "type"
37 porder =>
39 $MCE::Shared::Queue::HIGHEST # Highest priority items dequeue first
40 $MCE::Shared::Queue::LOWEST # Lowest priority items dequeue first
41 type =>
43 $MCE::Shared::Queue::FIFO # First in, first out
44 $MCE::Shared::Queue::LIFO # Last in, first out
45 $MCE::Shared::Queue::LILO # Synonym for FIFO
46 $MCE::Shared::Queue::FILO # Synonym for LIFO
47 # below, [ ... ] denotes optional parameters
49 $qu->await( [ $pending_threshold ] );
51 $qu->clear();
52 $qu->end();
53 $qu->enqueue( $item [, $item, ... ] );
55 $qu->enqueuep( $priority, $item [, $item, ... ] );
56 $item = $qu->dequeue();
58 @items = $qu->dequeue( $count );
59 $item = $qu->dequeue_nb();
60 @items = $qu->dequeue_nb( $count );
61 $qu->insert( $index, $item [, $item, ... ] );
63 $qu->insertp( $priority, $index, $item [, $item, ... ] );
64 $count = $qu->pending();
66 $item = $qu->peek( [ $index ] );
67 $item = $qu->peekp( $priority [, $index ] );
68 @array = $qu->heap();
69
71 MCE::Shared::Queue->new ( [ options ] )
72 MCE::Shared->queue ( [ options ] )
73 Constructs a new object. Supported options are queue, porder, type, and
74 await. Note: The barrier and fast options are silentently ignored (no-
75 op) if specified; starting with 1.867.
76 # non-shared or local construction for use by a single process
78 use MCE::Shared::Queue;
80 $q1 = MCE::Shared::Queue->new();
82 $q2 = MCE::Shared::Queue->new( queue => [ 0, 1, 2 ] );
83 $q3 = MCE::Shared::Queue->new( porder => $MCE::Shared::Queue::HIGHEST );
85 $q4 = MCE::Shared::Queue->new( porder => $MCE::Shared::Queue::LOWEST );
86 $q5 = MCE::Shared::Queue->new( type => $MCE::Shared::Queue::FIFO );
88 $q6 = MCE::Shared::Queue->new( type => $MCE::Shared::Queue::LIFO );
89 $q7 = MCE::Shared::Queue->new( await => 1, barrier => 0 );
91 $q8 = MCE::Shared::Queue->new( fast => 1 );
92 # construction for sharing with other threads and processes
94 use MCE::Shared;
96 use MCE::Shared::Queue;
97 $q1 = MCE::Shared->queue();
99 $q2 = MCE::Shared->queue( queue => [ 0, 1, 2 ] );
100 $q3 = MCE::Shared->queue( porder => $MCE::Shared::Queue::HIGHEST );
102 $q4 = MCE::Shared->queue( porder => $MCE::Shared::Queue::LOWEST );
103 $q5 = MCE::Shared->queue( type => $MCE::Shared::Queue::FIFO );
105 $q6 = MCE::Shared->queue( type => $MCE::Shared::Queue::LIFO );
106 $q7 = MCE::Shared->queue( await => 1, barrier => 0 );
108 $q8 = MCE::Shared->queue( fast => 1 );
109 The "await" option, when enabled, allows workers to block (semaphore-
111 like) until the number of items pending is equal or less than a
112 threshold value. The "await" method is described below.
113 Obsolete: On Unix platforms, "barrier" mode (enabled by default)
115 prevents many workers from dequeuing simultaneously to lessen overhead
116 for the OS kernel. Specify 0 to disable barrier mode and not allocate
117 sockets. The barrier option has no effect if constructing the queue
118 inside a thread or enabling "fast".
119 Obsolete: The "fast" option speeds up dequeues and is not enabled by
121 default. It is beneficial for queues not calling (->dequeue_nb) and
122 not altering the count value while running; e.g. ->dequeue($count).
123 await ( pending_threshold )
125 Waits until the queue drops down to threshold items. The "await" method
126 is beneficial when wanting to throttle worker(s) appending to the
127 queue. Perhaps, consumers are running a bit behind and wanting prevent
128 memory consumption from increasing too high. Below, the number of items
129 pending will never go above 20.
130 use Time::HiRes qw( sleep );
132 use MCE::Flow;
134 use MCE::Shared;
135 my $q = MCE::Shared->queue( await => 1, fast => 1 );
137 my ( $producers, $consumers ) = ( 1, 8 );
138 mce_flow {
140 task_name => [ 'producer', 'consumer' ],
141 max_workers => [ $producers, $consumers ],
142 },
143 sub {
144 ## producer
145 for my $item ( 1 .. 100 ) {
146 $q->enqueue($item);
147 ## blocks until the # of items pending reaches <= 10
149 if ($item % 10 == 0) {
150 MCE->say( 'pending: '.$q->pending() );
151 $q->await(10);
152 }
153 }
154 ## notify consumers no more work
156 $q->end();
157 },
159 sub {
160 ## consumers
161 while (defined (my $next = $q->dequeue())) {
162 MCE->say( MCE->task_wid().': '.$next );
163 sleep 0.100;
164 }
165 };
166 clear ( )
168 Clears the queue of any items.
169 $q->clear;
171 end ( )
173 Stops the queue from receiving more items. Any worker blocking on
174 "dequeue" will be unblocked automatically. Subsequent calls to
175 "dequeue" will behave like "dequeue_nb". Current API available since
176 MCE::Shared 1.814.
177 $q->end();
179 MCE Models (e.g. MCE::Flow) may persist between runs. In that case, one
181 might want to enqueue "undef"'s versus calling "end". The number of
182 "undef"'s depends on how many items workers dequeue at a time.
183 $q->enqueue((undef) x ($N_workers * 1)); # $q->dequeue() 1 item
185 $q->enqueue((undef) x ($N_workers * 2)); # $q->dequeue(2) 2 items
186 $q->enqueue((undef) x ($N_workers * N)); # $q->dequeue(N) N items
187 enqueue ( item [, item, ... ] )
189 Appends a list of items onto the end of the normal queue.
190 $q->enqueue( 'foo' );
192 $q->enqueue( 'bar', 'baz' );
193 enqueuep ( priority, item [, item, ... ] )
195 Appends a list of items onto the end of the priority queue with
196 priority.
197 $q->enqueue( $priority, 'foo' );
199 $q->enqueue( $priority, 'bar', 'baz' );
200 dequeue ( [ count ] )
202 Returns the requested number of items (default 1) from the queue.
203 Priority data will always dequeue first before any data from the normal
204 queue.
205 $q->dequeue( 2 );
207 $q->dequeue; # default 1
208 The method will block if the queue contains zero items. If the queue
210 contains fewer than the requested number of items, the method will not
211 block, but return whatever items there are on the queue.
212 The $count, used for requesting the number of items, is beneficial when
214 workers are passing parameters through the queue. For this reason,
215 always remember to dequeue using the same multiple for the count. This
216 is unlike Thread::Queue which will block until the requested number of
217 items are available.
218 # MCE::Shared::Queue 1.816 and prior releases
220 while ( my @items = $q->dequeue(2) ) {
221 last unless ( defined $items[0] );
222 ...
223 }
224 # MCE::Shared::Queue 1.817 and later
226 while ( my @items = $q->dequeue(2) ) {
227 ...
228 }
229 dequeue_nb ( [ count ] )
231 Returns the requested number of items (default 1) from the queue. Like
232 with dequeue, priority data will always dequeue first. This method is
233 non-blocking and returns "undef" in the absence of data.
234 $q->dequeue_nb( 2 );
236 $q->dequeue_nb; # default 1
237 insert ( index, item [, item, ... ] )
239 Adds the list of items to the queue at the specified index position (0
240 is the head of the list). The head of the queue is that item which
241 would be removed by a call to dequeue.
242 $q = MCE::Shared->queue( type => $MCE::Shared::Queue::FIFO );
244 $q->enqueue(1, 2, 3, 4);
245 $q->insert(1, 'foo', 'bar');
246 # Queue now contains: 1, foo, bar, 2, 3, 4
247 $q = MCE::Shared->queue( type => $MCE::Shared::Queue::LIFO );
249 $q->enqueue(1, 2, 3, 4);
250 $q->insert(1, 'foo', 'bar');
251 # Queue now contains: 1, 2, 3, 'foo', 'bar', 4
252 insertp ( priority, index, item [, item, ... ] )
254 Adds the list of items to the queue at the specified index position
255 with priority. The behavior is similarly to "$q->insert" otherwise.
256 pending ( )
258 Returns the number of items in the queue. The count includes both
259 normal and priority data. Returns "undef" if the queue has been ended,
260 and there are no more items in the queue.
261 $q = MCE::Shared->queue();
263 $q->enqueuep(5, 'foo', 'bar');
264 $q->enqueue('sunny', 'day');
265 print $q->pending(), "\n";
267 # Output: 4
268 peek ( [ index ] )
270 Returns an item from the normal queue, at the specified index, without
271 dequeuing anything. It defaults to the head of the queue if index is
272 not specified. The head of the queue is that item which would be
273 removed by a call to dequeue. Negative index values are supported,
274 similarly to arrays.
275 $q = MCE::Shared->queue( type => $MCE::Shared::Queue::FIFO );
277 $q->enqueue(1, 2, 3, 4, 5);
278 print $q->peek(1), ' ', $q->peek(-2), "\n";
280 # Output: 2 4
281 $q = MCE::Shared->queue( type => $MCE::Shared::Queue::LIFO );
283 $q->enqueue(1, 2, 3, 4, 5);
284 print $q->peek(1), ' ', $q->peek(-2), "\n";
286 # Output: 4 2
287 peekp ( priority [, index ] )
289 Returns an item from the queue with priority, at the specified index,
290 without dequeuing anything. It defaults to the head of the queue if
291 index is not specified. The behavior is similarly to "$q->peek"
292 otherwise.
293 peekh ( [ index ] )
295 Returns an item from the head of the heap or at the specified index.
296 $q = MCE::Shared->queue( porder => $MCE::Shared::Queue::HIGHEST );
298 $q->enqueuep(5, 'foo');
299 $q->enqueuep(6, 'bar');
300 $q->enqueuep(4, 'sun');
301 print $q->peekh(0), "\n";
303 # Output: 6
304 $q = MCE::Shared->queue( porder => $MCE::Shared::Queue::LOWEST );
306 $q->enqueuep(5, 'foo');
307 $q->enqueuep(6, 'bar');
308 $q->enqueuep(4, 'sun');
309 print $q->peekh(0), "\n";
311 # Output: 4
312 heap ( )
314 Returns an array containing the heap data. Heap data consists of
315 priority numbers, not the data.
316 @h = $q->heap; # $MCE::Shared::Queue::HIGHEST
318 # Heap contains: 6, 5, 4
319 @h = $q->heap; # $MCE::Shared::Queue::LOWEST
321 # Heap contains: 4, 5, 6
322
324 • List::BinarySearch
325 The bsearch_num_pos method was helpful for accommodating the highest
327 and lowest order in MCE::Shared::Queue.
328 • POE::Queue::Array
330 For extra optimization, two if statements were adopted for checking
332 if the item belongs at the end or head of the queue.
333 • List::Priority
335 MCE::Shared::Queue supports both normal and priority queues.
337 • Thread::Queue
339 Thread::Queue is used as a template for identifying and documenting
341 the methods. MCE::Shared::Queue is not fully compatible due to
342 supporting normal and priority queues simultaneously; e.g.
343 $q->enqueue( $item [, $item, ... ] ); # normal queue
345 $q->enqueuep( $p, $item [, $item, ... ] ); # priority queue
346 $q->dequeue( [ $count ] ); # priority data dequeues first
348 $q->dequeue_nb( [ $count ] );
349 $q->pending(); # counts both normal/priority queues
351
353 Perl must have IO::FDPass for constructing a shared "condvar" or
354 "queue" while the shared-manager process is running. For platforms
355 where IO::FDPass isn't possible, construct "condvar" and "queue" before
356 other classes. On systems without "IO::FDPass", the manager process is
357 delayed until sharing other classes or started explicitly.
358 use MCE::Shared;
360 my $has_IO_FDPass = $INC{'IO/FDPass.pm'} ? 1 : 0;
362 my $cv = MCE::Shared->condvar();
364 my $que = MCE::Shared->queue();
365 MCE::Shared->start() unless $has_IO_FDPass;
367 Regarding mce_open, "IO::FDPass" is needed for constructing a shared-
369 handle from a non-shared handle not yet available inside the shared-
370 manager process. The workaround is to have the non-shared handle made
371 before the shared-manager is started. Passing a file by reference is
372 fine for the three STD* handles.
373 # The shared-manager knows of \*STDIN, \*STDOUT, \*STDERR.
375 mce_open my $shared_in, "<", \*STDIN; # ok
377 mce_open my $shared_out, ">>", \*STDOUT; # ok
378 mce_open my $shared_err, ">>", \*STDERR; # ok
379 mce_open my $shared_fh1, "<", "/path/to/sequence.fasta"; # ok
380 mce_open my $shared_fh2, ">>", "/path/to/results.log"; # ok
381 mce_open my $shared_fh, ">>", \*NON_SHARED_FH; # requires IO::FDPass
383 The IO::FDPass module is known to work reliably on most platforms.
385 Install 1.1 or later to rid of limitations described above.
386 perl -MIO::FDPass -le "print 'Cheers! Perl has IO::FDPass.'"
388
390 MCE, MCE::Hobo, MCE::Shared
391
393 Mario E. Roy, <marioeroy AT gmail DOT com>
394perl v5.36.0 2022-07-22 MCE::Shared::Queue(3)