python
diff --git a/‎Lib/queue.py
Copy file name to clipboardExpand all lines: Lib/queue.py
+46Lines changed: 46 additions & 0 deletions b/‎Lib/queue.py
Copy file name to clipboardExpand all lines: Lib/queue.py
+46Lines changed: 46 additions & 0 deletions
@@ -25,6 +25,15 @@ class Full(Exception):
     pass
 
 
+class ShutDown(Exception):
+    '''Raised when put/get with shut-down queue.'''
+
+
+_queue_alive = "alive"
+_queue_shutdown = "shutdown"
+_queue_shutdown_immediate = "shutdown-immediate"
+
+
 class Queue:
     '''Create a queue object with a given maximum size.
 
@@ -54,6 +63,9 @@ def __init__(self, maxsize=0):
         self.all_tasks_done = threading.Condition(self.mutex)
         self.unfinished_tasks = 0
 
+        # Queue shut-down state
+        self.shutdown_state = _queue_alive
+
     def task_done(self):
         '''Indicate that a formerly enqueued task is complete.
 
@@ -87,6 +99,8 @@ def join(self):
         '''
         with self.all_tasks_done:
             while self.unfinished_tasks:
+                if self.shutdown_state == _queue_shutdown_immediate:
+                    return
                 self.all_tasks_done.wait()
 
     def qsize(self):
@@ -130,6 +144,8 @@ def put(self, item, block=True, timeout=None):
         is immediately available, else raise the Full exception ('timeout'
         is ignored in that case).
         '''
+        if self.shutdown_state != _queue_alive:
+            raise ShutDown
         with self.not_full:
             if self.maxsize > 0:
                 if not block:
@@ -138,6 +154,8 @@ def put(self, item, block=True, timeout=None):
                 elif timeout is None:
                     while self._qsize() >= self.maxsize:
                         self.not_full.wait()
+                        if self.shutdown_state != _queue_alive:
+                            raise ShutDown
                 elif timeout < 0:
                     raise ValueError("'timeout' must be a non-negative number")
                 else:
@@ -147,6 +165,8 @@ def put(self, item, block=True, timeout=None):
                         if remaining <= 0.0:
                             raise Full
                         self.not_full.wait(remaining)
+                        if self.shutdown_state != _queue_alive:
+                            raise ShutDown
             self._put(item)
             self.unfinished_tasks += 1
             self.not_empty.notify()
@@ -162,13 +182,17 @@ def get(self, block=True, timeout=None):
         available, else raise the Empty exception ('timeout' is ignored
         in that case).
         '''
+        if self.shutdown_state == _queue_shutdown_immediate:
+            raise ShutDown
         with self.not_empty:
             if not block:
                 if not self._qsize():
                     raise Empty
             elif timeout is None:
                 while not self._qsize():
                     self.not_empty.wait()
+                    if self.shutdown_state == _queue_shutdown_immediate:
+                        raise ShutDown
             elif timeout < 0:
                 raise ValueError("'timeout' must be a non-negative number")
             else:
@@ -178,6 +202,8 @@ def get(self, block=True, timeout=None):
                     if remaining <= 0.0:
                         raise Empty
                     self.not_empty.wait(remaining)
+                    if self.shutdown_state == _queue_shutdown_immediate:
+                        raise ShutDown
             item = self._get()
             self.not_full.notify()
             return item
@@ -198,6 +224,26 @@ def get_nowait(self):
         '''
         return self.get(block=False)
 
+    def shutdown(self, immediate=False):
+        '''Shut-down the queue, making queue gets and puts raise.
+
+        By default, gets will only raise once the queue is empty. Set
+        'immediate' to True to make gets raise immediately instead.
+
+        All blocked callers of put(), get() and join() will be
+        unblocked. The ShutDown exception is raised.
+        '''
+        if immediate:
+            self.shutdown_state = _queue_shutdown_immediate
+            with self.not_empty:
+                self.not_empty.notify_all()
+            with self.all_tasks_done:
+                self.all_tasks_done.notify_all()
+        else:
+            self.shutdown_state = _queue_shutdown
+        with self.not_full:
+            self.not_full.notify_all()
+
     # Override these methods to implement other queue organizations
     # (e.g. stack or priority queue).
     # These will only be called with appropriate locks held