From 3006d6e730748931a023ad37ed887417c321d2d0 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Mon, 31 Aug 2026 14:45:35 +0100 Subject: [PATCH 1/4] fix proactor error hang Fixes #156698 --- Lib/asyncio/proactor_events.py | 6 ++ Lib/test/test_asyncio/test_events.py | 73 +++++++++++++++++++ ...-08-31-00-00-00.gh-issue-156698.gyDxUe.rst | 4 + 3 files changed, 83 insertions(+) create mode 100644 Misc/NEWS.d/next/Library/2026-08-31-00-00-00.gh-issue-156698.gyDxUe.rst diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py index f18a7fe58558155..764f86d154a7222 100644 --- a/Lib/asyncio/proactor_events.py +++ b/Lib/asyncio/proactor_events.py @@ -534,6 +534,12 @@ def _loop_writing(self, fut=None): addr=addr) except OSError as exc: self._protocol.error_received(exc) + if self._buffer and not self._conn_lost: + # Re-arm the write loop so buffered data isn't stranded and + # a paused protocol is eventually resumed (gh-156698). + self._loop.call_soon(self._loop_writing) + else: + self._maybe_resume_protocol() except Exception as exc: self._fatal_error(exc, 'Fatal write error on datagram transport') else: diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py index db316fae090280a..7817487911f3dd5 100644 --- a/Lib/test/test_asyncio/test_events.py +++ b/Lib/test/test_asyncio/test_events.py @@ -1583,6 +1583,79 @@ def create_socket(): transport_1.close() transport_2.close() + def _test_datagram_write_error_resumes_paused_protocol(self, first, second): + # See https://github.com/python/cpython/issues/156698: a + # datagram write error must not strand data left in the write + # buffer, nor leave a paused protocol paused forever. + loop = self.loop + + class Protocol(asyncio.DatagramProtocol): + def connection_made(self, transport): + self.transport = transport + self.paused = False + self.resumed = False + self.errors = [] + self.error_received_event = loop.create_future() + + def pause_writing(self): + self.paused = True + + def resume_writing(self): + self.resumed = True + + def error_received(self, exc): + self.errors.append(exc) + if not self.error_received_event.done(): + self.error_received_event.set_result(None) + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setblocking(False) + sock.bind(('127.0.0.1', 0)) + transport, protocol = loop.run_until_complete( + loop.create_datagram_endpoint(Protocol, sock=sock)) + addr = sock.getsockname() + + # A high water mark of 0 makes pausing deterministic whenever + # anything is left in the write buffer. + transport.set_write_buffer_limits(0) + + # The first sendto() may arm an in-flight write, so the second + # one can end up queued behind it; queuing is what trips + # pause_writing() at a high water mark of 0. + transport.sendto(first, addr) + transport.sendto(second, addr) + + loop.run_until_complete( + asyncio.wait_for(protocol.error_received_event, 10)) + self.assertTrue(protocol.errors) + self.assertIsInstance(protocol.errors[0], OSError) + + # The write buffer must not be left stranded. + test_utils.run_until( + loop, lambda: transport.get_write_buffer_size() == 0) + + # A protocol that got paused must eventually be resumed too -- + # without requiring an unsolicited extra sendto() to un-stick it. + if protocol.paused: + test_utils.run_until(loop, lambda: protocol.resumed) + + transport.close() + test_utils.run_briefly(loop) + + def test_datagram_write_error_resumes_paused_protocol_in_flight(self): + # oversized datagram fails while in flight; a normal datagram + # queued right behind it must not be stranded. + oversized = b'\x00' * 70000 + self._test_datagram_write_error_resumes_paused_protocol( + oversized, b'queued') + + def test_datagram_write_error_resumes_paused_protocol_from_callback(self): + # oversized datagram fails once it reaches the front of the + # buffer; the protocol must not stay paused forever. + oversized = b'\x00' * 70000 + self._test_datagram_write_error_resumes_paused_protocol( + b'ok', oversized) + def test_internal_fds(self): loop = self.create_event_loop() if not isinstance(loop, selector_events.BaseSelectorEventLoop): diff --git a/Misc/NEWS.d/next/Library/2026-08-31-00-00-00.gh-issue-156698.gyDxUe.rst b/Misc/NEWS.d/next/Library/2026-08-31-00-00-00.gh-issue-156698.gyDxUe.rst new file mode 100644 index 000000000000000..4e3a292ce9b8217 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-08-31-00-00-00.gh-issue-156698.gyDxUe.rst @@ -0,0 +1,4 @@ +Fix :class:`asyncio.ProactorEventLoop` UDP transports so that a write +error no longer strands a paused protocol: the write loop is now +rescheduled when data remains buffered, and the protocol is resumed +when the buffer has drained. From d57cb23e9bbdbde9332de9dae1a668f6615d987f Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Fri, 4 Sep 2026 07:19:41 +0100 Subject: [PATCH 2/4] fix DatagramTransport when errror_recieved calls sendto --- Lib/asyncio/proactor_events.py | 10 ++++- Lib/test/test_asyncio/test_events.py | 56 ++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 2 deletions(-) diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py index 764f86d154a7222..03307058ff1cbec 100644 --- a/Lib/asyncio/proactor_events.py +++ b/Lib/asyncio/proactor_events.py @@ -534,10 +534,16 @@ def _loop_writing(self, fut=None): addr=addr) except OSError as exc: self._protocol.error_received(exc) - if self._buffer and not self._conn_lost: + if self._buffer: # Re-arm the write loop so buffered data isn't stranded and # a paused protocol is eventually resumed (gh-156698). - self._loop.call_soon(self._loop_writing) + def resume_writing(): + # a sendto() may have armed a write in the meantime; + # its own callback will drain the rest of the buffer. + if self._write_fut is None: + self._loop_writing() + + self._loop.call_soon(resume_writing) else: self._maybe_resume_protocol() except Exception as exc: diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py index 7817487911f3dd5..0dc78856531dd63 100644 --- a/Lib/test/test_asyncio/test_events.py +++ b/Lib/test/test_asyncio/test_events.py @@ -1656,6 +1656,62 @@ def test_datagram_write_error_resumes_paused_protocol_from_callback(self): self._test_datagram_write_error_resumes_paused_protocol( b'ok', oversized) + def test_datagram_write_error_reentrant_sendto(self): + # See https://github.com/python/cpython/issues/156698: an + # error_received() callback that sends more data synchronously + # can itself arm a new write. The write-loop restart scheduled + # for the failed write must notice that and not try to start a + # second, conflicting one. + loop = self.loop + unhandled = [] + loop.set_exception_handler(lambda loop, context: unhandled.append(context)) + + class Protocol(asyncio.DatagramProtocol): + def connection_made(self, transport): + self.transport = transport + self.sent_extra = False + self.errors = [] + self.done = loop.create_future() + + def datagram_received(self, data, addr): + if not self.done.done(): + self.done.set_result(None) + + def error_received(self, exc): + self.errors.append(exc) + if not self.sent_extra: + # Reentrantly kicks off another write while the + # failing one is still unwinding on the stack. + self.sent_extra = True + self.transport.sendto(b'extra', self.addr) + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setblocking(False) + sock.bind(('127.0.0.1', 0)) + transport, protocol = loop.run_until_complete( + loop.create_datagram_endpoint(Protocol, sock=sock)) + protocol.addr = addr = sock.getsockname() + + oversized = b'\x00' * 70000 + transport.sendto(oversized, addr) + transport.sendto(b'queued', addr) + + # The 'extra' datagram sent from error_received() is delivered + # back to the same socket; waiting for it proves the write loop + # kept running instead of wedging or crashing. + loop.run_until_complete(asyncio.wait_for(protocol.done, 10)) + + test_utils.run_until( + loop, lambda: transport.get_write_buffer_size() == 0) + + transport.close() + test_utils.run_briefly(loop) + + self.assertTrue(protocol.errors) + self.assertFalse( + unhandled, + f'unhandled exception in the write loop: {unhandled}') + def test_internal_fds(self): loop = self.create_event_loop() if not isinstance(loop, selector_events.BaseSelectorEventLoop): From 37a9b5b40c0d2296f2835748647cc336e5758950 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sat, 5 Sep 2026 17:33:59 +0100 Subject: [PATCH 3/4] Update Lib/asyncio/proactor_events.py Co-authored-by: Kumar Aditya --- Lib/asyncio/proactor_events.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py index 2354b9b02c4b473..1be0a8f777974b2 100644 --- a/Lib/asyncio/proactor_events.py +++ b/Lib/asyncio/proactor_events.py @@ -535,7 +535,7 @@ def _loop_writing(self, fut=None): except OSError as exc: self._protocol.error_received(exc) if self._buffer: - # Re-arm the write loop so buffered data isn't stranded and + # Reschedule the write loop so buffered data isn't stranded and # a paused protocol is eventually resumed (gh-156698). def resume_writing(): # a sendto() may have armed a write in the meantime; From 9eda8d041c9de05012eaa5994d7182b708c31e47 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 6 Sep 2026 11:34:23 +0100 Subject: [PATCH 4/4] don't resume a closed transport --- Lib/asyncio/proactor_events.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py index 1be0a8f777974b2..854f3a254938edc 100644 --- a/Lib/asyncio/proactor_events.py +++ b/Lib/asyncio/proactor_events.py @@ -534,7 +534,7 @@ def _loop_writing(self, fut=None): addr=addr) except OSError as exc: self._protocol.error_received(exc) - if self._buffer: + if self._buffer or self._closing: # Reschedule the write loop so buffered data isn't stranded and # a paused protocol is eventually resumed (gh-156698). def resume_writing():