gh-96471: Add threading queue shutdown

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.13
+
+
 .. _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.13
+
+
 SimpleQueue Objects
 -------------------
 

Lib/queue.py

@@ -25,6 +25,14 @@ 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 +62,9 @@ def __init__(self, maxsize=0):
         self.all_tasks_done = threading.Condition(self.mutex)
         self.unfinished_tasks = 0
 
+        # Queue shutdown state
+        self.shutdown_state = _queue_alive
+
     def task_done(self):
         '''Indicate that a formerly enqueued task is complete.
 
@@ -67,8 +78,12 @@ def task_done(self):
 
         Raises a ValueError if called more times than there were items
         placed in the queue.
+
+        Raises ShutDown if the queue has been shut down immediately.
         '''
         with self.all_tasks_done:
+            if self._is_shutdown_immediate():
+                raise ShutDown
             unfinished = self.unfinished_tasks - 1
             if unfinished <= 0:
                 if unfinished < 0:
@@ -84,10 +99,16 @@ def join(self):
         to indicate the item was retrieved and all work on it is complete.
 
         When the count of unfinished tasks drops to zero, join() unblocks.
+
+        Raises ShutDown if the queue has been shut down immediately.
         '''
         with self.all_tasks_done:
+            if self._is_shutdown_immediate():
+                raise ShutDown
             while self.unfinished_tasks:
                 self.all_tasks_done.wait()
+                if self._is_shutdown_immediate():
+                    raise ShutDown
 
     def qsize(self):
         '''Return the approximate size of the queue (not reliable!).'''
@@ -129,15 +150,21 @@ def put(self, item, block=True, timeout=None):
         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).
+
+        Raises ShutDown if the queue has been shut down.
         '''
         with self.not_full:
+            if not self._is_alive():
+                raise ShutDown
             if self.maxsize > 0:
                 if not block:
                     if self._qsize() >= self.maxsize:
                         raise Full
                 elif timeout is None:
                     while self._qsize() >= self.maxsize:
                         self.not_full.wait()
+                        if not self._is_alive():
+                            raise ShutDown
                 elif timeout < 0:
                     raise ValueError("'timeout' must be a non-negative number")
                 else:
@@ -147,6 +174,8 @@ def put(self, item, block=True, timeout=None):
                         if remaining <= 0.0:
                             raise Full
                         self.not_full.wait(remaining)
+                        if not self._is_alive():
+                            raise ShutDown
             self._put(item)
             self.unfinished_tasks += 1
             self.not_empty.notify()
@@ -161,14 +190,22 @@ def get(self, block=True, timeout=None):
         Otherwise ('block' is false), return an item if one is immediately
         available, else raise the Empty exception ('timeout' is ignored
         in that case).
+
+        Raises ShutDown if the queue has been shut down and is empty,
+        or if the queue has been shut down immediately.
         '''
         with self.not_empty:
+            if self._is_shutdown_immediate() or\
+                (self._is_shutdown() and not self._qsize()):
+                raise ShutDown
             if not block:
                 if not self._qsize():
                     raise Empty
             elif timeout is None:
                 while not self._qsize():
                     self.not_empty.wait()
+                    if self._is_shutdown_immediate():
+                        raise ShutDown
             elif timeout < 0:
                 raise ValueError("'timeout' must be a non-negative number")
             else:
@@ -178,6 +215,8 @@ def get(self, block=True, timeout=None):
                     if remaining <= 0.0:
                         raise Empty
                     self.not_empty.wait(remaining)
+                    if self._is_shutdown_immediate():
+                        raise ShutDown
             item = self._get()
             self.not_full.notify()
             return item
@@ -198,6 +237,43 @@ 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() will be unblocked, and also get()
+        and join() if 'immediate'. The ShutDown exception is raised.
+        '''
+        with self.mutex:
+            if self._is_shutdown_immediate():
+                return
+            if immediate:
+                self._set_shutdown_immediate()
+                self.not_empty.notify_all()
+                # release all blocked threads in `join()`
+                self.all_tasks_done.notify_all()
+            else:
+                self._set_shutdown()
+            self.not_full.notify_all()
+
+    def _is_alive(self):
+        return self.shutdown_state == _queue_alive
+
+    def _is_shutdown(self):
+        return self.shutdown_state == _queue_shutdown
+
+    def _is_shutdown_immediate(self):
+        return self.shutdown_state == _queue_shutdown_immediate
+
+    def _set_shutdown(self):
+        self.shutdown_state = _queue_shutdown
+
+    def _set_shutdown_immediate(self):
+        self.shutdown_state = _queue_shutdown_immediate
+
+
     # Override these methods to implement other queue organizations
     # (e.g. stack or priority queue).
     # These will only be called with appropriate locks held