From 1316d351e42ea7b000072f62d926ad4be0914a9d Mon Sep 17 00:00:00 2001 From: CMGS Date: Mon, 5 Oct 2026 12:48:12 +0800 Subject: [PATCH] sdk/python: a parked connection's liveness check works above fd 1024 select.select raises ValueError for an fd at or above FD_SETSIZE, which the quiet check read as a dead peer, so in a process with 1024+ open fds every parked relay and control-plane connection was closed on take and each call dialed afresh. Sockets below FD_SETSIZE keep the select path, since poll costs 8 us per check on macOS against 0.5 us for select; higher fds use poll, and platforms without poll (Windows, whose select has no fd-number cap) keep select. --- sdk/python/cocoonsandbox/client.py | 15 ++------------- sdk/python/cocoonsandbox/conn.py | 21 ++++++++++++++++----- sdk/python/tests/conftest.py | 15 +++++++++++++++ sdk/python/tests/test_client.py | 6 +++--- sdk/python/tests/test_keepalive.py | 7 ++++--- 5 files changed, 40 insertions(+), 24 deletions(-) diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py index 23de7017..6bc282c1 100644 --- a/sdk/python/cocoonsandbox/client.py +++ b/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, diff --git a/sdk/python/cocoonsandbox/conn.py b/sdk/python/cocoonsandbox/conn.py index 0dc4a264..63991fa8 100644 --- a/sdk/python/cocoonsandbox/conn.py +++ b/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") diff --git a/sdk/python/tests/conftest.py b/sdk/python/tests/conftest.py index 3fcc0fd5..a6d41399 100644 --- a/sdk/python/tests/conftest.py +++ b/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 = [] diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index 920c302b..723ae7b8 100644 --- a/sdk/python/tests/test_client.py +++ b/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) diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py index 22b5ce90..b0668f12 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/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: