Tweak dropped connection detection (#185)

* Tweak sync dropped connection detection

* Nits

* Use common helper across sync, asyncio, curio, anyio

* Update

* Add test case

* Rename is_connection_dropped -> is_socket_readable, tweak comments

* Switch to stealing util from trio

Co-authored-by: Tom Christie <[email protected]>

Author: florimondmanca

httpcore/_async/connection.py

@@ -165,8 +165,8 @@ def state(self) -> ConnectionState:
             return ConnectionState.PENDING
         return self.connection.get_state()
 
-    def is_connection_dropped(self) -> bool:
-        return self.connection is not None and self.connection.is_connection_dropped()
+    def is_socket_readable(self) -> bool:
+        return self.connection is not None and self.connection.is_socket_readable()
 
     def mark_as_ready(self) -> None:
         if self.connection is not None:

httpcore/_async/connection_pool.py

@@ -245,7 +245,14 @@ async def _get_connection_from_pool(
                 seen_http11 = True
 
             if connection.state == ConnectionState.IDLE:
-                if connection.is_connection_dropped():
+                if connection.is_socket_readable():
+                    # If the socket is readable while the connection is idle (meaning
+                    # we don't expect the server to send any data), then the only valid
+                    # reason is that the other end has disconnected, which means we
+                    # should drop the connection too.
+                    # (For a detailed run-through of what a "readable" socket is, and
+                    # why this is the best thing for us to do here, see:
+                    # https://github.com/encode/httpx/pull/143#issuecomment-515181778)
                     logger.trace("removing dropped idle connection=%r", connection)
                     # IDLE connections that have been dropped should be
                     # removed from the pool.

httpcore/_async/http.py

@@ -20,9 +20,9 @@ def mark_as_ready(self) -> None:
         """
         raise NotImplementedError()  # pragma: nocover
 
-    def is_connection_dropped(self) -> bool:
+    def is_socket_readable(self) -> bool:
         """
-        Return 'True' if the connection has been dropped by the remote end.
+        Return 'True' if the underlying network socket is readable.
         """
         raise NotImplementedError()  # pragma: nocover
 

httpcore/_async/http11.py

@@ -201,5 +201,5 @@ async def aclose(self) -> None:
 
             await self.socket.aclose()
 
-    def is_connection_dropped(self) -> bool:
-        return self.socket.is_connection_dropped()
+    def is_socket_readable(self) -> bool:
+        return self.socket.is_readable()

httpcore/_async/http2.py

@@ -158,8 +158,8 @@ async def send_connection_init(self, timeout: TimeoutDict) -> None:
     def is_closed(self) -> bool:
         return False
 
-    def is_connection_dropped(self) -> bool:
-        return self.socket.is_connection_dropped()
+    def is_socket_readable(self) -> bool:
+        return self.socket.is_readable()
 
     async def aclose(self) -> None:
         logger.trace("close_connection=%r", self)

httpcore/_backends/anyio.py

@@ -1,4 +1,3 @@
-import select
 from ssl import SSLContext
 from typing import Optional
 
@@ -17,6 +16,7 @@
     WriteTimeout,
 )
 from .._types import TimeoutDict
+from .._utils import is_socket_readable
 from .base import AsyncBackend, AsyncLock, AsyncSemaphore, AsyncSocketStream
 
 
@@ -85,10 +85,9 @@ async def aclose(self) -> None:
             except BrokenResourceError as exc:
                 raise CloseError from exc
 
-    def is_connection_dropped(self) -> bool:
-        raw_socket = self.stream.extra(SocketAttribute.raw_socket)
-        rready, _wready, _xready = select.select([raw_socket], [], [], 0)
-        return bool(rready)
+    def is_readable(self) -> bool:
+        sock = self.stream.extra(SocketAttribute.raw_socket)
+        return is_socket_readable(sock.fileno())
 
 
 class Lock(AsyncLock):

httpcore/_backends/asyncio.py

@@ -1,4 +1,5 @@
 import asyncio
+import socket
 from ssl import SSLContext
 from typing import Optional
 
@@ -13,6 +14,7 @@
     map_exceptions,
 )
 from .._types import TimeoutDict
+from .._utils import is_socket_readable
 from .base import AsyncBackend, AsyncLock, AsyncSemaphore, AsyncSocketStream
 
 SSL_MONKEY_PATCH_APPLIED = False
@@ -171,21 +173,10 @@ async def aclose(self) -> None:
             with map_exceptions({OSError: CloseError}):
                 self.stream_writer.close()
 
-    def is_connection_dropped(self) -> bool:
-        # Counter-intuitively, what we really want to know here is whether the socket is
-        # *readable*, i.e. whether it would return immediately with empty bytes if we
-        # called `.recv()` on it, indicating that the other end has closed the socket.
-        # See: https://github.com/encode/httpx/pull/143#issuecomment-515181778
-        #
-        # As it turns out, asyncio checks for readability in the background
-        # (see: https://github.com/encode/httpx/pull/276#discussion_r322000402),
-        # so checking for EOF or readability here would yield the same result.
-        #
-        # At the cost of rigour, we check for EOF instead of readability because asyncio
-        # does not expose any public API to check for readability.
-        # (For a solution that uses private asyncio APIs, see:
-        # https://github.com/encode/httpx/pull/143#issuecomment-515202982)
-        return self.stream_reader.at_eof()
+    def is_readable(self) -> bool:
+        transport = self.stream_reader._transport  # type: ignore
+        sock: socket.socket = transport.get_extra_info("socket")
+        return is_socket_readable(sock.fileno())
 
 
 class Lock(AsyncLock):

httpcore/_backends/base.py

@@ -63,7 +63,7 @@ async def write(self, data: bytes, timeout: TimeoutDict) -> None:
     async def aclose(self) -> None:
         raise NotImplementedError()  # pragma: no cover
 
-    def is_connection_dropped(self) -> bool:
+    def is_readable(self) -> bool:
         raise NotImplementedError()  # pragma: no cover
 
 

httpcore/_backends/curio.py

@@ -1,4 +1,3 @@
-import select
 from ssl import SSLContext, SSLSocket
 from typing import Optional
 
@@ -15,7 +14,7 @@
     map_exceptions,
 )
 from .._types import TimeoutDict
-from .._utils import get_logger
+from .._utils import get_logger, is_socket_readable
 from .base import AsyncBackend, AsyncLock, AsyncSemaphore, AsyncSocketStream
 
 logger = get_logger(__name__)
@@ -132,10 +131,8 @@ async def aclose(self) -> None:
         await self.stream.close()
         await self.socket.close()
 
-    def is_connection_dropped(self) -> bool:
-        rready, _, _ = select.select([self.socket.fileno()], [], [], 0)
-
-        return bool(rready)
+    def is_readable(self) -> bool:
+        return is_socket_readable(self.socket.fileno())
 
 
 class CurioBackend(AsyncBackend):

httpcore/_backends/sync.py

@@ -1,4 +1,3 @@
-import select
 import socket
 import threading
 import time
@@ -17,6 +16,7 @@
     map_exceptions,
 )
 from .._types import TimeoutDict
+from .._utils import is_socket_readable
 
 
 class SyncSocketStream:
@@ -77,9 +77,8 @@ def close(self) -> None:
             with map_exceptions({socket.error: CloseError}):
                 self.sock.close()
 
-    def is_connection_dropped(self) -> bool:
-        rready, _wready, _xready = select.select([self.sock], [], [], 0)
-        return bool(rready)
+    def is_readable(self) -> bool:
+        return is_socket_readable(self.sock.fileno())
 
 
 class SyncLock: