fix: attach ring buffers in the parent for fork producers

`Producer.__call__` attached the five ring-buffer shared-memory segments from
inside the child for every start method. Under `fork` that is not safe:
`SharedMemory(name=..., create=False)` calls
`multiprocessing.resource_tracker.register`, which takes a module-level
`threading.RLock` that CPython does not reinitialize after `fork`. A child that
inherits that lock held by a thread which does not exist in the child blocks on
the attach forever -- no timeout, nothing to interrupt it.

`WorkerBase.__init__` already handles this: it sets `_lazy_init = start_method
!= "fork"` and, for `fork`, initializes logging and attaches the rings in the
parent so the child inherits the mappings. `Producer` never got the same
treatment. It does now, mirroring the worker exactly.

Also logs the attach on both sides, as the worker already does, so a child that
does wedge there says so instead of going silent between two unrelated lines.

Verified on Linux (Python 3.12.13, TensorFlow 2.21.0):

- The new test counts shared-memory attaches performed outside the process that
  set the session up. On `main` a fork producer child performs five; with this
  change, none. It fails on `main` and passes here, deterministically -- the
  deadlock is a race, so the invariant is what gets asserted, not the hang.
- `fork` lane: 7 passed, 1 skipped.
- `inference` + `inference_pipeline` + `v2_4` + `core_py`, non-fork: 352 passed.
- 18 full `fork`-lane runs on this machine (10 unpinned, 8 pinned to 2 cores)
  never reproduced the CI hang, so this is not shown to change its rate.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: winedarksea

CHANGELOG.md

@@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Bugfixes
 
+- Producers no longer attach the ring buffers from inside a `fork` child. `SharedMemory(create=False)` calls `multiprocessing.resource_tracker.register`, whose module-level lock CPython does not reinitialize after `fork`, so a child that inherits it held by a thread which does not exist in the child blocks on the attach forever. The workers already avoided this by attaching in the parent and inheriting the mappings (`WorkerBase.__init__`); producers now follow the same rule. Measured on Linux: five attaches per producer child before, none after. This is a candidate cause of the intermittent 600 s hangs in the `fork` CI lane, where a wedged producer child logs its logging init and never reaches "Waiting for start signal" -- exactly the window the attaches sit in.
 - Fixed corrupt rows in acoustic prediction and encoding output: growing the internal result buffer via `numpy.ndarray.resize` (together with an off-by-one in the initial segment count) could leave stale or uninitialized values in some segments. Buffers are now reallocated and copied, so all output rows are correct (#50).
 - Geo model v3.0 caches now self-heal across releases: previously a cached ProtoBuf SavedModel or generated label files from an older release were never detected as stale, so a geo model version bump could keep serving outdated species labels/counts. The SavedModel now records its source release and the label files are validated against the current labels, so both are re-fetched/regenerated when they no longer match (#41).
 

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

@@ -154,7 +154,21 @@ def __init__(
 
     self._cancel_event = cancel_event
 
+    # Same rule the workers follow (see WorkerBase.__init__): only "fork"
+    # children inherit parent state, so only then may logging and the ring
+    # buffers be set up here in the parent. Attaching them in the child instead
+    # is not fork-safe -- SharedMemory(create=False) calls
+    # resource_tracker.register(), which takes a module-level threading lock
+    # that CPython does not reinitialize after fork. A child that inherits it
+    # held by a thread which no longer exists blocks on the attach forever.
+    self._lazy_init = start_method != "fork"
+
+    if not self._lazy_init:
+      self._init_logging()
+      self._load_ring_buffers()
+
   def _load_ring_buffers(self) -> None:
+    self._log("Attaching ring buffers...")
     self._shm_file_indices, self._ring_file_indices = (
       self._rf_file_indices.attach_and_get_array()
     )
@@ -168,6 +182,7 @@ def _load_ring_buffers(self) -> None:
       self._rf_batch_sizes.attach_and_get_array()
     )
     self._shm_ring_flags, self._ring_flags = self._rf_flags.attach_and_get_array()
+    self._log("Attached ring buffers.")
 
   def get_segments_from_input(
     self, input_idx: int, inp_data: Path | tuple[np.ndarray, int]
@@ -541,10 +556,12 @@ def _log(self, message: str) -> None:
     self._logger.debug(f"P_{os.getpid()}: {message}")
 
   def __call__(self) -> None:
-    self._init_logging()
+    if self._lazy_init:
+      self._init_logging()
 
     try:
-      self._load_ring_buffers()
+      if self._lazy_init:
+        self._load_ring_buffers()
 
       self._run_main_loop()
     except Exception as e:

src/birdnet_tests/acoustic_models/inference/session_py/test_fork_child_shm_attach.py

@@ -0,0 +1,76 @@
+"""Under ``fork``, no pipeline child may attach the ring buffers itself.
+
+``SharedMemory(name=..., create=False)`` calls
+``multiprocessing.resource_tracker.register``, which takes a module-level
+``threading.RLock`` that CPython does not reinitialize after ``fork``. A child
+that inherits that lock held by a thread which does not exist in the child
+blocks on the attach forever, with no timeout and nothing to interrupt it.
+
+``fork`` children inherit the parent's mappings, so they never need to attach:
+the parent does it before forking and the child uses what it inherited. This
+pins that invariant by counting the attaches that happen in a process other than
+the one that set the session up. It is deterministic -- the deadlock itself is a
+race, so it is not what gets asserted here.
+"""
+
+import multiprocessing as mp
+import os
+from multiprocessing import resource_tracker
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from birdnet.model_loader import load
+from birdnet_tests.helper import use_fork_or_skip
+
+SAMPLE_RATE = 48_000
+CLIP_DURATION_S = 9.0
+
+_original_register = resource_tracker.register
+_parent_pid: int | None = None
+_registrations_path: Path | None = None
+
+
+def _tracing_register(name: str, rtype: str) -> None:
+  # Runs in the parent and in every fork child. Only the child copies matter,
+  # and a child cannot report back through memory, so they append to a file.
+  if os.getpid() != _parent_pid and rtype == "shared_memory":
+    assert _registrations_path is not None
+    with _registrations_path.open("a", encoding="utf-8") as f:
+      f.write(f"{os.getpid()} {name}\n")
+  _original_register(name, rtype)
+
+
+@pytest.mark.fork
+def test_fork_children_do_not_attach_shared_memory_themselves(tmp_path: Path) -> None:
+  use_fork_or_skip()
+  assert mp.get_start_method(allow_none=True) == "fork"
+
+  global _parent_pid, _registrations_path
+  _parent_pid = os.getpid()
+  _registrations_path = tmp_path / "shm_registrations.txt"
+
+  rng = np.random.default_rng(0)
+  audio = rng.standard_normal(int(SAMPLE_RATE * CLIP_DURATION_S)).astype(np.float32)
+
+  resource_tracker.register = _tracing_register  # type: ignore[assignment]
+  try:
+    model = load("acoustic", "2.4", "tf", library="tflite")
+    with model.predict_session(n_workers=1, top_k=None) as session:
+      session.run_arrays((audio * 0.1, SAMPLE_RATE))
+  finally:
+    resource_tracker.register = _original_register  # type: ignore[assignment]
+
+  registrations = (
+    _registrations_path.read_text(encoding="utf-8").splitlines()
+    if _registrations_path.exists()
+    else []
+  )
+
+  assert not registrations, (
+    f"{len(registrations)} shared-memory attach(es) happened in a fork child "
+    f"instead of being inherited from the parent. Each one calls "
+    f"resource_tracker.register, whose lock is not reinitialized after fork and "
+    f"can deadlock the child permanently: {registrations}"
+  )