From eabca9f803f94be5eed032d30dae619cc20b6f85 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Mon, 7 Sep 2026 01:13:45 +0200 Subject: [PATCH] Make StatusInfoReader re-openable after close() StatusInfoReader subclasses both SourceReader and threading.Thread, and open() ended with self.start(). A Thread can only be started once, so the second open() always raised "threads can only be started once" -- breaking the reuse contract SourceReader's own class docstring demonstrates, and breaking StreamBuffer, which calls source_reader.open() on every start. open() now stops and joins the previous run's worker, then starts a fresh daemon thread. The join is required rather than cosmetic: open() clears the stop flag, and a previous worker still inside its time.sleep would never observe the stop, leaving two producers appending to one queue (counts drift 9 / 18 / 27 instead of holding at 9). The first open() is unchanged -- it still starts self -- so is_alive(), join() and the rest of the inherited Thread API behave exactly as before for single-use readers, isinstance(reader, threading.Thread) stays True, and close() stays non-blocking. Only the second open(), which until now raised unconditionally, behaves differently. How long open() waits is exposed as a keyword-only reopen_timeout_s, defaulting to None (wait for as long as it takes). A caller that sets a finite timeout and hits it gets an explanatory RuntimeError instead of two silent producers. Adds test_reader_can_be_reopened (three open/close cycles produce a steady sample count) and test_reopen_does_not_leak_worker_threads (a timing-free check that only the newest worker is ever alive). Both fail on the unmodified code and both also fail against a join-less implementation. Claude-Session: https://claude.ai/code/session_01L1aQPB34n7PU7jmbztSjBe --- pchealthstream2py/pchealth.py | 59 +++++++++++++++++++++++++- pchealthstream2py/tests/simple_test.py | 51 ++++++++++++++++++++++ 2 files changed, 109 insertions(+), 1 deletion(-) diff --git a/pchealthstream2py/pchealth.py b/pchealthstream2py/pchealth.py index 9aca477..cd223e4 100644 --- a/pchealthstream2py/pchealth.py +++ b/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,6 +230,8 @@ 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) @@ -234,6 +239,7 @@ def __init__( 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 diff --git a/pchealthstream2py/tests/simple_test.py b/pchealthstream2py/tests/simple_test.py index ca700c8..19ebec8 100644 --- a/pchealthstream2py/tests/simple_test.py +++ b/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()