summaryrefslogtreecommitdiff
path: root/Master/tlpkg/tlperl/lib/Thread/Queue.pm
diff options
context:
space:
mode:
Diffstat (limited to 'Master/tlpkg/tlperl/lib/Thread/Queue.pm')
-rw-r--r--Master/tlpkg/tlperl/lib/Thread/Queue.pm185
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>>