gh-96471: Add queue shutdown, next step.

Doc/library/asyncio-queue.rst

@@ -62,6 +62,9 @@ Queue
       Remove and return an item from the queue. If queue is empty,
       wait until an item is available.
 
+      Raises :exc:`QueueShutDown` if the queue has been shut down and
+      is empty, or if the queue has been shut down immediately.
+
    .. method:: get_nowait()
 
       Return an item if one is immediately available, else raise
@@ -77,11 +80,16 @@ Queue
       work on it is complete.  When the count of unfinished tasks drops
       to zero, :meth:`join` unblocks.
 
+      Raises :exc:`QueueShutDown` if the queue has been shut down
+      immediately.
+
    .. coroutinemethod:: put(item)
 
       Put an item into the queue. If the queue is full, wait until a
       free slot is available before adding the item.
 
+      Raises :exc:`QueueShutDown` if the queue has been shut down.
+
    .. method:: put_nowait(item)
 
       Put an item into the queue without blocking.
@@ -92,6 +100,19 @@ Queue
 
       Return the number of items in the queue.
 
+   .. method:: shutdown(immediate=False)
+
+      Shut-down the queue, making queue gets and puts raise
+      :exc:`QueueShutDown`.
+
+      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
+
    .. method:: task_done()
 
       Indicate that a formerly enqueued task is complete.
@@ -108,6 +129,9 @@ Queue
       Raises :exc:`ValueError` if called more times than there were
       items placed in the queue.
 
+      Raises :exc:`QueueShutDown` if the queue has been shut down
+      immediately.
+
 
 Priority Queue
 ==============
@@ -145,6 +169,14 @@ Exceptions
    on a queue that has reached its *maxsize*.
 
 
+.. exception:: QueueShutDown
+
+   Exception raised when getting an item from or putting an item onto a
+   queue which has been shut down.
+
+   .. versionadded:: 3.13
+
+
 Examples
 ========
 

Doc/library/multiprocessing.rst

@@ -845,7 +845,8 @@ For an example of the usage of queues for interprocess communication see
       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 :exc:`queue.Full` exception (*timeout* is
-      ignored in that case).
+      ignored in that case).  Raises :exc:`ShutDown` if the queue has been shut
+      down.
 
       .. versionchanged:: 3.8
          If the queue is closed, :exc:`ValueError` is raised instead of
@@ -863,7 +864,9 @@ For an example of the usage of queues for interprocess communication see
       it blocks at most *timeout* seconds and raises the :exc:`queue.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
-      :exc:`queue.Empty` exception (*timeout* is ignored in that case).
+      :exc:`queue.Empty` exception (*timeout* is ignored in that case).  Raises
+      :exc:`queue.ShutDown` if the queue has been shut down and is empty, or if
+      the queue has been shut down immediately.
 
       .. versionchanged:: 3.8
          If the queue is closed, :exc:`ValueError` is raised instead of
@@ -873,6 +876,19 @@ For an example of the usage of queues for interprocess communication see
 
       Equivalent to ``get(False)``.
 
+   .. method:: shutdown(immediate=False)
+
+      Shut-down the queue, making queue gets and puts raise
+      :exc:`queue.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
+
    :class:`multiprocessing.Queue` has a few additional methods not found in
    :class:`queue.Queue`.  These methods are usually unnecessary for most
    code:
@@ -962,6 +978,8 @@ For an example of the usage of queues for interprocess communication see
       Raises a :exc:`ValueError` if called more times than there were items
       placed in the queue.
 
+      Raises :exc:`queue.ShutDown` if the queue has been shut down immediately.
+
 
    .. method:: join()
 
@@ -973,6 +991,8 @@ For an example of the usage of queues for interprocess communication see
       it is complete.  When the count of unfinished tasks drops to zero,
       :meth:`~queue.Queue.join` unblocks.
 
+      Raises :exc:`queue.ShutDown` if the queue has been shut down immediately.
+
 
 Miscellaneous
 ~~~~~~~~~~~~~

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
 -------------------