diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index fa05a88e7..8e77288cb 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -23,6 +23,8 @@ This library adheres to `Semantic Versioning 2.0 `_.
module name. (The default name for a task spawned with ``TaskGroup.start_soon`` or
``TaskGroup.start`` typically includes the module name.)
(`#1234 `_; PR by @gschaffner)
+- Fixed UDP socket closing hanging on Windows if a datagram send was still in flight
+ (`#1237 `_; PR by @Guflly)
- 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
@@ -52,6 +54,9 @@ This library adheres to `Semantic Versioning 2.0 `_.
which triggers ``PytestRemovedIn10Warning`` on ``pytest>=9.2`` and crashes pytest at
startup when ``filterwarnings = error`` is configured
(`#1271 `_; PR by @matthewfeickert)
+- Fixed concurrent ``aclose_forcefully()`` calls on asyncio socket streams returning
+ before the underlying socket was closed
+ (`#1273 `_; PR by @hansu650)
**4.14.2**
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index d18eab2bb..a178e5cb0 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -1393,15 +1393,18 @@ async def send_eof(self) -> None:
async def aclose(self) -> None:
self._closed = True
- if not self._transport.is_closing():
- try:
- self._transport.write_eof()
- except OSError:
- pass
+ try:
+ if not self._transport.is_closing():
+ try:
+ self._transport.write_eof()
+ except OSError:
+ pass
- self._transport.close()
- await sleep(0)
+ self._transport.close()
+ await sleep(0)
+ finally:
self._transport.abort()
+ await AsyncIOBackend.cancel_shielded_checkpoint()
class _RawSocketMixin:
@@ -1684,6 +1687,11 @@ async def aclose(self) -> None:
self._closed = True
if not self._transport.is_closing():
self._transport.close()
+ try:
+ await sleep(0)
+ finally:
+ self._transport.abort()
+ await AsyncIOBackend.cancel_shielded_checkpoint()
await self._protocol.closed_event.wait()
@@ -1734,6 +1742,11 @@ async def aclose(self) -> None:
self._closed = True
if not self._transport.is_closing():
self._transport.close()
+ try:
+ await sleep(0)
+ finally:
+ self._transport.abort()
+ await AsyncIOBackend.cancel_shielded_checkpoint()
await self._protocol.closed_event.wait()
diff --git a/tests/test_sockets.py b/tests/test_sockets.py
index f47567627..7a15ca37a 100644
--- a/tests/test_sockets.py
+++ b/tests/test_sockets.py
@@ -34,6 +34,7 @@
from pytest import FixtureRequest
from pytest_mock.plugin import MockerFixture
+import anyio
from anyio import (
BrokenResourceError,
BusyResourceError,
@@ -43,6 +44,7 @@
TCPConnectable,
TypedAttributeLookupError,
UNIXConnectable,
+ aclose_forcefully,
as_connectable,
connect_tcp,
connect_unix,
@@ -507,6 +509,23 @@ async def interrupt() -> None:
with pytest.raises(ClosedResourceError):
await stream.receive()
+ async def test_concurrent_aclose_forcefully_waits_for_socket_close(
+ self, server_addr: tuple[str, int]
+ ) -> None:
+ stream = await connect_tcp(*server_addr)
+ raw_socket = stream.extra(SocketAttribute.raw_socket)
+ file_descriptors: list[int] = []
+
+ async def close_stream() -> None:
+ await aclose_forcefully(stream)
+ file_descriptors.append(raw_socket.fileno())
+
+ async with create_task_group() as task_group:
+ task_group.start_soon(close_stream)
+ task_group.start_soon(close_stream)
+
+ assert file_descriptors == [-1, -1]
+
async def test_receive_after_close(self, server_addr: tuple[str, int]) -> None:
stream = await connect_tcp(*server_addr)
await stream.aclose()
@@ -519,6 +538,14 @@ async def test_send_after_close(self, server_addr: tuple[str, int]) -> None:
with pytest.raises(ClosedResourceError):
await stream.send(b"foo")
+ @pytest.mark.parametrize("anyio_backend", asyncio_params)
+ async def test_aclose_forcefully(self, server_addr: tuple[str, int]) -> None:
+ stream = await connect_tcp(*server_addr)
+ sock = stream.extra(SocketAttribute.raw_socket)
+ await stream.send(b"x")
+ await anyio.aclose_forcefully(stream)
+ assert sock.fileno() == -1
+
async def test_receive_after_peer_closed(
self, family: AnyIPAddressFamily, request: FixtureRequest
) -> None:
@@ -1779,6 +1806,17 @@ async def test_aclose_waits_for_fd_release(
udp = await UDPSocket.from_socket(sock)
await udp.aclose()
+ @pytest.mark.skipif(sys.platform != "win32", reason="Windows only")
+ @pytest.mark.parametrize("anyio_backend", asyncio_params)
+ async def test_aclose_during_send(self) -> None:
+ udp = await create_udp_socket(local_host="127.0.0.1")
+ sock = udp.extra(SocketAttribute.raw_socket)
+ await udp.sendto(b"x", "127.0.0.1", 9999)
+ with fail_after(1):
+ await anyio.aclose_forcefully(udp)
+
+ assert sock.fileno() == -1
+
async def test_extra_attributes(self, family: AnyIPAddressFamily) -> None:
async with await create_udp_socket(
family=family, local_host="localhost"
@@ -1884,6 +1922,86 @@ async def close_when_blocked() -> None:
with pytest.raises(ClosedResourceError):
await udp.receive()
+ async def test_concurrent_aclose_forcefully_waits_for_socket_close(self) -> None:
+ udp = await create_udp_socket(
+ family=AddressFamily.AF_INET, local_host="localhost"
+ )
+ raw_socket = udp.extra(SocketAttribute.raw_socket)
+ file_descriptors: list[int] = []
+
+ async def close_socket() -> None:
+ await aclose_forcefully(udp)
+ file_descriptors.append(raw_socket.fileno())
+
+ async with create_task_group() as task_group:
+ task_group.start_soon(close_socket)
+ task_group.start_soon(close_socket)
+
+ assert file_descriptors == [-1, -1]
+
+ async def test_aclose_forcefully_with_pending_send(self) -> None:
+ """Force-close with an overlapped send still in flight.
+
+ On Windows' proactor loop ``sendto()`` always buffers and starts an
+ overlapped operation, so ``transport.close()`` declines to schedule
+ ``connection_lost`` while ``_write_fut`` is pending
+ (``proactor_events.py:108``), and ``_loop_writing()`` then bails out at
+ ``if self._conn_lost: return`` — which ``close()`` just incremented — so
+ ``connection_lost`` is never called at all and the wait on
+ ``closed_event`` blocks forever.
+
+ There must be no checkpoint between the send and the close: anyio puts
+ ``send()``'s checkpoint *before* ``transport.sendto()``, so the overlapped
+ operation is only still in flight if nothing yields afterwards. A single
+ event loop iteration is enough for the proactor to reap it.
+ """
+ udp = await create_udp_socket(
+ family=AddressFamily.AF_INET, local_host="localhost"
+ )
+ host, port = cast(tuple[str, int], udp.extra(SocketAttribute.local_address))
+ raw_socket = udp.extra(SocketAttribute.raw_socket)
+
+ await udp.sendto(b"x", host, port)
+ with fail_after(5):
+ await aclose_forcefully(udp)
+
+ assert raw_socket.fileno() == -1
+
+ async def test_concurrent_aclose_forcefully_with_pending_send(self) -> None:
+ """A second force-close arriving while the first is still in progress.
+
+ ``aclose()`` waits on ``closed_event`` unshielded, and the second caller
+ skips the ``is_closing()`` block entirely, so it never aborts the
+ transport — it only waits. This checks that it still doesn't return
+ before the FD has been released.
+ """
+ udp = await create_udp_socket(
+ family=AddressFamily.AF_INET, local_host="localhost"
+ )
+ host, port = cast(tuple[str, int], udp.extra(SocketAttribute.local_address))
+ raw_socket = udp.extra(SocketAttribute.raw_socket)
+ first_close_started = Event()
+ file_descriptors: list[int] = []
+
+ async def close_socket() -> None:
+ await first_close_started.wait()
+ await aclose_forcefully(udp)
+ file_descriptors.append(raw_socket.fileno())
+
+ with fail_after(5):
+ async with create_task_group() as task_group:
+ task_group.start_soon(close_socket)
+ await wait_all_tasks_blocked()
+
+ # No checkpoint from here on, so the send is still in flight and
+ # the second closer only runs once this one yields inside aclose().
+ await udp.sendto(b"x", host, port)
+ first_close_started.set()
+ await aclose_forcefully(udp)
+ file_descriptors.append(raw_socket.fileno())
+
+ assert file_descriptors == [-1, -1]
+
async def test_receive_after_close(self) -> None:
udp = await create_udp_socket(
family=AddressFamily.AF_INET, local_host="localhost"
@@ -1956,6 +2074,14 @@ async def test_aclose_waits_for_fd_release(
finally:
peer.close()
+ @pytest.mark.skipif(sys.platform != "win32", reason="Windows only")
+ @pytest.mark.parametrize("anyio_backend", asyncio_params)
+ async def test_aclose_during_send(self) -> None:
+ udp = await create_connected_udp_socket("127.0.0.1", 9999)
+ await udp.send(b"x")
+ with fail_after(1):
+ await udp.aclose()
+
async def test_extra_attributes(self, family: AnyIPAddressFamily) -> None:
async with await create_connected_udp_socket(
"localhost", 5000, family=family
@@ -2067,6 +2193,51 @@ async def close_when_blocked() -> None:
with pytest.raises(ClosedResourceError):
await udp.receive()
+ async def test_concurrent_aclose_forcefully_waits_for_socket_close(self) -> None:
+ udp = await create_connected_udp_socket(
+ "localhost", 5000, local_host="localhost", family=AddressFamily.AF_INET
+ )
+ raw_socket = udp.extra(SocketAttribute.raw_socket)
+ file_descriptors: list[int] = []
+
+ async def close_socket() -> None:
+ await aclose_forcefully(udp)
+ file_descriptors.append(raw_socket.fileno())
+
+ async with create_task_group() as task_group:
+ task_group.start_soon(close_socket)
+ task_group.start_soon(close_socket)
+
+ assert file_descriptors == [-1, -1]
+
+ async def test_aclose_forcefully_with_pending_send(self) -> None:
+ """Force-close with an overlapped send still in flight.
+
+ See ``TestUDPSocket.test_aclose_forcefully_with_pending_send`` for why an
+ in-flight send matters and why nothing may yield between the send and the
+ close. A real peer is bound here so the datagram doesn't provoke an ICMP
+ port unreachable, which would break the connected socket instead.
+ """
+ peer = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ peer.bind(("127.0.0.1", 0))
+ try:
+ peer_host, peer_port = peer.getsockname()
+ udp = await create_connected_udp_socket(
+ peer_host,
+ peer_port,
+ local_host="localhost",
+ family=AddressFamily.AF_INET,
+ )
+ raw_socket = udp.extra(SocketAttribute.raw_socket)
+
+ await udp.send(b"x")
+ with fail_after(5):
+ await aclose_forcefully(udp)
+
+ assert raw_socket.fileno() == -1
+ finally:
+ peer.close()
+
async def test_receive_after_close(self, family: AnyIPAddressFamily) -> None:
udp = await create_connected_udp_socket(
"localhost", 5000, local_host="localhost", family=family
diff --git a/tests/test_subprocesses.py b/tests/test_subprocesses.py
index a87f8b450..8dd41f9e9 100644
--- a/tests/test_subprocesses.py
+++ b/tests/test_subprocesses.py
@@ -2,6 +2,7 @@
import os
import platform
+import signal
import sys
from collections.abc import Callable
from pathlib import Path
@@ -17,6 +18,7 @@
CancelScope,
ClosedResourceError,
EndOfStream,
+ aclose_forcefully,
create_task_group,
fail_after,
open_process,
@@ -440,6 +442,22 @@ async def test_close_while_reading() -> None:
process.terminate()
+async def test_concurrent_aclose_forcefully_waits_for_process_exit() -> None:
+ process = await open_process([sys.executable, "-c", "import time; time.sleep(3)"])
+ expected_returncode = 1 if sys.platform == "win32" else -signal.SIGKILL
+ returncodes: list[int | None] = []
+
+ async def close_process() -> None:
+ await aclose_forcefully(process)
+ returncodes.append(process.returncode)
+
+ async with create_task_group() as task_group:
+ task_group.start_soon(close_process)
+ task_group.start_soon(close_process)
+
+ assert returncodes == [expected_returncode, expected_returncode]
+
+
async def test_wait_returns_on_process_exit_with_open_stdout() -> None:
"""
wait() should return once the process exits, rather than waiting for the piped