gh-96471: Add threading queue shutdown

Doc/library/queue.rst

@@ -239,13 +239,15 @@ them down.
 
 .. method:: Queue.shutdown(immediate=False)
 
-   Shut-down the queue, making queue gets and puts raise :exc:`ShutDown`.
+   Shut down the queue, making :meth:`~Queue.get` and :meth:`~Queue.put` raise
+   :exc:`ShutDown`.
 
-   By default, gets will only raise once the queue is empty. Set
-   *immediate* to true to make gets raise immediately instead.
+   By default, :meth:`~Queue.get` on a shut down queue will only raise once the
+   queue is empty. Set *immediate* to true to make :meth:`~Queue.get` raise
+   immediately instead.
 
-   All blocked callers of put() will be unblocked, and also get()
-   and join() if *immediate* is true.
+   All blocked callers of :meth:`~Queue.put` will be unblocked. If *immediate*
+   is true, also unblock callers of :meth:`~Queue.get` and :meth:`~Queue.join`.
 
    .. versionadded:: 3.13
 

Lib/queue.py

@@ -241,7 +241,7 @@ def shutdown(self, immediate=False):
                 n_items = self._qsize()
                 while self._qsize():
                     self._get()
-                self.unfinished_tasks -= n_items
+                self.unfinished_tasks = max(self.unfinished_tasks - n_items, 0)
                 self.not_empty.notify_all()
                 # release all blocked threads in `join()`
                 self.all_tasks_done.notify_all()

Lib/test/test_queue.py

@@ -10,16 +10,6 @@
 from test.support import import_helper
 from test.support import threading_helper
 
-import typing as t
-
-if t.TYPE_CHECKING:
-    import queue
-
-    parent_class = unittest.TestCase
-
-else:
-    parent_class = object
-
 # queue module depends on threading primitives
 threading_helper.requires_working_threading(module=True)
 
@@ -65,10 +55,7 @@ def run(self):
 # is supposed to raise an exception, call do_exceptional_blocking_test()
 # instead.
 
-class BlockingTestMixin(parent_class):
-    if t.TYPE_CHECKING:
-        queue = queue
-        type2test: t.Type[queue.Queue]
+class BlockingTestMixin:
 
     def do_blocking_test(self, block_func, block_args, trigger_func, trigger_args):
         thread = _TriggerThread(trigger_func, trigger_args)
@@ -588,14 +575,13 @@ def _shutdown_put_join(self, immediate):
         for func, params in thrds:
             threads.append(threading.Thread(target=func, args=params))
             threads[-1].start()
-        if not immediate or immediate:  # TODO: dedent (minimising Git diff)
-            self.assertEqual(q.unfinished_tasks, nb)
-            for i in range(nb):
-                t = threading.Thread(target=q.task_done)
-                t.start()
-                threads.append(t)
-        go.set()
+        self.assertEqual(q.unfinished_tasks, nb)
+        for i in range(nb):
+            t = threading.Thread(target=q.task_done)
+            t.start()
+            threads.append(t)
         q.shutdown(immediate)
+        go.set()
         for t in threads:
             t.join()
 
@@ -810,10 +796,7 @@ class CFailingQueueTest(FailingQueueTest, unittest.TestCase):
     queue = c_queue
 
 
-class BaseSimpleQueueTest(parent_class):
-    if t.TYPE_CHECKING:
-        queue = queue
-        type2test: t.Type[queue.Queue]
+class BaseSimpleQueueTest:
 
     def setUp(self):
         self.q = self.type2test()

Misc/NEWS.d/next/Library/2023-05-06-04-57-10.gh-issue-96471.C9wAU7.rst

@@ -1 +1 @@
-Add queue.Queue termination with ``shutdown`` method
+Add :py:class:`queue.Queue` termination with :py:meth:`~queue.Queue.shutdown`.