Mercurial > cpython-lru_cache-weakref
diff Lib/Queue.py @ 37198:bed3f52f0dc6 legacy-trunk
SF Patch #1455676: Simplify using Queues with daemon consumer threads
Adds join() and task_done() methods to track when all enqueued tasks have
been gotten and fully processed by daemon consumer threads.
| author | Raymond Hettinger <python@rcn.com> |
|---|---|
| date | Fri, 24 Mar 2006 20:43:29 +0000 |
| parents | d42efc2b45e9 |
| children | 49622d65c29d |
line wrap: on
line diff
--- a/Lib/Queue.py Fri Mar 24 13:36:33 2006 +0000 +++ b/Lib/Queue.py Fri Mar 24 20:43:29 2006 +0000 @@ -35,6 +35,50 @@ # Notify not_full whenever an item is removed from the queue; # a thread waiting to put is notified then. self.not_full = threading.Condition(self.mutex) + # Notify all_tasks_done whenever the number of unfinished tasks + # drops to zero; thread waiting to join() is notified to resume + self.all_tasks_done = threading.Condition(self.mutex) + self.unfinished_tasks = 0 + + def task_done(self): + """Indicate that a formerly enqueued task is complete. + + Used by Queue consumer threads. For each get() used to fetch a task, + a subsequent call to task_done() tells the queue that the processing + on the task is complete. + + If a join() is currently blocking, it will resume when all items + have been processed (meaning that a task_done() call was received + for every item that had been put() into the queue). + + Raises a ValueError if called more times than there were items + placed in the queue. + """ + self.all_tasks_done.acquire() + try: + self.unfinished_tasks = unfinished = self.unfinished_tasks - 1 + if unfinished <= 0: + if unfinished < 0: + raise ValueError('task_done() called too many times') + self.all_tasks_done.notifyAll() + finally: + self.all_tasks_done.release() + + def join(self): + """Blocks until all items in the Queue have been gotten and processed. + + The count of unfinished tasks goes up whenever an item is added to the + queue. The count goes down whenever a consumer thread calls task_done() + to indicate the item was retrieved and all work on it is complete. + + When the count of unfinished tasks drops to zero, join() unblocks. + """ + self.all_tasks_done.acquire() + try: + while self.unfinished_tasks: + self.all_tasks_done.wait() + finally: + self.all_tasks_done.release() def qsize(self): """Return the approximate size of the queue (not reliable!).""" @@ -86,6 +130,7 @@ raise Full self.not_full.wait(remaining) self._put(item) + self.unfinished_tasks += 1 self.not_empty.notify() finally: self.not_full.release()
