diff options
| -rw-r--r-- | Doc/lib/libqueue.tex | 46 | ||||
| -rw-r--r-- | Lib/Queue.py | 86 | ||||
| -rw-r--r-- | Lib/test/test_queue.py | 39 | ||||
| -rw-r--r-- | Misc/NEWS | 2 |
4 files changed, 137 insertions, 36 deletions
diff --git a/Doc/lib/libqueue.tex b/Doc/lib/libqueue.tex index 3a1d446..0770bfe 100644 --- a/Doc/lib/libqueue.tex +++ b/Doc/lib/libqueue.tex @@ -54,35 +54,47 @@ semantics, this number is not reliable. \end{methoddesc} \begin{methoddesc}{empty}{} -Return \code{1} if the queue is empty, \code{0} otherwise. Because -of multithreading semantics, this is not reliable. +Return \code{True} if the queue is empty, \code{False} otherwise. +Becauseof multithreading semantics, this is not reliable. \end{methoddesc} \begin{methoddesc}{full}{} -Return \code{1} if the queue is full, \code{0} otherwise. Because of -multithreading semantics, this is not reliable. +Return \code{True} if the queue is full, \code{False} otherwise. +Because of multithreading semantics, this is not reliable. \end{methoddesc} -\begin{methoddesc}{put}{item\optional{, block}} -Put \var{item} into the queue. If optional argument \var{block} is 1 -(the default), block if necessary until a free slot is available. -Otherwise (\var{block} is 0), put \var{item} on the queue if a free +\begin{methoddesc}{put}{item\optional{, block\optional{, timeout}}} +Put \var{item} into the queue. If optional args \var{block} is true +and \var{timeout} is None (the default), block if necessary until a +free slot is available. If \var{timeout} is a positive number, it +blocks at most \var{timeout} seconds and raises the \exception{Full} +exception if no free slot was available within that time. +Otherwise (\var{block} is false), put an item on the queue if a free slot is immediately available, else raise the \exception{Full} -exception. +exception (\var{timeout} is ignored in that case). + +\versionadded[the timeout parameter]{2.3} + \end{methoddesc} \begin{methoddesc}{put_nowait}{item} -Equivalent to \code{put(\var{item}, 0)}. +Equivalent to \code{put(\var{item}, False)}. \end{methoddesc} -\begin{methoddesc}{get}{\optional{block}} -Remove and return an item from the queue. If optional argument -\var{block} is 1 (the default), block if necessary until an item is -available. Otherwise (\var{block} is 0), return an item if one is -immediately available, else raise the -\exception{Empty} exception. +\begin{methoddesc}{get}{\optional{block\optional{, timeout}}} +Remove and return an item from the queue. If optional args +\var{block} is true and \var{timeout} is None (the default), +block if necessary until an item is available. If \var{timeout} is +a positive number, it blocks at most \var{timeout} seconds and raises +the \exception{Empty} exception if no item was available within that +time. Otherwise (\var{block} is false), return an item if one is +immediately available, else raise the \exception{Empty} exception +(\var{timeout} is ignored in that case). + +\versionadded[the timeout parameter]{2.3} + \end{methoddesc} \begin{methoddesc}{get_nowait}{} -Equivalent to \code{get(0)}. +Equivalent to \code{get(False)}. \end{methoddesc} 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 @@ class Queue: 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 @@ class Queue: 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 @@ class Queue: 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 @@ def FailingQueueTest(q): 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 @@ def FailingQueueTest(q): # 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 @@ def FailingQueueTest(q): 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 @@ def SimpleQueueTest(q): 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 @@ def SimpleQueueTest(q): 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) @@ -348,6 +348,8 @@ Extension modules Library ------- +- Queue.Queue.{put,get} now support an optional timeout argument. + - Various features of Tk 8.4 are exposed in Tkinter.py. The multiple option of tkFileDialog is exposed as function askopenfile{,name}s. |