Repository: agronholm/anyio

Issue:
[asyncio] socket.send() raised exception.
I get this very weird print on the console in my application:
"socket.send() raised exception."

The program essentially looks like the following

```
def handle(client):
    async with client:
        async for msg in receive_stream:
            await client.send(msg)

async with await anyio.create_tcp_listener(...) as listener:
    listener.serve(handle)
```

It looks to me as if there is an issue when the client disconnects:

https://github.com/agronholm/anyio/blob/b5dffbe7ccf28afd97d43113ae9e98fca61c6627/src/anyio/_backends/_asyncio.py#L764-L772

I "fixed" it locally by checking the state of the transport and the the stream. I don't know why I don't get this RuntimeError.

```
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index 4b4e6c2..65c0cbe 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -761,6 +761,11 @@ class SocketStream(abc.SocketStream):
     async def send(self, item: bytes) -> None:
         with self._send_guard:
             await checkpoint()
+            if self._closed:
+                raise ClosedResourceError
+            elif self._transport.is_closing():
+                raise BrokenResourceError
+
             try:
                 self._transport.write(item)
             except RuntimeError as exc:
```

I'm using anyio 2.0.2 on Ubuntu 20.04 (Python 3.8.5)

Discussion:

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index 8afd380..103b6fd 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -7,6 +7,8 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
 - Fixed ``Event`` objects on the trio backend not inheriting from ``anyio.abc.Event``
 - Fixed ``run_sync_in_worker_thread()`` raising ``UnboundLocalError`` on asyncio when cancelled
+- Fixed ``send()`` on socket streams not raising any exception on asyncio, and an unwrapped
+  ``BrokenPipeError`` on trio and curio when the peer has disconnected
 
 **2.0.2**
 
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index 3013cb3..6238fcb 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -662,20 +662,27 @@ _write_events: Dict[socket.SocketType, asyncio.Event] = {}
 class StreamProtocol(asyncio.Protocol):
     read_queue: Deque[bytes]
     read_event: asyncio.Event
-    write_event: asyncio.Event
+    write_future: asyncio.Future
     exception: Optional[Exception] = None
 
     def connection_made(self, transport: asyncio.BaseTransport) -> None:
         self.read_queue = deque()
         self.read_event = asyncio.Event()
-        self.write_event = asyncio.Event()
-        self.write_event.set()
+        self.write_future = asyncio.Future()
+        self.write_future.set_result(None)
         cast(asyncio.Transport, transport).set_write_buffer_limits(0)
 
     def connection_lost(self, exc: Optional[Exception]) -> None:
-        self.exception = exc
+        if exc:
+            self.exception = BrokenResourceError()
+            self.exception.__cause__ = exc
+
         self.read_event.set()
-        self.write_event.set()
+        self.write_future = asyncio.Future()
+        if self.exception:
+            self.write_future.set_exception(self.exception)
+        else:
+            self.write_future.set_result(None)
 
     def data_received(self, data: bytes) -> None:
         self.read_queue.append(data)
@@ -686,10 +693,10 @@ class StreamProtocol(asyncio.Protocol):
         return None
 
     def pause_writing(self) -> None:
-        self.write_event.clear()
+        self.write_future = asyncio.Future()
 
     def resume_writing(self) -> None:
-        self.write_event.set()
+        self.write_future.set_result(None)
 
 
 class DatagramProtocol(asyncio.DatagramProtocol):
@@ -750,7 +757,7 @@ class SocketStream(abc.SocketStream):
                 if self._closed:
                     raise ClosedResourceError from None
                 elif self._protocol.exception:
-                    raise BrokenResourceError from self._protocol.exception
+                    raise self._protocol.exception
                 else:
                     raise EndOfStream
 
@@ -774,7 +781,7 @@ class SocketStream(abc.SocketStream):
                 else:
                     raise
 
-            await self._protocol.write_event.wait()
+            await self._protocol.write_future
 
     async def send_eof(self) -> None:
         try:
diff --git a/src/anyio/_backends/_curio.py b/src/anyio/_backends/_curio.py
index 52d2238..6842429 100644
--- a/src/anyio/_backends/_curio.py
+++ b/src/anyio/_backends/_curio.py
@@ -592,13 +592,12 @@ class _CurioSocketMixin(Generic[T_SockAddr]):
                 await wtask.cancel(blocking=False, exc=ClosedResourceError)
 
     def _convert_socket_error(self, exc: Union[OSError, AttributeError]) -> NoReturn:
-        if self._curio_socket.fileno() < 0:
-            if self._closed:
-                raise ClosedResourceError from None
-            else:
-                raise BrokenResourceError from exc
-
-        raise exc
+        if self._curio_socket.fileno() < 0 and self._closed:
+            raise ClosedResourceError from None
+        elif isinstance(exc, OSError):
+            raise BrokenResourceError from exc
+        else:
+            raise exc
 
 
 class SocketStream(_CurioSocketMixin, abc.SocketStream):
diff --git a/src/anyio/_backends/_trio.py b/src/anyio/_backends/_trio.py
index e7ca79d..5eec692 100644
--- a/src/anyio/_backends/_trio.py
+++ b/src/anyio/_backends/_trio.py
@@ -307,13 +307,12 @@ class _TrioSocketMixin(Generic[T_SockAddr]):
     def _convert_socket_error(self, exc: BaseException) -> 'NoReturn':
         if isinstance(exc, trio.ClosedResourceError):
             raise ClosedResourceError from exc
-        elif isinstance(exc, OSError) and self._trio_socket.fileno() < 0:
-            if self._closed:
-                raise ClosedResourceError from None
-            else:
-                raise BrokenResourceError from exc
-
-        raise exc
+        elif self._trio_socket.fileno() < 0 and self._closed:
+            raise ClosedResourceError from None
+        elif isinstance(exc, OSError):
+            raise BrokenResourceError from exc
+        else:
+            raise exc
 
 
 class SocketStream(_TrioSocketMixin, abc.SocketStream):
diff --git a/src/anyio/abc/sockets.py b/src/anyio/abc/sockets.py
index 9bc0c72..32a3160 100644
--- a/src/anyio/abc/sockets.py
+++ b/src/anyio/abc/sockets.py
@@ -102,8 +102,7 @@ class SocketListener(Generic[T_SockAddr], Listener[SocketStream[T_SockAddr]],
             # Can be replaced with AsyncExitStack once on py3.7+
             context_manager = _NullAsyncContextManager()
 
-        # There is a mypy bug here
-        async with context_manager:  # type: ignore[attr-defined]
+        async with context_manager:
             while True:
                 stream = await self.accept()
                 await task_group.spawn(handler, stream)
diff --git a/src/anyio/streams/stapled.py b/src/anyio/streams/stapled.py
index a0927c2..81cf222 100644
--- a/src/anyio/streams/stapled.py
+++ b/src/anyio/streams/stapled.py
@@ -107,8 +107,7 @@ class MultiListener(Generic[T_Stream], Listener[T_Stream]):
                     task_group: Optional[TaskGroup] = None) -> None:
         from .. import create_task_group
 
-        # There is a mypy bug here
-        async with create_task_group() as tg:  # type: ignore[attr-defined]
+        async with create_task_group() as tg:
             for listener in self.listeners:
                 await tg.spawn(listener.serve, handler, task_group)
 


Test patch:
diff --git a/tests/test_sockets.py b/tests/test_sockets.py
index 24e59b7..9ebd7b9 100644
--- a/tests/test_sockets.py
+++ b/tests/test_sockets.py
@@ -9,10 +9,10 @@ from threading import Event, Thread
 import pytest
 
 from anyio import (
-    BusyResourceError, ClosedResourceError, ExceptionGroup, TypedAttributeLookupError, connect_tcp,
-    connect_unix, create_connected_udp_socket, create_event, create_task_group,
-    create_tcp_listener, create_udp_socket, create_unix_listener, getaddrinfo, getnameinfo,
-    move_on_after, wait_all_tasks_blocked)
+    BrokenResourceError, BusyResourceError, ClosedResourceError, ExceptionGroup,
+    TypedAttributeLookupError, connect_tcp, connect_unix, create_connected_udp_socket,
+    create_event, create_task_group, create_tcp_listener, create_udp_socket, create_unix_listener,
+    getaddrinfo, getnameinfo, move_on_after, wait_all_tasks_blocked)
 from anyio.abc.sockets import SocketAttribute
 from anyio.streams.stapled import MultiListener
 
@@ -266,6 +266,27 @@ class TestTCPStream:
         with pytest.raises(ClosedResourceError):
             await stream.send(b'foo')
 
+    async def test_send_after_peer_closed(self, family):
+        def serve_once():
+            client_sock, _ = server_sock.accept()
+            client_sock.close()
+            server_sock.close()
+
+        server_sock = socket.socket(family, socket.SOCK_STREAM)
+        server_sock.settimeout(1)
+        server_sock.bind(('localhost', 0))
+        server_addr = server_sock.getsockname()[:2]
+        server_sock.listen()
+        thread = Thread(target=serve_once, daemon=True)
+        thread.start()
+
+        with pytest.raises(BrokenResourceError):
+            async with await connect_tcp(*server_addr) as stream:
+                for _ in range(1000):
+                    await stream.send(b'foo')
+
+        thread.join()
+
     async def test_connect_tcp_with_tls(self, server_context, client_context, server_sock,
                                         server_addr):
         def serve():