Added explicit teardown (#62)

* Added explicit teardown

* fixed teardown test

Author: Josef-Haupt

src/birdnet/acoustic/inference/core/perf_tracker.py

@@ -18,7 +18,7 @@
 
 import birdnet.acoustic.inference.core.logs as bn_logging
 from birdnet.acoustic.inference.core.shm import RingField
-from birdnet.acoustic.inference.core.sync import CountedSemaphore
+from birdnet.acoustic.inference.core.sync import CountedSemaphore, abandon_queue_feeders
 from birdnet.globals import READABLE_FLAG, READING_FLAG, WRITABLE_FLAG
 
 
@@ -224,6 +224,17 @@ def __call__(self) -> None:
       )
       self._cancel_event.set()
 
+    # On cancellation the parent stops reading these queues; drop any buffered
+    # results/stats so this process can exit without blocking on its feeder
+    # threads (see abandon_queue_feeders).
+    if self._cancel_event.is_set():
+      abandon_queue_feeders(
+        self._perf_res,
+        self._callback_queue,
+        self._wkr_stats_queue,
+        self._prd_stats_queue,
+      )
+
     self._uninit_logging()
 
   def run_main_loop(self) -> None:

src/birdnet/acoustic/inference/core/producer.py

@@ -19,7 +19,7 @@
 
 import birdnet.acoustic.inference.core.logs as bn_logging
 from birdnet.acoustic.inference.core.shm import RingField
-from birdnet.acoustic.inference.core.sync import CountedSemaphore
+from birdnet.acoustic.inference.core.sync import CountedSemaphore, abandon_queue_feeders
 from birdnet.globals import (
   READABLE_FLAG,
   READING_FLAG,
@@ -552,6 +552,16 @@ def __call__(self) -> None:
       )
       self._cancel_event.set()
 
+    # On cancellation the parent stops reading these queues; drop any buffered
+    # stats/markers so this process can exit without blocking on its feeder
+    # threads (see abandon_queue_feeders).
+    if self._cancel_event.is_set():
+      abandon_queue_feeders(
+        self._prod_stats_queue,
+        self._unprocessed_inputs_queue,
+        self._completion_queue,
+      )
+
     self._uninit_logging()
 
   def _run_main_loop(self) -> None:

src/birdnet/acoustic/inference/core/sync.py

@@ -1,10 +1,33 @@
 from __future__ import annotations
 
 import multiprocessing as mp
+from multiprocessing import Queue
 from multiprocessing.sharedctypes import Synchronized
 from types import TracebackType
 
 
+def abandon_queue_feeders(*queues: Queue | None) -> None:
+  """Let the current process exit without flushing these queues' feeders.
+
+  On the cancellation path the parent stops reading the child->parent queues, so
+  any items still buffered in a child's feeder thread would otherwise block the
+  child's shutdown: its exit handler joins the feeder, which is stuck writing to
+  a full pipe that nobody drains. ``cancel_join_thread`` drops that buffered data
+  and lets the process exit immediately. Losing the data is fine here -- the run
+  was cancelled and the results are discarded anyway.
+
+  Exiting cleanly this way (instead of being force-terminated by the parent) also
+  avoids leaking the semaphores the child inherited: a killed process never runs
+  the finalizers that unregister them from the multiprocessing resource tracker.
+
+  Only call this once cancellation is certain; ``None`` entries are skipped so
+  callers can pass optional queues directly.
+  """
+  for q in queues:
+    if q is not None:
+      q.cancel_join_thread()
+
+
 class CountedSemaphore:
   """
   Drop-in replacement for ``mp.Semaphore`` whose ``get_value()`` works on

src/birdnet/acoustic/inference/core/worker.py

@@ -14,7 +14,7 @@
 
 import birdnet.acoustic.inference.core.logs as bn_logging
 from birdnet.acoustic.inference.core.shm import RingField
-from birdnet.acoustic.inference.core.sync import CountedSemaphore
+from birdnet.acoustic.inference.core.sync import CountedSemaphore, abandon_queue_feeders
 from birdnet.core.backends import BackendLoader, BatchT, VersionedBackendProtocol
 from birdnet.globals import (
   READABLE_FLAG,
@@ -200,6 +200,12 @@ def __call__(self) -> None:
       )
       self._cancel_event.set()
 
+    # On cancellation the consumer has stopped reading these queues; drop any
+    # buffered results/stats so this process can exit without blocking on its
+    # feeder threads (see abandon_queue_feeders).
+    if self._cancel_event.is_set():
+      abandon_queue_feeders(self._out_q, self._wkr_stats_queue)
+
     self._uninit_logging()
 
   def run_main_loop(self) -> None:

src/birdnet/acoustic/inference/process_manager.py

@@ -40,6 +40,12 @@
 if TYPE_CHECKING:
   from multiprocessing import Queue
 
+# After a cancelled process is asked to stop (SIGTERM), how long to wait for it
+# to actually exit before escalating to SIGKILL. Terminated processes normally
+# die within a moment; this only bounds a child that ignores SIGTERM so teardown
+# can never hang on it.
+_TERMINATE_JOIN_TIMEOUT_S = 5.0
+
 
 class ProcessManager:
   def __init__(
@@ -497,6 +503,7 @@ def _join_processes_after_cancel(
     queues = [q for q in queues if q is not None]
 
     deadline = time.monotonic() + grace_period_s
+    terminated = False
 
     while True:
       alive = [p for p in processes if p.is_alive()]
@@ -515,12 +522,20 @@ def _join_processes_after_cancel(
             f"Process '{p.name}' did not exit after cancellation; terminating."
           )
           p.terminate()
+        terminated = True
         break
 
       time.sleep(0.05)
 
+    # Reap every process. Ones that exited on their own join instantly; ones we
+    # just terminated get a bounded wait and are then SIGKILLed if they ignored
+    # SIGTERM, so a single unresponsive child can never hang teardown.
     for p in processes:
-      p.join()
+      p.join(timeout=_TERMINATE_JOIN_TIMEOUT_S if terminated else None)
+      if p.is_alive():
+        logger.warning(f"Process '{p.name}' ignored termination; killing.")
+        p.kill()
+        p.join()
       logger.debug(f"Process '{p.name}' finished.")
 
     self._producer_processes = None
@@ -532,3 +547,30 @@ def join_logging(self) -> None:
     assert self._logging_thread is not None
     self._logging_thread.join()
     self._logging_thread = None
+
+  def close_queues(self) -> None:
+    """Release the parent's handles on the multiprocessing queues.
+
+    Call once during teardown, after all child processes and the logging thread
+    have been joined. Closing is non-blocking (``cancel_join_thread`` first, so
+    ``close`` never waits on a feeder) and lets the OS reclaim the pipes and
+    semaphores promptly instead of leaving it to garbage collection -- which the
+    caller may skip entirely (e.g. ``os._exit``).
+    """
+    res = self._res
+    stats = res.stats_resources
+    queues: list[Queue | None] = [
+      res.producer_resources.input_queue,
+      res.producer_resources.unprocessed_inputs_queue,
+      res.worker_resources.results_queue,
+      res.file_completion_resources.marker_queue,
+      res.logging_resources.logging_queue,
+      stats.wkr_stats_queue,
+      stats.prd_stats_queue,
+      stats.perf_res_queue,
+      stats.callback_queue,
+    ]
+    for q in queues:
+      if q is not None:
+        q.cancel_join_thread()
+        q.close()

src/birdnet/acoustic/inference/session.py

@@ -228,6 +228,7 @@ def __exit__(self, *args) -> None:  # noqa: ANN002
     self._resources.logging_resources.stop_logging_event.set()
 
     self._process_manager.join_logging()
+    self._process_manager.close_queues()
     self._process_manager = None
 
     shutil.copyfile(

src/birdnet_tests/acoustic_models/v2_4/model_py/test_predict/test_cancel_teardown_v2_4.py

@@ -0,0 +1,92 @@
+"""Regression tests for cancelling a running prediction session (issue #51).
+
+Cancelling mid-run (e.g. a GUI "stop" button wired to a progress callback) must
+raise inside ``run()`` and then let the ``with`` block exit promptly, instead of
+hanging in ``ProcessManager.join()`` while it drains the worker queues.
+"""
+
+import shutil
+import threading
+from pathlib import Path
+
+import pytest
+
+from birdnet.acoustic.inference.core.perf_tracker import AcousticProgressStats
+from birdnet.model_loader import load
+from birdnet_tests.test_files import TEST_FILE_LONG
+
+
+def _load_model():  # noqa: ANN202
+  return load("acoustic", "2.4", "tf", precision="fp32", library="tflite")
+
+
+# Generous upper bound on cancel-to-teardown time: we cancel at ~10% progress, so
+# only a little inference runs and a clean teardown finishes well under this. Kept
+# below the global 300s per-test timeout, and enforced from a helper thread, so a
+# regression fails the assertion fast instead of hanging and killing the worker.
+_TEARDOWN_DEADLINE_S = 120.0
+
+
+def test_cancel_from_progress_callback_tears_down_cleanly(tmp_path: Path) -> None:
+  model = _load_model()
+
+  # Enough audio across multiple workers that the run lasts well beyond the first
+  # progress callback, so cancelling then reliably lands mid-run with many
+  # segments (and buffered result batches) still outstanding -- exactly the state
+  # that used to deadlock teardown. The run is cancelled almost immediately, so
+  # the large file list does not make the test slow.
+  #
+  # validate_input_files de-duplicates inputs by absolute path (it collects them
+  # into a set), so passing the same file N times collapses to a SINGLE input --
+  # the run then finishes before the first progress callback can cancel it, and
+  # run() returns without raising. Materialise N distinct copies so the run
+  # genuinely spans multiple files and workers.
+  src = Path(TEST_FILE_LONG)
+  files = [
+    str(shutil.copyfile(src, tmp_path / f"copy_{i}{src.suffix}")) for i in range(8)
+  ]
+
+  holder: dict = {}
+  cancelled = threading.Event()
+
+  def on_progress(stats: AcousticProgressStats) -> None:
+    # on_progress is only invoked once at least one prediction has been made, so
+    # the very first call already means the run is under way with work remaining.
+    session = holder.get("session")
+    if session is not None and not cancelled.is_set():
+      cancelled.set()
+      # Mirrors cancelling from a GUI "stop" button; runs on the dispatcher thread.
+      session.cancel()
+
+  errors: list[BaseException] = []
+  finished = threading.Event()
+
+  def run_session() -> None:
+    try:
+      with model.predict_session(
+        n_workers=2,
+        top_k=5,
+        show_stats="progress",
+        progress_callback=on_progress,
+      ) as session:
+        holder["session"] = session
+        with pytest.raises(RuntimeError, match="cancelled"):
+          session.run(files)
+      # Reaching here means __exit__ (join + teardown) returned without hanging.
+    except BaseException as exc:  # noqa: BLE001 - surfaced via the assertions below
+      errors.append(exc)
+    finally:
+      finished.set()
+
+  worker = threading.Thread(
+    target=run_session, name="cancel-teardown-test", daemon=True
+  )
+  worker.start()
+
+  completed = finished.wait(timeout=_TEARDOWN_DEADLINE_S)
+  assert completed, (
+    f"session did not tear down within {_TEARDOWN_DEADLINE_S:.0f}s after "
+    f"cancel() -- ProcessManager.join() likely hung"
+  )
+  assert cancelled.is_set(), "progress callback never reached the cancel threshold"
+  assert not errors, f"unexpected error during cancelled run/teardown: {errors!r}"