From b6700c6ded98ee63ac50b76307850b9b7eb0c63e Mon Sep 17 00:00:00 2001 From: Junaid Khan Date: Mon, 19 Oct 2026 12:52:02 +0530 Subject: [PATCH] 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. Co-authored-by: Cursor --- 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 7f4f225c0ad4fbd..51e58f6d5de833b 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 be2fc3edf5dce66..b3c1106a4164b44 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 dafbda862c51c24..db26021ab495d6d 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 000000000000000..058df3b123edde1 --- /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.