Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions Lib/asyncio/proactor_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
73 changes: 73 additions & 0 deletions Lib/test/test_asyncio/test_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
Original file line number Diff line number Diff line change
@@ -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.
Loading