From f17d1d82c2ef4e1472a9e31dd17329ce2306b7cf Mon Sep 17 00:00:00 2001 From: Timothy Allen Date: Sat, 10 Oct 2026 10:59:34 +0200 Subject: [PATCH] Answer each client in its own thread The server handled one client at a time, so a client that connected and sent nothing blocked every other client. The request handling is now in handle_client, and the main loop starts one thread for each connection. The server closes a connection that stays silent for 5 seconds, so its thread ends. --- README.md | 3 + bmspy/server.py | 196 ++++++++++++++++++++++--------------------- tests/test_server.py | 118 +++++++++++++------------- 3 files changed, 163 insertions(+), 154 deletions(-) diff --git a/README.md b/README.md index 5f380a4..35747ac 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,9 @@ failed one. It reads a serial BMS every 5 seconds and retries a failed serial read after 1 second. Unavailable devices are omitted from the response. Only the first request after startup waits, until each device completes one read attempt. +The server answers each client in its own thread, so a slow client does not +delay the other clients. It closes a connection that stays silent for 5 seconds. + Each Bluetooth read listens for 10 seconds. If a required report is still missing, the read continues until the report arrives, for at most 30 seconds in total. A read that reaches this limit returns a partial snapshot. Its diff --git a/bmspy/server.py b/bmspy/server.py index b2c0242..b4c375a 100755 --- a/bmspy/server.py +++ b/bmspy/server.py @@ -44,6 +44,8 @@ SERIAL_RETRY_SECONDS = 1 SERIAL_STALE_SECONDS = 30 BLUETOOTH_POLL_SECONDS = 60 FIRST_POLL_TIMEOUT = 120 +# Time limit for each read from a client and each write to it. +CLIENT_TIMEOUT_SECONDS = 5 # One adapter: collect from one Bluetooth device at a time. bluetooth_lock = threading.Lock() @@ -238,6 +240,99 @@ def start_polling( return threads +def handle_client( + connection: socket.socket, ups_devices: dict[str, DeviceState], debug: int = 0, +) -> None: + """Answer one request, then close the connection. + + Each client has its own thread, so a slow or silent client does not delay + the other clients. + """ + try: + # A client that sends no request, or does not read its reply, must not + # keep its thread forever. + connection.settimeout(CLIENT_TIMEOUT_SECONDS) + try: + request_data = read_request(connection, debug) + except Exception as e: + print("socket ERROR: {}".format(e)) + return + + if not isinstance(request_data, dict): + request_data = {} + client = request_data.get("client") or "unknown" + command = request_data.get("command") + + match command: + case "REGISTER": + connected_clients.append(client) + send_response( + connection, + {"status": "REGISTERED", "client": client}, + client, + debug, + ) + + case "DEREGISTER": + try: + connected_clients.remove(client) + except Exception: + pass + send_response( + connection, + {"status": "DEREGISTERED", "client": client}, + client, + debug, + ) + + case "DEVICES": + # Configured names, including devices that have no data yet. + send_response(connection, {"devices": list(ups_devices)}, client, debug) + + case "GET": + ups_filter = request_data.get("ups") + targets = ( + {ups_filter: ups_devices[ups_filter]} + if ups_filter and ups_filter in ups_devices + else ups_devices + ) + + result = {} + for name, device in targets.items(): + if debug > 0: + debugger( + "reading data for '{}', timestamp={}, time={}".format( + name, device.timestamp, time.time() + ) + ) + # Wait for the first collection attempt only, so that a + # client at startup sees every device that can answer. + device.polled.wait(FIRST_POLL_TIMEOUT) + data = device.data + if data is not None: + result[name] = data + + send_response(connection, result, client, debug) + + case _: + # Answer and continue: one bad request must not end the + # server for every other client. + debugger("socket: invalid request from {}".format(client)) + send_response( + connection, + {"status": "ERROR", "message": "unknown command: {!r}".format(command)}, + client, + debug, + ) + + except OSError as e: + # For example, the client disconnected before its reply. + debugger("socket ERROR: {}".format(e)) + + finally: + connection.close() + + def main(): import argparse import pwd @@ -394,104 +489,15 @@ def main(): start_polling(ups_devices, stop_polling, debug) while True: - connection = None - client_address = None - + if debug > 2: + debugger("socket: waiting for a connection") try: - if debug > 2: - debugger("socket: waiting for a connection") - connection, client_address = sock.accept() - - request_data = dict() - try: - request_data = read_request(connection, debug) - except Exception as e: - print("socket ERROR: {}".format(e)) - continue - - if not isinstance(request_data, dict): - request_data = {} - client = request_data.get("client") or "unknown" - command = request_data.get("command") - - match command: - case "REGISTER": - connected_clients.append(client) - send_response( - connection, - {"status": "REGISTERED", "client": client}, - client, - debug, - ) - - case "DEREGISTER": - try: - connected_clients.remove(client) - except Exception: - pass - send_response( - connection, - {"status": "DEREGISTERED", "client": client}, - client, - debug, - ) - - case "DEVICES": - # Configured names, including devices that have no data yet. - send_response(connection, {"devices": list(ups_devices)}, client, debug) - - case "GET": - ups_filter = request_data.get("ups") - targets = ( - {ups_filter: ups_devices[ups_filter]} - if ups_filter and ups_filter in ups_devices - else ups_devices - ) - - result = {} - for name, device in targets.items(): - if debug > 0: - debugger( - "reading data for '{}', timestamp={}, time={}".format( - name, device.timestamp, time.time() - ) - ) - # Wait for the first collection attempt only, so that a - # client at startup sees every device that can answer. - device.polled.wait(FIRST_POLL_TIMEOUT) - data = device.data - if data is not None: - result[name] = data - - send_response(connection, result, client, debug) - - case _: - # Answer and continue: one bad request must not end the - # server for every other client. - debugger("socket: invalid request from {}".format(client)) - send_response( - connection, - {"status": "ERROR", "message": "unknown command: {!r}".format(command)}, - client, - debug, - ) - - except OSError as e: - if connection is None: - # accept() failed, so the listening socket is not usable. - raise - # A client that disconnects before its reply must not end the - # server for every other client. - debugger("socket ERROR: {}".format(e)) - + connection, _ = sock.accept() except KeyboardInterrupt: - if connection: - connection.close() sys.exit(1) - - finally: - if connection: - connection.close() + threading.Thread( + target=handle_client, args=(connection, ups_devices, debug), name="client", daemon=True, + ).start() if __name__ == "__main__": diff --git a/tests/test_server.py b/tests/test_server.py index f7443c5..1cf903c 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -5,6 +5,7 @@ import struct import subprocess import sys import threading +import time from unittest.mock import patch @@ -702,66 +703,20 @@ class TestServerMain: finally: self._stop_server(sock_path, t) - def test_keyboard_interrupt_closes_connection(self, tmp_path): - """Test KeyboardInterrupt handler closes connection when connection is active.""" + def test_keyboard_interrupt_ends_the_server(self, tmp_path): + """An interrupt reaches main() while it waits for a client.""" sock_path = str(tmp_path / "server_kbi.sock") - ready = threading.Event() - - from unittest.mock import MagicMock, patch - from bmspy.server import main as server_main, read_request as original_read_request - from bmspy.jbd_bms import JBDBMS - - fake_bms = JBDBMS() - call_count = [0] - - def _do_main_kbi(): - argv = ["bmspy-server", "--socket", sock_path, "--device", "/dev/ttyUSB0"] - - original_listen = socket_module.socket.listen - - def _patched_listen(self, backlog=1): - result = original_listen(self, backlog) - ready.set() - return result - - def _patched_read_request(conn, debug=0): - call_count[0] += 1 - if call_count[0] >= 2: - raise KeyboardInterrupt("test interrupt") - return original_read_request(conn, debug) - - with patch("sys.argv", argv), \ - patch("bmspy.server.signal.signal"), \ - patch("bmspy.server.initialise_serial", return_value=MagicMock()), \ - patch("bmspy.server.collect_data", return_value=fake_bms), \ - patch("bmspy.server.start_polling", _poll_once), \ - patch("bmspy.server.time.sleep"), \ - patch("bmspy.server.os.path.isdir", return_value=True), \ - patch("bmspy.server.os.path.exists", return_value=False), \ - patch("bmspy.server.read_request", side_effect=_patched_read_request), \ - patch.object(socket_module.socket, "listen", _patched_listen): - try: - server_main() - except SystemExit: - pass - - t = threading.Thread(target=_do_main_kbi, daemon=True) - t.start() - ready.wait(timeout=2) - import time as _t - _t.sleep(0.2) - # Send first request to register call_count[0] = 1 - try: - self._send_command(sock_path, {"command": "REGISTER", "client": "test"}) - except Exception: - pass - _t.sleep(0.1) - # Send second request to trigger KeyboardInterrupt with active connection - try: - self._send_command(sock_path, {"command": "REGISTER", "client": "test2"}) - except Exception: - pass - self._stop_server(sock_path, t) + with patch("sys.argv", ["bmspy-server", "--socket", sock_path, "--device", "/dev/ttyUSB0"]), \ + patch("bmspy.server.signal.signal"), \ + patch("bmspy.server.atexit.register"), \ + patch("bmspy.server.initialise_serial"), \ + patch("bmspy.server.start_polling"), \ + patch("bmspy.server.os.path.isdir", return_value=True), \ + patch.object(socket_module.socket, "accept", side_effect=KeyboardInterrupt): + from bmspy.server import main as server_main + with pytest.raises(SystemExit) as exit_info: + server_main() + assert exit_info.value.code == 1 def test_socket_read_error_logs_and_continues(self, tmp_path, capsys): """Test that read_request errors are caught and logged.""" @@ -1036,6 +991,51 @@ class TestServerMain: self._stop_server(sock_path, t) assert "socket ERROR: unable to encode response" in capsys.readouterr().out + def test_clients_are_served_at_the_same_time(self, tmp_path, monkeypatch): + """A client that is slow to send its request must not delay another client.""" + monkeypatch.setattr("bmspy.server.CLIENT_TIMEOUT_SECONDS", 30) + sock_path = str(tmp_path / "server_two.sock") + ready = threading.Event() + t = self._make_server_thread(sock_path, ready, threading.Event()) + t.start() + ready.wait(timeout=2) + slow = socket_module.socket(socket_module.AF_UNIX, socket_module.SOCK_STREAM) + slow.settimeout(5) + try: + # The first client connects and sends nothing yet. + slow.connect(sock_path) + start = time.monotonic() + response = self._send_command(sock_path, {"command": "REGISTER", "client": "fast"}) + assert response == {"status": "REGISTERED", "client": "fast"} + assert time.monotonic() - start < 5 + # The first connection is still open, and gets its own answer. + payload = json.dumps({"command": "DEVICES", "client": "slow"}).encode() + slow.sendall(struct.pack("!I", len(payload)) + payload) + length = struct.unpack("!I", slow.recv(4))[0] + assert json.loads(slow.recv(length)) == {"devices": ["ttyUSB0"]} + finally: + slow.close() + self._stop_server(sock_path, t) + + def test_silent_client_is_disconnected_at_the_time_limit(self, tmp_path, capsys, monkeypatch): + """A client that never sends a request must not keep its thread forever.""" + monkeypatch.setattr("bmspy.server.CLIENT_TIMEOUT_SECONDS", 0.2) + sock_path = str(tmp_path / "server_silent.sock") + ready = threading.Event() + t = self._make_server_thread(sock_path, ready, threading.Event()) + t.start() + ready.wait(timeout=2) + silent = socket_module.socket(socket_module.AF_UNIX, socket_module.SOCK_STREAM) + silent.settimeout(5) + try: + silent.connect(sock_path) + # An empty read means that the server closed the connection. + assert silent.recv(1) == b"" + finally: + silent.close() + self._stop_server(sock_path, t) + assert "socket ERROR: unable to read request length from socket: timed out" in capsys.readouterr().out + def test_failed_accept_still_ends_the_server(self, tmp_path): """A listening socket that fails must not become an endless loop.""" sock_path = str(tmp_path / "server_accept.sock")