Fixed socket aclose() delayed close, missing checkpoint and Windows hang

docs/versionhistory.rst

@@ -81,6 +81,15 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 - Fixed ``CapacityLimiter`` on the asyncio backend over-granting tokens when
   ``total_tokens`` was raised while the limiter was over-subscribed
   (`#1223 <https://github.com/agronholm/anyio/pull/1223>`_; PR by @zelinewang)
+- 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 @graingert)
+- 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 @graingert)
+- Fixed ``aclose()`` on sockets not checkpointing, and thus failing to raise a
+  cancellation exception when called inside a cancelled scope (affected UNIX socket
+  streams and UNIX datagram sockets on asyncio, and all sockets on Trio)
+  (`#1288 <https://github.com/agronholm/anyio/issues/1288>`_; PR by @graingert)
 - Fixed asyncio task groups leaking unawaited coroutines when a custom task constructor
   fails; default task creation is unaffected
   (`#1274 <https://github.com/agronholm/anyio/issues/1274>`_; PR by @dsfaccini)

src/anyio/_backends/_asyncio.py

@@ -1301,20 +1301,17 @@ class DatagramProtocol(asyncio.DatagramProtocol):
     read_queue: deque[tuple[bytes, IPSockAddrType]]
     read_event: asyncio.Event
     write_event: asyncio.Event
-    closed_event: asyncio.Event
     exception: Exception | None = None
 
     def connection_made(self, transport: asyncio.BaseTransport) -> None:
         self.read_queue = deque(maxlen=100)  # arbitrary value
         self.read_event = asyncio.Event()
         self.write_event = asyncio.Event()
-        self.closed_event = asyncio.Event()
         self.write_event.set()
 
     def connection_lost(self, exc: Exception | None) -> None:
         self.read_event.set()
         self.write_event.set()
-        self.closed_event.set()
 
     def datagram_received(self, data: bytes, addr: IPSockAddrType) -> None:
         addr = convert_ipv6_sockaddr(addr)
@@ -1408,15 +1405,8 @@ 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
-
-            self._transport.close()
-            await sleep(0)
-            self._transport.abort()
+        self._transport.abort()
+        await AsyncIOBackend.checkpoint()
 
 
 class _RawSocketMixin:
@@ -1454,16 +1444,17 @@ def callback(f: object) -> None:
         return f
 
     async def aclose(self) -> None:
-        if not self._closing:
-            self._closing = True
-            if self.__raw_socket.fileno() != -1:
-                self.__raw_socket.close()
-
+        closing = self._closing
+        self._closing = True
+        self.__raw_socket.close()
+        if not closing:
             if self._receive_future and not self._receive_future.done():
                 self._receive_future.set_result(None)
             if self._send_future and not self._send_future.done():
                 self._send_future.set_result(None)
 
+        await AsyncIOBackend.checkpoint()
+
 
 class UNIXSocketStream(_RawSocketMixin, abc.UNIXSocketStream):
     async def send_eof(self) -> None:
@@ -1675,6 +1666,7 @@ async def accept(self) -> abc.SocketStream:
     async def aclose(self) -> None:
         self._closed = True
         self.__raw_socket.close()
+        await AsyncIOBackend.checkpoint()
 
     @property
     def _raw_socket(self) -> socket.socket:
@@ -1697,10 +1689,8 @@ def _raw_socket(self) -> socket.socket:
 
     async def aclose(self) -> None:
         self._closed = True
-        if not self._transport.is_closing():
-            self._transport.close()
-
-        await self._protocol.closed_event.wait()
+        self._transport.abort()
+        await AsyncIOBackend.checkpoint()
 
     async def receive(self) -> tuple[bytes, IPSockAddrType]:
         with self._receive_guard:
@@ -1747,10 +1737,8 @@ def _raw_socket(self) -> socket.socket:
 
     async def aclose(self) -> None:
         self._closed = True
-        if not self._transport.is_closing():
-            self._transport.close()
-
-        await self._protocol.closed_event.wait()
+        self._transport.abort()
+        await AsyncIOBackend.checkpoint()
 
     async def receive(self) -> bytes:
         with self._receive_guard:

src/anyio/_backends/_trio.py

@@ -463,9 +463,9 @@ def _raw_socket(self) -> socket.socket:
         return self._trio_socket._sock  # type: ignore[attr-defined]
 
     async def aclose(self) -> None:
-        if self._trio_socket.fileno() >= 0:
-            self._closed = True
-            self._trio_socket.close()
+        self._closed = True
+        self._trio_socket.close()
+        await trio.lowlevel.checkpoint()
 
     def _convert_socket_error(self, exc: BaseException) -> NoReturn:
         if isinstance(exc, trio.ClosedResourceError):