Read each device in a background thread

A failed JBD read retried forever inside a client request and blocked every other device and client. The server now starts one polling thread for each device, and a request returns the last snapshot. Only the first request after startup waits, until each device completes one read attempt.

One lock makes sure that only one Bluetooth collection runs at a time. A serial BMS retries a failed read after 1 second and keeps its last snapshot for 30 seconds. An error in a collection does not end the polling of that device.

README.md

@@ -94,8 +94,12 @@ Omit `--ups` for all devices. Add `--ups river2` to select one. `--ups all` mean
 device named `all`, not all devices. To include a serial BMS, add a server argument
 such as `-d network:/dev/ttyUSB0`.
 
-The server caches Bluetooth data for 60 seconds. Failed reads wait 60 seconds
-before retrying. Unavailable devices are omitted from the response.
+The server reads each device in its own background thread, so a slow or offline
+device does not delay the other devices or the clients. It reads one Bluetooth
+device at a time and waits 60 seconds after each Bluetooth read, including a
+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.
 
 #### Diagnostics
 

bmspy/server.py

@@ -13,8 +13,9 @@ import signal
 import json
 import re
 import struct
+import threading
 import serial
-from dataclasses import asdict as dataclass_asdict, dataclass
+from dataclasses import asdict as dataclass_asdict, dataclass, field
 from types import FrameType
 from typing import Any, NoReturn
 
@@ -34,6 +35,18 @@ class DeviceState:
     data: UPS | None = None
     timestamp: float = 0.0                 # Last collection attempt
     ecoflow: EcoFlowBTConfig | None = None  # EcoFlow BT config
+    # Set after the first collection attempt, successful or not.
+    polled: threading.Event = field(default_factory=threading.Event, repr=False, compare=False)
+
+
+SERIAL_POLL_SECONDS = 5
+SERIAL_RETRY_SECONDS = 1
+SERIAL_STALE_SECONDS = 30
+BLUETOOTH_POLL_SECONDS = 60
+FIRST_POLL_TIMEOUT = 120
+
+# One adapter: collect from one Bluetooth device at a time.
+bluetooth_lock = threading.Lock()
 
 # Expected kernel log output when the USB-serial adapter is plugged in:
 # usb 1-1.4: new full-speed USB device number 4 using xhci_hcd
@@ -172,23 +185,57 @@ def collect_device_data(device: DeviceState, debug: int = 0) -> UPS | None:
     if device.ser is not None:
         return collect_data(device.ser, debug) or None
     if device.ecoflow is not None:
+        with bluetooth_lock:
             return ecoflow_collect_data(device.ecoflow, debug) or None
     return None
 
 
-def get_device_data(device: DeviceState, debug: int = 0) -> UPS | None:
-    """Return cached data, refreshing it when the backend's interval expires."""
-    min_age = 60 if device.ecoflow is not None else 5
-    if device.timestamp <= time.time() - min_age:
-        device.data = collect_device_data(device, debug)
-        if device.ecoflow is None:
-            # Preserve serial retries. BLE already retries within collect_data;
-            # retrying it forever here would block all clients when it is offline.
-            while device.data is None:
-                time.sleep(1)
-                device.data = collect_device_data(device, debug)
+def poll_device(device: DeviceState, stop: threading.Event, debug: int = 0) -> None:
+    """Keep *device.data* current until *stop* is set.
+
+    Each device has its own thread, so a slow or offline device delays neither
+    the clients nor the other devices. A collection replaces the snapshot and
+    never changes one that clients can already read, so no lock is needed.
+    """
+    last_success = 0.0
+    while not stop.is_set():
+        try:
+            data = collect_device_data(device, debug)
+        except Exception as e:
+            # A failed read must not end polling for this device.
+            debugger("server: collection error: {}".format(e))
+            data = None
         device.timestamp = time.time()
-    return device.data
+        if data is not None:
+            device.data = data
+            last_success = device.timestamp
+            delay = BLUETOOTH_POLL_SECONDS if device.ecoflow is not None else SERIAL_POLL_SECONDS
+        elif device.ecoflow is not None:
+            # BLE already retries within collect_data. Do not serve stale data.
+            device.data = None
+            delay = BLUETOOTH_POLL_SECONDS
+        else:
+            # Serial reads fail briefly at times. Retry soon, and keep the last
+            # snapshot for a short time so that one failed read stays invisible.
+            if device.timestamp - last_success > SERIAL_STALE_SECONDS:
+                device.data = None
+            delay = SERIAL_RETRY_SECONDS
+        device.polled.set()
+        stop.wait(delay)
+
+
+def start_polling(
+    ups_devices: dict[str, DeviceState], stop: threading.Event, debug: int = 0,
+) -> list[threading.Thread]:
+    """Start one polling thread per device."""
+    threads = []
+    for name, device in ups_devices.items():
+        thread = threading.Thread(
+            target=poll_device, args=(device, stop, debug), name="poll-" + name, daemon=True,
+        )
+        thread.start()
+        threads.append(thread)
+    return threads
 
 
 def main():
@@ -342,6 +389,10 @@ def main():
 
     sock.listen(1)
 
+    stop_polling = threading.Event()
+    atexit.register(stop_polling.set)
+    start_polling(ups_devices, stop_polling, debug)
+
     while True:
         connection = None
         client_address = None
@@ -398,7 +449,10 @@ def main():
                                     name, device.timestamp, time.time()
                                 )
                             )
-                        data = get_device_data(device, debug)
+                        # 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
 

tests/test_server.py

@@ -11,7 +11,17 @@ import serial
 
 from bmspy.classes import BMSScalarField, BMSInfoField, UPS
 from bmspy.jbd_bms import JBDBMS
-from bmspy.server import DeviceState, parse_device, read_request, send_response
+from bmspy.server import (
+    DeviceState, collect_device_data, parse_device, read_request, send_response,
+)
+
+
+def _poll_once(devices, stop, debug=0):
+    """Collect once in the server thread. main() stops its polling threads only
+    at process exit, so real ones would outlive a test and call later mocks."""
+    for device in devices.values():
+        device.data = collect_device_data(device, debug)
+        device.polled.set()
 
 
 # ---------------------------------------------------------------------------
@@ -340,6 +350,7 @@ class TestServerMain:
                  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), \
@@ -525,6 +536,7 @@ class TestServerMain:
                  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), \
@@ -576,6 +588,7 @@ class TestServerMain:
                  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.exists", return_value=False), \
                  patch("bmspy.server.os.path.isdir", return_value=False), \
@@ -686,6 +699,7 @@ class TestServerMain:
                  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), \
@@ -777,6 +791,7 @@ class TestServerMain:
                  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=False), \
                  patch("bmspy.server.os.makedirs"), \
@@ -838,6 +853,7 @@ class TestServerMain:
                  patch("bmspy.server.signal.signal"), \
                  patch("bmspy.server.initialise_serial", return_value=MagicMock()), \
                  patch("bmspy.server.collect_data", return_value=JBDBMS()), \
+                 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), \
@@ -901,6 +917,7 @@ class TestServerMain:
                  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), \

tests/test_server_ecoflow.py

@@ -1,5 +1,6 @@
 """Server routing and caching tests with synthetic device identifiers."""
 
+import threading
 from unittest.mock import Mock
 
 import pytest
@@ -74,46 +75,100 @@ def test_collection_dispatches_to_the_selected_backend(monkeypatch):
     assert server.collect_device_data(server.DeviceState()) is None
 
 
+class StopAfter:
+    """Stand-in for the stop event: record each wait and stop after *loops* polls."""
+
+    def __init__(self, loops, observe=lambda: None):
+        self.loops, self.observe = loops, observe
+        self.waits, self.seen = [], []
+
+    def is_set(self):
+        return len(self.waits) >= self.loops
+
+    def wait(self, delay):
+        self.waits.append(delay)
+        self.seen.append(self.observe())
+
+
 @pytest.mark.parametrize("bluetooth,interval", [(False, 5), (True, 60)])
-def test_cache_interval_depends_on_backend(monkeypatch, bluetooth, interval):
-    old, fresh = JBDBMS(), JBDBMS()
+def test_poll_interval_depends_on_backend(monkeypatch, bluetooth, interval):
+    fresh = JBDBMS()
     config = initialise_device(ADDRESS, SERIAL, "test-user") if bluetooth else None
-    state = server.DeviceState(data=old, timestamp=100, ecoflow=config)
+    state = server.DeviceState(ser=None if bluetooth else Mock(), ecoflow=config)
     read = Mock(return_value=fresh)
     monkeypatch.setattr(server, "collect_device_data", read)
-    monkeypatch.setattr(server.time, "time", lambda: 100 + interval - 1)
-    assert server.get_device_data(state) is old
-    read.assert_not_called()
-    monkeypatch.setattr(server.time, "time", lambda: 100 + interval)
-    assert server.get_device_data(state) is fresh
-    read.assert_called_once_with(state, 0)
+    monkeypatch.setattr(server.time, "time", lambda: 100)
+    stop = StopAfter(1)
+    server.poll_device(state, stop, 2)
+    assert state.data is fresh
+    assert state.timestamp == 100
+    assert state.polled.is_set()
+    assert stop.waits == [interval]
+    read.assert_called_once_with(state, 2)
 
 
-def test_offline_bluetooth_does_not_retry_forever_or_serve_stale_data(monkeypatch):
+def test_offline_bluetooth_does_not_retry_at_once_or_serve_stale_data(monkeypatch):
     state = server.DeviceState(
         data=JBDBMS(), timestamp=1, ecoflow=initialise_device(ADDRESS, SERIAL, "test-user"),
     )
-    read = Mock(return_value=None)
-    monkeypatch.setattr(server, "collect_device_data", read)
+    monkeypatch.setattr(server, "collect_device_data", Mock(return_value=None))
     monkeypatch.setattr(server.time, "time", lambda: 100)
-    assert server.get_device_data(state) is None
+    stop = StopAfter(1)
+    server.poll_device(state, stop)
+    assert state.data is None
     assert state.timestamp == 100
-    # A failed poll also has a backoff, rather than reconnecting on every GET.
-    assert server.get_device_data(state) is None
-    read.assert_called_once_with(state, 0)
-    monkeypatch.setattr(server.time, "time", lambda: 160)
-    assert server.get_device_data(state) is None
-    assert read.call_count == 2
+    # A client must not wait for a device that is known to be offline.
+    assert state.polled.is_set()
+    assert stop.waits == [60]
 
 
-def test_serial_collection_still_retries(monkeypatch):
+def test_failed_serial_read_retries_soon_and_hides_a_short_failure(monkeypatch):
     data = JBDBMS()
     state = server.DeviceState(ser=Mock())
-    read = Mock(side_effect=[None, data])
-    sleep = Mock()
+    monkeypatch.setattr(server, "collect_device_data", Mock(side_effect=[data, None, None]))
+    clock = iter([100, 110, 131])
+    monkeypatch.setattr(server.time, "time", lambda: next(clock))
+    stop = StopAfter(3, observe=lambda: state.data)
+    server.poll_device(state, stop)
+    # The last snapshot stays for 30 seconds, then the device reads as absent.
+    assert stop.seen == [data, data, None]
+    assert stop.waits == [5, 1, 1]
+
+
+def test_collection_error_does_not_end_polling(monkeypatch, capsys):
+    data = JBDBMS()
+    state = server.DeviceState(ser=Mock())
+    monkeypatch.setattr(server, "collect_device_data", Mock(side_effect=[RuntimeError("boom"), data]))
+    stop = StopAfter(2)
+    server.poll_device(state, stop)
+    assert state.data is data
+    assert stop.waits == [1, 5]
+    assert "collection error: boom" in capsys.readouterr().out
+
+
+def test_polling_threads_do_not_wait_for_each_other_and_stop(monkeypatch):
+    data = JBDBMS()
+    offline_started, release = threading.Event(), threading.Event()
+    working, offline = server.DeviceState(ser=Mock()), server.DeviceState(ser=Mock())
+
+    def read(device, debug=0):
+        if device is offline:
+            offline_started.set()
+            release.wait(5)
+            return None
+        return data
+
     monkeypatch.setattr(server, "collect_device_data", read)
-    monkeypatch.setattr(server.time, "time", lambda: 100)
-    monkeypatch.setattr(server.time, "sleep", sleep)
-    assert server.get_device_data(state) is data
-    assert read.call_count == 2
-    sleep.assert_called_once_with(1)
+    stop = threading.Event()
+    threads = server.start_polling({"offline": offline, "working": working}, stop)
+    try:
+        assert offline_started.wait(5)
+        assert working.polled.wait(5)
+        assert working.data is data
+        assert not offline.polled.is_set()
+    finally:
+        stop.set()
+        release.set()
+        for thread in threads:
+            thread.join(5)
+    assert not any(thread.is_alive() for thread in threads)