From e7f1aac59fbdca6c7529ed2c2cd6066aa51e5643 Mon Sep 17 00:00:00 2001 From: Shantanu Bala Date: Thu, 17 Sep 2026 07:05:13 -0400 Subject: [PATCH] Serialise stderr capture around video source opens capture_process_stderr redirects process-wide fd 2 into a pipe for the duration of each cv2.VideoCapture open. Sources open on their own threads, so two opens starting together interleave their redirects: the first to finish blocks forever reading a pipe whose write end the second still holds, and fd 2 is finally restored to that pipe instead of real stderr. Take a process-wide lock around the capture and drain the pipe on a thread so output larger than the pipe buffer cannot block the writer either. --- .../camera/stream_error_classifier.py | 48 ++++++++----- .../camera/test_stream_error_classifier.py | 67 ++++++++++++++++++- 2 files changed, 99 insertions(+), 16 deletions(-) diff --git a/inference/core/interfaces/camera/stream_error_classifier.py b/inference/core/interfaces/camera/stream_error_classifier.py index 9fac1870de..94a88c1185 100644 --- a/inference/core/interfaces/camera/stream_error_classifier.py +++ b/inference/core/interfaces/camera/stream_error_classifier.py @@ -4,6 +4,7 @@ import os import re +import threading from contextlib import contextmanager from typing import Iterator, Optional @@ -31,23 +32,40 @@ ) +_capture_stderr_lock = threading.Lock() + + +def _drain_pipe(read_fd: int, captured_chunks: list[str]) -> None: + with os.fdopen(read_fd, "rb", buffering=0) as pipe: + for chunk in iter(lambda: pipe.read(65536), b""): + captured_chunks.append(chunk.decode("utf-8", errors="replace")) + + @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] = [] - 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")) + """Capture OS stderr (fd 2) for native backends such as FFmpeg/GStreamer. + + fd 2 is process-wide, so captures are serialised: interleaved redirects + from concurrent opens would leave one thread blocked reading a pipe whose + write end another still holds. The pipe is drained on a thread so output + larger than the pipe buffer cannot block the writer either. + """ + with _capture_stderr_lock: + read_fd, write_fd = os.pipe() + captured_chunks: list[str] = [] + reader = threading.Thread( + target=_drain_pipe, args=(read_fd, captured_chunks), daemon=True + ) + reader.start() + saved_stderr_fd = os.dup(2) + 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) + reader.join() def extract_stream_open_error(stderr: str) -> str: diff --git a/tests/inference/unit_tests/core/interfaces/camera/test_stream_error_classifier.py b/tests/inference/unit_tests/core/interfaces/camera/test_stream_error_classifier.py index 4dae566e86..a0358672b6 100644 --- a/tests/inference/unit_tests/core/interfaces/camera/test_stream_error_classifier.py +++ b/tests/inference/unit_tests/core/interfaces/camera/test_stream_error_classifier.py @@ -1,6 +1,12 @@ +from __future__ import annotations + +import os +import threading + from inference.core.interfaces.camera.exceptions import SourceConnectionError from inference.core.interfaces.camera.stream_error_classifier import ( build_source_connection_error_message, + capture_process_stderr, classify_stream_error_message, extract_stream_open_error, wrap_source_connection_error, @@ -66,7 +72,9 @@ def test_classify_default() -> None: def test_classify_empty_message() -> None: assert classify_stream_error_message("") == StreamErrorCode.STREAM_CONNECTION_FAILED - assert classify_stream_error_message(None) == StreamErrorCode.STREAM_CONNECTION_FAILED + assert ( + classify_stream_error_message(None) == StreamErrorCode.STREAM_CONNECTION_FAILED + ) def test_classify_status_code_false_positives() -> None: @@ -162,3 +170,60 @@ def test_wrap_source_connection_error_classifies_underlying_ffmpeg_error() -> No source_reference="rtsp://camera.example/stream", ) assert error.code == StreamErrorCode.STREAM_TLS_HANDSHAKE + + +def _run_all(threads: list[threading.Thread], timeout: float = 5.0) -> None: + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=timeout) + assert not any(thread.is_alive() for thread in threads) + + +def test_capture_process_stderr_serialises_interleaved_opens() -> None: + before = os.fstat(2) + first_inside, second_inside, first_done = ( + threading.Event(), + threading.Event(), + threading.Event(), + ) + captured: dict[str, str] = {} + + def first() -> None: + with capture_process_stderr() as chunks: + os.write(2, b"first: connection refused\n") + first_inside.set() + second_inside.wait(timeout=0.5) + first_done.set() + captured["first"] = "".join(chunks) + + def second() -> None: + first_inside.wait() + with capture_process_stderr() as chunks: + os.write(2, b"second: 401 Unauthorized\n") + second_inside.set() + first_done.wait(timeout=2) + captured["second"] = "".join(chunks) + + _run_all([threading.Thread(target=first), threading.Thread(target=second)]) + + assert captured == { + "first": "first: connection refused\n", + "second": "second: 401 Unauthorized\n", + } + after = os.fstat(2) + assert (after.st_dev, after.st_ino) == (before.st_dev, before.st_ino) + + +def test_capture_process_stderr_drains_output_larger_than_pipe_buffer() -> None: + payload = b"x" * (256 * 1024) + captured: list[str] = [] + + def writer() -> None: + with capture_process_stderr() as chunks: + os.write(2, payload) + captured.extend(chunks) + + _run_all([threading.Thread(target=writer)]) + + assert len("".join(captured)) == len(payload)