diff --git a/src/plugins/sensing/AGENTS.md b/src/plugins/sensing/AGENTS.md index a8abfd9bc2..5e5277c0d2 100644 --- a/src/plugins/sensing/AGENTS.md +++ b/src/plugins/sensing/AGENTS.md @@ -21,3 +21,12 @@ Container preview must carry the desktop display and authorization into the imag Support SSH launch with a local DISPLAY: when XAUTHORITY is unset, check the current user's GDM authority file as well as ~/.Xauthority. + +A frame lease keeps its capture backend alive, so a held frame is an ownership +cycle. Every cleanup path, including a failed start, +must join the worker threads and drop the held frames, not only normal stop. +A camera session that will not stop ends the process; do not retry in-process. +Only a camera's owner thread calls its buffer stream; a frame lease posts its +buffer to that owner's mailbox and makes no SDK call. +Check an SDK call's return type before testing it for success: Argus +stopRepeat returns a capture-id range, not a Status. diff --git a/src/plugins/sensing/CMakeLists.txt b/src/plugins/sensing/CMakeLists.txt index a86a6057e9..d0ea637282 100644 --- a/src/plugins/sensing/CMakeLists.txt +++ b/src/plugins/sensing/CMakeLists.txt @@ -29,51 +29,55 @@ set(_tegra_lib_dirs "/usr/lib/aarch64-linux-gnu/nvidia" "/usr/lib/aarch64-linux- # ============================================================================== # SIPL # ============================================================================== -# Headers ship in Jetson_SIPL_API_R_aarch64.tbz2, published alongside the -# L4T release. Not in any apt pool, so setup_container.sh fetches it by URL. -if(NOT SIPL_API_DIR) - find_path(SIPL_API_DIR sipl/include/NvSIPLCamera.hpp - PATHS "/usr/src/jetson_sipl_api" - "/opt/nvidia/jetson_sipl_api" - "$ENV{SIPL_API_DIR}" - ) -endif() - -if(NOT SIPL_API_DIR) - message(FATAL_ERROR - "================================================================================\n" - "SENSING camera plugin requires the Jetson SIPL API headers.\n" - "\n" - "Install the SIPL API package for your L4T release and point CMake at it:\n" - " cmake -B build -DSIPL_API_DIR=/path/to/jetson_sipl_api\n" - "\n" - "Or disable this plugin:\n" - " cmake -B build -DBUILD_PLUGIN_SENSING=OFF\n" - "================================================================================\n" - ) -endif() +# Off for an Argus-only rig, which needs neither the SIPL headers nor its libraries. +option(SENSING_WITH_SIPL "Build the SIPL capture backend" ON) +if(SENSING_WITH_SIPL) + # Headers ship in Jetson_SIPL_API_R_aarch64.tbz2, published alongside the + # L4T release. Not in any apt pool, so setup_container.sh fetches it by URL. + if(NOT SIPL_API_DIR) + find_path(SIPL_API_DIR sipl/include/NvSIPLCamera.hpp + PATHS "/usr/src/jetson_sipl_api" + "/opt/nvidia/jetson_sipl_api" + "$ENV{SIPL_API_DIR}" + ) + endif() -set(SIPL_INCLUDE_DIRS - "${SIPL_API_DIR}/sipl/include" - "${SIPL_API_DIR}/sipl/include/query/include" - "${SIPL_API_DIR}/sipl/include/nvsci" -) -message(STATUS "SENSING plugin: SIPL=${SIPL_API_DIR}") - -# L4T ships no .so dev symlinks for these, hence the explicit sonames. -foreach(_lib nvsipl nvsipl_query nvscibuf nvscisync nvbufsurface_nvsci) - string(TOUPPER "${_lib}" _upper) - find_library(${_upper}_LIBRARY - NAMES ${_lib} lib${_lib}.so.1 lib${_lib}.so.1.0.0 - PATHS ${_tegra_lib_dirs} - ) - if(NOT ${_upper}_LIBRARY) + if(NOT SIPL_API_DIR) message(FATAL_ERROR - "lib${_lib} not found. It is supplied by the L4T BSP and, in a container, by the " - "NVIDIA container runtime. Run src/plugins/sensing/scripts/verify.sh to see what is missing.") + "================================================================================\n" + "SENSING camera plugin requires the Jetson SIPL API headers.\n" + "\n" + "Install the SIPL API package for your L4T release and point CMake at it:\n" + " cmake -B build -DSIPL_API_DIR=/path/to/jetson_sipl_api\n" + "\n" + "Or disable this plugin:\n" + " cmake -B build -DBUILD_PLUGIN_SENSING=OFF\n" + "================================================================================\n" + ) endif() - list(APPEND SIPL_LIBRARIES ${${_upper}_LIBRARY}) -endforeach() + + set(SIPL_INCLUDE_DIRS + "${SIPL_API_DIR}/sipl/include" + "${SIPL_API_DIR}/sipl/include/query/include" + "${SIPL_API_DIR}/sipl/include/nvsci" + ) + message(STATUS "SENSING plugin: SIPL=${SIPL_API_DIR}") + + # L4T ships no .so dev symlinks for these, hence the explicit sonames. + foreach(_lib nvsipl nvsipl_query nvscibuf nvscisync nvbufsurface_nvsci) + string(TOUPPER "${_lib}" _upper) + find_library(${_upper}_LIBRARY + NAMES ${_lib} lib${_lib}.so.1 lib${_lib}.so.1.0.0 + PATHS ${_tegra_lib_dirs} + ) + if(NOT ${_upper}_LIBRARY) + message(FATAL_ERROR + "lib${_lib} not found. It is supplied by the L4T BSP and, in a container, by the " + "NVIDIA container runtime. Run src/plugins/sensing/scripts/verify.sh to see what is missing.") + endif() + list(APPEND SIPL_LIBRARIES ${${_upper}_LIBRARY}) + endforeach() +endif() # EGL stays on the capture path: NvBufSurfaceMapEglImage is how a SIPL # NvSciBufObj becomes a CUeglFrame. Interop path B in the design notes would @@ -85,10 +89,9 @@ if(NOT EGL_INCLUDE_DIRS OR NOT EGL_LIBRARIES) endif() # ============================================================================== -# Jetson V4L2 encoder +# Jetson Multimedia API tree # ============================================================================== -# Jetson has no libnvidia-encode (the Video Codec SDK is dGPU-only), so H.264 -# comes from the V4L2 M2M engine via NvVideoEncoder, vendored from the Multimedia API (see below). +# Found before the Argus block below, whose header search is relative to it. if(NOT JETSON_MMAPI_DIR) find_path(JETSON_MMAPI_DIR include/nvbufsurface.h PATHS "/usr/src/jetson_multimedia_api" @@ -106,6 +109,35 @@ if(NOT JETSON_MMAPI_DIR) endif() message(STATUS "SENSING plugin: Jetson Multimedia API=${JETSON_MMAPI_DIR}") +# ============================================================================== +# Argus (optional) +# ============================================================================== +# The Argus backend captures NV12 into DMA-BUFs the hardware encoder reads. It needs the Argus headers from the +# Multimedia API and libnvargus; none of this is built or run by default. +option(SENSING_WITH_ARGUS "Build the Argus capture backend (not built or run by the author)" OFF) +# Argus normally runs through nvargus-daemon, and its client library is libnvargus_socketclient. The single-process +# libnvargus does not connect to the daemon, so it is an explicit choice. +option(SENSING_ARGUS_SINGLE_PROCESS "Link single-process libnvargus instead of the nvargus-daemon client" OFF) +if(SENSING_WITH_ARGUS) + if(SENSING_ARGUS_SINGLE_PROCESS) + set(_argus_lib_name nvargus) + else() + set(_argus_lib_name nvargus_socketclient) + endif() + find_path(ARGUS_INCLUDE_DIR Argus/Argus.h HINTS "${JETSON_MMAPI_DIR}/argus/include") + find_library(ARGUS_LIBRARY NAMES ${_argus_lib_name} PATHS ${_tegra_lib_dirs}) + if(NOT ARGUS_INCLUDE_DIR OR NOT ARGUS_LIBRARY) + message(FATAL_ERROR + "SENSING_WITH_ARGUS needs the Argus headers (/argus/include) and lib${_argus_lib_name}.") + endif() + message(STATUS "SENSING plugin: Argus=${ARGUS_INCLUDE_DIR} (${ARGUS_LIBRARY})") +endif() + +# ============================================================================== +# Jetson V4L2 encoder +# ============================================================================== +# Jetson has no libnvidia-encode (the Video Codec SDK is dGPU-only), so H.264 +# comes from the V4L2 M2M engine via NvVideoEncoder, vendored from the Multimedia API (see below). # NVIDIA's encoder classes are vendored (BSD-3-Clause; third_party/jetson_multimedia_api/README.md), so they build the # same on every JetPack and carry no stdout debug prints. The installed Multimedia API tree supplies only what is not # vendored: nvbufsurface.h, libv4l2.h, v4l2_nv_extensions.h and the Argus headers. @@ -153,19 +185,36 @@ message(STATUS "SENSING plugin: V4L2=${V4L2_LIBRARY}") set(_livekit_local ON) if(NOT LIVEKIT_SDK_DIR) set(_livekit_local OFF) + # The prebuilt links against the glibc of the Ubuntu release it was built on: the 24.04 build does not load on + # JetPack 6 (Ubuntu 22.04). Pick the release matching the host; LIVEKIT_SDK_UBUNTU overrides the detection. + if(NOT LIVEKIT_SDK_UBUNTU) + set(LIVEKIT_SDK_UBUNTU "24.04") + if(EXISTS /etc/os-release) + file(STRINGS /etc/os-release _os_version_id REGEX "^VERSION_ID=") + if(_os_version_id MATCHES "\"?22\\.04\"?$") + set(LIVEKIT_SDK_UBUNTU "22.04") + endif() + endif() + endif() if(CMAKE_SYSTEM_PROCESSOR MATCHES "aarch64|arm64") set(_livekit_arch "arm64") - set(_livekit_hash "e3a97f71943c0bf9fa4ebd70927ca2906b31dcd6c14b42312b1225b4b868e4ad") + set(_livekit_hash_24.04 "e3a97f71943c0bf9fa4ebd70927ca2906b31dcd6c14b42312b1225b4b868e4ad") + set(_livekit_hash_22.04 "7a698bbd8f61f019744fd99b3a5d2b6c6118787fd1f81140c9a6fc8732fc87e0") elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "x86_64|AMD64") set(_livekit_arch "x64") - set(_livekit_hash "5e0feb2e519c6ff131dd2d1c45149115057ae4e21a324fdb249e1789d5614e7b") + set(_livekit_hash_24.04 "5e0feb2e519c6ff131dd2d1c45149115057ae4e21a324fdb249e1789d5614e7b") + set(_livekit_hash_22.04 "9e568ee6b2bb3c7b0d07faba694930767096a14d0e09422ba01f34077ffb03cb") else() message(FATAL_ERROR "No prebuilt LiveKit SDK for ${CMAKE_SYSTEM_PROCESSOR}; set LIVEKIT_SDK_DIR.") endif() + if(NOT DEFINED _livekit_hash_${LIVEKIT_SDK_UBUNTU}) + message(FATAL_ERROR "No prebuilt LiveKit SDK for Ubuntu ${LIVEKIT_SDK_UBUNTU}; set LIVEKIT_SDK_DIR.") + endif() + set(_livekit_hash "${_livekit_hash_${LIVEKIT_SDK_UBUNTU}}") include(FetchContent) FetchContent_Declare( livekit_sdk - URL https://github.com/livekit/client-sdk-cpp/releases/download/v1.12.4/livekit-sdk-ubuntu-24.04-${_livekit_arch}-1.12.4.tar.gz + URL https://github.com/livekit/client-sdk-cpp/releases/download/v1.12.4/livekit-sdk-ubuntu-${LIVEKIT_SDK_UBUNTU}-${_livekit_arch}-1.12.4.tar.gz URL_HASH SHA256=${_livekit_hash} ) FetchContent_MakeAvailable(livekit_sdk) diff --git a/src/plugins/sensing/README.md b/src/plugins/sensing/README.md index 22570cd87b..1906b555ab 100644 --- a/src/plugins/sensing/README.md +++ b/src/plugins/sensing/README.md @@ -47,6 +47,10 @@ hardware encoder and publishes one LiveKit video track per camera. The [SENSING user guide](https://nvidia.github.io/IsaacCapture/main/device/camera/sensing.html) has the build and run steps. -`--camera-backend=auto|sipl|argus` selects the capture backend once at startup. Only SIPL is built in today: `argus` fails -with a message saying so, and `auto` picks the only matching backend (and, with several, names the override instead of -guessing). +`--camera-backend=auto|sipl|argus` selects the capture backend once at startup: an explicit choice must match or fail +with its own reason, and `auto` picks the only matching backend (with several it names the override instead of +guessing). SIPL is built by default (`-DSENSING_WITH_SIPL=OFF` for an Argus-only rig). The Argus backend +(`-DSENSING_WITH_ARGUS=ON`, Argus device index with `--add-stream=sensor=N,...`, `--argus-mode=K`) captures NV12 into +DMA-BUFs that the hardware encoder reads without a copy. **The Argus code has never been built or run: there is no +Argus SDK or Argus camera available to the author.** Argus DMA-BUF input cannot be combined with +`--livekit-scale-div` above 1. An Argus-only build (`-DSENSING_WITH_SIPL=OFF -DSENSING_WITH_ARGUS=ON`) also builds on JetPack 6.2 / L4T R36.4.3. diff --git a/src/plugins/sensing/app/main.cpp b/src/plugins/sensing/app/main.cpp index ae4e461cd9..19a4dee0ad 100644 --- a/src/plugins/sensing/app/main.cpp +++ b/src/plugins/sensing/app/main.cpp @@ -2,8 +2,15 @@ // SPDX-License-Identifier: Apache-2.0 #include +#include #include -#include + +#ifdef SENSING_HAS_SIPL +# include +#endif +#ifdef SENSING_HAS_ARGUS +# include +#endif #include #include @@ -72,7 +79,7 @@ static StreamConfig parse_stream_arg(const std::string& arg) } if (!has_sensor || cfg.livekit_track.empty()) - throw std::runtime_error("--add-stream requires sensor=,livekit="); + throw std::runtime_error("--add-stream requires sensor=,livekit="); return cfg; } @@ -119,14 +126,15 @@ static void print_usage(const char* program_name) { std::cout << "Usage: " << program_name << " --livekit-url=URL --livekit-token-file=PATH --add-stream=... [options]\n" << "\n" - << "Publishes each SENSING sensor as a LiveKit camera track: SIPL capture, hardware H.264,\n" + << "Publishes each SENSING sensor as a LiveKit camera track: SIPL or Argus capture, hardware H.264,\n" << "sent as it is.\n" << "\n" << "Streams (repeat for each sensor):\n" - << " --add-stream=sensor=N,livekit=NAME SIPL pipeline index N, published as track NAME\n" - << " --list-sensors print the pipelines, then exit\n" + << " --add-stream=sensor=N,livekit=NAME camera N (SIPL or Argus device), published as track NAME\n" + << " --list-sensors print the cameras, then exit\n" << " --camera-backend=auto|sipl|argus auto picks the one backend that matches (default);\n" - << " only sipl is built in\n" + << " (a backend must also be built in)\n" + << " --argus-mode=N Argus sensor mode index (default: 0)\n" << "\n" << "Platform:\n" << " --platform-config=PATH platform JSON (default: configs/shw5g.json beside this\n" @@ -204,6 +212,10 @@ try { list_sensors = true; } + else if (arg.find("--argus-mode=") == 0) + { + camera_config.argus_sensor_mode = static_cast(std::stoul(arg.substr(13))); + } else if (arg.find("--camera-backend=") == 0) { const auto parsed = parse_camera_backend(arg.substr(17)); @@ -266,50 +278,92 @@ try } } - if (camera_config.platform_config_json.empty()) - { - camera_config.platform_config_json = default_platform_config(); - } - if (camera_config.platform_config_json.empty()) - { - std::cerr << "Error: no platform config. configs/shw5g.json was not found beside this\n" - "binary, so pass --platform-config=PATH. SIPL cannot enumerate sensors\n" - "without one." - << std::endl; - print_usage(argv[0]); - return 1; - } + // This process captures and never renders, but EGL is on its critical path: + // NvBufSurfaceMapEglImage() is how a SIPL NvSciBufObj reaches CUDA, and + // libnvbufsurface resolves its own display via + // eglGetDisplay(EGL_DEFAULT_DISPLAY). GLVND hands out Mesa's EGL rather than + // Tegra's whenever DISPLAY names an X server Tegra EGL cannot drive (Xvfb, + // X11 forwarding), and that choice cannot be made per-call -- DISPLAY has to + // be gone from the environment before the first EGL call, including the probes below. + ::unsetenv("DISPLAY"); - // Query is hardware-free -- it only parses the driver database and the JSON -- so --list-sensors - // works on a rig whose cameras are unplugged, and the poll interval below can be derived before - // anything is opened. - // Only SIPL is built in, so it is the only probe. Its query parses the configuration; it does not prove - // that the cameras are present, so a matching rig is a configuration that names sensors. - std::vector sensors; + // Each backend reports whether it can open the configured rig; probing parses configuration or enumerates + // devices and opens no camera, so --list-sensors works on a rig whose cameras are unplugged. + std::vector camera_ids; + for (const auto& entry : stream_map) + camera_ids.push_back(entry.first); + + std::vector sipl_cameras; + std::vector argus_cameras; BackendProbe sipl_probe; sipl_probe.backend = CameraBackend::Sipl; + BackendProbe argus_probe; + argus_probe.backend = CameraBackend::Argus; + +#ifdef SENSING_HAS_SIPL sipl_probe.available = true; - try + // An explicit --camera-backend names the one backend to use: the other SDK is not initialised just to be probed. + if (requested_backend != CameraBackend::Argus) { - sensors = SiplCamera::query( - camera_config.platform_config_json, camera_config.platform_config_name, camera_config.link_masks); - sipl_probe.matches = !sensors.empty(); - if (!sipl_probe.matches) - sipl_probe.detail = "the platform configuration names no sensors"; + if (camera_config.platform_config_json.empty()) + camera_config.platform_config_json = default_platform_config(); + if (camera_config.platform_config_json.empty()) + { + sipl_probe.detail = + "no platform config: configs/shw5g.json was not found beside this binary; pass " + "--platform-config=PATH"; + } + else + { + // The query parses the configuration; it does not prove that the cameras are present. + try + { + for (const auto& s : SiplCamera::query(camera_config.platform_config_json, + camera_config.platform_config_name, camera_config.link_masks)) + sipl_cameras.push_back({ s.id, s.name, s.width, s.height, s.fps }); + sipl_probe.matches = !sipl_cameras.empty(); + if (!sipl_probe.matches) + sipl_probe.detail = "the platform configuration names no sensors"; + } + catch (const std::exception& e) + { + sipl_probe.detail = e.what(); + } + } } - catch (const std::exception& e) +#else + sipl_probe.detail = "this build has no SIPL support"; +#endif + +#ifdef SENSING_HAS_ARGUS + argus_probe.available = true; + if (requested_backend != CameraBackend::Sipl) { - sipl_probe.detail = e.what(); + try + { + argus_cameras = ArgusCapture::query(camera_config.argus_sensor_mode); + // Every requested camera must exist; with none requested (--list-sensors) any camera counts. + match_requested_cameras(argus_cameras, camera_ids, "Argus", argus_probe); + } + catch (const std::exception& e) + { + argus_probe.detail = e.what(); + } } - const CameraBackend backend = select_camera_backend(requested_backend, { sipl_probe }); +#else + argus_probe.detail = "this build has no Argus support (configure with -DSENSING_WITH_ARGUS=ON)"; +#endif + + const CameraBackend backend = select_camera_backend(requested_backend, { sipl_probe, argus_probe }); + const std::vector& cameras = backend == CameraBackend::Argus ? argus_cameras : sipl_cameras; std::cout << "Camera backend: " << camera_backend_name(backend) << std::endl; if (list_sensors) { - std::cout << camera_config.platform_config_name << ": " << sensors.size() << " sensor(s)\n"; - for (const auto& s : sensors) + std::cout << cameras.size() << " camera(s)\n"; + for (const auto& c : cameras) { - std::cout << " sensor=" << s.id << " " << s.name << " " << s.width << "x" << s.height << " @ " << s.fps + std::cout << " sensor=" << c.id << " " << c.name << " " << c.width << "x" << c.height << " @ " << c.fps << " fps\n"; } return 0; @@ -346,23 +400,14 @@ try if (camera_config.livekit.token.empty()) throw std::runtime_error("--livekit-token-file " + livekit_token_file + " is empty"); - // This process captures and never renders, but EGL is on its critical path: - // NvBufSurfaceMapEglImage() is how a SIPL NvSciBufObj reaches CUDA, and - // libnvbufsurface resolves its own display via - // eglGetDisplay(EGL_DEFAULT_DISPLAY). GLVND hands out Mesa's EGL rather than - // Tegra's whenever DISPLAY names an X server Tegra EGL cannot drive (Xvfb, - // X11 forwarding), and that choice cannot be made per-call -- DISPLAY has to - // be gone from the environment before the first EGL call. - ::unsetenv("DISPLAY"); - std::signal(SIGINT, signal_handler); std::signal(SIGTERM, signal_handler); std::cout << "============================================================" << std::endl; - std::cout << "SENSING Camera Plugin Starting (SIPL)" << std::endl; + std::cout << "SENSING Camera Plugin Starting (" << camera_backend_name(backend) << ")" << std::endl; std::cout << "============================================================" << std::endl; - SensingCamera camera(camera_config, stream_configs); + SensingCamera camera(camera_config, make_capture(backend, camera_config, camera_ids), stream_configs); std::cout << "------------------------------------------------------------" << std::endl; std::cout << "Running capture loop. Press Ctrl+C to stop." << std::endl; @@ -370,13 +415,13 @@ try constexpr auto stats_interval = std::chrono::seconds(5); auto last_stats_time = std::chrono::steady_clock::now(); - // SiplCamera::latest() is a non-blocking mailbox read, so poll at roughly - // twice the fastest sensor's frame rate rather than spinning a core flat - // out. Capture itself runs on SIPL's own threads and is unaffected. + // acquire_latest() is non-blocking, so poll at roughly twice the fastest sensor's frame rate rather than + // spinning a core flat out. SIPL captures on its own threads and is unaffected; with Argus this loop also + // returns buffers to the camera, so a stalled loop stalls capture. double fastest_fps = 30.0; - for (const auto& s : sensors) + for (const auto& c : cameras) { - fastest_fps = std::max(fastest_fps, s.fps); + fastest_fps = std::max(fastest_fps, c.fps); } const auto poll_interval = std::chrono::microseconds(static_cast(500'000.0 / fastest_fps)); diff --git a/src/plugins/sensing/core/CMakeLists.txt b/src/plugins/sensing/core/CMakeLists.txt index ed8badb1cb..2f1998a3da 100644 --- a/src/plugins/sensing/core/CMakeLists.txt +++ b/src/plugins/sensing/core/CMakeLists.txt @@ -4,16 +4,17 @@ # SIPL capture, colour conversion, the hardware encoder and the LiveKit track output. add_library(sensing_core STATIC sensing_camera.cpp - sipl_camera.cpp - sipl_capture.cpp camera_backend.cpp + capture_factory.cpp jetson_encoder.cpp livekit_video_output.cpp - yuv_to_rgba.cu rgba_to_nv12.cu rgba_box_downscale.cu + inc/sensing_core/argus_capture.hpp inc/sensing_core/camera_backend.hpp inc/sensing_core/camera_capture.hpp + inc/sensing_core/capture_factory.hpp + inc/sensing_core/jetson_encoder.hpp inc/sensing_core/sensing_camera.hpp inc/sensing_core/sensing_types.hpp inc/sensing_core/sipl_camera.hpp @@ -21,6 +22,19 @@ add_library(sensing_core STATIC ) add_library(sensing::core ALIAS sensing_core) +if(SENSING_WITH_SIPL) + target_sources(sensing_core PRIVATE sipl_camera.cpp sipl_capture.cpp yuv_to_rgba.cu) + target_compile_definitions(sensing_core PUBLIC SENSING_HAS_SIPL) +endif() + +# The Argus backend is optional: it needs the Argus headers and libnvargus, which a SIPL-only rig does not have. +if(SENSING_WITH_ARGUS) + target_sources(sensing_core PRIVATE argus_capture.cpp) + target_compile_definitions(sensing_core PUBLIC SENSING_HAS_ARGUS) + target_include_directories(sensing_core PRIVATE ${ARGUS_INCLUDE_DIR}) + target_link_libraries(sensing_core PRIVATE ${ARGUS_LIBRARY}) +endif() + # RESOLVE_DEVICE_SYMBOLS as well as SEPARABLE_COMPILATION: without it the device-link step only happens # for consumers in a directory where CUDA is an enabled language. Resolving here makes the archive # self-contained. diff --git a/src/plugins/sensing/core/argus_capture.cpp b/src/plugins/sensing/core/argus_capture.cpp new file mode 100644 index 0000000000..91410f1507 --- /dev/null +++ b/src/plugins/sensing/core/argus_capture.cpp @@ -0,0 +1,630 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// Argus buffer-stream capture into NvBufSurface DMA-BUFs. The Argus calls follow the Jetson Multimedia API's +// client-owned-buffer pattern. Built and run on an AGX Orin by a reviewer before each camera's buffer stream was given +// a single owner thread (buffer_owner.hpp); that structure and the bounded teardown have not been built or run. + +#include "inc/sensing_core/argus_capture.hpp" + +#include "buffer_owner.hpp" + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace plugins +{ +namespace sensing +{ + +namespace +{ + +void check(Argus::Status status, const std::string& what) +{ + if (status != Argus::STATUS_OK) + throw std::runtime_error("Argus: " + what + " failed (status " + std::to_string(static_cast(status)) + ")"); +} + +// A blocking acquire returns as soon as a frame arrives; the wait only bounds how long a camera with no frames +// (an empty port) takes to notice a stop. The value needs validating on the Orin. +constexpr uint64_t kAcquireTimeoutNs = 100'000'000; +// Per camera, how long a stopped session may take to go idle before it is treated as hung. +constexpr uint64_t kIdleTimeoutNs = 3'000'000'000; +// Whole-teardown budget. An SDK call that never returns cannot be cancelled, so past this the process ends: no +// cleanup may run on state the camera hardware could still be writing to. +constexpr std::chrono::seconds kStopDeadline{ 20 }; +// How long the owner may take to stop new captures (a stopRepeat) before that is treated as hung. +constexpr std::chrono::seconds kStopAckTimeout{ 5 }; + +std::shared_ptr logger() +{ + return isaaccapture::Logger::get("isaaccapture.plugins.sensing.ArgusCapture"); +} + +// The SDK call teardown is in, recorded before the call is made so a call that never returns is still named. +struct ActiveCall +{ + std::atomic what{ nullptr }; + std::atomic camera{ -1 }; +}; +ActiveCall g_active_call; + +[[noreturn]] void fatal(const char* what, int64_t camera_id) +{ + std::fprintf(stderr, "Argus: %s (camera %lld); ending the process\n", what, static_cast(camera_id)); + std::_Exit(EXIT_FAILURE); +} + +// Ends the process if teardown does not finish in time. Scope it around the blocking SDK calls. +class StopWatchdog +{ +public: + explicit StopWatchdog(std::chrono::seconds limit) + : m_thread( + [this, limit] + { + std::unique_lock lock(m_mutex); + if (!m_cv.wait_for(lock, limit, [this] { return m_done; })) + { + const char* what = g_active_call.what.load(); + fatal((std::string("teardown did not finish in ") + (what ? what : "an unknown step")).c_str(), + g_active_call.camera.load()); + } + }) + { + } + StopWatchdog(const StopWatchdog&) = delete; + StopWatchdog& operator=(const StopWatchdog&) = delete; + ~StopWatchdog() + { + { + std::lock_guard lock(m_mutex); + m_done = true; + } + m_cv.notify_all(); + m_thread.join(); + } + +private: + std::mutex m_mutex; + std::condition_variable m_cv; + bool m_done = false; + std::thread m_thread; // last: it starts using the members above +}; + +// Runs one blocking SDK call and names it in the log when it is slow, so a hang points at the exact call. +template +auto timed(const char* what, int64_t camera_id, F&& call) +{ + g_active_call.camera.store(camera_id); + g_active_call.what.store(what); + const auto begin = std::chrono::steady_clock::now(); + struct Guard + { + const char* what; + int64_t camera_id; + std::chrono::steady_clock::time_point begin; + ~Guard() + { + g_active_call.what.store(nullptr); + const auto ms = + std::chrono::duration_cast(std::chrono::steady_clock::now() - begin); + if (ms.count() >= 1000) + logger()->warn("camera {}: {} took {} ms", camera_id, what, ms.count()); + } + } guard{ what, camera_id, begin }; + return call(); +} + +uint64_t clock_ns(clockid_t clock) +{ + timespec ts{}; + clock_gettime(clock, &ts); + return static_cast(ts.tv_sec) * 1'000'000'000ull + static_cast(ts.tv_nsec); +} + +// One NV12 DMA-BUF the camera writes into. Block-linear: only the hardware encoder reads it. +struct CaptureBuffer +{ + uint32_t index = 0; ///< position in the camera's pool; names the buffer in the return mailbox + int fd = -1; + NvBufSurface* surface = nullptr; + bool egl_mapped = false; + Argus::UniqueObj argus; + + ~CaptureBuffer() + { + argus.reset(); // the Argus buffer refers to the EGL image, so it goes first + if (egl_mapped) + NvBufSurfaceUnMapEglImage(surface, 0); + if (surface) + NvBufSurfaceDestroy(surface); + } +}; + +} // namespace + +struct ArgusCapture::Impl : std::enable_shared_from_this +{ + struct Camera + { + CameraInfo info; + Argus::CameraDevice* device = nullptr; + Argus::SensorMode* mode = nullptr; + Argus::UniqueObj session; + Argus::UniqueObj stream; + Argus::IBufferOutputStream* buffers = nullptr; + Argus::UniqueObj request; + std::vector> pool; + /// Numbers frames whose capture has no metadata, so the consumer never mistakes them for one re-read frame. + uint32_t fallback_sequence = 0; + + /// The camera's buffer-stream calls, and the one thread that makes them (see BufferOwner). The owner is + /// declared last so it, and its thread, go before the port and the objects above. + std::unique_ptr port; + std::unique_ptr owner; + }; + + // An initialised EGL display is released by this member whether construction completes or throws: a destructor + // that never runs for a failed constructor cannot be the only place that calls eglTerminate. + struct EglDisplay + { + EGLDisplay handle = EGL_NO_DISPLAY; + EglDisplay() = default; + EglDisplay(const EglDisplay&) = delete; + EglDisplay& operator=(const EglDisplay&) = delete; + ~EglDisplay() + { + terminate(); + } + void terminate() + { + if (handle != EGL_NO_DISPLAY) + eglTerminate(std::exchange(handle, EGL_NO_DISPLAY)); + } + }; + + Argus::UniqueObj provider; + EglDisplay egl; + std::vector> cameras; + std::vector infos; + std::map by_id; + bool started = false; + bool stopped = false; + + Impl(const ArgusConfig& config) + { + // Peak demand per camera: the encoder's pinned buffers, the output's pending frame, its worker's frame and + // one fresh acquire. + if (config.buffer_count < kEncoderInputBuffers + 3) + throw std::runtime_error("Argus: buffer_count must be at least " + std::to_string(kEncoderInputBuffers + 3) + + " so the camera never waits for a buffer the encoder holds"); + provider.reset(Argus::CameraProvider::create()); + auto* provider_iface = Argus::interface_cast(provider); + if (!provider_iface) + throw std::runtime_error("Argus: no camera provider (is nvargus-daemon running?)"); + + egl.handle = eglGetDisplay(EGL_DEFAULT_DISPLAY); + if (egl.handle == EGL_NO_DISPLAY || !eglInitialize(egl.handle, nullptr, nullptr)) + throw std::runtime_error("Argus: EGL display initialisation failed"); + + std::vector devices; + check(provider_iface->getCameraDevices(&devices), "getCameraDevices"); + for (uint32_t index : config.devices) + { + if (index >= devices.size()) + throw std::runtime_error("Argus: no camera device " + std::to_string(index) + " (found " + + std::to_string(devices.size()) + ")"); + auto camera = std::make_unique(); + camera->device = devices[index]; + camera->mode = sensor_mode(camera->device, config.sensor_mode); + auto* mode_iface = Argus::interface_cast(camera->mode); + if (!mode_iface) + throw std::runtime_error("Argus: sensor mode has no mode interface"); + const auto size = mode_iface->getResolution(); + const uint64_t period_ns = mode_iface->getFrameDurationRange().min(); + if (period_ns == 0) + throw std::runtime_error("Argus: sensor mode has a zero frame duration"); + camera->info = { index, "argus" + std::to_string(index), size.width(), + size.height(), 1e9 / static_cast(period_ns), PixelKind::DmaBufNv12 }; + open(*camera, provider_iface, config.buffer_count); + camera->port = std::make_unique(*this, *camera); + camera->owner = std::make_unique( + *camera->port, config.buffer_count, std::chrono::nanoseconds(kAcquireTimeoutNs)); + infos.push_back(camera->info); + by_id[index] = camera.get(); + cameras.push_back(std::move(camera)); + } + } + + ~Impl() + { + stop(); + // Closing the owners, destroying the Argus objects and the EGL display is SDK work too and can hang: one + // deadline covers all of it. + const StopWatchdog watchdog(kStopDeadline); + g_active_call.camera.store(-1); + g_active_call.what.store("returning the last buffers"); + // Every lease has been dropped by now (each keeps this object alive), so each owner returns its last buffers + // while the streams still exist, then ends. + for (auto& camera : cameras) + { + camera->owner->close(); + // The owner waited for the session to go idle once every buffer was back. The hardware may still own the + // session's buffers if it did not, and a retry in this process would pile new sessions on top of it. This + // executable is dedicated to the cameras, so that ends the process; starting it again is the supervisor's + // job (the deployed unit restarts it on failure). + if (!camera->owner->closed_cleanly()) + fatal("a session did not go idle after stopRepeat", camera->info.id); + } + g_active_call.what.store("destroying the camera objects"); + for (auto& camera : cameras) + { + camera->request.reset(); + camera->pool.clear(); // Argus buffers, then the DMA-BUFs behind them + camera->stream.reset(); + camera->session.reset(); + } + cameras.clear(); + provider.reset(); + g_active_call.what.store("terminating the EGL display"); + egl.terminate(); // after the Argus objects above; the member's destructor then has nothing left to do + } + + static Argus::SensorMode* sensor_mode(Argus::CameraDevice* device, uint32_t index) + { + auto* properties = Argus::interface_cast(device); + std::vector modes; + if (!properties || properties->getAllSensorModes(&modes) != Argus::STATUS_OK || index >= modes.size()) + throw std::runtime_error("Argus: sensor mode " + std::to_string(index) + " is not available"); + return modes[index]; + } + + // Session, buffer stream, DMA-BUF pool and repeating request for one camera. + void open(Camera& camera, Argus::ICameraProvider* provider_iface, uint32_t buffer_count) + { + Argus::Status status = Argus::STATUS_OK; + camera.session.reset(provider_iface->createCaptureSession(camera.device, &status)); + auto* session = Argus::interface_cast(camera.session); + if (!session) + throw std::runtime_error("Argus: createCaptureSession failed (status " + + std::to_string(static_cast(status)) + ")"); + + Argus::UniqueObj settings( + session->createOutputStreamSettings(Argus::STREAM_TYPE_BUFFER)); + auto* settings_iface = Argus::interface_cast(settings); + auto* buffer_settings_iface = Argus::interface_cast(settings); + if (!settings_iface || !buffer_settings_iface) + throw std::runtime_error("Argus: createOutputStreamSettings failed"); + settings_iface->setCameraDevice(camera.device); + check(buffer_settings_iface->setBufferType(Argus::BUFFER_TYPE_EGL_IMAGE), "setBufferType"); + buffer_settings_iface->setMetadataEnable(true); + + camera.stream.reset(session->createOutputStream(settings.get())); + camera.buffers = Argus::interface_cast(camera.stream); + if (!camera.buffers) + throw std::runtime_error("Argus: createOutputStream failed"); + + Argus::UniqueObj egl_settings(camera.buffers->createBufferSettings()); + auto* egl_iface = Argus::interface_cast(egl_settings); + if (!egl_iface) + throw std::runtime_error("Argus: createBufferSettings failed"); + check(egl_iface->setEGLDisplay(egl.handle), "setEGLDisplay"); + + for (uint32_t i = 0; i < buffer_count; ++i) + camera.pool.push_back(make_buffer(camera, egl_settings.get(), egl_iface)); + + camera.request.reset(session->createRequest(Argus::CAPTURE_INTENT_PREVIEW)); + auto* request = Argus::interface_cast(camera.request); + if (!request) + throw std::runtime_error("Argus: createRequest failed"); + check(request->enableOutputStream(camera.stream.get()), "enableOutputStream"); + auto* source = Argus::interface_cast(request->getSourceSettings()); + if (!source) + throw std::runtime_error("Argus: no source settings"); + check(source->setSensorMode(camera.mode), "setSensorMode"); + // Hold the sensor mode's own frame rate: auto-exposure may otherwise stretch the frame period in dim light. + const uint64_t period_ns = Argus::interface_cast(camera.mode)->getFrameDurationRange().min(); + check(source->setFrameDurationRange(Argus::Range(period_ns, period_ns)), "setFrameDurationRange"); + } + + // One DMA-BUF handed to Argus as a capture buffer; it starts out released, ready to be written. + std::unique_ptr make_buffer(Camera& camera, + Argus::BufferSettings* egl_settings, + Argus::IEGLImageBufferSettings* egl_iface) + { + auto buffer = std::make_unique(); + buffer->index = static_cast(camera.pool.size()); + // The driver's own allocator (libnvbufsurface), not the Multimedia API sample wrapper around it. + NvBufSurfaceAllocateParams params; + std::memset(¶ms, 0, sizeof(params)); + params.memtag = NvBufSurfaceTag_CAMERA; + params.params.width = camera.info.width; + params.params.height = camera.info.height; + params.params.colorFormat = NVBUF_COLOR_FORMAT_NV12; + params.params.layout = NVBUF_LAYOUT_BLOCK_LINEAR; + params.params.memType = NVBUF_MEM_SURFACE_ARRAY; + if (NvBufSurfaceAllocate(&buffer->surface, 1, ¶ms) != 0 || !buffer->surface) + throw std::runtime_error("Argus: allocating a capture DMA-BUF failed"); + // One buffer, one filled surface; its DMA-BUF fd is what the encoder takes. + buffer->surface->numFilled = 1; + buffer->fd = buffer->surface->surfaceList[0].bufferDesc; + if (NvBufSurfaceMapEglImage(buffer->surface, 0) != 0) + throw std::runtime_error("Argus: mapping a capture DMA-BUF to an EGL image failed"); + buffer->egl_mapped = true; + check(egl_iface->setEGLImage(buffer->surface->surfaceList[0].mappedAddr.eglImage), "setEGLImage"); + + buffer->argus.reset(camera.buffers->createBuffer(egl_settings)); + auto* iface = Argus::interface_cast(buffer->argus); + if (!iface) + throw std::runtime_error("Argus: createBuffer failed"); + iface->setClientData(buffer.get()); + check(camera.buffers->releaseBuffer(buffer->argus.get()), "releaseBuffer (initial)"); + return buffer; + } + + // Tied to the lifetime the issued frame leases keep alive; called once the shared_ptr exists. + void attach() + { + for (auto& camera : cameras) + camera->owner->set_keepalive(weak_from_this()); + } + + void start() + { + if (started) + throw std::logic_error("ArgusCapture: capture cannot be restarted; create a new instance"); + started = true; + std::vector repeating; + try + { + for (auto& camera : cameras) + { + auto* session = Argus::interface_cast(camera->session); + check(session->repeat(camera->request.get()), "repeat"); + repeating.push_back(camera.get()); + } + // Runtime ownership of each stream passes to its owner thread here; nothing above runs after this. + for (auto& camera : cameras) + camera->owner->start(); + } + catch (...) + { + undo_start(repeating); + throw; + } + } + + // A failed start: end whichever owner threads began, then stop the sessions that were repeating. With the + // owners gone nothing else touches the streams, so the calls are made here. + void undo_start(const std::vector& repeating) + { + stopped = true; + const StopWatchdog watchdog(kStopDeadline); + for (auto& camera : cameras) + timed("ending the owner thread", camera->info.id, + [&] + { + camera->owner->close(); + return 0; + }); + for (Camera* camera : repeating) + if (!stop_session(*camera)) + fatal("a session did not go idle after stopRepeat", camera->info.id); + } + + // Stops new captures on every camera's owner thread: all are asked before any is waited for, so one camera's + // slow stop does not delay the others'. Frames the consumers still hold stay valid and their buffers keep coming + // back; the wait for the sessions to go idle is deferred to the end (see ~Impl), when no consumer holds one. + void stop() + { + if (!started || stopped) + return; + stopped = true; + const StopWatchdog watchdog(kStopDeadline); + for (auto& camera : cameras) + camera->owner->request_stop(); + for (auto& camera : cameras) + { + const bool ok = timed("waiting for the owner to stop capture", camera->info.id, + [&] { return camera->owner->wait_stopped(kStopAckTimeout); }); + if (!ok) + fatal("a session did not stop", camera->info.id); + } + } + + bool stop_session(Camera& camera) + { + auto* session = Argus::interface_cast(camera.session); + timed("stopRepeat", camera.info.id, [&] { return session->stopRepeat(); }); + return timed("waitForIdle", camera.info.id, [&] { return session->waitForIdle(kIdleTimeoutNs); }) == + Argus::STATUS_OK; + } + + // The SDK calls of one camera's buffer stream. BufferOwner makes them all, on its one thread. + class Port : public BufferPort + { + public: + Port(Impl& impl, Camera& camera) : m_impl(impl), m_camera(camera) + { + } + + Acquired acquire(uint64_t timeout_ns) override + { + Acquired out; + Argus::Status status = Argus::STATUS_OK; + Argus::Buffer* buffer = m_camera.buffers->acquireBuffer(timeout_ns, &status); + if (!buffer) + { + if (status == Argus::STATUS_TIMEOUT) + return out; + out.result = Result::Failed; + out.error = "Argus: acquireBuffer failed (status " + std::to_string(static_cast(status)) + ")"; + return out; + } + auto* iface = Argus::interface_cast(buffer); + const auto* data = iface ? static_cast(iface->getClientData()) : nullptr; + if (!data) + { + // Its slot is unknown, so it cannot go through the mailbox: give it straight back. + m_camera.buffers->releaseBuffer(buffer); + out.result = Result::Failed; + out.error = "Argus: a captured buffer carries no client data"; + return out; + } + out.holds_buffer = true; + out.slot = data->index; + try + { + out.frame = m_impl.make_frame(m_camera, buffer, *data); + out.result = Result::Frame; + } + catch (const std::exception& e) + { + out.result = Result::Failed; + out.error = e.what(); + } + return out; + } + + bool release(uint32_t slot) override + { + if (slot >= m_camera.pool.size()) + return false; + return m_camera.buffers->releaseBuffer(m_camera.pool[slot]->argus.get()) == Argus::STATUS_OK; + } + + bool stop_capture() override + { + auto* session = Argus::interface_cast(m_camera.session); + // stopRepeat returns the range of capture ids it queued, not a Status: a nonempty range is not a failure. + timed("stopRepeat", m_camera.info.id, [&] { return session->stopRepeat(); }); + return true; + } + + bool wait_idle() override + { + auto* session = Argus::interface_cast(m_camera.session); + return timed("waitForIdle", m_camera.info.id, [&] { return session->waitForIdle(kIdleTimeoutNs); }) == + Argus::STATUS_OK; + } + + private: + Impl& m_impl; + Camera& m_camera; + }; + + // The frame for a buffer that has just arrived; its stamps are the arrival time. The owner adds the lease. + CaptureFrame make_frame(Camera& camera, Argus::Buffer* buffer, const CaptureBuffer& data) + { + CaptureFrame frame; + frame.camera_id = camera.info.id; + frame.width = camera.info.width; + frame.height = camera.info.height; + frame.monotonic_ns = clock_ns(CLOCK_MONOTONIC); + frame.unix_ns = clock_ns(CLOCK_REALTIME); + + auto* iface = Argus::interface_cast(buffer); + // The sensor's own timestamp, unscaled and in Argus's clock domain; the capture id numbers the frames. + const auto* metadata = Argus::interface_cast(iface->getMetadata()); + if (metadata) + { + frame.sensor_time = metadata->getSensorTimestamp(); + frame.sequence = metadata->getCaptureId(); + } + else + { + frame.sequence = ++camera.fallback_sequence; + } + frame.pixels = DmaBufNv12Pixels{ data.fd }; + return frame; + } +}; + +ArgusCapture::ArgusCapture(const ArgusConfig& config) : m_impl(std::make_shared(config)) +{ + m_impl->attach(); +} + +ArgusCapture::~ArgusCapture() +{ + // Outstanding leases keep the Impl, and with it the Argus objects, alive until they are released. + m_impl->stop(); +} + +std::vector ArgusCapture::query(uint32_t sensor_mode) +{ + Argus::UniqueObj provider(Argus::CameraProvider::create()); + auto* provider_iface = Argus::interface_cast(provider); + if (!provider_iface) + throw std::runtime_error("Argus: no camera provider (is nvargus-daemon running?)"); + std::vector devices; + check(provider_iface->getCameraDevices(&devices), "getCameraDevices"); + + std::vector infos; + for (uint32_t index = 0; index < devices.size(); ++index) + { + // A device without this mode cannot be opened with it, but must not hide the other cameras. + auto* properties = Argus::interface_cast(devices[index]); + std::vector modes; + if (!properties || properties->getAllSensorModes(&modes) != Argus::STATUS_OK || sensor_mode >= modes.size()) + continue; + const auto* mode = Argus::interface_cast(modes[sensor_mode]); + if (!mode) + continue; + const auto size = mode->getResolution(); + const uint64_t period_ns = mode->getFrameDurationRange().min(); + infos.push_back({ index, "argus" + std::to_string(index), size.width(), size.height(), + period_ns ? 1e9 / static_cast(period_ns) : 0.0, PixelKind::DmaBufNv12 }); + } + return infos; +} + +const std::vector& ArgusCapture::cameras() const +{ + return m_impl->infos; +} + +void ArgusCapture::start() +{ + m_impl->start(); +} + +void ArgusCapture::stop() +{ + m_impl->stop(); +} + +std::optional ArgusCapture::acquire_latest(uint32_t camera_id) +{ + const auto found = m_impl->by_id.find(camera_id); + if (found == m_impl->by_id.end()) + throw std::runtime_error("Argus: no camera " + std::to_string(camera_id)); + // The camera's owner thread keeps the newest frame; take it, if there is one. + return found->second->owner->take_latest(); +} + +} // namespace sensing +} // namespace plugins diff --git a/src/plugins/sensing/core/buffer_owner.hpp b/src/plugins/sensing/core/buffer_owner.hpp new file mode 100644 index 0000000000..5f8a15233a --- /dev/null +++ b/src/plugins/sensing/core/buffer_owner.hpp @@ -0,0 +1,422 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// One thread owns every runtime call on a camera's buffer stream. Frame leases are released from any thread, so they +// only post the buffer's slot to a mailbox; the owner makes the actual release. No vendor SDK type appears here, so +// the ownership rules are testable with a fake port. + +#pragma once + +#include "inc/sensing_core/camera_capture.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace plugins +{ +namespace sensing +{ + +/// Buffers waiting to go back to the stream. post() is for any thread and makes no outside call; the owner drains. +/// Capacity is the pool size, fixed up front: a slot is posted at most once per lease, so it never reallocates. +class BufferReturnMailbox +{ +public: + explicit BufferReturnMailbox(size_t capacity) : m_capacity(capacity) + { + m_pending.reserve(capacity); + } + + /// False, and counted, if the mailbox is full: that means a slot was returned twice, which is a bug. + bool post(uint32_t slot) + { + { + std::lock_guard lock(m_mutex); + if (m_pending.size() >= m_capacity) + { + ++m_rejected; + return false; + } + m_pending.push_back(slot); + } + m_cv.notify_one(); + return true; + } + + /// Wakes the owner's wait without posting anything. + void wake() + { + { + std::lock_guard lock(m_mutex); + m_woken = true; + } + m_cv.notify_one(); + } + + /// Owner: moves every pending slot into `out`. + void take(std::vector& out) + { + std::lock_guard lock(m_mutex); + out.insert(out.end(), m_pending.begin(), m_pending.end()); + m_pending.clear(); + } + + /// Owner: blocks until a slot is posted, wake() is called, or the time is up. + void wait(std::chrono::nanoseconds timeout) + { + std::unique_lock lock(m_mutex); + m_cv.wait_for(lock, timeout, [this] { return !m_pending.empty() || m_woken; }); + m_woken = false; + } + + uint64_t rejected() const + { + std::lock_guard lock(m_mutex); + return m_rejected; + } + +private: + const size_t m_capacity; + mutable std::mutex m_mutex; + std::condition_variable m_cv; + std::vector m_pending; + bool m_woken = false; + uint64_t m_rejected = 0; +}; + +/// The thin SDK calls the owner makes. All of them run on the owner's thread, never two at once. +class BufferPort +{ +public: + enum class Result + { + Frame, + Timeout, + Failed, + }; + + struct Acquired + { + Result result = Result::Timeout; + /// True when the stream handed over a buffer, whether or not the frame could be built from it: that buffer + /// is then the owner's to return. The frame's lease is left empty; the owner sets it. + bool holds_buffer = false; + uint32_t slot = 0; + CaptureFrame frame; + std::string error; + }; + + virtual ~BufferPort() = default; + /// Waits up to `timeout_ns` for the next frame. + virtual Acquired acquire(uint64_t timeout_ns) = 0; + /// Gives the buffer in `slot` back to the stream. False: the stream refused it. + virtual bool release(uint32_t slot) = 0; + /// Stops new captures from starting (no waiting). False: it could not. + virtual bool stop_capture() = 0; + /// Waits for capture that was stopped to go idle. The owner calls it once every buffer is back, so the wait + /// never depends on a buffer a consumer still holds. False: it did not go idle. + virtual bool wait_idle() = 0; +}; + +/** + * @brief Runs one camera's buffer stream on its own thread and keeps only the newest frame. + * + * The thread blocks on the port's finite acquire, stamps nothing itself, and replaces the single latest frame; a + * replaced frame goes back through the same mailbox as any other. When every buffer is out it waits on the mailbox + * instead of in an acquire that needs a returned buffer. Stopping has two parts: request_stop() stops new captures and + * clears the latest frame but keeps returning buffers; close() ends the thread after draining the mailbox, and only + * then waits for the stream to go idle. Frames handed + * out stay valid across both; the owner must outlive every lease it issued, which `keepalive` can tie to a larger + * object. + */ +class BufferOwner +{ +public: + BufferOwner(BufferPort& port, size_t pool_size, std::chrono::nanoseconds wait = std::chrono::milliseconds(100)) + : m_port(port), m_pool_size(pool_size), m_wait(wait), m_mailbox(pool_size) + { + } + BufferOwner(const BufferOwner&) = delete; + BufferOwner& operator=(const BufferOwner&) = delete; + ~BufferOwner() + { + close(); + } + + /// Objects the issued leases keep alive until they are released; weak, so the owner never owns its owner. + void set_keepalive(std::weak_ptr keepalive) + { + m_keepalive = std::move(keepalive); + } + + /// @throws std::system_error if the thread cannot be created. + void start() + { + if (m_started) + throw std::logic_error("BufferOwner: already started"); + m_thread = std::thread([this] { run(); }); + m_started = true; + } + + /// Asks the owner thread to stop capture; returns at once. + void request_stop() + { + if (!m_started || !m_thread.joinable()) + { + finish_stop(true); // no thread is running, so there is nothing to stop through it + return; + } + Command expected = Command::Run; + m_command.compare_exchange_strong(expected, Command::Stop); + m_mailbox.wake(); + } + + /// True once capture has stopped and gone idle; false if that took longer than `timeout` or the port said no. + bool wait_stopped(std::chrono::nanoseconds timeout) + { + std::unique_lock lock(m_state_mutex); + return m_state_cv.wait_for(lock, timeout, [this] { return m_stop_done; }) && m_stop_ok; + } + + /// Ends the thread after returning every buffer posted so far; it does not stop capture (see request_stop). + /// Idempotent. Not for the owner thread itself. + void close() + { + m_command.store(Command::Exit); + m_mailbox.wake(); + if (m_thread.joinable()) + m_thread.join(); + } + + /// The newest frame not yet taken. @throws std::runtime_error with the reason if the owner failed. + std::optional take_latest() + { + std::lock_guard lock(m_state_mutex); + if (!m_error.empty()) + throw std::runtime_error(m_error); + std::optional frame = std::move(m_latest); + m_latest.reset(); + return frame; + } + + /// After close(): every buffer was returned and, if capture was stopped, the stream went idle with nothing out. + bool closed_cleanly() const + { + std::lock_guard lock(m_state_mutex); + return m_closed_clean; + } + + uint64_t rejected_returns() const + { + return m_mailbox.rejected(); + } + +private: + enum class Command + { + Run, + Stop, + Exit, + }; + + void run() + { + std::vector returned; + returned.reserve(m_pool_size); + bool stopped = false; + for (;;) + { + drain(returned); + const Command command = m_command.load(); + if (command == Command::Exit) + break; + if (command == Command::Stop && !stopped) + { + // Return the held frame's buffer before the blocking idle wait: the owner cannot return anything while + // it is inside that wait. + clear_latest(); + drain(returned); + bool ok = false; + try + { + ok = m_port.stop_capture(); + } + catch (const std::exception& e) + { + fail(e.what()); + } + finish_stop(ok); + stopped = true; + continue; + } + // Nothing to acquire into, or nothing more to acquire: wait for returns or a command instead. + if (stopped || !m_acquiring || m_outstanding >= m_pool_size) + { + m_mailbox.wait(m_wait); + continue; + } + BufferPort::Acquired got; + try + { + got = m_port.acquire(static_cast(m_wait.count())); + } + catch (const std::exception& e) + { + // A port reports failure through its result; a throw is a bug, and it took no buffer we know of. + fail(e.what()); + m_acquiring = false; + continue; + } + if (got.holds_buffer) + ++m_outstanding; + if (got.result == BufferPort::Result::Timeout) + continue; + if (got.result == BufferPort::Result::Failed) + { + if (got.holds_buffer) + give_back(got.slot); + fail(got.error.empty() ? "the buffer stream failed" : got.error); + m_acquiring = false; + continue; + } + got.frame.lease = make_lease(got.slot); + std::optional replaced; + { + std::lock_guard lock(m_state_mutex); + replaced = std::move(m_latest); + m_latest = std::move(got.frame); + } + // `replaced` goes out of scope here, outside the lock: its lease posts the buffer back. + } + // The thread is ending: every posted buffer goes back, then (if capture was stopped) the stream must go idle + // with nothing out. A buffer still out, or a stream that will not go idle, is reported to close()'s caller. + drain(returned); + clear_latest(); + drain(returned); + bool clean = m_outstanding == 0; + if (stopped) + { + bool idle = false; + try + { + idle = m_port.wait_idle(); + } + catch (const std::exception& e) + { + fail(e.what()); + } + if (!idle) + fail("the buffer stream did not go idle"); + clean = clean && idle; + drain(returned); + clean = clean && m_outstanding == 0; + } + std::lock_guard lock(m_state_mutex); + m_closed_clean = clean; + } + + // Returns every posted buffer to the stream. A buffer the stream refuses stays counted as out, and acquiring + // stops: the stream is failing and the pool no longer adds up. + void drain(std::vector& returned) + { + returned.clear(); + m_mailbox.take(returned); + for (uint32_t slot : returned) + if (!give_back(slot)) + m_acquiring = false; + } + + bool give_back(uint32_t slot) + { + bool ok = false; + try + { + ok = m_port.release(slot); + } + catch (const std::exception& e) + { + fail(e.what()); + return false; + } + if (!ok) + { + fail("the buffer stream refused a returned buffer"); + return false; + } + if (m_outstanding > 0) + --m_outstanding; + return true; + } + + void clear_latest() + { + std::optional dropped; + { + std::lock_guard lock(m_state_mutex); + dropped = std::move(m_latest); + m_latest.reset(); + } + } + + FrameLease make_lease(uint32_t slot) + { + // The owner outlives its leases (see the class comment), so `this` is valid when a lease is dropped. A + // rejected post means a buffer came back twice: that breaks the pool count, so it fails the owner. + return FrameLease( + [this, keep = m_keepalive.lock(), slot]() + { + if (!m_mailbox.post(slot)) + fail("a buffer was returned twice"); + }); + } + + void fail(const std::string& why) + { + std::lock_guard lock(m_state_mutex); + if (m_error.empty()) + m_error = why; + } + + void finish_stop(bool ok) + { + { + std::lock_guard lock(m_state_mutex); + m_stop_done = true; + m_stop_ok = ok; + } + m_state_cv.notify_all(); + } + + BufferPort& m_port; + const size_t m_pool_size; + const std::chrono::nanoseconds m_wait; + BufferReturnMailbox m_mailbox; + std::weak_ptr m_keepalive; + std::atomic m_command{ Command::Run }; + bool m_started = false; + std::thread m_thread; + + size_t m_outstanding = 0; ///< owner thread only: buffers acquired and not yet released + bool m_acquiring = true; ///< owner thread only: false once the stream has failed + + mutable std::mutex m_state_mutex; ///< guards the members below + std::condition_variable m_state_cv; + std::optional m_latest; + std::string m_error; + bool m_stop_done = false; + bool m_stop_ok = false; + bool m_closed_clean = true; ///< set by the owner thread as it ends; true if no thread ever ran +}; + +} // namespace sensing +} // namespace plugins diff --git a/src/plugins/sensing/core/camera_backend.cpp b/src/plugins/sensing/core/camera_backend.cpp index 8bb84a003f..693e9655e6 100644 --- a/src/plugins/sensing/core/camera_backend.cpp +++ b/src/plugins/sensing/core/camera_backend.cpp @@ -48,6 +48,25 @@ std::string describe(const BackendProbe& probe) } // namespace +void match_requested_cameras(const std::vector& found, + const std::vector& requested, + const std::string& label, + BackendProbe& probe) +{ + probe.matches = !found.empty(); + if (found.empty()) + probe.detail = label + " reports no camera devices"; + for (uint32_t id : requested) + { + const bool present = std::any_of(found.begin(), found.end(), [id](const CameraInfo& c) { return c.id == id; }); + if (!present) + { + probe.matches = false; + probe.detail = "no " + label + " camera device " + std::to_string(id); + } + } +} + CameraBackend select_camera_backend(CameraBackend requested, const std::vector& probes) { if (requested != CameraBackend::Auto) diff --git a/src/plugins/sensing/core/capture_factory.cpp b/src/plugins/sensing/core/capture_factory.cpp new file mode 100644 index 0000000000..b0fa3e62e5 --- /dev/null +++ b/src/plugins/sensing/core/capture_factory.cpp @@ -0,0 +1,54 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "inc/sensing_core/capture_factory.hpp" + +#ifdef SENSING_HAS_SIPL +# include "sipl_capture.hpp" +#endif + +#ifdef SENSING_HAS_ARGUS +# include "inc/sensing_core/argus_capture.hpp" +#endif + +#include + +namespace plugins +{ +namespace sensing +{ + +std::unique_ptr make_capture(CameraBackend backend, + const SensingConfig& config, + const std::vector& camera_ids) +{ +#ifdef SENSING_HAS_SIPL + if (backend == CameraBackend::Sipl) + { + SiplConfig sipl{}; + sipl.platform_config_json = config.platform_config_json; + sipl.platform_config_name = config.platform_config_name; + sipl.link_masks = config.link_masks; + sipl.nito_dir = config.nito_dir; + sipl.gpu_id = config.gpu_id; + sipl.full_range = config.full_range; + sipl.swap_uv = config.swap_uv; + sipl.isp0_buffers = config.isp0_buffers; + return std::make_unique(sipl); + } +#endif +#ifdef SENSING_HAS_ARGUS + if (backend == CameraBackend::Argus) + { + ArgusConfig argus; + argus.devices = camera_ids; + argus.sensor_mode = config.argus_sensor_mode; + return std::make_unique(argus); + } +#endif + (void)camera_ids; + throw std::runtime_error("this build does not include the requested camera backend"); +} + +} // namespace sensing +} // namespace plugins diff --git a/src/plugins/sensing/core/inc/sensing_core/argus_capture.hpp b/src/plugins/sensing/core/inc/sensing_core/argus_capture.hpp new file mode 100644 index 0000000000..fceb722311 --- /dev/null +++ b/src/plugins/sensing/core/inc/sensing_core/argus_capture.hpp @@ -0,0 +1,54 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// The Argus backend behind the sender's capture interface: each camera streams NV12 into a pool of NvBufSurface +// DMA-BUFs that the Jetson hardware encoder reads directly. Built only with SENSING_WITH_ARGUS, which needs the +// Argus headers and libnvargus; the Argus types stay out of this header. + +#pragma once + +#include "camera_capture.hpp" +#include "jetson_encoder.hpp" + +#include +#include + +namespace plugins +{ +namespace sensing +{ + +struct ArgusConfig +{ + /// Argus camera device indices to capture; CameraInfo::id is the device index. + std::vector devices; + /// Index into each device's sensor modes (as listed by query()). + uint32_t sensor_mode = 0; + /// Capture buffers per camera. The encoder pins kEncoderInputBuffers, the output's pending mailbox and its + /// worker's frame hold one each, and a fresh acquire needs one more; fewer and the camera stalls. + uint32_t buffer_count = kEncoderInputBuffers + 4; +}; + +class ArgusCapture : public CameraCapture +{ +public: + /// @throws std::runtime_error naming the failing Argus or EGL step. + explicit ArgusCapture(const ArgusConfig& config); + ~ArgusCapture() override; + + /// Every Argus device with the given sensor mode's size and rate. Opens no camera. + static std::vector query(uint32_t sensor_mode); + + const std::vector& cameras() const override; + void start() override; + void stop() override; + std::optional acquire_latest(uint32_t camera_id) override; + +private: + // Shared with the frame leases it issues, so the Argus objects outlive every outstanding frame. + struct Impl; + std::shared_ptr m_impl; +}; + +} // namespace sensing +} // namespace plugins diff --git a/src/plugins/sensing/core/inc/sensing_core/camera_backend.hpp b/src/plugins/sensing/core/inc/sensing_core/camera_backend.hpp index 154914fe21..b2018e16e8 100644 --- a/src/plugins/sensing/core/inc/sensing_core/camera_backend.hpp +++ b/src/plugins/sensing/core/inc/sensing_core/camera_backend.hpp @@ -34,6 +34,17 @@ std::optional parse_camera_backend(const std::string& text); const char* camera_backend_name(CameraBackend backend); +/** + * @brief Whether the cameras a backend found include every requested one. + * + * Sets `probe.matches`, and `probe.detail` when it does not match. With no camera requested (a listing), any camera + * counts; none found never matches. `label` names the backend in the reason. + */ +void match_requested_cameras(const std::vector& found, + const std::vector& requested, + const std::string& label, + BackendProbe& probe); + /** * @brief The backend to open. * diff --git a/src/plugins/sensing/core/inc/sensing_core/camera_capture.hpp b/src/plugins/sensing/core/inc/sensing_core/camera_capture.hpp index 642d0cba1f..9108512630 100644 --- a/src/plugins/sensing/core/inc/sensing_core/camera_capture.hpp +++ b/src/plugins/sensing/core/inc/sensing_core/camera_capture.hpp @@ -27,6 +27,13 @@ enum class CameraBackend Sipl, }; +/// Which pixel representation a backend's frames use; fixed per backend, so the encoder input can be chosen once. +enum class PixelKind +{ + CudaRgba, + DmaBufNv12, +}; + /// One camera as a backend reports it. `id` is the backend's own index (a SIPL pipeline index, an Argus device /// index); the two are unrelated, so configuration names cameras by logical stream, not by id. struct CameraInfo @@ -36,6 +43,7 @@ struct CameraInfo uint32_t width = 0; uint32_t height = 0; double fps = 0.0; + PixelKind pixels = PixelKind::CudaRgba; }; /// RGBA8 in GPU memory, as the SIPL backend converts it. Borrowed: the backend recycles the buffer, so a consumer @@ -46,15 +54,13 @@ struct CudaRgbaPixels size_t pitch_bytes = 0; }; -/// NV12 in a DMA-BUF, as an Argus buffer stream produces it. Leased: valid until the frame's lease is released. -/// Contract: the producer has finished writing when the frame is returned, and the surface is pitch-linear NV12 with -/// the pitches and offsets below. A backend that cannot deliver that (block-linear, or still being written) must -/// convert or wait before returning, not describe it here for the encoder to guess. +/// NV12 in a DMA-BUF allocated through NvBufSurface, as an Argus buffer stream writes it. Leased: valid until the +/// frame's lease is released. The surface layout (block-linear or pitch-linear) is whatever the allocation chose and +/// is understood only by the Jetson hardware encoder, which takes the fd directly; no CPU or CUDA consumer may +/// assume a layout. The producer has finished writing when the frame is returned. struct DmaBufNv12Pixels { int fd = -1; - uint32_t pitch_bytes[2] = { 0, 0 }; - uint32_t offset_bytes[2] = { 0, 0 }; }; /// Returns a frame's buffer to its backend when destroyed or reset. Move-only: exactly one holder owns the buffer, diff --git a/src/plugins/sensing/core/inc/sensing_core/capture_factory.hpp b/src/plugins/sensing/core/inc/sensing_core/capture_factory.hpp new file mode 100644 index 0000000000..9ffa1bb2f8 --- /dev/null +++ b/src/plugins/sensing/core/inc/sensing_core/capture_factory.hpp @@ -0,0 +1,25 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include "camera_capture.hpp" +#include "sensing_types.hpp" + +#include +#include +#include + +namespace plugins +{ +namespace sensing +{ + +/// Opens the chosen backend for the given camera ids (SIPL pipeline indices, or Argus device indices). +/// @throws std::runtime_error if the backend is not built in or cannot open the cameras. +std::unique_ptr make_capture(CameraBackend backend, + const SensingConfig& config, + const std::vector& camera_ids); + +} // namespace sensing +} // namespace plugins diff --git a/src/plugins/sensing/core/jetson_encoder.hpp b/src/plugins/sensing/core/inc/sensing_core/jetson_encoder.hpp similarity index 82% rename from src/plugins/sensing/core/jetson_encoder.hpp rename to src/plugins/sensing/core/inc/sensing_core/jetson_encoder.hpp index aabd3dcfcd..84fdf97680 100644 --- a/src/plugins/sensing/core/jetson_encoder.hpp +++ b/src/plugins/sensing/core/inc/sensing_core/jetson_encoder.hpp @@ -3,6 +3,8 @@ #pragma once +#include "camera_capture.hpp" + #include #include #include @@ -12,6 +14,9 @@ namespace plugins namespace sensing { +/// Input buffers the encoder holds; with DMA-BUF input each one pins a capture buffer until its index is reused. +constexpr uint32_t kEncoderInputBuffers = 6; + struct EncoderConfig { uint32_t width = 0; @@ -24,6 +29,8 @@ struct EncoderConfig /// IDR period in frames; 0 defers to fps*5. uint32_t gop = 0; bool full_range = false; + /// What submit()/submit_dmabuf() will be given; fixed for the encoder's lifetime. + PixelKind input = PixelKind::CudaRgba; }; /// One encoded Annex-B unit. @@ -77,6 +84,15 @@ class JetsonEncoder */ bool submit(uintptr_t rgba_device_ptr, std::size_t row_pitch_bytes, int64_t timestamp_ns); + /** + * @brief Queue one NV12 DMA-BUF frame for encoding, with no copy. Only for `input == DmaBufNv12`. + * @param dmabuf_fd The NvBufSurface fd the encoder reads. + * @param lease Keeps the capture buffer alive: held until the encoder's input plane returns this buffer, and + * released at once if the frame is dropped. + * @return false when no input buffer is free (frame dropped). + */ + bool submit_dmabuf(int dmabuf_fd, int64_t timestamp_ns, FrameLease lease); + /** @brief Take any completed unit; empty during encoder warmup. */ EncodedUnit poll(); diff --git a/src/plugins/sensing/core/inc/sensing_core/sensing_camera.hpp b/src/plugins/sensing/core/inc/sensing_core/sensing_camera.hpp index b5271c0d41..d022e09036 100644 --- a/src/plugins/sensing/core/inc/sensing_core/sensing_camera.hpp +++ b/src/plugins/sensing/core/inc/sensing_core/sensing_camera.hpp @@ -21,14 +21,17 @@ class LiveKitVideoOutput; /** * @brief Multi-sensor SENSING camera publisher. * - * One SiplCamera owns every sensor, because INvSIPLCamera::GetInstance() is process-wide -- there is no - * per-sensor capture object to hold. Each update() polls every sensor's latest-frame mailbox and hands a new - * frame to that sensor's LiveKit track. Geometry is per sensor and comes from the platform config. + * One CameraCapture owns every camera of the rig (SIPL's camera object is process-wide, so there is no + * per-camera capture object to hold). Each update() polls every sensor's latest frame and hands a new frame + * to that sensor's LiveKit track. Geometry is per sensor and comes from the capture backend. */ class SensingCamera { public: - SensingCamera(const SensingConfig& config, const std::vector& streams); + /// `camera` is the opened backend; its camera ids are what StreamConfig::sensor_id names. + SensingCamera(const SensingConfig& config, + std::unique_ptr camera, + const std::vector& streams); ~SensingCamera(); SensingCamera(const SensingCamera&) = delete; diff --git a/src/plugins/sensing/core/inc/sensing_core/sensing_types.hpp b/src/plugins/sensing/core/inc/sensing_core/sensing_types.hpp index 9fb8fcba2a..b88d0e29cc 100644 --- a/src/plugins/sensing/core/inc/sensing_core/sensing_types.hpp +++ b/src/plugins/sensing/core/inc/sensing_core/sensing_types.hpp @@ -18,7 +18,7 @@ namespace sensing /// One captured sensor and the LiveKit track that carries it. struct StreamConfig { - /// SIPL pipeline index, validated against the platform config at startup. + /// The selected backend's own camera index (SIPL pipeline or Argus device), validated at startup. uint32_t sensor_id = 0; /// Name of the LiveKit camera track carrying this sensor. std::string livekit_track; @@ -58,6 +58,9 @@ struct SensingConfig /// ISP0 buffers per sensor, ~6 MB each at 2560x1984 NV12. uint32_t isp0_buffers = 6; + /// Argus only: the sensor mode index applied to every camera. + uint32_t argus_sensor_mode = 0; + LiveKitConfig livekit; }; diff --git a/src/plugins/sensing/core/jetson_encoder.cpp b/src/plugins/sensing/core/jetson_encoder.cpp index 5eb232758e..b61665ac83 100644 --- a/src/plugins/sensing/core/jetson_encoder.cpp +++ b/src/plugins/sensing/core/jetson_encoder.cpp @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -#include "jetson_encoder.hpp" +#include "inc/sensing_core/jetson_encoder.hpp" #include "rgba_to_nv12.cuh" @@ -25,7 +25,7 @@ namespace sensing namespace { -constexpr uint32_t kOutputBuffers = 6; +constexpr uint32_t kOutputBuffers = kEncoderInputBuffers; constexpr uint32_t kCaptureBuffers = 6; void check(int ret, const char* what) @@ -47,6 +47,17 @@ struct JetsonEncoder::Impl EncoderConfig config; NvVideoEncoder* encoder = nullptr; + // DMA-BUF input: a wrapper per output buffer carrying the fd the encoder reads, and the lease that keeps the + // capture buffer alive until the plane returns that index. Members destroy after ~Impl deleted the encoder. + struct DmaInput : NvBuffer + { + DmaInput() : NvBuffer(0, 0) + { + } + }; + std::vector> dma_inputs; + std::vector dma_leases; + // NV12 staging in device memory; the V4L2 output plane is host-mapped, so // each frame costs one device-to-host copy per plane. uint8_t* d_y = nullptr; @@ -86,6 +97,29 @@ struct JetsonEncoder::Impl error = what; } + // Fills v4l2_buf with a free output buffer (and `buffer` when the plane owns one). Until every output buffer + // has been queued once, index i is free by construction; after that a dequeue is what frees one. False: none + // is free yet, an ordinary dropped frame, or the device failed, which poll() reports. + bool take_input(uint32_t& queued, struct v4l2_buffer& v4l2_buf, NvBuffer** buffer) + { + if (queued < encoder->output_plane.getNumBuffers()) + { + if (buffer) + *buffer = encoder->output_plane.getNthBuffer(queued); + v4l2_buf.index = queued; + ++queued; + return true; + } + if (encoder->output_plane.dqBuffer(v4l2_buf, buffer, nullptr, 0) >= 0) + return true; + // The plane logs every failure but EAGAIN, which is the non-blocking "no input buffer free yet": drop this + // frame rather than block. Any other errno is the device failing. + const int err = errno; + if (err != EAGAIN && err != EINTR) + fail((std::string("dequeueing an input buffer failed: ") + std::strerror(err)).c_str()); + return false; + } + // Capture-plane dequeue callback; runs on the encoder's own thread. static bool capture_dq(struct v4l2_buffer* v4l2_buf, NvBuffer* buffer, NvBuffer* /*shared*/, void* arg) { @@ -183,7 +217,17 @@ JetsonEncoder::JetsonEncoder(const EncoderConfig& config) : m_impl(std::make_uni // FAST suits live streams; four native-size images at 30 fps may need ULTRAFAST, to be measured. check(enc->setHWPresetType(V4L2_ENC_HW_PRESET_FAST), "setHWPresetType"); - check(enc->output_plane.setupPlane(V4L2_MEMORY_MMAP, kOutputBuffers, true, false), "output setupPlane"); + if (config.input == PixelKind::DmaBufNv12) + { + check(enc->output_plane.setupPlane(V4L2_MEMORY_DMABUF, kOutputBuffers, false, false), "output setupPlane"); + for (uint32_t i = 0; i < kOutputBuffers; ++i) + m_impl->dma_inputs.push_back(std::make_unique()); + m_impl->dma_leases.resize(kOutputBuffers); + } + else + { + check(enc->output_plane.setupPlane(V4L2_MEMORY_MMAP, kOutputBuffers, true, false), "output setupPlane"); + } check(enc->capture_plane.setupPlane(V4L2_MEMORY_MMAP, kCaptureBuffers, true, false), "capture setupPlane"); check(enc->output_plane.setStreamStatus(true), "output setStreamStatus"); @@ -205,11 +249,15 @@ JetsonEncoder::JetsonEncoder(const EncoderConfig& config) : m_impl(std::make_uni check(enc->capture_plane.qBuffer(v4l2_buf, nullptr), "capture qBuffer"); } - check_cuda(cudaMallocPitch(reinterpret_cast(&m_impl->d_y), &m_impl->d_y_pitch, config.width, config.height), - "cudaMallocPitch(Y)"); - check_cuda( - cudaMallocPitch(reinterpret_cast(&m_impl->d_uv), &m_impl->d_uv_pitch, config.width, config.height / 2), - "cudaMallocPitch(UV)"); + if (config.input == PixelKind::CudaRgba) + { + check_cuda( + cudaMallocPitch(reinterpret_cast(&m_impl->d_y), &m_impl->d_y_pitch, config.width, config.height), + "cudaMallocPitch(Y)"); + check_cuda(cudaMallocPitch( + reinterpret_cast(&m_impl->d_uv), &m_impl->d_uv_pitch, config.width, config.height / 2), + "cudaMallocPitch(UV)"); + } } JetsonEncoder::~JetsonEncoder() = default; @@ -225,24 +273,11 @@ bool JetsonEncoder::submit(uintptr_t rgba_device_ptr, std::size_t row_pitch_byte std::memset(planes, 0, sizeof(planes)); v4l2_buf.m.planes = planes; + if (config.input != PixelKind::CudaRgba) + throw std::logic_error("JetsonEncoder: submit() needs a CUDA RGBA encoder; use submit_dmabuf()"); NvBuffer* buffer = nullptr; - // Until every output buffer has been queued once, index i is free by - // construction; after that a dequeue is what frees one. - if (m_queued < enc->output_plane.getNumBuffers()) - { - buffer = enc->output_plane.getNthBuffer(m_queued); - v4l2_buf.index = m_queued; - ++m_queued; - } - else if (enc->output_plane.dqBuffer(v4l2_buf, &buffer, nullptr, 0) < 0) - { - // The plane logs every failure but EAGAIN, which is the non-blocking "no input buffer free yet": drop this - // frame rather than block. Any other errno is the device failing, which poll() reports. - const int err = errno; - if (err != EAGAIN && err != EINTR) - m_impl->fail((std::string("dequeueing an input buffer failed: ") + std::strerror(err)).c_str()); + if (!m_impl->take_input(m_queued, v4l2_buf, &buffer)) return false; - } // An input buffer is ours now, so convert: a frame that would be dropped costs no conversion. A // failure from here is terminal for the encoder, as a failed qBuffer is. @@ -272,6 +307,39 @@ bool JetsonEncoder::submit(uintptr_t rgba_device_ptr, std::size_t row_pitch_byte return true; } +bool JetsonEncoder::submit_dmabuf(int dmabuf_fd, int64_t timestamp_ns, FrameLease lease) +{ + if (m_impl->config.input != PixelKind::DmaBufNv12) + throw std::logic_error("JetsonEncoder: submit_dmabuf() needs an encoder created for DMA-BUF input"); + + struct v4l2_buffer v4l2_buf; + struct v4l2_plane planes[MAX_PLANES]; + std::memset(&v4l2_buf, 0, sizeof(v4l2_buf)); + std::memset(planes, 0, sizeof(planes)); + v4l2_buf.m.planes = planes; + + // A dropped frame returns here, and `lease` releases the capture buffer as it goes out of scope. + if (!m_impl->take_input(m_queued, v4l2_buf, nullptr)) + return false; + + // The encoder is finished with whatever this index carried before: give that capture buffer back. + const uint32_t index = v4l2_buf.index; + m_impl->dma_leases[index].reset(); + + // Unverified contract: the output plane is configured NV12M (two planes) but only plane 0's fd is set, as the + // repository's Argus example does for the single-fd NvBufSurface Argus fills. Not run here; confirm on the target. + auto& input = *m_impl->dma_inputs[index]; + input.planes[0].fd = dmabuf_fd; + input.planes[0].bytesused = 1; + v4l2_buf.m.planes[0].bytesused = 1; + v4l2_buf.timestamp.tv_sec = static_cast(timestamp_ns / 1000000000LL); + v4l2_buf.timestamp.tv_usec = static_cast((timestamp_ns % 1000000000LL) / 1000LL); + + check(m_impl->encoder->output_plane.qBuffer(v4l2_buf, &input), "output qBuffer"); + m_impl->dma_leases[index] = std::move(lease); + return true; +} + EncodedUnit JetsonEncoder::poll() { std::lock_guard lock(m_impl->mutex); @@ -305,6 +373,11 @@ std::vector JetsonEncoder::drain() NvBuffer* buffer = nullptr; if (enc->output_plane.dqBuffer(v4l2_buf, &buffer, nullptr, 10) >= 0) { + if (m_impl->config.input == PixelKind::DmaBufNv12) + { + m_impl->dma_leases[v4l2_buf.index].reset(); + v4l2_buf.m.planes[0].m.fd = -1; + } v4l2_buf.m.planes[0].bytesused = 0; v4l2_buf.m.planes[1].bytesused = 0; enc->output_plane.qBuffer(v4l2_buf, nullptr); diff --git a/src/plugins/sensing/core/livekit_video_output.cpp b/src/plugins/sensing/core/livekit_video_output.cpp index 87170fe0f7..8dd38d371d 100644 --- a/src/plugins/sensing/core/livekit_video_output.cpp +++ b/src/plugins/sensing/core/livekit_video_output.cpp @@ -3,7 +3,7 @@ #include "livekit_video_output.hpp" -#include "jetson_encoder.hpp" +#include "inc/sensing_core/jetson_encoder.hpp" #include "rgba_box_downscale.cuh" #include @@ -14,6 +14,7 @@ #include #include #include +#include #include #include #include @@ -62,6 +63,8 @@ struct LiveKitVideoOutput::Impl out_height = config.source_height / config.scale_div; if (sink->width() != out_width || sink->height() != out_height) throw std::runtime_error("LiveKit output: the track size does not match the output size"); + if (config.input == PixelKind::DmaBufNv12 && config.scale_div != 1) + throw std::runtime_error("LiveKit output: DMA-BUF input cannot be reduced (scale_div must be 1)"); period_ns = 1'000'000'000ull / config.fps; if (config.bitrate_bps < kMinBitrateBps) { @@ -86,11 +89,14 @@ struct LiveKitVideoOutput::Impl encoder_config.fps = config.fps; encoder_config.gop = config.gop; encoder_config.full_range = config.full_range; + encoder_config.input = config.input; encoder = std::make_unique(encoder_config); applied_bitrate = config.bitrate_bps; for (auto& slot : slots) { + if (config.input != PixelKind::CudaRgba) + break; // DMA-BUF frames are not copied check_cuda(cudaMallocPitch(reinterpret_cast(&slot.ptr), &slot.pitch, static_cast(config.source_width) * 4, config.source_height), "cudaMallocPitch(snapshot)"); @@ -147,12 +153,13 @@ struct LiveKitVideoOutput::Impl } // Polling thread. - void submit(const CaptureFrame& frame) + void submit(CaptureFrame& frame) { + const bool dma = config.input == PixelKind::DmaBufNv12; const auto* rgba = std::get_if(&frame.pixels); - if (rgba == nullptr) + if (dma ? !std::holds_alternative(frame.pixels) : rgba == nullptr) { - fail("LiveKit output: this pixel type has no encoder input yet"); + fail("LiveKit output: the frame's pixel type does not match the encoder input"); return; } ++offered; @@ -167,6 +174,27 @@ struct LiveKitVideoOutput::Impl return; } + if (dma) + { + const uint64_t accepted_ns = frame.monotonic_ns; + // A frame still waiting is replaced by this newer one. Its lease is released after the lock is dropped: + // that is a camera API call, and must not run inside the section the worker waits on. + std::optional replaced_frame; + { + std::lock_guard lock(mutex); + if (pending_dma) + ++replaced; + replaced_frame = std::move(pending_dma); + pending_dma = std::move(frame); + } + replaced_frame.reset(); + ++snapshots; + have_last = true; + last_accepted_ns = accepted_ns; + wake.notify_one(); + return; + } + try { std::lock_guard lock(mutex); @@ -222,22 +250,30 @@ struct LiveKitVideoOutput::Impl for (;;) { int index = -1; + std::optional dma; bool flush_now = false; { std::unique_lock lock(mutex); // Frames in the encoder come out a little later: poll for them soon, and otherwise // wake now and then to pick up the receiving side's requests. const auto wait = in_encoder.empty() ? std::chrono::milliseconds(100) : std::chrono::milliseconds(2); - wake.wait_for(lock, wait, [&] { return stopping || pending >= 0 || flush_requested; }); + wake.wait_for(lock, wait, [&] { return stopping || pending >= 0 || pending_dma || flush_requested; }); if (stopping) return; index = pending; pending = -1; inflight = index; + dma = std::move(pending_dma); + pending_dma.reset(); flush_now = flush_requested; } apply_feedback(); + if (dma) + { + encode_dma(*dma); + dma.reset(); + } if (index >= 0) { encode(slots[index]); @@ -325,8 +361,32 @@ struct LiveKitVideoOutput::Impl ++encoder_busy; return; } - in_encoder.push_back( - { stamp_ns / 1000, static_cast(slot.sequence), slot.capture_tsc, slot.unix_ns / 1000 }); + track_in_flight( + static_cast(slot.sequence), slot.capture_tsc, slot.timestamp_ns / 1000, slot.unix_ns / 1000); + } + + // Worker thread: a leased DMA-BUF frame goes to the encoder as it is, and its lease stays with the encoder. + void encode_dma(CaptureFrame& frame) + { + if (keyframe_wanted) + { + encoder->request_keyframe(); + keyframe_wanted = false; + } + const int fd = std::get(frame.pixels).fd; + if (!encoder->submit_dmabuf(fd, static_cast(frame.monotonic_ns), std::move(frame.lease))) + { + ++encoder_busy; + return; + } + track_in_flight( + static_cast(frame.sequence), frame.sensor_time, frame.monotonic_ns / 1000, frame.unix_ns / 1000); + } + + // Worker thread: remember the capture a submitted frame came from, for the unit that comes back. + void track_in_flight(uint32_t sequence, uint64_t capture_tsc, uint64_t monotonic_us, uint64_t unix_us) + { + in_encoder.push_back({ sequence, capture_tsc, monotonic_us, unix_us }); } // Worker thread: one encoded unit becomes one published frame, with the sequence and capture time @@ -368,7 +428,7 @@ struct LiveKitVideoOutput::Impl frame.data = unit.data.data(); frame.size = unit.data.size(); frame.keyframe = unit.keyframe; - frame.timestamp_us = static_cast(meta.stamp_us); + frame.timestamp_us = meta.monotonic_us; frame.frame_id = meta.sequence; frame.capture_tsc = meta.capture_tsc; frame.user_timestamp_us = meta.unix_us; @@ -393,10 +453,11 @@ struct LiveKitVideoOutput::Impl struct FrameMeta { - int64_t stamp_us = 0; uint32_t sequence = 0; uint64_t capture_tsc = 0; - /// Unix time of the same instant as stamp_us, for the frame's application timestamp. + /// CLOCK_MONOTONIC when the frame was handed over, for the frame's transport timestamp. + uint64_t monotonic_us = 0; + /// Unix time when the frame was handed over, for the frame's application timestamp. uint64_t unix_us = 0; }; @@ -413,6 +474,8 @@ struct LiveKitVideoOutput::Impl std::chrono::steady_clock::time_point last_bitrate_change = std::chrono::steady_clock::now(); Slot slots[2]; + /// DMA-BUF input only: the newest accepted frame, waiting for the worker. Guarded by `mutex`. + std::optional pending_dma; uint8_t* d_scaled = nullptr; size_t scaled_pitch = 0; cudaStream_t snapshot_stream = nullptr; @@ -473,7 +536,7 @@ LiveKitVideoOutput::LiveKitVideoOutput(std::unique_ptr sink, con LiveKitVideoOutput::~LiveKitVideoOutput() = default; -void LiveKitVideoOutput::submit(const CaptureFrame& frame) +void LiveKitVideoOutput::submit(CaptureFrame& frame) { m_impl->submit(frame); } diff --git a/src/plugins/sensing/core/livekit_video_output.hpp b/src/plugins/sensing/core/livekit_video_output.hpp index 9f56ee6968..30bfd79f58 100644 --- a/src/plugins/sensing/core/livekit_video_output.hpp +++ b/src/plugins/sensing/core/livekit_video_output.hpp @@ -22,6 +22,8 @@ struct LiveKitOutputConfig uint32_t source_height = 0; /// Uniform output-only reduction; 1 publishes the sensor's native size. uint32_t scale_div = 1; + /// The frames submit() will be given; DMA-BUF input needs scale_div 1 (no CPU or CUDA copy exists to reduce). + PixelKind input = PixelKind::CudaRgba; /// Publication rate cap, independent of the capture rate. uint32_t fps = 30; int gpu_id = 0; @@ -51,9 +53,10 @@ struct LiveKitOutputStats /** * @brief Encodes one sensor's frames on hardware and feeds them to a pre-encoded video track. * - * submit() runs on the caller's polling loop. It takes a snapshot of the borrowed camera frame in - * GPU memory and finishes that copy before returning, because the camera may overwrite its slot at - * the second following poll. Everything else -- the optional reduction, the hardware encode and the + * submit() runs on the caller's polling loop. A pixel-buffer frame (SIPL) is copied into a snapshot in GPU + * memory before returning, because the camera may overwrite its slot at the second following poll. A DMA-BUF + * frame (Argus) is not copied: it goes to the encoder as it is, and its lease returns the buffer to the camera + * when the encoder is done with it. Everything else -- the optional reduction, the hardware encode and the * SDK call -- happens on this output's own worker thread, so a slow network cannot hold up capture * or the other sensors. * @@ -79,9 +82,10 @@ class LiveKitVideoOutput LiveKitVideoOutput(const LiveKitVideoOutput&) = delete; LiveKitVideoOutput& operator=(const LiveKitVideoOutput&) = delete; - /// Never blocks on the network. A frame sooner than the publication rate allows is skipped. Only borrowed - /// CUDA RGBA pixels are accepted for now; they are copied out before this returns. - void submit(const CaptureFrame& frame); + /// Never blocks on the network. A frame sooner than the publication rate allows is skipped. Borrowed CUDA + /// RGBA pixels are copied out before this returns; a leased DMA-BUF frame's lease is moved into the output and + /// released once the encoder is finished with it (or at once if the frame is dropped). + void submit(CaptureFrame& frame); LiveKitOutputStats stats() const; diff --git a/src/plugins/sensing/core/sensing_camera.cpp b/src/plugins/sensing/core/sensing_camera.cpp index 693888abf7..dbbaa049a4 100644 --- a/src/plugins/sensing/core/sensing_camera.cpp +++ b/src/plugins/sensing/core/sensing_camera.cpp @@ -4,7 +4,6 @@ #include "inc/sensing_core/sensing_camera.hpp" #include "livekit_video_output.hpp" -#include "sipl_capture.hpp" #include @@ -22,20 +21,6 @@ namespace sensing namespace { -SiplConfig make_sipl_config(const SensingConfig& config) -{ - SiplConfig sipl{}; - sipl.platform_config_json = config.platform_config_json; - sipl.platform_config_name = config.platform_config_name; - sipl.link_masks = config.link_masks; - sipl.nito_dir = config.nito_dir; - sipl.gpu_id = config.gpu_id; - sipl.full_range = config.full_range; - sipl.swap_uv = config.swap_uv; - sipl.isp0_buffers = config.isp0_buffers; - return sipl; -} - LiveKitOutputConfig make_output_config(const SensingConfig& config, const CameraInfo& sensor) { LiveKitOutputConfig output; @@ -46,6 +31,7 @@ LiveKitOutputConfig make_output_config(const SensingConfig& config, const Camera output.gpu_id = config.gpu_id; output.bitrate_bps = config.livekit.bitrate_bps; output.peak_bitrate_bps = config.livekit.bitrate_bps + config.livekit.bitrate_bps / 2; + output.input = sensor.pixels; output.gop = config.gop; output.full_range = config.full_range; return output; @@ -53,12 +39,17 @@ LiveKitOutputConfig make_output_config(const SensingConfig& config, const Camera } // namespace -SensingCamera::SensingCamera(const SensingConfig& config, const std::vector& streams) : m_config(config) +SensingCamera::SensingCamera(const SensingConfig& config, + std::unique_ptr camera, + const std::vector& streams) + : m_config(config) { if (streams.empty()) throw std::runtime_error("SensingCamera: no streams requested"); - m_camera = std::make_unique(make_sipl_config(config)); + m_camera = std::move(camera); + if (!m_camera) + throw std::runtime_error("SensingCamera: no camera backend"); const auto& sensors = m_camera->cameras(); // Everything that needs no network is checked first, so a bad name or divisor fails before any @@ -77,10 +68,13 @@ SensingCamera::SensingCamera(const SensingConfig& config, const std::vectorwidth, found->height, config.livekit.scale_div); + if (found->pixels == PixelKind::DmaBufNv12 && config.livekit.scale_div != 1) + throw std::runtime_error("sensor " + std::to_string(found->id) + + " delivers DMA-BUF frames, which cannot be reduced: use --livekit-scale-div=1"); } m_livekit = LiveKitSession::connect(config.livekit.url, config.livekit.token); diff --git a/src/plugins/sensing/tests/CMakeLists.txt b/src/plugins/sensing/tests/CMakeLists.txt index 966e8e9089..33819ad40b 100644 --- a/src/plugins/sensing/tests/CMakeLists.txt +++ b/src/plugins/sensing/tests/CMakeLists.txt @@ -14,3 +14,14 @@ target_include_directories(sensing_backend_tests PRIVATE ${CMAKE_CURRENT_SOURCE_ target_link_libraries(sensing_backend_tests PRIVATE Catch2::Catch2WithMain) catch_discover_tests(sensing_backend_tests) + +# The buffer owner is header-only and SDK-free, tested against a fake port. +add_executable(sensing_buffer_owner_tests + test_buffer_owner.cpp +) + +target_include_directories(sensing_buffer_owner_tests PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/..) + +target_link_libraries(sensing_buffer_owner_tests PRIVATE Catch2::Catch2WithMain Threads::Threads) + +catch_discover_tests(sensing_buffer_owner_tests) diff --git a/src/plugins/sensing/tests/test_buffer_owner.cpp b/src/plugins/sensing/tests/test_buffer_owner.cpp new file mode 100644 index 0000000000..e5fa043184 --- /dev/null +++ b/src/plugins/sensing/tests/test_buffer_owner.cpp @@ -0,0 +1,1020 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "core/buffer_owner.hpp" + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using Catch::Matchers::ContainsSubstring; +using plugins::sensing::BufferOwner; +using plugins::sensing::BufferPort; +using plugins::sensing::BufferReturnMailbox; +using plugins::sensing::CaptureFrame; +using plugins::sensing::DmaBufNv12Pixels; + +namespace +{ + +using namespace std::chrono_literals; + +constexpr auto kWait = 5ms; +constexpr auto kDeadline = 5s; +constexpr size_t kMaxSlots = 16; + +template +bool wait_until(Pred pred, + std::chrono::milliseconds limit = std::chrono::duration_cast(kDeadline)) +{ + const auto end = std::chrono::steady_clock::now() + limit; + while (!pred()) + { + if (std::chrono::steady_clock::now() >= end) + return false; + std::this_thread::sleep_for(1ms); + } + return true; +} + +/// A bounded pool of slots standing in for the stream. Frames are produced while `budget` is positive and a slot +/// is free; otherwise acquire sleeps the requested timeout and reports Timeout. +class FakePort : public BufferPort +{ +public: + explicit FakePort(size_t pool_size) : m_pool_size(pool_size) + { + } + + Acquired acquire(uint64_t timeout_ns) override + { + Guard guard(*this); + Acquired out; + int slot = -1; + bool produce = false; + { + std::lock_guard lock(m_mutex); + ++m_acquire_calls; + if (m_throw_on_acquire) + throw std::runtime_error("acquire threw"); + for (size_t i = 0; i < m_pool_size; ++i) + { + if (!m_busy[i]) + { + slot = static_cast(i); + break; + } + } + if (slot >= 0 && m_budget > 0) + { + --m_budget; + m_busy[slot] = true; + ++m_acquired[slot]; + ++m_acquired_total; + m_events.push_back({ Event::Kind::Acquire, static_cast(slot) }); + produce = true; + if (m_fail_next) + { + m_fail_next = false; + out.result = Result::Failed; + out.holds_buffer = true; + out.slot = static_cast(slot); + out.error = "stream exploded"; + return out; + } + } + } + if (!produce) + { + std::this_thread::sleep_for(std::chrono::nanoseconds(timeout_ns)); + out.result = Result::Timeout; + return out; + } + std::this_thread::sleep_for(200us); + out.result = Result::Frame; + out.holds_buffer = true; + out.slot = static_cast(slot); + out.frame.sequence = m_sequence.fetch_add(1) + 1; + out.frame.pixels = DmaBufNv12Pixels{ slot }; + return out; + } + + bool release(uint32_t slot) override + { + Guard guard(*this); + std::lock_guard lock(m_mutex); + ++m_release_calls; + m_events.push_back({ Event::Kind::Release, slot }); + if (m_throw_on_release) + throw std::runtime_error("release threw"); + if (!m_release_ok) + return false; // refused: the buffer stays out and is not counted as released + if (slot >= m_pool_size || !m_busy[slot]) + { + ++m_bad_releases; + return true; + } + m_busy[slot] = false; + ++m_released[slot]; + ++m_released_total; + return true; + } + + bool stop_capture() override + { + Guard guard(*this); + std::lock_guard lock(m_mutex); + ++m_stop_calls; + m_events.push_back({ Event::Kind::StopBegin, 0 }); + return m_stop_ok; + } + + bool wait_idle() override + { + Guard guard(*this); + std::lock_guard lock(m_mutex); + ++m_idle_calls; + m_events.push_back({ Event::Kind::WaitIdle, 0 }); + if (m_throw_on_idle) + throw std::runtime_error("wait_idle threw"); + return m_idle_ok; + } + + void set_budget(int64_t n) + { + std::lock_guard lock(m_mutex); + m_budget = n; + } + void set_throw_on_acquire(bool v) + { + std::lock_guard lock(m_mutex); + m_throw_on_acquire = v; + } + void set_fail_next(bool v) + { + std::lock_guard lock(m_mutex); + m_fail_next = v; + } + void set_release_ok(bool v) + { + std::lock_guard lock(m_mutex); + m_release_ok = v; + } + void set_throw_on_release(bool v) + { + std::lock_guard lock(m_mutex); + m_throw_on_release = v; + } + void set_stop_ok(bool v) + { + std::lock_guard lock(m_mutex); + m_stop_ok = v; + } + void set_idle_ok(bool v) + { + std::lock_guard lock(m_mutex); + m_idle_ok = v; + } + void set_throw_on_idle(bool v) + { + std::lock_guard lock(m_mutex); + m_throw_on_idle = v; + } + int idle_calls() const + { + std::lock_guard lock(m_mutex); + return m_idle_calls; + } + + uint64_t acquire_calls() const + { + std::lock_guard lock(m_mutex); + return m_acquire_calls; + } + uint64_t acquired_total() const + { + std::lock_guard lock(m_mutex); + return m_acquired_total; + } + uint64_t released_total() const + { + std::lock_guard lock(m_mutex); + return m_released_total; + } + uint64_t release_calls() const + { + std::lock_guard lock(m_mutex); + return m_release_calls; + } + uint64_t bad_releases() const + { + std::lock_guard lock(m_mutex); + return m_bad_releases; + } + /// Port calls in the order they happened, across all slots. + struct Event + { + enum class Kind + { + Acquire, + Release, + StopBegin, + WaitIdle, + }; + Kind kind; + uint32_t slot; + }; + std::vector events() const + { + std::lock_guard lock(m_mutex); + return m_events; + } + int stop_calls() const + { + std::lock_guard lock(m_mutex); + return m_stop_calls; + } + uint64_t acquired(size_t slot) const + { + std::lock_guard lock(m_mutex); + return m_acquired[slot]; + } + uint64_t released(size_t slot) const + { + std::lock_guard lock(m_mutex); + return m_released[slot]; + } + int max_in_flight() const + { + return m_max_in_flight.load(); + } + std::set threads() const + { + std::lock_guard lock(m_threads_mutex); + return m_threads; + } + + /// Every acquired buffer was released exactly once, slot by slot. + bool balanced() const + { + std::lock_guard lock(m_mutex); + for (size_t i = 0; i < m_pool_size; ++i) + { + if (m_acquired[i] != m_released[i] || m_busy[i]) + return false; + } + return m_bad_releases == 0; + } + +private: + /// Records the calling thread and how many calls are in flight at once. + struct Guard + { + explicit Guard(FakePort& port) : p(port) + { + const int now = ++p.m_in_flight; + int seen = p.m_max_in_flight.load(); + while (now > seen && !p.m_max_in_flight.compare_exchange_weak(seen, now)) + { + } + std::lock_guard lock(p.m_threads_mutex); + p.m_threads.insert(std::this_thread::get_id()); + } + ~Guard() + { + --p.m_in_flight; + } + FakePort& p; + }; + + const size_t m_pool_size; + mutable std::mutex m_mutex; + std::array m_busy{}; + std::array m_acquired{}; + std::array m_released{}; + int64_t m_budget = 0; + bool m_throw_on_acquire = false; + bool m_fail_next = false; + bool m_release_ok = true; + bool m_stop_ok = true; + bool m_throw_on_release = false; + bool m_idle_ok = true; + bool m_throw_on_idle = false; + int m_idle_calls = 0; + std::vector m_events; + uint64_t m_acquire_calls = 0; + uint64_t m_acquired_total = 0; + uint64_t m_released_total = 0; + uint64_t m_release_calls = 0; + uint64_t m_bad_releases = 0; + int m_stop_calls = 0; + std::atomic m_sequence{ 0 }; + + std::atomic m_in_flight{ 0 }; + std::atomic m_max_in_flight{ 0 }; + mutable std::mutex m_threads_mutex; + std::set m_threads; +}; + +constexpr int64_t kUnlimited = INT64_MAX / 2; + +/// Polls the owner until `count` frames have been taken, holding them (and their leases). +std::vector take_frames(BufferOwner& owner, size_t count) +{ + std::vector held; + wait_until( + [&] + { + if (auto frame = owner.take_latest()) + held.push_back(std::move(*frame)); + return held.size() >= count; + }); + return held; +} + +bool take_throws(BufferOwner& owner, std::string* what = nullptr) +{ + return wait_until( + [&] + { + try + { + owner.take_latest(); + } + catch (const std::runtime_error& e) + { + if (what) + *what = e.what(); + return true; + } + return false; + }); +} + +} // namespace + +TEST_CASE("every port call runs on one owner thread, never two at once", "[buffer_owner]") +{ + FakePort port(8); + port.set_budget(kUnlimited); + BufferOwner owner(port, 8, kWait); + owner.start(); + + std::atomic stop{ false }; + std::vector droppers; + for (int t = 0; t < 4; ++t) + { + droppers.emplace_back( + [&] + { + while (!stop) + { + // The temporary's lease goes back on this thread, not the owner's. + owner.take_latest(); + std::this_thread::sleep_for(100us); + } + }); + } + REQUIRE(wait_until([&] { return port.released_total() > 100; })); + owner.request_stop(); + CHECK(owner.wait_stopped(kDeadline)); + stop = true; + for (auto& t : droppers) + t.join(); + owner.close(); + + const auto threads = port.threads(); + REQUIRE(threads.size() == 1); + CHECK(*threads.begin() != std::this_thread::get_id()); + CHECK(port.max_in_flight() == 1); +} + +TEST_CASE("concurrent lease returns release every slot exactly once", "[buffer_owner]") +{ + FakePort port(8); + port.set_budget(kUnlimited); + BufferOwner owner(port, 8, kWait); + owner.start(); + + std::atomic stop{ false }; + std::vector consumers; + for (int t = 0; t < 6; ++t) + { + consumers.emplace_back( + [&] + { + std::vector held; + while (!stop) + { + if (auto frame = owner.take_latest()) + { + held.push_back(std::move(*frame)); + if (held.size() >= 2) + held.clear(); // drops leases in bulk + } + std::this_thread::sleep_for(50us); + } + }); + } + std::this_thread::sleep_for(300ms); + port.set_budget(0); + stop = true; + for (auto& t : consumers) + t.join(); + owner.close(); + + CHECK(port.acquired_total() > 50); + CHECK(port.acquired_total() == port.released_total()); + CHECK(port.balanced()); + CHECK(owner.rejected_returns() == 0); + CHECK(port.max_in_flight() == 1); +} + +TEST_CASE("a newer frame replaces an older one and the replaced buffer is released once", "[buffer_owner]") +{ + FakePort port(8); + port.set_budget(5); + BufferOwner owner(port, 8, kWait); + owner.start(); + + REQUIRE(wait_until([&] { return port.acquired_total() == 5; })); + REQUIRE(wait_until([&] { return port.released_total() == 4; })); + + auto frame = owner.take_latest(); + REQUIRE(frame.has_value()); + CHECK(frame->sequence == 5); + CHECK(port.released_total() == 4); + CHECK_FALSE(owner.take_latest().has_value()); + + frame.reset(); + REQUIRE(wait_until([&] { return port.released_total() == 5; })); + CHECK(port.balanced()); + CHECK(owner.rejected_returns() == 0); + owner.close(); +} + +TEST_CASE("an exhausted pool stops acquiring and resumes when a lease returns", "[buffer_owner]") +{ + constexpr size_t kPool = 3; + FakePort port(kPool); + port.set_budget(kUnlimited); + BufferOwner owner(port, kPool, kWait); + owner.start(); + + std::vector held = take_frames(owner, kPool); + REQUIRE(held.size() == kPool); + + // Let any in-flight acquire finish, then the count must stay put. + std::this_thread::sleep_for(50ms); + const uint64_t settled = port.acquire_calls(); + std::this_thread::sleep_for(150ms); + CHECK(port.acquire_calls() == settled); + CHECK_FALSE(owner.take_latest().has_value()); + + held.pop_back(); + REQUIRE(wait_until([&] { return port.acquire_calls() > settled; })); + auto next = take_frames(owner, 1); + CHECK(next.size() == 1); + + held.clear(); + next.clear(); + owner.close(); + CHECK(port.balanced()); + CHECK(owner.rejected_returns() == 0); +} + +TEST_CASE("stop clears the latest frame; held leases still return afterwards", "[buffer_owner]") +{ + FakePort port(8); + port.set_budget(kUnlimited); + BufferOwner owner(port, 8, kWait); + owner.start(); + + std::vector held = take_frames(owner, 2); + REQUIRE(held.size() == 2); + const uint64_t acquired_before_stop = port.acquired_total(); + REQUIRE(acquired_before_stop >= 2); + + owner.request_stop(); + REQUIRE(owner.wait_stopped(kDeadline)); + CHECK(port.stop_calls() == 1); + CHECK_FALSE(owner.take_latest().has_value()); + + // Nothing is acquired once stopped. + const uint64_t acquired_at_stop = port.acquired_total(); + std::this_thread::sleep_for(50ms); + CHECK(port.acquired_total() == acquired_at_stop); + + // Every acquired buffer comes back while the owner keeps draining, before close(). + const size_t held_count = held.size(); + held.clear(); + REQUIRE(wait_until([&] { return port.released_total() == port.acquired_total(); })); + CHECK(held_count == 2); + CHECK(port.stop_calls() == 1); + + owner.close(); + CHECK(port.balanced()); + CHECK(owner.rejected_returns() == 0); +} + +TEST_CASE("close releases every buffer posted before it and is idempotent", "[buffer_owner]") +{ + SECTION("buffers posted just before close") + { + FakePort port(8); + port.set_budget(kUnlimited); + BufferOwner owner(port, 8, kWait); + owner.start(); + std::vector held = take_frames(owner, 4); + REQUIRE(held.size() == 4); + port.set_budget(0); + held.clear(); + owner.close(); + CHECK(port.balanced()); + owner.close(); + CHECK(port.balanced()); + CHECK(owner.rejected_returns() == 0); + } + SECTION("destructor without start") + { + FakePort port(4); + { + BufferOwner owner(port, 4, kWait); + } + CHECK(port.acquire_calls() == 0); + CHECK(port.threads().empty()); + } + SECTION("close without start") + { + FakePort port(4); + BufferOwner owner(port, 4, kWait); + owner.close(); + owner.close(); + CHECK(port.threads().empty()); + } + SECTION("destructor without close") + { + FakePort port(4); + port.set_budget(kUnlimited); + { + BufferOwner owner(port, 4, kWait); + owner.start(); + REQUIRE(wait_until([&] { return port.acquired_total() >= 2; })); + } + CHECK(port.balanced()); + } + SECTION("stop without start completes") + { + FakePort port(4); + BufferOwner owner(port, 4, kWait); + owner.request_stop(); + CHECK(owner.wait_stopped(kDeadline)); + CHECK(port.stop_calls() == 0); + } + SECTION("starting twice throws") + { + FakePort port(4); + BufferOwner owner(port, 4, kWait); + owner.start(); + CHECK_THROWS_AS(owner.start(), std::logic_error); + owner.close(); + } +} + +TEST_CASE("failures surface from take_latest and never lose a buffer", "[buffer_owner]") +{ + SECTION("acquire failed while holding a buffer") + { + FakePort port(4); + port.set_fail_next(true); + port.set_budget(1); + BufferOwner owner(port, 4, kWait); + owner.start(); + std::string what; + REQUIRE(take_throws(owner, &what)); + CHECK_THAT(what, ContainsSubstring("stream exploded")); + REQUIRE(wait_until([&] { return port.released_total() == 1; })); + owner.close(); + CHECK(port.released_total() == 1); + CHECK(port.balanced()); + } + SECTION("release refused") + { + FakePort port(4); + port.set_release_ok(false); + port.set_budget(2); // the second frame replaces, and so releases, the first + BufferOwner owner(port, 4, kWait); + owner.start(); + std::string what; + REQUIRE(take_throws(owner, &what)); + CHECK_THAT(what, ContainsSubstring("refused")); + owner.close(); + } + SECTION("stop_capture refused") + { + FakePort port(4); + port.set_stop_ok(false); + BufferOwner owner(port, 4, kWait); + owner.start(); + owner.request_stop(); + CHECK_FALSE(owner.wait_stopped(kDeadline)); + CHECK(port.stop_calls() == 1); + owner.close(); + } + SECTION("wait_stopped times out when nobody asked to stop") + { + FakePort port(4); + BufferOwner owner(port, 4, kWait); + owner.start(); + CHECK_FALSE(owner.wait_stopped(20ms)); + owner.close(); + } + SECTION("a throwing acquire does not terminate the process") + { + FakePort port(4); + port.set_throw_on_acquire(true); + BufferOwner owner(port, 4, kWait); + owner.start(); + std::string what; + REQUIRE(take_throws(owner, &what)); + CHECK_THAT(what, ContainsSubstring("acquire threw")); + owner.request_stop(); + CHECK(owner.wait_stopped(kDeadline)); + owner.close(); + } +} + +TEST_CASE("a lease keeps its keepalive object alive; the owner holds it only weakly", "[buffer_owner]") +{ + struct Probe + { + explicit Probe(std::atomic& flag) : destroyed(flag) + { + } + ~Probe() + { + destroyed = true; + } + std::atomic& destroyed; + }; + + SECTION("with no leases the owner does not keep it") + { + FakePort port(4); + BufferOwner owner(port, 4, kWait); + std::atomic destroyed{ false }; + std::shared_ptr keep = std::make_shared(destroyed); + owner.set_keepalive(keep); + owner.start(); + keep.reset(); + CHECK(destroyed); + owner.close(); + } + SECTION("a held lease keeps it until dropped") + { + FakePort port(4); + port.set_budget(1); + BufferOwner owner(port, 4, kWait); + std::atomic destroyed{ false }; + std::shared_ptr keep = std::make_shared(destroyed); + owner.set_keepalive(keep); + owner.start(); + + std::vector held = take_frames(owner, 1); + REQUIRE(held.size() == 1); + keep.reset(); + CHECK_FALSE(destroyed); + + held.clear(); + CHECK(destroyed); + REQUIRE(wait_until([&] { return port.released_total() == 1; })); + owner.close(); + CHECK(port.balanced()); + } +} + +TEST_CASE("the mailbox rejects a post beyond its capacity", "[buffer_owner]") +{ + BufferReturnMailbox mailbox(2); + CHECK(mailbox.post(0)); + CHECK(mailbox.post(1)); + CHECK_FALSE(mailbox.post(1)); + CHECK(mailbox.rejected() == 1); + std::vector out; + mailbox.take(out); + CHECK(out == std::vector{ 0, 1 }); +} + +TEST_CASE("stop returns the held frame's buffer before stop_capture begins", "[buffer_owner]") +{ + FakePort port(8); + port.set_budget(kUnlimited); + BufferOwner owner(port, 8, kWait); + owner.start(); + + // The consumer never takes a frame, so the latest one is always held by the owner. + REQUIRE(wait_until([&] { return port.released_total() >= 3; })); + owner.request_stop(); + REQUIRE(owner.wait_stopped(kDeadline)); + + const auto events = port.events(); + size_t stop_at = events.size(); + for (size_t i = 0; i < events.size(); ++i) + { + if (events[i].kind == FakePort::Event::Kind::StopBegin) + { + stop_at = i; + break; + } + } + REQUIRE(stop_at < events.size()); + + // The last buffer acquired before the stop is the latest frame. + bool found_acquire = false; + uint32_t latest_slot = 0; + for (size_t i = 0; i < stop_at; ++i) + { + if (events[i].kind == FakePort::Event::Kind::Acquire) + { + found_acquire = true; + latest_slot = events[i].slot; + } + } + REQUIRE(found_acquire); + size_t acquired_before = 0; + size_t released_before = 0; + bool latest_released_first = false; + size_t last_acquire = 0; + for (size_t i = 0; i < stop_at; ++i) + if (events[i].kind == FakePort::Event::Kind::Acquire) + last_acquire = i; + for (size_t i = 0; i < stop_at; ++i) + { + acquired_before += events[i].kind == FakePort::Event::Kind::Acquire; + released_before += events[i].kind == FakePort::Event::Kind::Release; + if (i > last_acquire && events[i].kind == FakePort::Event::Kind::Release && events[i].slot == latest_slot) + latest_released_first = true; + } + CHECK(latest_released_first); + CHECK(released_before == acquired_before); + for (size_t i = stop_at + 1; i < events.size(); ++i) + CHECK(events[i].kind != FakePort::Event::Kind::Acquire); + + owner.close(); + CHECK(port.balanced()); + CHECK(owner.rejected_returns() == 0); +} + +TEST_CASE("a refused or throwing release keeps the buffer out and ends acquisition", "[buffer_owner]") +{ + constexpr size_t kPool = 4; + FakePort port(kPool); + port.set_budget(kUnlimited); + + std::string expected; + SECTION("refused") + { + port.set_release_ok(false); + expected = "refused"; + } + SECTION("throws") + { + port.set_throw_on_release(true); + expected = "release threw"; + } + + BufferOwner owner(port, kPool, kWait); + owner.start(); + std::string what; + REQUIRE(take_throws(owner, &what)); + CHECK_THAT(what, ContainsSubstring(expected)); + + // Let anything in flight finish; after that nothing more is acquired or released. + std::this_thread::sleep_for(50ms); + const uint64_t acquires = port.acquire_calls(); + const uint64_t acquired = port.acquired_total(); + const uint64_t releases = port.release_calls(); + std::this_thread::sleep_for(150ms); + CHECK(port.acquire_calls() == acquires); + CHECK(port.release_calls() == releases); + CHECK(acquired <= kPool); + for (size_t i = 0; i < kPool; ++i) + CHECK(port.acquired(i) <= 1); // the refused slot is never handed out again + CHECK(port.release_calls() == 1); // the refused buffer is not retried + CHECK(port.released_total() == 0); // nothing the stream refused counts as returned + CHECK_THROWS_AS(owner.take_latest(), std::runtime_error); + + owner.request_stop(); + CHECK(owner.wait_stopped(kDeadline)); + owner.close(); + CHECK(port.acquire_calls() == acquires); + CHECK(port.acquired_total() == acquired); + CHECK(port.released_total() == 0); + for (size_t i = 0; i < kPool; ++i) + CHECK(port.acquired(i) <= 1); + CHECK_FALSE(owner.closed_cleanly()); // the unresolved buffers are still out + CHECK(port.idle_calls() == 1); +} + +TEST_CASE("a refused release with a full pool does not over-acquire", "[buffer_owner]") +{ + // The stream has more slots than the owner's pool, so only the owner's own count can stop it. + constexpr size_t kPool = 2; + FakePort port(8); + port.set_budget(kUnlimited); + port.set_release_ok(false); + BufferOwner owner(port, kPool, kWait); + owner.start(); + REQUIRE(take_throws(owner)); + std::this_thread::sleep_for(50ms); + const uint64_t acquires = port.acquire_calls(); + std::this_thread::sleep_for(100ms); + CHECK(port.acquire_calls() == acquires); + CHECK(port.acquired_total() <= kPool); + owner.close(); + CHECK_FALSE(owner.closed_cleanly()); +} + +namespace +{ + +/// Index of the single event of `kind`, or events.size() if there is none or more than one. +size_t only_event(const std::vector& events, FakePort::Event::Kind kind) +{ + size_t found = events.size(); + for (size_t i = 0; i < events.size(); ++i) + { + if (events[i].kind == kind) + { + if (found != events.size()) + return events.size(); + found = i; + } + } + return found; +} + +} // namespace + +TEST_CASE("only close waits for the stream to go idle, after every buffer is back", "[buffer_owner]") +{ + FakePort port(8); + port.set_budget(kUnlimited); + BufferOwner owner(port, 8, kWait); + owner.start(); + + // Frames an encoder still holds when stop is requested; they come back before close(). + std::vector held = take_frames(owner, 2); + REQUIRE(held.size() == 2); + + owner.request_stop(); + REQUIRE(owner.wait_stopped(kDeadline)); + CHECK(port.idle_calls() == 0); + + held.clear(); + owner.close(); + + CHECK(port.idle_calls() == 1); + CHECK(owner.closed_cleanly()); + CHECK(port.balanced()); + CHECK(port.max_in_flight() == 1); + CHECK(port.threads().size() == 1); + CHECK_FALSE(owner.take_latest().has_value()); + + const auto events = port.events(); + const size_t stop_at = only_event(events, FakePort::Event::Kind::StopBegin); + const size_t idle_at = only_event(events, FakePort::Event::Kind::WaitIdle); + REQUIRE(stop_at < events.size()); + REQUIRE(idle_at < events.size()); + CHECK(stop_at < idle_at); + size_t acquires = 0; + size_t releases = 0; + for (size_t i = 0; i < events.size(); ++i) + { + acquires += events[i].kind == FakePort::Event::Kind::Acquire; + if (events[i].kind == FakePort::Event::Kind::Release) + { + ++releases; + CHECK(i < idle_at); + } + if (i > stop_at) + CHECK(events[i].kind != FakePort::Event::Kind::Acquire); + } + CHECK(acquires == releases); + CHECK(idle_at == events.size() - 1); +} + +TEST_CASE("close with leases never dropped ends the owner and reports it unclean", "[buffer_owner]") +{ + FakePort port(8); + port.set_budget(kUnlimited); + BufferOwner owner(port, 8, kWait); + owner.start(); + std::vector held = take_frames(owner, 2); + REQUIRE(held.size() == 2); + + SECTION("after a stop request") + { + owner.request_stop(); + REQUIRE(owner.wait_stopped(kDeadline)); + owner.close(); + CHECK(port.idle_calls() == 1); + } + SECTION("without a stop request") + { + owner.close(); + CHECK(port.idle_calls() == 0); + } + CHECK_FALSE(owner.closed_cleanly()); + CHECK(port.released_total() + 2 == port.acquired_total()); + held.clear(); // lands in a mailbox nobody drains; must be harmless + CHECK(owner.rejected_returns() == 0); +} + +TEST_CASE("the outcome of wait_idle decides closed_cleanly", "[buffer_owner]") +{ + FakePort port(4); + port.set_budget(kUnlimited); + bool expect_clean = false; + std::string expected; + SECTION("returns false") + { + port.set_idle_ok(false); + expected = "did not go idle"; + } + SECTION("throws") + { + port.set_throw_on_idle(true); + expected = "wait_idle threw"; + } + SECTION("returns true") + { + expect_clean = true; + } + + BufferOwner owner(port, 4, kWait); + owner.start(); + REQUIRE(wait_until([&] { return port.acquired_total() >= 2; })); + owner.request_stop(); + REQUIRE(owner.wait_stopped(kDeadline)); + owner.close(); + + CHECK(port.idle_calls() == 1); + CHECK(port.balanced()); + CHECK(owner.closed_cleanly() == expect_clean); + if (expect_clean) + { + CHECK_NOTHROW(owner.take_latest()); + } + else + { + std::string what; + try + { + owner.take_latest(); + } + catch (const std::runtime_error& e) + { + what = e.what(); + } + CHECK_THAT(what, ContainsSubstring(expected)); + } +} + +TEST_CASE("close without a stop request does not wait for idle", "[buffer_owner]") +{ + FakePort port(4); + port.set_budget(kUnlimited); + BufferOwner owner(port, 4, kWait); + owner.start(); + REQUIRE(wait_until([&] { return port.acquired_total() >= 2; })); + owner.close(); + CHECK(port.idle_calls() == 0); + CHECK(port.stop_calls() == 0); + CHECK(owner.closed_cleanly()); + CHECK(port.balanced()); +} + +TEST_CASE("an owner that never started is cleanly closed", "[buffer_owner]") +{ + FakePort port(4); + BufferOwner owner(port, 4, kWait); + CHECK(owner.closed_cleanly()); + owner.close(); + owner.close(); + CHECK(owner.closed_cleanly()); + CHECK(port.idle_calls() == 0); + CHECK(port.threads().empty()); +} diff --git a/src/plugins/sensing/tests/test_camera_backend.cpp b/src/plugins/sensing/tests/test_camera_backend.cpp index 4cb7837343..57bd538fef 100644 --- a/src/plugins/sensing/tests/test_camera_backend.cpp +++ b/src/plugins/sensing/tests/test_camera_backend.cpp @@ -6,6 +6,7 @@ #include #include +#include #include #include #include @@ -14,7 +15,9 @@ using Catch::Matchers::ContainsSubstring; using plugins::sensing::BackendProbe; using plugins::sensing::CameraBackend; +using plugins::sensing::CameraInfo; using plugins::sensing::FrameLease; +using plugins::sensing::match_requested_cameras; using plugins::sensing::parse_camera_backend; using plugins::sensing::select_camera_backend; @@ -165,3 +168,36 @@ TEST_CASE("A default FrameLease releases nothing", "[sensing][lease]") FrameLease lease; lease.reset(); } + +TEST_CASE("Requested cameras must all be found", "[sensing][backend]") +{ + CameraInfo a; + a.id = 0; + CameraInfo b; + b.id = 4; + BackendProbe p; + + SECTION("every requested id present matches") + { + match_requested_cameras({ a, b }, { 0, 4 }, "Argus", p); + CHECK(p.matches); + CHECK(p.detail.empty()); + } + SECTION("a missing id names it and does not match") + { + match_requested_cameras({ a, b }, { 0, 8 }, "Argus", p); + CHECK_FALSE(p.matches); + CHECK_THAT(p.detail, ContainsSubstring("no Argus camera device 8")); + } + SECTION("with none requested any camera matches") + { + match_requested_cameras({ a }, {}, "Argus", p); + CHECK(p.matches); + } + SECTION("no cameras found never matches") + { + match_requested_cameras({}, {}, "Argus", p); + CHECK_FALSE(p.matches); + CHECK_THAT(p.detail, ContainsSubstring("reports no camera devices")); + } +}