Repository navigation
Make StatusInfoReader re-openable after close() - #3
Merged
Merged
Conversation
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
thorwhalen
added a commit
that referenced
this pull request
Sep 22, 2026
#3 ran the first open() in self and later ones in fresh threads, so after a reopen reader.is_alive() was False while the reader was reading and reader.join() returned at once. Every open() now gets its own worker thread and is_alive/join follow it. Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Fixes i2mint/pchealthstream2py#1 —
StatusInfoReader.open()calledself.start()unconditionally, so a secondopen()afterclose()always raisedRuntimeError: threads can only be started once, violating the reuse contractSourceReader's own class docstring demonstrates.open()now stops and joins the previous run's worker thread before starting a new one, then reusesselffor the very first run (athreading.Threadcan only be started once) and a freshthreading.Thread(target=self.run, ...)for every subsequent reopen.reopen_timeout_s(defaultNone= wait indefinitely) bounds how longopen()waits for the previous worker to stop; if it doesn't stop in time, raises a clearRuntimeErrorexplaining why (rather than silently ending up with two producers feeding the same queue) instead of hanging forever with noNonedefault in place of an unbounded wait — additive, keyword-only, fully backward compatible.test_reader_can_be_reopened(drives 3 open/close cycles, asserts item counts don't drift — a leaked worker would double/triple the count each cycle) andtest_reopen_does_not_leak_worker_threads(inspects the actual worker threads directly, asserting only the newest is alive after each reopen).Verification
Branch already sat directly on current
master(no rebase needed; the unrelated already-merged PR #2, a packaging/CI modernization, is already in this history). fleet_dependents.json lists no importers ofpchealthstream2py.wads ci-local: PASSED — ruff format + lint, pytest py3.10/3.12 (5 passed, including both new reopen regression tests), build.Closes #1
🤖 Generated with Claude Code