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

@@ -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

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
+        if debug > 2:
-        client_address = None
+            debugger("socket: waiting for a connection")
-
         try:
-            if debug > 2:
+            connection, _ = sock.accept()
-                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))
-
         except KeyboardInterrupt:
-            if connection:
-                connection.close()
             sys.exit(1)
-
+        threading.Thread(
-        finally:
+            target=handle_client, args=(connection, ups_devices, debug), name="client", daemon=True,
-            if connection:
+        ).start()
-                connection.close()
 
 
 if __name__ == "__main__":

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):
+    def test_keyboard_interrupt_ends_the_server(self, tmp_path):
-        """Test KeyboardInterrupt handler closes connection when connection is active."""
+        """An interrupt reaches main() while it waits for a client."""
         sock_path = str(tmp_path / "server_kbi.sock")
-        ready = threading.Event()
+        with patch("sys.argv", ["bmspy-server", "--socket", sock_path, "--device", "/dev/ttyUSB0"]), \
-
+             patch("bmspy.server.signal.signal"), \
-        from unittest.mock import MagicMock, patch
+             patch("bmspy.server.atexit.register"), \
-        from bmspy.server import main as server_main, read_request as original_read_request
+             patch("bmspy.server.initialise_serial"), \
-        from bmspy.jbd_bms import JBDBMS
+             patch("bmspy.server.start_polling"), \
-
+             patch("bmspy.server.os.path.isdir", return_value=True), \
-        fake_bms = JBDBMS()
+             patch.object(socket_module.socket, "accept", side_effect=KeyboardInterrupt):
-        call_count = [0]
+            from bmspy.server import main as server_main
-
+            with pytest.raises(SystemExit) as exit_info:
-        def _do_main_kbi():
+                server_main()
-            argv = ["bmspy-server", "--socket", sock_path, "--device", "/dev/ttyUSB0"]
+        assert exit_info.value.code == 1
-
-            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)
 
     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")