From 6a5fc24f5e286fed301acd601e631dc11748ca80 Mon Sep 17 00:00:00 2001
From: "Yves.Duprat" <yduprat@gmail.com>
Date: Thu, 11 Sep 2025 14:36:54 +0200
Subject: [PATCH 01/10] Initial commits

---
 Lib/multiprocessing/queues.py     | 224 ++++++++++++---
 Lib/test/_test_multiprocessing.py | 449 +++++++++++++++++++++++++++++-
 2 files changed, 640 insertions(+), 33 deletions(-)

diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py
index 925f043900004e..f59d328bd5f83e 100644
--- a/Lib/multiprocessing/queues.py
+++ b/Lib/multiprocessing/queues.py
@@ -17,8 +17,9 @@
 import types
 import weakref
 import errno
+from contextlib import contextmanager
 
-from queue import Empty, Full
+from queue import Empty, Full, ShutDown
 
 from . import connection
 from . import context
@@ -45,21 +46,37 @@ def __init__(self, maxsize=0, *, ctx):
         else:
             self._wlock = ctx.Lock()
         self._sem = ctx.BoundedSemaphore(maxsize)
+
+        self._lock_shutdown = ctx.Lock()
+        # Cannot use a ctx.Value because 'ctypes' library is
+        # not always available on all Linux platforms.
+        # Use of Semaphores instead of an array from `heap.BufferWrapper'
+        # is here more efficient and explicit.
+        self._sem_flag_shutdown = ctx.Semaphore(0)
+        self._sem_flag_shutdown_immediate = ctx.Semaphore(0)
+        self._sem_pending_getters = ctx.Semaphore(0)
+        self._sem_pending_putters = ctx.Semaphore(0)
+
         # For use by concurrent.futures
         self._ignore_epipe = False
         self._reset()
-
         if sys.platform != 'win32':
             register_after_fork(self, Queue._after_fork)
 
     def __getstate__(self):
         context.assert_spawning(self)
         return (self._ignore_epipe, self._maxsize, self._reader, self._writer,
-                self._rlock, self._wlock, self._sem, self._opid)
+                self._rlock, self._wlock, self._sem, self._opid,
+                self._lock_shutdown,
+                self._sem_flag_shutdown, self._sem_flag_shutdown_immediate,
+                self._sem_pending_getters, self._sem_pending_putters)
 
     def __setstate__(self, state):
         (self._ignore_epipe, self._maxsize, self._reader, self._writer,
-         self._rlock, self._wlock, self._sem, self._opid) = state
+         self._rlock, self._wlock, self._sem, self._opid,
+         self._lock_shutdown,
+         self._sem_flag_shutdown, self._sem_flag_shutdown_immediate,
+         self._sem_pending_getters, self._sem_pending_putters) = state
         self._reset()
 
     def _after_fork(self):
@@ -81,11 +98,47 @@ def _reset(self, after_fork=False):
         self._recv_bytes = self._reader.recv_bytes
         self._poll = self._reader.poll
 
+    def _is_shutdown(self):
+        return not self._sem_flag_shutdown.locked()
+
+    def _set_shutdown(self, immediate=False):
+        self._sem_flag_shutdown.release()
+        if immediate:
+            self._sem_flag_shutdown_immediate.release()
+
+    @contextmanager
+    def _handle_pending_processes(self, sem):
+        # Count pending getter or putter processes in a dedicated
+        # semaphore. These 2 semaphores are only used when queue
+        # shuts down to release one by one all pending processes.
+        sem.release()
+        try:
+            # Wraps potentialy blocking calls:
+            #  sem._sem.acquire() in put method,
+            #  _recv_bytes()/_poll(*args) in get method.
+            yield
+        finally:
+            sem.acquire()
+
+    def _release_pending_putters(self):
+        with self._lock_shutdown:
+            if not self._sem_pending_putters.locked():
+                self._sem.release()
+
     def put(self, obj, block=True, timeout=None):
         if self._closed:
             raise ValueError(f"Queue {self!r} is closed")
-        if not self._sem.acquire(block, timeout):
-            raise Full
+
+        if self._is_shutdown():
+            raise ShutDown
+        try:
+            with self._handle_pending_processes(self._sem_pending_putters):
+                if not self._sem.acquire(block, timeout):
+                    raise Full
+        finally:
+            if self._is_shutdown():
+                self._release_pending_putters()
+                raise ShutDown
 
         with self._notempty:
             if self._thread is None:
@@ -93,31 +146,53 @@ def put(self, obj, block=True, timeout=None):
             self._buffer.append(obj)
             self._notempty.notify()
 
+    def _release_pending_getters(self):
+        with self._lock_shutdown:
+            if not self._sem_pending_getters.locked():
+                self._put_sentinel()
+
     def get(self, block=True, timeout=None):
         if self._closed:
             raise ValueError(f"Queue {self!r} is closed")
-        if block and timeout is None:
-            with self._rlock:
-                res = self._recv_bytes()
-            self._sem.release()
-        else:
-            if block:
-                deadline = time.monotonic() + timeout
-            if not self._rlock.acquire(block, timeout):
-                raise Empty
-            try:
-                if block:
-                    timeout = deadline - time.monotonic()
-                    if not self._poll(timeout):
+
+        if (empty := self.empty()) and self._is_shutdown():
+            raise ShutDown
+        try:
+            with self._handle_pending_processes(self._sem_pending_getters):
+                if block and timeout is None:
+                    with self._rlock:
+                        res = self._recv_bytes()
+                    self._sem.release()
+                else:
+                    if block:
+                        deadline = time.monotonic() + timeout
+                    if not self._rlock.acquire(block, timeout):
                         raise Empty
-                elif not self._poll():
-                    raise Empty
-                res = self._recv_bytes()
-                self._sem.release()
-            finally:
-                self._rlock.release()
-        # unserialize the data after having released the lock
-        return _ForkingPickler.loads(res)
+                    try:
+                        if block:
+                            timeout = deadline - time.monotonic()
+                            if not self._poll(timeout):
+                                raise Empty
+                        elif not self._poll():
+                            raise Empty
+                        res = self._recv_bytes()
+                        self._sem.release()
+                    finally:
+                        self._rlock.release()
+        finally:
+            if self._is_shutdown() and empty:
+                self._release_pending_getters()
+                raise ShutDown
+
+        item = _ForkingPickler.loads(res)
+        if self._is_shutdown() \
+            and isinstance(item, _ShutdownSentinel):
+            # A pending getter process is just unblocked,
+            # Unblock a next one if exists.
+            self._release_pending_getters()
+            raise ShutDown
+
+        return item
 
     def qsize(self):
         # Raises NotImplementedError on Mac OSX because of broken sem_getvalue()
@@ -135,6 +210,57 @@ def get_nowait(self):
     def put_nowait(self, obj):
         return self.put(obj, False)
 
+    def _clear(self):
+        with self._rlock:
+            while self._poll():
+                self._recv_bytes()
+
+    def _put_sentinel(self):
+        # When put a sentinel into an empty queue,
+        # dont forget to call to _sem.acquire in order to
+        # maintain a correct count of acquire/release
+        # calls for BoudedSempaphore.
+        self._sem.acquire()
+
+        with self._notempty:
+            if self._thread is None:
+                self._start_thread()
+            self._buffer.append(_sentinel_shutdown)
+            self._notempty.notify()
+
+    def shutdown(self, immediate=False):
+        if self._closed:
+            raise ValueError(f"Queue {self!r} is closed")
+
+        with self._lock_shutdown:
+            if self._is_shutdown():
+                raise RuntimeError(f"Queue {self!r} already shut down")
+
+            is_pending_getters = not self._sem_pending_getters.locked()
+            is_pending_putters = not self._sem_pending_putters.locked()
+            str_shutdown =  f"shutdown -> immediate:{immediate}"
+            str_shutdown += f"/PGetters:{is_pending_getters}" \
+                            f"/PPutters:{is_pending_putters}" \
+                            f"/Empty:{self.empty()}" \
+                            f"/Full:{self.full()}"
+            debug(str_shutdown)
+            self._set_shutdown(immediate)
+
+            # Shut down is immediatly and there is no pending getter,
+            # we purge the queue (pipe). If there are datas into the buffer
+            # the 'feeder' thread should erase all of them.
+            if immediate and not is_pending_getters:
+                self._clear()
+
+            # Starting release one pending getter process.
+            # Put a first shutdown sentinel data into the pipe.
+            if self.empty() and is_pending_getters:
+                self._put_sentinel()
+
+            # Starting release one pending putter processes.
+            if is_pending_putters:
+                self._sem.release()
+
     def close(self):
         self._closed = True
         close = self._close
@@ -180,7 +306,7 @@ def _start_thread(self):
             args=(self._buffer, self._notempty, self._send_bytes,
                   self._wlock, self._reader.close, self._writer.close,
                   self._ignore_epipe, self._on_queue_feeder_error,
-                  self._sem),
+                  self._sem, self._sem_flag_shutdown_immediate),
             name='QueueFeederThread',
             daemon=True,
         )
@@ -228,7 +354,8 @@ def _finalize_close(buffer, notempty):
 
     @staticmethod
     def _feed(buffer, notempty, send_bytes, writelock, reader_close,
-              writer_close, ignore_epipe, onerror, queue_sem):
+              writer_close, ignore_epipe, onerror, queue_sem,
+              flag_shutdown_immediate):
         debug('starting thread to feed data to pipe')
         nacquire = notempty.acquire
         nrelease = notempty.release
@@ -240,7 +367,7 @@ def _feed(buffer, notempty, send_bytes, writelock, reader_close,
             wrelease = writelock.release
         else:
             wacquire = None
-
+        is_shutdown_immediate = lambda: not flag_shutdown_immediate.locked()
         while 1:
             try:
                 nacquire()
@@ -258,6 +385,14 @@ def _feed(buffer, notempty, send_bytes, writelock, reader_close,
                             writer_close()
                             return
 
+                        # When queue shuts down immediatly, do not insert
+                        # regular data in pipe, only shutdown sentinel.
+                        if is_shutdown_immediate() \
+                            and not isinstance(obj, _ShutdownSentinel):
+                            debug("Queue shuts down immediatly, " \
+                                  "don't feed regular data to pipe")
+                            continue
+
                         # serialize the data before acquiring the lock
                         obj = _ForkingPickler.dumps(obj)
                         if wacquire is None:
@@ -301,6 +436,12 @@ def _on_queue_feeder_error(e, obj):
     __class_getitem__ = classmethod(types.GenericAlias)
 
 
+# Sentinel item used to release pending getter processes
+# when queue shuts down.
+class _ShutdownSentinel: pass
+_sentinel_shutdown = _ShutdownSentinel()
+
+
 _sentinel = object()
 
 #
@@ -328,8 +469,16 @@ def __setstate__(self, state):
     def put(self, obj, block=True, timeout=None):
         if self._closed:
             raise ValueError(f"Queue {self!r} is closed")
-        if not self._sem.acquire(block, timeout):
-            raise Full
+        if self._is_shutdown():
+            raise ShutDown
+        try:
+            with self._handle_pending_processes(self._sem_pending_putters):
+                if not self._sem.acquire(block, timeout):
+                    raise Full
+        finally:
+            if self._is_shutdown():
+                self._release_pending_putters()
+                raise ShutDown
 
         with self._notempty, self._cond:
             if self._thread is None:
@@ -350,6 +499,17 @@ def join(self):
             if not self._unfinished_tasks._semlock._is_zero():
                 self._cond.wait()
 
+    def _clear(self):
+        super()._clear()
+
+        # Data could be in the buffer, not in the pipe.
+        # Call acquire method of '_unfinished_tasks' Semaphore
+        # until counter is zero.
+        with self._cond:
+            while not self._unfinished_tasks.locked():
+                self._unfinished_tasks.acquire(block=False)
+            self._cond.notify_all()
+
 #
 # Simplified Queue type -- really just a locked pipe
 #
diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
index d9e572961152b3..b19306a526e74f 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -453,7 +453,7 @@ def _test_report_parent_status(cls, wconn):
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_process(self):
         q = self.Queue(1)
-        e = self.Event()
+        #e = self.Event()
         args = (q, 1, 2)
         kwargs = {'hello':23, 'bye':2.54}
         name = 'SomeProcess'
@@ -1474,6 +1474,453 @@ def test_closed_queue_put_get_exceptions(self):
                 q.put('foo')
             with self.assertRaisesRegex(ValueError, 'is closed'):
                 q.get()
+
+    #
+    # Issue gh-96471: multiprocessing queue shutdown.
+    #
+
+    @classmethod
+    def _wait(cls):
+        # See _wait function.
+        _wait()
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_twice(self):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        q = self.Queue()
+        q.shutdown(immediate=False)
+        self.assertTrue(q._is_shutdown())
+
+        with self.assertRaises(RuntimeError):
+            q.shutdown(immediate=False)
+
+        close_queue(q)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_empty(self):
+        if self.TYPE != 'processes':
+            self.skipTest('test not appropriate for {}'.format(self.TYPE))
+
+        q = self.Queue()
+        self.assertTrue(q.empty())
+
+        q.shutdown(immediate=False)
+        self.assertTrue(q._is_shutdown())
+
+        with self.assertRaises(pyqueue.ShutDown):
+            q.put("data")
+        with self.assertRaises(pyqueue.ShutDown):
+            q.get()
+
+        close_queue(q)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_notempty(self):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        q = self.Queue()
+        q.put("Y")
+        q.put("D")
+        self._wait()
+
+        self.assertFalse(q.empty())
+        q.shutdown(immediate=False)
+        self.assertTrue(q._is_shutdown())
+
+        self.assertEqual(q.get(), "Y")
+        self.assertEqual(q.get(), "D")
+        with self.assertRaises(pyqueue.ShutDown):
+            q.get()
+
+        close_queue(q)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_full(self):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        q = self.Queue(maxsize=2)
+        q.put("YY")
+        q.put("DD")
+        self._wait()
+
+        self.assertTrue(q.full())
+        q.shutdown(immediate=False)
+        self.assertTrue(q._is_shutdown())
+
+        self.assertEqual(q.get(), "YY")
+        self.assertEqual(q.get(), "DD")
+        with self.assertRaises(pyqueue.ShutDown):
+            q.put("data")
+        with self.assertRaises(pyqueue.ShutDown):
+            q.get()
+        self.assertFalse(q.full())
+
+        close_queue(q)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_immediate_notempty(self):
+        # when queue shuts down immediatly, all datas inserted
+        # into the pipe or into the buffer are erased.
+        # The queue must be empty.
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        q = self.Queue(maxsize=3)
+        q.put("data")
+        self._wait()
+
+        self.assertFalse(q.full())
+        q.shutdown(immediate=True)
+        self.assertTrue(q.empty())
+        self.assertTrue(q._is_shutdown())
+
+        with self.assertRaises(pyqueue.ShutDown):
+            q.get()
+
+        close_queue(q)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_immediate_full(self):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        q = self.Queue(maxsize=2)
+        q.put("YD")
+        q.put("LO")
+        self._wait()
+
+        self.assertTrue(q.full())
+        q.shutdown(immediate=True)
+        self.assertTrue(q.empty())
+        self.assertTrue(q._is_shutdown())
+
+        with self.assertRaises(pyqueue.ShutDown):
+            q.get()
+
+        close_queue(q)
+
+    def _joinablequeue_shutdown_all_methods(self, immediate):
+        size = 2
+        q = self.JoinableQueue(maxsize=size)
+        q.put("L")
+        q.put_nowait("O")
+        self._wait()
+
+        q.shutdown(immediate)
+        if immediate:
+            self.assertTrue(q.empty())
+        self.assertTrue(q._is_shutdown())
+
+        with self.assertRaises(pyqueue.ShutDown):
+            q.put("E")
+        with self.assertRaises(pyqueue.ShutDown):
+            q.put_nowait("W")
+        if immediate:
+            # All items into the queue are removed.
+            with self.assertRaises(pyqueue.ShutDown):
+                q.get()
+            with self.assertRaises(pyqueue.ShutDown):
+                q.get_nowait()
+            with self.assertRaises(pyqueue.ShutDown):
+                q.get(True, 0.125)
+        else:
+            self.assertEqual(q.get(), "L")
+            q.task_done()
+            self.assertEqual(q.get_nowait(), "O")
+            q.task_done()
+
+            # when `shutdown` queue is empty,
+            # should raise ShutDown Exception
+            with self.assertRaises(pyqueue.ShutDown):
+                q.get() # p.get(True)
+            with self.assertRaises(pyqueue.ShutDown):
+                q.get_nowait() # p.get(False)
+            with self.assertRaises(pyqueue.ShutDown):
+                q.get(True, 0.125)
+        q.join()
+
+        close_queue(q)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_all_methods(self):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        self._joinablequeue_shutdown_all_methods(immediate=False)
+        self._joinablequeue_shutdown_all_methods(immediate=True)
+
+    def _start_processes_pending(self, n, target, args):
+        self.ps = []
+        for i in range(n):
+            self.ps.append(self.Process(target=target,
+                                        args=args))
+        for p in self.ps:
+            p.start()
+
+    @classmethod
+    def _pending_put(cls, q, barrier, results):
+        val = os.getpid()
+        try:
+            barrier.wait()
+            q.put(val)
+            results.append(val)
+        except pyqueue.ShutDown:
+            results.append(pyqueue.ShutDown)
+
+    @unittest.skipIf(sys.platform == 'darwin',
+                     "'get_value' is not implemented on MacOSX")
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_count_pending_put(self):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        # Queue must have a size
+        size = 2
+        q = self.Queue(maxsize=size)
+        results = multiprocessing.Manager().list()
+        n = 30
+        b = self.Barrier(n+1)
+
+        self._start_processes_pending(n, target=self._pending_put,
+                                      args=(q, b, results))
+        # Wait for all procesess start.
+        b.wait()
+
+        # to be sure that queue is full, and all 'n-size' others processes
+        # are pending.
+        self._wait()
+
+        self.assertEqual(q._sem_pending_putters.get_value(), n-size)
+        self.assertEqual(q._sem_pending_getters.get_value(), 0)
+        q.shutdown(immediate=False)
+        self.assertTrue(q._is_shutdown())
+        for p in self.ps:
+            p.join()
+
+        self.assertEqual(q._sem_pending_putters.get_value(), 0)
+        self.assertEqual(q._sem_pending_getters.get_value(), 0)
+        self.assertEqual(len(results), n)
+
+        close_queue(q)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_immediate_pending_put(self):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        # Regardless of the value of the immediate variable,
+        # the tests are identical.
+        size = 5
+        q = self.Queue(maxsize=size)
+        results = multiprocessing.Manager().list()
+        n = 30
+        b = self.Barrier(n+1)
+        self._start_processes_pending(n, target=self._pending_put,
+                                      args=(q, b, results))
+        # Wait for all procesess start.
+        b.wait()
+
+        # We need to call _wait to be sure that queue is full,
+        # and all others processes are pending.
+        self._wait()
+
+        q.shutdown(immediate=True)
+        self.assertTrue(q.empty())
+        self.assertTrue(q._is_shutdown())
+        for p in self.ps:
+            p.join()
+
+        self.assertEqual(len(results), n)
+        self.assertEqual(results.count(pyqueue.ShutDown), n-size)
+
+        self.assertTrue(q.empty())
+        close_queue(q)
+
+    @classmethod
+    def _pending_get(cls, q, barrier, results):
+        try:
+            barrier.wait()
+            results.append(q.get())
+        except pyqueue.ShutDown:
+            results.append(pyqueue.ShutDown)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    @unittest.skipIf(sys.platform == 'darwin',
+                     "'get_value' is not implemented on MacOSX")
+    def test_queue_shutdown_count_pending_get(self):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        q = self.Queue()
+        results = multiprocessing.Manager().list()
+        n = 25
+        b = self.Barrier(n+1)
+        self._start_processes_pending(n, target=self._pending_get,
+                                      args=(q, b, results))
+        # Wait for all procesess start.
+        b.wait()
+
+        # wait for all pending get processes to be blocked.
+        self._wait()
+
+        self.assertTrue(q.empty())
+        self.assertEqual(q._sem_pending_getters.get_value(), n)
+        self.assertEqual(q._sem_pending_putters.get_value(), 0)
+        q.shutdown(immediate=False)
+        self.assertTrue(q._is_shutdown())
+        for p in self.ps:
+            p.join()
+
+        self.assertEqual(q._sem_pending_putters.get_value(), 0)
+        self.assertEqual(q._sem_pending_putters.get_value(), 0)
+        self.assertEqual(len(results), n)
+
+        close_queue(q)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_immediate_pending_get(self):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        q = self.Queue()
+        results = multiprocessing.Manager().list()
+        n = 25
+        b = self.Barrier(n+1)
+        self._start_processes_pending(n, target=self._pending_get,
+                                      args=(q, b, results))
+        # Wait for all procesess start.
+        b.wait()
+
+        # wait for all pending get processes to be blocked.
+        self._wait()
+
+        q.shutdown(immediate=True)
+        self.assertTrue(q.empty())
+        self._wait() # adding data as 'sentinel shutdown'.
+        self.assertTrue(q._is_shutdown())
+        for p in self.ps:
+            p.join()
+
+        self.assertEqual(len(results), n)
+        self.assertEqual(results.count(pyqueue.ShutDown), n)
+
+        self.assertTrue(q.empty())
+        close_queue(q)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_pending_get(self):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        q = self.Queue()
+        q.put("AZ67")
+        self._wait()
+
+        results = multiprocessing.Manager().list()
+        n = 25
+        b = self.Barrier(n+1)
+        self._start_processes_pending(n, target=self._pending_get,
+                                      args=(q, b, results))
+        # Wait for all procesess start.
+        b.wait()
+
+        # wait for all pending get processes to be blocked.
+        self._wait()
+
+        q.shutdown(immediate=False)
+        self.assertTrue(q._is_shutdown())
+        for p in self.ps:
+            p.join()
+
+        self.assertEqual(len(results), n)
+        self.assertEqual(results.count("AZ67"), 1)
+
+        close_queue(q)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def queue_shutdown_immediate_check_purge_buffer(self):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        logger = multiprocessing.get_logger()
+        original_level = logger.level
+        logger.setLevel(logging.DEBUG)
+        stream = io.StringIO()
+        handler = logging.StreamHandler(stream)
+        logging_format = '[%(levelname)s] [%(filename)s] %(message)s'
+        handler.setFormatter(logging.Formatter(logging_format))
+        logger.addHandler(handler)
+
+        try:
+            q = self.JoinableQueue()
+            for i in range(1500):
+                q.put("tictoc")
+            q.shutdown(immediate=True)
+            self.assertTrue(q.empty())
+            q.join()
+
+            log_record = stream.getvalue()
+            self.assertIn("don't feed regular data to pipe", log_record)
+
+            close_queue(q)
+        finally:
+            logger.setLevel(original_level)
+            logger.removeHandler(handler)
+            handler.close()
+
+    @classmethod
+    def _shutdown(cls, q, immediate, results, retcode):
+        q.shutdown(immediate)
+        cls._wait()
+        if not immediate:
+            while not q.empty():
+                results.append(q.get())
+                q.task_done()
+        results.insert(0, retcode)
+
+    def _join_joinablequeue(self, immediate):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        results = multiprocessing.Manager().list()
+        n = 10
+        q = self.JoinableQueue()
+        for i in range(n):
+            q.put(i)
+        self._wait()
+
+        return_process = 1000
+        p = self.Process(target=self._shutdown,
+                         args=(q, immediate, results,
+                              return_process+immediate
+                              )
+                        )
+        p.start()
+        q.join()
+        p.join()
+
+        if immediate:
+            self.assertEqual(len(results), 1)
+            self.assertTrue(q.empty())
+        else:
+            self.assertEqual(len(results), n+1)
+            self.assertListEqual(results[1:], list(range(n)))
+        self.assertEqual(results[0], return_process+immediate)
+
+        close_queue(q)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_immediate_join_joinablequeue(self):
+        return self._join_joinablequeue(True)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_queue_shutdown_join_joinablequeue(self):
+        return self._join_joinablequeue(False)
+
 #
 #
 #

From 7a98c3a0e812d748005f31121725c80e2c22ab33 Mon Sep 17 00:00:00 2001
From: "blurb-it[bot]" <43283697+blurb-it[bot]@users.noreply.github.com>
Date: Fri, 12 Sep 2025 12:02:06 +0000
Subject: [PATCH 02/10] =?UTF-8?q?=F0=9F=93=9C=F0=9F=A4=96=20Added=20by=20b?=
 =?UTF-8?q?lurb=5Fit.?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst   | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst

diff --git a/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst b/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst
new file mode 100644
index 00000000000000..c5c21c40c559b0
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst
@@ -0,0 +1 @@
+Add :py:class:`multiprocessing.Queue` termination with :py:meth:`~multiprocessing.Queue.shutdown` method.

From 297a460bd0cccd59ce73ad6169ae1097482332e4 Mon Sep 17 00:00:00 2001
From: Duprat <yduprat@gmail.com>
Date: Fri, 12 Sep 2025 14:21:55 +0200
Subject: [PATCH 03/10] Fix nits

Docs is not updated, so ref to `shutdown`method is wrong.
---
 .../next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst  | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst b/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst
index c5c21c40c559b0..a0ae0c823241a2 100644
--- a/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst
+++ b/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst
@@ -1 +1 @@
-Add :py:class:`multiprocessing.Queue` termination with :py:meth:`~multiprocessing.Queue.shutdown` method.
+Add :py:class:`multiprocessing.Queue` termination with ``multiprocessing.Queue.shutdown`` method.

From 7c6f69b0303871c4bf5e3a708ba75c275dc9926b Mon Sep 17 00:00:00 2001
From: "Yves.Duprat" <yduprat@gmail.com>
Date: Fri, 19 Jun 2026 19:43:50 +0200
Subject: [PATCH 04/10] Add ``Queue.ShutDown`` documentation. + add
 ``Queue.shutdown` method, + update ``Queue.get``and ``Queue.put`` methods +
 add versionadded to ``get``and ``put``

---
 Doc/library/multiprocessing.rst | 22 ++++++++++++++++++++--
 1 file changed, 20 insertions(+), 2 deletions(-)

diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst
index 2d13053915830b..a2ca4dd9691340 100644
--- a/Doc/library/multiprocessing.rst
+++ b/Doc/library/multiprocessing.rst
@@ -932,8 +932,7 @@ For an example of the usage of queues for interprocess communication see
    standard library's :mod:`queue` module are raised to signal timeouts.
 
    :class:`Queue` implements all the methods of :class:`queue.Queue` except for
-   :meth:`~queue.Queue.task_done`, :meth:`~queue.Queue.join`, and
-   :meth:`~queue.Queue.shutdown`.
+   :meth:`~queue.Queue.task_done`, :meth:`~queue.Queue.join`.
 
    .. method:: qsize()
 
@@ -970,6 +969,10 @@ For an example of the usage of queues for interprocess communication see
          If the queue is closed, :exc:`ValueError` is raised instead of
          :exc:`AssertionError`.
 
+      Can raise the :exc:`queue.ShutDown` when queue shuts down.
+
+      .. versionadded:: 3.16
+
    .. method:: put_nowait(obj)
 
       Equivalent to ``put(obj, False)``.
@@ -988,10 +991,25 @@ For an example of the usage of queues for interprocess communication see
          If the queue is closed, :exc:`ValueError` is raised instead of
          :exc:`OSError`.
 
+      Can raise the :exc:`queue.ShutDown` when queue shuts down and is empty.
+
+      .. versionadded:: 3.16
+
    .. method:: get_nowait()
 
       Equivalent to ``get(False)``.
 
+   .. method:: shutdown([immediate])
+
+      Shuts down the queue. If optional args *immediate* is ``False``
+      (the default), all remaining items could be retrieve until queue is empty.
+      If optional args *immediate* is ``True``, all remaining items from the
+      queue are erased. When queue is empty, get an item raises
+      a :exc:`queue.ShutDown` exception. Adding a new item into the queue
+      also raises this same exception.
+
+   .. versionadded:: 3.16
+
    :class:`multiprocessing.Queue` has a few additional methods not found in
    :class:`queue.Queue`.  These methods are usually unnecessary for most
    code:

From 640798f30d3d948d3588a9208d1c2ce3351491d9 Mon Sep 17 00:00:00 2001
From: Duprat <yduprat@gmail.com>
Date: Sun, 21 Jun 2026 19:44:50 +0200
Subject: [PATCH 05/10] Remove inappropiate assertion

---
 Lib/test/_test_multiprocessing.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
index d83902090095e0..494a3ac19d6352 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -1853,7 +1853,6 @@ def test_queue_shutdown_immediate_pending_get(self):
         self._wait()
 
         q.shutdown(immediate=True)
-        self.assertTrue(q.empty())
         self._wait() # adding data as 'sentinel shutdown'.
         self.assertTrue(q._is_shutdown())
         for p in self.ps:

From 3d47640dc59a8be893ee1f9736a7218cf57529ac Mon Sep 17 00:00:00 2001
From: "Yves.Duprat" <yduprat@gmail.com>
Date: Tue, 23 Jun 2026 14:42:44 +0200
Subject: [PATCH 06/10] Refactor two tests

---
 Lib/test/_test_multiprocessing.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
index 494a3ac19d6352..17b7f89948b6bb 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -1748,7 +1748,8 @@ def test_queue_shutdown_count_pending_put(self):
         # are pending.
         self._wait()
 
-        self.assertEqual(q._sem_pending_putters.get_value(), n-size)
+        self.assertLessEqual(q._sem_pending_putters.get_value(), n-size)
+        self.assertGreater(q._sem_pending_putters.get_value(), 0)
         self.assertEqual(q._sem_pending_getters.get_value(), 0)
         q.shutdown(immediate=False)
         self.assertTrue(q._is_shutdown())
@@ -1821,8 +1822,8 @@ def test_queue_shutdown_count_pending_get(self):
         # wait for all pending get processes to be blocked.
         self._wait()
 
-        self.assertTrue(q.empty())
-        self.assertEqual(q._sem_pending_getters.get_value(), n)
+        self.assertLessEqual(q._sem_pending_getters.get_value(), n)
+        self.assertGreater(q._sem_pending_getters.get_value(), 0)
         self.assertEqual(q._sem_pending_putters.get_value(), 0)
         q.shutdown(immediate=False)
         self.assertTrue(q._is_shutdown())

From e540ce31cd9d65e24d20cb99b5c210bb163e5d7a Mon Sep 17 00:00:00 2001
From: "Gregory P. Smith" <68491+gpshead@users.noreply.github.com>
Date: Wed, 11 Nov 2026 14:46:21 -0800
Subject: [PATCH 07/10] s/3.16/next/

---
 Doc/library/multiprocessing.rst | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst
index a25181abe615bb..882ab3488a849d 100644
--- a/Doc/library/multiprocessing.rst
+++ b/Doc/library/multiprocessing.rst
@@ -981,7 +981,7 @@ For an example of the usage of queues for interprocess communication see
 
       Can raise the :exc:`queue.ShutDown` when queue shuts down.
 
-      .. versionadded:: 3.16
+      .. versionadded:: next
 
    .. method:: put_nowait(obj)
 

From f71d8bc26adc9b5c5a81e28ce516ca1ba9fe301f Mon Sep 17 00:00:00 2001
From: "Gregory P. Smith" <68491+gpshead@users.noreply.github.com>
Date: Wed, 11 Nov 2026 14:46:43 -0800
Subject: [PATCH 08/10] s/3.16/next/

---
 Doc/library/multiprocessing.rst | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst
index 882ab3488a849d..dcd256983dc4ec 100644
--- a/Doc/library/multiprocessing.rst
+++ b/Doc/library/multiprocessing.rst
@@ -1003,7 +1003,7 @@ For an example of the usage of queues for interprocess communication see
 
       Can raise the :exc:`queue.ShutDown` when queue shuts down and is empty.
 
-      .. versionadded:: 3.16
+      .. versionadded:: next
 
    .. method:: get_nowait()
 

From 2297146fe6e30fe07c6477a996cb90988e7aab54 Mon Sep 17 00:00:00 2001
From: "Gregory P. Smith" <68491+gpshead@users.noreply.github.com>
Date: Wed, 11 Nov 2026 14:47:06 -0800
Subject: [PATCH 09/10] retrieved per smontanaro

---
 Doc/library/multiprocessing.rst | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst
index dcd256983dc4ec..f693cbf9d0b60b 100644
--- a/Doc/library/multiprocessing.rst
+++ b/Doc/library/multiprocessing.rst
@@ -1012,7 +1012,7 @@ For an example of the usage of queues for interprocess communication see
    .. method:: shutdown([immediate])
 
       Shuts down the queue. If optional args *immediate* is ``False``
-      (the default), all remaining items could be retrieve until queue is empty.
+      (the default), all remaining items could be retrieved until queue is empty.
       If optional args *immediate* is ``True``, all remaining items from the
       queue are erased. When queue is empty, get an item raises
       a :exc:`queue.ShutDown` exception. Adding a new item into the queue

From a5bf5f30ffc1d20902a56a3403076746bb3c0568 Mon Sep 17 00:00:00 2001
From: "Gregory P. Smith" <greg@krypto.org>
Date: Wed, 11 Nov 2026 17:55:14 -0800
Subject: [PATCH 10/10] gh-96471: Reimplement multiprocessing.Queue.shutdown()

Replace the pending-getter and pending-putter counters with a design
which does not need to know who is waiting:

* A count of the items accepted by put() in any process and not yet
  received tells a getter whether the queue is really empty, including
  the items still in another process's feeder thread buffer.
* shutdown() writes a single empty message to the pipe.  The getter
  which reads it writes it back, so every getter is woken in its turn,
  and raises ShutDown only when nothing is outstanding.
* shutdown() releases the queue's semaphore once if the queue is full.
  A putter which acquires it and finds the queue shut down releases it
  again, waking the next one.
* shutdown(immediate=True) discards the items in the pipe and in every
  feeder thread buffer, accounting for each of them.  The unfinished
  task count of a JoinableQueue only goes down by the number of items
  discarded, as in queue.Queue.

Items put just before shutdown() are no longer lost, getters and
shutdown() itself no longer hang, and calling shutdown() again is
allowed, as in queue.Queue.

The tests are rewritten.  Most of them also run on queue.Queue, through
the 'threads' and 'manager' types, to check the behaviour is the same.

Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>
---
 Doc/library/multiprocessing.rst               |  71 +-
 Doc/whatsnew/3.16.rst                         |   5 +
 Lib/multiprocessing/queues.py                 | 506 ++++++++------
 Lib/test/_test_multiprocessing.py             | 649 ++++++++----------
 ...5-09-12-12-02-02.gh-issue-96471.R6sqGE.rst |   4 +-
 5 files changed, 651 insertions(+), 584 deletions(-)

diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst
index 395d31e66643cb..4f5c0443551ebb 100644
--- a/Doc/library/multiprocessing.rst
+++ b/Doc/library/multiprocessing.rst
@@ -942,7 +942,7 @@ For an example of the usage of queues for interprocess communication see
    standard library's :mod:`queue` module are raised to signal timeouts.
 
    :class:`Queue` implements all the methods of :class:`queue.Queue` except for
-   :meth:`~queue.Queue.task_done`, :meth:`~queue.Queue.join`.
+   :meth:`~queue.Queue.task_done` and :meth:`~queue.Queue.join`.
 
    .. method:: qsize()
 
@@ -979,9 +979,10 @@ For an example of the usage of queues for interprocess communication see
          If the queue is closed, :exc:`ValueError` is raised instead of
          :exc:`AssertionError`.
 
-      Can raise the :exc:`queue.ShutDown` when queue shuts down.
+      Raises :exc:`queue.ShutDown` if the queue has been shut down.
 
-      .. versionadded:: next
+      .. versionchanged:: next
+         :exc:`queue.ShutDown` is raised if the queue has been shut down.
 
    .. method:: put_nowait(obj)
 
@@ -1001,24 +1002,64 @@ For an example of the usage of queues for interprocess communication see
          If the queue is closed, :exc:`ValueError` is raised instead of
          :exc:`OSError`.
 
-      Can raise the :exc:`queue.ShutDown` when queue shuts down and is empty.
+      Raises :exc:`queue.ShutDown` if the queue has been shut down and is
+      empty, or if the queue has been shut down immediately.
 
-      .. versionadded:: next
+      .. versionchanged:: next
+         :exc:`queue.ShutDown` is raised if the queue has been shut down.
 
    .. method:: get_nowait()
 
       Equivalent to ``get(False)``.
 
-   .. method:: shutdown([immediate])
+   .. method:: shutdown(immediate=False)
+
+      Put the queue into a shutdown mode, in every process which uses it.
+
+      The queue can no longer grow.
+      Future calls to :meth:`put` raise :exc:`queue.ShutDown`.
+      Currently blocked callers of :meth:`put` will be unblocked
+      and will raise :exc:`queue.ShutDown` in the formerly blocked process.
+
+      If *immediate* is false (the default), the queue can be wound
+      down normally with :meth:`get` calls to extract the items which have
+      already been put, including those which a process has put but which
+      its feeder thread has not yet written to the pipe.
+
+      Once the queue is empty, future calls to :meth:`get` will raise
+      :exc:`queue.ShutDown`.  Currently blocked callers of :meth:`get` are
+      unblocked and raise :exc:`queue.ShutDown` as well.
+
+      If *immediate* is true, the items remaining in the queue are discarded
+      instead, including those which other processes have put but not yet
+      written to the pipe.  All callers of :meth:`get` are unblocked.
+
+      Calling :meth:`shutdown` more than once is allowed.  A queue which has
+      been shut down cannot be put back into service.
 
-      Shuts down the queue. If optional args *immediate* is ``False``
-      (the default), all remaining items could be retrieved until queue is empty.
-      If optional args *immediate* is ``True``, all remaining items from the
-      queue are erased. When queue is empty, get an item raises
-      a :exc:`queue.ShutDown` exception. Adding a new item into the queue
-      also raises this same exception.
+      Raises :exc:`ValueError` if the queue has been closed in this process
+      with :meth:`close`.
 
-   .. versionadded:: 3.16
+      .. note::
+
+         A getter is woken up by the getter woken before it, so the
+         :exc:`queue.ShutDown` exception can only reach all of the blocked
+         processes as long as none of them is killed, or exits without
+         flushing its feeder thread (see :meth:`cancel_join_thread`), as it
+         is being woken.
+
+         Similarly, if a process is killed or exits without flushing its
+         feeder thread after putting items on the queue, the queue may never
+         be considered empty and :meth:`get` may block instead of raising
+         :exc:`queue.ShutDown`.
+
+      .. note::
+
+         On Windows, an item which another process is writing to the pipe
+         while ``shutdown(immediate=True)`` is running may still be returned
+         by a later :meth:`get`.
+
+      .. versionadded:: next
 
    :class:`multiprocessing.Queue` has a few additional methods not found in
    :class:`queue.Queue`.  These methods are usually unnecessary for most
@@ -1122,6 +1163,10 @@ 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.
 
+      .. versionchanged:: next
+         Items discarded by ``shutdown(immediate=True)`` count as processed:
+         :meth:`task_done` must not be called for them.
+
 
    .. method:: join()
 
diff --git a/Doc/whatsnew/3.16.rst b/Doc/whatsnew/3.16.rst
index da81d8513da99d..a3dcfea29690d1 100644
--- a/Doc/whatsnew/3.16.rst
+++ b/Doc/whatsnew/3.16.rst
@@ -638,6 +638,11 @@ multiprocessing
   ``N=buffersize-processes`` arguments).
   (Contributed by Oleksandr Baltian in :gh:`136871`.)
 
+* Add :meth:`multiprocessing.Queue.shutdown` to terminate a
+  :class:`multiprocessing.Queue` or a :class:`multiprocessing.JoinableQueue`
+  in every process which uses it, like :meth:`queue.Queue.shutdown`.
+  (Contributed by Yves Duprat and Gregory P. Smith in :gh:`96471`.)
+
 
 os
 --
diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py
index 86f983f0ddd835..0d59dfe32c3993 100644
--- a/Lib/multiprocessing/queues.py
+++ b/Lib/multiprocessing/queues.py
@@ -17,7 +17,6 @@
 import types
 import weakref
 import errno
-from contextlib import contextmanager
 
 from queue import Empty, Full, ShutDown
 
@@ -47,19 +46,27 @@ def __init__(self, maxsize=0, *, ctx):
             self._wlock = ctx.Lock()
         self._sem = ctx.BoundedSemaphore(maxsize)
 
-        self._lock_shutdown = ctx.Lock()
-        # Cannot use a ctx.Value because 'ctypes' library is
-        # not always available on all Linux platforms.
-        # Use of Semaphores instead of an array from `heap.BufferWrapper'
-        # is here more efficient and explicit.
-        self._sem_flag_shutdown = ctx.Semaphore(0)
-        self._sem_flag_shutdown_immediate = ctx.Semaphore(0)
-        self._sem_pending_getters = ctx.Semaphore(0)
-        self._sem_pending_putters = ctx.Semaphore(0)
+        # State used by shutdown().  It is all kept in semaphores, which are
+        # only ever tested for zero, because neither ctypes nor a working
+        # sem_getvalue() is available on every platform.
+        #
+        # Where sem_getvalue() is broken (macOS), the zero test acquires
+        # the semaphore and releases it again, so a second test made at the
+        # same moment can wrongly find it zero.  Every test whose answer
+        # matters is therefore made while holding _state_lock.
+        self._state_lock = ctx.Lock()
+        # Non-zero once shutdown() has been called.
+        self._shutdown_flag = ctx.Semaphore(0)
+        # Non-zero once shutdown(immediate=True) has been called.
+        self._immediate_flag = ctx.Semaphore(0)
+        # Number of items accepted by put() in any process and not yet
+        # received: those in the pipe and in every process's buffer.
+        self._outstanding = ctx.Semaphore(0)
 
         # For use by concurrent.futures
         self._ignore_epipe = False
         self._reset()
+
         if sys.platform != 'win32':
             register_after_fork(self, Queue._after_fork)
 
@@ -67,16 +74,14 @@ def __getstate__(self):
         context.assert_spawning(self)
         return (self._ignore_epipe, self._maxsize, self._reader, self._writer,
                 self._rlock, self._wlock, self._sem, self._opid,
-                self._lock_shutdown,
-                self._sem_flag_shutdown, self._sem_flag_shutdown_immediate,
-                self._sem_pending_getters, self._sem_pending_putters)
+                self._state_lock, self._shutdown_flag, self._immediate_flag,
+                self._outstanding)
 
     def __setstate__(self, state):
         (self._ignore_epipe, self._maxsize, self._reader, self._writer,
          self._rlock, self._wlock, self._sem, self._opid,
-         self._lock_shutdown,
-         self._sem_flag_shutdown, self._sem_flag_shutdown_immediate,
-         self._sem_pending_getters, self._sem_pending_putters) = state
+         self._state_lock, self._shutdown_flag, self._immediate_flag,
+         self._outstanding) = state
         self._reset()
 
     def _after_fork(self):
@@ -94,51 +99,52 @@ def _reset(self, after_fork=False):
         self._joincancelled = False
         self._closed = False
         self._close = None
+        # The shutdown flag is never cleared, so it can be cached once seen.
+        self._shutdown_seen = False
         self._send_bytes = self._writer.send_bytes
         self._recv_bytes = self._reader.recv_bytes
         self._poll = self._reader.poll
 
     def _is_shutdown(self):
-        return not self._sem_flag_shutdown.locked()
+        if not self._shutdown_seen:
+            with self._state_lock:
+                self._shutdown_seen = (
+                    not self._shutdown_flag._semlock._is_zero())
+        return self._shutdown_seen
 
-    def _set_shutdown(self, immediate=False):
-        self._sem_flag_shutdown.release()
-        if immediate:
-            self._sem_flag_shutdown_immediate.release()
-
-    @contextmanager
-    def _handle_pending_processes(self, sem):
-        # Count pending getter or putter processes in a dedicated
-        # semaphore. These 2 semaphores are only used when queue
-        # shuts down to release one by one all pending processes.
-        sem.release()
+    def _release_slot(self):
         try:
-            # Wraps potentialy blocking calls:
-            #  sem._sem.acquire() in put method,
-            #  _recv_bytes()/_poll(*args) in get method.
-            yield
-        finally:
-            sem.acquire()
-
-    def _release_pending_putters(self):
-        with self._lock_shutdown:
-            if not self._sem_pending_putters.locked():
-                self._sem.release()
+            self._sem.release()
+        except ValueError:
+            # shutdown() may have released _sem once more than there were
+            # items, to wake the blocked putters.
+            if not self._is_shutdown():
+                raise
+
+    def _acquire_slot(self, block, timeout):
+        # Reserve room for one item and count it as outstanding, or raise
+        # Full or ShutDown.
+        if self._shutdown_seen:
+            raise ShutDown
+        if not self._sem.acquire(block, timeout):
+            raise ShutDown if self._is_shutdown() else Full
+        if not self._shutdown_seen:
+            # Testing the flag and counting the item must be atomic with
+            # respect to shutdown(): once a getter has seen the queue shut
+            # down with nothing outstanding, no put() may succeed.
+            with self._state_lock:
+                if self._shutdown_flag._semlock._is_zero():
+                    self._outstanding.release()
+                    return
+                self._shutdown_seen = True
+        # Hand the unit on, so that the next blocked putter wakes up too.
+        self._release_slot()
+        raise ShutDown
 
     def put(self, obj, block=True, timeout=None):
         if self._closed:
             raise ValueError(f"Queue {self!r} is closed")
-
-        if self._is_shutdown():
-            raise ShutDown
-        try:
-            with self._handle_pending_processes(self._sem_pending_putters):
-                if not self._sem.acquire(block, timeout):
-                    raise Full
-        finally:
-            if self._is_shutdown():
-                self._release_pending_putters()
-                raise ShutDown
+        self._acquire_slot(block, timeout)
 
         with self._notempty:
             if self._thread is None:
@@ -146,59 +152,201 @@ def put(self, obj, block=True, timeout=None):
             self._buffer.append(obj)
             self._notempty.notify()
 
-    def _release_pending_getters(self):
-        with self._lock_shutdown:
-            if not self._sem_pending_getters.locked():
-                self._put_sentinel()
+    def _received(self):
+        # Called with _rlock held, for each item read from the pipe.
+        if not self._outstanding.acquire(False):
+            # Either a zero test was holding the semaphore (see __init__),
+            # or the item was put by a subclass which overrides put()
+            # without calling _acquire_slot() and was never counted.
+            with self._state_lock:
+                self._outstanding.acquire(False)
+
+    def _after_marker(self, block, deadline):
+        # Called with _rlock held, after reading the shutdown marker instead
+        # of an item.  Return the next item, or raise ShutDown or Empty.
+        #
+        # shutdown() writes a single marker and whoever reads it must write
+        # it back, so that every getter, blocked or yet to come, is woken in
+        # its turn.  Extra markers are harmless: they are only wake-ups, and
+        # a reader never writes back more than one.
+        try:
+            while True:
+                with self._state_lock:
+                    drained = self._outstanding._semlock._is_zero()
+                    immediate = not self._immediate_flag._semlock._is_zero()
+                if drained:
+                    # No item is outstanding.  Still return anything found
+                    # in the pipe, such as an item put by a subclass which
+                    # does not count them, rather than strand it.
+                    if not self._poll():
+                        raise ShutDown
+                elif not block:
+                    if not self._poll():
+                        raise Empty
+                elif deadline is not None:
+                    if not self._poll(max(0.0, deadline - time.monotonic())):
+                        raise Empty
+                # Otherwise an item is still in some process's buffer: wait
+                # for it.  If it gets discarded instead, whoever discards it
+                # writes another marker.
+                res = self._recv_bytes()
+                if not res:
+                    continue
+                if immediate:
+                    self._make_discarder()(1)
+                    continue
+                return res
+        finally:
+            self._put_marker()
 
     def get(self, block=True, timeout=None):
         if self._closed:
             raise ValueError(f"Queue {self!r} is closed")
-
-        if (empty := self.empty()) and self._is_shutdown():
-            raise ShutDown
-        try:
-            with self._handle_pending_processes(self._sem_pending_getters):
-                if block and timeout is None:
-                    with self._rlock:
-                        res = self._recv_bytes()
-                    self._sem.release()
-                else:
+        if block and timeout is None:
+            with self._rlock:
+                res = self._recv_bytes()
+                if not res:
+                    # An empty message is the shutdown marker.
+                    res = self._after_marker(True, None)
+                self._received()
+            self._release_slot()
+        else:
+            if block:
+                deadline = time.monotonic() + timeout
+            try:
+                if not self._rlock.acquire(block, timeout):
+                    raise Empty
+                try:
                     if block:
-                        deadline = time.monotonic() + timeout
-                    if not self._rlock.acquire(block, timeout):
-                        raise Empty
-                    try:
-                        if block:
-                            timeout = deadline - time.monotonic()
-                            if not self._poll(timeout):
-                                raise Empty
-                        elif not self._poll():
+                        timeout = deadline - time.monotonic()
+                        if not self._poll(timeout):
                             raise Empty
-                        res = self._recv_bytes()
-                        self._sem.release()
-                    finally:
-                        self._rlock.release()
-        finally:
-            if self._is_shutdown() and empty:
-                self._release_pending_getters()
-                raise ShutDown
-
-        item = _ForkingPickler.loads(res)
-        if self._is_shutdown() \
-            and isinstance(item, _ShutdownSentinel):
-            # A pending getter process is just unblocked,
-            # Unblock a next one if exists.
-            self._release_pending_getters()
-            raise ShutDown
+                    elif not self._poll():
+                        raise Empty
+                    res = self._recv_bytes()
+                    if not res:
+                        res = self._after_marker(block,
+                                                 deadline if block else None)
+                    self._received()
+                    self._release_slot()
+                finally:
+                    self._rlock.release()
+            except Empty:
+                # The pipe of a shut down queue is briefly empty while a
+                # getter writes the marker back.
+                if self._is_shutdown() and self._is_drained():
+                    raise ShutDown from None
+                raise
+        # unserialize the data after having released the lock
+        return _ForkingPickler.loads(res)
+
+    def _is_drained(self):
+        with self._state_lock:
+            return self._outstanding._semlock._is_zero()
+
+    def _put_marker(self):
+        # The marker goes through the feeder thread so that it follows
+        # everything this process has already put, and so that writing it
+        # can never block the caller.
+        with self._notempty:
+            if self._thread is None:
+                self._start_thread()
+            self._buffer.append(_shutdown_marker)
+            self._notempty.notify()
+
+    def _make_discarder(self):
+        # Return a function accounting for N items thrown away by
+        # shutdown(immediate=True).  It must not refer to self: the feeder
+        # thread keeps it, and is only told to exit when the queue is
+        # garbage collected.
+        sem, outstanding, state_lock = (
+            self._sem, self._outstanding, self._state_lock)
+        def discard(n):
+            for _ in range(n):
+                if not outstanding.acquire(False):
+                    with state_lock:
+                        outstanding.acquire(False)
+                try:
+                    sem.release()
+                except ValueError:
+                    pass
+        return discard
 
-        return item
+    def shutdown(self, immediate=False):
+        if self._closed:
+            raise ValueError(f"Queue {self!r} is closed")
+        with self._state_lock:
+            first = self._shutdown_flag._semlock._is_zero()
+            if first:
+                self._shutdown_flag.release()
+            if immediate and self._immediate_flag._semlock._is_zero():
+                self._immediate_flag.release()
+        self._shutdown_seen = True
+        if first:
+            # Wake a putter blocked on a full queue; it wakes the next one.
+            # Putters which acquire a unit from now on give it back, so
+            # there is nothing to do if the queue is not full.
+            if self._sem._semlock._is_zero():
+                self._release_slot()
+            # Wake a blocked getter; it wakes the next one.
+            self._put_marker()
+        if immediate:
+            self._discard_all()
+
+    def _discard_all(self):
+        discard = self._make_discarder()
+        # Items this process has buffered.  Feeder threads, here and in
+        # other processes, discard what they hold on seeing the flag.
+        with self._notempty:
+            keep = [obj for obj in self._buffer
+                    if obj is _sentinel or obj is _shutdown_marker]
+            n = len(self._buffer) - len(keep)
+            self._buffer.clear()
+            self._buffer.extend(keep)
+        if n:
+            discard(n)
+            # A getter may be waiting for one of them.
+            self._put_marker()
+        # Items already in the pipe.
+        wlock = self._wlock
+        n = 0
+        with self._rlock:
+            while True:
+                while self._poll():
+                    if self._recv_bytes():
+                        n += 1
+                if wlock is None or wlock.acquire(timeout=0.01):
+                    break
+                # A feeder thread is in the middle of writing, and may be
+                # waiting for room in the pipe: keep reading.
+            if wlock is None:
+                # Windows: there is no write lock to stop other processes
+                # from writing, so an item may still get in ahead of the
+                # marker and be returned by get().
+                self._put_marker()
+            else:
+                # No one else can write now.  Empty the pipe and leave the
+                # marker in it, ahead of any item a feeder thread had
+                # already decided to send: the getter which reads the marker
+                # throws those away before letting another getter in.
+                try:
+                    while self._poll():
+                        if self._recv_bytes():
+                            n += 1
+                    self._send_bytes(b'')
+                finally:
+                    wlock.release()
+            discard(n)
 
     def qsize(self):
         # Raises NotImplementedError on Mac OSX because of broken sem_getvalue()
-        return self._maxsize - self._sem.get_value()
+        return max(0, self._maxsize - self._sem.get_value())
 
     def empty(self):
+        if self._is_shutdown():
+            # The pipe now holds the shutdown marker, so whether it can be
+            # read says nothing about items.
+            return self._is_drained()
         return not self._poll()
 
     def full(self):
@@ -210,57 +358,6 @@ def get_nowait(self):
     def put_nowait(self, obj):
         return self.put(obj, False)
 
-    def _clear(self):
-        with self._rlock:
-            while self._poll():
-                self._recv_bytes()
-
-    def _put_sentinel(self):
-        # When put a sentinel into an empty queue,
-        # dont forget to call to _sem.acquire in order to
-        # maintain a correct count of acquire/release
-        # calls for BoudedSempaphore.
-        self._sem.acquire()
-
-        with self._notempty:
-            if self._thread is None:
-                self._start_thread()
-            self._buffer.append(_sentinel_shutdown)
-            self._notempty.notify()
-
-    def shutdown(self, immediate=False):
-        if self._closed:
-            raise ValueError(f"Queue {self!r} is closed")
-
-        with self._lock_shutdown:
-            if self._is_shutdown():
-                raise RuntimeError(f"Queue {self!r} already shut down")
-
-            is_pending_getters = not self._sem_pending_getters.locked()
-            is_pending_putters = not self._sem_pending_putters.locked()
-            str_shutdown =  f"shutdown -> immediate:{immediate}"
-            str_shutdown += f"/PGetters:{is_pending_getters}" \
-                            f"/PPutters:{is_pending_putters}" \
-                            f"/Empty:{self.empty()}" \
-                            f"/Full:{self.full()}"
-            debug(str_shutdown)
-            self._set_shutdown(immediate)
-
-            # Shut down is immediatly and there is no pending getter,
-            # we purge the queue (pipe). If there are datas into the buffer
-            # the 'feeder' thread should erase all of them.
-            if immediate and not is_pending_getters:
-                self._clear()
-
-            # Starting release one pending getter process.
-            # Put a first shutdown sentinel data into the pipe.
-            if self.empty() and is_pending_getters:
-                self._put_sentinel()
-
-            # Starting release one pending putter processes.
-            if is_pending_putters:
-                self._sem.release()
-
     def close(self):
         self._closed = True
         close = self._close
@@ -306,7 +403,9 @@ def _start_thread(self):
             args=(self._buffer, self._notempty, self._send_bytes,
                   self._wlock, self._reader.close, self._writer.close,
                   self._ignore_epipe, self._on_queue_feeder_error,
-                  self._sem, self._sem_flag_shutdown_immediate),
+                  self._sem, self._make_discarder(),
+                  (self._state_lock, self._shutdown_flag,
+                   self._immediate_flag, self._outstanding)),
             name='QueueFeederThread',
             daemon=True,
         )
@@ -355,20 +454,34 @@ def _finalize_close(buffer, notempty):
     @staticmethod
     def _feed(buffer, notempty, send_bytes, writelock, reader_close,
               writer_close, ignore_epipe, onerror, queue_sem,
-              flag_shutdown_immediate):
+              discard, shutdown_state):
         debug('starting thread to feed data to pipe')
         nacquire = notempty.acquire
         nrelease = notempty.release
         nwait = notempty.wait
         bpopleft = buffer.popleft
         sentinel = _sentinel
+        marker = _shutdown_marker
+        state_lock, shutdown_flag, immediate_flag, outstanding = shutdown_state
+        not_immediate = immediate_flag._semlock._is_zero
         if sys.platform != 'win32':
             wacquire = writelock.acquire
             wrelease = writelock.release
         else:
             wacquire = None
-        is_shutdown_immediate = lambda: not flag_shutdown_immediate.locked()
+
+        def send(data):
+            if wacquire is None:
+                send_bytes(data)
+            else:
+                wacquire()
+                try:
+                    send_bytes(data)
+                finally:
+                    wrelease()
+
         while 1:
+            is_item = False
             try:
                 nacquire()
                 try:
@@ -376,35 +489,43 @@ def _feed(buffer, notempty, send_bytes, writelock, reader_close,
                         nwait()
                 finally:
                     nrelease()
+                ndiscarded = 0
                 try:
                     while 1:
+                        is_item = False
                         obj = bpopleft()
-                        if obj is sentinel:
-                            debug('feeder thread got sentinel -- exiting')
-                            reader_close()
-                            writer_close()
-                            return
-
-                        # When queue shuts down immediatly, do not insert
-                        # regular data in pipe, only shutdown sentinel.
-                        if is_shutdown_immediate() \
-                            and not isinstance(obj, _ShutdownSentinel):
-                            debug("Queue shuts down immediatly, " \
-                                  "don't feed regular data to pipe")
+                        if obj is sentinel or obj is marker:
+                            if ndiscarded:
+                                # A getter may be waiting for one of the
+                                # discarded items: wake it.
+                                discard(ndiscarded)
+                                ndiscarded = 0
+                                send(b'')
+                            if obj is sentinel:
+                                debug('feeder thread got sentinel -- exiting')
+                                reader_close()
+                                writer_close()
+                                return
+                            # An empty message is never a valid pickle.
+                            send(b'')
+                            continue
+
+                        # After shutdown(immediate=True), throw away what
+                        # has not been sent yet.  This test is not made
+                        # under state_lock, to keep the feeder cheap: at
+                        # worst an item is sent which should not have been.
+                        if not not_immediate():
+                            ndiscarded += 1
                             continue
 
+                        is_item = True
                         # serialize the data before acquiring the lock
                         obj = _ForkingPickler.dumps(obj)
-                        if wacquire is None:
-                            send_bytes(obj)
-                        else:
-                            wacquire()
-                            try:
-                                send_bytes(obj)
-                            finally:
-                                wrelease()
+                        send(obj)
                 except IndexError:
-                    pass
+                    if ndiscarded:
+                        discard(ndiscarded)
+                        send(b'')
             except Exception as e:
                 if ignore_epipe and getattr(e, 'errno', 0) == errno.EPIPE:
                     return
@@ -421,7 +542,14 @@ def _feed(buffer, notempty, send_bytes, writelock, reader_close,
                     # if the object had been silently removed from the queue
                     # and this step is necessary to have a properly working
                     # queue.
-                    queue_sem.release()
+                    if is_item:
+                        queue_sem.release()
+                        with state_lock:
+                            outstanding.acquire(False)
+                            shutdown = not shutdown_flag._semlock._is_zero()
+                        if shutdown:
+                            # A getter may be waiting for this item.
+                            buffer.appendleft(marker)
                     onerror(e, obj)
 
     @staticmethod
@@ -436,13 +564,10 @@ def _on_queue_feeder_error(e, obj):
     __class_getitem__ = classmethod(types.GenericAlias)
 
 
-# Sentinel item used to release pending getter processes
-# when queue shuts down.
-class _ShutdownSentinel: pass
-_sentinel_shutdown = _ShutdownSentinel()
-
-
 _sentinel = object()
+# Put in a feeder thread's buffer to have the shutdown marker, an empty
+# message, written to the pipe.
+_shutdown_marker = object()
 
 #
 # A queue type which also supports join() and task_done() methods
@@ -469,16 +594,7 @@ def __setstate__(self, state):
     def put(self, obj, block=True, timeout=None):
         if self._closed:
             raise ValueError(f"Queue {self!r} is closed")
-        if self._is_shutdown():
-            raise ShutDown
-        try:
-            with self._handle_pending_processes(self._sem_pending_putters):
-                if not self._sem.acquire(block, timeout):
-                    raise Full
-        finally:
-            if self._is_shutdown():
-                self._release_pending_putters()
-                raise ShutDown
+        self._acquire_slot(block, timeout)
 
         with self._notempty, self._cond:
             if self._thread is None:
@@ -499,16 +615,20 @@ def join(self):
             if not self._unfinished_tasks._semlock._is_zero():
                 self._cond.wait()
 
-    def _clear(self):
-        super()._clear()
-
-        # Data could be in the buffer, not in the pipe.
-        # Call acquire method of '_unfinished_tasks' Semaphore
-        # until counter is zero.
-        with self._cond:
-            while not self._unfinished_tasks.locked():
-                self._unfinished_tasks.acquire(block=False)
-            self._cond.notify_all()
+    def _make_discarder(self):
+        discard_items = super()._make_discarder()
+        cond, unfinished_tasks = self._cond, self._unfinished_tasks
+        def discard(n):
+            discard_items(n)
+            # Only the discarded items are done: a task which a worker has
+            # already taken still needs its task_done().
+            with cond:
+                for _ in range(n):
+                    if not unfinished_tasks.acquire(False):
+                        break
+                if unfinished_tasks._semlock._is_zero():
+                    cond.notify_all()
+        return discard
 
 #
 # Simplified Queue type -- really just a locked pipe
diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
index 947789af141c89..80bfd79edaa23d 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -508,7 +508,7 @@ def _test_report_parent_status(cls, wconn):
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_process(self):
         q = self.Queue(1)
-        #e = self.Event()
+        e = self.Event()
         args = (q, 1, 2)
         kwargs = {'hello':23, 'bye':2.54}
         name = 'SomeProcess'
@@ -1538,450 +1538,345 @@ def test_closed_queue_put_get_exceptions(self):
                 q.get()
 
     #
-    # Issue gh-96471: multiprocessing queue shutdown.
+    # shutdown()
+    #
+    # Except where they look at the internals, these tests also run on
+    # queue.Queue, through the 'threads' and 'manager' types: the
+    # behaviour must be the same.
     #
 
     @classmethod
-    def _wait(cls):
-        # See _wait function.
-        _wait()
-
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_twice(self):
-        if self.TYPE != 'processes':
-            self.skipTest(f'test not appropriate for {self.TYPE}')
-
-        q = self.Queue()
-        q.shutdown(immediate=False)
-        self.assertTrue(q._is_shutdown())
-
-        with self.assertRaises(RuntimeError):
-            q.shutdown(immediate=False)
+    def _shutdown_getter(cls, q, results, ready):
+        ready.set()
+        items = []
+        try:
+            while True:
+                items.append(q.get())
+        except pyqueue.ShutDown:
+            pass
+        results.put(items)
 
-        close_queue(q)
+    @classmethod
+    def _shutdown_putter(cls, q, results, ready):
+        ready.set()
+        try:
+            q.put('extra')
+        except pyqueue.ShutDown:
+            results.put('ShutDown')
+        else:
+            results.put('put')
+
+    def _start_shutdown_workers(self, n, target, q):
+        results = self.Queue()
+        self.addCleanup(close_queue, results)
+        readies = [self.Event() for _ in range(n)]
+        procs = [self.Process(target=target, args=(q, results, ready))
+                 for ready in readies]
+        for p in procs:
+            p.daemon = True
+            p.start()
+            self.addCleanup(join_process, p)
+        for ready in readies:
+            self.assertTrue(ready.wait(support.SHORT_TIMEOUT))
+        # Let the workers block in get() or put().  This is only to make
+        # the interesting case the likely one: the tests are valid
+        # whatever the workers are doing when shutdown() is called.
+        time.sleep(0.1)
+        return [functools.partial(results.get, timeout=support.SHORT_TIMEOUT)
+                for p in procs]
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_shutdown_empty(self):
+        for immediate in False, True:
+            with self.subTest(immediate=immediate):
+                q = self.Queue()
+                q.shutdown(immediate=immediate)
+                with self.assertRaises(pyqueue.ShutDown):
+                    q.put(1)
+                with self.assertRaises(pyqueue.ShutDown):
+                    q.put_nowait(1)
+                with self.assertRaises(pyqueue.ShutDown):
+                    q.put(1, timeout=0.01)
+                with self.assertRaises(pyqueue.ShutDown):
+                    q.get()
+                with self.assertRaises(pyqueue.ShutDown):
+                    q.get_nowait()
+                with self.assertRaises(pyqueue.ShutDown):
+                    q.get(timeout=0.01)
+                self.assertTrue(queue_empty(q))
+                close_queue(q)
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_empty(self):
-        if self.TYPE != 'processes':
-            self.skipTest('test not appropriate for {}'.format(self.TYPE))
-
+    def test_shutdown_drain(self):
         q = self.Queue()
-        self.assertTrue(q.empty())
-
-        q.shutdown(immediate=False)
-        self.assertTrue(q._is_shutdown())
-
+        for i in range(3):
+            q.put(i)
+        # Do not wait: the items put just before shutdown() must not be
+        # lost, even those the feeder thread has not written yet.
+        q.shutdown()
         with self.assertRaises(pyqueue.ShutDown):
-            q.put("data")
+            q.put(3)
+        self.assertEqual(q.get(), 0)
+        self.assertEqual(q.get(timeout=support.SHORT_TIMEOUT), 1)
+        self.assertEqual(q.get(), 2)
         with self.assertRaises(pyqueue.ShutDown):
             q.get()
-
-        close_queue(q)
-
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_notempty(self):
-        if self.TYPE != 'processes':
-            self.skipTest(f'test not appropriate for {self.TYPE}')
-
-        q = self.Queue()
-        q.put("Y")
-        q.put("D")
-        self._wait()
-
-        self.assertFalse(q.empty())
-        q.shutdown(immediate=False)
-        self.assertTrue(q._is_shutdown())
-
-        self.assertEqual(q.get(), "Y")
-        self.assertEqual(q.get(), "D")
         with self.assertRaises(pyqueue.ShutDown):
-            q.get()
-
-        close_queue(q)
-
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_full(self):
-        if self.TYPE != 'processes':
-            self.skipTest(f'test not appropriate for {self.TYPE}')
-
-        q = self.Queue(maxsize=2)
-        q.put("YY")
-        q.put("DD")
-        self._wait()
-
-        self.assertTrue(q.full())
-        q.shutdown(immediate=False)
-        self.assertTrue(q._is_shutdown())
-
-        self.assertEqual(q.get(), "YY")
-        self.assertEqual(q.get(), "DD")
-        with self.assertRaises(pyqueue.ShutDown):
-            q.put("data")
+            q.get_nowait()
         with self.assertRaises(pyqueue.ShutDown):
-            q.get()
-        self.assertFalse(q.full())
-
+            q.get(timeout=0.01)
+        self.assertTrue(queue_empty(q))
         close_queue(q)
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_immediate_notempty(self):
-        # when queue shuts down immediatly, all datas inserted
-        # into the pipe or into the buffer are erased.
-        # The queue must be empty.
-        if self.TYPE != 'processes':
-            self.skipTest(f'test not appropriate for {self.TYPE}')
-
+    def test_shutdown_immediate(self):
         q = self.Queue(maxsize=3)
-        q.put("data")
-        self._wait()
-
-        self.assertFalse(q.full())
+        for i in range(3):
+            q.put(i)
         q.shutdown(immediate=True)
-        self.assertTrue(q.empty())
-        self.assertTrue(q._is_shutdown())
-
         with self.assertRaises(pyqueue.ShutDown):
             q.get()
-
+        with self.assertRaises(pyqueue.ShutDown):
+            q.get_nowait()
+        with self.assertRaises(pyqueue.ShutDown):
+            q.put(3)
+        self.assertTrue(queue_empty(q))
+        self.assertFalse(queue_full(q, 3))
+        self.assertReturnsIfImplemented(0, q.qsize)
         close_queue(q)
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_immediate_full(self):
-        if self.TYPE != 'processes':
-            self.skipTest(f'test not appropriate for {self.TYPE}')
-
-        q = self.Queue(maxsize=2)
-        q.put("YD")
-        q.put("LO")
-        self._wait()
-
-        self.assertTrue(q.full())
+    def test_shutdown_repeated(self):
+        q = self.Queue()
+        q.put(1)
+        q.put(2)
+        q.shutdown()
+        q.shutdown()
+        self.assertEqual(q.get(), 1)
+        # An immediate shutdown can follow a graceful one.
         q.shutdown(immediate=True)
-        self.assertTrue(q.empty())
-        self.assertTrue(q._is_shutdown())
-
         with self.assertRaises(pyqueue.ShutDown):
             q.get()
-
-        close_queue(q)
-
-    def _joinablequeue_shutdown_all_methods(self, immediate):
-        size = 2
-        q = self.JoinableQueue(maxsize=size)
-        q.put("L")
-        q.put_nowait("O")
-        self._wait()
-
-        q.shutdown(immediate)
-        if immediate:
-            self.assertTrue(q.empty())
-        self.assertTrue(q._is_shutdown())
-
-        with self.assertRaises(pyqueue.ShutDown):
-            q.put("E")
-        with self.assertRaises(pyqueue.ShutDown):
-            q.put_nowait("W")
-        if immediate:
-            # All items into the queue are removed.
-            with self.assertRaises(pyqueue.ShutDown):
-                q.get()
-            with self.assertRaises(pyqueue.ShutDown):
-                q.get_nowait()
-            with self.assertRaises(pyqueue.ShutDown):
-                q.get(True, 0.125)
-        else:
-            self.assertEqual(q.get(), "L")
-            q.task_done()
-            self.assertEqual(q.get_nowait(), "O")
-            q.task_done()
-
-            # when `shutdown` queue is empty,
-            # should raise ShutDown Exception
-            with self.assertRaises(pyqueue.ShutDown):
-                q.get() # p.get(True)
-            with self.assertRaises(pyqueue.ShutDown):
-                q.get_nowait() # p.get(False)
-            with self.assertRaises(pyqueue.ShutDown):
-                q.get(True, 0.125)
-        q.join()
-
+        q.shutdown()
+        self.assertTrue(queue_empty(q))
         close_queue(q)
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_all_methods(self):
+    def test_shutdown_closed(self):
         if self.TYPE != 'processes':
             self.skipTest(f'test not appropriate for {self.TYPE}')
+        q = self.Queue()
+        q.close()
+        with self.assertRaisesRegex(ValueError, 'is closed'):
+            q.shutdown()
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_shutdown_blocked_getters(self):
+        for immediate in False, True:
+            with self.subTest(immediate=immediate):
+                q = self.Queue()
+                results = self._start_shutdown_workers(
+                    4, self._shutdown_getter, q)
+                q.shutdown(immediate=immediate)
+                self.assertEqual([get() for get in results], [[]] * 4)
+                close_queue(q)
 
-        self._joinablequeue_shutdown_all_methods(immediate=False)
-        self._joinablequeue_shutdown_all_methods(immediate=True)
-
-    def _start_processes_pending(self, n, target, args):
-        self.ps = []
-        for i in range(n):
-            self.ps.append(self.Process(target=target,
-                                        args=args))
-        for p in self.ps:
-            p.start()
-
-    @classmethod
-    def _pending_put(cls, q, barrier, results):
-        val = os.getpid()
-        try:
-            barrier.wait()
-            q.put(val)
-            results.append(val)
-        except pyqueue.ShutDown:
-            results.append(pyqueue.ShutDown)
-
-    @unittest.skipIf(sys.platform == 'darwin',
-                     "'get_value' is not implemented on MacOSX")
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_count_pending_put(self):
-        if self.TYPE != 'processes':
-            self.skipTest(f'test not appropriate for {self.TYPE}')
-
-        # Queue must have a size
-        size = 2
-        q = self.Queue(maxsize=size)
-        results = multiprocessing.Manager().list()
-        n = 30
-        b = self.Barrier(n+1)
-
-        self._start_processes_pending(n, target=self._pending_put,
-                                      args=(q, b, results))
-        # Wait for all procesess start.
-        b.wait()
-
-        # to be sure that queue is full, and all 'n-size' others processes
-        # are pending.
-        self._wait()
-
-        self.assertLessEqual(q._sem_pending_putters.get_value(), n-size)
-        self.assertGreater(q._sem_pending_putters.get_value(), 0)
-        self.assertEqual(q._sem_pending_getters.get_value(), 0)
-        q.shutdown(immediate=False)
-        self.assertTrue(q._is_shutdown())
-        for p in self.ps:
-            p.join()
-
-        self.assertEqual(q._sem_pending_putters.get_value(), 0)
-        self.assertEqual(q._sem_pending_getters.get_value(), 0)
-        self.assertEqual(len(results), n)
-
+    def test_shutdown_blocked_getter_gets_items(self):
+        # The items put before shutdown() go to a getter which was already
+        # waiting.
+        q = self.Queue()
+        result, = self._start_shutdown_workers(1, self._shutdown_getter, q)
+        for i in range(10):
+            q.put(i)
+        q.shutdown()
+        self.assertEqual(result(), list(range(10)))
         close_queue(q)
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_immediate_pending_put(self):
-        if self.TYPE != 'processes':
-            self.skipTest(f'test not appropriate for {self.TYPE}')
-
-        # Regardless of the value of the immediate variable,
-        # the tests are identical.
-        size = 5
-        q = self.Queue(maxsize=size)
-        results = multiprocessing.Manager().list()
-        n = 30
-        b = self.Barrier(n+1)
-        self._start_processes_pending(n, target=self._pending_put,
-                                      args=(q, b, results))
-        # Wait for all procesess start.
-        b.wait()
-
-        # We need to call _wait to be sure that queue is full,
-        # and all others processes are pending.
-        self._wait()
-
-        q.shutdown(immediate=True)
-        self.assertTrue(q.empty())
-        self.assertTrue(q._is_shutdown())
-        for p in self.ps:
-            p.join()
-
-        self.assertEqual(len(results), n)
-        self.assertEqual(results.count(pyqueue.ShutDown), n-size)
+    def test_shutdown_many_getters(self):
+        # No getter may be left waiting for an item that another one took.
+        for nitems in 1, 100:
+            with self.subTest(nitems=nitems):
+                q = self.Queue()
+                for i in range(nitems):
+                    q.put(i)
+                q.shutdown()
+                results = self._start_shutdown_workers(
+                    4, self._shutdown_getter, q)
+                items = [item for get in results for item in get()]
+                self.assertEqual(sorted(items), list(range(nitems)))
+                close_queue(q)
 
-        self.assertTrue(q.empty())
-        close_queue(q)
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_shutdown_blocked_putters(self):
+        for immediate in False, True:
+            with self.subTest(immediate=immediate):
+                q = self.Queue(maxsize=2)
+                q.put(1)
+                q.put(2)
+                results = self._start_shutdown_workers(
+                    4, self._shutdown_putter, q)
+                q.shutdown(immediate=immediate)
+                self.assertEqual([get() for get in results],
+                                 ['ShutDown'] * 4)
+                if not immediate:
+                    self.assertEqual(q.get(), 1)
+                    self.assertEqual(q.get(), 2)
+                with self.assertRaises(pyqueue.ShutDown):
+                    q.get()
+                self.assertTrue(queue_empty(q))
+                close_queue(q)
 
     @classmethod
-    def _pending_get(cls, q, barrier, results):
-        try:
-            barrier.wait()
-            results.append(q.get())
-        except pyqueue.ShutDown:
-            results.append(pyqueue.ShutDown)
+    def _shutdown_producer(cls, q, nitems, done, exit):
+        for i in range(nitems):
+            q.put((i, bytes(1000)))
+        done.set()
+        # Exiting would wait for the feeder thread to write everything to
+        # the pipe.  Stay alive instead, so that items remain in its buffer.
+        exit.wait(support.LONG_TIMEOUT)
+
+    def _start_shutdown_producer(self, q, nitems):
+        done = self.Event()
+        exit = self.Event()
+        self.addCleanup(exit.set)
+        p = self.Process(target=self._shutdown_producer,
+                         args=(q, nitems, done, exit))
+        p.daemon = True
+        p.start()
+        self.assertTrue(done.wait(support.SHORT_TIMEOUT))
+        return p, exit
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    @unittest.skipIf(sys.platform == 'darwin',
-                     "'get_value' is not implemented on MacOSX")
-    def test_queue_shutdown_count_pending_get(self):
-        if self.TYPE != 'processes':
-            self.skipTest(f'test not appropriate for {self.TYPE}')
-
+    def test_shutdown_other_process_items(self):
+        # Most of the items are too big for the pipe: they are still in the
+        # buffer of the process which put them when the queue is shut down.
         q = self.Queue()
-        results = multiprocessing.Manager().list()
-        n = 25
-        b = self.Barrier(n+1)
-        self._start_processes_pending(n, target=self._pending_get,
-                                      args=(q, b, results))
-        # Wait for all procesess start.
-        b.wait()
-
-        # wait for all pending get processes to be blocked.
-        self._wait()
-
-        self.assertLessEqual(q._sem_pending_getters.get_value(), n)
-        self.assertGreater(q._sem_pending_getters.get_value(), 0)
-        self.assertEqual(q._sem_pending_putters.get_value(), 0)
-        q.shutdown(immediate=False)
-        self.assertTrue(q._is_shutdown())
-        for p in self.ps:
-            p.join()
-
-        self.assertEqual(q._sem_pending_putters.get_value(), 0)
-        self.assertEqual(q._sem_pending_putters.get_value(), 0)
-        self.assertEqual(len(results), n)
-
+        nitems = 1000
+        p, exit = self._start_shutdown_producer(q, nitems)
+        q.shutdown()
+        for i in range(nitems):
+            self.assertEqual(q.get(timeout=support.SHORT_TIMEOUT)[0], i)
+        with self.assertRaises(pyqueue.ShutDown):
+            q.get()
+        self.assertTrue(queue_empty(q))
+        exit.set()
+        join_process(p)
         close_queue(q)
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_immediate_pending_get(self):
-        if self.TYPE != 'processes':
-            self.skipTest(f'test not appropriate for {self.TYPE}')
-
+    def test_shutdown_immediate_other_process_items(self):
+        if self.TYPE == 'processes' and sys.platform == 'win32':
+            self.skipTest('an item being written may be left on Windows')
         q = self.Queue()
-        results = multiprocessing.Manager().list()
-        n = 25
-        b = self.Barrier(n+1)
-        self._start_processes_pending(n, target=self._pending_get,
-                                      args=(q, b, results))
-        # Wait for all procesess start.
-        b.wait()
-
-        # wait for all pending get processes to be blocked.
-        self._wait()
-
+        p, exit = self._start_shutdown_producer(q, 1000)
         q.shutdown(immediate=True)
-        self._wait() # adding data as 'sentinel shutdown'.
-        self.assertTrue(q._is_shutdown())
-        for p in self.ps:
-            p.join()
-
-        self.assertEqual(len(results), n)
-        self.assertEqual(results.count(pyqueue.ShutDown), n)
-
-        self.assertTrue(q.empty())
+        with self.assertRaises(pyqueue.ShutDown):
+            q.get()
+        self.assertTrue(queue_empty(q))
+        self.assertReturnsIfImplemented(0, q.qsize)
+        exit.set()
+        join_process(p)
         close_queue(q)
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_pending_get(self):
+    def test_shutdown_feeder_error(self):
+        # A getter waiting for an item which the feeder thread fails to
+        # send is woken up.
         if self.TYPE != 'processes':
             self.skipTest(f'test not appropriate for {self.TYPE}')
 
+        class NotSerializable(object):
+            def __reduce__(self):
+                raise AttributeError
         q = self.Queue()
-        q.put("AZ67")
-        self._wait()
-
-        results = multiprocessing.Manager().list()
-        n = 25
-        b = self.Barrier(n+1)
-        self._start_processes_pending(n, target=self._pending_get,
-                                      args=(q, b, results))
-        # Wait for all procesess start.
-        b.wait()
-
-        # wait for all pending get processes to be blocked.
-        self._wait()
-
-        q.shutdown(immediate=False)
-        self.assertTrue(q._is_shutdown())
-        for p in self.ps:
-            p.join()
-
-        self.assertEqual(len(results), n)
-        self.assertEqual(results.count("AZ67"), 1)
-
-        close_queue(q)
+        result, = self._start_shutdown_workers(1, self._shutdown_getter, q)
+        with test.support.captured_stderr():
+            q.put(NotSerializable())
+            q.shutdown()
+            self.assertEqual(result(), [])
+            close_queue(q)
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def queue_shutdown_immediate_check_purge_buffer(self):
+    def test_shutdown_subclass_put(self):
+        # A subclass may override put() without calling _acquire_slot(), so
+        # that its items are not counted.  They must not be lost.
         if self.TYPE != 'processes':
             self.skipTest(f'test not appropriate for {self.TYPE}')
 
-        logger = multiprocessing.get_logger()
-        original_level = logger.level
-        logger.setLevel(logging.DEBUG)
-        stream = io.StringIO()
-        handler = logging.StreamHandler(stream)
-        logging_format = '[%(levelname)s] [%(filename)s] %(message)s'
-        handler.setFormatter(logging.Formatter(logging_format))
-        logger.addHandler(handler)
-
-        try:
-            q = self.JoinableQueue()
-            for i in range(1500):
-                q.put("tictoc")
-            q.shutdown(immediate=True)
-            self.assertTrue(q.empty())
-            q.join()
-
-            log_record = stream.getvalue()
-            self.assertIn("don't feed regular data to pipe", log_record)
+        class MyQueue(multiprocessing.queues.Queue):
+            def put(self, obj, block=True, timeout=None):
+                if not self._sem.acquire(block, timeout):
+                    raise pyqueue.Full
+                with self._notempty:
+                    if self._thread is None:
+                        self._start_thread()
+                    self._buffer.append(obj)
+                    self._notempty.notify()
+
+        q = MyQueue(ctx=multiprocessing.get_context())
+        q.put(1)
+        q.put(2)
+        self.assertEqual(q.get(), 1)
+        q.shutdown()
+        self.assertEqual(q.get(), 2)
+        with self.assertRaises(pyqueue.ShutDown):
+            q.get()
+        close_queue(q)
 
-            close_queue(q)
-        finally:
-            logger.setLevel(original_level)
-            logger.removeHandler(handler)
-            handler.close()
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_shutdown_joinable(self):
+        q = self.JoinableQueue()
+        q.put(1)
+        q.put(2)
+        q.shutdown()
+        self.assertEqual(q.get(), 1)
+        q.task_done()
+        self.assertEqual(q.get(), 2)
+        q.task_done()
+        q.join()
+        with self.assertRaises(pyqueue.ShutDown):
+            q.get()
+        close_queue(q)
 
     @classmethod
-    def _shutdown(cls, q, immediate, results, retcode):
-        q.shutdown(immediate)
-        cls._wait()
-        if not immediate:
-            while not q.empty():
-                results.append(q.get())
-                q.task_done()
-        results.insert(0, retcode)
-
-    def _join_joinablequeue(self, immediate):
-        if self.TYPE != 'processes':
-            self.skipTest(f'test not appropriate for {self.TYPE}')
+    def _shutdown_joiner(cls, q, joined):
+        q.join()
+        joined.set()
 
-        results = multiprocessing.Manager().list()
-        n = 10
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_shutdown_immediate_joinable(self):
         q = self.JoinableQueue()
-        for i in range(n):
+        for i in range(3):
             q.put(i)
-        self._wait()
-
-        return_process = 1000
-        p = self.Process(target=self._shutdown,
-                         args=(q, immediate, results,
-                              return_process+immediate
-                              )
-                        )
-        p.start()
-        q.join()
-        p.join()
-
-        if immediate:
-            self.assertEqual(len(results), 1)
-            self.assertTrue(q.empty())
-        else:
-            self.assertEqual(len(results), n+1)
-            self.assertListEqual(results[1:], list(range(n)))
-        self.assertEqual(results[0], return_process+immediate)
+        self.assertEqual(q.get(), 0)
+        q.shutdown(immediate=True)
+        with self.assertRaises(pyqueue.ShutDown):
+            q.get()
 
+        # The discarded items are done, the one which was taken is not.
+        joined = self.Event()
+        p = self.Process(target=self._shutdown_joiner, args=(q, joined))
+        p.daemon = True
+        p.start()
+        self.assertFalse(joined.wait(0.1))
+        q.task_done()
+        self.assertTrue(joined.wait(support.SHORT_TIMEOUT))
+        join_process(p)
+        with self.assertRaises(ValueError):
+            q.task_done()
         close_queue(q)
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_immediate_join_joinablequeue(self):
-        return self._join_joinablequeue(True)
-
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def test_queue_shutdown_join_joinablequeue(self):
-        return self._join_joinablequeue(False)
+    def test_shutdown_immediate_joinable_blocked_getter(self):
+        # join() returns although no task_done() is ever called.
+        q = self.JoinableQueue()
+        result, = self._start_shutdown_workers(1, self._shutdown_getter, q)
+        q.shutdown(immediate=True)
+        self.assertEqual(result(), [])
+        q.join()
+        close_queue(q)
 
 #
 #
diff --git a/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst b/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst
index a0ae0c823241a2..9ee370289436ad 100644
--- a/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst
+++ b/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst
@@ -1 +1,3 @@
-Add :py:class:`multiprocessing.Queue` termination with ``multiprocessing.Queue.shutdown`` method.
+Add :meth:`multiprocessing.Queue.shutdown` to terminate a
+:class:`multiprocessing.Queue` or a :class:`multiprocessing.JoinableQueue`
+in every process which uses it, like :meth:`queue.Queue.shutdown`.