abf33ae2ca7f70180ce9cbc644c7231ebc6f8da1
[perl.git] / lib / Thread / Queue.pm
1 package Thread::Queue;
2
3 use strict;
4 use warnings;
5
6 our $VERSION = '2.08';
7
8 use threads::shared 1.21;
9 use Scalar::Util 1.10 qw(looks_like_number blessed reftype refaddr);
10
11 # Carp errors from threads::shared calls should complain about caller
12 our @CARP_NOT = ("threads::shared");
13
14 # Predeclarations for internal functions
15 my ($validate_count, $validate_index);
16
17 # Create a new queue possibly pre-populated with items
18 sub new
19 {
20     my $class = shift;
21     my @queue :shared = map { shared_clone($_) } @_;
22     return bless(\@queue, $class);
23 }
24
25 # Add items to the tail of a queue
26 sub enqueue
27 {
28     my $queue = shift;
29     lock(@$queue);
30     push(@$queue, map { shared_clone($_) } @_)
31         and cond_signal(@$queue);
32 }
33
34 # Return a count of the number of items on a queue
35 sub pending
36 {
37     my $queue = shift;
38     lock(@$queue);
39     return scalar(@$queue);
40 }
41
42 # Return 1 or more items from the head of a queue, blocking if needed
43 sub dequeue
44 {
45     my $queue = shift;
46     lock(@$queue);
47
48     my $count = @_ ? $validate_count->(shift) : 1;
49
50     # Wait for requisite number of items
51     cond_wait(@$queue) until (@$queue >= $count);
52     cond_signal(@$queue) if (@$queue > $count);
53
54     # Return single item
55     return shift(@$queue) if ($count == 1);
56
57     # Return multiple items
58     my @items;
59     push(@items, shift(@$queue)) for (1..$count);
60     return @items;
61 }
62
63 # Return items from the head of a queue with no blocking
64 sub dequeue_nb
65 {
66     my $queue = shift;
67     lock(@$queue);
68
69     my $count = @_ ? $validate_count->(shift) : 1;
70
71     # Return single item
72     return shift(@$queue) if ($count == 1);
73
74     # Return multiple items
75     my @items;
76     for (1..$count) {
77         last if (! @$queue);
78         push(@items, shift(@$queue));
79     }
80     return @items;
81 }
82
83 # Return an item without removing it from a queue
84 sub peek
85 {
86     my $queue = shift;
87     lock(@$queue);
88     my $index = @_ ? $validate_index->(shift) : 0;
89     return $$queue[$index];
90 }
91
92 # Insert items anywhere into a queue
93 sub insert
94 {
95     my $queue = shift;
96     lock(@$queue);
97
98     my $index = $validate_index->(shift);
99
100     return if (! @_);   # Nothing to insert
101
102     # Support negative indices
103     if ($index < 0) {
104         $index += @$queue;
105         if ($index < 0) {
106             $index = 0;
107         }
108     }
109
110     # Dequeue items from $index onward
111     my @tmp;
112     while (@$queue > $index) {
113         unshift(@tmp, pop(@$queue))
114     }
115
116     # Add new items to the queue
117     push(@$queue, map { shared_clone($_) } @_);
118
119     # Add previous items back onto the queue
120     push(@$queue, @tmp);
121
122     # Soup's up
123     cond_signal(@$queue);
124 }
125
126 # Remove items from anywhere in a queue
127 sub extract
128 {
129     my $queue = shift;
130     lock(@$queue);
131
132     my $index = @_ ? $validate_index->(shift) : 0;
133     my $count = @_ ? $validate_count->(shift) : 1;
134
135     # Support negative indices
136     if ($index < 0) {
137         $index += @$queue;
138         if ($index < 0) {
139             $count += $index;
140             return if ($count <= 0);            # Beyond the head of the queue
141             return $queue->dequeue_nb($count);  # Extract from the head
142         }
143     }
144
145     # Dequeue items from $index+$count onward
146     my @tmp;
147     while (@$queue > ($index+$count)) {
148         unshift(@tmp, pop(@$queue))
149     }
150
151     # Extract desired items
152     my @items;
153     unshift(@items, pop(@$queue)) while (@$queue > $index);
154
155     # Add back any removed items
156     push(@$queue, @tmp);
157
158     # Return single item
159     return $items[0] if ($count == 1);
160
161     # Return multiple items
162     return @items;
163 }
164
165 ### Internal Functions ###
166
167 # Check value of the requested index
168 $validate_index = sub {
169     my $index = shift;
170
171     if (! looks_like_number($index) || (int($index) != $index)) {
172         require Carp;
173         my ($method) = (caller(1))[3];
174         $method =~ s/Thread::Queue:://;
175         $index = 'undef' if (! defined($index));
176         Carp::croak("Invalid 'index' argument ($index) to '$method' method");
177     }
178
179     return $index;
180 };
181
182 # Check value of the requested count
183 $validate_count = sub {
184     my $count = shift;
185
186     if ((! looks_like_number($count)) || (int($count) != $count) || ($count < 1)) {
187         require Carp;
188         my ($method) = (caller(1))[3];
189         $method =~ s/Thread::Queue:://;
190         $count = 'undef' if (! defined($count));
191         Carp::croak("Invalid 'count' argument ($count) to '$method' method");
192     }
193
194     return $count;
195 };
196
197 1;
198
199 =head1 NAME
200
201 Thread::Queue - Thread-safe queues
202
203 =head1 VERSION
204
205 This document describes Thread::Queue version 2.08
206
207 =head1 SYNOPSIS
208
209     use strict;
210     use warnings;
211
212     use threads;
213     use Thread::Queue;
214
215     my $q = Thread::Queue->new();    # A new empty queue
216
217     # Worker thread
218     my $thr = threads->create(sub {
219                                 while (my $item = $q->dequeue()) {
220                                     # Do work on $item
221                                 }
222                              })->detach();
223
224     # Send work to the thread
225     $q->enqueue($item1, ...);
226
227
228     # Count of items in the queue
229     my $left = $q->pending();
230
231     # Non-blocking dequeue
232     if (defined(my $item = $q->dequeue_nb())) {
233         # Work on $item
234     }
235
236     # Get the second item in the queue without dequeuing anything
237     my $item = $q->peek(1);
238
239     # Insert two items into the queue just behind the head
240     $q->insert(1, $item1, $item2);
241
242     # Extract the last two items on the queue
243     my ($item1, $item2) = $q->extract(-2, 2);
244
245 =head1 DESCRIPTION
246
247 This module provides thread-safe FIFO queues that can be accessed safely by
248 any number of threads.
249
250 Any data types supported by L<threads::shared> can be passed via queues:
251
252 =over
253
254 =item Ordinary scalars
255
256 =item Array refs
257
258 =item Hash refs
259
260 =item Scalar refs
261
262 =item Objects based on the above
263
264 =back
265
266 Ordinary scalars are added to queues as they are.
267
268 If not already thread-shared, the other complex data types will be cloned
269 (recursively, if needed, and including any C<bless>ings and read-only
270 settings) into thread-shared structures before being placed onto a queue.
271
272 For example, the following would cause L<Thread::Queue> to create a empty,
273 shared array reference via C<&shared([])>, copy the elements 'foo', 'bar'
274 and 'baz' from C<@ary> into it, and then place that shared reference onto
275 the queue:
276
277     my @ary = qw/foo bar baz/;
278     $q->enqueue(\@ary);
279
280 However, for the following, the items are already shared, so their references
281 are added directly to the queue, and no cloning takes place:
282
283     my @ary :shared = qw/foo bar baz/;
284     $q->enqueue(\@ary);
285
286     my $obj = &shared({});
287     $$obj{'foo'} = 'bar';
288     $$obj{'qux'} = 99;
289     bless($obj, 'My::Class');
290     $q->enqueue($obj);
291
292 See L</"LIMITATIONS"> for caveats related to passing objects via queues.
293
294 =head1 QUEUE CREATION
295
296 =over
297
298 =item ->new()
299
300 Creates a new empty queue.
301
302 =item ->new(LIST)
303
304 Creates a new queue pre-populated with the provided list of items.
305
306 =back
307
308 =head1 BASIC METHODS
309
310 The following methods deal with queues on a FIFO basis.
311
312 =over
313
314 =item ->enqueue(LIST)
315
316 Adds a list of items onto the end of the queue.
317
318 =item ->dequeue()
319
320 =item ->dequeue(COUNT)
321
322 Removes the requested number of items (default is 1) from the head of the
323 queue, and returns them.  If the queue contains fewer than the requested
324 number of items, then the thread will be blocked until the requisite number
325 of items are available (i.e., until other threads <enqueue> more items).
326
327 =item ->dequeue_nb()
328
329 =item ->dequeue_nb(COUNT)
330
331 Removes the requested number of items (default is 1) from the head of the
332 queue, and returns them.  If the queue contains fewer than the requested
333 number of items, then it immediately (i.e., non-blocking) returns whatever
334 items there are on the queue.  If the queue is empty, then C<undef> is
335 returned.
336
337 =item ->pending()
338
339 Returns the number of items still in the queue.
340
341 =back
342
343 =head1 ADVANCED METHODS
344
345 The following methods can be used to manipulate items anywhere in a queue.
346
347 To prevent the contents of a queue from being modified by another thread
348 while it is being examined and/or changed, L<lock|threads::shared/"lock
349 VARIABLE"> the queue inside a local block:
350
351     {
352         lock($q);   # Keep other threads from changing the queue's contents
353         my $item = $q->peek();
354         if ($item ...) {
355             ...
356         }
357     }
358     # Queue is now unlocked
359
360 =over
361
362 =item ->peek()
363
364 =item ->peek(INDEX)
365
366 Returns an item from the queue without dequeuing anything.  Defaults to the
367 the head of queue (at index position 0) if no index is specified.  Negative
368 index values are supported as with L<arrays|perldata/"Subscripts"> (i.e., -1
369 is the end of the queue, -2 is next to last, and so on).
370
371 If no items exists at the specified index (i.e., the queue is empty, or the
372 index is beyond the number of items on the queue), then C<undef> is returned.
373
374 Remember, the returned item is not removed from the queue, so manipulating a
375 C<peek>ed at reference affects the item on the queue.
376
377 =item ->insert(INDEX, LIST)
378
379 Adds the list of items to the queue at the specified index position (0
380 is the head of the list).  Any existing items at and beyond that position are
381 pushed back past the newly added items:
382
383     $q->enqueue(1, 2, 3, 4);
384     $q->insert(1, qw/foo bar/);
385     # Queue now contains:  1, foo, bar, 2, 3, 4
386
387 Specifying an index position greater than the number of items in the queue
388 just adds the list to the end.
389
390 Negative index positions are supported:
391
392     $q->enqueue(1, 2, 3, 4);
393     $q->insert(-2, qw/foo bar/);
394     # Queue now contains:  1, 2, foo, bar, 3, 4
395
396 Specifying a negative index position greater than the number of items in the
397 queue adds the list to the head of the queue.
398
399 =item ->extract()
400
401 =item ->extract(INDEX)
402
403 =item ->extract(INDEX, COUNT)
404
405 Removes and returns the specified number of items (defaults to 1) from the
406 specified index position in the queue (0 is the head of the queue).  When
407 called with no arguments, C<extract> operates the same as C<dequeue_nb>.
408
409 This method is non-blocking, and will return only as many items as are
410 available to fulfill the request:
411
412     $q->enqueue(1, 2, 3, 4);
413     my $item  = $q->extract(2)     # Returns 3
414                                    # Queue now contains:  1, 2, 4
415     my @items = $q->extract(1, 3)  # Returns (2, 4)
416                                    # Queue now contains:  1
417
418 Specifying an index position greater than the number of items in the
419 queue results in C<undef> or an empty list being returned.
420
421     $q->enqueue('foo');
422     my $nada = $q->extract(3)      # Returns undef
423     my @nada = $q->extract(1, 3)   # Returns ()
424
425 Negative index positions are supported.  Specifying a negative index position
426 greater than the number of items in the queue may return items from the head
427 of the queue (similar to C<dequeue_nb>) if the count overlaps the head of the
428 queue from the specified position (i.e. if queue size + index + count is
429 greater than zero):
430
431     $q->enqueue(qw/foo bar baz/);
432     my @nada = $q->extract(-6, 2);   # Returns ()         - (3+(-6)+2) <= 0
433     my @some = $q->extract(-6, 4);   # Returns (foo)      - (3+(-6)+4) > 0
434                                      # Queue now contains:  bar, baz
435     my @rest = $q->extract(-3, 4);   # Returns (bar, baz) - (2+(-3)+4) > 0
436
437 =back
438
439 =head1 NOTES
440
441 Queues created by L<Thread::Queue> can be used in both threaded and
442 non-threaded applications.
443
444 =head1 LIMITATIONS
445
446 Passing objects on queues may not work if the objects' classes do not support
447 sharing.  See L<threads::shared/"BUGS AND LIMITATIONS"> for more.
448
449 Passing array/hash refs that contain objects may not work for Perl prior to
450 5.10.0.
451
452 =head1 SEE ALSO
453
454 Thread::Queue Discussion Forum on CPAN:
455 L<http://www.cpanforum.com/dist/Thread-Queue>
456
457 Annotated POD for Thread::Queue:
458 L<http://annocpan.org/~JDHEDDEN/Thread-Queue-2.08/lib/Thread/Queue.pm>
459
460 Source repository:
461 L<http://code.google.com/p/thread-queue/>
462
463 L<threads>, L<threads::shared>
464
465 =head1 MAINTAINER
466
467 Jerry D. Hedden, S<E<lt>jdhedden AT cpan DOT orgE<gt>>
468
469 =head1 LICENSE
470
471 This program is free software; you can redistribute it and/or modify it under
472 the same terms as Perl itself.
473
474 =cut