gh-158570: Fix ProcessPoolExecutor shutdown deadlock when the result pipe is full

Lib/concurrent/futures/process.py

@@ -249,8 +249,6 @@ def _process_worker(call_queue, result_queue, initializer, initargs, max_tasks=N
     while True:
         call_item = call_queue.get(block=True)
         if call_item is None:
-            # Wake up queue management thread
-            result_queue.put(os.getpid())
             return
 
         if max_tasks is not None:
@@ -531,9 +529,6 @@ def wait_result_broken_or_wakeup(self):
         return result_item, is_broken, cause
 
     def process_result_item(self, result_item):
-        # Process the received a result_item. This can be either the PID of a
-        # worker that exited gracefully or a _ResultItem
-
         # Received a _ResultItem so mark the future as completed.
         work_item = self.pending_work_items.pop(result_item.work_id, None)
         # work_item can be None if another process terminated (see above)

Lib/test/test_concurrent_futures/test_deadlock.py

@@ -2,6 +2,7 @@
 import queue
 import signal
 import sys
+import threading
 import time
 import unittest
 import unittest.mock
@@ -10,7 +11,7 @@
 from concurrent.futures.process import BrokenProcessPool, _ThreadWakeup
 
 from test import support
-from test.support import warnings_helper
+from test.support import threading_helper, warnings_helper
 
 from .util import (
     create_executor_tests, setup_module,
@@ -329,6 +330,55 @@ def clear(self):
                     signal.signal(signal.SIGALRM, old_handler)
 
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_shutdown_should_not_deadlock_if_result_pipe_full(self):
+        # Workers must not write to the result queue on shutdown: the
+        # executor manager thread does not read it while it joins them.
+        # Fill the pipe once the executor manager thread stops reading
+        # results, right before the workers are told to exit.
+        self.executor.shutdown(wait=True)
+
+        data = b"a" * support.PIPE_MAX_SIZE
+        join_executor_internals = (
+            futures.process._ExecutorManagerThread._join_executor_internals)
+        def mock_join_executor_internals(self, broken=False):
+            # The put() blocks holding the result queue's write lock until
+            # the pipe is drained, so any worker writing to the result queue
+            # on exit blocks behind it.
+            filler = threading.Thread(target=self.result_queue.put,
+                                      args=(data,))
+            filler.start()
+            wlock = self.result_queue._wlock
+            if wlock is not None:
+                # Wait for the filler to hold the write lock.
+                while wlock.acquire(block=False):
+                    wlock.release()
+                    time.sleep(0.001)
+            join_executor_internals(self, broken)
+            # Unblock the filler.
+            self.result_queue.get()
+            filler.join()
+
+        executor = self.executor_type(max_workers=2,
+                                      mp_context=self.get_context())
+        self.executor = executor  # Allow clean up in fail_on_deadlock
+        with unittest.mock.patch.object(futures.process._ExecutorManagerThread,
+                                        '_join_executor_internals',
+                                        mock_join_executor_internals):
+            self.assertEqual(list(executor.map(int, range(10))),
+                             list(range(10)))
+            manager_thread = executor._executor_manager_thread
+            processes = tuple(executor._processes.values())
+            executor.shutdown(wait=False)
+            try:
+                threading_helper.join_thread(manager_thread, self.TIMEOUT)
+            finally:
+                if manager_thread.is_alive():
+                    for process in processes:
+                        process.terminate()
+                    threading_helper.join_thread(manager_thread, self.TIMEOUT)
+
+
 create_executor_tests(globals(), ExecutorDeadlockTest,
                       executor_mixins=(ProcessPoolForkMixin,
                                        ProcessPoolForkserverMixin,

Misc/NEWS.d/next/Library/2026-10-01-18-46-33.gh-issue-158570.rixXY7.rst

@@ -0,0 +1,4 @@
+Fix a deadlock in :meth:`ProcessPoolExecutor.shutdown()
+<concurrent.futures.Executor.shutdown>` when the messages that exiting
+workers send back fill up the pipe used to return results. This could
+happen with many workers or a small pipe buffer.