diff --git a/inference/core/env.py b/inference/core/env.py index 795fbe8d92..7c8be62b08 100644 --- a/inference/core/env.py +++ b/inference/core/env.py @@ -1196,6 +1196,17 @@ os.getenv("DISABLE_GSTREAMER_VIDEO_SOURCES", "False") ) +# Opt-out from capturing the process-wide stderr (fd 2) around native video +# backend opens (FFmpeg/GStreamer). The capture is what turns an otherwise +# silent `cv2.VideoCapture` failure into a classified stream error code, so it +# is on by default; set this to True to leave fd 2 untouched at the cost of +# losing the underlying error text in `SourceConnectionError` messages. +# Note that while a capture is active, fd 2 belongs to it, so stderr written +# by the rest of the process is captured rather than logged. +DISABLE_NATIVE_STDERR_CAPTURE = str2bool( + os.getenv("DISABLE_NATIVE_STDERR_CAPTURE", "False") +) + # Instance-segmentation tensor blocks request dense (on-device) masks from the # inference_models adapter instead of the default RLE carrier. Dense masks let # GPU consumers (e.g. the mask-visualization compositor) skip the host-side diff --git a/inference/core/interfaces/camera/stream_error_classifier.py b/inference/core/interfaces/camera/stream_error_classifier.py index 9fac1870de..902fcfed19 100644 --- a/inference/core/interfaces/camera/stream_error_classifier.py +++ b/inference/core/interfaces/camera/stream_error_classifier.py @@ -2,11 +2,25 @@ from __future__ import annotations +import array import os import re +import select +import threading +import time from contextlib import contextmanager -from typing import Iterator, Optional +from typing import Iterator, List, Optional, Set, Tuple +try: # POSIX only: lets the drain be settled without writing into the pipe + import fcntl + import termios + + _FIONREAD: Optional[int] = termios.FIONREAD +except (ImportError, AttributeError): # pragma: no cover - Windows + fcntl = None # type: ignore[assignment] + _FIONREAD = None + +from inference.core.env import DISABLE_NATIVE_STDERR_CAPTURE from inference.core.interfaces.camera.exceptions import SourceConnectionError from inference.core.interfaces.camera.source_reference_sanitizer import ( redact_credentials_in_text, @@ -31,23 +45,444 @@ ) +STDERR_FILENO = 2 + +# fd 2 is a process-wide resource, so the redirection below is shared by every +# thread that happens to be opening a native video source at the same time. +# Retaining a quarter of a megabyte is far more than the handful of FFmpeg +# lines the classifier needs while bounding memory when a chatty backend spams. +_MAX_CAPTURED_BYTES = 256 * 1024 +_DRAIN_CHUNK_BYTES = 65536 +_DRAIN_POLL_SECONDS = 0.2 + +# How long a participant will wait for the drain thread to catch up before it +# settles for what has been buffered so far. This is on the reconnect path - a +# source retrying every second against a 30 s open timeout crosses it twice per +# attempt - and the text it delimits is only a hint for error classification, +# so a fraction of a second is the right order of magnitude. It is only ever +# reached when some other thread is writing to stderr continuously; normally +# the wait ends on the first notification, in well under a millisecond. +_QUIESCE_TIMEOUT_SECONDS = 0.25 +_QUIESCE_POLL_SECONDS = 0.02 + +# Capability marker for integrators that used to serialise native opens behind +# a lock of their own to work around the historical non-reentrancy of this +# helper: it is also set on `capture_process_stderr` itself, so such a +# workaround can detect a build that no longer needs it and disable itself. +NATIVE_STDERR_CAPTURE_HANDLES_CONCURRENT_OPENS = True + +_capture_state_lock = threading.Lock() +_active_capture: Optional["_SharedStderrCapture"] = None + +# Every capture that still owns a descriptor, whether or not it is the one +# currently installed: a capture is released - fd 2 restored, write end closed - +# before its drain thread has finished with the read end. Guarded by +# ``_capture_state_lock``, which is also the fork lock, so ``fork()`` cannot land +# between a drain's final close and its removal from here. The forked child has +# no drain threads, so it uses this to close what it inherited. +_captures_owning_descriptors: "Set[_SharedStderrCapture]" = set() + + +if hasattr(select, "poll"): + + def _wait_readable(read_fd: int, timeout: float) -> bool: + # poll() rather than select(): select() raises ValueError once a + # descriptor number reaches FD_SETSIZE (1024), which a server holding + # many cameras and sockets reaches easily, and a capture that cannot + # test for readability never drains its pipe. + poller = select.poll() + poller.register(read_fd, select.POLLIN | select.POLLHUP | select.POLLERR) + try: + return bool(poller.poll(timeout * 1000.0)) + except (OSError, ValueError): + return False + +else: # pragma: no cover - Windows has neither poll() nor pipe-capable select() + + def _wait_readable(read_fd: int, timeout: float) -> bool: + # Fall back to a blocking read, ended by EOF when the write ends close. + return True + + +def _pipe_unread_bytes(read_fd: Optional[int]) -> Optional[int]: + """Bytes sitting in the pipe, or None where that cannot be asked.""" + if read_fd is None or fcntl is None or _FIONREAD is None: + return None + try: + buffer = array.array("i", [0]) + fcntl.ioctl(read_fd, _FIONREAD, buffer, True) + return int(buffer[0]) + except (OSError, ValueError): + return None + + +class _SharedStderrCapture: + """One redirection of fd 2 into a pipe drained by a background thread. + + Instances are shared by every concurrent participant and reference counted: + the first entrant installs the redirection, the last leaver removes it. All + fd manipulation happens under ``_capture_state_lock`` and never spans the + body of the context manager, so a slow native open never blocks anybody. + + Descriptors live in ``_fds`` and are closed through :meth:`_close_owned`, + which pops the entry before closing. ``dict.pop`` is atomic, so each is + closed exactly once however many cleanup paths run - a closed number is + immediately reusable, and closing it twice shuts an unrelated file. + """ + + def __init__(self, read_fd: int, write_fd: int, restore_fd: int) -> None: + self._fds = {"read": read_fd, "write": write_fd, "restore": restore_fd} + self._condition = threading.Condition() + self._buffer = bytearray() + self._buffer_start = 0 # absolute offset of self._buffer[0] + self._total_bytes = 0 # absolute offset just past self._buffer[-1] + self._idle = False + self._stop = threading.Event() + self._restored = False + self._drain_entered = False + # Set only by the at-fork child handler, and only read - never written - + # by a context manager unwinding in the child. + self.discarded = False + self.ref_count = 0 + self._drain_thread = threading.Thread( + target=self._drain, + name="native-stderr-capture", + daemon=True, + ) + + def _close_owned(self, name: str) -> None: + file_descriptor = self._fds.pop(name, None) + if file_descriptor is None: + return + try: + os.close(file_descriptor) + except OSError: + pass + + @classmethod + def start(cls) -> Optional["_SharedStderrCapture"]: + """Install the redirection, or return None if fd 2 cannot be captured. + + Caller holds ``_capture_state_lock``. + """ + # Duplicate the real stderr *before* allocating the pipe. If fd 2 is + # closed, os.pipe() would be handed descriptor 2 and the capture would + # later "restore" a pipe end onto it; duplicating first fails cleanly + # instead, and the caller carries on with no capture at all. + try: + restore_fd = os.dup(STDERR_FILENO) + except OSError: + return None + capture = None + read_fd = write_fd = None + try: + read_fd, write_fd = os.pipe() + capture = cls(read_fd=read_fd, write_fd=write_fd, restore_fd=restore_fd) + _captures_owning_descriptors.add(capture) + os.dup2(write_fd, STDERR_FILENO) + capture._drain_thread.start() + return capture + except BaseException as error: + # Includes an interrupt landing inside Thread.start(), which may + # already have spawned the thread: if it is running, it owns the + # read end and closes it itself, so only take it back when it is + # certainly not. + if capture is None: + for file_descriptor in (read_fd, write_fd, restore_fd): + if file_descriptor is not None: + try: + os.close(file_descriptor) + except OSError: + pass + else: + capture.restore_stderr() + capture._stop.set() + capture._close_owned("write") + if not capture._drain_entered and not capture._drain_thread.is_alive(): + capture._close_owned("read") + _captures_owning_descriptors.discard(capture) + if isinstance(error, Exception): + # Capturing stderr is a diagnostics nicety, never a reason to + # fail an open. Interrupts still propagate. + return None + raise + + def _drain(self) -> None: + global _active_capture + self._drain_entered = True + try: + while not self._stop.is_set(): + read_fd = self._fds.get("read") + if read_fd is None: + break + if not self._wait_for_input(read_fd): + continue + try: + chunk = os.read(read_fd, _DRAIN_CHUNK_BYTES) + except (OSError, ValueError): + break + if not chunk: # EOF: every write end is gone + break + self._append(chunk) + finally: + with _capture_state_lock: + # Only matters if the loop stopped while still installed: fd 2 + # must not point at a pipe nobody reads, or a full pipe would + # block every native writer. + if _active_capture is self: + self.restore_stderr() + _active_capture = None + # Under the lock, so the last close and giving up the claim are + # one step as far as fork() is concerned. + self._close_owned("read") + _captures_owning_descriptors.discard(self) + with self._condition: + self._idle = True + self._condition.notify_all() + + def _wait_for_input(self, read_fd: int) -> bool: + with self._condition: + self._idle = True + self._condition.notify_all() + try: + return _wait_readable(read_fd, _DRAIN_POLL_SECONDS) + finally: + with self._condition: + self._idle = False + + def _append(self, chunk: bytes) -> None: + with self._condition: + self._buffer += chunk + self._total_bytes += len(chunk) + overflow = len(self._buffer) - _MAX_CAPTURED_BYTES + if overflow > 0: + del self._buffer[:overflow] + self._buffer_start += overflow + self._condition.notify_all() + + def quiesce(self) -> int: + """Absolute offset once everything written so far has been buffered. + + Nothing is written into the pipe to find this out - the pipe is asked + how much it still holds, and the drain thread reports when it is idle. + Bounded by ``_QUIESCE_TIMEOUT_SECONDS``; on timeout the current offset + is returned, which can only make a window slightly less complete. + """ + deadline = time.monotonic() + _QUIESCE_TIMEOUT_SECONDS + previous_total: Optional[int] = None + while True: + pending = _pipe_unread_bytes(self._fds.get("read")) + with self._condition: + total = self._total_bytes + if not self._drain_thread.is_alive(): + return total + if pending == 0 and self._idle: + return total + if pending is None and previous_total == total: + # No way to ask the pipe: settle for the byte counter + # having stopped moving across one poll interval. + return total + previous_total = total + remaining = deadline - time.monotonic() + if remaining <= 0: + return total + self._condition.wait(min(remaining, _QUIESCE_POLL_SECONDS)) + + def text_between(self, start: int, end: int) -> str: + with self._condition: + begin = max(start, self._buffer_start) + stop = max(begin, min(end, self._total_bytes)) + raw = bytes( + self._buffer[begin - self._buffer_start : stop - self._buffer_start] + ) + return raw.decode("utf-8", errors="replace") + + def restore_stderr(self) -> None: + """Point fd 2 back at the real stderr. + + Caller holds ``_capture_state_lock`` (the at-fork child handler is the + exception: that process has exactly one thread). One-shot, so a late + cleanup path can never dup a stale descriptor over a redirection + installed by a newer capture. + """ + if self._restored: + return + self._restored = True + restore_fd = self._fds.get("restore") + if restore_fd is not None: + try: + os.dup2(restore_fd, STDERR_FILENO) + except OSError: + pass + self._close_owned("restore") + + def shut_down(self) -> None: + """Stop the drain and hand the pipe back. Caller holds the state lock. + + The read end is left to the drain thread, the only owner that knows + when it has stopped using it; it notices the stop within one poll + interval, or at once through EOF when this closes the last write end. + """ + self._stop.set() + self._close_owned("write") + + def discard_in_forked_child(self) -> None: + """Undo an inherited redirection. Runs in a single-threaded child. + + Touches no per-instance lock: another thread could have been holding + the condition when ``fork()`` ran, and it will never be released here. + """ + self.discarded = True + self._stop.set() + self.restore_stderr() + self._close_owned("write") + self._close_owned("read") + + +def _reset_after_fork() -> None: + # The drain thread does not survive fork(), so the inherited redirection + # would silently swallow the child's stderr forever. + global _active_capture + try: + capture = _active_capture + _active_capture = None + if capture is not None: + capture.discard_in_forked_child() + # A capture released moments before the fork has already given fd 2 + # back, but its drain thread - which does not exist here - still owned + # the read end. Nothing drains it in the child, so just close it. + for orphan in list(_captures_owning_descriptors): + if orphan is not capture: + orphan.discard_in_forked_child() + _captures_owning_descriptors.clear() + finally: + _capture_state_lock.release() + + +if hasattr(os, "register_at_fork"): + os.register_at_fork( + before=_capture_state_lock.acquire, + after_in_parent=_capture_state_lock.release, + after_in_child=_reset_after_fork, + ) + + +def _reset_capture_state_for_tests() -> None: + """Tear down any capture left behind by a failed test.""" + global _active_capture + with _capture_state_lock: + capture, _active_capture = _active_capture, None + if capture is not None: + capture.restore_stderr() + capture.shut_down() + + +def _join_or_start_capture() -> Optional[Tuple["_SharedStderrCapture", int]]: + global _active_capture + with _capture_state_lock: + capture = _active_capture + if capture is None: + capture = _SharedStderrCapture.start() + if capture is None: + return None + _active_capture = capture + capture.ref_count = 1 + # Nothing can have been written yet, so no settling needed. + return capture, 0 + capture.ref_count += 1 + try: + start_offset = capture.quiesce() + except BaseException: + # An interruption between taking the reference and returning it to the + # caller would otherwise strand the capture, and fd 2 with it. + _release_capture(capture) + raise + return capture, start_offset + + +def _release_capture(capture: "_SharedStderrCapture") -> None: + global _active_capture + with _capture_state_lock: + capture.ref_count -= 1 + if capture.ref_count > 0: + return + if _active_capture is capture: + _active_capture = None + # Restoring fd 2 under the lock is what makes the next entrant dup the + # *real* stderr rather than this capture's pipe. + capture.restore_stderr() + capture.shut_down() + # Deliberately no join: the drain owns and closes the read end itself, the + # next capture allocates fresh descriptors, and making every open wait out + # a thread exit would add latency to the reconnect path for no guarantee a + # bounded join could actually give. + + @contextmanager -def capture_process_stderr() -> Iterator[list[str]]: - """Capture OS stderr (fd 2) for native backends such as FFmpeg/GStreamer.""" - read_fd, write_fd = os.pipe() - saved_stderr_fd = os.dup(2) - captured_chunks: list[str] = [] +def capture_process_stderr() -> Iterator[List[str]]: + """Capture OS stderr (fd 2) for native backends such as FFmpeg/GStreamer. + + fd 2 is process-wide, so concurrent callers share a single redirection that + is reference counted: the first entrant installs it, the last leaver takes + it down and fd 2 is restored even when the body raises. A background thread + drains the pipe continuously, so a backend that outpaces the pipe buffer + cannot block the caller, and window boundaries are found by asking the pipe + how much it still holds rather than by writing anything into it. + + Because the redirection is shared, overlapping captures may see each + other's lines. That is acceptable: the text is only a hint used to classify + a failed open (see ``extract_stream_open_error``), and it is consulted only + when the open actually failed, so a successful open is never turned into an + error by another source's output. + + Set ``DISABLE_NATIVE_STDERR_CAPTURE=True`` to leave fd 2 alone entirely. + + ``capture_process_stderr.handles_concurrent_opens`` marks a build on which + concurrent entry is safe, for integrators that previously serialised native + opens around this helper and want to drop that workaround. + + Known limitation, unchanged from before this was made concurrency-safe: + while a capture is active fd 2 is the pipe, so stderr written by the rest + of the process is captured rather than logged, and a subprocess spawned + during a capture inherits the pipe and sees EPIPE on stderr writes made + after the capture ends. + """ + captured_chunks: List[str] = [] + if DISABLE_NATIVE_STDERR_CAPTURE: + yield captured_chunks + return + joined = _join_or_start_capture() + if joined is None: + yield captured_chunks + return + capture, start_offset = joined + # Nothing between here and the `try` may touch state the `finally` has to + # undo. An asynchronous exception can still land on any bytecode boundary; + # that cannot be made airtight in pure Python, and in practice the only + # such exception, KeyboardInterrupt, is delivered to the main thread while + # these opens run on per-pipeline worker threads. try: - os.dup2(write_fd, 2) yield captured_chunks finally: - os.dup2(saved_stderr_fd, 2) - os.close(saved_stderr_fd) - os.close(write_fd) - captured = os.read(read_fd, 65536) - os.close(read_fd) - if captured: - captured_chunks.append(captured.decode("utf-8", errors="replace")) + if not capture.discarded: + try: + end_offset = capture.quiesce() + captured = capture.text_between(start_offset, end_offset) + if captured: + captured_chunks.append(captured) + finally: + # Unconditional: an interruption anywhere in the bookkeeping + # above must still give the reference back and restore fd 2. + _release_capture(capture) + # else: a fork happened inside the body. The child inherited this + # object but not its drain thread, and another thread may have been + # holding its condition when fork() ran, so touch none of it - the + # at-fork handler has already put the child's fd 2 back. + + +capture_process_stderr.handles_concurrent_opens = ( + NATIVE_STDERR_CAPTURE_HANDLES_CONCURRENT_OPENS +) def extract_stream_open_error(stderr: str) -> str: diff --git a/tests/inference/unit_tests/core/interfaces/camera/test_stream_error_classifier_concurrency.py b/tests/inference/unit_tests/core/interfaces/camera/test_stream_error_classifier_concurrency.py new file mode 100644 index 0000000000..37bb1c1e6d --- /dev/null +++ b/tests/inference/unit_tests/core/interfaces/camera/test_stream_error_classifier_concurrency.py @@ -0,0 +1,658 @@ +"""Concurrency and lifetime guarantees for the native (fd 2) stderr capture. + +Every test here runs its workers on daemon threads and joins them with a hard +timeout, or isolates them in a subprocess, so a regression that re-introduces an +unbounded block fails the suite instead of hanging it. +""" + +import os +import pathlib +import subprocess +import sys +import threading +import time +from typing import Callable, List, Optional + +import pytest + +from inference.core.interfaces.camera import stream_error_classifier, video_source +from inference.core.interfaces.camera.stream_error_classifier import ( + capture_process_stderr, +) + +# Long enough that a slow CI box never trips it, short enough that a real +# deadlock is reported quickly. +HARD_TIMEOUT_SECONDS = 20.0 + +# A child pays for a fresh interpreter and the `inference` import before it +# reaches the code under test, so it gets a larger budget - still a hard bound, +# so a wedge fails the test rather than hanging the run. +SUBPROCESS_TIMEOUT_SECONDS = 120.0 + +# Linux pipes hold 64 KiB by default; write comfortably past that. +OVER_PIPE_CAPACITY_LINES = 200 +LINE_PAYLOAD = b"x" * 1024 + b"\n" + +# The select-limit test opens descriptors until one is numbered above 1100, and +# the capture then needs a few more on top of that. +DESCRIPTORS_NEEDED_BY_SELECT_LIMIT_TEST = 1200 + + +def _descriptor_limit_allows(descriptor_count: int) -> bool: + try: + import resource + except ImportError: # pragma: no cover - Windows + return False + soft_limit, _ = resource.getrlimit(resource.RLIMIT_NOFILE) + return soft_limit == resource.RLIM_INFINITY or soft_limit >= descriptor_count + + +# Platform requirements are decided once, at collection time, on the test +# thread. pytest.skip() raised inside a worker thread never reaches pytest: it +# only kills that thread, and the test then fails instead of skipping. +requires_procfs = pytest.mark.skipif( + not os.path.isdir("/proc/self/fd"), + reason="/proc/self/fd is unavailable on this platform", +) +requires_high_descriptor_limit = pytest.mark.skipif( + not _descriptor_limit_allows(DESCRIPTORS_NEEDED_BY_SELECT_LIMIT_TEST), + reason="RLIMIT_NOFILE is too low to open a descriptor above FD_SETSIZE", +) + + +def _reset_module_state() -> None: + # Bounded: a wedged capture must not turn cleanup into a second hang. + reset = getattr(stream_error_classifier, "_reset_capture_state_for_tests", None) + if reset is None: + return + worker = threading.Thread(target=reset, daemon=True) + worker.start() + worker.join(timeout=HARD_TIMEOUT_SECONDS) + + +@pytest.fixture(autouse=True) +def restore_process_stderr(): + """Undo any fd 2 damage a failing test leaves behind.""" + real_stderr_fd = os.dup(2) + try: + yield + finally: + _reset_module_state() + os.dup2(real_stderr_fd, 2) + os.close(real_stderr_fd) + + +def _wait_for_drains_to_exit() -> None: + """Block until no drain thread is still holding its descriptors. + + A capture's teardown does not join its drain thread - the thread notices + the stop itself and closes the read end on its way out - so a descriptor + count taken the instant a capture returns can still see one on the way out. + That is a measurement artefact, not a leak, but it has to be waited out + before counting or the count is a race. + """ + + def drains_alive() -> bool: + return any( + thread.name == "native-stderr-capture" and thread.is_alive() + for thread in threading.enumerate() + ) + + deadline = time.monotonic() + HARD_TIMEOUT_SECONDS + while drains_alive() and time.monotonic() < deadline: + time.sleep(0.005) + assert not drains_alive() + + +def _call_on_daemon_thread(target: Callable[[], None]) -> dict: + """Run ``target`` where a wedge cannot take the test runner with it.""" + outcome: dict = {} + + def runner() -> None: + try: + target() + except BaseException as error: # noqa: BLE001 - the point is to catch it + outcome["error"] = error + + thread = threading.Thread(target=runner, daemon=True) + thread.start() + thread.join(timeout=HARD_TIMEOUT_SECONDS) + assert not thread.is_alive(), "the worker did not finish within the timeout" + return outcome + + +def _open_fd_count() -> int: + # Callers must be marked with `requires_procfs`; this may run on a worker + # thread, where skipping is not possible. + return len(os.listdir("/proc/self/fd")) + + +def _stderr_identity() -> tuple: + stat = os.fstat(2) + return stat.st_dev, stat.st_ino + + +def _run_daemon_threads(targets: List[Callable[[], None]]) -> None: + threads = [threading.Thread(target=target, daemon=True) for target in targets] + for thread in threads: + thread.start() + stuck = [] + for index, thread in enumerate(threads): + thread.join(timeout=HARD_TIMEOUT_SECONDS) + if thread.is_alive(): + stuck.append(index) + assert not stuck, f"worker threads did not finish within the timeout: {stuck}" + + +def _run_in_subprocess(script: str) -> subprocess.CompletedProcess: + """Run a snippet against this interpreter's imports, with a hard timeout.""" + environment = dict(os.environ) + environment["PYTHONPATH"] = os.pathsep.join(path for path in sys.path if path) + return subprocess.run( + [sys.executable, "-u", "-c", script], + capture_output=True, + text=True, + timeout=SUBPROCESS_TIMEOUT_SECONDS, + env=environment, + ) + + +def test_second_capture_entering_before_first_leaves_does_not_block() -> None: + # Forces the exact interleaving that wedges an unsynchronised + # dup2-based capture: A enters, B enters, A leaves, B leaves. + first_entered = threading.Event() + second_entered = threading.Event() + first_left = threading.Event() + second_left = threading.Event() + + def first_worker() -> None: + with capture_process_stderr(): + first_entered.set() + second_entered.wait(timeout=HARD_TIMEOUT_SECONDS) + first_left.set() + + def second_worker() -> None: + first_entered.wait(timeout=HARD_TIMEOUT_SECONDS) + with capture_process_stderr(): + second_entered.set() + first_left.wait(timeout=HARD_TIMEOUT_SECONDS) + second_left.set() + + _run_daemon_threads([first_worker, second_worker]) + + assert first_left.is_set(), "the first capture never returned from its exit path" + assert second_left.is_set(), "the second capture never returned from its exit path" + + +def test_capture_does_not_block_when_output_exceeds_pipe_capacity() -> None: + captured: List[str] = [] + + def worker() -> None: + with capture_process_stderr() as chunks: + for _ in range(OVER_PIPE_CAPACITY_LINES): + os.write(2, LINE_PAYLOAD) + captured.append("".join(chunks)) + + _run_daemon_threads([worker]) + + assert captured, "the capture body never completed" + # The buffer is bounded, so do not require every byte back - only that far + # more than one pipe's worth survived and nothing blocked. + assert captured[0].count("x") > 128 * 1024 + + +@requires_high_descriptor_limit +def test_drains_when_the_read_descriptor_is_above_the_select_limit() -> None: + # select() raises ValueError for descriptors at or above FD_SETSIZE (1024), + # which a server holding many cameras and sockets reaches; a capture that + # cannot test its pipe for readability never drains and never restores fd 2. + # Isolated in a subprocess because the failure mode is an unbounded block. + script = """ +import os, sys +from inference.core.interfaces.camera.stream_error_classifier import ( + capture_process_stderr, +) + +hogs = [] +while True: + fd = os.open(os.devnull, os.O_RDONLY) + hogs.append(fd) + if fd > 1100: + break +for fd in hogs[-4:]: + os.close(fd) + +with capture_process_stderr() as chunks: + os.write(2, b"y" * 65536) +sys.stdout.write("CAPTURED=%d" % len("".join(chunks))) +""" + result = _run_in_subprocess(script) + + assert result.returncode == 0, result.stderr[-2000:] + assert "CAPTURED=65536" in result.stdout + + +def test_stderr_fd_is_restored_after_interleaved_concurrent_captures() -> None: + before = _stderr_identity() + worker_count = 8 + barrier = threading.Barrier(worker_count, timeout=HARD_TIMEOUT_SECONDS) + inner_barrier = threading.Barrier(worker_count, timeout=HARD_TIMEOUT_SECONDS) + + def worker(index: int) -> None: + barrier.wait() + with capture_process_stderr(): + os.write(2, f"worker-{index}\n".encode("utf-8")) + # Guarantee every worker is inside the capture at the same time. + inner_barrier.wait() + + _run_daemon_threads([lambda index=i: worker(index) for i in range(worker_count)]) + + assert _stderr_identity() == before + + +def test_stderr_fd_is_restored_when_capture_bodies_raise() -> None: + before = _stderr_identity() + worker_count = 6 + barrier = threading.Barrier(worker_count, timeout=HARD_TIMEOUT_SECONDS) + inner_barrier = threading.Barrier(worker_count, timeout=HARD_TIMEOUT_SECONDS) + raised = [] + + def worker(index: int) -> None: + barrier.wait() + try: + with capture_process_stderr(): + os.write(2, f"worker-{index}\n".encode("utf-8")) + inner_barrier.wait() + if index % 2 == 0: + raise RuntimeError("boom") + except RuntimeError: + raised.append(index) + + _run_daemon_threads([lambda index=i: worker(index) for i in range(worker_count)]) + + assert sorted(raised) == [0, 2, 4] + assert _stderr_identity() == before + + +def test_interrupt_during_entry_bookkeeping_releases_the_reference( + monkeypatch, +) -> None: + holder_entered = threading.Event() + holder_release = threading.Event() + + def holder() -> None: + with capture_process_stderr(): + holder_entered.set() + holder_release.wait(timeout=HARD_TIMEOUT_SECONDS) + + holder_thread = threading.Thread(target=holder, daemon=True) + holder_thread.start() + try: + assert holder_entered.wait(timeout=HARD_TIMEOUT_SECONDS) + capture = stream_error_classifier._active_capture + assert capture is not None and capture.ref_count == 1 + + def interrupted_quiesce(self) -> int: + raise KeyboardInterrupt("interrupted during entry bookkeeping") + + monkeypatch.setattr( + stream_error_classifier._SharedStderrCapture, + "quiesce", + interrupted_quiesce, + ) + + def entrant() -> None: + with capture_process_stderr(): + pass + + outcome = _call_on_daemon_thread(entrant) + monkeypatch.undo() + assert isinstance(outcome.get("error"), KeyboardInterrupt) + + # The interrupted entrant must not have kept its reference. + assert capture.ref_count == 1 + finally: + holder_release.set() + holder_thread.join(timeout=HARD_TIMEOUT_SECONDS) + + assert not holder_thread.is_alive() + assert stream_error_classifier._active_capture is None + + +def test_interrupt_during_exit_bookkeeping_restores_stderr(monkeypatch) -> None: + before = _stderr_identity() + original = stream_error_classifier._SharedStderrCapture.quiesce + state = {"armed": False} + + def flaky_quiesce(self) -> int: + if state["armed"]: + state["armed"] = False + raise KeyboardInterrupt("interrupted during exit bookkeeping") + return original(self) + + monkeypatch.setattr( + stream_error_classifier._SharedStderrCapture, "quiesce", flaky_quiesce + ) + + def participant() -> None: + with capture_process_stderr(): + os.write(2, b"payload\n") + state["armed"] = True + + outcome = _call_on_daemon_thread(participant) + monkeypatch.undo() + assert isinstance(outcome.get("error"), KeyboardInterrupt) + + assert stream_error_classifier._active_capture is None + assert _stderr_identity() == before + + +def test_each_participant_sees_the_output_of_its_own_window() -> None: + worker_count = 4 + barrier = threading.Barrier(worker_count, timeout=HARD_TIMEOUT_SECONDS) + inner_barrier = threading.Barrier(worker_count, timeout=HARD_TIMEOUT_SECONDS) + results: List[Optional[str]] = [None] * worker_count + + def worker(index: int) -> None: + barrier.wait() + with capture_process_stderr() as chunks: + inner_barrier.wait() + os.write(2, f"marker-for-worker-{index}\n".encode("utf-8")) + results[index] = "".join(chunks) + + _run_daemon_threads([lambda index=i: worker(index) for i in range(worker_count)]) + + for index, text in enumerate(results): + assert text is not None + assert f"marker-for-worker-{index}" in text + + +def test_output_written_before_a_capture_starts_is_not_attributed_to_it() -> None: + noise_written = threading.Event() + holder_release = threading.Event() + late_capture: List[str] = [] + + def holder() -> None: + with capture_process_stderr(): + os.write(2, b"noise-from-an-earlier-open\n") + noise_written.set() + holder_release.wait(timeout=HARD_TIMEOUT_SECONDS) + + def latecomer() -> None: + noise_written.wait(timeout=HARD_TIMEOUT_SECONDS) + with capture_process_stderr() as chunks: + os.write(2, b"own-output\n") + late_capture.append("".join(chunks)) + holder_release.set() + + _run_daemon_threads([holder, latecomer]) + + assert late_capture + assert "own-output" in late_capture[0] + assert "noise-from-an-earlier-open" not in late_capture[0] + + +def test_nothing_is_written_into_the_capture_pipe() -> None: + # Windows are delimited by asking the pipe how much it still holds, never + # by injecting a marker; a marker can be truncated mid-sequence and then + # surfaces as garbage in a user-facing error message. + source = pathlib.Path(stream_error_classifier.__file__).read_text() + + assert '"write"' in source, "sanity check: the write end is still tracked" + assert "os.write(" not in source, "the module must never write to a descriptor" + assert "\\x00" not in source + + +def test_capture_is_a_no_op_when_disabled_by_env(monkeypatch) -> None: + monkeypatch.setattr( + stream_error_classifier, "DISABLE_NATIVE_STDERR_CAPTURE", True, raising=False + ) + before = _stderr_identity() + seen: List[tuple] = [] + results: List[List[str]] = [] + + def worker() -> None: + with capture_process_stderr() as chunks: + seen.append(_stderr_identity()) + os.write(2, b"") + results.append(chunks) + + outcome = _call_on_daemon_thread(worker) + + assert "error" not in outcome, outcome + assert seen == [before], "fd 2 must not be touched while the switch is off" + assert results == [[]] + assert _stderr_identity() == before + + +@requires_procfs +def test_repeated_sequential_captures_do_not_leak_file_descriptors() -> None: + counts: List[int] = [] + + def worker() -> None: + for _ in range(5): + with capture_process_stderr(): + os.write(2, b"warmup\n") + _wait_for_drains_to_exit() + counts.append(_open_fd_count()) + for _ in range(50): + with capture_process_stderr(): + os.write(2, b"line\n") + _wait_for_drains_to_exit() + counts.append(_open_fd_count()) + + _run_daemon_threads([worker]) + + assert len(counts) == 2 + # Must not grow. It may legitimately shrink: an unrelated descriptor left + # over from an earlier test can be reaped between the two samples. + assert counts[1] <= counts[0] + + +def test_capability_marker_is_exposed_on_the_public_callable() -> None: + # Integrators that serialised native opens around this helper detect a + # build that no longer needs the workaround through this attribute, so it + # has to live on the callable itself, under both names it is imported as. + assert getattr(capture_process_stderr, "handles_concurrent_opens", False) is True + assert ( + getattr(video_source.capture_process_stderr, "handles_concurrent_opens", False) + is True + ) + assert stream_error_classifier.NATIVE_STDERR_CAPTURE_HANDLES_CONCURRENT_OPENS + + +@requires_procfs +def test_interrupt_inside_thread_start_leaves_no_orphan_state(monkeypatch) -> None: + # Thread.start() may already have spawned the drain when the interrupt + # lands, so startup has to hand fd 2 back and drop its descriptors without + # closing a read end the drain now owns. + before_identity = _stderr_identity() + before_descriptors = _open_fd_count() + original_start = threading.Thread.start + + def interrupted_start(self) -> None: + if self.name == "native-stderr-capture": + raise KeyboardInterrupt("interrupted inside Thread.start") + original_start(self) + + monkeypatch.setattr(threading.Thread, "start", interrupted_start) + + def entrant() -> None: + with capture_process_stderr(): + pass + + outcome = _call_on_daemon_thread(entrant) + monkeypatch.undo() + + assert isinstance(outcome.get("error"), KeyboardInterrupt) + assert stream_error_classifier._active_capture is None + assert _stderr_identity() == before_identity + _wait_for_drains_to_exit() + assert _open_fd_count() <= before_descriptors + + +def test_closing_a_descriptor_gives_up_ownership_of_its_number() -> None: + # A closed descriptor number is immediately reusable, so a second cleanup + # path that still remembers it would close something else's file. + captured: List[stream_error_classifier._SharedStderrCapture] = [] + + def worker() -> None: + with capture_process_stderr(): + capture = stream_error_classifier._active_capture + assert capture is not None + captured.append(capture) + + _run_daemon_threads([worker]) + capture = captured[0] + _wait_for_drains_to_exit() + assert not capture._fds, "teardown should have given up every descriptor" + + # Take the freed numbers for something unrelated, then run every remaining + # cleanup path over the spent capture. + replacements = [os.open(os.devnull, os.O_RDONLY) for _ in range(4)] + try: + capture.restore_stderr() + capture.shut_down() + capture.discard_in_forked_child() + for file_descriptor in replacements: + os.fstat(file_descriptor) # raises if the capture closed it again + finally: + for file_descriptor in replacements: + os.close(file_descriptor) + + +def test_capture_is_a_no_op_when_stderr_is_already_closed() -> None: + # Duplicating the real stderr before allocating the pipe is what stops + # os.pipe() being handed descriptor 2 and later "restored" onto it. + # + # The assertion is "fd 2 is not left as a pipe" rather than "fd 2 is still + # closed": a background thread of the imported package occasionally opens a + # socket, which lands on the just-freed descriptor 2 and makes the stricter + # form flaky (measured about once in sixty runs). Capturing onto whatever + # genuinely occupies fd 2 is correct behaviour; leaving a pipe there is the + # defect, and that is what this pins down. (The older single-read + # implementation ended up closing descriptor 2 rather than leaving a pipe + # on it, so this is a regression test against the drained design.) + script = """ +import os, stat, sys +from inference.core.interfaces.camera.stream_error_classifier import ( + capture_process_stderr, +) + +os.close(2) +with capture_process_stderr() as chunks: + pass +try: + mode = os.fstat(2).st_mode +except OSError: + sys.stdout.write("FD2_CLOSED chunks=%r" % (chunks,)) +else: + kind = "FIFO" if stat.S_ISFIFO(mode) else "OTHER" + sys.stdout.write("FD2_OPEN_%s chunks=%r" % (kind, chunks)) +""" + result = _run_in_subprocess(script) + + assert result.returncode == 0, result.stderr[-2000:] + assert "FD2_OPEN_FIFO" not in result.stdout, result.stdout + assert "chunks=[]" in result.stdout + + +def test_forked_child_leaves_an_inherited_capture_without_deadlocking() -> None: + # fork() while another thread holds the capture's condition: the child + # inherits the locked condition but not the thread that would release it, + # so unwinding the context there must touch none of the capture's locks. + script = """ +import os, sys, threading, time +from inference.core.interfaces.camera import stream_error_classifier +from inference.core.interfaces.camera.stream_error_classifier import ( + capture_process_stderr, +) + +holding = threading.Event() +release = threading.Event() + +def hold(capture): + with capture._condition: + holding.set() + release.wait(20) + +child = False +with capture_process_stderr(): + capture = stream_error_classifier._active_capture + threading.Thread(target=hold, args=(capture,), daemon=True).start() + holding.wait(10) + pid = os.fork() + child = pid == 0 + if not child: + # The child's copy of the condition stays locked forever; let the + # parent's holder go so the parent can unwind normally. + release.set() +# both processes unwind the context here +if child: + os._exit(0) + +deadline = time.monotonic() + 10 +status = None +while time.monotonic() < deadline: + finished, status = os.waitpid(pid, os.WNOHANG) + if finished: + break + time.sleep(0.01) +if status is None or not finished: + os.kill(pid, 9) + sys.stdout.write("CHILD_STUCK") +else: + sys.stdout.write("CHILD_EXIT=%d" % status) +""" + result = _run_in_subprocess(script) + + assert result.returncode == 0, result.stderr[-2000:] + assert "CHILD_EXIT=0" in result.stdout + + +@requires_procfs +def test_fork_closes_the_read_end_of_a_capture_that_is_still_winding_down() -> None: + # A released capture has already given fd 2 back, but its drain thread is + # still inside its poll interval and still owns the read end. That thread + # does not exist in a forked child, so the child must close the descriptor + # rather than carry it for the rest of its life. + script = """ +import os, stat, sys, time +from inference.core.interfaces.camera.stream_error_classifier import ( + capture_process_stderr, +) + +def pipe_fds(): + found = set() + for name in os.listdir("/proc/self/fd"): + try: + file_descriptor = int(name) + if stat.S_ISFIFO(os.fstat(file_descriptor).st_mode): + found.add(file_descriptor) + except (OSError, ValueError): + pass + return found + +before = pipe_fds() +with capture_process_stderr(): + # A duplicate of the redirected fd 2 is a write end, so the pipe cannot + # reach EOF and the drain is still polling when the fork happens. + held_writer = os.dup(2) + +pid = os.fork() +if pid == 0: + os.close(held_writer) + time.sleep(0.4) + sys.stdout.write("CHILD_EXTRA_PIPE_FDS=%d " % len(pipe_fds() - before)) + sys.stdout.flush() + os._exit(0) + +os.close(held_writer) +_, status = os.waitpid(pid, 0) +sys.stdout.write("PARENT_STATUS=%d" % status) +""" + result = _run_in_subprocess(script) + + assert result.returncode == 0, result.stderr[-2000:] + assert "CHILD_EXTRA_PIPE_FDS=0" in result.stdout, result.stdout + assert "PARENT_STATUS=0" in result.stdout, result.stdout