CI probe: concurrent aclose_forcefully() on UDP sockets and Process

docs/versionhistory.rst

@@ -23,6 +23,8 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
   module name. (The default name for a task spawned with ``TaskGroup.start_soon`` or
   ``TaskGroup.start`` typically includes the module name.)
   (`#1234 <https://github.com/agronholm/anyio/pull/1234>`_; PR by @gschaffner)
+- Fixed UDP socket closing hanging on Windows if a datagram send was still in flight
+  (`#1237 <https://github.com/agronholm/anyio/issues/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 <http://semver.org/>`_.
   which triggers ``PytestRemovedIn10Warning`` on ``pytest>=9.2`` and crashes pytest at
   startup when ``filterwarnings = error`` is configured
   (`#1271 <https://github.com/agronholm/anyio/issues/1271>`_; PR by @matthewfeickert)
+- Fixed concurrent ``aclose_forcefully()`` calls on asyncio socket streams returning
+  before the underlying socket was closed
+  (`#1273 <https://github.com/agronholm/anyio/issues/1273>`_; PR by @hansu650)
 
 **4.14.2**
 

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()
 

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

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