Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 33 additions & 14 deletions pchealthstream2py/pchealth.py
Original file line number Diff line number Diff line change
Expand Up @@ -252,7 +252,7 @@ def __init__(
# `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`).
# been opened (see `_start_worker`).
self._worker: Optional[threading.Thread] = None

def open(self):
Expand Down Expand Up @@ -295,21 +295,40 @@ def _stop_previous_worker(self):
)

def _start_worker(self):
"""Run `run` in a daemon thread, reusing `self` for the very first run.
"""Run `run` in a fresh daemon thread: one per `open()`.

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.
A `threading.Thread` can only be started once, so a reusable reader
can't run its work in `self` -- not even the first time, or `self` would
be a finished thread standing in for the reader from the second `open()`
on. `is_alive()` and `join()` are overridden to follow the current
worker, so they keep meaning "is it reading" and "wait for it to stop".
"""
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()
self._worker = threading.Thread(
target=self.run, daemon=True, name=f"{self.name}-worker"
)
self._worker.start()

def _current_worker_is_another_thread(self) -> bool:
worker = self._worker
return worker is not None and worker is not self

def is_alive(self) -> bool:
"""Whether the reader is currently reading, whichever thread is doing it.

The work runs in a per-`open()` worker thread (see `_start_worker`), not
in `self`, so the inherited `Thread.is_alive` would describe the wrong
thread. (A reader started the legacy way, with `start()` rather than
`open()`, has no worker and answers for `self` as before.)
"""
if self._current_worker_is_another_thread():
return self._worker.is_alive()
return super().is_alive()

def join(self, timeout: Optional[float] = None) -> None:
"""Wait for the current run's worker thread to finish (see `is_alive`)."""
if self._current_worker_is_another_thread():
return self._worker.join(timeout)
return super().join(timeout)

def read(self):
"""Returns one data item
Expand Down
24 changes: 24 additions & 0 deletions pchealthstream2py/tests/simple_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -103,3 +103,27 @@ def test_reopen_does_not_leak_worker_threads():
for worker in workers:
worker.join(timeout=10)
assert not worker.is_alive()


def test_thread_api_follows_the_current_worker_after_reopen():
"""`is_alive()` / `join()` must describe the run in progress, not the first one.

#3 ran the first `open()` in `self` and later ones in fresh threads, so after
a reopen the inherited `Thread.is_alive` answered False for a reader that was
busy reading (it described the finished first-run thread), and `join()`
returned at once instead of waiting for the worker to stop.
"""
reader = StatusInfoReader(read_interval_ms=50)
with reader:
time.sleep(0.1)
reader.join(timeout=10)

reader.open()
try:
time.sleep(0.1)
assert reader.is_alive()
finally:
reader.close()
reader.join(timeout=10)
assert not reader.is_alive()
assert not reader._worker.is_alive()
Loading