sdk/python: a parked connection's liveness check works above fd 1024

sdk/python/cocoonsandbox/client.py

@@ -5,7 +5,6 @@
 import http.client
 import json
 import queue
-import select
 import ssl
 import threading
 import time
@@ -14,7 +13,7 @@
 from typing import Any, TypeVar, cast
 
 from .checkpoint import Checkpoint
-from .conn import Conn, dial_agent, remaining_timeout
+from .conn import Conn, dial_agent, is_quiet, remaining_timeout
 from .endpoint import _endpoint_url
 from .errors import APIError, SandboxTimeout, require_field
 from .sandbox import Sandbox
@@ -259,7 +258,7 @@ def _take(self, key: tuple[str, str], timeout: float) -> http.client.HTTPConnect
             idle = self._idle.get(key, [])
             while idle:
                 conn, parked = idle.pop()
-                if time.monotonic() - parked < self.keep_alive and _quiet(conn.sock):
+                if time.monotonic() - parked < self.keep_alive and is_quiet(conn.sock):
                     conn.sock.settimeout(timeout)
                     return conn
                 conn.close()
@@ -275,16 +274,6 @@ def _tls(self) -> ssl.SSLContext:
         return self._ssl_context
 
 
-def _quiet(sock: Any) -> bool:
-    if sock is None:
-        return False
-    try:
-        readable, _, _ = select.select([sock], [], [], 0)
-    except (OSError, ValueError):
-        return False
-    return not readable
-
-
 def _claim_body(
     template: str,
     net: str,

sdk/python/cocoonsandbox/conn.py

@@ -16,6 +16,7 @@
 from .frames import MAX_FRAME, decode_response, encode_request
 
 KEEP_ALIVE_CONNS = 8
+_FD_SETSIZE = 1024
 
 _CloseableT = TypeVar("_CloseableT", bound="_Closeable")
 
@@ -81,11 +82,7 @@ def recv_until(self, *terminal: str) -> Iterator[dict[str, Any]]:
 
     def quiet(self) -> bool:
         """Reports whether the peer has neither hung up nor spoken since the last frame."""
-        try:
-            readable, _, _ = select.select([self._sock], [], [], 0)
-        except (OSError, ValueError):
-            return False
-        return not readable
+        return is_quiet(self._sock)
 
     def close(self) -> None:
         self.abort()
@@ -237,6 +234,20 @@ def remaining_timeout(timeout: float, deadline: float | None, what: str = "agent
     return min(timeout, remaining)
 
 
+def is_quiet(sock: socket.socket | None) -> bool:
+    """Reports whether a parked socket's peer has neither hung up nor spoken, for any fd number."""
+    if sock is None:
+        return False
+    try:
+        if sock.fileno() < _FD_SETSIZE or not hasattr(select, "poll"):
+            return not select.select([sock], [], [], 0)[0]
+        poller = select.poll()
+        poller.register(sock, select.POLLIN)
+        return not poller.poll(0)
+    except (OSError, ValueError):
+        return False
+
+
 def _handshake_error(what: str, exc: OSError) -> SandboxError:
     if isinstance(exc, (TimeoutError, socket.timeout)):
         return SandboxTimeout(f"{what} timed out")

sdk/python/tests/conftest.py

@@ -2,6 +2,8 @@
 that refuse a connection, and the relay-side upgrade handshake."""
 
 import json
+import os
+import resource
 import socket
 import threading
 import time
@@ -70,6 +72,19 @@ def wait_until(cond: Callable[[], bool], message: str) -> None:
     raise AssertionError(message)
 
 
+@pytest.fixture(params=[0, 1024], ids=["low_fd", "high_fd"])
+def fd_floor(request):
+    soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
+    resource.setrlimit(resource.RLIMIT_NOFILE, (max(soft, min(4096, hard)), hard))
+    fds = [os.open(os.devnull, os.O_RDONLY)]
+    while fds[-1] < request.param:
+        fds.append(os.open(os.devnull, os.O_RDONLY))
+    yield request.param
+    for fd in fds:
+        os.close(fd)
+    resource.setrlimit(resource.RLIMIT_NOFILE, (soft, hard))
+
+
 @pytest.fixture
 def spawn_node():
     servers = []

sdk/python/tests/test_client.py

@@ -343,15 +343,15 @@ def claim(body, path):
 
 
 @pytest.mark.parametrize(("keep_alive", "conns"), [(30.0, 1), (0, 2)])
-def test_control_calls_reuse_one_connection(kept_node, keep_alive, conns):
+def test_control_calls_reuse_one_connection(kept_node, keep_alive, conns, fd_floor):
     KeptNode.routes[("POST", "/v1/claim")] = lambda body, path: (200, {"id": "sb_1", "token": "tok"})
     client = Client(kept_node, keep_alive=keep_alive)
     client.new("rt:24.04")
     client.new("rt:24.04")
-    assert KeptNode.conns == conns
+    assert KeptNode.conns == conns, KeptNode.conns
 
 
-def test_a_parked_connection_the_node_closed_is_not_reused(kept_node):
+def test_a_parked_connection_the_node_closed_is_not_reused(kept_node, fd_floor):
     KeptNode.routes[("POST", "/v1/claim")] = lambda body, path: (200, {"id": "sb_1", "token": "tok"})
     KeptNode.drop_after_reply = True
     client = Client(kept_node)

sdk/python/tests/test_keepalive.py

@@ -16,17 +16,18 @@
 INPUT_OPS = ("stdin", "stdin_close", "data", "data_end")
 
 
-def test_calls_share_one_connection():
+def test_calls_share_one_connection(fd_floor):
     agent = FakeAgent()
     sb = sandbox_at(agent.addr)
     try:
         for _ in range(3):
             assert sb.stat("/")["kind"] == "dir"
         assert sb.exec("echo", "42") == "42\n"
+        assert sb._pool._parked[0][1]._sock.fileno() >= fd_floor
     finally:
         sb._pool.drain()
         agent.stop()
-    assert agent.upgrades == 1
+    assert agent.upgrades == 1, agent.upgrades
 
 
 def test_a_refused_stream_open_keeps_its_connection():
@@ -75,7 +76,7 @@ def test_idle_connection_closes():
         agent.stop()
 
 
-def test_peer_hang_up_is_noticed_before_reuse():
+def test_peer_hang_up_is_noticed_before_reuse(fd_floor):
     agent = FakeAgent(hang_up_after=2)
     sb = sandbox_at(agent.addr)
     try: