Add threading queue shutdown

* Include docs

Author: EpicWink

Doc/library/queue.rst

@@ -93,6 +93,14 @@ The :mod:`queue` module defines the following classes and exceptions:
    on a :class:`Queue` object which is full.
 
 
+.. exception:: ShutDown
+
+   Exception raised when :meth:`~Queue.put` or :meth:`~Queue.get` is called on
+   a :class:`Queue` object which has been shut down.
+
+   .. versionadded:: 3.12
+
+
 .. _queueobjects:
 
 Queue Objects
@@ -135,6 +143,8 @@ provide the public methods described below.
    immediately available, else raise the :exc:`Full` exception (*timeout* is
    ignored in that case).
 
+   Raises :exc:`ShutDown` if the queue has been shut down.
+
 
 .. method:: Queue.put_nowait(item)
 
@@ -155,6 +165,9 @@ provide the public methods described below.
    an uninterruptible wait on an underlying lock.  This means that no exceptions
    can occur, and in particular a SIGINT will not trigger a :exc:`KeyboardInterrupt`.
 
+   Raises :exc:`ShutDown` if the queue has been shut down and is empty, or if
+   the queue has been shut down immediately.
+
 
 .. method:: Queue.get_nowait()
 
@@ -177,6 +190,8 @@ fully processed by daemon consumer threads.
    Raises a :exc:`ValueError` if called more times than there were items placed in
    the queue.
 
+   Raises :exc:`ShutDown` if the queue has been shut down immediately.
+
 
 .. method:: Queue.join()
 
@@ -187,6 +202,8 @@ fully processed by daemon consumer threads.
    indicate that the item was retrieved and all work on it is complete.  When the
    count of unfinished tasks drops to zero, :meth:`join` unblocks.
 
+   Raises :exc:`ShutDown` if the queue has been shut down immediately.
+
 
 Example of how to wait for enqueued tasks to be completed::
 
@@ -214,6 +231,25 @@ Example of how to wait for enqueued tasks to be completed::
     print('All work completed')
 
 
+Terminating queues
+^^^^^^^^^^^^^^^^^^
+
+:class:`Queue` objects can be made to prevent further interaction by shutting
+them down.
+
+.. method:: Queue.shutdown(immediate=False)
+
+   Shut-down the queue, making queue gets and puts raise :exc:`ShutDown`.
+
+   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() will be unblocked, and also get()
+   and join() if *immediate* is true.
+
+   .. versionadded:: 3.12
+
+
 SimpleQueue Objects
 -------------------
 

Lib/asyncio/queues.py

@@ -1,4 +1,11 @@
-__all__ = ('Queue', 'PriorityQueue', 'LifoQueue', 'QueueFull', 'QueueEmpty')
+__all__ = (
+    'Queue',
+    'PriorityQueue',
+    'LifoQueue',
+    'QueueFull',
+    'QueueEmpty',
+    'QueueShutDown',
+)
 
 import collections
 import heapq
@@ -18,6 +25,16 @@ class QueueFull(Exception):
     pass
 
 
+class QueueShutDown(Exception):
+    """Raised when putting on to or getting from a shut-down Queue."""
+    pass
+
+
+_queue_alive = "alive"
+_queue_shutdown = "shutdown"
+_queue_shutdown_immediate = "shutdown-immediate"
+
+
 class Queue(mixins._LoopBoundMixin):
     """A queue, useful for coordinating producer and consumer coroutines.
 
@@ -41,6 +58,7 @@ def __init__(self, maxsize=0):
         self._finished = locks.Event()
         self._finished.set()
         self._init(maxsize)
+        self.shutdown_state = _queue_alive
 
     # These three are overridable in subclasses.
 
@@ -113,6 +131,8 @@ async def put(self, item):
         Put an item into the queue. If the queue is full, wait until a free
         slot is available before adding item.
         """
+        if self.shutdown_state != _queue_alive:
+            raise QueueShutDown
         while self.full():
             putter = self._get_loop().create_future()
             self._putters.append(putter)
@@ -132,13 +152,17 @@ async def put(self, item):
                     # the call.  Wake up the next in line.
                     self._wakeup_next(self._putters)
                 raise
+            if self.shutdown_state != _queue_alive:
+                raise QueueShutDown
         return self.put_nowait(item)
 
     def put_nowait(self, item):
         """Put an item into the queue without blocking.
 
         If no free slot is immediately available, raise QueueFull.
         """
+        if self.shutdown_state != _queue_alive:
+            raise QueueShutDown
         if self.full():
             raise QueueFull
         self._put(item)
@@ -151,7 +175,11 @@ async def get(self):
 
         If queue is empty, wait until an item is available.
         """
+        if self.shutdown_state == _queue_shutdown_immediate:
+            raise QueueShutDown
         while self.empty():
+            if self.shutdown_state != _queue_alive:
+                raise QueueShutDown
             getter = self._get_loop().create_future()
             self._getters.append(getter)
             try:
@@ -170,6 +198,8 @@ async def get(self):
                     # the call.  Wake up the next in line.
                     self._wakeup_next(self._getters)
                 raise
+            if self.shutdown_state == _queue_shutdown_immediate:
+                raise QueueShutDown
         return self.get_nowait()
 
     def get_nowait(self):
@@ -178,7 +208,11 @@ def get_nowait(self):
         Return an item if one is immediately available, else raise QueueEmpty.
         """
         if self.empty():
+            if self.shutdown_state != _queue_alive:
+                raise QueueShutDown
             raise QueueEmpty
+        elif self.shutdown_state == _queue_shutdown_immediate:
+            raise QueueShutDown
         item = self._get()
         self._wakeup_next(self._putters)
         return item
@@ -214,6 +248,27 @@ async def join(self):
         if self._unfinished_tasks > 0:
             await self._finished.wait()
 
+    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() will be unblocked, and also get()
+        and join() if 'immediate'. The QueueShutDown exception is raised.
+        """
+        if immediate:
+            self.shutdown_state = _queue_shutdown_immediate
+            while self._getters:
+                getter = self._getters.popleft()
+                if not getter.done():
+                    getter.set_result(None)
+        else:
+            self.shutdown_state = _queue_shutdown
+        while self._putters:
+            putter = self._putters.popleft()
+            if not putter.done():
+                putter.set_result(None)
 
 class PriorityQueue(Queue):
     """A subclass of Queue; retrieves entries in priority order (lowest first).

Lib/multiprocessing/queues.py

@@ -17,8 +17,9 @@
 import types
 import weakref
 import errno
+import ctypes
 
-from queue import Empty, Full
+from queue import Empty, Full, ShutDown
 
 import _multiprocessing
 
@@ -28,6 +29,10 @@
 
 from .util import debug, info, Finalize, register_after_fork, is_exiting
 
+_queue_alive = 0
+_queue_shutdown = 1
+_queue_shutdown_immediate = 2
+
 #
 # Queue type using a pipe, buffer and thread
 #
@@ -50,6 +55,9 @@ def __init__(self, maxsize=0, *, ctx):
         # For use by concurrent.futures
         self._ignore_epipe = False
         self._reset()
+        self._shutdown_state = context._default_context.Value(
+            ctypes.c_uint8, lock=self._rlock
+        )
 
         if sys.platform != 'win32':
             register_after_fork(self, Queue._after_fork)
@@ -86,20 +94,28 @@ def _reset(self, after_fork=False):
     def put(self, obj, block=True, timeout=None):
         if self._closed:
             raise ValueError(f"Queue {self!r} is closed")
+        if self._shutdown_state.value != _queue_alive:
+            raise ShutDown
         if not self._sem.acquire(block, timeout):
             raise Full
 
         with self._notempty:
+            if self._shutdown_state.value != _queue_alive:
+                raise ShutDown
             if self._thread is None:
                 self._start_thread()
             self._buffer.append(obj)
             self._notempty.notify()
 
     def get(self, block=True, timeout=None):
+        if self._shutdown_state.value == _queue_shutdown_immediate:
+            raise ShutDown
         if self._closed:
             raise ValueError(f"Queue {self!r} is closed")
         if block and timeout is None:
             with self._rlock:
+                if self._shutdown_state.value != _queue_alive:
+                    raise ShutDown
                 res = self._recv_bytes()
             self._sem.release()
         else:
@@ -111,13 +127,19 @@ def get(self, block=True, timeout=None):
                 if block:
                     timeout = deadline - time.monotonic()
                     if not self._poll(timeout):
+                        if self._shutdown_state.value != _queue_alive:
+                            raise ShutDown
                         raise Empty
+                    if self._shutdown_state.value != _queue_alive       :
+                        raise ShutDown
                 elif not self._poll():
                     raise Empty
                 res = self._recv_bytes()
                 self._sem.release()
             finally:
                 self._rlock.release()
+            if self._shutdown_state.value == _queue_shutdown:
+                raise ShutDown
         # unserialize the data after having released the lock
         return _ForkingPickler.loads(res)
 
@@ -329,6 +351,8 @@ def task_done(self):
 
     def join(self):
         with self._cond:
+            if self._shutdown_state.value == _queue_shutdown_immediate:
+                return
             if not self._unfinished_tasks._semlock._is_zero():
                 self._cond.wait()