Upgrade to Thread::Queue 3.12

9 files changed, 73 insertions, 25 deletions

diff --git a/dist/Thread-Queue/lib/Thread/Queue.pm b/dist/Thread-Queue/lib/Thread/Queue.pm
index 9f896b72ea..c0d2180653 100644
--- a/dist/Thread-Queue/lib/Thread/Queue.pm
+++ b/dist/Thread-Queue/lib/Thread/Queue.pm
@@ -3,7 +3,7 @@ package Thread::Queue;
 use strict;
 use warnings;
 
-our $VERSION = '3.11';
+our $VERSION = '3.12';
 $VERSION = eval $VERSION;
 
 use threads::shared 1.21;
@@ -65,8 +65,8 @@ sub end
     lock(%$self);
     # No more data is coming
     $$self{'ENDED'} = 1;
-    # Try to release at least one blocked thread
-    cond_signal(%$self);
+
+    cond_signal(%$self);  # Unblock possibly waiting threads
 }
 
 # Return 1 or more items from the head of a queue, blocking if needed
@@ -80,17 +80,21 @@ sub dequeue
 
     # Wait for requisite number of items
     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);
+    if ($count == 1) {
+        my $item = shift(@$queue);
+        cond_signal(%$self);  # Unblock possibly waiting threads
+        return $item;
+    }
 
     # Return multiple items
     my @items;
     push(@items, shift(@$queue)) for (1..$count);
+    cond_signal(%$self);  # Unblock possibly waiting threads
     return @items;
 }
 
@@ -104,7 +108,11 @@ sub dequeue_nb
     my $count = @_ ? $self->_validate_count(shift) : 1;
 
     # Return single item
-    return shift(@$queue) if ($count == 1);
+    if ($count == 1) {
+        my $item = shift(@$queue);
+        cond_signal(%$self);  # Unblock possibly waiting threads
+        return $item;
+    }
 
     # Return multiple items
     my @items;
@@ -112,6 +120,7 @@ sub dequeue_nb
         last if (! @$queue);
         push(@items, shift(@$queue));
     }
+    cond_signal(%$self);  # Unblock possibly waiting threads
     return @items;
 }
 
@@ -135,7 +144,6 @@ sub dequeue_timed
     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);
@@ -187,8 +195,7 @@ sub insert
     # Add previous items back onto the queue
     push(@$queue, @tmp);
 
-    # Soup's up
-    cond_signal(%$self);
+    cond_signal(%$self);  # Unblock possibly waiting threads
 }
 
 # Remove items from anywhere in a queue
@@ -206,7 +213,7 @@ sub extract
         $index += @$queue;
         if ($index < 0) {
             $count += $index;
-            return if ($count <= 0);            # Beyond the head of the queue
+            return if ($count <= 0);           # Beyond the head of the queue
             return $self->dequeue_nb($count);  # Extract from the head
         }
     }
@@ -224,6 +231,8 @@ sub extract
     # Add back any removed items
     push(@$queue, @tmp);
 
+    cond_signal(%$self);  # Unblock possibly waiting threads
+
     # Return single item
     return $items[0] if ($count == 1);
 
@@ -263,14 +272,19 @@ sub _validate_count
     if (! defined($count) ||
         ! looks_like_number($count) ||
         (int($count) != $count) ||
-        ($count < 1))
+        ($count < 1) ||
+        ($$self{'LIMIT'} && $count > $$self{'LIMIT'}))
     {
         require Carp;
         my ($method) = (caller(1))[3];
         my $class_name = ref($self);
         $method =~ s/$class_name\:://;
         $count = 'undef' if (! defined($count));
-        Carp::croak("Invalid 'count' argument ($count) to '$method' method");
+        if ($$self{'LIMIT'} && $count > $$self{'LIMIT'}) {
+            Carp::croak("'count' argument ($count) to '$method' method exceeds queue size limit ($$self{'LIMIT'})");
+        } else {
+            Carp::croak("Invalid 'count' argument ($count) to '$method' method");
+        }
     }
 
     return $count;
@@ -304,7 +318,7 @@ Thread::Queue - Thread-safe queues
 
 =head1 VERSION
 
-This document describes Thread::Queue version 3.11
+This document describes Thread::Queue version 3.12
 
 =head1 SYNOPSIS
 
@@ -494,6 +508,9 @@ C<limit> does not prevent enqueuing items beyond that count:
                          # 'undef')
  $q->limit = 0;          # Queue size is now unlimited
 
+Calling any of the dequeue methods with C<COUNT> greater than a queue's
+C<limit> will generate an error.
+
 =item ->end()
 
 Declares that no more items will be added to the queue.
diff --git a/dist/Thread-Queue/t/01_basic.t b/dist/Thread-Queue/t/01_basic.t
index 4ec51957ae..2983f0b700 100644
--- a/dist/Thread-Queue/t/01_basic.t
+++ b/dist/Thread-Queue/t/01_basic.t
@@ -13,7 +13,7 @@ use threads;
 use Thread::Queue;
 
 if ($] == 5.008) {
-    require './t/test.pl';   # Test::More work-alike for Perl 5.8.0
+    require 't/test.pl';   # Test::More work-alike for Perl 5.8.0
 } else {
     require Test::More;
 }
diff --git a/dist/Thread-Queue/t/02_refs.t b/dist/Thread-Queue/t/02_refs.t
index fdf8f6bad2..0cebdc1db3 100644
--- a/dist/Thread-Queue/t/02_refs.t
+++ b/dist/Thread-Queue/t/02_refs.t
@@ -14,7 +14,7 @@ use threads::shared;
 use Thread::Queue;
 
 if ($] == 5.008) {
-    require './t/test.pl';   # Test::More work-alike for Perl 5.8.0
+    require 't/test.pl';   # Test::More work-alike for Perl 5.8.0
 } else {
     require Test::More;
 }
diff --git a/dist/Thread-Queue/t/03_peek.t b/dist/Thread-Queue/t/03_peek.t
index 29ef75e7fe..d543b59469 100644
--- a/dist/Thread-Queue/t/03_peek.t
+++ b/dist/Thread-Queue/t/03_peek.t
@@ -13,7 +13,7 @@ use threads;
 use Thread::Queue;
 
 if ($] == 5.008) {
-    require './t/test.pl';   # Test::More work-alike for Perl 5.8.0
+    require 't/test.pl';   # Test::More work-alike for Perl 5.8.0
 } else {
     require Test::More;
 }
diff --git a/dist/Thread-Queue/t/05_extract.t b/dist/Thread-Queue/t/05_extract.t
index d8cb417be9..de0e78bfd0 100644
--- a/dist/Thread-Queue/t/05_extract.t
+++ b/dist/Thread-Queue/t/05_extract.t
@@ -13,7 +13,7 @@ use threads;
 use Thread::Queue;
 
 if ($] == 5.008) {
-    require './t/test.pl';   # Test::More work-alike for Perl 5.8.0
+    require 't/test.pl';   # Test::More work-alike for Perl 5.8.0
 } else {
     require Test::More;
 }
diff --git a/dist/Thread-Queue/t/06_insert.t b/dist/Thread-Queue/t/06_insert.t
index 93617e13a3..4f9d1dff5e 100644
--- a/dist/Thread-Queue/t/06_insert.t
+++ b/dist/Thread-Queue/t/06_insert.t
@@ -13,7 +13,7 @@ use threads;
 use Thread::Queue;
 
 if ($] == 5.008) {
-    require './t/test.pl';   # Test::More work-alike for Perl 5.8.0
+    require 't/test.pl';   # Test::More work-alike for Perl 5.8.0
 } else {
     require Test::More;
 }
diff --git a/dist/Thread-Queue/t/07_lock.t b/dist/Thread-Queue/t/07_lock.t
index 633722103c..b20e0604ca 100644
--- a/dist/Thread-Queue/t/07_lock.t
+++ b/dist/Thread-Queue/t/07_lock.t
@@ -14,7 +14,7 @@ use Thread::Queue;
 use Thread::Semaphore;
 
 if ($] == 5.008) {
-    require './t/test.pl';   # Test::More work-alike for Perl 5.8.0
+    require 't/test.pl';   # Test::More work-alike for Perl 5.8.0
 } else {
     require Test::More;
 }
diff --git a/dist/Thread-Queue/t/10_timed.t b/dist/Thread-Queue/t/10_timed.t
index da8b03a7eb..8404720ed6 100644
--- a/dist/Thread-Queue/t/10_timed.t
+++ b/dist/Thread-Queue/t/10_timed.t
@@ -13,7 +13,7 @@ use threads;
 use Thread::Queue;
 
 if ($] == 5.008) {
-    require './t/test.pl';   # Test::More work-alike for Perl 5.8.0
+    require 't/test.pl';   # Test::More work-alike for Perl 5.8.0
 } else {
     require Test::More;
 }
diff --git a/dist/Thread-Queue/t/11_limit.t b/dist/Thread-Queue/t/11_limit.t
index 1bd88b39a1..12f351bc74 100644
--- a/dist/Thread-Queue/t/11_limit.t
+++ b/dist/Thread-Queue/t/11_limit.t
@@ -19,7 +19,7 @@ use Thread::Queue;
 
 use Test::More;
 
-plan tests => 8;
+plan tests => 13;
 
 my $q = Thread::Queue->new();
 my $rpt = Thread::Queue->new();
@@ -82,12 +82,12 @@ $rpt->enqueue($q->pending);
 # q = (4, 5, 'foo'); r = (4, 3, 4, 3)
 
 # Read all items from queue
-my @item = $q->dequeue(3);
-is_deeply(\@item, [4, 5, 'foo'], 'Dequeued 3 items');
+my @items = $q->dequeue(3);
+is_deeply(\@items, [4, 5, 'foo'], 'Dequeued 3 items');
 # Thread is now unblocked
 
-@item = $q->dequeue(2);
-is_deeply(\@item, [6, 7], 'Dequeued 2 items');
+@items = $q->dequeue(2);
+is_deeply(\@items, [6, 7], 'Dequeued 2 items');
 
 # Thread is now unblocked
 # Handshake with thread
@@ -96,6 +96,37 @@ $rpt->enqueue('go');
 # (7) - Done
 $th->join;
 
+# It's an error to call dequeue methods with COUNT > LIMIT
+eval { $q->dequeue(5); };
+like($@, qr/exceeds queue size limit/, $@);
+
+# Bug #120157
+#  Fix deadlock from combination of dequeue_nb, enqueue and queue size limit
+
+# (1) Fill queue
+$q->enqueue(1..3);
+is($q->pending, 3, 'Queue loaded');
+
+# (2) Thread will block trying to add to full queue
+$th = threads->create( sub {
+    $q->enqueue(99);
+    return('OK');
+});
+threads->yield();
+
+# (3) Dequeue an item so that thread can unblock
+is($q->dequeue_nb(), 1, 'Dequeued item');
+
+# (4) Thread unblocks
+is($th->join(), 'OK', 'Thread exited');
+
+# (5) Fetch queue to show thread's item was enqueued
+@items = ();
+while (my $item = $q->dequeue_nb()) {
+    push(@items, $item);
+}
+is_deeply(\@items, [2,3,99], 'Dequeued remaining');
+
 exit(0);
 
 # EOF