# Lib/asyncio/proactor_events.py, _ProactorDatagramTransport._loop_writing except OSError as exc: self._protocol.error_received(exc) except Exception as exc: self._fatal_error(exc, 'Fatal write error on datagram transport') else: self._write_fut.add_done_callback(self._loop_writing) self._maybe_resume_protocol()

 except OSError as exc: self._protocol.error_received(exc) if self._buffer and not self._conn_lost: self._loop.call_soon(self._loop_writing) else: self._maybe_resume_protocol()

""" Does _ProactorDatagramTransport._loop_writing() strand a paused protocol?  Run on Windows (ProactorEventLoop). See README.md for what the output means. """ from __future__ import annotations import asyncio import sys OVERSIZED = b"\x00" * 70000 # guaranteed oversend error (WSAEMSGSIZE or, on some # builds, WinError 1784 ERROR_INVALID_USER_BUFFER); either way it fails the write class Protocol(asyncio.DatagramProtocol): def connection_made(self, transport: asyncio.DatagramTransport) -> None: self.transport = transport self.events: list[str] = [] self.error = asyncio.get_running_loop().create_future() # a high water mark of 0 makes the pause deterministic; it is what # anyio's UDP sockets use transport.set_write_buffer_limits(0) def datagram_received(self, data: bytes, addr: object) -> None: self.events.append(f"datagram_received({len(data)} bytes)") def pause_writing(self) -> None: self.events.append("pause_writing") def resume_writing(self) -> None: self.events.append("resume_writing") def error_received(self, exc: Exception) -> None: self.events.append(f"error_received({exc})") if not self.error.done(): self.error.set_result(exc) async def scenario(name: str, first: bytes, second: bytes) -> None: loop = asyncio.get_running_loop() transport, protocol = await loop.create_datagram_endpoint( Protocol, local_addr=("127.0.0.1", 0) ) addr = transport.get_extra_info("sockname") # the first sendto() arms the overlapped write, so the second one has to be # buffered behind it, which trips pause_writing() at a high water mark of 0 transport.sendto(first, addr) transport.sendto(second, addr) try: await asyncio.wait_for(protocol.error, 5) except asyncio.TimeoutError: protocol.events.append("<no error reported>") await asyncio.sleep(0.5) paused = "pause_writing" in protocol.events resumed = "resume_writing" in protocol.events buffered = transport.get_write_buffer_size() print(f"{name}:") print(f" events = {protocol.events}") print(f" buffered = {buffered}") print(f" verdict = {'STRANDED' if paused and not resumed else 'ok'}") # recovery probe: an application send is the only thing that can re-arm the # write loop, which a protocol honouring pause_writing() would never do if paused and not resumed: transport.sendto(b"probe", addr) await asyncio.sleep(0.5) print(f" after an unsolicited sendto(): {protocol.events}") transport.close() await asyncio.sleep(0.1) async def main() -> None: print(sys.version) print(type(asyncio.get_running_loop()).__name__) print() # covers WSAEMSGSIZE being reported at completion of the in-flight write await scenario("fail-in-flight", OVERSIZED, b"queued") # covers WSAEMSGSIZE being reported synchronously by WSASendTo, from the # completion callback of the preceding write await scenario("fail-from-callback", b"ok", OVERSIZED)