diff --git a/jsocket/_version.py b/jsocket/_version.py index a8e4cfd..135156a 100644 --- a/jsocket/_version.py +++ b/jsocket/_version.py @@ -1,3 +1,3 @@ """Package version.""" -__version__ = "2.0.1" +__version__ = "2.0.2" diff --git a/jsocket/jsocket_base.py b/jsocket/jsocket_base.py index 85e2c03..4131dcc 100644 --- a/jsocket/jsocket_base.py +++ b/jsocket/jsocket_base.py @@ -71,6 +71,7 @@ def __init__( self.max_message_size = max_message_size self._last_client_addr = None self._is_server = False + self._is_listening = False if create_socket: self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.conn = self.socket @@ -93,7 +94,15 @@ def _send(self, msg): """Send all bytes in `msg` to the peer.""" sent = 0 while sent < len(msg): - sent += self.conn.send(msg[sent:]) + try: + chunk = self.conn.send(msg[sent:]) + except OSError as e: + self._close_connection() + raise RuntimeError("socket connection broken") from e + if chunk == 0: + self._close_connection() + raise RuntimeError("socket connection broken") + sent += chunk def _read(self, size, allow_timeout=False): """Read exactly `size` bytes from the peer or raise on disconnect.""" @@ -168,6 +177,7 @@ def _close_socket(self): pass except OSError: pass + self._is_listening = False def _close_connection(self): """Best-effort shutdown and close of the connection socket.""" @@ -301,9 +311,13 @@ def _close_connection(self): def _bind(self): self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.socket.bind((self.address, self.port)) + self._is_listening = False def _listen(self): + if self._is_listening: + return self.socket.listen(5) + self._is_listening = True def _accept(self): return self.socket.accept() @@ -341,17 +355,30 @@ def __init__(self, address='127.0.0.1', port=5489, timeout=2.0, recv_timeout=Non def connect(self): """Attempt to connect to the server up to 10 times with backoff.""" + def _recreate_socket(): + self._close_socket() + self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.socket.settimeout(self._recv_timeout) + self.conn = self.socket + for attempt in range(1, 11): + sock = getattr(self, "socket", None) + needs_fresh_socket = sock is None + if not needs_fresh_socket: + try: + needs_fresh_socket = sock.fileno() == -1 + except OSError: + needs_fresh_socket = True + if needs_fresh_socket: + _recreate_socket() + logger.debug("created fresh socket before connect attempt %d to %s:%s", attempt, self.address, self.port) try: logger.debug("connect attempt %d to %s:%s", attempt, self.address, self.port) self.socket.connect((self.address, self.port)) except socket.error as msg: logger.error("SockThread Error: %s", msg) # Recreate the socket to avoid retrying on a potentially bad fd. - self._close_socket() - self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - self.socket.settimeout(self._recv_timeout) - self.conn = self.socket + _recreate_socket() logger.debug("recreated socket for retry %d to %s:%s", attempt, self.address, self.port) time.sleep(3) continue diff --git a/jsocket/tserver.py b/jsocket/tserver.py index c2a753c..326fbb2 100644 --- a/jsocket/tserver.py +++ b/jsocket/tserver.py @@ -76,6 +76,9 @@ def _init_wakeup(self): except (AttributeError, OSError): pass # Fallback: create a loopback TCP pair + listener = None + writer = None + reader = None try: listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) @@ -84,12 +87,24 @@ def _init_wakeup(self): writer = socket.socket(socket.AF_INET, socket.SOCK_STREAM) writer.connect(listener.getsockname()) reader, _ = listener.accept() - listener.close() reader.setblocking(False) writer.setblocking(False) self._wakeup_r, self._wakeup_w = reader, writer except OSError: + for sock in (reader, writer): + if sock is None: + continue + try: + sock.close() + except OSError: + pass self._wakeup_r, self._wakeup_w = None, None + finally: + if listener is not None: + try: + listener.close() + except OSError: + pass def _signal_wakeup(self): """Wake the accept loop if it is blocked.""" @@ -398,7 +413,6 @@ def run(self): if not self._is_alive: self._is_alive = True while self._is_alive: - tmp = self._thread_type(**self._thread_args) self._purge_threads() while not self.connected and self._is_alive: if not self._wait_for_accept(): @@ -409,7 +423,10 @@ def run(self): logger.debug("factory accept timeout on %s:%s: %s", self.address, self.port, e) continue except Exception as e: # pylint: disable=broad-exception-caught - logger.exception("accept error: %s", e) + if self._is_alive: + logger.exception("factory accept error on %s:%s: %s", self.address, self.port, e) + else: + logger.debug("factory stopping; accept loop exiting (%s:%s)", self.address, self.port) continue else: # Hand off the accepted connection to the worker @@ -427,10 +444,42 @@ def run(self): except OSError: pass break - tmp.swap_socket(accepted_conn) - tmp.start() + try: + tmp = self._thread_type(**self._thread_args) + tmp.swap_socket(accepted_conn) + tmp.start() + except Exception as e: # pylint: disable=broad-exception-caught + # Worker construction/hand-off failed; ensure the accepted connection is closed + try: + accepted_conn.shutdown(socket.SHUT_RDWR) + except OSError: + pass + try: + accepted_conn.close() + except OSError: + pass + if self._is_alive: + logger.exception( + "factory worker handoff error on %s:%s: %s", + self.address, + self.port, + e, + ) + else: + logger.debug( + "factory stopping; worker handoff aborted (%s:%s)", + self.address, + self.port, + ) + # Continue accepting new connections + continue with self._threads_lock: self._threads.append(tmp) + try: + addr = accepted_conn.getpeername() + except OSError: + addr = None + logger.debug("factory spawned worker %s for %s", tmp.name, _format_client_id(addr)) break self._wait_to_exit() diff --git a/tests/test_additional_coverage.py b/tests/test_additional_coverage.py index e523799..e21ed8e 100644 --- a/tests/test_additional_coverage.py +++ b/tests/test_additional_coverage.py @@ -29,6 +29,67 @@ def fail_connect(self, addr): # pylint: disable=unused-argument assert client.connect() is False +def test_client_connect_recreates_closed_socket_without_backoff(monkeypatch): + """connect() should refresh a closed fd before dialing, avoiding backoff sleep.""" + + events = [] + + class ClosedSocket: + def fileno(self): + return -1 + + def settimeout(self, timeout): + events.append(("closed_settimeout", timeout)) + + def connect(self, addr): # pragma: no cover - should never be called + events.append(("closed_connect", addr)) + raise AssertionError("connect() attempted on closed socket") + + def shutdown(self, how): # pylint: disable=unused-argument + return None + + def close(self): + return None + + class FreshSocket: + def fileno(self): + return 42 + + def settimeout(self, timeout): + events.append(("fresh_settimeout", timeout)) + + def connect(self, addr): + events.append(("fresh_connect", addr)) + return None + + def shutdown(self, how): # pylint: disable=unused-argument + return None + + def close(self): + return None + + def fake_socket(*args, **kwargs): # pylint: disable=unused-argument + return FreshSocket() + + def no_sleep(*args, **kwargs): # pylint: disable=unused-argument + raise AssertionError("unexpected connect backoff sleep") + + monkeypatch.setattr(jsocket.jsocket_base.socket, "socket", fake_socket, raising=True) + monkeypatch.setattr(jsocket.jsocket_base.time, "sleep", no_sleep, raising=True) + + client = jsocket.jsocket_base.JsonClient.__new__(jsocket.jsocket_base.JsonClient) + client.socket = ClosedSocket() + client.conn = client.socket + client._recv_timeout = 0.25 + client._address = "127.0.0.1" + client._port = 5489 + client._is_listening = False + + assert client.connect() is True + assert ("fresh_connect", ("127.0.0.1", 5489)) in events + assert not [e for e in events if e[0] == "closed_connect"] + + def test_close_idempotent_and_connected_guard(): """close() is safe to call multiple times; connected guard tolerates None.""" # Client close idempotence