Patch #572628: Optional timeouts for put and get.
diff --git a/Lib/Queue.py b/Lib/Queue.py
index cd035da..39c86f2 100644
--- a/Lib/Queue.py
+++ b/Lib/Queue.py
@@ -1,5 +1,7 @@
"""A multi-producer, multi-consumer queue."""
+from time import time as _time, sleep as _sleep
+
class Empty(Exception):
"Exception raised by Queue.get(block=0)/get_nowait()."
pass
@@ -29,29 +31,54 @@
return n
def empty(self):
- """Return 1 if the queue is empty, 0 otherwise (not reliable!)."""
+ """Return True if the queue is empty, False otherwise (not reliable!)."""
self.mutex.acquire()
n = self._empty()
self.mutex.release()
return n
def full(self):
- """Return 1 if the queue is full, 0 otherwise (not reliable!)."""
+ """Return True if the queue is full, False otherwise (not reliable!)."""
self.mutex.acquire()
n = self._full()
self.mutex.release()
return n
- def put(self, item, block=1):
+ def put(self, item, block=True, timeout=None):
"""Put an item into the queue.
- If optional arg 'block' is 1 (the default), block if
- necessary until a free slot is available. Otherwise (block
- is 0), put an item on the queue if a free slot is immediately
- available, else raise the Full exception.
+ If optional args 'block' is true and 'timeout' is None (the default),
+ block if necessary until a free slot is available. If 'timeout' is
+ a positive number, it blocks at most 'timeout' seconds and raises
+ the Full exception if no free slot was available within that time.
+ Otherwise ('block' is false), put an item on the queue if a free slot
+ is immediately available, else raise the Full exception ('timeout'
+ is ignored in that case).
"""
if block:
- self.fsema.acquire()
+ if timeout is None:
+ # blocking, w/o timeout, i.e. forever
+ self.fsema.acquire()
+ elif timeout >= 0:
+ # waiting max. 'timeout' seconds.
+ # this code snipped is from threading.py: _Event.wait():
+ # Balancing act: We can't afford a pure busy loop, so we
+ # have to sleep; but if we sleep the whole timeout time,
+ # we'll be unresponsive. The scheme here sleeps very
+ # little at first, longer as time goes on, but never longer
+ # than 20 times per second (or the timeout time remaining).
+ delay = 0.0005 # 500 us -> initial delay of 1 ms
+ endtime = _time() + timeout
+ while True:
+ if self.fsema.acquire(0):
+ break
+ remaining = endtime - _time()
+ if remaining <= 0: #time is over and no slot was free
+ raise Full
+ delay = min(delay * 2, remaining, .05)
+ _sleep(delay) #reduce CPU usage by using a sleep
+ else:
+ raise ValueError("'timeout' must be a positive number")
elif not self.fsema.acquire(0):
raise Full
self.mutex.acquire()
@@ -80,18 +107,43 @@
Only enqueue the item if a free slot is immediately available.
Otherwise raise the Full exception.
"""
- return self.put(item, 0)
+ return self.put(item, False)
- def get(self, block=1):
+ def get(self, block=True, timeout=None):
"""Remove and return an item from the queue.
- If optional arg 'block' is 1 (the default), block if
- necessary until an item is available. Otherwise (block is 0),
- return an item if one is immediately available, else raise the
- Empty exception.
+ If optional args 'block' is true and 'timeout' is None (the default),
+ block if necessary until an item is available. If 'timeout' is
+ a positive number, it blocks at most 'timeout' seconds and raises
+ the Empty exception if no item was available within that time.
+ Otherwise ('block' is false), return an item if one is immediately
+ available, else raise the Empty exception ('timeout' is ignored
+ in that case).
"""
if block:
- self.esema.acquire()
+ if timeout is None:
+ # blocking, w/o timeout, i.e. forever
+ self.esema.acquire()
+ elif timeout >= 0:
+ # waiting max. 'timeout' seconds.
+ # this code snipped is from threading.py: _Event.wait():
+ # Balancing act: We can't afford a pure busy loop, so we
+ # have to sleep; but if we sleep the whole timeout time,
+ # we'll be unresponsive. The scheme here sleeps very
+ # little at first, longer as time goes on, but never longer
+ # than 20 times per second (or the timeout time remaining).
+ delay = 0.0005 # 500 us -> initial delay of 1 ms
+ endtime = _time() + timeout
+ while 1:
+ if self.esema.acquire(0):
+ break
+ remaining = endtime - _time()
+ if remaining <= 0: #time is over and no element arrived
+ raise Empty
+ delay = min(delay * 2, remaining, .05)
+ _sleep(delay) #reduce CPU usage by using a sleep
+ else:
+ raise ValueError("'timeout' must be a positive number")
elif not self.esema.acquire(0):
raise Empty
self.mutex.acquire()
@@ -115,10 +167,10 @@
def get_nowait(self):
"""Remove and return an item from the queue without blocking.
- Only get an item if one is immediately available. Otherwise
+ Only get an item if one is immediately available. Otherwise
raise the Empty exception.
"""
- return self.get(0)
+ return self.get(False)
# Override these methods to implement other queue organizations
# (e.g. stack or priority queue).
diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py
index a2c744f..19cd321 100644
--- a/Lib/test/test_queue.py
+++ b/Lib/test/test_queue.py
@@ -60,17 +60,23 @@
raise RuntimeError, "Call this function with an empty queue"
for i in range(queue_size-1):
q.put(i)
- q.fail_next_put = True
# Test a failing non-blocking put.
+ q.fail_next_put = True
try:
q.put("oops", block=0)
raise TestFailed("The queue didn't fail when it should have")
except FailingQueueException:
pass
+ q.fail_next_put = True
+ try:
+ q.put("oops", timeout=0.1)
+ raise TestFailed("The queue didn't fail when it should have")
+ except FailingQueueException:
+ pass
q.put("last")
verify(q.full(), "Queue should be full")
- q.fail_next_put = True
# Test a failing blocking put
+ q.fail_next_put = True
try:
_doBlockingTest( q.put, ("full",), q.get, ())
raise TestFailed("The queue didn't fail when it should have")
@@ -79,6 +85,16 @@
# Check the Queue isn't damaged.
# put failed, but get succeeded - re-add
q.put("last")
+ # Test a failing timeout put
+ q.fail_next_put = True
+ try:
+ _doBlockingTest( q.put, ("full", True, 0.2), q.get, ())
+ raise TestFailed("The queue didn't fail when it should have")
+ except FailingQueueException:
+ pass
+ # Check the Queue isn't damaged.
+ # put failed, but get succeeded - re-add
+ q.put("last")
verify(q.full(), "Queue should be full")
q.get()
verify(not q.full(), "Queue should not be full")
@@ -98,6 +114,13 @@
except FailingQueueException:
pass
verify(not q.empty(), "Queue should not be empty")
+ q.fail_next_get = True
+ try:
+ q.get(timeout=0.1)
+ raise TestFailed("The queue didn't fail when it should have")
+ except FailingQueueException:
+ pass
+ verify(not q.empty(), "Queue should not be empty")
q.get()
verify(q.empty(), "Queue should be empty")
q.fail_next_get = True
@@ -128,8 +151,14 @@
raise TestFailed("Didn't appear to block with a full queue")
except Queue.Full:
pass
+ try:
+ q.put("full", timeout=0.1)
+ raise TestFailed("Didn't appear to time-out with a full queue")
+ except Queue.Full:
+ pass
# Test a blocking put
_doBlockingTest( q.put, ("full",), q.get, ())
+ _doBlockingTest( q.put, ("full", True, 0.2), q.get, ())
# Empty it
for i in range(queue_size):
q.get()
@@ -139,8 +168,14 @@
raise TestFailed("Didn't appear to block with an empty queue")
except Queue.Empty:
pass
+ try:
+ q.get(timeout=0.1)
+ raise TestFailed("Didn't appear to time-out with an empty queue")
+ except Queue.Empty:
+ pass
# Test a blocking get
_doBlockingTest( q.get, (), q.put, ('empty',))
+ _doBlockingTest( q.get, (True, 0.2), q.put, ('empty',))
def test():
q=Queue.Queue(queue_size)