1 files changed, 153 insertions, 32 deletions

diff --git a/Master/tlpkg/tlperl/lib/Thread/Queue.pm b/Master/tlpkg/tlperl/lib/Thread/Queue.pm
index 8588ed58458..027dd56c8a5 100644
--- a/Master/tlpkg/tlperl/lib/Thread/Queue.pm
+++ b/Master/tlpkg/tlperl/lib/Thread/Queue.pm
@@ -3,7 +3,7 @@ package Thread::Queue;
 use strict;
 use warnings;
 
-our $VERSION = '2.12';
+our $VERSION = '3.02';
 $VERSION = eval $VERSION;
 
 use threads::shared 1.21;
@@ -13,44 +13,65 @@ use Scalar::Util 1.10 qw(looks_like_number blessed reftype refaddr);
 our @CARP_NOT = ("threads::shared");
 
 # Predeclarations for internal functions
-my ($validate_count, $validate_index);
+my ($validate_count, $validate_index, $validate_timeout);
 
 # Create a new queue possibly pre-populated with items
 sub new
 {
     my $class = shift;
     my @queue :shared = map { shared_clone($_) } @_;
-    return bless(\@queue, $class);
+    my %self :shared = ( 'queue' => \@queue );
+    return bless(\%self, $class);
 }
 
 # Add items to the tail of a queue
 sub enqueue
 {
-    my $queue = shift;
-    lock(@$queue);
-    push(@$queue, map { shared_clone($_) } @_)
-        and cond_signal(@$queue);
+    my $self = shift;
+    lock(%$self);
+    if ($$self{'ENDED'}) {
+        require Carp;
+        Carp::croak("'enqueue' method called on queue that has been 'end'ed");
+    }
+    push(@{$$self{'queue'}}, map { shared_clone($_) } @_)
+        and cond_signal(%$self);
 }
 
 # Return a count of the number of items on a queue
 sub pending
 {
-    my $queue = shift;
-    lock(@$queue);
-    return scalar(@$queue);
+    my $self = shift;
+    lock(%$self);
+    return if ($$self{'ENDED'} && ! @{$$self{'queue'}});
+    return scalar(@{$$self{'queue'}});
+}
+
+# Indicate that no more data will enter the queue
+sub end
+{
+    my $self = shift;
+    lock $self;
+    # No more data is coming
+    $$self{'ENDED'} = 1;
+    # Try to release at least one blocked thread
+    cond_signal(%$self);
 }
 
 # Return 1 or more items from the head of a queue, blocking if needed
 sub dequeue
 {
-    my $queue = shift;
-    lock(@$queue);
+    my $self = shift;
+    lock(%$self);
+    my $queue = $$self{'queue'};
 
     my $count = @_ ? $validate_count->(shift) : 1;
 
     # Wait for requisite number of items
-    cond_wait(@$queue) until (@$queue >= $count);
-    cond_signal(@$queue) if (@$queue > $count);
+    cond_wait(%$self) while ((@$queue < $count) && ! $$self{'ENDED'});
+    cond_signal(%$self) if ((@$queue > $count) || $$self{'ENDED'});
+
+    # If no longer blocking, try getting whatever is left on the queue
+    return $self->dequeue_nb($count) if ($$self{'ENDED'});
 
     # Return single item
     return shift(@$queue) if ($count == 1);
@@ -64,8 +85,9 @@ sub dequeue
 # Return items from the head of a queue with no blocking
 sub dequeue_nb
 {
-    my $queue = shift;
-    lock(@$queue);
+    my $self = shift;
+    lock(%$self);
+    my $queue = $$self{'queue'};
 
     my $count = @_ ? $validate_count->(shift) : 1;
 
@@ -81,20 +103,53 @@ sub dequeue_nb
     return @items;
 }
 
+# Return items from the head of a queue, blocking if needed up to a timeout
+sub dequeue_timed
+{
+    my $self = shift;
+    lock(%$self);
+    my $queue = $$self{'queue'};
+
+    # Timeout may be relative or absolute
+    my $timeout = @_ ? $validate_timeout->(shift) : -1;
+    # Convert to an absolute time for use with cond_timedwait()
+    if ($timeout < 32000000) {   # More than one year
+        $timeout += time();
+    }
+
+    my $count = @_ ? $validate_count->(shift) : 1;
+
+    # Wait for requisite number of items, or until timeout
+    while ((@$queue < $count) && ! $$self{'ENDED'}) {
+        last if (! cond_timedwait(%$self, $timeout));
+    }
+    cond_signal(%$self) if ((@$queue > $count) || $$self{'ENDED'});
+
+    # Get whatever we need off the queue if available
+    return $self->dequeue_nb($count);
+}
+
 # Return an item without removing it from a queue
 sub peek
 {
-    my $queue = shift;
-    lock(@$queue);
+    my $self = shift;
+    lock(%$self);
     my $index = @_ ? $validate_index->(shift) : 0;
-    return $$queue[$index];
+    return $$self{'queue'}[$index];
 }
 
 # Insert items anywhere into a queue
 sub insert
 {
-    my $queue = shift;
-    lock(@$queue);
+    my $self = shift;
+    lock(%$self);
+
+    if ($$self{'ENDED'}) {
+        require Carp;
+        Carp::croak("'insert' method called on queue that has been 'end'ed");
+    }
+
+    my $queue = $$self{'queue'};
 
     my $index = $validate_index->(shift);
 
@@ -121,14 +176,15 @@ sub insert
     push(@$queue, @tmp);
 
     # Soup's up
-    cond_signal(@$queue);
+    cond_signal(%$self);
 }
 
 # Remove items from anywhere in a queue
 sub extract
 {
-    my $queue = shift;
-    lock(@$queue);
+    my $self = shift;
+    lock(%$self);
+    my $queue = $$self{'queue'};
 
     my $index = @_ ? $validate_index->(shift) : 0;
     my $count = @_ ? $validate_count->(shift) : 1;
@@ -139,7 +195,7 @@ sub extract
         if ($index < 0) {
             $count += $index;
             return if ($count <= 0);            # Beyond the head of the queue
-            return $queue->dequeue_nb($count);  # Extract from the head
+            return $self->dequeue_nb($count);  # Extract from the head
         }
     }
 
@@ -202,6 +258,23 @@ $validate_count = sub {
     return $count;
 };
 
+# Check value of the requested timeout
+$validate_timeout = sub {
+    my $timeout = shift;
+
+    if (! defined($timeout) ||
+        ! looks_like_number($timeout))
+    {
+        require Carp;
+        my ($method) = (caller(1))[3];
+        $method =~ s/Thread::Queue:://;
+        $timeout = 'undef' if (! defined($timeout));
+        Carp::croak("Invalid 'timeout' argument ($timeout) to '$method' method");
+    }
+
+    return $timeout;
+};
+
 1;
 
 =head1 NAME
@@ -210,7 +283,7 @@ Thread::Queue - Thread-safe queues
 
 =head1 VERSION
 
-This document describes Thread::Queue version 2.12
+This document describes Thread::Queue version 3.02
 
 =head1 SYNOPSIS
 
@@ -223,15 +296,24 @@ This document describes Thread::Queue version 2.12
     my $q = Thread::Queue->new();    # A new empty queue
 
     # Worker thread
-    my $thr = threads->create(sub {
-                                while (my $item = $q->dequeue()) {
-                                    # Do work on $item
-                                }
-                             })->detach();
+    my $thr = threads->create(
+        sub {
+            # Thread will loop until no more work
+            while (defined(my $item = $q->dequeue())) {
+                # Do work on $item
+                ...
+            }
+        }
+    );
 
     # Send work to the thread
     $q->enqueue($item1, ...);
+    # Signal that there is no more work to be sent
+    $q->end();
+    # Join up with the thread when it finishes
+    $thr->join();
 
+    ...
 
     # Count of items in the queue
     my $left = $q->pending();
@@ -241,6 +323,11 @@ This document describes Thread::Queue version 2.12
         # Work on $item
     }
 
+    # Blocking dequeue with 5-second timeout
+    if (defined(my $item = $q->dequeue_timed(5))) {
+        # Work on $item
+    }
+
     # Get the second item in the queue without dequeuing anything
     my $item = $q->peek(1);
 
@@ -342,9 +429,41 @@ number of items, then it immediately (i.e., non-blocking) returns whatever
 items there are on the queue.  If the queue is empty, then C<undef> is
 returned.
 
+=item ->dequeue_timed(TIMEOUT)
+
+=item ->dequeue_timed(TIMEOUT, COUNT)
+
+Removes the requested number of items (default is 1) from the head of the
+queue, and returns them.  If the queue contains fewer than the requested
+number of items, then the thread will be blocked until the requisite number of
+items are available, or until the timeout is reached.  If the timeout is
+reached, it returns whatever items there are on the queue, or C<undef> if the
+queue is empty.
+
+The timeout may be a number of seconds relative to the current time (e.g., 5
+seconds from when the call is made), or may be an absolute timeout in I<epoch>
+seconds the same as would be used with
+L<cond_timedwait()|threads::shared/"cond_timedwait VARIABLE, ABS_TIMEOUT">.
+Fractional seconds (e.g., 2.5 seconds) are also supported (to the extent of
+the underlying implementation).
+
+If C<TIMEOUT> is missing, c<undef>, or less than or equal to 0, then this call
+behaves the same as C<dequeue_nb>.
+
 =item ->pending()
 
-Returns the number of items still in the queue.
+Returns the number of items still in the queue.  Returns C<undef> if the queue
+has been ended (see below), and there are no more items in the queue.
+
+=item ->end()
+
+Declares that no more items will be added to the queue.
+
+All threads blocking on C<dequeue()> calls will be unblocked with any
+remaining items in the queue and/or C<undef> being returned.  Any subsequent
+calls to C<dequeue()> will behave like C<dequeue_nb()>.
+
+Once ended, no more items may be placed in the queue.
 
 =back
 
@@ -464,6 +583,8 @@ L<http://www.cpanforum.com/dist/Thread-Queue>
 
 L<threads>, L<threads::shared>
 
+Sample code in the I<examples> directory of this distribution on CPAN.
+
 =head1 MAINTAINER
 
 Jerry D. Hedden, S<E<lt>jdhedden AT cpan DOT orgE<gt>>