From 9f8bdcc1a200d8c4b73de736bdbf50ca8b045c1c Mon Sep 17 00:00:00 2001 From: Junaid Khan Date: Mon, 19 Oct 2026 12:52:02 +0530 Subject: [PATCH 1/4] gh-159689: Fix ProcessPoolExecutor hang when a worker is killed mid-result A worker killed after writing part of a large result left the manager thread blocked in Connection.recv(). The parent holds the result pipe's write end, so that death is not delivered as EOF. Watch worker sentinels for the rest of the message and raise BrokenProcessPool. --- Lib/concurrent/futures/process.py | 11 +- Lib/multiprocessing/connection.py | 124 +++++++++++++++--- .../test_process_pool.py | 116 +++++++++++++++- ...0-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst | 3 + 4 files changed, 231 insertions(+), 23 deletions(-) create mode 100644 Misc/NEWS.d/next/Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py index 7f4f225c0ad4fb..51e58f6d5de833 100644 --- a/Lib/concurrent/futures/process.py +++ b/Lib/concurrent/futures/process.py @@ -52,6 +52,7 @@ # This import is required to load the multiprocessing.connection submodule # so that it can be accessed later as `mp.connection` import multiprocessing.connection +from multiprocessing.connection import _SentinelReady from multiprocessing.queues import Queue import threading import weakref @@ -518,8 +519,16 @@ def wait_result_broken_or_wakeup(self): result_item = None if result_reader in ready: try: - result_item = result_reader.recv() + # Keep watching worker sentinels for the rest of the + # message. recv() otherwise blocks in plain reads, and the + # parent's write end of this pipe prevents a killed worker + # from delivering EOF. gh-159689. + result_item = result_reader.recv(sentinels=worker_sentinels) is_broken = False + except _SentinelReady: + # A worker exited while its result was only partly written. + # Leave is_broken set so the exit code is reported. + pass except BaseException as exc: cause = format_exception(exc) diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py index be2fc3edf5dce6..b3c1106a4164b4 100644 --- a/Lib/multiprocessing/connection.py +++ b/Lib/multiprocessing/connection.py @@ -65,6 +65,20 @@ def _init_timeout(timeout=CONNECTION_TIMEOUT): def _check_timeout(t): return time.monotonic() > t + +class _SentinelReady(Exception): + """A wait sentinel became readable while a connection read was in progress. + + Used by ProcessPoolExecutor so a worker killed mid-result does not leave + the manager thread blocked in a read that can never see EOF. The parent's + copy of the result pipe's write end stays open for the executor's + lifetime, so the death of the writer is not delivered as end-of-file. + """ + + def __init__(self, ready): + super().__init__(ready) + self.ready = ready + # # # @@ -253,11 +267,18 @@ def recv_bytes_into(self, buf, offset=0): (offset + size) // itemsize]) return size - def recv(self): - """Receive a (picklable) object""" + def recv(self, *, sentinels=()): + """Receive a (picklable) object. + + *sentinels* is an optional sequence of objects accepted by + :func:`wait`, such as process sentinels. Before each read, those + objects are waited on together with this connection. If one of them + is ready and this connection is not, :class:`_SentinelReady` is + raised instead of blocking until the rest of the message arrives. + """ self._check_closed() self._check_readable() - buf = self._recv_bytes() + buf = self._recv_bytes(sentinels=sentinels) return _ForkingPickler.loads(buf.getbuffer()) def poll(self, timeout=0.0): @@ -317,7 +338,31 @@ def _send_bytes(self, buf): assert err == 0 assert nwritten == len(buf) - def _recv_bytes(self, maxsize=None): + def _wait_overlapped_or_sentinels(self, ov, sentinels): + """Block until *ov* completes, or a sentinel is signaled.""" + if not sentinels: + waitres = _winapi.WaitForMultipleObjects( + [ov.event], False, INFINITE) + assert waitres == WAIT_OBJECT_0 + return + sentinels = list(sentinels) + handles = [ov.event, *(int(item) for item in sentinels)] + waitres = _winapi.WaitForMultipleObjects(handles, False, INFINITE) + if waitres == WAIT_OBJECT_0: + return + ov.cancel() + try: + ov.GetOverlappedResult(True) + except OSError: + pass + idx = waitres - WAIT_OBJECT_0 + if not 0 < idx <= len(sentinels): + idx = waitres - WAIT_ABANDONED_0 + if 0 < idx <= len(sentinels): + raise _SentinelReady([sentinels[idx - 1]]) + raise OSError("failed waiting for pipe data") + + def _recv_bytes(self, maxsize=None, sentinels=()): if self._got_empty_message: self._got_empty_message = False return io.BytesIO() @@ -329,23 +374,29 @@ def _recv_bytes(self, maxsize=None): sentinel = object() return_value = sentinel + sentinel_ready = False try: try: if err == _winapi.ERROR_IO_PENDING: - waitres = _winapi.WaitForMultipleObjects( - [ov.event], False, INFINITE) - assert waitres == WAIT_OBJECT_0 + self._wait_overlapped_or_sentinels( + ov, sentinels) + except _SentinelReady: + # Already cancelled and reaped above. + sentinel_ready = True + raise except: ov.cancel() raise finally: - nread, err = ov.GetOverlappedResult(True) - if err == 0: - f = io.BytesIO() - f.write(ov.getbuffer()) - return_value = f - elif err == _winapi.ERROR_MORE_DATA: - return_value = self._get_more_data(ov, maxsize) + if not sentinel_ready: + nread, err = ov.GetOverlappedResult(True) + if err == 0: + f = io.BytesIO() + f.write(ov.getbuffer()) + return_value = f + elif err == _winapi.ERROR_MORE_DATA: + return_value = self._get_more_data( + ov, maxsize, sentinels) except: if return_value is sentinel: raise @@ -365,11 +416,28 @@ def _poll(self, timeout): return True return bool(wait([self], timeout)) - def _get_more_data(self, ov, maxsize): + def _get_more_data(self, ov, maxsize, sentinels=()): buf = ov.getbuffer() f = io.BytesIO() f.write(buf) left = _winapi.PeekNamedPipe(self._handle)[1] + while left == 0 and sentinels: + # The rest of this message has not been written yet. The + # parent keeps its own write end open, so a killed writer + # does not surface as EOF. Poll the sentinels directly: + # waiting on this pipe here would start another overlapped + # read and consume the message. + handles = [int(item) for item in sentinels] + waitres = _winapi.WaitForMultipleObjects(handles, False, 50) + if waitres == WAIT_TIMEOUT: + left = _winapi.PeekNamedPipe(self._handle)[1] + continue + idx = waitres - WAIT_OBJECT_0 + if not 0 <= idx < len(handles): + idx = waitres - WAIT_ABANDONED_0 + if 0 <= idx < len(handles): + raise _SentinelReady([sentinels[idx]]) + raise OSError("failed waiting for pipe data") assert left > 0 if maxsize is not None and len(buf) + left > maxsize: self._bad_message_length() @@ -407,11 +475,27 @@ def _send(self, buf, write=_write): break buf = buf[n:] - def _recv(self, size, read=_read): + def _abandon_read_if_sentinel_ready(self, sentinels): + """Raise if a sentinel is ready and this connection has no data. + + A non-blocking poll covers the common case where the next chunk is + already buffered, so a large result does not block once per chunk. + """ + ready = wait([self, *sentinels], timeout=0) + if self in ready: + return + if not ready: + ready = wait([self, *sentinels]) + if self not in ready: + raise _SentinelReady(ready) + + def _recv(self, size, sentinels=(), read=_read): buf = io.BytesIO() handle = self._handle remaining = size while remaining > 0: + if sentinels: + self._abandon_read_if_sentinel_ready(sentinels) to_read = min(BUFSIZE, remaining) chunk = read(handle, to_read) n = len(chunk) @@ -447,15 +531,15 @@ def _send_bytes(self, buf): # to avoid "broken pipe" errors if the other end closed the pipe. self._send(header + buf) - def _recv_bytes(self, maxsize=None): - buf = self._recv(4) + def _recv_bytes(self, maxsize=None, sentinels=()): + buf = self._recv(4, sentinels) size, = struct.unpack("!i", buf.getvalue()) if size == -1: - buf = self._recv(8) + buf = self._recv(8, sentinels) size, = struct.unpack("!Q", buf.getvalue()) if maxsize is not None and size > maxsize: return None - return self._recv(size) + return self._recv(size, sentinels) def _poll(self, timeout): r = wait([self], timeout) diff --git a/Lib/test/test_concurrent_futures/test_process_pool.py b/Lib/test/test_concurrent_futures/test_process_pool.py index dafbda862c51c2..db26021ab495d6 100644 --- a/Lib/test/test_concurrent_futures/test_process_pool.py +++ b/Lib/test/test_concurrent_futures/test_process_pool.py @@ -1,5 +1,8 @@ +import multiprocessing as mp import os import queue +import signal +import struct import sys import threading import time @@ -16,8 +19,8 @@ from .executor import ExecutorTest, mul from .util import ( - ProcessPoolForkMixin, ProcessPoolForkserverMixin, ProcessPoolSpawnMixin, - create_executor_tests, setup_module) + BaseTestCase, ProcessPoolForkMixin, ProcessPoolForkserverMixin, + ProcessPoolSpawnMixin, create_executor_tests, setup_module) class EventfulGCObj(): @@ -50,6 +53,40 @@ def _report_wait_return(queue, event, value): return value +def _send_result_header_only(writer): + """Write a length header and exit before the message body.""" + writer._send(struct.pack("!i", 100_000)) + + +def _send_object(writer): + writer.send({"ok": True}) + + +def _return_large_result(): + # Larger than Connection._send_bytes' 16384-byte inline threshold, so the + # length header is written separately from the body. + return b"x" * (32 * 1024) + + +def _pause_after_result_header(writer): + """Stop this worker after the result length is written, before the body.""" + import multiprocessing.connection as connection + + original = connection.Connection._send_bytes + + def _send_bytes(self, buf): + n = len(buf) + if n > 16384: + self._send(struct.pack("!i", n)) + writer.send_bytes(b"1") + os.kill(os.getpid(), signal.SIGSTOP) + self._send(buf) + else: + original(self, buf) + + connection.Connection._send_bytes = _send_bytes + + class ProcessPoolExecutorTest(ExecutorTest): @unittest.skipUnless(sys.platform=='win32', 'Windows-only process limit') @@ -58,6 +95,33 @@ def test_max_workers_too_large(self): "max_workers must be <= 61"): futures.ProcessPoolExecutor(max_workers=62) + @unittest.skipUnless(hasattr(signal, "SIGSTOP"), + "requires SIGSTOP to pause a worker mid-send") + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_killed_child_during_result_send(self): + # gh-159689: a worker killed after writing the result length but + # before the body must fail the future. The parent holds the + # result pipe's write end, so this is not delivered as EOF. + ctx = self.get_context() + reader, writer = ctx.Pipe(duplex=False) + executor = self.executor_type( + max_workers=1, + mp_context=ctx, + initializer=_pause_after_result_header, + initargs=(writer,)) + try: + future = executor.submit(_return_large_result) + self.assertTrue(reader.poll(support.SHORT_TIMEOUT)) + self.assertEqual(reader.recv_bytes(), b"1") + worker = next(iter(executor._processes.values())) + os.kill(worker.pid, signal.SIGKILL) + with self.assertRaises(BrokenProcessPool): + future.result(timeout=support.SHORT_TIMEOUT) + finally: + executor.shutdown(wait=True, cancel_futures=True) + writer.close() + reader.close() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_killed_child(self): # When a child process is abruptly terminated, the whole pool gets @@ -543,6 +607,54 @@ def test_force_shutdown_workers_stops_pool(self, function_name): break +@unittest.skipUnless(sys.platform != "win32", + "partial pipe reads are specific to Unix connections") +class ResultPipeSentinelTest(BaseTestCase): + """Connection.recv(sentinels=...) when a writer dies mid-message. + + The parent keeps its copy of the write end, which is how + ProcessPoolExecutor's result queue is set up. gh-159689. + """ + + def _context(self): + return mp.get_context("spawn") + + def test_partial_message_raises_when_writer_exits(self): + ctx = self._context() + reader, writer = ctx.Pipe(duplex=False) + proc = ctx.Process(target=_send_result_header_only, args=(writer,)) + try: + proc.start() + proc.join(support.SHORT_TIMEOUT) + self.assertEqual(proc.exitcode, 0) + started = time.monotonic() + with self.assertRaises(mp.connection._SentinelReady): + reader.recv(sentinels=[proc.sentinel]) + self.assertLess(time.monotonic() - started, support.SHORT_TIMEOUT) + finally: + writer.close() + reader.close() + proc.join(support.SHORT_TIMEOUT) + proc.close() + + def test_complete_message_still_received(self): + ctx = self._context() + reader, writer = ctx.Pipe(duplex=False) + proc = ctx.Process(target=_send_object, args=(writer,)) + try: + proc.start() + proc.join(support.SHORT_TIMEOUT) + self.assertEqual(proc.exitcode, 0) + writer.close() + self.assertEqual(reader.recv(sentinels=[proc.sentinel]), + {"ok": True}) + finally: + writer.close() + reader.close() + proc.join(support.SHORT_TIMEOUT) + proc.close() + + create_executor_tests(globals(), ProcessPoolExecutorTest, executor_mixins=(ProcessPoolForkMixin, ProcessPoolForkserverMixin, diff --git a/Misc/NEWS.d/next/Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst b/Misc/NEWS.d/next/Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst new file mode 100644 index 00000000000000..058df3b123edde --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst @@ -0,0 +1,3 @@ +Fix :class:`concurrent.futures.ProcessPoolExecutor` hanging when a worker +is killed while writing a large result. The pending future now raises +:exc:`concurrent.futures.BrokenProcessPool`, and the interpreter can exit. From bcbd6edff36a8cec815a409424f64465f342320e Mon Sep 17 00:00:00 2001 From: Junaid Khan Date: Mon, 19 Oct 2026 13:14:57 +0530 Subject: [PATCH 2/4] gh-159689: Use a class reference for BrokenProcessPool in the news entry --- .../Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Misc/NEWS.d/next/Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst b/Misc/NEWS.d/next/Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst index 058df3b123edde..62cd05eb2370cc 100644 --- a/Misc/NEWS.d/next/Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst +++ b/Misc/NEWS.d/next/Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst @@ -1,3 +1,3 @@ Fix :class:`concurrent.futures.ProcessPoolExecutor` hanging when a worker is killed while writing a large result. The pending future now raises -:exc:`concurrent.futures.BrokenProcessPool`, and the interpreter can exit. +:class:`concurrent.futures.BrokenProcessPool`, and the interpreter can exit. From 5626bf2294c04598e8a81c81283c18339270c4e6 Mon Sep 17 00:00:00 2001 From: Junaid Khan Date: Mon, 19 Oct 2026 13:23:21 +0530 Subject: [PATCH 3/4] gh-159689: Point the news entry at the documented BrokenProcessPool exception --- .../Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Misc/NEWS.d/next/Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst b/Misc/NEWS.d/next/Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst index 62cd05eb2370cc..509b23c3abe4b9 100644 --- a/Misc/NEWS.d/next/Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst +++ b/Misc/NEWS.d/next/Library/2026-10-19-12-20-00.gh-issue-159689.kH3nP9xQ.rst @@ -1,3 +1,3 @@ Fix :class:`concurrent.futures.ProcessPoolExecutor` hanging when a worker is killed while writing a large result. The pending future now raises -:class:`concurrent.futures.BrokenProcessPool`, and the interpreter can exit. +:exc:`~concurrent.futures.process.BrokenProcessPool`, and the interpreter can exit. From 0f945f2949db73f724836842339ffe85800a6d6e Mon Sep 17 00:00:00 2001 From: Junaid Khan Date: Mon, 19 Oct 2026 14:11:52 +0530 Subject: [PATCH 4/4] gh-159689: Skip sentinel pipe tests when multiprocessing is unavailable --- .../test_process_pool.py | 27 ++++++++++++++----- 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/Lib/test/test_concurrent_futures/test_process_pool.py b/Lib/test/test_concurrent_futures/test_process_pool.py index db26021ab495d6..2d3a6f3ca4216b 100644 --- a/Lib/test/test_concurrent_futures/test_process_pool.py +++ b/Lib/test/test_concurrent_futures/test_process_pool.py @@ -11,7 +11,7 @@ import unittest.mock import weakref from concurrent import futures -from concurrent.futures.process import BrokenProcessPool +from concurrent.futures.process import BrokenProcessPool, _check_system_limits from test import support from test.support import hashlib_helper, threading_helper, warnings_helper @@ -616,6 +616,13 @@ class ResultPipeSentinelTest(BaseTestCase): ProcessPoolExecutor's result queue is set up. gh-159689. """ + def setUp(self): + super().setUp() + try: + _check_system_limits() + except NotImplementedError: + self.skipTest("ProcessPoolExecutor unavailable on this system") + def _context(self): return mp.get_context("spawn") @@ -623,26 +630,31 @@ def test_partial_message_raises_when_writer_exits(self): ctx = self._context() reader, writer = ctx.Pipe(duplex=False) proc = ctx.Process(target=_send_result_header_only, args=(writer,)) + started = False try: proc.start() + started = True proc.join(support.SHORT_TIMEOUT) self.assertEqual(proc.exitcode, 0) - started = time.monotonic() + t0 = time.monotonic() with self.assertRaises(mp.connection._SentinelReady): reader.recv(sentinels=[proc.sentinel]) - self.assertLess(time.monotonic() - started, support.SHORT_TIMEOUT) + self.assertLess(time.monotonic() - t0, support.SHORT_TIMEOUT) finally: writer.close() reader.close() - proc.join(support.SHORT_TIMEOUT) - proc.close() + if started: + proc.join(support.SHORT_TIMEOUT) + proc.close() def test_complete_message_still_received(self): ctx = self._context() reader, writer = ctx.Pipe(duplex=False) proc = ctx.Process(target=_send_object, args=(writer,)) + started = False try: proc.start() + started = True proc.join(support.SHORT_TIMEOUT) self.assertEqual(proc.exitcode, 0) writer.close() @@ -651,8 +663,9 @@ def test_complete_message_still_received(self): finally: writer.close() reader.close() - proc.join(support.SHORT_TIMEOUT) - proc.close() + if started: + proc.join(support.SHORT_TIMEOUT) + proc.close() create_executor_tests(globals(), ProcessPoolExecutorTest,