gh-94777: Fix deadlock in ProcessPoolExecutor (#94784)

Fixes a hang in multiprocessing process pool executor when a child process crashes and code could otherwise block on writing to the pipe.  See GH-94777 for more details.

diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py
index 816edab99f63e33d7da61e6acb84a1bc17e66d70..301207f59de37a5e312de93765268883d7401b20 100644 (file)
--- a/Lib/concurrent/futures/process.py
+++ b/Lib/concurrent/futures/process.py
@@ -499,6 +499,10 @@ class _ExecutorManagerThread(threading.Thread):
         for p in self.processes.values():
             p.terminate()
 
+        # Prevent queue writing to a pipe which is no longer read.
+        # https://github.com/python/cpython/issues/94777
+        self.call_queue._reader.close()
+
         # clean up resources
         self.join_executor_internals()
 

diff --git a/Lib/test/test_concurrent_futures.py b/Lib/test/test_concurrent_futures.py
index a20cb844a293c929e8a1f34221d5a87d5791de1b..39dbe234e765e85402e65f23e43c5e96adc7bdd0 100644 (file)
--- a/Lib/test/test_concurrent_futures.py
+++ b/Lib/test/test_concurrent_futures.py
@@ -1172,6 +1172,11 @@ def _crash(delay=None):
     faulthandler._sigsegv()
 
 
+def _crash_with_data(data):
+    """Induces a segfault with dummy data in input."""
+    _crash()
+
+
 def _exit():
     """Induces a sys exit with exitcode 1."""
     sys.exit(1)
@@ -1371,6 +1376,19 @@ class ExecutorDeadlockTest:
         # dangling threads
         executor_manager.join()
 
+    def test_crash_big_data(self):
+        # Test that there is a clean exception instad of a deadlock when a
+        # child process crashes while some data is being written into the
+        # queue.
+        # https://github.com/python/cpython/issues/94777
+        self.executor.shutdown(wait=True)
+        data = "a" * support.PIPE_MAX_SIZE
+        with self.executor_type(max_workers=2,
+                                mp_context=self.get_context()) as executor:
+            self.executor = executor  # Allow clean up in fail_on_deadlock
+            with self.assertRaises(BrokenProcessPool):
+                list(executor.map(_crash_with_data, [data] * 10))
+
 
 create_executor_tests(ExecutorDeadlockTest,
                       executor_mixins=(ProcessPoolForkMixin,

diff --git a/Misc/NEWS.d/next/Library/2022-07-12-18-45-13.gh-issue-94777.mOybx7.rst b/Misc/NEWS.d/next/Library/2022-07-12-18-45-13.gh-issue-94777.mOybx7.rst
new file mode 100644 (file)
index 0000000..2c04a35
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2022-07-12-18-45-13.gh-issue-94777.mOybx7.rst
@@ -0,0 +1 @@
+Fix hanging :mod:`multiprocessing` ``ProcessPoolExecutor`` when a child process crashes while data is being written in the call queue.