Make StatusInfoReader re-openable after close()

pchealthstream2py/pchealth.py

@@ -22,6 +22,9 @@
 __all__ = ["StatusInfo", "StatusInfoReader"]
 
 DFLT_STATUS_INFO_READ_INTERVAL = 1000  # in ms
+# How long `StatusInfoReader.open()` waits for the previous run's worker thread
+# to stop before starting a new one. `None` means wait for as long as it takes.
+DFLT_REOPEN_TIMEOUT_S = None
 
 
 class StatusInfo:
@@ -227,13 +230,16 @@ def __init__(
         read_interval_ms=DFLT_STATUS_INFO_READ_INTERVAL,
         include_network_download_speed: bool = False,
         include_network_upload_speed: bool = False,
+        *,
+        reopen_timeout_s: Optional[float] = DFLT_REOPEN_TIMEOUT_S,
     ):
 
         threading.Thread.__init__(self, daemon=True)
 
         self.read_interval_ms = read_interval_ms
         self.include_network_download_speed = include_network_download_speed
         self.include_network_upload_speed = include_network_upload_speed
+        self.reopen_timeout_s = reopen_timeout_s
 
         # Per-instance state. These used to be *class* attributes, which meant
         # every StatusInfoReader shared one queue and one stop flag.
@@ -245,14 +251,65 @@ def __init__(
         # `join()` and `is_alive()`), so shadowing it with an Event breaks
         # `join()` with "TypeError: 'Event' object is not callable".
         self._stop_event: threading.Event = threading.Event()
+        # The thread currently running `run`, or None while the reader has never
+        # been opened. It is `self` for the first run (see `_start_worker`).
+        self._worker: Optional[threading.Thread] = None
 
     def open(self):
+        """Start reading status info in the background.
+
+        Can be called again after `close()`: a reader is reusable, as
+        `SourceReader`'s reuse contract requires. Each call stops and waits for
+        the worker started by the previous call, so exactly one thread feeds
+        the queue at a time.
+        """
+        self._stop_previous_worker()
+
         self._data.clear()
         self._bt = self.get_timestamp()
         self._index = 0
 
         self._stop_event.clear()
-        self.start()
+        self._start_worker()
+
+    def _stop_previous_worker(self):
+        """Signal and wait for the worker started by an earlier `open()`, if any.
+
+        Waiting is required, not cosmetic: `open()` clears the stop flag, and a
+        previous worker still inside its `time.sleep` would never observe the
+        stop, leaving two threads appending to the same queue.
+        """
+        worker = self._worker
+        if worker is None or not worker.is_alive():
+            return
+
+        self._stop_event.set()
+        worker.join(self.reopen_timeout_s)
+        if worker.is_alive():
+            raise RuntimeError(
+                f"{type(self).__name__}.open(): the worker thread of the previous"
+                f" run did not stop within reopen_timeout_s={self.reopen_timeout_s}"
+                f" seconds, so a new one can't be started without ending up with"
+                f" two producers feeding the same queue. Give the reader a longer"
+                f" reopen_timeout_s (or None to wait for as long as it takes)."
+            )
+
+    def _start_worker(self):
+        """Run `run` in a daemon thread, reusing `self` for the very first run.
+
+        A `threading.Thread` can only be started once, so from the second
+        `open()` on the work has to go in a fresh thread. The first `open()`
+        still starts `self`, which keeps `is_alive()`, `join()` and the rest of
+        the inherited `Thread` API behaving exactly as they always have.
+        """
+        if self._worker is None:
+            self._worker = self
+            self.start()
+        else:
+            self._worker = threading.Thread(
+                target=self.run, daemon=True, name=f"{self.name}-worker"
+            )
+            self._worker.start()
 
     def read(self):
         """Returns one data item

pchealthstream2py/tests/simple_test.py

@@ -52,3 +52,54 @@ def test_stop_does_not_shadow_thread_internals():
         reader.close()
     reader.join(timeout=10)
     assert not reader.is_alive()
+
+
+def test_reader_can_be_reopened():
+    """A reader must be usable again after `close()`.
+
+    Regression test: `StatusInfoReader` is also a `threading.Thread`, and
+    `open()` used to call `self.start()` unconditionally, so the second
+    `open()` always raised "threads can only be started once" -- violating the
+    reuse contract that `SourceReader`'s own class docstring demonstrates.
+
+    The count assertion covers the other half of the bug: if `open()` restarts
+    without joining the previous worker, that worker is still sleeping when the
+    stop flag is cleared, so it never exits and each cycle adds another producer
+    to the same queue (counts drift 9 / 18 / 27 instead of holding at 9).
+    """
+    read_interval_ms = 50
+    reader = StatusInfoReader(read_interval_ms=read_interval_ms)
+
+    counts = []
+    for _ in range(3):
+        with reader:
+            time.sleep(10 * read_interval_ms / 1000)
+        counts.append(reader._data.len())
+
+    assert all(count > 0 for count in counts), counts
+    # One producer per cycle, not one more on every cycle. A leaked producer
+    # multiplies the count by the cycle number, so the ratio separates the bug
+    # from ordinary scheduling jitter.
+    assert max(counts) < 2 * min(counts), counts
+
+
+def test_reopen_does_not_leak_worker_threads():
+    """Reopening must leave exactly one live worker, not one more each time.
+
+    The timing-free half of `test_reader_can_be_reopened`: it inspects the
+    worker threads themselves rather than how much they produced.
+    """
+    reader = StatusInfoReader(read_interval_ms=50)
+    workers = []
+
+    for _ in range(3):
+        with reader:
+            time.sleep(0.2)
+            workers.append(reader._worker)
+            # Only the newest worker runs; `open()` joined the earlier ones.
+            expected = [False] * (len(workers) - 1) + [True]
+            assert [worker.is_alive() for worker in workers] == expected
+
+    for worker in workers:
+        worker.join(timeout=10)
+        assert not worker.is_alive()