fix(bg): serialize daemon-thread start under a lifecycle lock
start_drain_thread and start_session_channel_reaper checked is_alive() and then created + started the daemon thread without holding a lock, a check-then-act race: two concurrent callers could both see "not alive" and each spawn a thread. The loser's thread was never stored in the module global and ran forever, un-joinable by the matching stop_*. Both check-then-start sequences now run under a dedicated _THREAD_LIFECYCLE_LOCK (kept separate from the purpose-bound SESSION_CHANNELS_LOCK / _EMIT_COALESCE_LOCK), so exactly one thread is ever created. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
460f3eb47d
commit
d505b45545
2 files changed
+128
-20
No files matched your search
+30
-20
@@ -57,6 +57,14 @@ _REAPER_THREAD: Optional[threading.Thread] = None
|
||||
_REAPER_STOP = threading.Event()
|
||||
_REAPER_INTERVAL_SECS = 60.0
|
||||
|
||||
# Serializes the check-then-start of the module's daemon threads
|
||||
# (``start_drain_thread`` / ``start_session_channel_reaper``). Without it two
|
||||
# concurrent callers can both observe ``is_alive() == False`` and each spawn a
|
||||
# thread; the loser's thread is never referenced by the module global and runs
|
||||
# forever, un-joinable. A dedicated lock (not the purpose-bound
|
||||
# ``SESSION_CHANNELS_LOCK`` / ``_EMIT_COALESCE_LOCK``) keeps this narrow.
|
||||
_THREAD_LIFECYCLE_LOCK = threading.Lock()
|
||||
|
||||
# T3: per-session coalesce gate for the public bg_task_complete SSE emit.
|
||||
# The server-side wakeup path remains immediate; only the browser-observation
|
||||
# frame is throttled so a burst of background task completions does not flood an
|
||||
@@ -413,16 +421,17 @@ def _reaper_loop() -> None:
|
||||
def start_session_channel_reaper() -> bool:
|
||||
"""Start the SessionChannel reaper thread. Idempotent; returns True on first start."""
|
||||
global _REAPER_THREAD
|
||||
if _REAPER_THREAD is not None and _REAPER_THREAD.is_alive():
|
||||
return False
|
||||
_REAPER_STOP.clear()
|
||||
_REAPER_THREAD = threading.Thread(
|
||||
target=_reaper_loop,
|
||||
name="hermes-webui-session-channel-reaper",
|
||||
daemon=True,
|
||||
)
|
||||
_REAPER_THREAD.start()
|
||||
return True
|
||||
with _THREAD_LIFECYCLE_LOCK:
|
||||
if _REAPER_THREAD is not None and _REAPER_THREAD.is_alive():
|
||||
return False
|
||||
_REAPER_STOP.clear()
|
||||
_REAPER_THREAD = threading.Thread(
|
||||
target=_reaper_loop,
|
||||
name="hermes-webui-session-channel-reaper",
|
||||
daemon=True,
|
||||
)
|
||||
_REAPER_THREAD.start()
|
||||
return True
|
||||
|
||||
|
||||
def stop_session_channel_reaper(timeout: float = 2.0) -> None:
|
||||
@@ -1372,16 +1381,17 @@ def forget_bg_task_completion_dedup(session_id: str) -> None:
|
||||
def start_drain_thread() -> bool:
|
||||
"""Start the background drain thread idempotently. Returns True on first start."""
|
||||
global _DRAIN_THREAD
|
||||
if _DRAIN_THREAD is not None and _DRAIN_THREAD.is_alive():
|
||||
return False
|
||||
_DRAIN_STOP.clear()
|
||||
_DRAIN_THREAD = threading.Thread(
|
||||
target=_drain_loop,
|
||||
name="hermes-webui-bg-task-complete-drain",
|
||||
daemon=True,
|
||||
)
|
||||
_DRAIN_THREAD.start()
|
||||
return True
|
||||
with _THREAD_LIFECYCLE_LOCK:
|
||||
if _DRAIN_THREAD is not None and _DRAIN_THREAD.is_alive():
|
||||
return False
|
||||
_DRAIN_STOP.clear()
|
||||
_DRAIN_THREAD = threading.Thread(
|
||||
target=_drain_loop,
|
||||
name="hermes-webui-bg-task-complete-drain",
|
||||
daemon=True,
|
||||
)
|
||||
_DRAIN_THREAD.start()
|
||||
return True
|
||||
|
||||
|
||||
def stop_drain_thread(timeout: float = 2.0) -> None:
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
"""Regression test for the thread-start check-then-act race in background_process.
|
||||
|
||||
``start_drain_thread`` and ``start_session_channel_reaper`` checked
|
||||
``is_alive()`` and then created + started the daemon thread WITHOUT a lock, so
|
||||
two concurrent callers could both observe "not alive" and each spawn a thread.
|
||||
The loser's thread was never stored in the module global and ran forever,
|
||||
un-joinable. Both check-then-start sequences are now serialized under
|
||||
``_THREAD_LIFECYCLE_LOCK``, so exactly one thread is ever created.
|
||||
|
||||
The test drives N callers through a barrier so they hit the check-and-start
|
||||
simultaneously, and asserts exactly one caller reports "started" (returns True)
|
||||
and exactly one live thread exists.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# (start_fn, stop_fn, loop_attr, thread_attr, thread_name_fragment)
|
||||
_CASES = [
|
||||
(
|
||||
"start_drain_thread",
|
||||
"stop_drain_thread",
|
||||
"_drain_loop",
|
||||
"_DRAIN_THREAD",
|
||||
"bg-task-complete-drain",
|
||||
),
|
||||
(
|
||||
"start_session_channel_reaper",
|
||||
"stop_session_channel_reaper",
|
||||
"_reaper_loop",
|
||||
"_REAPER_THREAD",
|
||||
"session-channel-reaper",
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"start_name,stop_name,loop_attr,thread_attr,name_frag", _CASES
|
||||
)
|
||||
def test_concurrent_start_creates_exactly_one_thread(
|
||||
monkeypatch, start_name, stop_name, loop_attr, thread_attr, name_frag
|
||||
):
|
||||
from api import background_process as bp
|
||||
|
||||
release = threading.Event()
|
||||
|
||||
def _blocking_loop() -> None:
|
||||
# Keep the started thread alive during the assertions so a genuinely
|
||||
# started thread reads as is_alive()==True deterministically.
|
||||
release.wait(5.0)
|
||||
|
||||
monkeypatch.setattr(bp, loop_attr, _blocking_loop, raising=True)
|
||||
# Fresh slate: no pre-existing thread reference.
|
||||
monkeypatch.setattr(bp, thread_attr, None, raising=True)
|
||||
|
||||
start_fn = getattr(bp, start_name)
|
||||
n = 12
|
||||
barrier = threading.Barrier(n)
|
||||
results: list[bool] = []
|
||||
results_lock = threading.Lock()
|
||||
|
||||
def _caller() -> None:
|
||||
barrier.wait()
|
||||
started = start_fn()
|
||||
with results_lock:
|
||||
results.append(started)
|
||||
|
||||
callers = [threading.Thread(target=_caller) for _ in range(n)]
|
||||
try:
|
||||
for c in callers:
|
||||
c.start()
|
||||
for c in callers:
|
||||
c.join(timeout=5.0)
|
||||
|
||||
# Exactly one caller won the create; the rest saw a live thread.
|
||||
assert sum(1 for r in results if r) == 1, (
|
||||
f"expected exactly one start to win, got {results}"
|
||||
)
|
||||
# Exactly one live daemon thread carrying this loop's name.
|
||||
live = [
|
||||
t
|
||||
for t in threading.enumerate()
|
||||
if name_frag in t.name and t.is_alive()
|
||||
]
|
||||
assert len(live) == 1, f"expected 1 live {name_frag} thread, got {len(live)}"
|
||||
# The module global points at that one live thread.
|
||||
assert getattr(bp, thread_attr) is live[0]
|
||||
finally:
|
||||
release.set()
|
||||
getattr(bp, stop_name)()
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover - manual invocation
|
||||
raise SystemExit(pytest.main([__file__, "-v"]))
|
||||
Reference in new issue
Block a user