diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst index fa05a88e7..b30d9c99d 100644 --- a/docs/versionhistory.rst +++ b/docs/versionhistory.rst @@ -23,6 +23,14 @@ 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 ``aclose_forcefully()`` (and, more generally, concurrent calls to + ``SocketStream.aclose()``) returning before the underlying TCP socket was actually + released on asyncio: a checkpoint hit while the cancel scope used by + ``aclose_forcefully()`` was already cancelled could skip the transport's + ``abort()`` call, and a second, concurrent ``aclose()`` call could see the + transport already marked as closing and return immediately without waiting for + the socket to actually be released + (`#1273 `_) - 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 d18eab2bb..47ece3a98 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -1323,6 +1323,7 @@ def __init__(self, transport: asyncio.Transport, protocol: StreamProtocol): self._receive_guard = ResourceGuard("reading from") self._send_guard = ResourceGuard("writing to") self._closed = False + self._aclose_event: asyncio.Event | None = None @property def _raw_socket(self) -> socket.socket: @@ -1393,15 +1394,31 @@ 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 + if self._aclose_event is not None: + # Another task is already closing (or has already closed) the + # transport. Wait for it to actually finish, shielded from our own + # cancellation, so that we never return before the socket is + # really released (see #1273). + with CancelScope(shield=True): + await self._aclose_event.wait() - self._transport.close() - await sleep(0) - self._transport.abort() + return + + self._aclose_event = event = asyncio.Event() + try: + if not self._transport.is_closing(): + try: + self._transport.write_eof() + except OSError: + pass + + try: + self._transport.close() + await sleep(0) + finally: + self._transport.abort() + finally: + event.set() class _RawSocketMixin: diff --git a/tests/test_sockets.py b/tests/test_sockets.py index f47567627..b1213eefb 100644 --- a/tests/test_sockets.py +++ b/tests/test_sockets.py @@ -43,6 +43,7 @@ TCPConnectable, TypedAttributeLookupError, UNIXConnectable, + aclose_forcefully, as_connectable, connect_tcp, connect_unix, @@ -507,6 +508,23 @@ async def interrupt() -> None: with pytest.raises(ClosedResourceError): await stream.receive() + async def test_concurrent_aclose_forcefully_returns_before_socket_closes( + self, server_addr: tuple[str, int] + ) -> None: + stream = await connect_tcp(*server_addr) + raw_socket = stream.extra(SocketAttribute.raw_socket) + fds: list[int] = [] + + async def do_aclose() -> None: + await aclose_forcefully(stream) + fds.append(raw_socket.fileno()) + + async with create_task_group() as tg: + tg.start_soon(do_aclose) + tg.start_soon(do_aclose) + + assert fds == [-1, -1] + async def test_receive_after_close(self, server_addr: tuple[str, int]) -> None: stream = await connect_tcp(*server_addr) await stream.aclose()