From f460010e217c22a8d39fd71ee6cc8efd0393a177 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Wed, 26 Aug 2026 08:49:24 +0100 Subject: [PATCH 1/8] call set_write_buffer_limits(0) for Datagram sockets --- docs/versionhistory.rst | 5 ++ src/anyio/_backends/_asyncio.py | 1 + tests/test_sockets.py | 153 ++++++++++++++++++++++++++++++++ 3 files changed, 159 insertions(+) diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst index 3acc15fb3..7830d0cfa 100644 --- a/docs/versionhistory.rst +++ b/docs/versionhistory.rst @@ -46,6 +46,11 @@ This library adheres to `Semantic Versioning 2.0 `_. building a lookup table from the ``if TYPE_CHECKING:`` block. A fallback mode has been provided for installations where the source code is unavailable (e.g. PyInstaller). (`#1169 `_) +- Changed the asyncio backend to set the write buffer high water mark to 0 on UDP + sockets (both connected and unconnected), so that ``send()`` waits until the + datagram has actually been passed to the operating system instead of letting the + transport buffer it + (`#1294 `_; PR by @graingert) - Fixed free-threading compatibility issues arising from the fact that on Python 3.14 free-threading builds, newly created threads inherit the current context by default, causing AnyIO to behave erroneously in relation to ``start_blocking_portal()`` and diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index 35b0cf371..6b91c32ba 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -1299,6 +1299,7 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None: self.write_event = asyncio.Event() self.closed_event = asyncio.Event() self.write_event.set() + cast(asyncio.WriteTransport, transport).set_write_buffer_limits(0) def connection_lost(self, exc: Exception | None) -> None: self.read_event.set() diff --git a/tests/test_sockets.py b/tests/test_sockets.py index cc661b79a..7e6be380d 100644 --- a/tests/test_sockets.py +++ b/tests/test_sockets.py @@ -117,6 +117,10 @@ not sys.platform.startswith("linux"), reason="Abstract namespace sockets is a Linux only feature", ) +skip_no_dgram_backpressure_mark = pytest.mark.skipif( + not sys.platform.startswith("linux"), + reason="Datagram sockets are only known to refuse datagrams with EAGAIN on Linux", +) @pytest.fixture @@ -1761,6 +1765,72 @@ async def handle(stream: SocketStream) -> None: assert client_addresses == expected_addresses +DGRAM_BACKPRESSURE_PAYLOAD = b"\x00" * 64 + + +def drain_datagram_socket(sock: socket.socket) -> None: + """Receive and discard every datagram currently queued on ``sock``.""" + try: + while True: + sock.recv(65536) + except BlockingIOError: + pass + + +def make_backpressured_pair( + tmp_path: Path, *, connect: bool +) -> tuple[socket.socket, socket.socket, str, int]: + """ + Create a datagram socket that the operating system will refuse datagrams from, + along with the peer it sends to. + + UNIX datagram sockets are used here because UDP sockets never exert any + back-pressure on the loopback interface: the kernel accounts for the datagram + only until it has been delivered or dropped, so ``sendto()`` always succeeds, no + matter how small the send buffer is or how full the peer's receive buffer is. + UNIX datagram sockets, on the other hand, refuse datagrams with ``EAGAIN`` once + the send buffer is full, and they go through the very same code path in both + backends. + + The send buffer is deliberately made as small as the OS allows, so that only a + handful of datagrams are needed to fill it up. The buffer is left empty on + return. + + :return: the socket, its peer, the peer's path, and the number of datagrams the + OS accepts before it starts refusing them + + """ + peer_path = str(tmp_path / "peer.sock") + peer = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) + peer.setblocking(False) + peer.bind(peer_path) + sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) + sock.bind(str(tmp_path / "local.sock")) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024) + if connect: + sock.connect(peer_path) + + sock.setblocking(False) + + # Find out how many datagrams the OS accepts before it refuses any more, and then + # empty both buffers again + capacity = 0 + try: + while True: + if connect: + sock.send(DGRAM_BACKPRESSURE_PAYLOAD) + else: + sock.sendto(DGRAM_BACKPRESSURE_PAYLOAD, peer_path) + + capacity += 1 + except BlockingIOError: + pass + + assert capacity >= 2, f"the send buffer only fits {capacity} datagram(s)" + drain_datagram_socket(peer) + return sock, peer, peer_path, capacity + + @pytest.mark.network @pytest.mark.usefixtures("check_asyncio_bug") class TestUDPSocket: @@ -1832,6 +1902,48 @@ async def test_send_receive(self, family: AnyIPAddressFamily) -> None: assert response == b"halb" assert addr == (host, port) + @skip_no_dgram_backpressure_mark + async def test_send_waits_for_the_os_to_accept_the_datagram( + self, tmp_path: Path + ) -> None: + """ + ``send()`` must not return before the OS has accepted the datagram. + + The asyncio backend sets the transport's write buffer high water mark to 0, + so the transport pauses the sender as soon as it has had to buffer a datagram + the OS refused, instead of quietly buffering up to 64 KiB worth of them. The + trio backend never buffers datagrams to begin with. + """ + sock, peer, peer_path, capacity = make_backpressured_pair( + tmp_path, connect=False + ) + addr = cast(IPSockAddrType, peer_path) + with peer: + async with await UDPSocket.from_socket(sock) as udp: + # Fill up the send buffer again + for _ in range(capacity): + await udp.send((DGRAM_BACKPRESSURE_PAYLOAD, addr)) + + send_completed = False + + async def send_two_more() -> None: + nonlocal send_completed + await udp.send((DGRAM_BACKPRESSURE_PAYLOAD, addr)) + await udp.send((DGRAM_BACKPRESSURE_PAYLOAD, addr)) + send_completed = True + + with fail_after(5): + async with create_task_group() as tg: + tg.start_soon(send_two_more) + await wait_all_tasks_blocked() + blocked = not send_completed + + # Make room in the send buffer so the sender can continue + drain_datagram_socket(peer) + + assert blocked, "send() returned before the OS accepted the datagram" + assert send_completed + async def test_iterate(self, family: AnyIPAddressFamily) -> None: async def serve() -> None: async for packet, addr in server: @@ -2011,6 +2123,47 @@ async def test_send_receive(self, family: AnyIPAddressFamily) -> None: response = await udp2.receive() assert response == b"halb" + @skip_no_dgram_backpressure_mark + async def test_send_waits_for_the_os_to_accept_the_datagram( + self, tmp_path: Path + ) -> None: + """ + ``send()`` must not return before the OS has accepted the datagram. + + The asyncio backend sets the transport's write buffer high water mark to 0, + so the transport pauses the sender as soon as it has had to buffer a datagram + the OS refused, instead of quietly buffering up to 64 KiB worth of them. The + trio backend never buffers datagrams to begin with. + """ + sock, peer, _peer_path, capacity = make_backpressured_pair( + tmp_path, connect=True + ) + with peer: + async with await ConnectedUDPSocket.from_socket(sock) as udp: + # Fill up the send buffer again + for _ in range(capacity): + await udp.send(DGRAM_BACKPRESSURE_PAYLOAD) + + send_completed = False + + async def send_two_more() -> None: + nonlocal send_completed + await udp.send(DGRAM_BACKPRESSURE_PAYLOAD) + await udp.send(DGRAM_BACKPRESSURE_PAYLOAD) + send_completed = True + + with fail_after(5): + async with create_task_group() as tg: + tg.start_soon(send_two_more) + await wait_all_tasks_blocked() + blocked = not send_completed + + # Make room in the send buffer so the sender can continue + drain_datagram_socket(peer) + + assert blocked, "send() returned before the OS accepted the datagram" + assert send_completed + async def test_iterate(self, family: AnyIPAddressFamily) -> None: async def serve() -> None: async for packet in udp2: From 1ccfe0ae3dfbad2bd24d79180e2f540b523c29d2 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Thu, 27 Aug 2026 08:14:59 +0100 Subject: [PATCH 2/8] wait for UDP socket to drain after sending --- docs/versionhistory.rst | 7 +++++-- src/anyio/_backends/_asyncio.py | 20 ++++++++++++++------ tests/test_sockets.py | 10 ++++------ 3 files changed, 23 insertions(+), 14 deletions(-) diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst index 7830d0cfa..1da666019 100644 --- a/docs/versionhistory.rst +++ b/docs/versionhistory.rst @@ -47,9 +47,12 @@ This library adheres to `Semantic Versioning 2.0 `_. provided for installations where the source code is unavailable (e.g. PyInstaller). (`#1169 `_) - Changed the asyncio backend to set the write buffer high water mark to 0 on UDP - sockets (both connected and unconnected), so that ``send()`` waits until the + sockets (both connected and unconnected), and to wait on the write event *after* + handing the datagram to the transport, so that each ``send()`` waits until its own datagram has actually been passed to the operating system instead of letting the - transport buffer it + transport buffer it. Note that on asyncio, a ``send()`` cancelled while waiting may + still result in the datagram being delivered, as the transport has already accepted + it; on trio, a cancelled ``send()`` never sends (`#1294 `_; PR by @graingert) - Fixed free-threading compatibility issues arising from the fact that on Python 3.14 free-threading builds, newly created threads inherit the current context by default, diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index 6b91c32ba..2adb7535f 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -1712,13 +1712,17 @@ async def receive(self) -> tuple[bytes, IPSockAddrType]: async def send(self, item: UDPPacketType) -> None: with self._send_guard: await AsyncIOBackend.checkpoint() - await self._protocol.write_event.wait() if self._closed: raise ClosedResourceError elif self._transport.is_closing(): raise BrokenResourceError - else: - self._transport.sendto(*item) + + self._transport.sendto(*item) + + # If the OS refused the datagram, the transport has buffered it and + # (because the high water mark is 0) already cleared the write event, so + # this waits until this call's own datagram has been handed to the OS + await self._protocol.write_event.wait() class ConnectedUDPSocket(abc.ConnectedUDPSocket): @@ -1764,13 +1768,17 @@ async def receive(self) -> bytes: async def send(self, item: bytes) -> None: with self._send_guard: await AsyncIOBackend.checkpoint() - await self._protocol.write_event.wait() if self._closed: raise ClosedResourceError elif self._transport.is_closing(): raise BrokenResourceError - else: - self._transport.sendto(item) + + self._transport.sendto(item) + + # If the OS refused the datagram, the transport has buffered it and + # (because the high water mark is 0) already cleared the write event, so + # this waits until this call's own datagram has been handed to the OS + await self._protocol.write_event.wait() class UNIXDatagramSocket(_RawSocketMixin, abc.UNIXDatagramSocket): diff --git a/tests/test_sockets.py b/tests/test_sockets.py index 7e6be380d..b165017e1 100644 --- a/tests/test_sockets.py +++ b/tests/test_sockets.py @@ -1926,15 +1926,14 @@ async def test_send_waits_for_the_os_to_accept_the_datagram( send_completed = False - async def send_two_more() -> None: + async def send_one_more() -> None: nonlocal send_completed await udp.send((DGRAM_BACKPRESSURE_PAYLOAD, addr)) - await udp.send((DGRAM_BACKPRESSURE_PAYLOAD, addr)) send_completed = True with fail_after(5): async with create_task_group() as tg: - tg.start_soon(send_two_more) + tg.start_soon(send_one_more) await wait_all_tasks_blocked() blocked = not send_completed @@ -2146,15 +2145,14 @@ async def test_send_waits_for_the_os_to_accept_the_datagram( send_completed = False - async def send_two_more() -> None: + async def send_one_more() -> None: nonlocal send_completed await udp.send(DGRAM_BACKPRESSURE_PAYLOAD) - await udp.send(DGRAM_BACKPRESSURE_PAYLOAD) send_completed = True with fail_after(5): async with create_task_group() as tg: - tg.start_soon(send_two_more) + tg.start_soon(send_one_more) await wait_all_tasks_blocked() blocked = not send_completed From ae7b7d7de02253b0c96a8f207da281e4b352aa9a Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sat, 29 Aug 2026 18:42:33 +0100 Subject: [PATCH 3/8] keep the pre-send write_event wait on UDP sockets A send() cancelled while blocked on the post-send wait leaves its datagram in the asyncio transport's buffer with the protocol still paused. Without a wait before sendto(), the next send() hands its datagram straight to the transport, which appends it to that non-empty buffer without ever trying the OS, so repeated cancellation grows the buffer despite the zero high water mark. Co-Authored-By: Claude Opus 5 --- docs/versionhistory.rst | 12 ++-- src/anyio/_backends/_asyncio.py | 12 ++++ tests/test_sockets.py | 98 +++++++++++++++++++++++++++++++++ 3 files changed, 116 insertions(+), 6 deletions(-) diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst index 1da666019..6e36f8c13 100644 --- a/docs/versionhistory.rst +++ b/docs/versionhistory.rst @@ -47,12 +47,12 @@ This library adheres to `Semantic Versioning 2.0 `_. provided for installations where the source code is unavailable (e.g. PyInstaller). (`#1169 `_) - Changed the asyncio backend to set the write buffer high water mark to 0 on UDP - sockets (both connected and unconnected), and to wait on the write event *after* - handing the datagram to the transport, so that each ``send()`` waits until its own - datagram has actually been passed to the operating system instead of letting the - transport buffer it. Note that on asyncio, a ``send()`` cancelled while waiting may - still result in the datagram being delivered, as the transport has already accepted - it; on trio, a cancelled ``send()`` never sends + sockets (both connected and unconnected), and to wait on the write event both before + and after handing the datagram to the transport, so that each ``send()`` waits until + its own datagram has actually been passed to the operating system instead of letting + the transport buffer it. Note that on asyncio, a ``send()`` cancelled while waiting + may still result in the datagram being delivered, as the transport has already + accepted it; on trio, a cancelled ``send()`` never sends (`#1294 `_; PR by @graingert) - Fixed free-threading compatibility issues arising from the fact that on Python 3.14 free-threading builds, newly created threads inherit the current context by default, diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index 2adb7535f..b09a1e8b4 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -1712,6 +1712,12 @@ async def receive(self) -> tuple[bytes, IPSockAddrType]: async def send(self, item: UDPPacketType) -> None: with self._send_guard: await AsyncIOBackend.checkpoint() + + # Wait until the transport has flushed any datagram the OS previously + # refused, so that this datagram is not merely appended to the transport's + # buffer (which would happen if a previous send() was cancelled while + # waiting below) + await self._protocol.write_event.wait() if self._closed: raise ClosedResourceError elif self._transport.is_closing(): @@ -1768,6 +1774,12 @@ async def receive(self) -> bytes: async def send(self, item: bytes) -> None: with self._send_guard: await AsyncIOBackend.checkpoint() + + # Wait until the transport has flushed any datagram the OS previously + # refused, so that this datagram is not merely appended to the transport's + # buffer (which would happen if a previous send() was cancelled while + # waiting below) + await self._protocol.write_event.wait() if self._closed: raise ClosedResourceError elif self._transport.is_closing(): diff --git a/tests/test_sockets.py b/tests/test_sockets.py index b165017e1..dc68734bf 100644 --- a/tests/test_sockets.py +++ b/tests/test_sockets.py @@ -1777,6 +1777,23 @@ def drain_datagram_socket(sock: socket.socket) -> None: pass +async def receive_datagrams_until(sock: socket.socket, sentinel: bytes) -> list[bytes]: + """ + Receive datagrams from ``sock`` until ``sentinel`` has been received. + + :return: every datagram received, in the order they arrived, ``sentinel`` last + + """ + received: list[bytes] = [] + while not received or received[-1] != sentinel: + try: + received.append(sock.recv(65536)) + except BlockingIOError: + await wait_readable(sock) + + return received + + def make_backpressured_pair( tmp_path: Path, *, connect: bool ) -> tuple[socket.socket, socket.socket, str, int]: @@ -1943,6 +1960,47 @@ async def send_one_more() -> None: assert blocked, "send() returned before the OS accepted the datagram" assert send_completed + @skip_no_dgram_backpressure_mark + async def test_cancelled_send_does_not_buffer_the_next_datagram( + self, tmp_path: Path + ) -> None: + """ + A ``send()`` cancelled while blocked must not let the next one skip ahead. + + On asyncio, the datagram of the *first* cancelled send has unavoidably been + buffered by the transport already, but any subsequent ``send()`` must wait for + that buffer to be flushed instead of appending to it — otherwise repeated + cancellation would grow the transport's buffer without bound, despite the zero + high water mark. + """ + sock, peer, peer_path, capacity = make_backpressured_pair( + tmp_path, connect=False + ) + addr = cast(IPSockAddrType, peer_path) + with peer: + async with await UDPSocket.from_socket(sock) as udp: + # Fill up the send buffer again + for _ in range(capacity): + await udp.send((DGRAM_BACKPRESSURE_PAYLOAD, addr)) + + async def send_and_cancel(payload: bytes) -> None: + async with create_task_group() as tg: + tg.start_soon(udp.send, (payload, addr)) + await wait_all_tasks_blocked() + tg.cancel_scope.cancel() + + with fail_after(5): + # The first cancelled send may leave its datagram in the asyncio + # transport's buffer, but the second one must never reach it + await send_and_cancel(b"one") + await send_and_cancel(b"two") + + async with create_task_group() as tg: + tg.start_soon(udp.send, (b"three", addr)) + received = await receive_datagrams_until(peer, b"three") + + assert b"two" not in received + async def test_iterate(self, family: AnyIPAddressFamily) -> None: async def serve() -> None: async for packet, addr in server: @@ -2162,6 +2220,46 @@ async def send_one_more() -> None: assert blocked, "send() returned before the OS accepted the datagram" assert send_completed + @skip_no_dgram_backpressure_mark + async def test_cancelled_send_does_not_buffer_the_next_datagram( + self, tmp_path: Path + ) -> None: + """ + A ``send()`` cancelled while blocked must not let the next one skip ahead. + + On asyncio, the datagram of the *first* cancelled send has unavoidably been + buffered by the transport already, but any subsequent ``send()`` must wait for + that buffer to be flushed instead of appending to it — otherwise repeated + cancellation would grow the transport's buffer without bound, despite the zero + high water mark. + """ + sock, peer, _peer_path, capacity = make_backpressured_pair( + tmp_path, connect=True + ) + with peer: + async with await ConnectedUDPSocket.from_socket(sock) as udp: + # Fill up the send buffer again + for _ in range(capacity): + await udp.send(DGRAM_BACKPRESSURE_PAYLOAD) + + async def send_and_cancel(payload: bytes) -> None: + async with create_task_group() as tg: + tg.start_soon(udp.send, payload) + await wait_all_tasks_blocked() + tg.cancel_scope.cancel() + + with fail_after(5): + # The first cancelled send may leave its datagram in the asyncio + # transport's buffer, but the second one must never reach it + await send_and_cancel(b"one") + await send_and_cancel(b"two") + + async with create_task_group() as tg: + tg.start_soon(udp.send, b"three") + received = await receive_datagrams_until(peer, b"three") + + assert b"two" not in received + async def test_iterate(self, family: AnyIPAddressFamily) -> None: async def serve() -> None: async for packet in udp2: From 6b1ab4208ad259fc01c2552e341f9d440b66f74c Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 30 Aug 2026 15:29:21 +0100 Subject: [PATCH 4/8] trim the added comments and docstrings down to the essentials Co-Authored-By: Claude Opus 5 --- src/anyio/_backends/_asyncio.py | 20 ++++------- tests/test_sockets.py | 62 ++++----------------------------- 2 files changed, 12 insertions(+), 70 deletions(-) diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index b09a1e8b4..be26ef143 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -1713,10 +1713,7 @@ async def send(self, item: UDPPacketType) -> None: with self._send_guard: await AsyncIOBackend.checkpoint() - # Wait until the transport has flushed any datagram the OS previously - # refused, so that this datagram is not merely appended to the transport's - # buffer (which would happen if a previous send() was cancelled while - # waiting below) + # Wait for any datagram the transport had to buffer to be flushed first await self._protocol.write_event.wait() if self._closed: raise ClosedResourceError @@ -1725,9 +1722,8 @@ async def send(self, item: UDPPacketType) -> None: self._transport.sendto(*item) - # If the OS refused the datagram, the transport has buffered it and - # (because the high water mark is 0) already cleared the write event, so - # this waits until this call's own datagram has been handed to the OS + # The high water mark is 0, so the write event has been cleared if the OS + # refused the datagram; wait until it has been handed over await self._protocol.write_event.wait() @@ -1775,10 +1771,7 @@ async def send(self, item: bytes) -> None: with self._send_guard: await AsyncIOBackend.checkpoint() - # Wait until the transport has flushed any datagram the OS previously - # refused, so that this datagram is not merely appended to the transport's - # buffer (which would happen if a previous send() was cancelled while - # waiting below) + # Wait for any datagram the transport had to buffer to be flushed first await self._protocol.write_event.wait() if self._closed: raise ClosedResourceError @@ -1787,9 +1780,8 @@ async def send(self, item: bytes) -> None: self._transport.sendto(item) - # If the OS refused the datagram, the transport has buffered it and - # (because the high water mark is 0) already cleared the write event, so - # this waits until this call's own datagram has been handed to the OS + # The high water mark is 0, so the write event has been cleared if the OS + # refused the datagram; wait until it has been handed over await self._protocol.write_event.wait() diff --git a/tests/test_sockets.py b/tests/test_sockets.py index dc68734bf..421b3cfea 100644 --- a/tests/test_sockets.py +++ b/tests/test_sockets.py @@ -1769,7 +1769,6 @@ async def handle(stream: SocketStream) -> None: def drain_datagram_socket(sock: socket.socket) -> None: - """Receive and discard every datagram currently queued on ``sock``.""" try: while True: sock.recv(65536) @@ -1778,12 +1777,7 @@ def drain_datagram_socket(sock: socket.socket) -> None: async def receive_datagrams_until(sock: socket.socket, sentinel: bytes) -> list[bytes]: - """ - Receive datagrams from ``sock`` until ``sentinel`` has been received. - - :return: every datagram received, in the order they arrived, ``sentinel`` last - - """ + """Receive datagrams until ``sentinel`` arrives, and return all of them.""" received: list[bytes] = [] while not received or received[-1] != sentinel: try: @@ -1798,23 +1792,13 @@ def make_backpressured_pair( tmp_path: Path, *, connect: bool ) -> tuple[socket.socket, socket.socket, str, int]: """ - Create a datagram socket that the operating system will refuse datagrams from, - along with the peer it sends to. - - UNIX datagram sockets are used here because UDP sockets never exert any - back-pressure on the loopback interface: the kernel accounts for the datagram - only until it has been delivered or dropped, so ``sendto()`` always succeeds, no - matter how small the send buffer is or how full the peer's receive buffer is. - UNIX datagram sockets, on the other hand, refuse datagrams with ``EAGAIN`` once - the send buffer is full, and they go through the very same code path in both - backends. + Create a datagram socket with a full send buffer, along with the peer it sends to. - The send buffer is deliberately made as small as the OS allows, so that only a - handful of datagrams are needed to fill it up. The buffer is left empty on - return. + UNIX datagram sockets are used because UDP sockets never exert back-pressure on the + loopback interface, but they take the same code path in both backends. - :return: the socket, its peer, the peer's path, and the number of datagrams the - OS accepts before it starts refusing them + :return: the socket, its peer, the peer's path, and the number of datagrams the OS + accepts before it starts refusing them """ peer_path = str(tmp_path / "peer.sock") @@ -1923,14 +1907,6 @@ async def test_send_receive(self, family: AnyIPAddressFamily) -> None: async def test_send_waits_for_the_os_to_accept_the_datagram( self, tmp_path: Path ) -> None: - """ - ``send()`` must not return before the OS has accepted the datagram. - - The asyncio backend sets the transport's write buffer high water mark to 0, - so the transport pauses the sender as soon as it has had to buffer a datagram - the OS refused, instead of quietly buffering up to 64 KiB worth of them. The - trio backend never buffers datagrams to begin with. - """ sock, peer, peer_path, capacity = make_backpressured_pair( tmp_path, connect=False ) @@ -1964,15 +1940,6 @@ async def send_one_more() -> None: async def test_cancelled_send_does_not_buffer_the_next_datagram( self, tmp_path: Path ) -> None: - """ - A ``send()`` cancelled while blocked must not let the next one skip ahead. - - On asyncio, the datagram of the *first* cancelled send has unavoidably been - buffered by the transport already, but any subsequent ``send()`` must wait for - that buffer to be flushed instead of appending to it — otherwise repeated - cancellation would grow the transport's buffer without bound, despite the zero - high water mark. - """ sock, peer, peer_path, capacity = make_backpressured_pair( tmp_path, connect=False ) @@ -2184,14 +2151,6 @@ async def test_send_receive(self, family: AnyIPAddressFamily) -> None: async def test_send_waits_for_the_os_to_accept_the_datagram( self, tmp_path: Path ) -> None: - """ - ``send()`` must not return before the OS has accepted the datagram. - - The asyncio backend sets the transport's write buffer high water mark to 0, - so the transport pauses the sender as soon as it has had to buffer a datagram - the OS refused, instead of quietly buffering up to 64 KiB worth of them. The - trio backend never buffers datagrams to begin with. - """ sock, peer, _peer_path, capacity = make_backpressured_pair( tmp_path, connect=True ) @@ -2224,15 +2183,6 @@ async def send_one_more() -> None: async def test_cancelled_send_does_not_buffer_the_next_datagram( self, tmp_path: Path ) -> None: - """ - A ``send()`` cancelled while blocked must not let the next one skip ahead. - - On asyncio, the datagram of the *first* cancelled send has unavoidably been - buffered by the transport already, but any subsequent ``send()`` must wait for - that buffer to be flushed instead of appending to it — otherwise repeated - cancellation would grow the transport's buffer without bound, despite the zero - high water mark. - """ sock, peer, _peer_path, capacity = make_backpressured_pair( tmp_path, connect=True ) From ba33c637cc3ace7b984f6f14a77703cdd7d7c93b Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 30 Aug 2026 16:10:37 +0100 Subject: [PATCH 5/8] trim the changelog entry down Co-Authored-By: Claude Opus 5 --- docs/versionhistory.rst | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst index 6e36f8c13..49229e263 100644 --- a/docs/versionhistory.rst +++ b/docs/versionhistory.rst @@ -46,13 +46,8 @@ This library adheres to `Semantic Versioning 2.0 `_. building a lookup table from the ``if TYPE_CHECKING:`` block. A fallback mode has been provided for installations where the source code is unavailable (e.g. PyInstaller). (`#1169 `_) -- Changed the asyncio backend to set the write buffer high water mark to 0 on UDP - sockets (both connected and unconnected), and to wait on the write event both before - and after handing the datagram to the transport, so that each ``send()`` waits until - its own datagram has actually been passed to the operating system instead of letting - the transport buffer it. Note that on asyncio, a ``send()`` cancelled while waiting - may still result in the datagram being delivered, as the transport has already - accepted it; on trio, a cancelled ``send()`` never sends +- Changed UDP sockets on the asyncio backend to make ``send()`` wait until the + datagram has been passed to the operating system (`#1294 `_; PR by @graingert) - Fixed free-threading compatibility issues arising from the fact that on Python 3.14 free-threading builds, newly created threads inherit the current context by default, From 9dff9202c163c17eaee86941b706e83ce760ba7d Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 30 Aug 2026 16:18:57 +0100 Subject: [PATCH 6/8] reduce number of checkpoints in UDP sockets --- src/anyio/_backends/_asyncio.py | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index be26ef143..7edc0f8cb 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -1711,10 +1711,12 @@ async def receive(self) -> tuple[bytes, IPSockAddrType]: async def send(self, item: UDPPacketType) -> None: with self._send_guard: - await AsyncIOBackend.checkpoint() - # Wait for any datagram the transport had to buffer to be flushed first - await self._protocol.write_event.wait() + if self._protocol.write_event.is_set(): + await AsyncIOBackend.checkpoint() + else: + await self._protocol.write_event.wait() + if self._closed: raise ClosedResourceError elif self._transport.is_closing(): @@ -1769,10 +1771,12 @@ async def receive(self) -> bytes: async def send(self, item: bytes) -> None: with self._send_guard: - await AsyncIOBackend.checkpoint() - # Wait for any datagram the transport had to buffer to be flushed first - await self._protocol.write_event.wait() + if self._protocol.write_event.is_set(): + await AsyncIOBackend.checkpoint() + else: + await self._protocol.write_event.wait() + if self._closed: raise ClosedResourceError elif self._transport.is_closing(): From 482c60e961014f8fcb440e3be0dd647a74f81463 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 30 Aug 2026 16:32:29 +0100 Subject: [PATCH 7/8] use checkpoint_if_cancelled() and yield after the send instead Co-Authored-By: Claude Opus 5 --- src/anyio/_backends/_asyncio.py | 64 +++++++++++++++++++++------------ 1 file changed, 42 insertions(+), 22 deletions(-) diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index 7edc0f8cb..723f52b31 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -1711,22 +1711,32 @@ async def receive(self) -> tuple[bytes, IPSockAddrType]: async def send(self, item: UDPPacketType) -> None: with self._send_guard: + await AsyncIOBackend.checkpoint_if_cancelled() + yielded = False + # Wait for any datagram the transport had to buffer to be flushed first - if self._protocol.write_event.is_set(): - await AsyncIOBackend.checkpoint() - else: + if not self._protocol.write_event.is_set(): + yielded = True await self._protocol.write_event.wait() - if self._closed: - raise ClosedResourceError - elif self._transport.is_closing(): - raise BrokenResourceError + try: + if self._closed: + raise ClosedResourceError + elif self._transport.is_closing(): + raise BrokenResourceError - self._transport.sendto(*item) + self._transport.sendto(*item) - # The high water mark is 0, so the write event has been cleared if the OS - # refused the datagram; wait until it has been handed over - await self._protocol.write_event.wait() + # The high water mark is 0, so the write event has been cleared if the + # OS refused the datagram; wait until it has been handed over + if not self._protocol.write_event.is_set(): + yielded = True + await self._protocol.write_event.wait() + finally: + # Nothing blocked, so make the mandatory yield here instead, where it + # can no longer lose an already sent datagram to cancellation + if not yielded: + await AsyncIOBackend.cancel_shielded_checkpoint() class ConnectedUDPSocket(abc.ConnectedUDPSocket): @@ -1771,22 +1781,32 @@ async def receive(self) -> bytes: async def send(self, item: bytes) -> None: with self._send_guard: + await AsyncIOBackend.checkpoint_if_cancelled() + yielded = False + # Wait for any datagram the transport had to buffer to be flushed first - if self._protocol.write_event.is_set(): - await AsyncIOBackend.checkpoint() - else: + if not self._protocol.write_event.is_set(): + yielded = True await self._protocol.write_event.wait() - if self._closed: - raise ClosedResourceError - elif self._transport.is_closing(): - raise BrokenResourceError + try: + if self._closed: + raise ClosedResourceError + elif self._transport.is_closing(): + raise BrokenResourceError - self._transport.sendto(item) + self._transport.sendto(item) - # The high water mark is 0, so the write event has been cleared if the OS - # refused the datagram; wait until it has been handed over - await self._protocol.write_event.wait() + # The high water mark is 0, so the write event has been cleared if the + # OS refused the datagram; wait until it has been handed over + if not self._protocol.write_event.is_set(): + yielded = True + await self._protocol.write_event.wait() + finally: + # Nothing blocked, so make the mandatory yield here instead, where it + # can no longer lose an already sent datagram to cancellation + if not yielded: + await AsyncIOBackend.cancel_shielded_checkpoint() class UNIXDatagramSocket(_RawSocketMixin, abc.UNIXDatagramSocket): From 316c50ccbc08262f972c95edce8a53f6a439c878 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 30 Aug 2026 17:47:58 +0100 Subject: [PATCH 8/8] don't checkpoint when send() raises without sending Unlike Trio, which runs a cancel_shielded_checkpoint() on the way out of a failed socket call, the closed and broken paths now only get the checkpoint_if_cancelled() at the top. Co-Authored-By: Claude Opus 5 --- src/anyio/_backends/_asyncio.py | 56 +++++++++++++-------------------- 1 file changed, 22 insertions(+), 34 deletions(-) diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index 723f52b31..cb3c57459 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -1714,29 +1714,23 @@ async def send(self, item: UDPPacketType) -> None: await AsyncIOBackend.checkpoint_if_cancelled() yielded = False - # Wait for any datagram the transport had to buffer to be flushed first + # Wait out any datagram the transport had to buffer if not self._protocol.write_event.is_set(): yielded = True await self._protocol.write_event.wait() - try: - if self._closed: - raise ClosedResourceError - elif self._transport.is_closing(): - raise BrokenResourceError + if self._closed: + raise ClosedResourceError + elif self._transport.is_closing(): + raise BrokenResourceError - self._transport.sendto(*item) + self._transport.sendto(*item) - # The high water mark is 0, so the write event has been cleared if the - # OS refused the datagram; wait until it has been handed over - if not self._protocol.write_event.is_set(): - yielded = True - await self._protocol.write_event.wait() - finally: - # Nothing blocked, so make the mandatory yield here instead, where it - # can no longer lose an already sent datagram to cancellation - if not yielded: - await AsyncIOBackend.cancel_shielded_checkpoint() + # The high water mark is 0, so the event is clear if the OS refused it + if not self._protocol.write_event.is_set(): + await self._protocol.write_event.wait() + elif not yielded: + await AsyncIOBackend.cancel_shielded_checkpoint() class ConnectedUDPSocket(abc.ConnectedUDPSocket): @@ -1784,29 +1778,23 @@ async def send(self, item: bytes) -> None: await AsyncIOBackend.checkpoint_if_cancelled() yielded = False - # Wait for any datagram the transport had to buffer to be flushed first + # Wait out any datagram the transport had to buffer if not self._protocol.write_event.is_set(): yielded = True await self._protocol.write_event.wait() - try: - if self._closed: - raise ClosedResourceError - elif self._transport.is_closing(): - raise BrokenResourceError + if self._closed: + raise ClosedResourceError + elif self._transport.is_closing(): + raise BrokenResourceError - self._transport.sendto(item) + self._transport.sendto(item) - # The high water mark is 0, so the write event has been cleared if the - # OS refused the datagram; wait until it has been handed over - if not self._protocol.write_event.is_set(): - yielded = True - await self._protocol.write_event.wait() - finally: - # Nothing blocked, so make the mandatory yield here instead, where it - # can no longer lose an already sent datagram to cancellation - if not yielded: - await AsyncIOBackend.cancel_shielded_checkpoint() + # The high water mark is 0, so the event is clear if the OS refused it + if not self._protocol.write_event.is_set(): + await self._protocol.write_event.wait() + elif not yielded: + await AsyncIOBackend.cancel_shielded_checkpoint() class UNIXDatagramSocket(_RawSocketMixin, abc.UNIXDatagramSocket):