diff --git a/src/plugins/sensing/CMakeLists.txt b/src/plugins/sensing/CMakeLists.txt index 5f0ecf8d3f..236a6748f7 100644 --- a/src/plugins/sensing/CMakeLists.txt +++ b/src/plugins/sensing/CMakeLists.txt @@ -188,12 +188,64 @@ message(STATUS "SENSING plugin: V4L2=${V4L2_LIBRARY}") # the tegra lib dir ships no libnvjpeg to link against. endif() +# ============================================================================== +# LiveKit video output (optional) +# ============================================================================== +# Publishes each selected sensor as a LiveKit camera track. Off by default so a build that does not +# stream needs nothing new. The SDK is a prebuilt package (headers and shared libraries), pinned to a +# release and checked against its hash; point LIVEKIT_SDK_DIR at an unpacked copy to use another. +option(SENSING_LIVEKIT "Add LiveKit video streaming to the SENSING plugin (downloads the LiveKit C++ SDK)" OFF) +if(SENSING_LIVEKIT) + if(NOT LIVEKIT_SDK_DIR) + if(CMAKE_SYSTEM_PROCESSOR MATCHES "aarch64|arm64") + set(_livekit_arch "arm64") + set(_livekit_hash "e3a97f71943c0bf9fa4ebd70927ca2906b31dcd6c14b42312b1225b4b868e4ad") + elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "x86_64|AMD64") + set(_livekit_arch "x64") + set(_livekit_hash "5e0feb2e519c6ff131dd2d1c45149115057ae4e21a324fdb249e1789d5614e7b") + else() + message(FATAL_ERROR "No prebuilt LiveKit SDK for ${CMAKE_SYSTEM_PROCESSOR}; set LIVEKIT_SDK_DIR.") + endif() + 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_HASH SHA256=${_livekit_hash} + ) + FetchContent_MakeAvailable(livekit_sdk) + set(LIVEKIT_SDK_DIR "${livekit_sdk_SOURCE_DIR}") + endif() + # The package ships no licence files: install the SDK's own, pinned to the same release. + foreach(_notice LICENSE NOTICE) + if(_notice STREQUAL "LICENSE") + set(_notice_hash "09e8a9bcec8067104652c168685ab0931e7868f9c8284b66f5ae6edae5f1130b") + else() + set(_notice_hash "6d0f7c5e0deb284362296c5da110dd353831da2e67ca77cb5abc051b951b0fac") + endif() + FetchContent_Declare( + livekit_sdk_${_notice} + URL https://raw.githubusercontent.com/livekit/client-sdk-cpp/v1.12.4/${_notice} + URL_HASH SHA256=${_notice_hash} + DOWNLOAD_NO_EXTRACT TRUE + ) + FetchContent_MakeAvailable(livekit_sdk_${_notice}) + set(SENSING_LIVEKIT_${_notice}_FILE "${livekit_sdk_${_notice}_SOURCE_DIR}/${_notice}" CACHE INTERNAL "") + endforeach() + find_package(LiveKit CONFIG REQUIRED PATHS "${LIVEKIT_SDK_DIR}" NO_DEFAULT_PATH) + message(STATUS "SENSING plugin: LiveKit SDK=${LIVEKIT_SDK_DIR}") + # Cache, so tests/ (a sibling directory) can find the same SDK. + set(SENSING_LIVEKIT_SDK_DIR "${LIVEKIT_SDK_DIR}" CACHE INTERNAL "") +endif() + # ============================================================================== # Targets # ============================================================================== message(STATUS "Building SENSING camera plugin") add_subdirectory(cuda_ipc) +if(SENSING_LIVEKIT) + add_subdirectory(livekit) +endif() add_subdirectory(core) add_subdirectory(app) add_subdirectory(tools) diff --git a/src/plugins/sensing/README.md b/src/plugins/sensing/README.md index 9560d9dee2..67a417559b 100644 --- a/src/plugins/sensing/README.md +++ b/src/plugins/sensing/README.md @@ -25,6 +25,7 @@ That guide owns the operating instructions; this README maps the source tree. host-side container setup. - [`configs/`](configs/): the platform configuration and its provenance. - [`core/mock_*`](core/): the mock backend (see below). +- [`livekit/`](livekit/): the LiveKit SDK behind a small interface, for streaming (see below). ## Platform configuration provenance @@ -101,3 +102,101 @@ Environment, read once at construction: A pass on the mock says nothing about SIPL, the ISP output format, NvSci buffer lifetime, the V4L2 encoder or real frame rates. Those are only checked on an Orin with SENSING cameras. + +## LiveKit streaming + +`livekit=` on an `--add-stream` publishes that sensor as a LiveKit camera track, from the real +cameras or the mock alike: the code is the same on both backends. It is off unless built with +`-DSENSING_LIVEKIT=ON`, which downloads the pinned LiveKit C++ SDK package (Ubuntu 24.04, x64 or +arm64, checked against its hash) at configure time; set `LIVEKIT_SDK_DIR` to use an unpacked copy +instead. + +```bash +camera_plugin_sensing --livekit-url=wss://livekit.example.org --livekit-token-file=join.token \ + --add-stream=sensor=0,livekit=sensing-left --add-stream=sensor=1,livekit=sensing-right +``` + +- **Hardware-encoded passthrough.** Each track carries H.264 that this machine's hardware encoder + produced, published through the SDK's pre-encoded source (`EncodedVideoSource`, encoder backend + `PreEncoded`): the SDK forwards each access unit and does not encode. The encoder is the plugin's + `JetsonEncoder` class, so it is the V4L2 engine on a Jetson and NVENC through libavcodec on the + mock. There is no software encoder and no fallback: if the hardware encoder cannot start, + construction fails. The status line names the encoder and the passthrough mode. +- **Token.** The operator supplies it in a file, never on the command line. It names the room and + the participant and must allow publishing; viewers use a separate subscribe-only token. It is not + printed. +- **Tracks.** One H.264 track per sensor, no simulcast. The default size is the sensor's configured + size. `--livekit-scale-div=N` reduces every track by an integer divisor that must divide the + sensor exactly and leave even sizes (2 gives 1280x992 from 2560x1984); it is refused otherwise, + never rounded or stretched, and the encoder runs at the reduced size. `--livekit-fps` caps the + publication rate on the capture clock (default 30). `--livekit-bitrate` sets the encoder's target + (default 8 Mbit/s; the VBR ceiling is half as much again), unlike the recording's `--bitrate`. +- **Frames.** `update()` takes a GPU snapshot of the camera frame and finishes that copy before + returning, because the camera may rewrite its slot at the second following poll. A worker per + sensor then reduces it if asked, hands it to the hardware encoder and publishes each encoded unit + as one access unit, with its own stamp. A slow network drops frames before encoding and cannot + hold up capture or the other sensor: one snapshot is being encoded, one waits, and a newer capture + replaces the waiting one. +- **Feedback.** A viewer joining, or packet loss, makes the SDK ask for a keyframe: the next frame + is an IDR with SPS/PPS. A frame the SDK refuses breaks the reference chain, so the next frame is + an IDR and nothing is published in between. Bitrate targets from congestion control are clamped + to the configured range and applied to the encoder: the Jetson engine changes rate while running; + the mock reopens its session at the new rate, which starts with an IDR. +- **Metadata.** Each frame carries the camera's sequence as its frame ID, the capture time on the host + clock (CLOCK_MONOTONIC microseconds) as its user timestamp, and the sensor's capture TSC as eight + little-endian bytes of user data. Two separately published tracks are not exposure-synchronised; + the shared TSC is what pairs them. +- **Recording.** A stream with `output=` and `livekit=` runs two encoders, one for the file at the + recording's settings and one for the track at the live settings. `--mcap-filename` metadata is + separate and unchanged. +- **Failure.** Losing the room, or a failed publish, ends capture with an error; there is no + fallback to files or another transport. The SDK's own reconnection runs first; on loopback, losing + the server was reported after 15 to 30 s. + +**Verified.** On a desktop GPU against a local `livekit-server`, with a second participant decoding +what arrives: tracks, sizes and picture content, frame IDs and capture TSCs, a viewer joining late +and rejoining, keyframe requests, recovery after a refused frame, the end-of-stream drain, lower +bitrate targets, and the transport's stats naming the passthrough source. The arm64 SDK package has +the same pre-encoded API headers as the x64 one. + +**Not verified.** Nothing here ran on an Orin: not real cameras, not the V4L2 encoder feeding +LiveKit, not the arm64 package, and not the `JETSON` build with LiveKit. The Jetson encoder's +keyframe, rate and drain code is written but was not compiled off the rig. That the H.264 the +Jetson engine produces (High profile, Level 5.2) is accepted by a WebRTC receiver is likewise +unproven; the mock's NVENC output, with the same settings, decoded in the tests. + +**Receiver page.** [`livekit/receiver/index.html`](livekit/receiver/index.html) is a static page that +uses the LiveKit browser SDK (from a CDN, integrity-checked): enter the server URL and a subscriber +token, Connect, and each subscribed video track appears under its name with its size, in its own +aspect ratio. There is no backend or token generator; it is installed beside the plugin under +`receiver/` and can be opened from disk or any static server. + +**Capacity on a Jetson (not measured).** The numbers below are calculated encoder workloads from the +module datasheets, not achieved frame rates; nothing has run on an Orin. They assume head images at +2560x1984 and wrist images at 1920x1536. + +| Images | Rate | Encoded pixels/s | Assessment | +| --- | --- | --- | --- | +| Head stereo, native | 30 fps | 305 million | plausible first target | +| Head stereo and both wrists, native | 30 fps | 482 million | depends on the preset; little headroom | +| All four at half width and height | 30 fps | 120 million | more headroom | +| All four, native | 60 fps | 963 million | not a supported target | + +The datasheets list H.264 ceilings of 930/530/280 MPixel/s (UHP/HP/HQ) for the AGX Orin 64GB module, +730/420/220 for the 32GB module and 680/470/220 for Orin NX; four native images at 30 fps already +exceed the NX and 32GB HP ceilings. Orin Nano has no NVENC, so it cannot meet the hardware-encoder +requirement. The encoder selects the V4L2 `FAST` hardware preset explicitly (a control separate from +max-performance clocks); four native images may need `ULTRAFAST`, to be measured. A stream with both +`output=` and `livekit=` runs two encoder instances on the one hardware engine, so budget the sum of +their pixel rates. Sustained rates also depend on power mode and cooling. A frame the encoder has no +input buffer for is dropped before any conversion work. Moving frames without the intermediate +copies (a direct NV12 or DMA-BUF path) is a separate change to make only if measurements call for it. + +`cmake --install` copies `liblivekit.so` and `liblivekit_ffi.so` beside the plugin, which finds them +through `$ORIGIN`, and the SDK's own `LICENSE` and `NOTICE` (fetched at configure time, hash +checked) under `licenses/livekit-sdk/`. `liblivekit_ffi.so` bundles third-party components, WebRTC +among them, whose notices the SDK package does not ship: they need a compliance review before the +binaries are distributed. + +Tests that need a LiveKit service start a local `livekit-server` when `LIVEKIT_SERVER` names its +binary, and are skipped without it; the rest need only a GPU. diff --git a/src/plugins/sensing/app/CMakeLists.txt b/src/plugins/sensing/app/CMakeLists.txt index 4d94874f19..fe69368593 100644 --- a/src/plugins/sensing/app/CMakeLists.txt +++ b/src/plugins/sensing/app/CMakeLists.txt @@ -23,3 +23,19 @@ install(TARGETS camera_plugin_sensing install(DIRECTORY "${CMAKE_CURRENT_SOURCE_DIR}/../configs/" DESTINATION plugins/sensing_camera/configs ) + +# The plugin links the LiveKit runtime, whose RUNPATH points into the build tree; ship it beside the +# binary, where INSTALL_RPATH=$ORIGIN (and liblivekit.so's own $ORIGIN) finds it. +if(SENSING_LIVEKIT) + install(FILES + "${SENSING_LIVEKIT_SDK_DIR}/lib/liblivekit.so" + "${SENSING_LIVEKIT_SDK_DIR}/lib/liblivekit_ffi.so" + DESTINATION plugins/sensing_camera + ) + install(FILES "${SENSING_LIVEKIT_LICENSE_FILE}" "${SENSING_LIVEKIT_NOTICE_FILE}" + DESTINATION plugins/sensing_camera/licenses/livekit-sdk + ) + install(FILES "${CMAKE_CURRENT_SOURCE_DIR}/../livekit/receiver/index.html" + DESTINATION plugins/sensing_camera/receiver + ) +endif() diff --git a/src/plugins/sensing/app/main.cpp b/src/plugins/sensing/app/main.cpp index 5c163e7a09..79fbe3facb 100644 --- a/src/plugins/sensing/app/main.cpp +++ b/src/plugins/sensing/app/main.cpp @@ -11,8 +11,10 @@ #include #include #include +#include #include #include +#include #include #include #include @@ -67,6 +69,10 @@ static StreamConfig parse_stream_arg(const std::string& arg) { cfg.ipc_socket_path = val; } + else if (key == "livekit") + { + cfg.livekit_track = val; + } else { throw std::runtime_error("Unknown key in --add-stream: '" + key + "'"); @@ -75,8 +81,8 @@ static StreamConfig parse_stream_arg(const std::string& arg) if (!has_sensor) throw std::runtime_error("--add-stream requires sensor="); - if (cfg.output_path.empty() && cfg.ipc_socket_path.empty()) - throw std::runtime_error("--add-stream requires output= or ipc=, or both"); + if (cfg.output_path.empty() && cfg.ipc_socket_path.empty() && cfg.livekit_track.empty()) + throw std::runtime_error("--add-stream requires at least one of output=, ipc=, livekit="); return cfg; } @@ -123,13 +129,15 @@ void print_usage(const char* program_name) { std::cout << "Usage: " << program_name << " [options] --add-stream ...\n" << "\nStream Configuration (repeatable):\n" - << " --add-stream sensor=[,output=][,ipc=]\n" + << " --add-stream sensor=[,output=][,ipc=][,livekit=]\n" << " sensor: SIPL pipeline index, as --list-sensors reports it. This is NOT\n" << " the GMSL link index or the JSON sensorInfo.id.\n" << " output: file path for this stream's H.264 data\n" << " ipc: Unix socket serving raw RGBA8 frames as CUDA memory to\n" << " another process (camera_viz `type: cuda_ipc`). No encode.\n" - << " At least one of output/ipc is required; both may be given.\n" + << " livekit: name of the LiveKit camera track carrying this sensor (needs the\n" + << " --livekit-* options below and a build with -DSENSING_LIVEKIT=ON).\n" + << " At least one of output/ipc/livekit is required; any combination may be given.\n" << "\nPlatform (resolution and frame rate come from here, not from flags):\n" << " --platform-config=PATH platform JSON (default: configs/shw5g.json beside this\n" << " binary; point it at the vendor package to test a\n" @@ -142,6 +150,17 @@ void print_usage(const char* program_name) << " --bitrate=N H.264 average bitrate in bps (default: 40000000)\n" << " --peak-bitrate=N VBR ceiling in bps; 0 selects CBR (default: 60000000)\n" << " --gop=N IDR period in frames (default: fps*5)\n" + << "\nLiveKit (for livekit= streams):\n" + << " --livekit-url=URL server, e.g. wss://livekit.example.org\n" + << " --livekit-token-file=PATH file holding the join token; it names the room and the\n" + << " participant, and must allow publishing. Never printed.\n" + << " --livekit-fps=N publication rate cap (default: 30)\n" + << " --livekit-scale-div=N reduce every track by this integer divisor, e.g. 2 gives\n" + << " 1280x992 from 2560x1984; must divide exactly (default: 1)\n" + << " --livekit-bitrate=N target bitrate of each track's hardware encoder in bps; the\n" + << " ceiling is half as much again (default: 8000000)\n" + << " Each track carries H.264 encoded on this machine's hardware encoder and forwarded as is;\n" + << " there is no software encoder.\n" << "\nGeneral:\n" << " --gpu-id=N CUDA device index (default: 0)\n" << " --isp0-buffers=N ISP0 buffers per sensor (default: 6)\n" @@ -152,7 +171,10 @@ void print_usage(const char* program_name) << "\nExamples:\n" << " " << program_name << " --list-sensors\n" << " " << program_name << " --add-stream=sensor=0,ipc=/tmp/sensing0.sock \\\n" - << " --add-stream=sensor=1,ipc=/tmp/sensing1.sock\n"; + << " --add-stream=sensor=1,ipc=/tmp/sensing1.sock\n" + << " " << program_name << " --livekit-url=wss://livekit.example.org \\\n" + << " --livekit-token-file=join.token --add-stream=sensor=0,livekit=sensing-left \\\n" + << " --add-stream=sensor=1,livekit=sensing-right\n"; } // ============================================================================= @@ -165,6 +187,7 @@ try SensingConfig camera_config; std::map stream_map; std::string mcap_filename; + std::string livekit_token_file; bool list_sensors = false; for (int i = 1; i < argc; ++i) @@ -207,6 +230,26 @@ try { list_sensors = true; } + else if (arg.find("--livekit-url=") == 0) + { + camera_config.livekit.url = arg.substr(14); + } + else if (arg.find("--livekit-token-file=") == 0) + { + livekit_token_file = arg.substr(21); + } + else if (arg.find("--livekit-fps=") == 0) + { + camera_config.livekit.fps = static_cast(std::stoul(arg.substr(14))); + } + else if (arg.find("--livekit-bitrate=") == 0) + { + camera_config.livekit.bitrate_bps = static_cast(std::stoul(arg.substr(18))); + } + else if (arg.find("--livekit-scale-div=") == 0) + { + camera_config.livekit.scale_div = static_cast(std::stoul(arg.substr(20))); + } else if (arg.find("--bitrate=") == 0) { camera_config.bitrate_bps = static_cast(std::stoul(arg.substr(10))); @@ -296,6 +339,39 @@ try stream_configs.push_back(std::move(cfg)); } + // LiveKit: check the settings that need no hardware or network before anything is opened. + const bool uses_livekit = std::any_of( + stream_configs.begin(), stream_configs.end(), [](const StreamConfig& s) { return !s.livekit_track.empty(); }); + if (uses_livekit) + { + std::set track_names; + for (const auto& s : stream_configs) + { + if (!s.livekit_track.empty() && !track_names.insert(s.livekit_track).second) + throw std::runtime_error("LiveKit track name '" + s.livekit_track + "' is used twice"); + } + if (camera_config.livekit.url.empty() || livekit_token_file.empty()) + throw std::runtime_error("livekit= streams need --livekit-url and --livekit-token-file"); + if (camera_config.livekit.fps == 0) + throw std::runtime_error("--livekit-fps must be at least 1"); + if (camera_config.livekit.bitrate_bps < 100'000) + throw std::runtime_error("--livekit-bitrate must be at least 100000"); + + // The token is read from a file so it never appears in a process listing, and is not printed. + std::ifstream token_file(livekit_token_file); + if (!token_file) + throw std::runtime_error("cannot read --livekit-token-file " + livekit_token_file); + std::getline(token_file, camera_config.livekit.token); + const auto last = camera_config.livekit.token.find_last_not_of(" \t\r\n"); + camera_config.livekit.token.erase(last == std::string::npos ? 0 : last + 1); + if (camera_config.livekit.token.empty()) + throw std::runtime_error("--livekit-token-file " + livekit_token_file + " is empty"); + } + else if (!camera_config.livekit.url.empty() || !livekit_token_file.empty()) + { + throw std::runtime_error("--livekit-* options were given but no stream has livekit="); + } + // 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 diff --git a/src/plugins/sensing/core/CMakeLists.txt b/src/plugins/sensing/core/CMakeLists.txt index 9ddc30adc8..f0aa2d5731 100644 --- a/src/plugins/sensing/core/CMakeLists.txt +++ b/src/plugins/sensing/core/CMakeLists.txt @@ -92,6 +92,20 @@ target_link_libraries(sensing_core oxr::oxr_utils ) +# The LiveKit output is optional: without it the camera compiles the same code minus the streaming path. +if(SENSING_LIVEKIT) + target_sources(sensing_core PRIVATE + livekit_video_output.cpp + livekit_video_output.hpp + rgba_box_downscale.cu + rgba_box_downscale.cuh + ) + target_link_libraries(sensing_core PUBLIC sensing::livekit) + target_compile_definitions(sensing_core PUBLIC SENSING_WITH_LIVEKIT=1) +else() + target_compile_definitions(sensing_core PUBLIC SENSING_WITH_LIVEKIT=0) +endif() + if(SENSING_BACKEND_RESOLVED STREQUAL "JETSON") target_include_directories(sensing_core PRIVATE diff --git a/src/plugins/sensing/core/jetson_encoder.cpp b/src/plugins/sensing/core/jetson_encoder.cpp index 39b6a9c735..b282c93423 100644 --- a/src/plugins/sensing/core/jetson_encoder.cpp +++ b/src/plugins/sensing/core/jetson_encoder.cpp @@ -11,6 +11,7 @@ #include #include #include +#include #include #include #include @@ -55,6 +56,7 @@ struct JetsonEncoder::Impl std::mutex mutex; std::deque ready; bool eos_reached = false; + bool eos_sent = false; // driving thread only ~Impl() { @@ -83,6 +85,7 @@ struct JetsonEncoder::Impl unit.data.assign(data, data + buffer->planes[0].bytesused); unit.timestamp_ns = static_cast(v4l2_buf->timestamp.tv_sec) * 1000000000LL + static_cast(v4l2_buf->timestamp.tv_usec) * 1000LL; + unit.keyframe = (v4l2_buf->flags & V4L2_BUF_FLAG_KEYFRAME) != 0; std::lock_guard lock(self->mutex); self->ready.push_back(std::move(unit)); } @@ -152,6 +155,9 @@ JetsonEncoder::JetsonEncoder(const EncoderConfig& config) : m_impl(std::make_uni check(enc->setNumBFrames(0), "setNumBFrames"); check(enc->setInsertSpsPpsAtIdrEnabled(true), "setInsertSpsPpsAtIdrEnabled"); check(enc->setMaxPerfMode(1), "setMaxPerfMode"); + // A separate control from the max-performance clocks: it trades quality for encoder throughput. + // 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"); check(enc->capture_plane.setupPlane(V4L2_MEMORY_MMAP, kCaptureBuffers, true, false), "capture setupPlane"); @@ -195,12 +201,6 @@ bool JetsonEncoder::submit(uintptr_t rgba_device_ptr, std::size_t row_pitch_byte auto* enc = m_impl->encoder; const auto& config = m_impl->config; - launch_rgba_to_nv12(reinterpret_cast(rgba_device_ptr), static_cast(row_pitch_bytes), m_impl->d_y, - static_cast(m_impl->d_y_pitch), m_impl->d_uv, static_cast(m_impl->d_uv_pitch), - static_cast(config.width), static_cast(config.height), config.full_range, nullptr); - check_cuda(cudaGetLastError(), "rgba_to_nv12 launch"); - check_cuda(cudaStreamSynchronize(nullptr), "rgba_to_nv12 sync"); - struct v4l2_buffer v4l2_buf; struct v4l2_plane planes[MAX_PLANES]; std::memset(&v4l2_buf, 0, sizeof(v4l2_buf)); @@ -221,6 +221,14 @@ bool JetsonEncoder::submit(uintptr_t rgba_device_ptr, std::size_t row_pitch_byte return false; // encoder still busy; drop this frame rather than block } + // 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. + launch_rgba_to_nv12(reinterpret_cast(rgba_device_ptr), static_cast(row_pitch_bytes), m_impl->d_y, + static_cast(m_impl->d_y_pitch), m_impl->d_uv, static_cast(m_impl->d_uv_pitch), + static_cast(config.width), static_cast(config.height), config.full_range, nullptr); + check_cuda(cudaGetLastError(), "rgba_to_nv12 launch"); + check_cuda(cudaStreamSynchronize(nullptr), "rgba_to_nv12 sync"); + check_cuda(cudaMemcpy2D(buffer->planes[0].data, buffer->planes[0].fmt.stride, m_impl->d_y, m_impl->d_y_pitch, config.width, config.height, cudaMemcpyDeviceToHost), "cudaMemcpy2D(Y)"); @@ -253,38 +261,65 @@ EncodedUnit JetsonEncoder::poll() return out; } -EncodedUnit JetsonEncoder::end_of_stream() +std::vector JetsonEncoder::drain() { auto* enc = m_impl->encoder; - // A zero-length output buffer is the EOS marker for the V4L2 encoder. - 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; - - NvBuffer* buffer = nullptr; - if (enc->output_plane.dqBuffer(v4l2_buf, &buffer, nullptr, 10) >= 0) + // A zero-length output buffer is the EOS marker for the V4L2 encoder. It is sent once: a second + // drain only returns what is left. + if (!m_impl->eos_sent) { - v4l2_buf.m.planes[0].bytesused = 0; - v4l2_buf.m.planes[1].bytesused = 0; - enc->output_plane.qBuffer(v4l2_buf, nullptr); - } + m_impl->eos_sent = true; + 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; - enc->capture_plane.waitForDQThread(2000); + NvBuffer* buffer = nullptr; + if (enc->output_plane.dqBuffer(v4l2_buf, &buffer, nullptr, 10) >= 0) + { + v4l2_buf.m.planes[0].bytesused = 0; + v4l2_buf.m.planes[1].bytesused = 0; + enc->output_plane.qBuffer(v4l2_buf, nullptr); + } + + enc->capture_plane.waitForDQThread(2000); + } - EncodedUnit out; std::lock_guard lock(m_impl->mutex); + std::vector units( + std::make_move_iterator(m_impl->ready.begin()), std::make_move_iterator(m_impl->ready.end())); + m_impl->ready.clear(); + return units; +} + +EncodedUnit JetsonEncoder::end_of_stream() +{ // The tail is concatenated, so it carries the stamp of the last unit in it. - for (auto& chunk : m_impl->ready) + EncodedUnit out; + for (auto& chunk : drain()) { out.data.insert(out.data.end(), chunk.data.begin(), chunk.data.end()); out.timestamp_ns = chunk.timestamp_ns; } - m_impl->ready.clear(); return out; } +void JetsonEncoder::request_keyframe() +{ + check(m_impl->encoder->forceIDR(), "forceIDR"); +} + +bool JetsonEncoder::set_bitrate(uint32_t bitrate_bps) +{ + return m_impl->encoder->setBitrate(bitrate_bps) >= 0; +} + +const char* JetsonEncoder::backend_name() +{ + return "Jetson V4L2 NVENC"; +} + } // namespace sensing } // namespace plugins diff --git a/src/plugins/sensing/core/jetson_encoder.hpp b/src/plugins/sensing/core/jetson_encoder.hpp index da3742ca72..67454bb6e8 100644 --- a/src/plugins/sensing/core/jetson_encoder.hpp +++ b/src/plugins/sensing/core/jetson_encoder.hpp @@ -31,6 +31,9 @@ struct EncodedUnit { std::vector data; int64_t timestamp_ns = 0; + /// The unit is an IDR (with its parameter sets), decodable on its own. Not set on the + /// concatenated tail from end_of_stream(). + bool keyframe = false; bool empty() const { @@ -44,6 +47,9 @@ struct EncodedUnit * Jetson has no libnvidia-encode — the NVIDIA Video Codec SDK is dGPU-only — * so this wraps NvVideoEncoder from the Jetson Multimedia API instead. * + * One thread must drive an encoder: submit(), poll(), request_keyframe() and set_bitrate() are not + * safe to call concurrently. + * * Submission is asynchronous: the encoder needs several input frames before it * emits the first bitstream unit, so submit() never blocks waiting for output * and poll() drains whatever a capture-plane thread has completed. @@ -72,9 +78,24 @@ class JetsonEncoder /** @brief Take any completed unit; empty during encoder warmup. */ EncodedUnit poll(); - /** @brief Signal EOS and drain the remaining bitstream. */ + /** @brief Signal EOS and drain the remaining bitstream, concatenated into one unit. */ EncodedUnit end_of_stream(); + /** @brief Signal EOS and drain the remaining bitstream, one unit per frame with its own stamp. */ + std::vector drain(); + + /** @brief Make the next submitted frame an IDR, preceded by SPS/PPS. */ + void request_keyframe(); + + /** + * @brief Change the target bitrate of the running stream. + * @return false when this backend cannot change it while encoding. + */ + bool set_bitrate(uint32_t bitrate_bps); + + /** @brief Names the encoder behind this class, for diagnostics. */ + static const char* backend_name(); + private: struct Impl; std::unique_ptr m_impl; diff --git a/src/plugins/sensing/core/livekit_video_output.cpp b/src/plugins/sensing/core/livekit_video_output.cpp new file mode 100644 index 0000000000..5495f750fa --- /dev/null +++ b/src/plugins/sensing/core/livekit_video_output.cpp @@ -0,0 +1,486 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "livekit_video_output.hpp" + +#include "jetson_encoder.hpp" +#include "rgba_box_downscale.cuh" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace plugins +{ +namespace sensing +{ + +namespace +{ + +void check_cuda(cudaError_t err, const char* what) +{ + if (err != cudaSuccess) + throw std::runtime_error(std::string("LiveKit output: ") + what + ": " + cudaGetErrorString(err)); +} + +// Congestion control never drives the encoder below this. +constexpr uint32_t kMinBitrateBps = 100'000; +// A target is applied only when it differs from the current one by this fraction, and this long after +// the last change. +constexpr double kMinBitrateChange = 0.25; +constexpr std::chrono::seconds kMinBitrateInterval{ 3 }; + +} // namespace + +struct LiveKitVideoOutput::Impl +{ + struct Slot + { + uint8_t* ptr = nullptr; + size_t pitch = 0; + uint64_t timestamp_ns = 0; + uint64_t capture_tsc_ns = 0; + uint64_t sequence = 0; + }; + + Impl(std::unique_ptr track, const LiveKitOutputConfig& cfg) : sink(std::move(track)), config(cfg) + { + validate_geometry(config.source_width, config.source_height, config.scale_div); + if (config.fps == 0) + throw std::runtime_error("LiveKit output: the publication rate must be at least 1 fps"); + out_width = config.source_width / config.scale_div; + 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"); + period_ns = 1'000'000'000ull / config.fps; + if (config.bitrate_bps < kMinBitrateBps) + { + throw std::runtime_error("LiveKit output: the bitrate must be at least " + std::to_string(kMinBitrateBps) + + " bps"); + } + // Congestion control moves the target between the floor and the ceiling; the configured target + // is never below the floor, so the range is never inverted. + max_bitrate = std::max(config.bitrate_bps, config.peak_bitrate_bps); + + // A constructor that throws never runs the destructor, so free what was allocated here. + try + { + check_cuda(cudaSetDevice(config.gpu_id), "cudaSetDevice"); + + // First, so a machine without a usable hardware encoder fails before anything is allocated. + EncoderConfig encoder_config; + encoder_config.width = out_width; + encoder_config.height = out_height; + encoder_config.bitrate_bps = config.bitrate_bps; + encoder_config.peak_bitrate_bps = config.peak_bitrate_bps; + encoder_config.fps = config.fps; + encoder_config.gop = config.gop; + encoder_config.full_range = config.full_range; + encoder = std::make_unique(encoder_config); + applied_bitrate = config.bitrate_bps; + + for (auto& slot : slots) + { + check_cuda(cudaMallocPitch(reinterpret_cast(&slot.ptr), &slot.pitch, + static_cast(config.source_width) * 4, config.source_height), + "cudaMallocPitch(snapshot)"); + } + if (config.scale_div > 1) + { + check_cuda(cudaMallocPitch(reinterpret_cast(&d_scaled), &scaled_pitch, + static_cast(out_width) * 4, out_height), + "cudaMallocPitch(reduced)"); + } + check_cuda(cudaStreamCreateWithFlags(&snapshot_stream, cudaStreamNonBlocking), "cudaStreamCreate"); + check_cuda(cudaStreamCreateWithFlags(&work_stream, cudaStreamNonBlocking), "cudaStreamCreate"); + + worker = std::thread([this] { run(); }); + } + catch (...) + { + release(); + throw; + } + } + + ~Impl() + { + { + std::lock_guard lock(mutex); + stopping = true; + } + wake.notify_all(); + if (worker.joinable()) + worker.join(); + release(); + } + + void release() + { + for (auto& slot : slots) + cudaFree(slot.ptr); + cudaFree(d_scaled); + if (snapshot_stream) + cudaStreamDestroy(snapshot_stream); + if (work_stream) + cudaStreamDestroy(work_stream); + } + + void fail(const std::string& message) + { + std::lock_guard lock(failure_mutex); + if (!failed_flag.load()) + { + failure_message = message; + failed_flag.store(true); + } + } + + // Polling thread. + void submit(const FrameView& view) + { + ++offered; + if (failed_flag.load()) + return; + + // Pace on the capture clock, so poll jitter does not drop a frame that is due. A tenth of a + // period of slack lets 60 fps capture feed a 30 fps track every second capture. + if (have_last && view.timestamp_ns < last_accepted_ns + period_ns * 9 / 10) + { + ++paced_out; + return; + } + + try + { + std::lock_guard lock(mutex); + int target = -1; + for (int i = 0; i < 2; ++i) + { + if (i != inflight && i != pending) + { + target = i; + break; + } + } + if (target < 0) + { + // The worker holds one slot and the other waits: the newer capture takes the waiting one. + target = pending; + } + // A capture still waiting is overwritten by this newer one. + if (pending >= 0) + ++replaced; + + // Complete before returning: the camera may rewrite its slot at the second following poll. + Slot& slot = slots[target]; + check_cuda(cudaMemcpy2DAsync(slot.ptr, slot.pitch, reinterpret_cast(view.ptr), view.pitch, + static_cast(config.source_width) * 4, config.source_height, + cudaMemcpyDeviceToDevice, snapshot_stream), + "snapshot copy"); + check_cuda(cudaStreamSynchronize(snapshot_stream), "snapshot sync"); + + slot.timestamp_ns = view.timestamp_ns; + slot.capture_tsc_ns = view.capture_tsc_ns; + slot.sequence = view.sequence; + pending = target; + } + catch (const std::exception& e) + { + fail(e.what()); + return; + } + ++snapshots; + have_last = true; + last_accepted_ns = view.timestamp_ns; + wake.notify_one(); + } + + // Worker thread. + void run() + { + try + { + check_cuda(cudaSetDevice(config.gpu_id), "cudaSetDevice(worker)"); + for (;;) + { + int index = -1; + 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; }); + if (stopping) + return; + index = pending; + pending = -1; + inflight = index; + flush_now = flush_requested; + } + + apply_feedback(); + if (index >= 0) + { + encode(slots[index]); + std::lock_guard lock(mutex); + inflight = -1; + } + if (flush_now) + { + for (auto& unit : encoder->drain()) + publish_unit(unit); + std::lock_guard lock(mutex); + flush_requested = false; + flushed.notify_all(); + } + else + { + for (auto unit = encoder->poll(); !unit.empty(); unit = encoder->poll()) + publish_unit(unit); + } + } + } + catch (const std::exception& e) + { + fail(e.what()); + std::lock_guard lock(mutex); + flushed.notify_all(); + } + } + + // Worker thread: what the receiving side asked for since the last frame. + void apply_feedback() + { + const EncoderFeedback feedback = sink->take_feedback(); + if (feedback.keyframe_requested) + { + ++keyframes_requested; + keyframe_wanted = true; + } + if (feedback.target_bitrate_bps) + { + const uint32_t target = std::clamp(*feedback.target_bitrate_bps, kMinBitrateBps, max_bitrate); + // Congestion control re-estimates all the time. Follow it only for a real change, and not + // more than every few seconds: on the mock a change reopens the encoder session. + const auto now = std::chrono::steady_clock::now(); + const double change = std::abs(static_cast(target) - applied_bitrate) / applied_bitrate; + if (change >= kMinBitrateChange && now - last_bitrate_change >= kMinBitrateInterval && + encoder->set_bitrate(target)) + { + applied_bitrate = target; + last_bitrate_change = now; + } + } + } + + // Worker thread. + void encode(const Slot& slot) + { + const uint8_t* rgba = slot.ptr; + size_t pitch = slot.pitch; + if (config.scale_div > 1) + { + launch_rgba_box_downscale(rgba, static_cast(pitch), d_scaled, static_cast(scaled_pitch), + static_cast(out_width), static_cast(out_height), + static_cast(config.scale_div), work_stream); + check_cuda(cudaGetLastError(), "reduction kernel"); + // The encoder reads on the default stream, which a non-blocking stream does not order against. + check_cuda(cudaStreamSynchronize(work_stream), "reduction sync"); + rgba = d_scaled; + pitch = scaled_pitch; + } + + if (keyframe_wanted) + { + encoder->request_keyframe(); + keyframe_wanted = false; + } + const int64_t stamp_ns = static_cast(slot.timestamp_ns); + if (!encoder->submit(reinterpret_cast(rgba), pitch, stamp_ns)) + { + // No input buffer: dropped before encoding, so no reference is lost. A keyframe already + // requested of the encoder applies to the next frame it takes. + ++encoder_busy; + return; + } + in_encoder.push_back({ stamp_ns / 1000, static_cast(slot.sequence), slot.capture_tsc_ns }); + } + + // Worker thread: one encoded unit becomes one published frame, with the sequence and capture time + // of the capture it came from. + void publish_unit(const EncodedUnit& unit) + { + const int64_t stamp_us = unit.timestamp_ns / 1000; + // The encoder keeps submission order; a frame it never returned was dropped inside it. + while (!in_encoder.empty() && in_encoder.front().stamp_us < stamp_us) + in_encoder.pop_front(); + if (in_encoder.empty() || in_encoder.front().stamp_us != stamp_us) + throw std::runtime_error("LiveKit output: the encoder returned a unit for a frame it was not given"); + const FrameMeta meta = in_encoder.front(); + in_encoder.pop_front(); + + // After a refusal the stream is only usable from its next keyframe. + if (awaiting_keyframe && !unit.keyframe) + return; + + EncodedVideoFrame frame; + frame.data = unit.data.data(); + frame.size = unit.data.size(); + frame.keyframe = unit.keyframe; + frame.timestamp_us = static_cast(stamp_us); + frame.frame_id = meta.sequence; + frame.capture_tsc_ns = meta.capture_tsc_ns; + if (sink->submit(frame)) + { + awaiting_keyframe = false; + ++published; + } + else + { + ++rejected; + awaiting_keyframe = true; + keyframe_wanted = true; + } + } + + std::unique_ptr sink; + LiveKitOutputConfig config; + uint32_t out_width = 0; + uint32_t out_height = 0; + uint64_t period_ns = 0; + + struct FrameMeta + { + int64_t stamp_us = 0; + uint32_t sequence = 0; + uint64_t capture_tsc_ns = 0; + }; + + std::unique_ptr encoder; + uint32_t max_bitrate = 0; + + // Worker thread only. + std::deque in_encoder; ///< Submitted to the encoder, no unit back yet. + bool keyframe_wanted = false; + bool awaiting_keyframe = false; + uint32_t applied_bitrate = 0; + std::chrono::steady_clock::time_point last_bitrate_change = std::chrono::steady_clock::now(); + + Slot slots[2]; + uint8_t* d_scaled = nullptr; + size_t scaled_pitch = 0; + cudaStream_t snapshot_stream = nullptr; + cudaStream_t work_stream = nullptr; + + // Guards pending, inflight, stopping and the snapshot slots' contents. + std::mutex mutex; + std::condition_variable wake; + int pending = -1; + int inflight = -1; + bool stopping = false; + bool flush_requested = false; + std::condition_variable flushed; + std::thread worker; + + // Polling thread only. + bool have_last = false; + uint64_t last_accepted_ns = 0; + + std::atomic offered{ 0 }; + std::atomic paced_out{ 0 }; + std::atomic replaced{ 0 }; + std::atomic snapshots{ 0 }; + std::atomic published{ 0 }; + std::atomic encoder_busy{ 0 }; + std::atomic rejected{ 0 }; + std::atomic keyframes_requested{ 0 }; + + std::mutex failure_mutex; + std::string failure_message; + std::atomic failed_flag{ false }; +}; + +void LiveKitVideoOutput::validate_geometry(uint32_t width, uint32_t height, uint32_t divisor) +{ + if (divisor == 0) + throw std::runtime_error("LiveKit output: the scale divisor must be at least 1"); + if (width == 0 || height == 0) + throw std::runtime_error("LiveKit output: the source size is empty"); + if (width % divisor != 0 || height % divisor != 0) + { + throw std::runtime_error("LiveKit output: divisor " + std::to_string(divisor) + " does not divide " + + std::to_string(width) + "x" + std::to_string(height) + " exactly"); + } + if ((width / divisor) % 2 != 0 || (height / divisor) % 2 != 0) + { + throw std::runtime_error("LiveKit output: " + std::to_string(width / divisor) + "x" + + std::to_string(height / divisor) + " is not even; NV12 needs even dimensions"); + } +} + +LiveKitVideoOutput::LiveKitVideoOutput(std::unique_ptr sink, const LiveKitOutputConfig& config) + : m_impl(std::make_unique(std::move(sink), config)) +{ +} + +LiveKitVideoOutput::~LiveKitVideoOutput() = default; + +void LiveKitVideoOutput::submit(const FrameView& view) +{ + m_impl->submit(view); +} + +LiveKitOutputStats LiveKitVideoOutput::stats() const +{ + LiveKitOutputStats s; + s.offered = m_impl->offered.load(); + s.paced_out = m_impl->paced_out.load(); + s.replaced = m_impl->replaced.load(); + s.snapshots = m_impl->snapshots.load(); + s.published = m_impl->published.load(); + s.encoder_busy = m_impl->encoder_busy.load(); + s.rejected = m_impl->rejected.load(); + s.keyframes_requested = m_impl->keyframes_requested.load(); + return s; +} + +bool LiveKitVideoOutput::failed() const +{ + return m_impl->failed_flag.load(); +} + +std::string LiveKitVideoOutput::failure() const +{ + std::lock_guard lock(m_impl->failure_mutex); + return m_impl->failure_message; +} + +std::string LiveKitVideoOutput::describe() +{ + std::string text = std::string(JetsonEncoder::backend_name()) + ", pre-encoded passthrough"; + const std::string transport = m_impl->sink->describe(); + if (!transport.empty()) + text += "; transport: " + transport; + return text; +} + +void LiveKitVideoOutput::flush() +{ + std::unique_lock lock(m_impl->mutex); + m_impl->flush_requested = true; + m_impl->wake.notify_one(); + m_impl->flushed.wait(lock, [&] { return !m_impl->flush_requested || m_impl->failed_flag.load(); }); +} + +} // namespace sensing +} // namespace plugins diff --git a/src/plugins/sensing/core/livekit_video_output.hpp b/src/plugins/sensing/core/livekit_video_output.hpp new file mode 100644 index 0000000000..a0526c17cc --- /dev/null +++ b/src/plugins/sensing/core/livekit_video_output.hpp @@ -0,0 +1,110 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include "sipl_camera.hpp" + +#include + +#include +#include +#include + +namespace plugins +{ +namespace sensing +{ + +struct LiveKitOutputConfig +{ + uint32_t source_width = 0; + uint32_t source_height = 0; + /// Uniform output-only reduction; 1 publishes the sensor's native size. + uint32_t scale_div = 1; + /// Publication rate cap, independent of the capture rate. + uint32_t fps = 30; + int gpu_id = 0; + /// Hardware encoder settings for the published pictures; the peak is also the ceiling for the + /// bitrate that congestion control may ask for. + uint32_t bitrate_bps = 8'000'000; + uint32_t peak_bitrate_bps = 0; + /// IDR period in frames; 0 defers to the encoder's default. + uint32_t gop = 0; + bool full_range = false; +}; + +struct LiveKitOutputStats +{ + uint64_t offered = 0; ///< Captures handed to submit(). + uint64_t paced_out = 0; ///< Skipped to hold the publication rate. + uint64_t replaced = 0; ///< Snapshots overwritten by a newer capture before the worker took them. + uint64_t snapshots = 0; ///< GPU snapshots taken. + uint64_t published = 0; ///< Frames the SDK accepted. Not proof that a viewer received them. + uint64_t encoder_busy = 0; ///< Snapshots the encoder had no input buffer for, dropped before encoding. + uint64_t rejected = 0; ///< Encoded frames the SDK refused; a keyframe follows. + uint64_t keyframes_requested = 0; ///< Keyframes the receiving side asked for. +}; + +/** + * @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 + * SDK call -- happens on this output's own worker thread, so a slow network cannot hold up capture + * or the other sensors. + * + * Two snapshot slots bound the memory: one the worker is encoding, one waiting. A newer capture + * replaces the waiting one; the worker never waits for it. The encoder is the JetsonEncoder class: + * the V4L2 engine on a Jetson, NVENC through libavcodec in the mock. There is no software encoder: if + * the hardware encoder cannot start, construction fails. + * + * The receiving side's keyframe requests and bitrate targets are applied before the next frame is + * encoded. A frame the SDK refuses breaks the reference chain, so the output asks for a keyframe and + * publishes nothing else until one arrives. + */ +class LiveKitVideoOutput +{ +public: + /// @throws std::runtime_error if the geometry is unusable (see validate_geometry), the hardware + /// encoder cannot start, or CUDA fails. + LiveKitVideoOutput(std::unique_ptr sink, const LiveKitOutputConfig& config); + + /// Stops the worker; a snapshot not yet published is dropped. + ~LiveKitVideoOutput(); + + LiveKitVideoOutput(const LiveKitVideoOutput&) = delete; + LiveKitVideoOutput& operator=(const LiveKitVideoOutput&) = delete; + + /// Never blocks on the network. A view older than the publication rate allows is skipped. + void submit(const FrameView& view); + + LiveKitOutputStats stats() const; + + bool failed() const; + std::string failure() const; + + /// Names the hardware encoder and the transport's view of the stream, for diagnostics. + std::string describe(); + + /** + * Encodes what is waiting, drains the encoder and publishes every remaining unit with its own + * stamp. Call once, at the end of the stream: the encoder cannot take frames afterwards. + */ + void flush(); + + /** + * Output size after dividing: both source dimensions must divide exactly and leave positive, + * even results. Nothing is rounded or stretched. + * @throws std::runtime_error naming the problem. + */ + static void validate_geometry(uint32_t width, uint32_t height, uint32_t divisor); + +private: + struct Impl; + std::unique_ptr m_impl; +}; + +} // namespace sensing +} // namespace plugins diff --git a/src/plugins/sensing/core/mock_jetson_encoder.cpp b/src/plugins/sensing/core/mock_jetson_encoder.cpp index e6a30853f9..63bb0cbc36 100644 --- a/src/plugins/sensing/core/mock_jetson_encoder.cpp +++ b/src/plugins/sensing/core/mock_jetson_encoder.cpp @@ -27,6 +27,7 @@ extern "C" #include #include #include +#include #include #include #include @@ -66,21 +67,20 @@ struct JetsonEncoder::Impl { uint32_t buffer = 0; int64_t timestamp_ns = 0; + bool keyframe = false; }; explicit Impl(const EncoderConfig& cfg) : config(cfg) { - const AVCodec* codec = avcodec_find_encoder_by_name("h264_nvenc"); + codec = avcodec_find_encoder_by_name("h264_nvenc"); if (!codec) throw std::runtime_error("mock JetsonEncoder: libavcodec has no h264_nvenc encoder"); - context = avcodec_alloc_context3(codec); - if (!context) - throw std::runtime_error("mock JetsonEncoder: cannot allocate the encoder context"); // A constructor that throws never runs the destructor, so free what was allocated here. try { - open(codec); + open_context(config.bitrate_bps); + alloc_frame(); check_cuda(cudaMallocPitch(reinterpret_cast(&d_y), &d_y_pitch, config.width, config.height), "cudaMallocPitch(Y)"); check_cuda(cudaMallocPitch(reinterpret_cast(&d_uv), &d_uv_pitch, config.width, config.height / 2), @@ -114,8 +114,19 @@ struct JetsonEncoder::Impl release(); } - void open(const AVCodec* codec) + void set_option(const char* name, const char* value) + { + check_av(av_opt_set(context->priv_data, name, value, 0), (std::string("av_opt_set ") + name).c_str()); + } + + // A fresh encoder session at `bitrate`: its first frame is an IDR with parameter sets. + void open_context(uint32_t bitrate) { + avcodec_free_context(&context); + context = avcodec_alloc_context3(codec); + if (!context) + throw std::runtime_error("mock JetsonEncoder: cannot allocate the encoder context"); + const uint32_t fps = config.fps ? config.fps : 30; const uint32_t gop = config.gop ? config.gop : fps * 5; context->width = static_cast(config.width); @@ -124,45 +135,34 @@ struct JetsonEncoder::Impl // Microseconds, as a V4L2 timeval carries them. context->time_base = { 1, 1'000'000 }; context->framerate = { static_cast(fps), 1 }; - context->bit_rate = config.bitrate_bps; + context->bit_rate = bitrate; context->gop_size = static_cast(gop); context->max_b_frames = 0; // Parameter sets stay in the stream, in front of every IDR, so a receiver can join mid-stream. context->flags &= ~AV_CODEC_FLAG_GLOBAL_HEADER; - check_av(av_opt_set(context->priv_data, "profile", "high", 0), - "av_opt_set(" - "profile" - ")"); - check_av(av_opt_set(context->priv_data, "level", "5.2", 0), - "av_opt_set(" - "level" - ")"); - check_av(av_opt_set(context->priv_data, "forced-idr", "1", 0), - "av_opt_set(" - "forced-idr" - ")"); - check_av(av_opt_set(context->priv_data, "zerolatency", "1", 0), - "av_opt_set(" - "zerolatency" - ")"); + set_option("profile", "high"); + set_option("level", "5.2"); + set_option("forced-idr", "1"); + set_option("zerolatency", "1"); + // Hand each unit back as soon as it is encoded rather than after a queue of frames. + set_option("delay", "0"); if (config.peak_bitrate_bps > config.bitrate_bps) { - context->rc_max_rate = config.peak_bitrate_bps; - context->rc_buffer_size = static_cast(config.peak_bitrate_bps); - check_av(av_opt_set(context->priv_data, "rc", "vbr", 0), - "av_opt_set(" - "rc" - ")"); + // The configured ratio between the ceiling and the target holds at any target. + context->rc_max_rate = static_cast(bitrate) * config.peak_bitrate_bps / config.bitrate_bps; + context->rc_buffer_size = static_cast(context->rc_max_rate); + set_option("rc", "vbr"); } else { - check_av(av_opt_set(context->priv_data, "rc", "cbr", 0), - "av_opt_set(" - "rc" - ")"); + set_option("rc", "cbr"); } check_av(avcodec_open2(context, codec, nullptr), "cannot open h264_nvenc"); + applied_bitrate = bitrate; + } + void alloc_frame() + { frame = av_frame_alloc(); packet = av_packet_alloc(); if (!frame || !packet) @@ -195,11 +195,30 @@ struct JetsonEncoder::Impl EncodedUnit unit; unit.data.assign(packet->data, packet->data + packet->size); unit.timestamp_ns = packet->pts * 1000; + unit.keyframe = (packet->flags & AV_PKT_FLAG_KEY) != 0; av_packet_unref(packet); out.push_back(std::move(unit)); } } + // Worker thread, which owns the codec. The libavcodec NVENC wrapper takes its rate control at + // open, so a new target drains the running session into `out` and opens another at that rate: the + // stream continues from an IDR with parameter sets. + void apply_bitrate(std::deque& out) + { + uint32_t target = 0; + { + std::lock_guard lock(mutex); + target = pending_bitrate; + pending_bitrate = 0; + } + if (target == 0 || target == applied_bitrate) + return; + check_av(avcodec_send_frame(context, nullptr), "avcodec_send_frame(flush)"); + collect(out); + open_context(target); + } + void encode(const Job& job) { check_av(av_frame_make_writable(frame), "av_frame_make_writable"); @@ -211,6 +230,8 @@ struct JetsonEncoder::Impl for (uint32_t row = 0; row < config.height / 2; ++row) std::memcpy(frame->data[1] + static_cast(row) * frame->linesize[1], uv + row * width, width); frame->pts = job.timestamp_ns / 1000; + // With forced-idr, an I picture is an IDR with its parameter sets. + frame->pict_type = job.keyframe ? AV_PICTURE_TYPE_I : AV_PICTURE_TYPE_NONE; check_av(avcodec_send_frame(context, frame), "avcodec_send_frame"); } @@ -234,6 +255,7 @@ struct JetsonEncoder::Impl std::string error; try { + apply_bitrate(produced); encode(job); collect(produced); } @@ -257,6 +279,7 @@ struct JetsonEncoder::Impl EncoderConfig config; + const AVCodec* codec = nullptr; AVCodecContext* context = nullptr; AVFrame* frame = nullptr; AVPacket* packet = nullptr; @@ -278,6 +301,9 @@ struct JetsonEncoder::Impl bool busy = false; bool closing = false; bool flushed = false; + bool keyframe_requested = false; // guarded by `mutex` + uint32_t pending_bitrate = 0; // guarded by `mutex`; 0 when nothing is waiting + uint32_t applied_bitrate = 0; // worker thread only std::thread worker; }; @@ -324,7 +350,8 @@ bool JetsonEncoder::submit(uintptr_t rgba_device_ptr, std::size_t row_pitch_byte { std::lock_guard lock(impl.mutex); - impl.jobs.push_back({ buffer, timestamp_ns }); + impl.jobs.push_back({ buffer, timestamp_ns, impl.keyframe_requested }); + impl.keyframe_requested = false; } impl.wake.notify_one(); return true; @@ -342,31 +369,56 @@ EncodedUnit JetsonEncoder::poll() return out; } -EncodedUnit JetsonEncoder::end_of_stream() +std::vector JetsonEncoder::drain() { auto& impl = *m_impl; std::unique_lock lock(impl.mutex); impl.idle.wait(lock, [&] { return impl.jobs.empty() && !impl.busy; }); if (!impl.failure.empty()) throw std::runtime_error(impl.failure); + if (!impl.flushed) + { + impl.flushed = true; + check_av(avcodec_send_frame(impl.context, nullptr), "avcodec_send_frame(flush)"); + impl.collect(impl.ready); + } + std::vector units( + std::make_move_iterator(impl.ready.begin()), std::make_move_iterator(impl.ready.end())); + impl.ready.clear(); + return units; +} - if (impl.flushed) - return {}; - impl.flushed = true; - - // Flush the encoder, then concatenate what is left, so the tail carries the stamp of the last unit. - check_av(avcodec_send_frame(impl.context, nullptr), "avcodec_send_frame(flush)"); - impl.collect(impl.ready); - +EncodedUnit JetsonEncoder::end_of_stream() +{ + // The tail is concatenated, so it carries the stamp of the last unit in it. EncodedUnit out; - for (auto& chunk : impl.ready) + for (auto& chunk : drain()) { out.data.insert(out.data.end(), chunk.data.begin(), chunk.data.end()); out.timestamp_ns = chunk.timestamp_ns; } - impl.ready.clear(); return out; } +void JetsonEncoder::request_keyframe() +{ + std::lock_guard lock(m_impl->mutex); + m_impl->keyframe_requested = true; +} + +bool JetsonEncoder::set_bitrate(uint32_t bitrate_bps) +{ + if (bitrate_bps == 0) + return false; + std::lock_guard lock(m_impl->mutex); + m_impl->pending_bitrate = bitrate_bps; + return true; +} + +const char* JetsonEncoder::backend_name() +{ + return "NVENC (libavcodec h264_nvenc)"; +} + } // namespace sensing } // namespace plugins diff --git a/src/plugins/sensing/core/rgba_box_downscale.cu b/src/plugins/sensing/core/rgba_box_downscale.cu new file mode 100644 index 0000000000..cc169b3b92 --- /dev/null +++ b/src/plugins/sensing/core/rgba_box_downscale.cu @@ -0,0 +1,57 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "rgba_box_downscale.cuh" + +namespace plugins +{ +namespace sensing +{ + +namespace +{ + +__global__ void box_downscale( + const uint8_t* src, int src_pitch, uint8_t* dst, int dst_pitch, int width, int height, int divisor) +{ + const int x = blockIdx.x * blockDim.x + threadIdx.x; + const int y = blockIdx.y * blockDim.y + threadIdx.y; + if (x >= width || y >= height) + return; + + int sum[4] = { 0, 0, 0, 0 }; + for (int j = 0; j < divisor; ++j) + { + const uint8_t* row = + src + static_cast(y * divisor + j) * src_pitch + static_cast(x * divisor) * 4; + for (int i = 0; i < divisor; ++i) + { + for (int c = 0; c < 4; ++c) + sum[c] += row[i * 4 + c]; + } + } + + const int count = divisor * divisor; + uint8_t* out = dst + static_cast(y) * dst_pitch + static_cast(x) * 4; + for (int c = 0; c < 4; ++c) + out[c] = static_cast((sum[c] + count / 2) / count); +} + +} // namespace + +void launch_rgba_box_downscale(const uint8_t* src, + int src_pitch, + uint8_t* dst, + int dst_pitch, + int dst_width, + int dst_height, + int divisor, + cudaStream_t stream) +{ + const dim3 block(16, 16); + const dim3 grid((dst_width + block.x - 1) / block.x, (dst_height + block.y - 1) / block.y); + box_downscale<<>>(src, src_pitch, dst, dst_pitch, dst_width, dst_height, divisor); +} + +} // namespace sensing +} // namespace plugins diff --git a/src/plugins/sensing/core/rgba_box_downscale.cuh b/src/plugins/sensing/core/rgba_box_downscale.cuh new file mode 100644 index 0000000000..4a96e37a80 --- /dev/null +++ b/src/plugins/sensing/core/rgba_box_downscale.cuh @@ -0,0 +1,33 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include + +namespace plugins +{ +namespace sensing +{ + +/** + * @brief Reduce RGBA8 by an integer divisor, averaging each divisor x divisor block. + * + * The aspect ratio is kept exactly: the source size must divide evenly, so nothing is stretched, + * cropped or rounded. + * + * @param dst_width src_width / divisor. + * @param dst_height src_height / divisor. + */ +void launch_rgba_box_downscale(const uint8_t* src, + int src_pitch, + uint8_t* dst, + int dst_pitch, + int dst_width, + int dst_height, + int divisor, + cudaStream_t stream); + +} // namespace sensing +} // namespace plugins diff --git a/src/plugins/sensing/core/sensing_camera.cpp b/src/plugins/sensing/core/sensing_camera.cpp index 03a60c0f31..a2ba019735 100644 --- a/src/plugins/sensing/core/sensing_camera.cpp +++ b/src/plugins/sensing/core/sensing_camera.cpp @@ -7,9 +7,30 @@ #include #include +#include #include #include +#if SENSING_WITH_LIVEKIT +# include "livekit_video_output.hpp" + +# include +#else +// Without LiveKit the members are never set; empty types let the camera's special members compile. +namespace plugins +{ +namespace sensing +{ +class LiveKitSession +{ +}; +class LiveKitVideoOutput +{ +}; +} // namespace sensing +} // namespace plugins +#endif + namespace plugins { namespace sensing @@ -58,6 +79,24 @@ SensingCamera::SensingCamera(const SensingConfig& config, m_camera = std::make_unique(make_sipl_config(config)); const auto& sensors = m_camera->sensors(); +#if SENSING_WITH_LIVEKIT + // Everything that needs no network is checked first, so a bad name or divisor fails before any + // connection is made. + std::set livekit_tracks; + for (const auto& stream_config : streams) + { + if (stream_config.livekit_track.empty()) + continue; + if (!livekit_tracks.insert(stream_config.livekit_track).second) + throw std::runtime_error("LiveKit track name '" + stream_config.livekit_track + "' is used twice"); + for (const auto& sensor : sensors) + { + if (sensor.id == stream_config.sensor_id) + LiveKitVideoOutput::validate_geometry(sensor.width, sensor.height, config.livekit.scale_div); + } + } +#endif + m_streams.reserve(streams.size()); for (const auto& stream_config : streams) { @@ -93,6 +132,36 @@ SensingCamera::SensingCamera(const SensingConfig& config, stream.publisher = std::make_unique(ipc); } + if (!stream_config.livekit_track.empty()) + { +#if SENSING_WITH_LIVEKIT + LiveKitOutputConfig output; + output.source_width = found->width; + output.source_height = found->height; + output.scale_div = config.livekit.scale_div; + output.fps = config.livekit.fps; + 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.gop = config.gop; + output.full_range = config.full_range; + if (!m_livekit) + m_livekit = LiveKitSession::connect(config.livekit.url, config.livekit.token); + + PublishSpec spec; + spec.name = stream_config.livekit_track; + spec.width = found->width / output.scale_div; + spec.height = found->height / output.scale_div; + spec.max_bitrate_bps = output.peak_bitrate_bps; + spec.max_fps = static_cast(std::min(output.fps, found->fps)); + stream.livekit = std::make_unique(m_livekit->publish_video(spec), output); + std::cout << "LiveKit track '" << spec.name << "': " << spec.width << "x" << spec.height << " @ up to " + << spec.max_fps << " fps" << std::endl; +#else + throw std::runtime_error("this build has no LiveKit support; configure with -DSENSING_LIVEKIT=ON"); +#endif + } + m_streams.push_back(std::move(stream)); std::cout << "Sensor " << found->id << " (" << found->name << "): " << found->width << "x" << found->height @@ -129,6 +198,12 @@ int64_t SensingCamera::take_capture_tsc(Stream& stream, int64_t timestamp_ns) void SensingCamera::update() { +#if SENSING_WITH_LIVEKIT + // The room is gone for good: stop rather than keep capturing for nobody. + if (m_livekit && m_livekit->failed()) + throw std::runtime_error(m_livekit->failure()); +#endif + for (auto& stream : m_streams) { // Accept consumers and reap released slots even on a frameless tick, @@ -149,6 +224,12 @@ void SensingCamera::update() if (stream.publisher) stream.publisher->publish(view->ptr, view->pitch, view->timestamp_ns); +#if SENSING_WITH_LIVEKIT + // A GPU snapshot only; the conversion and the network are on the output's own thread. + if (stream.livekit) + stream.livekit->submit(*view); +#endif + if (stream.encoder) { if (stream.encoder->submit(view->ptr, view->pitch, static_cast(view->timestamp_ns))) @@ -170,6 +251,11 @@ void SensingCamera::update() } } +#if SENSING_WITH_LIVEKIT + if (stream.livekit && stream.livekit->failed()) + throw std::runtime_error("sensor " + std::to_string(stream.sensor_id) + ": " + stream.livekit->failure()); +#endif + // Submission and output are decoupled: the V4L2 encoder needs several // input frames before it emits the first unit, so drain independently. // Each unit carries its own frame's stamp back from the output plane. @@ -188,6 +274,10 @@ void SensingCamera::flush() { for (auto& stream : m_streams) { +#if SENSING_WITH_LIVEKIT + if (stream.livekit) + stream.livekit->flush(); +#endif if (!stream.encoder) continue; auto unit = stream.encoder->end_of_stream(); @@ -234,6 +324,23 @@ void SensingCamera::print_stats() const << stream.publisher->dropped_count() << " dropped" << (stream.publisher->has_consumer() ? "" : ", no consumer"); } +#if SENSING_WITH_LIVEKIT + if (stream.livekit) + { + const auto s = stream.livekit->stats(); + std::cout << " | livekit " << s.published << " published, " << s.paced_out << " paced out, " << s.replaced + << " replaced"; + if (s.encoder_busy) + std::cout << ", " << s.encoder_busy << " encoder busy"; + if (s.rejected) + std::cout << ", " << s.rejected << " rejected"; + if (s.keyframes_requested) + std::cout << ", " << s.keyframes_requested << " keyframes requested"; + const std::string encoder = stream.livekit->describe(); + if (!encoder.empty()) + std::cout << " (" << encoder << ")"; + } +#endif std::cout << std::endl; } } diff --git a/src/plugins/sensing/core/sensing_camera.hpp b/src/plugins/sensing/core/sensing_camera.hpp index 71b47d5273..750d12cbeb 100644 --- a/src/plugins/sensing/core/sensing_camera.hpp +++ b/src/plugins/sensing/core/sensing_camera.hpp @@ -20,6 +20,8 @@ namespace sensing { class FrameSink; +class LiveKitSession; +class LiveKitVideoOutput; /** * @brief Multi-sensor SENSING camera manager. @@ -63,6 +65,8 @@ class SensingCamera std::unique_ptr encoder; /// Null when the stream requested no CUDA IPC socket. std::unique_ptr publisher; + /// Null when the stream requested no LiveKit track. + std::unique_ptr livekit; /// SIPL publish counter of the frame already handled; skips re-reads. uint64_t last_sequence = 0; bool have_last_sequence = false; @@ -83,6 +87,8 @@ class SensingCamera SensingConfig m_config; std::unique_ptr m_camera; + /// Declared before the streams so it is destroyed after their tracks have been unpublished. + std::unique_ptr m_livekit; std::vector m_streams; std::unique_ptr m_sink; }; diff --git a/src/plugins/sensing/core/sensing_types.hpp b/src/plugins/sensing/core/sensing_types.hpp index b2f29a92f3..2b84b89d61 100644 --- a/src/plugins/sensing/core/sensing_types.hpp +++ b/src/plugins/sensing/core/sensing_types.hpp @@ -19,9 +19,9 @@ namespace plugins namespace sensing { -/// One captured sensor and where its frames go. Either destination may be -/// empty: an ipc-only stream skips the encoder entirely, which is the point of -/// the CUDA path. +/// One captured sensor and where its frames go. Any destination may be empty: an +/// ipc-only stream skips the encoder entirely, and a livekit-only stream has no recording +/// encoder, only the live track's own. struct StreamConfig { /// SIPL pipeline index, validated against the platform config at startup. @@ -30,6 +30,23 @@ struct StreamConfig std::string output_path; /// Unix socket serving RGBA8 frames as CUDA memory; empty disables it. std::string ipc_socket_path; + /// Name of the LiveKit camera track carrying this sensor; empty disables it. + std::string livekit_track; +}; + +/// Where and how the livekit= streams are published. +struct LiveKitConfig +{ + std::string url; + /// The join token itself, not a path. It names the room and the participant. Never logged. + std::string token; + /// Publication rate cap, independent of the capture rate. + uint32_t fps = 30; + /// Output-only uniform reduction of every track; 1 publishes the sensor's native size. + uint32_t scale_div = 1; + /// Target bitrate of each track's hardware encoder; the VBR ceiling is half as much again. A + /// live transport, unlike the recording, cannot take the recording's rate. + uint32_t bitrate_bps = 8'000'000; }; struct SensingConfig @@ -58,6 +75,9 @@ struct SensingConfig bool swap_uv = false; /// ISP0 buffers per sensor, ~6 MB each at 2560x1984 NV12. uint32_t isp0_buffers = 6; + + /// Used when any stream names a livekit= track. + LiveKitConfig livekit; }; struct SensingFrame diff --git a/src/plugins/sensing/livekit/CMakeLists.txt b/src/plugins/sensing/livekit/CMakeLists.txt new file mode 100644 index 0000000000..7d0b061953 --- /dev/null +++ b/src/plugins/sensing/livekit/CMakeLists.txt @@ -0,0 +1,24 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# The LiveKit SDK behind a small interface: join a room, publish a video track, submit NV12 frames. +# No LiveKit type appears in an installed header, and nothing here touches CUDA. +add_library(sensing_livekit STATIC + livekit_session.cpp + inc/sensing_livekit/livekit_session.hpp + inc/sensing_livekit/video_track_sink.hpp +) + +target_include_directories(sensing_livekit + INTERFACE + $ +) + +target_link_libraries(sensing_livekit + PUBLIC + log_bridge::log_bridge_core + PRIVATE + LiveKit::livekit +) + +add_library(sensing::livekit ALIAS sensing_livekit) diff --git a/src/plugins/sensing/livekit/inc/sensing_livekit/livekit_session.hpp b/src/plugins/sensing/livekit/inc/sensing_livekit/livekit_session.hpp new file mode 100644 index 0000000000..7cf33bcfea --- /dev/null +++ b/src/plugins/sensing/livekit/inc/sensing_livekit/livekit_session.hpp @@ -0,0 +1,78 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include "video_track_sink.hpp" + +#include +#include +#include + +namespace plugins +{ +namespace sensing +{ + +struct LiveKitSessionImpl; + +/** + * @brief Keeps the LiveKit SDK initialised while it is held. + * + * The SDK is process-wide and must be shut down only when nothing in the process uses it, so every + * user holds one of these and the last release shuts it down. + */ +class LiveKitSdkReference +{ +public: + LiveKitSdkReference(); + ~LiveKitSdkReference(); + LiveKitSdkReference(const LiveKitSdkReference&) = delete; + LiveKitSdkReference& operator=(const LiveKitSdkReference&) = delete; +}; + +struct PublishSpec +{ + /// Name subscribers see for the track. + std::string name; + uint32_t width = 0; + uint32_t height = 0; + /// Upper bound the upstream encoder may use; WebRTC adapts below it. + uint64_t max_bitrate_bps = 0; + double max_fps = 30.0; +}; + +/** + * @brief One participant in one LiveKit room. + * + * The room and the participant's identity come from the join token. A terminal loss of the room + * (the server went away, the participant was removed) sets failed(); the SDK's own bounded + * reconnection runs first and is not wrapped in another supervisor. + */ +class LiveKitSession +{ +public: + /// @throws std::runtime_error if the room cannot be joined. The token is never logged. + static std::unique_ptr connect(const std::string& url, const std::string& token); + + /// Unpublish every track (destroy them first), then leave the room. + ~LiveKitSession(); + + LiveKitSession(const LiveKitSession&) = delete; + LiveKitSession& operator=(const LiveKitSession&) = delete; + + /// Publishes an H.264 camera track that carries pre-encoded access units; the SDK does not encode. + /// @throws std::runtime_error if publication fails. + std::unique_ptr publish_video(const PublishSpec& spec); + + /// True once the room has been lost for good. + bool failed() const; + std::string failure() const; + +private: + explicit LiveKitSession(std::shared_ptr impl); + std::shared_ptr m_impl; +}; + +} // namespace sensing +} // namespace plugins diff --git a/src/plugins/sensing/livekit/inc/sensing_livekit/video_track_sink.hpp b/src/plugins/sensing/livekit/inc/sensing_livekit/video_track_sink.hpp new file mode 100644 index 0000000000..a69a5ea1af --- /dev/null +++ b/src/plugins/sensing/livekit/inc/sensing_livekit/video_track_sink.hpp @@ -0,0 +1,70 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include +#include +#include + +namespace plugins +{ +namespace sensing +{ + +/// One complete encoded H.264 access unit, with what accompanies it. +struct EncodedVideoFrame +{ + /// Annex-B bytes: an IDR carries its SPS/PPS in front of it. + const uint8_t* data = nullptr; + size_t size = 0; + /// The unit is an IDR, decodable on its own. + bool keyframe = false; + /// CLOCK_MONOTONIC microseconds at capture: the host timeline, not sensor time. + uint64_t timestamp_us = 0; + /// The camera's own frame sequence. + uint32_t frame_id = 0; + /// The sensor's capture time on its own clock, carried as user data so a consumer can pair + /// frames from different sensors. + uint64_t capture_tsc_ns = 0; +}; + +/// What the receiving side asked of the encoder since the last call. +struct EncoderFeedback +{ + /// The next frame must be an IDR with its parameter sets. + bool keyframe_requested = false; + /// Latest bitrate target from congestion control, if one was requested. + std::optional target_bitrate_bps; +}; + +/** + * @brief One published video track that carries frames already encoded by hardware. + * + * The SDK forwards each access unit without encoding it again. A single thread must drive one sink. + */ +class VideoTrackSink +{ +public: + virtual ~VideoTrackSink() = default; + + /// Size of the encoded pictures this track was published for. + virtual uint32_t width() const = 0; + virtual uint32_t height() const = 0; + + /// Publishes one access unit; the bytes are copied before this returns. + /// @return false when the SDK did not accept the frame, which can break the reference chain: + /// the caller must follow with a keyframe. + /// @throws std::runtime_error when the frame cannot be published at all. + virtual bool submit(const EncodedVideoFrame& frame) = 0; + + /// Keyframe and rate requests accumulated since the previous call. + virtual EncoderFeedback take_feedback() = 0; + + /// One line naming the transport's view of the stream, or empty if unavailable. + virtual std::string describe() = 0; +}; + +} // namespace sensing +} // namespace plugins diff --git a/src/plugins/sensing/livekit/livekit_session.cpp b/src/plugins/sensing/livekit/livekit_session.cpp new file mode 100644 index 0000000000..70fa5853a7 --- /dev/null +++ b/src/plugins/sensing/livekit/livekit_session.cpp @@ -0,0 +1,301 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "inc/sensing_livekit/livekit_session.hpp" + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace plugins +{ +namespace sensing +{ + +namespace +{ + +std::mutex g_sdk_mutex; +int g_sdk_users = 0; + +const char* reason_name(livekit::DisconnectReason reason) +{ + using R = livekit::DisconnectReason; + switch (reason) + { + case R::ClientInitiated: + return "client initiated"; + case R::DuplicateIdentity: + return "another participant joined with the same identity"; + case R::ServerShutdown: + return "the server is shutting down"; + case R::ParticipantRemoved: + return "the participant was removed from the room"; + case R::RoomDeleted: + return "the room was deleted"; + default: + return "connection lost"; + } +} + +} // namespace + +LiveKitSdkReference::LiveKitSdkReference() +{ + std::lock_guard lock(g_sdk_mutex); + if (g_sdk_users == 0) + livekit::initialize(livekit::LogLevel::Warn); + ++g_sdk_users; +} + +LiveKitSdkReference::~LiveKitSdkReference() +{ + std::lock_guard lock(g_sdk_mutex); + if (--g_sdk_users == 0) + livekit::shutdown(); +} + +struct LiveKitSessionImpl : livekit::RoomDelegate +{ + ~LiveKitSessionImpl() override + { + // Leave the room while the SDK is still up: the reference below is released after the room. + room.disconnect(); + } + + void onDisconnected(livekit::Room&, const livekit::DisconnectedEvent& event) override + { + if (event.reason == livekit::DisconnectReason::ClientInitiated) + return; + fail(std::string("LiveKit room lost: ") + reason_name(event.reason)); + } + + void onReconnecting(livekit::Room&, const livekit::ReconnectingEvent&) override + { + logger->warn("LiveKit connection interrupted; the SDK is reconnecting"); + } + + void onReconnected(livekit::Room&, const livekit::ReconnectedEvent&) override + { + logger->info("LiveKit connection restored"); + } + + void fail(const std::string& message) + { + std::lock_guard lock(mutex); + if (!failed.load()) + { + failure = message; + failed.store(true); + logger->error("{}", message); + } + } + + // Declared before the room, so destroyed after it. + LiveKitSdkReference sdk; + livekit::Room room; + std::atomic failed{ false }; + mutable std::mutex mutex; + std::string failure; + std::shared_ptr logger = isaaccapture::Logger::get("isaaccapture.plugins.sensing.LiveKitSession"); +}; + +namespace +{ + +/// A track fed with H.264 access units that were encoded upstream. +class LiveKitVideoTrack : public VideoTrackSink +{ +public: + LiveKitVideoTrack(std::shared_ptr session, const PublishSpec& spec) + : m_session(std::move(session)), m_width(spec.width), m_height(spec.height) + { + m_source = std::make_shared( + livekit::VideoCodec::H264, static_cast(spec.width), static_cast(spec.height)); + m_track = livekit::LocalVideoTrack::createLocalVideoTrack(spec.name, m_source); + + livekit::TrackPublishOptions options; + options.source = livekit::TrackSource::SOURCE_CAMERA; + options.video_codec = livekit::VideoCodec::H264; + // Pre-encoded access units go through as they are: no software or SDK-selected encoder. + options.video_encoder = livekit::VideoEncoderBackend::PreEncoded; + options.simulcast = false; + options.video_encoding = livekit::VideoEncodingOptions{ spec.max_bitrate_bps, spec.max_fps }; + options.frame_metadata_features = livekit::FrameMetadataFeatures{ true, true, true }; + + auto participant = m_session->room.localParticipant().lock(); + if (!participant) + throw std::runtime_error("LiveKit: no local participant to publish '" + spec.name + "'"); + try + { + participant->publishTrack(m_track, options); + } + catch (const std::exception& e) + { + throw std::runtime_error("LiveKit: cannot publish track '" + spec.name + "': " + e.what()); + } + m_participant = participant; + } + + ~LiveKitVideoTrack() override + { + try + { + if (auto participant = m_participant.lock()) + participant->unpublishTrack(m_track->sid()); + } + catch (const std::exception& e) + { + m_session->logger->warn("LiveKit: unpublish failed: {}", e.what()); + } + } + + uint32_t width() const override + { + return m_width; + } + uint32_t height() const override + { + return m_height; + } + + bool submit(const EncodedVideoFrame& frame) override + { + if (m_session->failed.load()) + throw std::runtime_error(m_session->failure); + + livekit::EncodedVideoSource::Frame out; + out.is_keyframe = frame.keyframe; + out.width = m_width; + out.height = m_height; + out.timestamp_us = static_cast(frame.timestamp_us); + out.data.assign(frame.data, frame.data + frame.size); + livekit::VideoFrameMetadata metadata; + metadata.user_timestamp_us = frame.timestamp_us; + metadata.frame_id = frame.frame_id; + std::vector tsc(8); + for (int i = 0; i < 8; ++i) + tsc[i] = static_cast((frame.capture_tsc_ns >> (8 * i)) & 0xFF); + metadata.user_data = std::move(tsc); + out.metadata = std::move(metadata); + + try + { + return m_source->captureFrame(out); + } + catch (const std::exception& e) + { + throw std::runtime_error(std::string("LiveKit: cannot submit a frame: ") + e.what()); + } + } + + EncoderFeedback take_feedback() override + { + EncoderFeedback feedback; + try + { + const auto pending = m_source->takeFeedback(); + feedback.keyframe_requested = pending.keyframe_requested; + if (pending.rate_control && pending.rate_control->target_bitrate_bps > 0) + feedback.target_bitrate_bps = static_cast(pending.rate_control->target_bitrate_bps); + } + catch (const std::exception& e) + { + throw std::runtime_error(std::string("LiveKit: cannot read encoder feedback: ") + e.what()); + } + return feedback; + } + + // Called from the polling thread, which must not wait on the network: this returns the last answer + // and starts, or collects, a stats request that was asked for earlier. + std::string describe() override + { + try + { + if (m_stats.valid() && m_stats.wait_for(std::chrono::seconds(0)) == std::future_status::ready) + { + for (const auto& entry : m_stats.get()) + { + const auto* out = std::get_if(&entry.stats); + if (!out || out->outbound.frame_width == 0) + continue; + const auto& o = out->outbound; + m_description = "encoder " + o.encoder_implementation + ", " + std::to_string(o.frame_width) + "x" + + std::to_string(o.frame_height) + " @ " + + std::to_string(static_cast(o.frames_per_second)) + " fps, " + + std::to_string(static_cast(o.target_bitrate / 1000)) + " kbps target"; + break; + } + } + if (!m_stats.valid()) + m_stats = m_track->getStats(); + } + catch (const std::exception&) + { + } + return m_description; + } + +private: + std::shared_ptr m_session; + uint32_t m_width; + uint32_t m_height; + std::shared_ptr m_source; + std::shared_ptr m_track; + std::weak_ptr m_participant; + decltype(std::declval().getStats()) m_stats; + std::string m_description; +}; + +} // namespace + +LiveKitSession::LiveKitSession(std::shared_ptr impl) : m_impl(std::move(impl)) +{ +} + +LiveKitSession::~LiveKitSession() = default; + +std::unique_ptr LiveKitSession::connect(const std::string& url, const std::string& token) +{ + auto impl = std::make_shared(); + impl->room.setDelegate(impl.get()); + + livekit::RoomOptions options; + options.auto_subscribe = false; // this process only publishes + options.dynacast = false; + if (!impl->room.connect(url, token, options)) + { + throw std::runtime_error("LiveKit: cannot join the room at " + url + + " (check the URL, that the server is reachable and that the token is valid)"); + } + impl->logger->info("Joined LiveKit room at {}", url); + return std::unique_ptr(new LiveKitSession(std::move(impl))); +} + +std::unique_ptr LiveKitSession::publish_video(const PublishSpec& spec) +{ + return std::make_unique(m_impl, spec); +} + +bool LiveKitSession::failed() const +{ + return m_impl->failed.load(); +} + +std::string LiveKitSession::failure() const +{ + std::lock_guard lock(m_impl->mutex); + return m_impl->failure; +} + +} // namespace sensing +} // namespace plugins diff --git a/src/plugins/sensing/livekit/receiver/index.html b/src/plugins/sensing/livekit/receiver/index.html new file mode 100644 index 0000000000..86f4a470e6 --- /dev/null +++ b/src/plugins/sensing/livekit/receiver/index.html @@ -0,0 +1,108 @@ + + + + + + +SENSING LiveKit receiver + + + +
+ + + + +
+

Not connected.

+
+ + + + + diff --git a/tests/cpp/plugins/sensing_mock/CMakeLists.txt b/tests/cpp/plugins/sensing_mock/CMakeLists.txt index 1b13b47254..ec168d1b3f 100644 --- a/tests/cpp/plugins/sensing_mock/CMakeLists.txt +++ b/tests/cpp/plugins/sensing_mock/CMakeLists.txt @@ -33,4 +33,25 @@ target_compile_definitions(sensing_mock_tests PRIVATE SENSING_BUNDLED_GIF_DIR="${CMAKE_SOURCE_DIR}/src/plugins/sensing/mock_assets" ) +# LiveKit streaming tests. The unit tests use a fake track; the integration tests also need a +# livekit-server binary, named by LIVEKIT_SERVER, and skip without it. +if(SENSING_LIVEKIT) + find_package(LiveKit CONFIG REQUIRED PATHS "${SENSING_LIVEKIT_SDK_DIR}" NO_DEFAULT_PATH) + find_package(OpenSSL REQUIRED) + target_sources(sensing_mock_tests PRIVATE + livekit_test_support.hpp + test_livekit_cli.cpp + test_livekit_output.cpp + test_livekit_stream.cpp + ) + target_link_libraries(sensing_mock_tests + PRIVATE + sensing::livekit + LiveKit::livekit + OpenSSL::Crypto + ) # test_livekit_cli.cpp runs the plugin binary, so it has to be built first. + add_dependencies(sensing_mock_tests camera_plugin_sensing) + target_compile_definitions(sensing_mock_tests PRIVATE SENSING_PLUGIN_BINARY="$") +endif() + catch_discover_tests(sensing_mock_tests ADD_TAGS_AS_LABELS) diff --git a/tests/cpp/plugins/sensing_mock/livekit_test_support.hpp b/tests/cpp/plugins/sensing_mock/livekit_test_support.hpp new file mode 100644 index 0000000000..5d9f693131 --- /dev/null +++ b/tests/cpp/plugins/sensing_mock/livekit_test_support.hpp @@ -0,0 +1,308 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// A local LiveKit server, join tokens, and a subscriber that records what it receives, so the +// SENSING publisher can be checked against a real LiveKit service. + +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace livekit_test +{ + +constexpr const char* kApiKey = "devkey"; +constexpr const char* kApiSecret = "secret"; + +inline std::string base64url(const unsigned char* data, size_t size) +{ + static const char* alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"; + std::string out; + for (size_t i = 0; i < size; i += 3) + { + const uint32_t n = (data[i] << 16) | ((i + 1 < size ? data[i + 1] : 0) << 8) | (i + 2 < size ? data[i + 2] : 0); + out.push_back(alphabet[(n >> 18) & 63]); + out.push_back(alphabet[(n >> 12) & 63]); + if (i + 1 < size) + out.push_back(alphabet[(n >> 6) & 63]); + if (i + 2 < size) + out.push_back(alphabet[n & 63]); + } + return out; +} + +inline std::string base64url(const std::string& text) +{ + return base64url(reinterpret_cast(text.data()), text.size()); +} + +/// An HS256 LiveKit access token for the local server. +inline std::string make_token(const std::string& identity, + const std::string& room, + bool can_publish, + const std::string& secret = kApiSecret) +{ + const auto now = + std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count(); + const std::string header = R"({"alg":"HS256","typ":"JWT"})"; + const std::string claims = std::string(R"({"iss":")") + kApiKey + R"(","sub":")" + identity + R"(","name":")" + + identity + R"(","nbf":)" + std::to_string(now) + R"(,"exp":)" + + std::to_string(now + 3600) + R"(,"video":{"roomJoin":true,"room":")" + room + + R"(","canPublish":)" + (can_publish ? "true" : "false") + R"(,"canSubscribe":)" + + (can_publish ? "false" : "true") + "}}"; + const std::string signing = base64url(header) + "." + base64url(claims); + unsigned char mac[EVP_MAX_MD_SIZE]; + unsigned int mac_size = 0; + HMAC(EVP_sha256(), secret.data(), static_cast(secret.size()), + reinterpret_cast(signing.data()), signing.size(), mac, &mac_size); + return signing + "." + base64url(mac, mac_size); +} + +/// A livekit-server child process on private ports. Tests skip when LIVEKIT_SERVER is not set. +class LocalServer +{ +public: + static const char* binary() + { + return std::getenv("LIVEKIT_SERVER"); + } + + LocalServer() + { + static std::atomic counter{ 0 }; + m_port = 21000 + static_cast(::getpid() % 2000) * 4 + counter++ * 4000 % 7000; + const std::string config = "port: " + std::to_string(m_port) + + "\nrtc:\n tcp_port: " + std::to_string(m_port + 1) + + "\n udp_port: " + std::to_string(m_port + 2) + "\n use_external_ip: false\n"; + const std::string keys = std::string(kApiKey) + ": " + kApiSecret; + m_pid = ::fork(); + if (m_pid == 0) + { + // Child: silence the server's logs, then become it. + if (FILE* devnull = std::fopen("/dev/null", "w")) + { + ::dup2(::fileno(devnull), 1); + ::dup2(::fileno(devnull), 2); + } + ::execl(binary(), binary(), "--bind", "127.0.0.1", "--node-ip", "127.0.0.1", "--keys", keys.c_str(), + "--config-body", config.c_str(), static_cast(nullptr)); + ::_exit(127); + } + wait_until_listening(); + } + + ~LocalServer() + { + stop(); + } + + LocalServer(const LocalServer&) = delete; + LocalServer& operator=(const LocalServer&) = delete; + + std::string url() const + { + return "ws://127.0.0.1:" + std::to_string(m_port); + } + + /// Kills the server, as a crash or a network loss would. + void stop() + { + if (m_pid > 0) + { + ::kill(m_pid, SIGKILL); + ::waitpid(m_pid, nullptr, 0); + m_pid = -1; + } + } + +private: + void wait_until_listening() + { + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(15); + while (std::chrono::steady_clock::now() < deadline) + { + const int fd = ::socket(AF_INET, SOCK_STREAM, 0); + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_port = htons(static_cast(m_port)); + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + const bool up = ::connect(fd, reinterpret_cast(&address), sizeof(address)) == 0; + ::close(fd); + if (up) + return; + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + throw std::runtime_error("the local LiveKit server did not start"); + } + + pid_t m_pid = -1; + int m_port = 0; +}; + +/// What a subscriber has received on one track. +struct TrackSeen +{ + uint64_t frames = 0; + int width = 0; + int height = 0; + std::vector last_rgba; + std::vector frame_ids; + std::map tsc_by_frame; ///< capture TSC carried as user data + uint64_t with_user_timestamp = 0; +}; + +/// A second participant that subscribes to every video track and records it. +class Subscriber : public livekit::RoomDelegate +{ +public: + Subscriber(const std::string& url, const std::string& room) + { + m_room = std::make_unique(); + m_room->setDelegate(this); + livekit::RoomOptions options; + if (!m_room->connect(url, make_token("test-subscriber", room, false), options)) + throw std::runtime_error("the subscriber could not join the room"); + } + + ~Subscriber() override + { + // The SDK callback takes the mutex, so claim the streams and readers under it, then close and join + // with it released. + std::vector> streams; + std::vector readers; + { + std::lock_guard lock(m_mutex); + m_closing = true; + streams.swap(m_streams); + readers.swap(m_readers); + } + for (auto& stream : streams) + stream->close(); + for (auto& reader : readers) + reader.join(); + m_room->disconnect(); + m_room.reset(); + } + + void onTrackSubscribed(livekit::Room&, const livekit::TrackSubscribedEvent& event) override + { + if (!event.track || event.track->kind() != livekit::TrackKind::KIND_VIDEO) + return; + std::lock_guard lock(m_mutex); + if (m_closing) + return; + livekit::VideoStream::Options options; + options.capacity = 16; + auto stream = livekit::VideoStream::fromTrack(event.track, options); + const std::string name = event.track->name(); + m_streams.push_back(stream); + m_readers.emplace_back([this, stream, name] { read(stream, name); }); + } + + void onParticipantDisconnected(livekit::Room&, const livekit::ParticipantDisconnectedEvent&) override + { + ++m_participants_left; + } + + /// True once every named track has delivered at least `frames` frames. + bool wait_for_frames(const std::vector& names, uint64_t frames, std::chrono::milliseconds limit) + { + std::unique_lock lock(m_mutex); + return m_changed.wait_for(lock, limit, + [&] + { + for (const auto& name : names) + { + auto it = m_seen.find(name); + if (it == m_seen.end() || it->second.frames < frames) + return false; + } + return true; + }); + } + + TrackSeen seen(const std::string& name) + { + std::lock_guard lock(m_mutex); + auto it = m_seen.find(name); + return it == m_seen.end() ? TrackSeen{} : it->second; + } + + std::set track_names() + { + std::lock_guard lock(m_mutex); + std::set names; + for (const auto& entry : m_seen) + names.insert(entry.first); + return names; + } + + int participants_left() const + { + return m_participants_left.load(); + } + +private: + void read(const std::shared_ptr& stream, const std::string& name) + { + livekit::VideoFrameEvent event; + while (stream->read(event)) + { + std::lock_guard lock(m_mutex); + TrackSeen& seen = m_seen[name]; + ++seen.frames; + seen.width = event.frame.width(); + seen.height = event.frame.height(); + seen.last_rgba.assign(event.frame.data(), event.frame.data() + event.frame.dataSize()); + if (event.metadata) + { + if (event.metadata->frame_id) + seen.frame_ids.push_back(*event.metadata->frame_id); + if (event.metadata->user_timestamp_us) + ++seen.with_user_timestamp; + if (event.metadata->frame_id && event.metadata->user_data && event.metadata->user_data->size() == 8) + { + uint64_t tsc = 0; + for (int i = 7; i >= 0; --i) + tsc = (tsc << 8) | (*event.metadata->user_data)[i]; + seen.tsc_by_frame[*event.metadata->frame_id] = tsc; + } + } + m_changed.notify_all(); + } + } + + // Keeps the SDK up for as long as this subscriber's room exists, alongside the publisher's. + plugins::sensing::LiveKitSdkReference m_sdk; + std::unique_ptr m_room; + std::mutex m_mutex; + std::condition_variable m_changed; + std::map m_seen; + std::vector> m_streams; + std::vector m_readers; + std::atomic m_participants_left{ 0 }; + bool m_closing = false; +}; + +} // namespace livekit_test diff --git a/tests/cpp/plugins/sensing_mock/mock_test_support.hpp b/tests/cpp/plugins/sensing_mock/mock_test_support.hpp index 2bd18c035f..b39c9ed83a 100644 --- a/tests/cpp/plugins/sensing_mock/mock_test_support.hpp +++ b/tests/cpp/plugins/sensing_mock/mock_test_support.hpp @@ -7,6 +7,7 @@ #include #include +#include #include extern "C" @@ -15,6 +16,7 @@ extern "C" #include } +#include #include #include #include @@ -272,6 +274,8 @@ struct Decoded int width = 0; int height = 0; double luma = 0; + double luma_left = 0; ///< mean luma of the left half + double luma_right = 0; ///< mean luma of the right half }; /// Decodes an Annex-B stream with libavcodec's software H.264 decoder, in output order. Any parser or @@ -307,13 +311,20 @@ inline std::vector decode_annex_b(const std::vector& stream) picture.width = frame->width; picture.height = frame->height; double sum = 0; + double left = 0; for (int y = 0; y < frame->height; ++y) { const uint8_t* row = frame->data[0] + static_cast(y) * frame->linesize[0]; for (int x = 0; x < frame->width; ++x) + { sum += row[x]; + left += x < frame->width / 2 ? row[x] : 0; + } } - picture.luma = sum / (static_cast(frame->width) * frame->height); + const double pixels = static_cast(frame->width) * frame->height; + picture.luma = sum / pixels; + picture.luma_left = left / (pixels / 2); + picture.luma_right = (sum - left) / (pixels / 2); out.push_back(picture); } }; @@ -348,4 +359,39 @@ inline std::vector decode_annex_b(const std::vector& stream) return out; } +/// Runs SensingCamera::update() on its own thread, as the plugin's main loop would. +class UpdateLoop +{ +public: + explicit UpdateLoop(plugins::sensing::SensingCamera& camera) + : m_thread( + [this, &camera] + { + while (!m_stop.load()) + { + camera.update(); + ++m_ticks; + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + }) + { + } + + ~UpdateLoop() + { + m_stop.store(true); + m_thread.join(); + } + + uint64_t ticks() const + { + return m_ticks.load(); + } + +private: + std::atomic m_stop{ false }; + std::atomic m_ticks{ 0 }; + std::thread m_thread; +}; + } // namespace sensing_test diff --git a/tests/cpp/plugins/sensing_mock/test_cuda_ipc_consumer.cpp b/tests/cpp/plugins/sensing_mock/test_cuda_ipc_consumer.cpp index 95cf117abc..f6a8411ba2 100644 --- a/tests/cpp/plugins/sensing_mock/test_cuda_ipc_consumer.cpp +++ b/tests/cpp/plugins/sensing_mock/test_cuda_ipc_consumer.cpp @@ -157,41 +157,6 @@ class IpcConsumer CUdeviceptr m_ptr = 0; }; -/// Runs SensingCamera::update() on its own thread, as the plugin's main loop would. -class UpdateLoop -{ -public: - explicit UpdateLoop(SensingCamera& camera) - : m_thread( - [this, &camera] - { - while (!m_stop.load()) - { - camera.update(); - ++m_ticks; - std::this_thread::sleep_for(1ms); - } - }) - { - } - - ~UpdateLoop() - { - m_stop.store(true); - m_thread.join(); - } - - uint64_t ticks() const - { - return m_ticks.load(); - } - -private: - std::atomic m_stop{ false }; - std::atomic m_ticks{ 0 }; - std::thread m_thread; -}; - SensingConfig make_config() { SensingConfig config; diff --git a/tests/cpp/plugins/sensing_mock/test_livekit_cli.cpp b/tests/cpp/plugins/sensing_mock/test_livekit_cli.cpp new file mode 100644 index 0000000000..f86c6b282d --- /dev/null +++ b/tests/cpp/plugins/sensing_mock/test_livekit_cli.cpp @@ -0,0 +1,145 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// The plugin binary's LiveKit options, run as an operator would run it. None of these needs a server: +// each one must be refused before any connection is tried. + +#include "mock_test_support.hpp" + +#include +#include + +#include +#include +#include + +using namespace sensing_test; + +namespace +{ + +struct Run +{ + int exit_code = -1; + std::string output; // stdout and stderr together +}; + +Run run_plugin(const std::string& arguments) +{ + const std::string command = std::string("'") + SENSING_PLUGIN_BINARY + "' " + arguments + " 2>&1"; + Run run; + if (FILE* pipe = ::popen(command.c_str(), "r")) + { + std::array buffer{}; + while (size_t n = std::fread(buffer.data(), 1, buffer.size(), pipe)) + run.output.append(buffer.data(), n); + const int status = ::pclose(pipe); + run.exit_code = WIFEXITED(status) ? WEXITSTATUS(status) : -1; + } + return run; +} + +void write_text(const std::string& path, const std::string& text) +{ + std::ofstream(path) << text; +} + +bool mentions(const Run& run, const std::string& text) +{ + return run.output.find(text) != std::string::npos; +} + +// Settings that are fine, so a test can break exactly one thing. +std::string good(const TempFile& token) +{ + return "--livekit-url=ws://127.0.0.1:9 --livekit-token-file=" + token.path() + + " --add-stream=sensor=0,livekit=left --add-stream=sensor=1,livekit=right"; +} + +} // namespace + +TEST_CASE("the help describes the LiveKit options", "[unit][sensing][livekit]") +{ + const Run run = run_plugin("--help"); + CHECK(run.exit_code == 0); + for (const char* text : + { "livekit=", "--livekit-url", "--livekit-token-file", "--livekit-fps", "--livekit-scale-div" }) + CHECK(mentions(run, text)); +} + +TEST_CASE("livekit streams need a URL and a token file", "[unit][sensing][livekit]") +{ + const Run run = run_plugin("--add-stream=sensor=0,livekit=left"); + CHECK(run.exit_code == 1); + CHECK(mentions(run, "--livekit-url and --livekit-token-file")); +} + +TEST_CASE("LiveKit options without a livekit stream are refused", "[unit][sensing][livekit]") +{ + TempFile token(".token"); + write_text(token.path(), "x"); + const Run run = run_plugin("--livekit-url=ws://127.0.0.1:9 --livekit-token-file=" + token.path() + + " --add-stream=sensor=0,ipc=/tmp/nope.sock"); + CHECK(run.exit_code == 1); + CHECK(mentions(run, "no stream has livekit=")); +} + +TEST_CASE("track names must be unique", "[unit][sensing][livekit]") +{ + TempFile token(".token"); + write_text(token.path(), "x"); + const Run run = run_plugin("--livekit-url=ws://127.0.0.1:9 --livekit-token-file=" + token.path() + + " --add-stream=sensor=0,livekit=same --add-stream=sensor=1,livekit=same"); + CHECK(run.exit_code == 1); + CHECK(mentions(run, "used twice")); +} + +TEST_CASE("a missing or empty token file is refused", "[unit][sensing][livekit]") +{ + const Run missing = run_plugin( + "--livekit-url=ws://127.0.0.1:9 --livekit-token-file=/nonexistent/token --add-stream=sensor=0,livekit=left"); + CHECK(missing.exit_code == 1); + CHECK(mentions(missing, "cannot read --livekit-token-file")); + + TempFile empty(".token"); + write_text(empty.path(), "\n"); + const Run blank = run_plugin(good(empty)); + CHECK(blank.exit_code == 1); + CHECK(mentions(blank, "is empty")); +} + +TEST_CASE("the publication rate must be positive", "[unit][sensing][livekit]") +{ + TempFile token(".token"); + write_text(token.path(), "x"); + const Run run = run_plugin(good(token) + " --livekit-fps=0"); + CHECK(run.exit_code == 1); + CHECK(mentions(run, "--livekit-fps must be at least 1")); +} + +TEST_CASE("a divisor that does not fit the sensor is refused before connecting", "[unit][gpu][sensing][livekit]") +{ + TempFile token(".token"); + write_text(token.path(), "x"); + // The vendor config's 2560x1984: 3 does not divide it, and 64 leaves an odd height. + const Run three = run_plugin(good(token) + " --livekit-scale-div=3"); + CHECK(three.exit_code == 1); + CHECK(mentions(three, "exactly")); + const Run odd = run_plugin(good(token) + " --livekit-scale-div=64"); + CHECK(odd.exit_code == 1); + CHECK(mentions(odd, "even")); +} + +TEST_CASE("the join token is never printed", "[unit][gpu][sensing][livekit]") +{ + TempFile token(".token"); + const std::string secret = "eyJ-THIS-IS-A-SECRET-JOIN-TOKEN-0123456789"; + write_text(token.path(), secret + "\n"); + + // Nothing listens on the URL, so this fails to connect: the error path must not echo the token. + const Run run = run_plugin(good(token) + " --livekit-fps=15"); + CHECK(run.exit_code == 1); + CHECK(mentions(run, "cannot join")); + CHECK_FALSE(mentions(run, secret)); + CHECK_FALSE(mentions(run, "THIS-IS-A-SECRET")); +} diff --git a/tests/cpp/plugins/sensing_mock/test_livekit_output.cpp b/tests/cpp/plugins/sensing_mock/test_livekit_output.cpp new file mode 100644 index 0000000000..220c4d9517 --- /dev/null +++ b/tests/cpp/plugins/sensing_mock/test_livekit_output.cpp @@ -0,0 +1,557 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// LiveKitVideoOutput against a fake track, with no LiveKit server: the hardware-encoded units it hands +// over, pacing, the bounded staging, ownership of the borrowed camera frame, the reduction, the +// receiving side's keyframe requests, refused frames, the end-of-stream drain and failure handling. + +#include "mock_test_support.hpp" + +#include +#include +#include + +#include +#include +#include +#include + +using namespace plugins::sensing; +using namespace sensing_test; +using namespace std::chrono_literals; + +namespace +{ + +struct Received +{ + EncodedVideoFrame info; ///< `data` is not valid after submit() returns; see `bytes`. + std::vector bytes; +}; + +// A track that records what it is given. It can hold a submission open, make it fail, refuse frames +// as the SDK can, and hand back the receiving side's requests. +class FakeSink : public VideoTrackSink +{ +public: + FakeSink(uint32_t width, uint32_t height) : m_width(width), m_height(height) + { + } + + uint32_t width() const override + { + return m_width; + } + uint32_t height() const override + { + return m_height; + } + + bool submit(const EncodedVideoFrame& frame) override + { + std::unique_lock lock(m_mutex); + ++m_entered; + m_changed.notify_all(); + m_changed.wait(lock, [&] { return !m_gated; }); + if (m_fail) + throw std::runtime_error("injected sink failure"); + if (m_refuse > 0) + { + --m_refuse; + return false; + } + Received got; + got.info = frame; + got.bytes.assign(frame.data, frame.data + frame.size); + got.info.data = nullptr; + m_received.push_back(std::move(got)); + m_changed.notify_all(); + return true; + } + + EncoderFeedback take_feedback() override + { + std::lock_guard lock(m_mutex); + EncoderFeedback out = m_feedback; + m_feedback = {}; + return out; + } + + std::string describe() override + { + return "fake"; + } + + void gate(bool closed) + { + std::lock_guard lock(m_mutex); + m_gated = closed; + m_changed.notify_all(); + } + + void fail() + { + std::lock_guard lock(m_mutex); + m_fail = true; + } + + /// The next `count` frames are refused, as the SDK may. + void refuse(int count) + { + std::lock_guard lock(m_mutex); + m_refuse = count; + } + + void request_keyframe() + { + std::lock_guard lock(m_mutex); + m_feedback.keyframe_requested = true; + } + + void request_bitrate(uint32_t bps) + { + std::lock_guard lock(m_mutex); + m_feedback.target_bitrate_bps = bps; + } + + bool wait_entered(size_t count) + { + std::unique_lock lock(m_mutex); + return m_changed.wait_for(lock, 5s, [&] { return m_entered >= count; }); + } + + bool wait_received(size_t count) + { + std::unique_lock lock(m_mutex); + return m_changed.wait_for(lock, 5s, [&] { return m_received.size() >= count; }); + } + + std::vector received() + { + std::lock_guard lock(m_mutex); + return m_received; + } + + /// Every received unit, in order, as one Annex-B stream. + std::vector stream() + { + std::vector out; + for (const auto& unit : received()) + out.insert(out.end(), unit.bytes.begin(), unit.bytes.end()); + return out; + } + +private: + uint32_t m_width; + uint32_t m_height; + std::mutex m_mutex; + std::condition_variable m_changed; + bool m_gated = false; + bool m_fail = false; + int m_refuse = 0; + EncoderFeedback m_feedback; + size_t m_entered = 0; + std::vector m_received; +}; + +// An RGBA frame in device memory that a test rewrites, standing in for the camera's mailbox slot. +class DeviceFrame +{ +public: + DeviceFrame(uint32_t width, uint32_t height) : m_width(width), m_height(height) + { + cudaMallocPitch(reinterpret_cast(&m_ptr), &m_pitch, static_cast(width) * 4, height); + } + ~DeviceFrame() + { + cudaFree(m_ptr); + } + DeviceFrame(const DeviceFrame&) = delete; + DeviceFrame& operator=(const DeviceFrame&) = delete; + + // Left half `left`, right half `right`. + void fill(Rgb left, Rgb right) + { + std::vector host(static_cast(m_width) * m_height * 4); + for (uint32_t y = 0; y < m_height; ++y) + { + for (uint32_t x = 0; x < m_width; ++x) + { + const Rgb c = x < m_width / 2 ? left : right; + uint8_t* px = &host[(static_cast(y) * m_width + x) * 4]; + px[0] = static_cast(c.r); + px[1] = static_cast(c.g); + px[2] = static_cast(c.b); + px[3] = 255; + } + } + cudaMemcpy2D(m_ptr, m_pitch, host.data(), static_cast(m_width) * 4, static_cast(m_width) * 4, + m_height, cudaMemcpyHostToDevice); + } + void fill(Rgb colour) + { + fill(colour, colour); + } + + FrameView view(uint64_t sequence, uint64_t timestamp_ns, uint64_t tsc_ns = 0) const + { + FrameView v; + v.ptr = reinterpret_cast(m_ptr); + v.pitch = m_pitch; + v.width = m_width; + v.height = m_height; + v.timestamp_ns = timestamp_ns; + v.capture_tsc_ns = tsc_ns; + v.sequence = sequence; + return v; + } + +private: + uint32_t m_width; + uint32_t m_height; + uint8_t* m_ptr = nullptr; + size_t m_pitch = 0; +}; + +constexpr Rgb kRed{ 255, 0, 0 }; +constexpr Rgb kBlue{ 0, 0, 255 }; +constexpr Rgb kGreen{ 0, 255, 0 }; + +// Mean luma the encoder should produce for a solid colour: BT.601 limited range. +double luma_601(Rgb c) +{ + return 16 + 0.257 * c.r + 0.504 * c.g + 0.098 * c.b; +} + +LiveKitOutputConfig config(uint32_t width, uint32_t height, uint32_t fps = 30, uint32_t div = 1) +{ + LiveKitOutputConfig c; + c.source_width = width; + c.source_height = height; + c.fps = fps; + c.scale_div = div; + c.bitrate_bps = 4'000'000; + c.peak_bitrate_bps = 6'000'000; + return c; +} + +} // namespace + +TEST_CASE("the reduction must divide the sensor exactly and leave even sizes", "[unit][gpu][sensing][livekit]") +{ + CHECK_NOTHROW(LiveKitVideoOutput::validate_geometry(2560, 1984, 1)); + CHECK_NOTHROW(LiveKitVideoOutput::validate_geometry(2560, 1984, 2)); // 1280x992 + CHECK_NOTHROW(LiveKitVideoOutput::validate_geometry(2560, 1984, 4)); // 640x496 + CHECK_THROWS_WITH( + LiveKitVideoOutput::validate_geometry(2560, 1984, 3), Catch::Matchers::ContainsSubstring("exactly")); + CHECK_THROWS_WITH( + LiveKitVideoOutput::validate_geometry(2560, 1984, 0), Catch::Matchers::ContainsSubstring("at least 1")); + // 1984 / 64 = 31, which is odd: NV12 cannot hold it, and it is refused rather than rounded. + CHECK_THROWS_WITH(LiveKitVideoOutput::validate_geometry(2560, 1984, 64), Catch::Matchers::ContainsSubstring("even")); + CHECK_THROWS_AS(LiveKitVideoOutput::validate_geometry(0, 0, 1), std::runtime_error); +} + +TEST_CASE("a track whose size is not the output size is refused", "[unit][gpu][sensing][livekit]") +{ + CHECK_THROWS_AS( + LiveKitVideoOutput(std::make_unique(320, 240), config(320, 240, 30, 2)), std::runtime_error); + CHECK_THROWS_AS(LiveKitVideoOutput(std::make_unique(320, 240), config(320, 240, 0)), std::runtime_error); +} + +TEST_CASE("an encoder that cannot start fails construction, with no software fallback", "[unit][gpu][sensing][livekit]") +{ + // Far beyond what H.264 hardware encodes. + CHECK_THROWS_WITH(LiveKitVideoOutput(std::make_unique(16384, 16384), config(16384, 16384)), + Catch::Matchers::ContainsSubstring("h264_nvenc")); +} + +TEST_CASE("frames come out hardware-encoded as decodable units with the camera's identifiers", + "[unit][gpu][sensing][livekit]") +{ + auto sink = std::make_unique(320, 240); + FakeSink& track = *sink; + LiveKitVideoOutput output(std::move(sink), config(320, 240)); + + DeviceFrame frame(320, 240); + frame.fill(kRed, kBlue); + output.submit(frame.view(7, 5'000'000'000ull, 123456789)); + output.flush(); + REQUIRE(track.wait_received(1)); + + const auto got = track.received(); + REQUIRE(got.size() == 1); + CHECK(got[0].info.keyframe); + CHECK(got[0].info.frame_id == 7); + CHECK(got[0].info.timestamp_us == 5'000'000); + CHECK(got[0].info.capture_tsc_ns == 123456789); + + // SPS and PPS travel with the IDR, so a receiver can start on this unit alone. + std::vector types; + for (const auto& nal : split_annex_b(got[0].bytes)) + types.push_back(nal.type); + REQUIRE(types.size() >= 3); + CHECK(types[0] == kNalSps); + CHECK(types[1] == kNalPps); + + const auto pictures = decode_annex_b(got[0].bytes); + REQUIRE(pictures.size() == 1); + CHECK(pictures[0].width == 320); + CHECK(pictures[0].height == 240); + CHECK(std::abs(pictures[0].luma_left - luma_601(kRed)) < 12); + CHECK(std::abs(pictures[0].luma_right - luma_601(kBlue)) < 12); + CHECK(output.describe().find("pre-encoded passthrough") != std::string::npos); +} + +TEST_CASE("a reduced output is encoded at the reduced size and keeps the picture's layout", + "[unit][gpu][sensing][livekit]") +{ + auto sink = std::make_unique(320, 240); + FakeSink& track = *sink; + LiveKitVideoOutput output(std::move(sink), config(640, 480, 30, 2)); + + DeviceFrame frame(640, 480); + frame.fill(kRed, kBlue); + output.submit(frame.view(1, 1'000'000'000)); + output.flush(); + REQUIRE(track.wait_received(1)); + + const auto pictures = decode_annex_b(track.stream()); + REQUIRE(pictures.size() == 1); + CHECK(pictures[0].width == 320); + CHECK(pictures[0].height == 240); + CHECK(std::abs(pictures[0].luma_left - luma_601(kRed)) < 12); + CHECK(std::abs(pictures[0].luma_right - luma_601(kBlue)) < 12); +} + +TEST_CASE("the publication rate is held on the capture clock, and every encoded frame is drained", + "[unit][gpu][sensing][livekit]") +{ + auto sink = std::make_unique(320, 240); + FakeSink& track = *sink; + LiveKitVideoOutput output(std::move(sink), config(320, 240, 30)); + + DeviceFrame frame(320, 240); + frame.fill(kRed); + // Sixty captures over one second of capture time, offered to a 30 fps track. + for (uint64_t i = 0; i < 60; ++i) + { + output.submit(frame.view(i + 1, 1'000'000'000ull + i * 16'666'667ull)); + std::this_thread::sleep_for(3ms); // let the worker keep up, so nothing is replaced + } + output.flush(); + + const auto stats = output.stats(); + CHECK(stats.offered == 60); + CHECK(stats.snapshots + stats.paced_out == 60); + CHECK(stats.snapshots >= 29); + CHECK(stats.snapshots <= 31); + CHECK(stats.paced_out >= 29); + // Each frame the encoder took is published after the drain, one unit per frame. + CHECK(stats.published == stats.snapshots - stats.replaced - stats.encoder_busy); + CHECK(track.received().size() == stats.published); +} + +TEST_CASE("the end-of-stream drain publishes every unit with its own stamp", "[unit][gpu][sensing][livekit]") +{ + auto sink = std::make_unique(320, 240); + FakeSink& track = *sink; + LiveKitVideoOutput output(std::move(sink), config(320, 240, 1000)); + + DeviceFrame frame(320, 240); + frame.fill(kGreen); + for (uint64_t i = 1; i <= 5; ++i) + { + output.submit(frame.view(i, 2'000'000'000ull + i * 33'333'000ull, 1000 + i)); + std::this_thread::sleep_for(15ms); + } + output.flush(); + + const auto got = track.received(); + REQUIRE(got.size() == 5); + for (size_t i = 0; i < got.size(); ++i) + { + CHECK(got[i].info.frame_id == i + 1); + CHECK(got[i].info.timestamp_us == (2'000'000'000ull + (i + 1) * 33'333'000ull) / 1000); + CHECK(got[i].info.capture_tsc_ns == 1001 + i); + CHECK(got[i].info.keyframe == (i == 0)); + } + CHECK(decode_annex_b(track.stream()).size() == 5); +} + +TEST_CASE("a keyframe request from the receiving side makes the next frame an IDR", "[unit][gpu][sensing][livekit]") +{ + auto sink = std::make_unique(320, 240); + FakeSink& track = *sink; + LiveKitVideoOutput output(std::move(sink), config(320, 240, 1000)); + + DeviceFrame frame(320, 240); + frame.fill(kRed); + for (uint64_t i = 1; i <= 3; ++i) + { + output.submit(frame.view(i, 1'000'000'000ull + i * 33'000'000ull)); + std::this_thread::sleep_for(40ms); + } + track.request_keyframe(); + for (uint64_t i = 4; i <= 6; ++i) + { + output.submit(frame.view(i, 1'000'000'000ull + i * 33'000'000ull)); + std::this_thread::sleep_for(40ms); + } + output.flush(); + + const auto got = track.received(); + REQUIRE(got.size() == 6); + std::vector keyframes; + for (const auto& unit : got) + keyframes.push_back(unit.info.keyframe); + CHECK(keyframes == std::vector{ true, false, false, true, false, false }); + CHECK(output.stats().keyframes_requested == 1); + + // The re-sent IDR carries parameter sets again, so a late joiner can start on it. + const auto nals = split_annex_b(got[3].bytes); + REQUIRE(nals.size() >= 3); + CHECK(nals[0].type == kNalSps); + CHECK(nals[1].type == kNalPps); + + // And the whole stream, with the second IDR, decodes. + CHECK(decode_annex_b(track.stream()).size() == 6); +} + +TEST_CASE("a frame the SDK refuses is followed by a keyframe, and nothing between", "[unit][gpu][sensing][livekit]") +{ + auto sink = std::make_unique(320, 240); + FakeSink& track = *sink; + LiveKitVideoOutput output(std::move(sink), config(320, 240, 1000)); + + DeviceFrame frame(320, 240); + frame.fill(kRed); + output.submit(frame.view(1, 1'000'000'000ull)); + std::this_thread::sleep_for(60ms); + REQUIRE(track.wait_received(1)); + + track.refuse(1); + for (uint64_t i = 2; i <= 6; ++i) + { + output.submit(frame.view(i, 1'000'000'000ull + i * 33'000'000ull)); + std::this_thread::sleep_for(60ms); + } + output.flush(); + + // Frame 2 was refused; frame 3 was encoded as the IDR that repairs the stream. + const auto got = track.received(); + std::vector ids; + for (const auto& unit : got) + ids.push_back(unit.info.frame_id); + CHECK(ids == std::vector{ 1, 3, 4, 5, 6 }); + CHECK(got[1].info.keyframe); + CHECK(output.stats().rejected == 1); + CHECK(decode_annex_b(track.stream()).size() == 5); +} + +TEST_CASE("a bitrate target the backend cannot apply does not disturb the stream", "[unit][gpu][sensing][livekit]") +{ + auto sink = std::make_unique(320, 240); + FakeSink& track = *sink; + LiveKitVideoOutput output(std::move(sink), config(320, 240, 1000)); + + DeviceFrame frame(320, 240); + frame.fill(kRed); + track.request_bitrate(500'000'000); // above any ceiling: clamped, and the mock encoder keeps its rate + output.submit(frame.view(1, 1'000'000'000ull)); + output.flush(); + CHECK_FALSE(output.failed()); + CHECK(track.received().size() == 1); +} + +TEST_CASE("a stalled track cannot stall capture, and only the newest waiting frame is kept", + "[unit][gpu][sensing][livekit]") +{ + auto sink = std::make_unique(320, 240); + FakeSink& track = *sink; + track.gate(true); + LiveKitVideoOutput output(std::move(sink), config(320, 240, 1000)); + + DeviceFrame frame(320, 240); + frame.fill(kRed); + output.submit(frame.view(1, 1'000'000'000ull)); + REQUIRE(track.wait_entered(1)); // the worker holds frame 1 inside the stalled track + + // Fifty more captures arrive while it is stuck: the polling thread must not wait for any of them. + const auto start = std::chrono::steady_clock::now(); + for (uint64_t i = 2; i <= 51; ++i) + output.submit(frame.view(i, 1'000'000'000ull + i * 2'000'000ull)); + CHECK(std::chrono::steady_clock::now() - start < 2s); + + const auto stats = output.stats(); + CHECK(stats.replaced >= 48); // one slot waits, and each newer capture takes it over + + track.gate(false); + REQUIRE(track.wait_received(2)); + std::this_thread::sleep_for(200ms); + const auto got = track.received(); + REQUIRE(got.size() == 2); // frame 1, then only the newest + CHECK(got[0].info.frame_id == 1); + CHECK(got[1].info.frame_id == 51); +} + +TEST_CASE("the camera frame may be rewritten as soon as submit returns", "[unit][gpu][sensing][livekit]") +{ + auto sink = std::make_unique(320, 240); + FakeSink& track = *sink; + track.gate(true); + LiveKitVideoOutput output(std::move(sink), config(320, 240, 1000)); + + DeviceFrame source(320, 240); + source.fill(kRed); + output.submit(source.view(1, 1'000'000'000ull)); + REQUIRE(track.wait_entered(1)); // frame 1 is encoded and held in the track + + // Frame 2 waits behind it. The camera then rewrites its slot, as it does after the next poll. + source.fill(kBlue); + output.submit(source.view(2, 1'010'000'000ull)); + source.fill(kGreen); + + track.gate(false); + output.flush(); + REQUIRE(track.wait_received(2)); + const auto pictures = decode_annex_b(track.stream()); + REQUIRE(pictures.size() == 2); + CHECK(std::abs(pictures[0].luma - luma_601(kRed)) < 12); + // The snapshot was complete before submit returned, so the later overwrite cannot reach it. + CHECK(std::abs(pictures[1].luma - luma_601(kBlue)) < 12); +} + +TEST_CASE("a failing track fails the output and later captures return at once", "[unit][gpu][sensing][livekit]") +{ + auto sink = std::make_unique(320, 240); + FakeSink& track = *sink; + track.fail(); + LiveKitVideoOutput output(std::move(sink), config(320, 240, 1000)); + + DeviceFrame frame(320, 240); + frame.fill(kRed); + output.submit(frame.view(1, 1'000'000'000ull)); + REQUIRE(wait_for([&] { return output.failed(); })); + CHECK(output.failure().find("injected sink failure") != std::string::npos); + + const auto start = std::chrono::steady_clock::now(); + for (uint64_t i = 2; i < 100; ++i) + output.submit(frame.view(i, 1'000'000'000ull + i * 2'000'000ull)); + CHECK(std::chrono::steady_clock::now() - start < 1s); + CHECK(output.stats().published == 0); +} + +TEST_CASE("destroying an output while a frame waits drops it and returns", "[unit][gpu][sensing][livekit]") +{ + DeviceFrame frame(320, 240); + frame.fill(kRed); + + const auto start = std::chrono::steady_clock::now(); + { + LiveKitVideoOutput output(std::make_unique(320, 240), config(320, 240, 1000)); + for (uint64_t i = 1; i <= 20; ++i) + output.submit(frame.view(i, 1'000'000'000ull + i * 2'000'000ull)); + } + // The worker joined: nothing is left running against freed memory, and teardown is prompt. + CHECK(std::chrono::steady_clock::now() - start < 3s); +} diff --git a/tests/cpp/plugins/sensing_mock/test_livekit_stream.cpp b/tests/cpp/plugins/sensing_mock/test_livekit_stream.cpp new file mode 100644 index 0000000000..38a325dd86 --- /dev/null +++ b/tests/cpp/plugins/sensing_mock/test_livekit_stream.cpp @@ -0,0 +1,271 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// The SENSING plugin publishing to a real LiveKit server (a local livekit-server process) and a +// second participant decoding what arrives. Set LIVEKIT_SERVER to the server binary to run these. + +#include "livekit_test_support.hpp" +#include "mock_test_support.hpp" + +#include +#include +#include +#include + +#include +#include +#include +#include + +using namespace plugins::sensing; +using namespace sensing_test; +using namespace livekit_test; +using namespace std::chrono_literals; + +#define REQUIRE_SERVER() \ + do \ + { \ + if (!LocalServer::binary()) \ + SKIP("set LIVEKIT_SERVER to a livekit-server binary to run this test"); \ + } while (0) + +namespace +{ + +constexpr const char* kRoom = "sensing-room"; +constexpr uint64_t kMockTscBaseNs = 1'000'000'000'000'000ull; + +SensingConfig stream_config(const std::string& url, + uint32_t scale_div, + uint32_t fps = 30, + const std::string& secret = kApiSecret) +{ + SensingConfig config; + config.platform_config_json = SENSING_PLATFORM_CONFIG; + config.bitrate_bps = 8'000'000; + config.livekit.url = url; + config.livekit.token = make_token("sensing-publisher", kRoom, true, secret); + config.livekit.fps = fps; + config.livekit.scale_div = scale_div; + return config; +} + +std::vector both_sensors(const std::string& left = "sensing-left", + const std::string& right = "sensing-right") +{ + std::vector streams(2); + streams[0].sensor_id = 0; + streams[0].livekit_track = left; + streams[1].sensor_id = 1; + streams[1].livekit_track = right; + return streams; +} + +std::unique_ptr make_camera(const SensingConfig& config, const std::vector& streams) +{ + return std::make_unique(config, streams, std::make_unique(streams)); +} + +// One of black, red, blue or white, allowing for what H.264 does to a flat colour. +bool is_cycle_colour_lossy(const Rgb& c) +{ + return near(c, 0, 0, 0, 50) || near(c, 255, 0, 0, 50) || near(c, 0, 0, 255, 50) || near(c, 255, 255, 255, 50); +} + +} // namespace + +TEST_CASE("two sensors reach a real server as named tracks and decode correctly", "[unit][gpu][sensing][livekit][server]") +{ + REQUIRE_SERVER(); + LocalServer server; + FastMock fast; // 320x248 sensors playing a black, red, blue, white cycle + auto camera = make_camera(stream_config(server.url(), 2), both_sensors()); + UpdateLoop loop(*camera); + + Subscriber subscriber(server.url(), kRoom); + REQUIRE(subscriber.wait_for_frames({ "sensing-left", "sensing-right" }, 25, 30s)); + + CHECK(subscriber.track_names() == std::set{ "sensing-left", "sensing-right" }); + for (const char* name : { "sensing-left", "sensing-right" }) + { + INFO(name); + const TrackSeen seen = subscriber.seen(name); + // 320x248 reduced by 2. + CHECK(seen.width == 160); + CHECK(seen.height == 124); + REQUIRE(seen.last_rgba.size() == 160u * 124u * 4u); + Shot shot; + shot.width = 160; + shot.height = 124; + shot.rgba = seen.last_rgba; + CHECK(is_cycle_colour_lossy(shot.at(80, 62))); + + // The camera's own frame ids and capture clock survive to the viewer. + REQUIRE(seen.frame_ids.size() >= 20); + for (size_t i = 1; i < seen.frame_ids.size(); ++i) + CHECK(seen.frame_ids[i] > seen.frame_ids[i - 1]); + CHECK(seen.with_user_timestamp >= seen.frames - 2); + REQUIRE_FALSE(seen.tsc_by_frame.empty()); + for (const auto& [id, tsc] : seen.tsc_by_frame) + CHECK(tsc == kMockTscBaseNs + static_cast((id - 1) * (1e9 / 60.0))); + } +} + +TEST_CASE("the track carries hardware-encoded video passed through, with no software encoder", + "[unit][gpu][sensing][livekit][server]") +{ + REQUIRE_SERVER(); + LocalServer server; + FastMock fast; + std::vector streams(1); + streams[0].sensor_id = 0; + streams[0].livekit_track = "passthrough"; + auto camera = make_camera(stream_config(server.url(), 1), streams); + { + UpdateLoop loop(*camera); + Subscriber subscriber(server.url(), kRoom); + REQUIRE(subscriber.wait_for_frames({ "passthrough" }, 25, 30s)); + } + + // The first report starts the transport's stats request and the second collects it. + std::ostringstream report; + auto* previous = std::cout.rdbuf(report.rdbuf()); + camera->print_stats(); + std::this_thread::sleep_for(2s); + camera->print_stats(); + std::cout.rdbuf(previous); + + const std::string text = report.str(); + INFO(text); + CHECK(text.find("NVENC") != std::string::npos); + CHECK(text.find("pre-encoded passthrough") != std::string::npos); + CHECK(text.find("transport: encoder LiveKit pre-encoded passthrough") != std::string::npos); + CHECK(text.find("OpenH264") == std::string::npos); + CHECK(text.find("openh264") == std::string::npos); +} + +TEST_CASE("a track is published at the sensor's native geometry by default", "[unit][gpu][sensing][livekit][server]") +{ + REQUIRE_SERVER(); + LocalServer server; + ScopedEnv fps("SENSING_MOCK_FPS", "15"); // the bundled GIF at the real 2560x1984, at a rate any machine can encode + std::vector streams(1); + streams[0].sensor_id = 0; + streams[0].livekit_track = "native"; + auto camera = make_camera(stream_config(server.url(), 1, 15), streams); + UpdateLoop loop(*camera); + + Subscriber subscriber(server.url(), kRoom); + REQUIRE(subscriber.wait_for_frames({ "native" }, 5, 60s)); + const TrackSeen seen = subscriber.seen("native"); + CHECK(seen.width == 2560); + CHECK(seen.height == 1984); +} + +TEST_CASE("a viewer can join late, leave and join again", "[unit][gpu][sensing][livekit][server]") +{ + REQUIRE_SERVER(); + LocalServer server; + FastMock fast; + auto camera = make_camera(stream_config(server.url(), 2), both_sensors()); + UpdateLoop loop(*camera); + + { + Subscriber first(server.url(), kRoom); + REQUIRE(first.wait_for_frames({ "sensing-left", "sensing-right" }, 10, 30s)); + } + Subscriber second(server.url(), kRoom); + REQUIRE(second.wait_for_frames({ "sensing-left", "sensing-right" }, 10, 30s)); +} + +TEST_CASE("losing the server ends capture with a clear error", "[unit][gpu][sensing][livekit][server]") +{ + REQUIRE_SERVER(); + auto server = std::make_unique(); + FastMock fast; + auto camera = make_camera(stream_config(server->url(), 2), both_sensors()); + { + Subscriber subscriber(server->url(), kRoom); + // Poll as the plugin's main loop does, on this thread so a failure can be caught. + const auto warm = std::chrono::steady_clock::now() + 20s; + bool seen = false; + while (!seen && std::chrono::steady_clock::now() < warm) + { + camera->update(); + seen = subscriber.wait_for_frames({ "sensing-left" }, 5, 5ms); + } + REQUIRE(seen); + } + + server->stop(); + std::string error; + const auto start = std::chrono::steady_clock::now(); + while (error.empty() && std::chrono::steady_clock::now() - start < 90s) + { + try + { + camera->update(); + } + catch (const std::runtime_error& e) + { + error = e.what(); + } + std::this_thread::sleep_for(5ms); + } + INFO("took " << std::chrono::duration_cast(std::chrono::steady_clock::now() - start).count() + << " s"); + REQUIRE_FALSE(error.empty()); + CHECK(error.find("LiveKit") != std::string::npos); +} + +TEST_CASE("a token signed with the wrong secret is refused", "[unit][gpu][sensing][livekit][server]") +{ + REQUIRE_SERVER(); + LocalServer server; + FastMock fast; + CHECK_THROWS_WITH(make_camera(stream_config(server.url(), 2, 30, "not-the-secret"), both_sensors()), + Catch::Matchers::ContainsSubstring("cannot join")); +} + +TEST_CASE("a server that is not there is an error, not a hang", "[unit][gpu][sensing][livekit]") +{ + FastMock fast; + const auto start = std::chrono::steady_clock::now(); + CHECK_THROWS_WITH(make_camera(stream_config("ws://127.0.0.1:9", 2), both_sensors()), + Catch::Matchers::ContainsSubstring("cannot join")); + CHECK(std::chrono::steady_clock::now() - start < 60s); +} + +TEST_CASE("stopping the plugin unpublishes and leaves the room", "[unit][gpu][sensing][livekit][server]") +{ + REQUIRE_SERVER(); + LocalServer server; + FastMock fast; + Subscriber subscriber(server.url(), kRoom); + { + auto camera = make_camera(stream_config(server.url(), 2), both_sensors()); + UpdateLoop loop(*camera); + REQUIRE(subscriber.wait_for_frames({ "sensing-left", "sensing-right" }, 10, 30s)); + + // Destroy in the plugin's order: the loop stops, then the camera and its tracks and room. + } + const auto start = std::chrono::steady_clock::now(); + while (subscriber.participants_left() == 0 && std::chrono::steady_clock::now() - start < 10s) + std::this_thread::sleep_for(50ms); + CHECK(subscriber.participants_left() >= 1); +} + +TEST_CASE("bad LiveKit settings fail before any connection is made", "[unit][gpu][sensing][livekit]") +{ + FastMock fast; + // The URL points nowhere: these must be refused without trying it, so they return at once. The + // sensors here are 320x248: 3 does not divide them, and 8 leaves 40x31, which NV12 cannot hold. + const auto start = std::chrono::steady_clock::now(); + CHECK_THROWS_WITH(make_camera(stream_config("ws://127.0.0.1:9", 2), both_sensors("same", "same")), + Catch::Matchers::ContainsSubstring("used twice")); + CHECK_THROWS_WITH(make_camera(stream_config("ws://127.0.0.1:9", 3), both_sensors()), + Catch::Matchers::ContainsSubstring("exactly")); + CHECK_THROWS_WITH( + make_camera(stream_config("ws://127.0.0.1:9", 8), both_sensors()), Catch::Matchers::ContainsSubstring("even")); + CHECK(std::chrono::steady_clock::now() - start < 5s); +} diff --git a/tests/cpp/plugins/sensing_mock/test_mock_encoder.cpp b/tests/cpp/plugins/sensing_mock/test_mock_encoder.cpp index ac528986aa..d337ffba0f 100644 --- a/tests/cpp/plugins/sensing_mock/test_mock_encoder.cpp +++ b/tests/cpp/plugins/sensing_mock/test_mock_encoder.cpp @@ -7,6 +7,7 @@ #include #include +#include #include #include @@ -37,6 +38,16 @@ class Frame { cudaMemset2D(m_ptr, m_pitch, gray, kWidth * 4, kHeight); } + /// Different random pixels for each seed: content no encoder can predict, so the bitrate target decides + /// how large a unit is. + void randomize(uint32_t seed) + { + std::mt19937 rng(seed); + std::vector host(static_cast(kWidth) * kHeight * 4); + for (auto& byte : host) + byte = static_cast(rng()); + cudaMemcpy2D(m_ptr, m_pitch, host.data(), kWidth * 4, kWidth * 4, kHeight, cudaMemcpyHostToDevice); + } uintptr_t ptr() const { return reinterpret_cast(m_ptr); @@ -120,7 +131,10 @@ TEST_CASE("the output decodes, in order, at the configured size", "[unit][gpu][s for (int i = 0; i < 4; ++i) { frame.set(40 + 50 * i); - REQUIRE(wait_for([&] { return encoder.submit(frame.ptr(), frame.pitch(), 1'000'000'000 + i * 33'333'000); })); + REQUIRE(wait_for( + [&] { + return encoder.submit(frame.ptr(), frame.pitch(), int64_t{ 1'000'000'000 } + i * int64_t{ 33'333'000 }); + })); // The next frame is only written once the copy of this one is done: submit() has returned. } const auto tail = encoder.end_of_stream(); @@ -187,3 +201,77 @@ TEST_CASE("end of stream returns the tail as one unit carrying the last stamp", CHECK(encoder.poll().empty()); CHECK(encoder.end_of_stream().empty()); } + +TEST_CASE("a lower bitrate target changes the encoded output and the stream still decodes", "[unit][gpu][sensing]") +{ + Frame frame; + auto config = make_config(); + config.bitrate_bps = 4'000'000; + config.peak_bitrate_bps = 6'000'000; + JetsonEncoder encoder(config); + + constexpr int kBefore = 20; + constexpr int kAfter = 40; + std::vector units; + for (int i = 0; i < kBefore + kAfter; ++i) + { + if (i == kBefore) + REQUIRE(encoder.set_bitrate(400'000)); + frame.randomize(1000 + i); + REQUIRE(wait_for( + [&] { + return encoder.submit(frame.ptr(), frame.pitch(), int64_t{ 1'000'000'000 } + i * int64_t{ 33'333'000 }); + })); + std::this_thread::sleep_for(5ms); + for (auto unit = encoder.poll(); !unit.empty(); unit = encoder.poll()) + units.push_back(std::move(unit)); + } + for (auto& unit : encoder.drain()) + units.push_back(std::move(unit)); + REQUIRE(units.size() == kBefore + kAfter); + + auto mean_size = [&](int from, int to) + { + double total = 0; + for (int i = from; i < to; ++i) + total += static_cast(units[i].data.size()); + return total / (to - from); + }; + const double before = mean_size(3, kBefore); + const double after = mean_size(kBefore + 20, kBefore + kAfter); // after the rate control has settled + INFO("before " << before << " bytes, after " << after << " bytes"); + CHECK(after < before * 0.4); + + std::vector stream; + for (const auto& unit : units) + stream.insert(stream.end(), unit.data.begin(), unit.data.end()); + CHECK(decode_annex_b(stream).size() == units.size()); +} + +TEST_CASE("a keyframe request makes the next frame an IDR with parameter sets", "[unit][gpu][sensing]") +{ + Frame frame; + JetsonEncoder encoder(make_config()); // GOP of 150: no IDR but the first, unless asked for + std::vector units; + for (int i = 0; i < 6; ++i) + { + if (i == 4) + encoder.request_keyframe(); + REQUIRE(wait_for( + [&] { + return encoder.submit(frame.ptr(), frame.pitch(), int64_t{ 1'000'000'000 } + i * int64_t{ 33'333'000 }); + })); + std::this_thread::sleep_for(20ms); + } + for (auto& unit : encoder.drain()) + units.push_back(std::move(unit)); + REQUIRE(units.size() == 6); + + std::vector keyframes; + for (const auto& unit : units) + keyframes.push_back(unit.keyframe); + CHECK(keyframes == std::vector{ true, false, false, false, true, false }); + const auto nals = split_annex_b(units[4].data); + REQUIRE(nals.size() >= 3); + CHECK(nals[0].type == kNalSps); +}