| Subject: | deadlock using dequeue_nb on size limited queue |
| Date: | Thu, 9 Feb 2017 08:51:32 -0600 |
| To: | bug-thread-queue [...] rt.cpan.org |
| From: | Chad Fox <dchidelf [...] gmail.com> |
Apologies if this is a resubmission, but it did not appear as though my
original from chad [at] gigapogo.com made it through.
Issue
===========
Thread-Queue-3.11 can deadlock when using dequeue_nb() on a queue with a
set size limit.
The dequeue() and dequeue_timed() subs will signal a blocked enqueue, but
dequeue_nb() will not.
Test code
===========
This isn't representative of how someone may actually use Thread/Queue,
but recreates the issue pretty consistently
------------------------------------------------------------
-------------------
use threads;
use threads::shared;
use Thread::Queue;
my $thr;
my $q = Thread::Queue->new();
$q->limit = 5; # set queue limit of 5
$q->enqueue(1..5); # fill queue to capacity
my $sem :shared = 0; # "semaphore" to signal enqueue thread has started
{
lock($sem);
# Create a thread that will enqueue 1 item
# the queue is full, so it will block until we dequeue an item
$thr = threads->create(sub {
my ($sem) = @_;
{
lock($$sem);
cond_signal($$sem);
}
# main thread yields to us, via a sleep
$q->enqueue(1); # blocks until main thread dequeues
$q->end();
}, \$sem);
cond_wait($sem);
}
# at this point the enqueue thread just signaled the main thread
# we are sure they are synchronized at this point, but want to make
# sure the enqueue thread goes first, sleep for a second
print "Yield for 1 second\n";
sleep 1;
# dequeue all items from the queue
while (defined($q->dequeue_nb())) { }
printf "Queue now contains %d items\n", $q->pending();
# enqueue should have become unblocked, so join shouldn't hang
$thr->join();
print "SUCCESS: enqueue unblocked!\n";
------------------------------------------------------------
-------------------
Resolution
===========
By moving the cond_signal into dequeue_nb(), a blocked enqueue can be
signaled using any of the dequeue methods.
An additional check in _validate_count prevents a deadlock due to using a
count value greater than the queue limit.
Patch
===========
--- Thread/Queue.pm 2016-12-16 20:56:35.000000000 -0500
+++ Thread/Queue.pm 2016-12-16 20:56:35.000000000 -0500
@@ -80,11 +80,14 @@
# 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'});
+ # propagate a signal to other threads if queue ended, items remain,
+ # or a thread is potentially blocked on an enqueue at size limit
+ cond_signal(%$self) if ((@$queue >= $count) || $$self{'ENDED'});
+
# Return single item
return shift(@$queue) if ($count == 1);
@@ -103,6 +106,10 @@
my $count = @_ ? $self->_validate_count(shift) : 1;
+ # propagate a signal to other threads if queue ended, items remain,
+ # or a thread is potentially blocked on an enqueue at size limit
+ cond_signal(%$self) if ((@$queue >= $count) || $$self{'ENDED'});
+
# Return single item
return shift(@$queue) if ($count == 1);
@@ -135,7 +142,6 @@
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);
@@ -263,7 +269,8 @@
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];