race condition testing

Author: LC-Linkous

hackrfpy/src/hackrfpy/core.py

@@ -29,6 +29,7 @@
 import subprocess
 import sys
 import threading
+import time
 import weakref
 from types import TracebackType
 from typing import Any, Callable, Generator, Literal, Protocol
@@ -203,6 +204,14 @@ def _deadman_attach(proc: subprocess.Popen[Any]) -> Any:
         return None                      # POSIX path is the preexec
 
 
+_BUSY_MARKERS = ("Resource busy", "(-1000)")
+
+
+def _is_busy(text: str) -> bool:
+    # hackrf_open()'s "device still claimed" signature; see busy_retries.
+    return any(m in text for m in _BUSY_MARKERS)
+
+
 class _Stoppable(Protocol):
     # What the atexit backstop actually needs. _Process and PersistentReceiver
     # both satisfy it; a Protocol keeps _LIVE honest about holding both without
@@ -385,6 +394,14 @@ def __init__(self, tools_dir: str | None = None, verbose: bool = False,
         # TX handles are ALWAYS atexit-stopped. RX handles are too by default;
         # set backstop_rx=False for deliberate fire-and-forget receivers.
         self.backstop_rx = True
+        # Device-busy retry: hackrf_open() reports "Resource busy (-1000)"
+        # when the previous child's USB claim has not been released yet --
+        # on Linux the kernel takes a beat after a tool exits, so rapid
+        # back-to-back operations (scan_frequencies, test suites, any
+        # capture-then-sweep script) can lose the race. Bounded retries with
+        # backoff absorb it; set busy_retries=0 to fail immediately.
+        self.busy_retries = 3
+        self.busy_backoff = 0.25         # seconds; doubles per attempt
 
         # ---- parameter readback ----
         # Populated by every validated operation with the ACTUAL values used
@@ -739,6 +756,42 @@ def _run(self, argv: list[Any], *, mode: str = "blocking",
         if mode == "timed" and duration is None:
             raise HackRFValueError("timed mode requires duration")
 
+        for _busy_attempt in range(max(0, int(self.busy_retries)) + 1):
+            out, err, rc = self._run_once(resolved, mode, duration, text)
+            blob = self._text_of(err) + self._text_of(out)
+            if rc in (0, None) or not _is_busy(blob):
+                break
+            if _busy_attempt < self.busy_retries:
+                delay = self.busy_backoff * (2 ** _busy_attempt)
+                self.warn(f"device busy (previous claim not yet released); "
+                          f"retrying in {delay:.2f}s "
+                          f"({_busy_attempt + 1}/{self.busy_retries})")
+                time.sleep(delay)
+
+        # timed runs end via our SIGINT, so a non-zero rc there is expected.
+        if check and mode != "timed" and rc not in (0, None):
+            errtxt = self._text_of(err)
+            detail = errtxt.strip()
+            if not detail:
+                # some tools report failures on STDOUT with an empty stderr
+                # (Linux hackrf_info prints "No HackRF boards found." there,
+                # exit 1) -- without this fallback the error read
+                # "exited 1: " with the actual reason discarded
+                tail = [ln for ln in self._text_of(out).splitlines()
+                        if ln.strip()]
+                detail = tail[-1].strip() if tail else ""
+            raise HackRFDeviceError(
+                f"{C.TOOLS[argv[0]]} exited {rc}: {detail}")
+        return out, err, rc
+
+    @staticmethod
+    def _text_of(blob: Any) -> str:
+        if isinstance(blob, bytes):
+            return blob.decode(errors="replace")
+        return blob or ""
+
+    def _run_once(self, resolved: list[Any], mode: str,
+                  duration: float | None, text: bool) -> tuple[Any, Any, Any]:
         proc = subprocess.Popen(resolved, stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE, text=text,
                                 creationflags=_CREATION_FLAGS)
@@ -759,23 +812,7 @@ def _run(self, argv: list[Any], *, mode: str = "blocking",
             # capture would just start the next segment).
             self._sigint_and_reap(proc)
             raise
-
-        rc = proc.returncode
-        # timed runs end via our SIGINT, so a non-zero rc there is expected.
-        if check and mode != "timed" and rc not in (0, None):
-            errtxt = err.decode(errors="replace") if isinstance(err, bytes) else err
-            detail = errtxt.strip()
-            if not detail:
-                # some tools report failures on STDOUT with an empty stderr
-                # (Linux hackrf_info prints "No HackRF boards found." there,
-                # exit 1) -- without this fallback the error read
-                # "exited 1: " with the actual reason discarded
-                outtxt = out.decode(errors="replace") if isinstance(out, bytes) else (out or "")
-                tail = [ln for ln in outtxt.splitlines() if ln.strip()]
-                detail = tail[-1].strip() if tail else ""
-            raise HackRFDeviceError(
-                f"{C.TOOLS[argv[0]]} exited {rc}: {detail}")
-        return out, err, rc
+        return out, err, proc.returncode
 
     @staticmethod
     def _sigint_and_reap(proc: subprocess.Popen[Any],
@@ -790,6 +827,29 @@ def _sigint_and_reap(proc: subprocess.Popen[Any],
 
     def _stream(self, resolved: list[str], text: bool = False,
                 read_samples: int = 65536) -> Generator[Any, None, None]:
+        # Device-busy retry, stream flavor: if the child dies with the
+        # hackrf_open busy signature BEFORE anything was yielded, respawn
+        # with backoff (same policy as _run). Once data has flowed, a busy
+        # error can no longer be a stale-claim race and is raised as-is.
+        for _busy_attempt in range(max(0, int(self.busy_retries)) + 1):
+            yielded = False
+            try:
+                for item in self._stream_once(resolved, text, read_samples):
+                    yielded = True
+                    yield item
+                return
+            except HackRFDeviceError as e:
+                if (yielded or not _is_busy(str(e))
+                        or _busy_attempt >= self.busy_retries):
+                    raise
+                delay = self.busy_backoff * (2 ** _busy_attempt)
+                self.warn(f"device busy (previous claim not yet released); "
+                          f"retrying in {delay:.2f}s "
+                          f"({_busy_attempt + 1}/{self.busy_retries})")
+                time.sleep(delay)
+
+    def _stream_once(self, resolved: list[str], text: bool = False,
+                     read_samples: int = 65536) -> Generator[Any, None, None]:
         # Generator twin of _run. Launch, yield stdout as it arrives, clean up
         # on the caller breaking out (GeneratorExit) so we never leave a
         # transmitting/receiving process orphaned.
@@ -848,10 +908,39 @@ def _eat(stream: Any, sink: list[Any]) -> None:
         finally:
             if proc.poll() is None:
                 _interrupt(proc)
+                # brief window with the pipe OPEN: a well-behaved child's
+                # SIGINT handler runs (flush, marker, clean close) and it
+                # exits here
+                try:
+                    proc.wait(timeout=0.5)
+                except subprocess.TimeoutExpired:
+                    pass
+            if proc.poll() is None:
+                # Still alive: with the consumer gone, the 64 KB pipe fills
+                # in milliseconds at capture rates and the child is stuck
+                # inside write(); its SIGINT/SIGTERM handlers only set an
+                # exit flag a blocked write never returns to check. On real
+                # hardware the old interrupt->wait->terminate sequence left
+                # hackrf_transfer frozen in write(), HOLDING THE USB CLAIM
+                # until interpreter exit, and every later open failed
+                # "Resource busy". Closing our read end turns the frozen
+                # write into EPIPE/SIGPIPE so the child can die (the kernel
+                # releases the claim on any death) -- then finish the
+                # ladder, which previously stopped at an unreaped
+                # terminate().
+                try:
+                    stdout.close()
+                except OSError:
+                    pass
                 try:
                     proc.wait(timeout=2.0)
                 except subprocess.TimeoutExpired:
                     proc.terminate()
+                    try:
+                        proc.wait(timeout=1.0)
+                    except subprocess.TimeoutExpired:
+                        proc.kill()
+                        proc.wait()
         if not broke_out and proc.returncode not in (0, None):
             t.join(timeout=2.0)   # let the drain finish before reading it
             err = (b"" if not text else "").join(err_tail)

hackrfpy/tests/conftest.py

@@ -71,6 +71,23 @@ def fixtures_dir():
 EMIT_BYTES = {emit_bytes!r}
 TAIL_ON_INTERRUPT = {tail_on_interrupt!r}
 IGNORE_INTERRUPT = {ignore_interrupt!r}
+BUSY_FAILS = {busy_fails!r}
+STDOUT_FLOOD = {stdout_flood!r}
+
+if BUSY_FAILS:
+    # simulate hackrf_open()'s stale-claim race: fail with the Resource
+    # busy signature the first N invocations, then behave normally. State
+    # lives in a counter file next to the stub so it survives respawns.
+    _cf = __file__ + ".busycount"
+    try:
+        _n = int(open(_cf).read())
+    except (OSError, ValueError):
+        _n = 0
+    if _n < BUSY_FAILS:
+        open(_cf, "w").write(str(_n + 1))
+        sys.stderr.write("hackrf_open() failed: Resource busy (-1000)\\n")
+        sys.stderr.flush()
+        sys.exit(1)
 
 def _on_signal(signum, frame):
     # record WHICH signal arrived, so tests can distinguish the clean
@@ -121,6 +138,16 @@ def _on_signal(signum, frame):
     sys.stdout.write(line + "\\n")
 sys.stdout.flush()
 
+if STDOUT_FLOOD:
+    # keep writing far past pipe capacity so a consumer that stops reading
+    # leaves this child BLOCKED inside write() -- the frozen-writer state
+    # that held the USB claim on real hardware
+    _w = 0
+    while _w < STDOUT_FLOOD:
+        sys.stdout.buffer.write(b"\\x2a" * 4096)
+        _w += 4096
+    sys.stdout.buffer.flush()
+
 if IDLE:
     _idle_until = time.monotonic() + 30.0    # safety ceiling: never leak a
     while time.monotonic() < _idle_until:    # stub child on CI, even one
@@ -136,15 +163,16 @@ def _on_signal(signum, frame):
 def _write_stub(tools_dir, name, *, stdout_lines=(), stderr_lines=(),
                 stderr_flood=0, idle=False, exit_code=0, marker=None,
                 emit_bytes=None, tail_on_interrupt=None,
-                ignore_interrupt=False):
+                ignore_interrupt=False, busy_fails=0, stdout_flood=0):
     py_path = os.path.join(tools_dir, name + ".py")
     body = _STUB_TEMPLATE.format(
         marker=marker, stdout_lines=list(stdout_lines),
         stderr_lines=list(stderr_lines), stderr_flood=stderr_flood,
         idle=idle, exit_code=exit_code,
         emit_bytes=list(emit_bytes) if emit_bytes else None,
         tail_on_interrupt=tail_on_interrupt,
-        ignore_interrupt=ignore_interrupt)
+        ignore_interrupt=ignore_interrupt, busy_fails=busy_fails,
+        stdout_flood=stdout_flood)
     with open(py_path, "w") as f:
         f.write(body)
 

hackrfpy/tests/test_busy_retry.py

@@ -0,0 +1,83 @@
+#! /usr/bin/python3
+
+##--------------------------------------------------------------------\\
+#   hackrfpy  'tests/test_busy_retry.py'
+#   Device-busy retry: hackrf_open() reports "Resource busy (-1000)"
+#   when the PREVIOUS child's USB claim has not been released yet. On
+#   Linux the kernel takes a beat after a tool exits, so rapid back-to-
+#   back operations lose the race -- found on the first real-hardware
+#   Linux run (7 test failures, all this one error, all in tests that
+#   reopen the device immediately after another process used it).
+#   The library absorbs it with bounded, backed-off retries in every
+#   acquisition mode; these tests pin that with a stub that fails busy
+#   N times and then behaves.
+#
+#
+#   Author(s): Lauren Linkous
+##--------------------------------------------------------------------\\
+
+import os
+
+import pytest
+
+from hackrfpy.exceptions import HackRFDeviceError
+
+
+def _fast(h):
+    h.busy_backoff = 0.01           # keep test wall time negligible
+    return h
+
+
+def _spawn_count(h, tool="hackrf_operacake"):
+    # the counter sits next to the EXECUTED file: the extensionless launcher
+    # on POSIX, the .py on Windows (the .bat re-invokes it)
+    for suffix in ("", ".py"):
+        cf = os.path.join(h._tmp_path, tool + suffix + ".busycount")
+        if os.path.exists(cf):
+            return int(open(cf).read())
+    return 0
+
+
+# ---- blocking mode ----------------------------------------------------------
+def test_blocking_retries_through_busy(stub_device):
+    h = _fast(stub_device(operacake=dict(stdout_lines=["OK boards: none"],
+                                         busy_fails=2)))
+    out, err, rc = h.operacake("-l")
+    assert rc == 0 and "OK boards" in out
+    assert _spawn_count(h) == 2         # two busy failures were absorbed
+
+
+def test_blocking_raises_when_retries_exhausted(stub_device):
+    h = _fast(stub_device(operacake=dict(stdout_lines=["never seen"],
+                                         busy_fails=99)))
+    with pytest.raises(HackRFDeviceError, match="Resource busy"):
+        h.operacake("-l")
+    assert _spawn_count(h) == h.busy_retries + 1   # initial try + retries
+
+
+def test_busy_retries_zero_fails_immediately(stub_device):
+    h = _fast(stub_device(operacake=dict(busy_fails=1)))
+    h.busy_retries = 0
+    with pytest.raises(HackRFDeviceError, match="Resource busy"):
+        h.operacake("-l")
+    assert _spawn_count(h) == 1
+
+
+# ---- stream mode (sweep, monitor, and the persistent receiver ride on it) --
+_SWEEP_ROW = ("2026-06-18, 12:00:00.000000, 88000000, 88500000, 100000.00, "
+              "8192, -71.2, -70.1, -69.9, -72.4, -71.8")
+
+
+def test_stream_retries_before_first_row(stub_device):
+    h = _fast(stub_device(sweep=dict(stdout_lines=[_SWEEP_ROW],
+                                     busy_fails=1)))
+    rows = h.sweep_collect(88e6, 89e6, num_sweeps=1)
+    assert rows and rows[0]["hz_low"] == 88000000
+
+
+def test_receiver_open_retries_through_busy(stub_device):
+    h = _fast(stub_device(transfer=dict(emit_bytes=[0, 64] * 65536,
+                                        busy_fails=1)))
+    with h.open_receiver(100e6, 2e6) as rx:
+        iq = rx.read(1024)
+    assert len(iq) == 1024

hackrfpy/tests/test_stream_teardown.py

@@ -0,0 +1,64 @@
+#! /usr/bin/python3
+
+##--------------------------------------------------------------------\\
+#   hackrfpy  'tests/test_stream_teardown.py'
+#   The frozen-writer leak: a stream consumer that breaks out stops
+#   reading the pipe; at capture rates the 64 KB pipe fills in
+#   milliseconds and the child blocks inside write(). Signal handlers
+#   that only set an exit flag can never reach it from a blocked write,
+#   and the old teardown ended at an unreaped terminate() -- so on real
+#   hardware, hackrf_transfer stayed frozen, HOLDING THE USB CLAIM, and
+#   every later open in the process failed "Resource busy" (found on the
+#   Linux verification run: a wall of 7 failures starting immediately
+#   after the two stream-breakout tests). Teardown now gives the clean
+#   interrupt a window, then closes the read end to unblock the writer,
+#   then completes the terminate->kill ladder.
+#
+#
+#   Author(s): Lauren Linkous
+##--------------------------------------------------------------------\\
+
+import time
+
+
+def test_breakout_reaps_flooding_writer(stub_device):
+    # a deaf, flooding child: blocked in write() once we stop reading,
+    # immune to the flag-setting signals -- the hardware failure mode
+    h = stub_device(transfer=dict(stdout_flood=8_000_000,
+                                  ignore_interrupt=True))
+    gen = h._run(["transfer", "-r", "-"], mode="stream")
+    next(gen)                       # stream is live; now abandon it
+    t0 = time.monotonic()
+    gen.close()                     # GeneratorExit -> teardown under test
+    took = time.monotonic() - t0
+    assert took < 6.0, f"teardown took {took:.1f}s -- ladder not bounded"
+
+
+def test_breakout_then_immediate_reopen_works(stub_device):
+    # the user-visible contract: after abandoning one stream, the next
+    # device operation must not find a leaked child in the way
+    h = stub_device(transfer=dict(stdout_flood=8_000_000,
+                                  ignore_interrupt=True))
+    gen = h._run(["transfer", "-r", "-"], mode="stream")
+    next(gen)
+    gen.close()
+    gen2 = h._run(["transfer", "-r", "-"], mode="stream")
+    assert next(gen2)               # second stream delivers data
+    gen2.close()
+
+
+def test_breakout_clean_child_still_gets_interrupt_window(stub_device,
+                                                          tmp_path):
+    # the fix must NOT cost well-behaved children their clean exit: the
+    # SIGINT handler (marker write) runs before the pipe is closed
+    marker = str(tmp_path / "sig")
+    h = stub_device(transfer=dict(emit_bytes=[0, 64] * 4096, idle=True,
+                                  marker=marker))
+    gen = h._run(["transfer", "-r", "-"], mode="stream")
+    next(gen)
+    gen.close()
+    import os
+    deadline = time.monotonic() + 3.0
+    while time.monotonic() < deadline and not os.path.exists(marker):
+        time.sleep(0.05)
+    assert os.path.exists(marker), "clean interrupt window was lost"