diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 95680ca540..1ea2f91e2b 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -25,8 +25,17 @@ if(BUILD_VIZ) add_subdirectory(viz) endif() +# The LiveKit SDK and publisher library: the PC relay publishes to LiveKit; the camera producer still does until it sends +# direct to the relay. +if(BUILD_VIDEO_STAGE OR BUILD_PLUGIN_SENSING) + add_subdirectory(livekit_publisher) +endif() + if(BUILD_VIDEO_STAGE) add_subdirectory(video_stage) +elseif(BUILD_PLUGIN_SENSING) + # The LiveKit library uses the record codec and clock-exchange helpers. + add_subdirectory(video_stage/wire) endif() if(BUILD_PLUGINS) diff --git a/src/livekit_publisher/CMakeLists.txt b/src/livekit_publisher/CMakeLists.txt new file mode 100644 index 0000000000..2a3a4e3213 --- /dev/null +++ b/src/livekit_publisher/CMakeLists.txt @@ -0,0 +1,110 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# ============================================================================== +# LiveKit +# ============================================================================== +# Each selected sensor is published as a LiveKit camera track. 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. +set(_livekit_local ON) +if(NOT LIVEKIT_SDK_DIR) + set(_livekit_local OFF) + # The prebuilt links against the glibc of the Ubuntu release it was built on: the 24.04 build does not load on + # JetPack 6 (Ubuntu 22.04). Pick the release matching the host; LIVEKIT_SDK_UBUNTU overrides the detection. + if(NOT LIVEKIT_SDK_UBUNTU) + set(LIVEKIT_SDK_UBUNTU "24.04") + if(EXISTS /etc/os-release) + file(STRINGS /etc/os-release _os_version_id REGEX "^VERSION_ID=") + if(_os_version_id MATCHES "\"?22\\.04\"?$") + set(LIVEKIT_SDK_UBUNTU "22.04") + endif() + endif() + endif() + if(CMAKE_SYSTEM_PROCESSOR MATCHES "aarch64|arm64") + set(_livekit_arch "arm64") + set(_livekit_hash_24.04 "e3a97f71943c0bf9fa4ebd70927ca2906b31dcd6c14b42312b1225b4b868e4ad") + set(_livekit_hash_22.04 "7a698bbd8f61f019744fd99b3a5d2b6c6118787fd1f81140c9a6fc8732fc87e0") + elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "x86_64|AMD64") + set(_livekit_arch "x64") + set(_livekit_hash_24.04 "5e0feb2e519c6ff131dd2d1c45149115057ae4e21a324fdb249e1789d5614e7b") + set(_livekit_hash_22.04 "9e568ee6b2bb3c7b0d07faba694930767096a14d0e09422ba01f34077ffb03cb") + else() + message(FATAL_ERROR "No prebuilt LiveKit SDK for ${CMAKE_SYSTEM_PROCESSOR}; set LIVEKIT_SDK_DIR.") + endif() + if(NOT DEFINED _livekit_hash_${LIVEKIT_SDK_UBUNTU}) + message(FATAL_ERROR "No prebuilt LiveKit SDK for Ubuntu ${LIVEKIT_SDK_UBUNTU}; set LIVEKIT_SDK_DIR.") + endif() + set(_livekit_hash "${_livekit_hash_${LIVEKIT_SDK_UBUNTU}}") + include(FetchContent) + FetchContent_Declare( + livekit_sdk + URL https://github.com/livekit/client-sdk-cpp/releases/download/v1.12.4/livekit-sdk-ubuntu-${LIVEKIT_SDK_UBUNTU}-${_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. The pinned download supplies the SDK's own; a local SDK uses +# LICENSE/NOTICE from its directory, or LIVEKIT_SDK_LICENSE_FILE / LIVEKIT_SDK_NOTICE_FILE, and needs no network. +foreach(_notice LICENSE NOTICE) + set(_override "${LIVEKIT_SDK_${_notice}_FILE}") + if(_override) + set(_found "${_override}") + elseif(_livekit_local) + set(_found "") + foreach(_candidate "${LIVEKIT_SDK_DIR}/${_notice}" "${LIVEKIT_SDK_DIR}/../${_notice}") + if(NOT _found AND EXISTS "${_candidate}") + set(_found "${_candidate}") + endif() + endforeach() + if(NOT _found) + message(FATAL_ERROR + "LIVEKIT_SDK_DIR is set but ${_notice} was not found in it. Put the SDK's ${_notice} file there " + "or pass -DLIVEKIT_SDK_${_notice}_FILE=/path/to/${_notice}.") + endif() + else() + 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}) + # FetchContent publishes _SOURCE_DIR. + string(TOLOWER "${_notice}" _notice_lower) + set(_found "${livekit_sdk_${_notice_lower}_SOURCE_DIR}/${_notice}") + endif() + set(SENSING_LIVEKIT_${_notice}_FILE "${_found}" CACHE INTERNAL "") +endforeach() +find_package(LiveKit CONFIG REQUIRED PATHS "${LIVEKIT_SDK_DIR}" NO_DEFAULT_PATH) +message(STATUS "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 "") + +# The LiveKit SDK behind a small interface: join a room, publish a video track, submit H.264 access units. +# 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 + video_stage::wire + 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/livekit_publisher/inc/sensing_livekit/livekit_session.hpp similarity index 100% rename from src/plugins/sensing/livekit/inc/sensing_livekit/livekit_session.hpp rename to src/livekit_publisher/inc/sensing_livekit/livekit_session.hpp diff --git a/src/plugins/sensing/livekit/inc/sensing_livekit/video_track_sink.hpp b/src/livekit_publisher/inc/sensing_livekit/video_track_sink.hpp similarity index 64% rename from src/plugins/sensing/livekit/inc/sensing_livekit/video_track_sink.hpp rename to src/livekit_publisher/inc/sensing_livekit/video_track_sink.hpp index 85dac890df..2eb813e640 100644 --- a/src/plugins/sensing/livekit/inc/sensing_livekit/video_track_sink.hpp +++ b/src/livekit_publisher/inc/sensing_livekit/video_track_sink.hpp @@ -3,6 +3,8 @@ #pragma once +#include + #include #include #include @@ -13,26 +15,7 @@ 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; - /// Unix time in microseconds on the publishing host when the frame finished conversion: processing - /// time, not a capture time. Carried as the frame's application timestamp; the transport timestamp - /// above stays monotonic. - uint64_t user_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 = 0; -}; +using video_stage::EncodedVideoFrame; /// What the receiving side asked of the encoder since the last call. struct EncoderFeedback diff --git a/src/plugins/sensing/livekit/livekit_session.cpp b/src/livekit_publisher/livekit_session.cpp similarity index 97% rename from src/plugins/sensing/livekit/livekit_session.cpp rename to src/livekit_publisher/livekit_session.cpp index 929721ecc4..69c8c8847a 100644 --- a/src/plugins/sensing/livekit/livekit_session.cpp +++ b/src/livekit_publisher/livekit_session.cpp @@ -3,11 +3,10 @@ #include "inc/sensing_livekit/livekit_session.hpp" -#include "inc/sensing_livekit/clock_sync.hpp" - #include #include #include +#include #include #include @@ -268,7 +267,10 @@ class LiveKitVideoTrack : public VideoTrackSink 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.user_timestamp_us; + // Zero means the publisher cannot vouch for a capture time: the frame is sent without one, and a consumer + // that needs a timestamp (a recording) says so instead of using a guessed one. + if (frame.user_timestamp_us != 0) + metadata.user_timestamp_us = frame.user_timestamp_us; metadata.frame_id = frame.frame_id; std::vector tsc(8); for (int i = 0; i < 8; ++i) diff --git a/src/plugins/sensing/CMakeLists.txt b/src/plugins/sensing/CMakeLists.txt index 04dcea08e8..2e47a2e159 100644 --- a/src/plugins/sensing/CMakeLists.txt +++ b/src/plugins/sensing/CMakeLists.txt @@ -4,8 +4,7 @@ # ============================================================================== # SENSING Camera Plugin: Jetson SIPL capture, hardware H.264 encode, LiveKit publication # ============================================================================== -# Two leaf targets besides the LiveKit wrapper, one per directory (cmake/cmake-structure.md rule 1): -# livekit/ sensing_livekit -- the LiveKit SDK behind a small interface +# Two leaf targets, one per directory (cmake/cmake-structure.md rule 1); the LiveKit wrapper is src/livekit_publisher: # core/ sensing_core -- SIPL capture, encoder, conversion, track output # app/ camera_plugin_sensing # @@ -184,90 +183,6 @@ message(STATUS "SENSING plugin: V4L2=${V4L2_LIBRARY}") # dependency at all (readelf -d lists only EGL, cuda, dl, pthread, rt, c), and # the tegra lib dir ships no libnvjpeg to link against. -# ============================================================================== -# LiveKit -# ============================================================================== -# Each selected sensor is published as a LiveKit camera track. 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. -set(_livekit_local ON) -if(NOT LIVEKIT_SDK_DIR) - set(_livekit_local OFF) - # The prebuilt links against the glibc of the Ubuntu release it was built on: the 24.04 build does not load on - # JetPack 6 (Ubuntu 22.04). Pick the release matching the host; LIVEKIT_SDK_UBUNTU overrides the detection. - if(NOT LIVEKIT_SDK_UBUNTU) - set(LIVEKIT_SDK_UBUNTU "24.04") - if(EXISTS /etc/os-release) - file(STRINGS /etc/os-release _os_version_id REGEX "^VERSION_ID=") - if(_os_version_id MATCHES "\"?22\\.04\"?$") - set(LIVEKIT_SDK_UBUNTU "22.04") - endif() - endif() - endif() - if(CMAKE_SYSTEM_PROCESSOR MATCHES "aarch64|arm64") - set(_livekit_arch "arm64") - set(_livekit_hash_24.04 "e3a97f71943c0bf9fa4ebd70927ca2906b31dcd6c14b42312b1225b4b868e4ad") - set(_livekit_hash_22.04 "7a698bbd8f61f019744fd99b3a5d2b6c6118787fd1f81140c9a6fc8732fc87e0") - elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "x86_64|AMD64") - set(_livekit_arch "x64") - set(_livekit_hash_24.04 "5e0feb2e519c6ff131dd2d1c45149115057ae4e21a324fdb249e1789d5614e7b") - set(_livekit_hash_22.04 "9e568ee6b2bb3c7b0d07faba694930767096a14d0e09422ba01f34077ffb03cb") - else() - message(FATAL_ERROR "No prebuilt LiveKit SDK for ${CMAKE_SYSTEM_PROCESSOR}; set LIVEKIT_SDK_DIR.") - endif() - if(NOT DEFINED _livekit_hash_${LIVEKIT_SDK_UBUNTU}) - message(FATAL_ERROR "No prebuilt LiveKit SDK for Ubuntu ${LIVEKIT_SDK_UBUNTU}; set LIVEKIT_SDK_DIR.") - endif() - set(_livekit_hash "${_livekit_hash_${LIVEKIT_SDK_UBUNTU}}") - include(FetchContent) - FetchContent_Declare( - livekit_sdk - URL https://github.com/livekit/client-sdk-cpp/releases/download/v1.12.4/livekit-sdk-ubuntu-${LIVEKIT_SDK_UBUNTU}-${_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. The pinned download supplies the SDK's own; a local SDK uses -# LICENSE/NOTICE from its directory, or LIVEKIT_SDK_LICENSE_FILE / LIVEKIT_SDK_NOTICE_FILE, and needs no network. -foreach(_notice LICENSE NOTICE) - set(_override "${LIVEKIT_SDK_${_notice}_FILE}") - if(_override) - set(_found "${_override}") - elseif(_livekit_local) - set(_found "") - foreach(_candidate "${LIVEKIT_SDK_DIR}/${_notice}" "${LIVEKIT_SDK_DIR}/../${_notice}") - if(NOT _found AND EXISTS "${_candidate}") - set(_found "${_candidate}") - endif() - endforeach() - if(NOT _found) - message(FATAL_ERROR - "LIVEKIT_SDK_DIR is set but ${_notice} was not found in it. Put the SDK's ${_notice} file there " - "or pass -DLIVEKIT_SDK_${_notice}_FILE=/path/to/${_notice}.") - endif() - else() - 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(_found "${livekit_sdk_${_notice}_SOURCE_DIR}/${_notice}") - endif() - set(SENSING_LIVEKIT_${_notice}_FILE "${_found}" 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 "") - - # ============================================================================== # Targets # ============================================================================== @@ -276,7 +191,6 @@ message(STATUS "Building SENSING camera plugin") # The platform configs live beside this file; app/ stages and installs them from here. set(SENSING_CONFIGS_DIR "${CMAKE_CURRENT_SOURCE_DIR}/configs") -add_subdirectory(livekit) add_subdirectory(core) add_subdirectory(app) diff --git a/src/plugins/sensing/livekit/CMakeLists.txt b/src/plugins/sensing/livekit/CMakeLists.txt deleted file mode 100644 index 6151bae977..0000000000 --- a/src/plugins/sensing/livekit/CMakeLists.txt +++ /dev/null @@ -1,24 +0,0 @@ -# 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 H.264 access units. -# 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/clock_sync.hpp b/src/plugins/sensing/livekit/inc/sensing_livekit/clock_sync.hpp deleted file mode 100644 index dfefbed700..0000000000 --- a/src/plugins/sensing/livekit/inc/sensing_livekit/clock_sync.hpp +++ /dev/null @@ -1,86 +0,0 @@ -// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -// SPDX-License-Identifier: Apache-2.0 -// -// The publisher's half of the PC's four-timestamp clock exchange: parse a request, build the reply. All times are -// CLOCK_MONOTONIC nanoseconds, the clock the PC maps. Wire format (little-endian), shared with -// examples/egocap/backend/egocap_web/livekit_clock_sync.py and the egocap emulator: -// request, 13 bytes: 'Q', u32 sequence, i64 t1 (the requester's clock) -// reply, 29 bytes: 'R', u32 sequence, i64 t1 (both echoed), i64 t2 (receipt), i64 t3 (reply send) -// No LiveKit type appears here, so the format is testable on its own. - -#pragma once - -#include -#include -#include -#include -#include - -namespace plugins -{ -namespace sensing -{ - -inline constexpr const char* kClockSyncTopic = "isaaccapture.clock-sync"; -inline constexpr std::size_t kClockSyncRequestSize = 13; -inline constexpr std::size_t kClockSyncReplySize = 29; -/// Byte offset of t3 in a reply, so it can be written as late as possible, just before the send. -inline constexpr std::size_t kClockSyncReplyT3Offset = 21; - -struct ClockSyncRequest -{ - std::uint32_t sequence = 0; - std::int64_t t1 = 0; -}; - -inline std::int64_t monotonic_ns() -{ - timespec ts{}; - clock_gettime(CLOCK_MONOTONIC, &ts); - return static_cast(ts.tv_sec) * 1'000'000'000LL + ts.tv_nsec; -} - -inline void put_le(std::vector& out, std::size_t at, std::uint64_t value, std::size_t bytes) -{ - for (std::size_t i = 0; i < bytes; ++i) - out[at + i] = static_cast((value >> (8 * i)) & 0xFF); -} - -inline std::uint64_t get_le(const std::vector& in, std::size_t at, std::size_t bytes) -{ - std::uint64_t value = 0; - for (std::size_t i = 0; i < bytes; ++i) - value |= static_cast(in[at + i]) << (8 * i); - return value; -} - -/// The request in `data`, or nullopt for anything that is not exactly a request. -inline std::optional parse_clock_sync_request(const std::vector& data) -{ - if (data.size() != kClockSyncRequestSize || data[0] != 'Q') - return std::nullopt; - ClockSyncRequest request; - request.sequence = static_cast(get_le(data, 1, 4)); - request.t1 = static_cast(get_le(data, 5, 8)); - return request; -} - -inline std::vector build_clock_sync_reply(const ClockSyncRequest& request, std::int64_t t2, std::int64_t t3) -{ - std::vector reply(kClockSyncReplySize); - reply[0] = 'R'; - put_le(reply, 1, request.sequence, 4); - put_le(reply, 5, static_cast(request.t1), 8); - put_le(reply, 13, static_cast(t2), 8); - put_le(reply, kClockSyncReplyT3Offset, static_cast(t3), 8); - return reply; -} - -/// Writes t3 into an already built reply. -inline void set_clock_sync_reply_t3(std::vector& reply, std::int64_t t3) -{ - put_le(reply, kClockSyncReplyT3Offset, static_cast(t3), 8); -} - -} // namespace sensing -} // namespace plugins diff --git a/src/plugins/sensing/tests/CMakeLists.txt b/src/plugins/sensing/tests/CMakeLists.txt index eaf7b80670..fe62e0f347 100644 --- a/src/plugins/sensing/tests/CMakeLists.txt +++ b/src/plugins/sensing/tests/CMakeLists.txt @@ -26,16 +26,6 @@ target_link_libraries(sensing_buffer_owner_tests PRIVATE Catch2::Catch2WithMain catch_discover_tests(sensing_buffer_owner_tests) -# The clock-sync wire format is header-only and LiveKit-free, so the test needs only its include path. -add_executable(sensing_clock_sync_tests - test_clock_sync.cpp -) - -target_include_directories(sensing_clock_sync_tests PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../livekit/inc) - -target_link_libraries(sensing_clock_sync_tests PRIVATE Catch2::Catch2WithMain) - -catch_discover_tests(sensing_clock_sync_tests) # The sensor-clock mapping is header-only and injectable, so it needs no SDK. add_executable(sensing_sensor_clock_tests diff --git a/src/plugins/sensing/tests/test_clock_sync.cpp b/src/plugins/sensing/tests/test_clock_sync.cpp deleted file mode 100644 index db0176b7ec..0000000000 --- a/src/plugins/sensing/tests/test_clock_sync.cpp +++ /dev/null @@ -1,135 +0,0 @@ -// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -// SPDX-License-Identifier: Apache-2.0 - -#include "sensing_livekit/clock_sync.hpp" - -#include - -#include -#include -#include -#include - -using plugins::sensing::build_clock_sync_reply; -using plugins::sensing::ClockSyncRequest; -using plugins::sensing::kClockSyncReplySize; -using plugins::sensing::kClockSyncRequestSize; -using plugins::sensing::monotonic_ns; -using plugins::sensing::parse_clock_sync_request; -using plugins::sensing::set_clock_sync_reply_t3; - -namespace -{ - -using Bytes = std::vector; - -Bytes from_hex(const std::string& hex) -{ - Bytes out; - for (std::size_t i = 0; i + 1 < hex.size(); i += 2) - out.push_back(static_cast(std::stoul(hex.substr(i, 2), nullptr, 16))); - return out; -} - -// Golden bytes produced by egocap_web.livekit_clock_sync.encode_request and -// tools/livekit_orin_emulator.build_sync_reply for the inputs named beside each constant. -// sequence 1, t1 1, t2 2, t3 3 -const char* const kSimpleRequest = "51010000000100000000000000"; -const char* const kSimpleReply = "5201000000010000000000000002000000000000000300000000000000"; -// sequence 0xFFFFFFFF, t1 -1234567890123, t2 INT64_MAX, t3 INT64_MIN -const char* const kExtremeRequest = "51ffffffff35fb048ee0feffff"; -const char* const kExtremeReply = "52ffffffff35fb048ee0feffffffffffffffffff7f0000000000000080"; - -} // namespace - -TEST_CASE("parse_clock_sync_request accepts a well-formed request", "[clock_sync]") -{ - auto request = parse_clock_sync_request(from_hex(kSimpleRequest)); - REQUIRE(request.has_value()); - CHECK(request->sequence == 1); - CHECK(request->t1 == 1); - - request = parse_clock_sync_request(from_hex(kExtremeRequest)); - REQUIRE(request.has_value()); - CHECK(request->sequence == 0xFFFFFFFFu); - CHECK(request->t1 == -1234567890123LL); -} - -TEST_CASE("parse_clock_sync_request rejects anything else", "[clock_sync]") -{ - for (std::size_t size : { std::size_t{ 0 }, std::size_t{ 12 }, std::size_t{ 14 }, std::size_t{ 29 } }) - { - Bytes data(size, 0); - if (size > 0) - data[0] = 'Q'; - CHECK_FALSE(parse_clock_sync_request(data).has_value()); - } - - Bytes wrong_kind = from_hex(kSimpleRequest); - wrong_kind[0] = 'X'; - CHECK_FALSE(parse_clock_sync_request(wrong_kind).has_value()); - - CHECK_FALSE(parse_clock_sync_request(from_hex(kSimpleReply)).has_value()); - - Bytes reply_kind_at_request_size = from_hex(kSimpleRequest); - reply_kind_at_request_size[0] = 'R'; - CHECK_FALSE(parse_clock_sync_request(reply_kind_at_request_size).has_value()); -} - -TEST_CASE("build_clock_sync_reply lays out the documented fields", "[clock_sync]") -{ - const Bytes reply = build_clock_sync_reply(ClockSyncRequest{ 1, 1 }, 2, 3); - REQUIRE(reply.size() == kClockSyncReplySize); - CHECK(reply[0] == 'R'); - CHECK(reply == from_hex(kSimpleReply)); - - // Explicit offsets, little-endian. - CHECK(reply[1] == 1); - CHECK(reply[5] == 1); - CHECK(reply[13] == 2); - CHECK(reply[21] == 3); -} - -TEST_CASE("build_clock_sync_reply handles large and negative values", "[clock_sync]") -{ - const ClockSyncRequest request{ 0xFFFFFFFFu, -1234567890123LL }; - const Bytes reply = build_clock_sync_reply( - request, std::numeric_limits::max(), std::numeric_limits::min()); - CHECK(reply == from_hex(kExtremeReply)); -} - -TEST_CASE("clock sync request and reply round trip through parse", "[clock_sync]") -{ - const Bytes request_bytes = from_hex(kExtremeRequest); - REQUIRE(request_bytes.size() == kClockSyncRequestSize); - const auto request = parse_clock_sync_request(request_bytes); - REQUIRE(request.has_value()); - const Bytes reply = build_clock_sync_reply( - *request, std::numeric_limits::max(), std::numeric_limits::min()); - CHECK(reply == from_hex(kExtremeReply)); -} - -TEST_CASE("set_clock_sync_reply_t3 changes only the t3 bytes", "[clock_sync]") -{ - const Bytes before = build_clock_sync_reply(ClockSyncRequest{ 7, -5 }, 11, 13); - Bytes after = before; - set_clock_sync_reply_t3(after, 0x0102030405060708LL); - - REQUIRE(after.size() == before.size()); - for (std::size_t i = 0; i < before.size(); ++i) - { - if (i >= 21 && i <= 28) - CHECK(after[i] == static_cast(0x08 - (i - 21))); - else - CHECK(after[i] == before[i]); - } - CHECK(after == build_clock_sync_reply(ClockSyncRequest{ 7, -5 }, 11, 0x0102030405060708LL)); -} - -TEST_CASE("monotonic_ns does not go backwards", "[clock_sync]") -{ - const std::int64_t a = monotonic_ns(); - const std::int64_t b = monotonic_ns(); - CHECK(a > 0); - CHECK(b >= a); -} diff --git a/src/video_stage/CMakeLists.txt b/src/video_stage/CMakeLists.txt index 6f1465ba6a..d7bb15b103 100644 --- a/src/video_stage/CMakeLists.txt +++ b/src/video_stage/CMakeLists.txt @@ -2,12 +2,13 @@ # SPDX-License-Identifier: Apache-2.0 # ============================================================================== -# Video stage: the Orin-to-PC video transport +# Video stage: the Orin-to-PC video transport and the PC relay to LiveKit # ============================================================================== # wire/ video_stage_wire -- AVS1 records, clock exchange, H.264 access-unit helpers, frame correlator # sender/ video_stage_sender -- the Orin-side sender of AVS1 records over a direct link +# app/ video_stage_worker -- the PC relay: receives the Orin's records and publishes them to LiveKit or a file # -# Plain C++ and POSIX sockets: no CUDA, no LiveKit. +# No CUDA. The LiveKit SDK (src/livekit_publisher) is needed only by the relay. # # Configure with: # cmake -B build -DBUILD_VIDEO_STAGE=ON @@ -15,7 +16,9 @@ add_subdirectory(wire) add_subdirectory(sender) +add_subdirectory(app) if(BUILD_TESTING) add_subdirectory(wire/tests) + add_subdirectory(tests) endif() diff --git a/src/video_stage/README.md b/src/video_stage/README.md new file mode 100644 index 0000000000..120aa12eab --- /dev/null +++ b/src/video_stage/README.md @@ -0,0 +1,73 @@ + + +# Video stage + +Moves camera video from the Orin to the PC and on to LiveKit. The Orin sends its encoded H.264 access units to the PC +over a direct link; the PC relay (`video_stage_worker`) hands them, untouched, to LiveKit or a file. There is no decode, +encode or GPU in the relay, and it builds without CUDA. + +```text +Orin producer --DATA (AVS1 records, H.264)--> PC relay --> LiveKit track (or an Annex-B file) + <--CTRL (keyframe requests, clock exchange)-- +``` + +- `wire/` the AVS1 record codec, clock exchange, H.264 access-unit helpers and the frame correlator. +- `sender/` the Orin-side sender of AVS1 records. +- `app/` the relay. + +`video_stage_worker` serves one Orin camera per `--stream=PORT:TRACK`; all tracks are published under one LiveKit +participant. The recorder and the web page subscribe to the relayed tracks as they would to the Orin's own. + +- **Direct input.** The sender connects to each stream's port and sends every access unit as an AVS1 record over TCP + (`wire/`; codec 2 is H.264). It reconnects by itself, starts at a keyframe, and drops frames up to the next keyframe + rather than block when the link cannot keep up. Use a wired link. +- **Control link.** A second connection from the Orin carries the PC's keyframe requests and a four-timestamp clock + exchange (the same bytes the recorder uses over LiveKit). The relay converts each Orin frame-ready stamp onto its own + monotonic clock before publishing, and answers the recorder's and the page's clock sync with that clock. Video is never + withheld because of timing: while the Orin-to-PC mapping is missing or less certain than + `--max-upstream-uncertainty-ms` (5 by default) the frames are still published, without a capture timestamp, so a + recording that needs one stops and says so. When the Orin's clock jumps, the track is unpublished and published again at + the next keyframe. +- **Fixed rate and geometry.** The relay publishes what its source encodes, at that rate; a downstream bitrate request is + logged once and not acted on. A stream's picture size is fixed for the run; a change ends the stream with an error. +- **Overload.** A relay that cannot keep up discards its queued records and resumes at the next keyframe; it never drops + compressed frames one by one. + +Declare the relayed tracks like any other camera in the CaptureServer streams file, with the relay's participant as the +publisher and the monotonic basis it publishes: + +```yaml +livekit: + url: ws://PC_HOST:7880 + room: sensing-preview + publisher: stage-worker # the identity in the relay's token + timestamp_basis: monotonic_us + streams: + - {name: Head left, track: processed-left, width: 2560, height: 1984, fps: 30} + - {name: Head right, track: processed-right, width: 2560, height: 1984, fps: 30} +``` + +```bash +video_stage_worker --stream=5600:processed-left --stream=5601:processed-right \ + --livekit-url=ws://127.0.0.1:7880 --livekit-token-file=worker.token +# a file sink instead of LiveKit, to check a run: +video_stage_worker --stream=5600:file=out.h264 --once +``` + +`tests/tools/mock_orin_sender.py` stands in for the Orin, with no Orin or camera: it mirrors the sender's connection and +hello order, retry, keyframe start, bounded queue with drop-to-keyframe, keyframe requests and clock answers (on its own +offset clock), and feeds it Jetson-style H.264 (no AUDs, IDR every 5 s, constant bitrate). One `--port` per camera. Run it +against a relay writing a file, then check the file with `ffprobe`. + +## Build + +```bash +cmake -B build -DBUILD_VIDEO_STAGE=ON -DBUILD_PLUGINS=OFF -DBUILD_VIZ=OFF -DBUILD_EXAMPLES=OFF \ + -DBUILD_PYTHON_BINDINGS=OFF +cmake --build build && ctest --test-dir build -R 'video_stage|avs1|clock' +``` + +The LiveKit SDK is fetched, or taken from `LIVEKIT_SDK_DIR`. diff --git a/src/video_stage/app/CMakeLists.txt b/src/video_stage/app/CMakeLists.txt new file mode 100644 index 0000000000..0d29961af5 --- /dev/null +++ b/src/video_stage/app/CMakeLists.txt @@ -0,0 +1,18 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# The PC relay: receives AVS1 access units from the Orin over its direct link and hands them, untouched, to LiveKit or +# a file. +add_executable(video_stage_worker + main.cpp + control_channel.cpp + livekit_output.cpp + stream_worker.cpp + tcp_receiver.cpp +) + +target_link_libraries(video_stage_worker + PRIVATE + video_stage::wire + sensing::livekit +) diff --git a/src/video_stage/app/control_channel.cpp b/src/video_stage/app/control_channel.cpp new file mode 100644 index 0000000000..36c8fa1ea2 --- /dev/null +++ b/src/video_stage/app/control_channel.cpp @@ -0,0 +1,113 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "control_channel.hpp" + +#include +#include + +#include +#include + +namespace video_stage +{ + +namespace cs = plugins::sensing; + +ControlChannel::ControlChannel(TcpReceiver& link, uint64_t first_epoch) : m_link(link), m_epoch(first_epoch) +{ + m_reader = std::thread(&ControlChannel::read_loop, this); + m_requester = std::thread(&ControlChannel::request_loop, this); +} + +ControlChannel::~ControlChannel() +{ + { + std::lock_guard lock(m_wait_mutex); + m_stop = true; + } + m_wait.notify_all(); + m_link.shutdown_session(); // unblocks the reader + if (m_requester.joinable()) + m_requester.join(); + if (m_reader.joinable()) + m_reader.join(); +} + +std::optional ControlChannel::mapping() const +{ + std::lock_guard lock(m_mutex); + const auto mapping = m_estimator.mapping(monotonic_ns()); + if (!mapping) + return std::nullopt; + return UpstreamMapping{ *mapping, m_epoch }; +} + +void ControlChannel::request_loop() +{ + uint32_t sequence = 0; + while (true) + { + cs::ClockSyncRequest request; + request.sequence = ++sequence; + request.t1 = monotonic_ns(); + { + std::lock_guard lock(m_mutex); + m_outstanding[m_next_outstanding] = { request.sequence, request.t1, false }; + m_next_outstanding = (m_next_outstanding + 1) % kOutstanding; + } + if (!m_link.send_control(cs::build_clock_sync_request(request).data(), cs::kClockSyncRequestSize)) + return; + std::unique_lock lock(m_wait_mutex); + if (m_wait.wait_for(lock, std::chrono::seconds(1), [this] { return m_stop; })) + return; + } +} + +void ControlChannel::read_loop() +{ + while (true) + { + uint8_t first = 0; + if (!m_link.recv_control(&first, 1)) + return; + const int64_t t4 = monotonic_ns(); // stamped before anything else runs + if (first != 'R') + continue; // the Orin sends nothing else; ignore rather than guess + std::vector reply(cs::kClockSyncReplySize); + reply[0] = first; + if (!m_link.recv_control(reply.data() + 1, reply.size() - 1)) + return; + const auto parsed = cs::parse_clock_sync_reply(reply); + if (!parsed) + continue; + const Exchange exchange{ parsed->t1, parsed->t2, parsed->t3, t4 }; + std::lock_guard lock(m_mutex); + Outstanding* match = nullptr; + for (Outstanding& o : m_outstanding) + { + if (!o.used && o.sequence == parsed->sequence && o.t1 == parsed->t1) + match = &o; + } + if (match == nullptr) + { + ++m_rejected; // not the answer to a request of this session + continue; + } + match->used = true; + if (!m_estimator.add(exchange).empty()) + { + ++m_rejected; + continue; + } + ++m_exchanges; + if (m_estimator.discontinuity()) + { + ++m_discontinuities; + ++m_epoch; + m_estimator.clear_discontinuity(); + } + } +} + +} // namespace video_stage diff --git a/src/video_stage/app/control_channel.hpp b/src/video_stage/app/control_channel.hpp new file mode 100644 index 0000000000..0a6d7d821f --- /dev/null +++ b/src/video_stage/app/control_channel.hpp @@ -0,0 +1,90 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include "tcp_receiver.hpp" + +#include + +#include +#include +#include +#include +#include +#include + +namespace video_stage +{ + +/// The Orin-to-PC mapping and the epoch it belongs to. The epoch increases whenever the Orin's clock was found to have +/// jumped relative to this one; frames mapped under different epochs must not be part of one synchronized take. +struct UpstreamMapping +{ + ClockMapping mapping; + uint64_t epoch = 0; +}; + +/** + * @brief The PC's half of the Orin clock exchange, over the control connection. + * + * Once a second it sends a four-timestamp request (the same wire the recorder and the browser use over LiveKit) and + * feeds each reply to a ClockEstimator, so the Orin's monotonic stamps can be placed on this machine's monotonic clock. + * A reader thread stamps t4 the instant a reply's first byte arrives. On a discontinuity the estimator restarts from + * the new exchange; frames mapped in the interval before it was noticed carry the old offset. + */ +class ControlChannel +{ +public: + /// `first_epoch` differs for every session of a stream, so an epoch is never reused across a sender restart. + ControlChannel(TcpReceiver& link, uint64_t first_epoch); + ~ControlChannel(); + + ControlChannel(const ControlChannel&) = delete; + ControlChannel& operator=(const ControlChannel&) = delete; + + /// The current Orin-to-PC mapping with its uncertainty aged to now; nullopt before the first reply or when stale. + std::optional mapping() const; + + uint64_t exchanges() const + { + return m_exchanges.load(); + } + uint64_t rejected() const + { + return m_rejected.load(); + } + uint64_t discontinuities() const + { + return m_discontinuities.load(); + } + +private: + void request_loop(); + void read_loop(); + + TcpReceiver& m_link; + mutable std::mutex m_mutex; + ClockEstimator m_estimator; + std::atomic m_exchanges{ 0 }; + std::atomic m_rejected{ 0 }; + std::atomic m_discontinuities{ 0 }; + // Requests sent and not yet answered (sequence, t1): a reply is used only if it answers one of these. + struct Outstanding + { + uint32_t sequence = 0; + int64_t t1 = 0; + bool used = true; + }; + static constexpr size_t kOutstanding = 4; + Outstanding m_outstanding[kOutstanding]{}; + size_t m_next_outstanding = 0; + uint64_t m_epoch; + std::mutex m_wait_mutex; + std::condition_variable m_wait; + bool m_stop = false; + std::thread m_requester; + std::thread m_reader; +}; + +} // namespace video_stage diff --git a/src/video_stage/app/livekit_output.cpp b/src/video_stage/app/livekit_output.cpp new file mode 100644 index 0000000000..24c2cdc293 --- /dev/null +++ b/src/video_stage/app/livekit_output.cpp @@ -0,0 +1,163 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "livekit_output.hpp" + +#include +#include + +namespace video_stage +{ + +namespace sensing = plugins::sensing; + +SharedLiveKit::SharedLiveKit(const std::string& url, const std::string& token) +{ + m_session = sensing::LiveKitSession::connect(url, token); +} + +std::unique_ptr SharedLiveKit::publish(const sensing::PublishSpec& spec) +{ + std::lock_guard lock(m_mutex); + return m_session->publish_video(spec); +} + +bool SharedLiveKit::failed() const +{ + return m_session->failed(); +} + +std::string SharedLiveKit::failure() const +{ + return m_session->failure(); +} + +LiveKitOutput::LiveKitOutput(SharedLiveKit& livekit, + const LiveKitOutputConfig& config, + MappingSource mapping, + std::function request_idr) + : m_livekit(livekit), m_config(config), m_mapping(std::move(mapping)), m_request_idr(std::move(request_idr)) +{ +} + +void LiveKitOutput::break_publication() +{ + m_track.reset(); // destroying the sink unpublishes the track + m_awaiting_keyframe = true; + m_idr_requested_for_gap = false; +} + +void LiveKitOutput::end_session() +{ + break_publication(); +} + +void LiveKitOutput::write(const EncodedPacket& packet, const FrameMeta& meta, const Geometry& geometry) +{ + const int64_t write_start_ns = monotonic_ns(); + if (m_livekit.failed()) + throw std::runtime_error("LiveKit room lost: " + m_livekit.failure()); + + // A subscriber's keyframe or rate request reaches the encoder through the pipeline. + if (m_track) + { + const auto feedback = m_track->take_feedback(); + if (feedback.keyframe_requested) + m_request_idr(); + // This worker runs at the rate its source encodes; it does not adapt it to downstream congestion. Say so once + // instead of silently discarding the request. + if (feedback.target_bitrate_bps && !m_rate_feedback_noted) + { + m_rate_feedback_noted = true; + std::cerr << "stream " << m_config.track_name << ": downstream asked for " + << *feedback.target_bitrate_bps / 1000 + << " kbit/s; this worker forwards its source's rate and does not adapt it" << std::endl; + } + } + + // Video keeps flowing whatever the timing quality: a viewer that loses the picture every time the link's clock + // mapping wobbles is worse than one that sees video without a latency figure. What a frame may claim about time + // depends on the mapping: a qualified one carries the Orin's capture time on this PC's clock; otherwise the frame + // carries no timestamp, and a consumer that needs one (a recording) says so instead of using a guessed time. + const auto upstream = m_mapping(); + bool qualified = false; + if (meta.source_timestamp_ns == 0) + ++m_no_source_stamp; // the sender could not tie this unit to a capture: no capture time to map + else if (!upstream) + ++m_unmapped; + else if (upstream->mapping.uncertainty_ns > m_config.max_upstream_uncertainty_ns) + ++m_over_budget; + else + qualified = true; + + // This milestone publishes one fixed picture size: a change is refused rather than reconfiguring the track. + if (m_track && !(geometry == m_track_geometry)) + throw std::runtime_error("the stream's picture size changed; publishing a different size is not supported"); + // The Orin's clock jumped: end this publication so consumers see a break instead of a silent offset change. + if (upstream) + { + if (m_track && upstream->epoch != m_epoch) + { + ++m_republished; + break_publication(); + } + m_epoch = upstream->epoch; + } + + if (m_awaiting_keyframe && !packet.keyframe) + { + // Resume at a keyframe: ask for one once per gap instead of waiting out the encoder's own interval. + if (!m_idr_requested_for_gap) + { + m_idr_requested_for_gap = true; + m_request_idr(); + } + return; + } + if (!m_track) + { + sensing::PublishSpec spec; + spec.name = m_config.track_name; + spec.width = geometry.width; + spec.height = geometry.height; + spec.max_bitrate_bps = m_config.max_bitrate_bps; + spec.max_fps = m_config.fps; + m_track = m_livekit.publish(spec); + m_track_geometry = geometry; + } + m_awaiting_keyframe = false; + + // Without a qualified mapping the transport timestamp is the arrival time here and the frame has no capture time. + const int64_t pc_ns = + qualified ? upstream->mapping.to_local_ns(static_cast(meta.source_timestamp_ns)) : meta.received_ns; + sensing::EncodedVideoFrame frame; + frame.data = packet.data.data(); + frame.size = packet.data.size(); + frame.keyframe = packet.keyframe; + frame.timestamp_us = static_cast(pc_ns / 1000); + frame.user_timestamp_us = qualified ? frame.timestamp_us : 0; + frame.frame_id = static_cast(meta.frame_number); + const int64_t submit_start_ns = monotonic_ns(); + const bool accepted = m_track->submit(frame); + const int64_t now_ns = monotonic_ns(); + if (accepted) + { + ++m_published; + m_intake_to_write_ns.add(write_start_ns - meta.received_ns); + m_submit_ns.add(now_ns - submit_start_ns); + if (qualified) + { + m_uplink_ns.add(meta.received_ns - pc_ns); + m_source_to_publish_ns.add(now_ns - pc_ns); + m_upstream_uncertainty_ns = upstream->mapping.uncertainty_ns; + } + return; + } + // The SDK refused the frame, which can break the reference chain: continue from a keyframe. + ++m_refused; + m_awaiting_keyframe = true; + m_idr_requested_for_gap = true; + m_request_idr(); +} + +} // namespace video_stage diff --git a/src/video_stage/app/livekit_output.hpp b/src/video_stage/app/livekit_output.hpp new file mode 100644 index 0000000000..a94ed48872 --- /dev/null +++ b/src/video_stage/app/livekit_output.hpp @@ -0,0 +1,158 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include "control_channel.hpp" +#include "stats.hpp" + +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace video_stage +{ + +/// One LiveKit participant shared by every stream of the worker, as the Orin producer shares one for its cameras: one +/// identity, one room, one track per stream. Publishing is serialised because streams run on their own threads. +class SharedLiveKit +{ +public: + SharedLiveKit(const std::string& url, const std::string& token); + + std::unique_ptr publish(const plugins::sensing::PublishSpec& spec); + bool failed() const; + std::string failure() const; + +private: + plugins::sensing::LiveKitSdkReference m_sdk; + std::unique_ptr m_session; + std::mutex m_mutex; +}; + +struct LiveKitOutputConfig +{ + std::string track_name; + uint64_t max_bitrate_bps = 12'000'000; + double fps = 30.0; + /// Frames are withheld while the Orin-to-PC mapping is less certain than this. The recorder's and the page's own + /// uncertainty covers only the leg from this worker to them; this leg is on top of it. + int64_t max_upstream_uncertainty_ns = 5'000'000; +}; + +/** + * @brief Publishes the processed access units as an ordinary LiveKit camera track. + * + * The recorder and the web page subscribe to it exactly as they do to the Orin's own track, so it follows the same + * conventions: per-frame application timestamp and frame id in the packet trailer, and the publisher's own + * monotonic clock answered on the clock-sync topic. The Orin's frame-ready stamp is converted onto this machine's + * monotonic clock through the control-link mapping before publication, so consumers map one clock, never two. + * + * Timing breaks are explicit, because consumers cannot see this leg: a frame with no mapping, or a mapping worse than + * the budget, is withheld; and when the mapping's epoch changes (the Orin's clock jumped) the track is unpublished and + * published again at the next keyframe, which a recorder reports as a lost track instead of silently continuing a + * synchronized take through the jump. The picture size is fixed per stream: a change is an error. + */ +class LiveKitOutput final : public PacketSink +{ +public: + /// Where the Orin-to-PC clock mapping comes from; empty when there is none (no session, or a stale mapping). + using MappingSource = std::function()>; + + /// `request_idr` is called when the track needs a keyframe (a subscriber joined, a frame was refused, or a gap). + LiveKitOutput(SharedLiveKit& livekit, + const LiveKitOutputConfig& config, + MappingSource mapping, + std::function request_idr); + + void write(const EncodedPacket& packet, const FrameMeta& meta, const Geometry& geometry) override; + + /// The source session ended: unpublish, so a sender restart is never carried through as one continuous take. The + /// next session publishes a fresh track at its first keyframe. + void end_session(); + + uint64_t published() const + { + return m_published.load(); + } + /// Frames the sender sent without a capture time (it could not tie the unit to a capture). + uint64_t no_source_stamp() const + { + return m_no_source_stamp.load(); + } + uint64_t unmapped() const + { + return m_unmapped.load(); + } + uint64_t refused() const + { + return m_refused.load(); + } + uint64_t republished() const + { + return m_republished.load(); + } + uint64_t over_budget() const + { + return m_over_budget.load(); + } + /// Orin frame-ready to arrival at this PC (Orin encode, its queue and the link), per published frame. + const DurationStats& uplink_ns() const + { + return m_uplink_ns; + } + /// Orin frame-ready to the moment this worker hands the frame to LiveKit: everything before the SFU and the page. + const DurationStats& source_to_publish_ns() const + { + return m_source_to_publish_ns; + } + /// From the intake of the record to the start of this publish call: the worker's own time (queue, decode, warp, + /// encode). + const DurationStats& intake_to_write_ns() const + { + return m_intake_to_write_ns; + } + /// The LiveKit SDK call that takes the encoded frame. + const DurationStats& submit_ns() const + { + return m_submit_ns; + } + int64_t upstream_uncertainty_ns() const + { + return m_upstream_uncertainty_ns.load(); + } + +private: + void break_publication(); ///< unpublish; the next keyframe publishes a fresh track + + SharedLiveKit& m_livekit; + LiveKitOutputConfig m_config; + MappingSource m_mapping; + std::function m_request_idr; + std::unique_ptr m_track; + Geometry m_track_geometry; + uint64_t m_epoch = 0; + bool m_awaiting_keyframe = true; + bool m_idr_requested_for_gap = false; + bool m_rate_feedback_noted = false; + // Counted on the output's thread and read by the statistics thread. + std::atomic m_published{ 0 }; + std::atomic m_unmapped{ 0 }; + std::atomic m_no_source_stamp{ 0 }; + std::atomic m_refused{ 0 }; + std::atomic m_republished{ 0 }; + std::atomic m_over_budget{ 0 }; + DurationStats m_intake_to_write_ns; + DurationStats m_submit_ns; + DurationStats m_uplink_ns; + DurationStats m_source_to_publish_ns; + std::atomic m_upstream_uncertainty_ns{ 0 }; +}; + +} // namespace video_stage diff --git a/src/video_stage/app/main.cpp b/src/video_stage/app/main.cpp new file mode 100644 index 0000000000..4088c4e677 --- /dev/null +++ b/src/video_stage/app/main.cpp @@ -0,0 +1,165 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "livekit_output.hpp" +#include "stream_worker.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace video_stage; + +namespace +{ + +std::atomic g_stop{ false }; + +void on_signal(int) +{ + g_stop = true; +} + +void usage(const char* program) +{ + std::cout + << "Usage: " << program << " --stream=PORT:TRACK ... --livekit-url=URL --livekit-token-file=PATH\n" + << " " << program << " --stream=PORT:file=PATH.h264 ...\n" + << " [--bitrate=BPS] [--max-upstream-uncertainty-ms=MS] [--stats-interval=S] [--once]\n" + << "One --stream per camera. Each listens on PORT for the Orin's direct connection (AVS1 H.264 records plus a\n" + << "control link) and publishes the access units as LiveKit track TRACK (all tracks under one participant) or\n" + << "writes them to a file. The Orin's access units are forwarded untouched, with no decode, encode or GPU. Serves\n" + << "one Orin session after another unless --once. Prints a summary at the end.\n"; +} + +std::string read_token(const std::string& path) +{ + std::ifstream in(path); + std::string token; + if (!in || !std::getline(in, token) || token.empty()) + throw std::runtime_error("cannot read a token from " + path); + return token; +} + +// "PORT:TRACK" or "PORT:file=PATH". +StreamSpec parse_stream(const std::string& text) +{ + const auto colon = text.find(':'); + if (colon == std::string::npos || colon == 0 || colon + 1 == text.size()) + throw std::runtime_error("--stream must be PORT:TRACK or PORT:file=PATH, got '" + text + "'"); + StreamSpec spec; + spec.port = static_cast(std::stoul(text.substr(0, colon))); + const std::string rest = text.substr(colon + 1); + if (rest.rfind("file=", 0) == 0) + spec.file = rest.substr(5); + else + spec.track = rest; + return spec; +} + +} // namespace + +int main(int argc, char** argv) +try +{ + std::vector specs; + std::string livekit_url; + std::string token_file; + StreamSettings settings; + int stats_interval_s = 5; + for (int i = 1; i < argc; ++i) + { + const std::string arg = argv[i]; + if (arg.rfind("--stream=", 0) == 0) + specs.push_back(parse_stream(arg.substr(9))); + else if (arg.rfind("--livekit-url=", 0) == 0) + livekit_url = arg.substr(14); + else if (arg.rfind("--livekit-token-file=", 0) == 0) + token_file = arg.substr(21); + else if (arg.rfind("--stats-interval=", 0) == 0) + stats_interval_s = std::stoi(arg.substr(17)); + else if (arg == "--once") + settings.once = true; + else if (arg.rfind("--bitrate=", 0) == 0) + settings.livekit.max_bitrate_bps = static_cast(std::stoul(arg.substr(10))); + else if (arg.rfind("--max-upstream-uncertainty-ms=", 0) == 0) + settings.livekit.max_upstream_uncertainty_ns = static_cast(std::stod(arg.substr(30)) * 1e6); + else + { + usage(argv[0]); + return arg == "--help" ? 0 : 1; + } + } + if (specs.empty()) + { + usage(argv[0]); + return 1; + } + bool any_livekit = false; + for (const StreamSpec& s : specs) + any_livekit = any_livekit || s.file.empty(); + if (any_livekit && (livekit_url.empty() || token_file.empty())) + { + std::cerr << "a --stream with a track needs --livekit-url and --livekit-token-file\n"; + return 1; + } + + // One LiveKit participant for every stream, as the Orin producer joins once for all its cameras. + std::unique_ptr livekit; + if (any_livekit) + livekit = std::make_unique(livekit_url, read_token(token_file)); + + std::vector> workers; + for (const StreamSpec& spec : specs) + workers.push_back(std::make_unique(spec, settings, spec.file.empty() ? livekit.get() : nullptr)); + + std::signal(SIGINT, on_signal); + std::signal(SIGTERM, on_signal); + for (auto& w : workers) + w->start(); + + // With --once the streams end by themselves; otherwise run until interrupted. + std::atomic finished{ false }; + std::thread waiter( + [&] + { + auto next_stats = std::chrono::steady_clock::now() + std::chrono::seconds(stats_interval_s); + while (!g_stop && !finished) + { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + if (stats_interval_s > 0 && std::chrono::steady_clock::now() >= next_stats) + { + next_stats += std::chrono::seconds(stats_interval_s); + for (const auto& w : workers) + w->print_live(std::cout); + std::cout.flush(); + } + } + if (g_stop) + { + for (auto& w : workers) + w->request_stop(); + } + }); + bool ok = true; + for (auto& w : workers) + ok = w->join() && ok; + finished = true; + waiter.join(); + + for (const auto& w : workers) + w->print_summary(std::cout); + return ok ? 0 : 1; +} +catch (const std::exception& e) +{ + std::cerr << argv[0] << ": " << e.what() << std::endl; + return 1; +} diff --git a/src/video_stage/app/stats.hpp b/src/video_stage/app/stats.hpp new file mode 100644 index 0000000000..1be197b6c3 --- /dev/null +++ b/src/video_stage/app/stats.hpp @@ -0,0 +1,65 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include +#include +#include + +namespace video_stage +{ + +/// Bounded per-stage duration samples, summarised as percentiles at shutdown; nothing is logged per frame. +class DurationStats +{ +public: + explicit DurationStats(size_t capacity = 1U << 16) : m_capacity(capacity) + { + m_samples.reserve(capacity); + } + + void add(int64_t ns) + { + std::lock_guard lock(m_mutex); + if (m_samples.size() < m_capacity) + m_samples.push_back(ns); + else + ++m_overflow; + } + + size_t count() const + { + std::lock_guard lock(m_mutex); + return m_samples.size(); + } + size_t overflow() const + { + return m_overflow; + } + + /// The q-th percentile (0..100) in nanoseconds; 0 when empty. + int64_t percentile(double q) const + { + std::vector sorted; + { + std::lock_guard lock(m_mutex); + sorted = m_samples; + } + if (sorted.empty()) + return 0; + std::sort(sorted.begin(), sorted.end()); + const size_t index = + std::min(sorted.size() - 1, static_cast(q / 100.0 * static_cast(sorted.size()))); + return sorted[index]; + } + +private: + size_t m_capacity; + size_t m_overflow = 0; + mutable std::mutex m_mutex; + std::vector m_samples; +}; + +} // namespace video_stage diff --git a/src/video_stage/app/stream_worker.cpp b/src/video_stage/app/stream_worker.cpp new file mode 100644 index 0000000000..38a371ac95 --- /dev/null +++ b/src/video_stage/app/stream_worker.cpp @@ -0,0 +1,412 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "stream_worker.hpp" + +#include "control_channel.hpp" +#include "tcp_receiver.hpp" + +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace video_stage +{ + +namespace +{ + +// Writes the processed access units as one Annex-B stream, so a run can be checked with ffprobe or a player. +class FileSink final : public PacketSink +{ +public: + explicit FileSink(const std::string& path) : m_out(path, std::ios::binary) + { + if (!m_out) + throw std::runtime_error("cannot open " + path); + } + void write(const EncodedPacket& packet, const FrameMeta&, const Geometry&) override + { + m_out.write(reinterpret_cast(packet.data.data()), static_cast(packet.data.size())); + if (!m_out) + throw std::runtime_error("writing the output file failed"); + } + +private: + std::ofstream m_out; +}; + +// Records waiting for the processing thread. Bounded by count: past that, the stage is not keeping up. +class RecordQueue +{ +public: + struct Item + { + std::vector bytes; + int64_t arrived_ns = 0; + }; + + explicit RecordQueue(size_t capacity) : m_capacity(capacity) + { + } + + /// False when full. + bool push(Item&& item) + { + { + std::lock_guard lock(m_mutex); + if (m_items.size() >= m_capacity) + return false; + m_items.push_back(std::move(item)); + } + m_wake.notify_one(); + return true; + } + + /// Discards everything waiting; returns how many records. + size_t clear() + { + std::lock_guard lock(m_mutex); + const size_t n = m_items.size(); + m_items.clear(); + return n; + } + + void close() + { + { + std::lock_guard lock(m_mutex); + m_closed = true; + } + m_wake.notify_all(); + } + + /// Blocks for the next record; false once closed and empty. + bool pop(Item& out) + { + std::unique_lock lock(m_mutex); + m_wake.wait(lock, [this] { return m_closed || !m_items.empty(); }); + if (m_items.empty()) + return false; + out = std::move(m_items.front()); + m_items.pop_front(); + return true; + } + +private: + size_t m_capacity; + std::mutex m_mutex; + std::condition_variable m_wake; + std::deque m_items; + bool m_closed = false; +}; + +void print_stage(std::ostream& out, const char* name, const DurationStats& s) +{ + char line[160]; + std::snprintf(line, sizeof(line), " %-22s n=%zu p50 %.2f ms p95 %.2f ms p99 %.2f ms\n", name, s.count(), + s.percentile(50) / 1e6, s.percentile(95) / 1e6, s.percentile(99) / 1e6); + out << line; +} + +} // namespace + +struct StreamWorker::Impl +{ + StreamSpec spec; + StreamSettings settings; + SharedLiveKit* livekit; + + std::unique_ptr receiver; + std::atomic active_control{ nullptr }; + std::unique_ptr sink; + LiveKitOutput* livekit_output = nullptr; + + std::thread thread; + std::atomic stop{ false }; + std::atomic ok{ true }; + std::atomic overflows{ 0 }; + std::atomic discarded{ 0 }; + std::atomic sessions{ 0 }; + uint64_t epoch_base = 0; ///< advances for every session so a control channel's epoch is never reused + std::atomic clock_exchanges{ 0 }; + std::atomic clock_rejected{ 0 }; + std::atomic clock_discontinuities{ 0 }; + DurationStats backlog_bytes; + // The picture size from the last sequence header, and a count. + Geometry passthrough_geometry; + std::atomic passthrough_records{ 0 }; + + Impl(const StreamSpec& s, const StreamSettings& st, SharedLiveKit* lk) : spec(s), settings(st), livekit(lk) + { + if (livekit != nullptr) + { + LiveKitOutputConfig out = settings.livekit; + out.track_name = spec.track; + auto output = std::make_unique( + *livekit, out, + [this]() -> std::optional + { + ControlChannel* c = active_control.load(); + return c ? c->mapping() : std::nullopt; + }, + [this] + { + // A subscriber needs a keyframe: only the Orin's encoder can make one. + receiver->request_keyframe(); + }); + livekit_output = output.get(); + sink = std::move(output); + } + else + { + sink = std::make_unique(spec.file); + } + receiver = std::make_unique(spec.port); + } + + // The access unit goes to the sink untouched. The picture size comes from the stream's own sequence header; + // nothing is published before the first one. + void forward(const avs1::Record& record, int64_t arrived_ns) + { + const bool keyframe = avs1::contains_idr(record.payload, record.payload_size); + if (keyframe) + { + if (const auto sps = h264::parse_sps(record.payload, record.payload_size)) + passthrough_geometry = Geometry{ sps->width, sps->height }; + } + if (passthrough_geometry.width == 0) + return; + FrameMeta meta; + meta.frame_number = record.sample.frame_number; + meta.source_timestamp_ns = record.sample.frame_timestamp_ns; + meta.received_ns = arrived_ns; + const uint64_t index = passthrough_records.fetch_add(1); + meta.key = index; + EncodedPacket packet; + packet.data.assign(record.payload, record.payload + record.payload_size); + packet.keyframe = keyframe; + packet.key = index; + sink->write(packet, meta, passthrough_geometry); + } + + // Runs one Orin session to its end. Intake and processing are separate threads, so reading from the socket never + // waits for the GPU stages and the time a record spends waiting is measured. Compressed records are not dropped + // one by one: if the queue overflows, everything waiting is discarded and the stream resumes at the next keyframe, + // which is asked of the sender. A link or decoder error ends the session (reported), after the queued records are + // processed. + bool run_session() + { + constexpr size_t kQueueCapacity = 4; + RecordQueue queue(kQueueCapacity); + std::atomic session_ok{ true }; + std::thread intake( + [&] + { + try + { + bool catching_up = false; + std::vector bytes; + while (receiver->read_record(bytes)) + { + const int64_t arrived_ns = monotonic_ns(); + backlog_bytes.add(static_cast(receiver->pending_bytes())); + std::string error; + const auto record = avs1::parse(bytes.data(), bytes.size(), error); + if (!record) + throw std::runtime_error("bad AVS1 record: " + error); + if (catching_up) + { + if (!avs1::contains_idr(record->payload, record->payload_size)) + { + ++discarded; + continue; + } + catching_up = false; + } + if (!queue.push({ std::move(bytes), arrived_ns })) + { + discarded += queue.clear() + 1; + ++overflows; + catching_up = true; + receiver->request_keyframe(); + } + bytes = {}; + } + } + catch (const std::exception& e) + { + std::cerr << "stream " << spec.port << " intake: " << e.what() << std::endl; + session_ok = false; + } + queue.close(); + }); + + try + { + RecordQueue::Item item; + while (queue.pop(item)) + { + std::string error; + const auto record = avs1::parse(item.bytes.data(), item.bytes.size(), error); + if (!record) + throw std::runtime_error("bad AVS1 record: " + error); + forward(*record, item.arrived_ns); + } + } + catch (const std::exception& e) + { + std::cerr << "stream " << spec.port << " processing: " << e.what() << std::endl; + session_ok = false; + queue.close(); + receiver->shutdown_session(); + } + intake.join(); + return session_ok.load(); + } + + void run() + { + try + { + do + { + std::cout << "stream " << spec.port << ": listening" << std::endl; + receiver->accept_session(); + if (stop) + break; + std::cout << "stream " << spec.port << ": Orin session connected" << std::endl; + ++sessions; + { + // A new session starts clean: nothing of the previous sender's publication or numbering carries + // over. + if (livekit_output) + livekit_output->end_session(); + passthrough_geometry = Geometry{}; + epoch_base += 1'000'000; + ControlChannel control(*receiver, epoch_base + 1); + active_control = &control; + if (!run_session()) + ok = false; + active_control = nullptr; + if (livekit_output) + livekit_output->end_session(); + clock_exchanges += control.exchanges(); + clock_rejected += control.rejected(); + clock_discontinuities += control.discontinuities(); + } + std::cout << "stream " << spec.port << ": Orin session ended" << std::endl; + } while (!settings.once && !stop); + } + catch (const std::exception& e) + { + if (!stop) + { + std::cerr << "stream " << spec.port << ": " << e.what() << std::endl; + ok = false; + } + } + } +}; + +StreamWorker::StreamWorker(const StreamSpec& spec, const StreamSettings& settings, SharedLiveKit* livekit) + : m_impl(std::make_unique(spec, settings, livekit)) +{ +} + +StreamWorker::~StreamWorker() +{ + request_stop(); + join(); +} + +void StreamWorker::start() +{ + m_impl->thread = std::thread([this] { m_impl->run(); }); +} + +void StreamWorker::request_stop() +{ + m_impl->stop = true; + m_impl->receiver->stop_listening(); + m_impl->receiver->shutdown_session(); +} + +bool StreamWorker::join() +{ + if (m_impl->thread.joinable()) + m_impl->thread.join(); + return m_impl->ok.load(); +} + +void StreamWorker::print_live(std::ostream& out) const +{ + const Impl& d = *m_impl; + char line[512]; + int n = std::snprintf(line, sizeof(line), "[%u] backlog p99 %lld B", static_cast(d.spec.port), + static_cast(d.backlog_bytes.percentile(99))); + if (d.livekit_output && n > 0 && n < static_cast(sizeof(line))) + { + std::snprintf( + line + n, sizeof(line) - static_cast(n), + " | intake->write p50 %.1f p99 %.1f ms | submit p50 %.1f p99 %.1f ms | Orin->PC p50 %.1f p99 %.1f ms | Orin->publish p50 %.1f p99 %.1f ms | mapping +-%.1f ms | published %llu", + d.livekit_output->intake_to_write_ns().percentile(50) / 1e6, + d.livekit_output->intake_to_write_ns().percentile(99) / 1e6, + d.livekit_output->submit_ns().percentile(50) / 1e6, d.livekit_output->submit_ns().percentile(99) / 1e6, + d.livekit_output->uplink_ns().percentile(50) / 1e6, d.livekit_output->uplink_ns().percentile(99) / 1e6, + d.livekit_output->source_to_publish_ns().percentile(50) / 1e6, + d.livekit_output->source_to_publish_ns().percentile(99) / 1e6, + d.livekit_output->upstream_uncertainty_ns() / 1e6, + static_cast(d.livekit_output->published())); + } + out << line << "\n"; +} + +void StreamWorker::print_summary(std::ostream& out) const +{ + const Impl& d = *m_impl; + char line[256]; + std::snprintf( + line, sizeof(line), + "stream %u: sessions %llu, records %llu\n" + " overflows %llu, records discarded %llu, clock exchanges %llu (rejected %llu, discontinuities %llu)\n", + static_cast(d.spec.port), static_cast(d.sessions.load()), + static_cast(d.passthrough_records.load()), + static_cast(d.overflows.load()), static_cast(d.discarded.load()), + static_cast(d.clock_exchanges.load()), + static_cast(d.clock_rejected.load()), + static_cast(d.clock_discontinuities.load())); + out << line; + if (d.livekit_output) + { + std::snprintf( + line, sizeof(line), + " published %llu (frames sent without a capture time: no source stamp %llu, unmapped %llu, over budget " + "%llu), refused %llu, republished %llu\n", + static_cast(d.livekit_output->published()), + static_cast(d.livekit_output->no_source_stamp()), + static_cast(d.livekit_output->unmapped()), + static_cast(d.livekit_output->over_budget()), + static_cast(d.livekit_output->refused()), + static_cast(d.livekit_output->republished())); + out << line; + } + std::snprintf(line, sizeof(line), " socket backlog p50 %lld B p99 %lld B\n", + static_cast(d.backlog_bytes.percentile(50)), + static_cast(d.backlog_bytes.percentile(99))); + out << line; + if (d.livekit_output) + { + print_stage(out, "Orin -> PC arrival", d.livekit_output->uplink_ns()); + print_stage(out, "Orin -> publish", d.livekit_output->source_to_publish_ns()); + } +} + +} // namespace video_stage diff --git a/src/video_stage/app/stream_worker.hpp b/src/video_stage/app/stream_worker.hpp new file mode 100644 index 0000000000..afad93c4fa --- /dev/null +++ b/src/video_stage/app/stream_worker.hpp @@ -0,0 +1,63 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include "livekit_output.hpp" + +#include +#include +#include +#include +#include + +namespace video_stage +{ + +/// One camera's stream: where the Orin connects, and where the processed video goes. +struct StreamSpec +{ + uint16_t port = 0; + /// LiveKit track name; used when a LiveKit session is given. + std::string track; + /// Annex-B output file; used instead of LiveKit when set (for checking a run). + std::string file; +}; + +struct StreamSettings +{ + LiveKitOutputConfig livekit; + bool once = false; ///< serve one Orin session, then finish +}; + +/** + * @brief One stream's thread: accept an Orin session, run it to its end, accept the next. + * + * Owns that stream's receiver, control channel and output. When publishing, streams share one LiveKit participant and + * otherwise run independently, so one camera's trouble stays its own. The Orin's access units go on as they are. + */ +class StreamWorker +{ +public: + StreamWorker(const StreamSpec& spec, const StreamSettings& settings, SharedLiveKit* livekit); + ~StreamWorker(); + + StreamWorker(const StreamWorker&) = delete; + StreamWorker& operator=(const StreamWorker&) = delete; + + void start(); + /// Ends the wait for an Orin and shuts down any running session. + void request_stop(); + /// Waits for the thread; false if any session of this stream failed. + bool join(); + + void print_summary(std::ostream& out) const; + /// One compact line of the current percentiles; safe to call while the stream runs. + void print_live(std::ostream& out) const; + +private: + struct Impl; + std::unique_ptr m_impl; +}; + +} // namespace video_stage diff --git a/src/video_stage/app/tcp_receiver.cpp b/src/video_stage/app/tcp_receiver.cpp new file mode 100644 index 0000000000..91ab361d4d --- /dev/null +++ b/src/video_stage/app/tcp_receiver.cpp @@ -0,0 +1,241 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "tcp_receiver.hpp" + +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +namespace video_stage +{ + +namespace +{ + +// Reads exactly `size` bytes. Returns false only for end of stream before the first byte when `allow_eof`. +bool read_exact(int fd, uint8_t* out, size_t size, bool allow_eof) +{ + size_t done = 0; + while (done < size) + { + const ssize_t n = ::recv(fd, out + done, size - done, 0); + if (n == 0) + { + if (allow_eof && done == 0) + return false; + throw std::runtime_error("TcpReceiver: the connection ended inside a record"); + } + if (n < 0) + { + if (errno == EINTR) + continue; + // A peer that exits with unread data resets the connection: at a record boundary that is the end of the + // stream. + if ((errno == ECONNRESET || errno == EPIPE) && allow_eof && done == 0) + return false; + throw std::runtime_error(std::string("TcpReceiver: recv failed: ") + std::strerror(errno)); + } + done += static_cast(n); + } + return true; +} + +} // namespace + +TcpReceiver::TcpReceiver(uint16_t port) +{ + m_listen = ::socket(AF_INET, SOCK_STREAM, 0); + if (m_listen < 0) + throw std::runtime_error("TcpReceiver: cannot create a socket"); + const int one = 1; + ::setsockopt(m_listen, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)); + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_ANY); + addr.sin_port = htons(port); + if (::bind(m_listen, reinterpret_cast(&addr), sizeof(addr)) < 0 || ::listen(m_listen, 1) < 0) + { + const std::string why = std::strerror(errno); + ::close(m_listen); + throw std::runtime_error("TcpReceiver: cannot listen on port " + std::to_string(port) + ": " + why); + } +} + +int TcpReceiver::connection_fd() const +{ + std::lock_guard lock(m_fd_mutex); + return m_connection; +} + +int TcpReceiver::control_fd() const +{ + std::lock_guard lock(m_fd_mutex); + return m_control; +} + +TcpReceiver::~TcpReceiver() +{ + if (m_connection >= 0) + ::close(m_connection); + if (m_control >= 0) + ::close(m_control); + if (m_listen >= 0) + ::close(m_listen); +} + +void TcpReceiver::accept_session() +{ + { + // Retire the earlier session under the lock a concurrent shutdown_session() takes, so it never sees a + // descriptor that is being closed. + std::lock_guard lock(m_fd_mutex); + for (int* fd : { &m_connection, &m_control }) + { + if (*fd >= 0) + ::close(*fd); + *fd = -1; + } + } + // Only this thread publishes descriptors, so it may read them here without the lock. + while (m_connection < 0 || m_control < 0) + { + const int fd = ::accept(m_listen, nullptr, nullptr); + if (fd < 0) + throw std::runtime_error(std::string("TcpReceiver: accept failed: ") + std::strerror(errno)); + const int one = 1; + ::setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)); + // Notice a link that died (Wi-Fi roam, unplugged cable) in seconds, as the sender does. + ::setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &one, sizeof(one)); + const int idle = 5, interval = 2, count = 3; + ::setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, &idle, sizeof(idle)); + ::setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &interval, sizeof(interval)); + ::setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &count, sizeof(count)); + const unsigned user_timeout_ms = 5000; + ::setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, &user_timeout_ms, sizeof(user_timeout_ms)); + // A peer that connects and never finishes its hello must not hold the worker: bound the wait, and let a stop + // reach this socket through m_pending. + timeval timeout{ 2, 0 }; + ::setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); + { + std::lock_guard lock(m_fd_mutex); + m_pending = fd; + } + uint8_t hello[4]; + bool have_hello = false; + try + { + have_hello = read_exact(fd, hello, sizeof(hello), true); + } + catch (const std::exception&) + { + have_hello = false; + } + timeval none{ 0, 0 }; + ::setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &none, sizeof(none)); + { + // Hand the socket over (or close it) in the same critical section that stops advertising it as pending. + std::lock_guard lock(m_fd_mutex); + m_pending = -1; + if (have_hello && std::memcmp(hello, "DATA", 4) == 0 && m_connection < 0) + m_connection = fd; + else if (have_hello && std::memcmp(hello, "CTRL", 4) == 0 && m_control < 0) + m_control = fd; + else + ::close(fd); // no hello, an unknown one, or a duplicate + } + } +} + +void TcpReceiver::stop_listening() +{ + if (m_listen >= 0) + ::shutdown(m_listen, SHUT_RDWR); // makes a blocked accept() fail + std::lock_guard lock(m_fd_mutex); + if (m_pending >= 0) + ::shutdown(m_pending, SHUT_RDWR); // and a hello being read +} + +void TcpReceiver::shutdown_session() +{ + std::lock_guard lock(m_fd_mutex); + if (m_pending >= 0) + ::shutdown(m_pending, SHUT_RDWR); + for (const int fd : { m_connection, m_control }) + { + if (fd >= 0) + ::shutdown(fd, SHUT_RDWR); + } +} + +bool TcpReceiver::send_control(const uint8_t* data, size_t size) +{ + std::lock_guard lock(m_control_send); + const int fd = control_fd(); + size_t done = 0; + while (done < size) + { + const ssize_t n = ::send(fd, data + done, size - done, MSG_NOSIGNAL); + if (n < 0) + { + if (errno == EINTR) + continue; + return false; + } + done += static_cast(n); + } + return true; +} + +bool TcpReceiver::recv_control(uint8_t* out, size_t size) +{ + try + { + return read_exact(control_fd(), out, size, true); + } + catch (const std::exception&) + { + return false; + } +} + +size_t TcpReceiver::pending_bytes() const +{ + int pending = 0; + const int fd = connection_fd(); + if (fd < 0 || ::ioctl(fd, FIONREAD, &pending) < 0) + return 0; + return static_cast(pending); +} + +void TcpReceiver::request_keyframe() +{ + // Best effort: if the link is down the sender asks for a keyframe itself when it reconnects. + const uint8_t request = 'K'; + (void)send_control(&request, 1); +} + +bool TcpReceiver::read_record(std::vector& record) +{ + const int fd = connection_fd(); + uint8_t prefix[4]; + if (!read_exact(fd, prefix, 4, true)) + return false; + const uint32_t length = avs1::read_length(prefix); + if (length == 0 || length > avs1::kMaxRecordBytes) + throw std::runtime_error("TcpReceiver: invalid record length " + std::to_string(length)); + record.resize(length); + read_exact(fd, record.data(), length, false); + return true; +} + +} // namespace video_stage diff --git a/src/video_stage/app/tcp_receiver.hpp b/src/video_stage/app/tcp_receiver.hpp new file mode 100644 index 0000000000..e6abeed772 --- /dev/null +++ b/src/video_stage/app/tcp_receiver.hpp @@ -0,0 +1,65 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include +#include + +namespace video_stage +{ + +/// One listening TCP socket and one accepted connection that delivers length-prefixed AVS1 records. +class TcpReceiver +{ +public: + /// Listens on `port` (all interfaces). Throws std::runtime_error on failure. + explicit TcpReceiver(uint16_t port); + ~TcpReceiver(); + + TcpReceiver(const TcpReceiver&) = delete; + TcpReceiver& operator=(const TcpReceiver&) = delete; + + /// Blocks until the sender has opened both connections, each announced by a 4-byte hello: "DATA" carries the + /// records and "CTRL" the small control messages (keyframe requests and the clock exchange). TCP_NODELAY on both. + /// Any earlier session is closed: call this only after shutdown_session() and after the threads that read or + /// write the earlier session have stopped, since they use a copy of its descriptors. + void accept_session(); + + /// Ends a blocked accept_session() (it then throws or returns without a session) and stops accepting. + void stop_listening(); + + /// Unblocks every thread waiting in a read or write on this session. + void shutdown_session(); + + /// Control link: write all of `data` / read exactly `size` bytes. send_control is safe from several threads; + /// recv_control returns false when the link ended. + bool send_control(const uint8_t* data, size_t size); + bool recv_control(uint8_t* out, size_t size); + + /// Bytes already received and waiting in the socket (how far behind the reader is). + size_t pending_bytes() const; + + /// Asks the sender for a keyframe: one byte back on the connection. Safe to call from another thread. + void request_keyframe(); + + /// Reads one record's bytes (after the length prefix) into `record`. False on a clean end of stream; throws on a + /// truncated or oversized record. + bool read_record(std::vector& record); + +private: + /// The session descriptors have one owner at a time: this lock covers publishing a new one, retiring the old and + /// shutting them down from another thread. Blocking reads and writes happen outside it, on a copy. + int connection_fd() const; + int control_fd() const; + + int m_listen = -1; + mutable std::mutex m_fd_mutex; + int m_connection = -1; ///< DATA; guarded by m_fd_mutex + int m_control = -1; ///< CTRL; guarded by m_fd_mutex + int m_pending = -1; ///< an accepted socket whose hello is being read, visible to shutdown; guarded by m_fd_mutex + std::mutex m_control_send; +}; + +} // namespace video_stage diff --git a/src/video_stage/tests/CMakeLists.txt b/src/video_stage/tests/CMakeLists.txt new file mode 100644 index 0000000000..272ec56770 --- /dev/null +++ b/src/video_stage/tests/CMakeLists.txt @@ -0,0 +1,13 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# The relay's receiver compiled into its own test, so the test links no LiveKit. +add_executable(video_stage_tcp_receiver_tests + test_tcp_receiver.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/../app/tcp_receiver.cpp +) + +target_include_directories(video_stage_tcp_receiver_tests PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../app) +target_link_libraries(video_stage_tcp_receiver_tests PRIVATE video_stage::wire Catch2::Catch2WithMain) + +catch_discover_tests(video_stage_tcp_receiver_tests) diff --git a/src/video_stage/tests/test_tcp_receiver.cpp b/src/video_stage/tests/test_tcp_receiver.cpp new file mode 100644 index 0000000000..8c4fe9d8ed --- /dev/null +++ b/src/video_stage/tests/test_tcp_receiver.cpp @@ -0,0 +1,210 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace std::chrono_literals; +using video_stage::TcpReceiver; + +namespace +{ + +constexpr uint16_t kStopPort = 5741; +constexpr uint16_t kRacePort = 5742; +constexpr uint16_t kSessionPort = 5743; + +int connect_to(uint16_t port) +{ + const int fd = ::socket(AF_INET, SOCK_STREAM, 0); + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_port = htons(port); + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + if (fd < 0 || ::connect(fd, reinterpret_cast(&addr), sizeof(addr)) != 0) + { + if (fd >= 0) + ::close(fd); + return -1; + } + return fd; +} + +/// Opens a DATA and a CTRL connection with their hellos; the descriptors are closed on destruction. +struct Sender +{ + explicit Sender(uint16_t port) + { + data = connect_to(port); + if (data >= 0) + (void)::send(data, "DATA", 4, MSG_NOSIGNAL); + control = connect_to(port); + if (control >= 0) + (void)::send(control, "CTRL", 4, MSG_NOSIGNAL); + } + ~Sender() + { + for (const int fd : { data, control }) + if (fd >= 0) + ::close(fd); + } + Sender(const Sender&) = delete; + Sender& operator=(const Sender&) = delete; + int data = -1; + int control = -1; +}; + +} // namespace + +TEST_CASE("a session is accepted and carries a record and control bytes", "[tcp_receiver]") +{ + TcpReceiver receiver(kSessionPort); + Sender sender(kSessionPort); + REQUIRE(sender.data >= 0); + REQUIRE(sender.control >= 0); + receiver.accept_session(); + + const uint8_t record[] = { 0, 0, 0, 3, 'a', 'b', 'c' }; // 4-byte length prefix, then the record + REQUIRE(::send(sender.data, record, sizeof(record), MSG_NOSIGNAL) == static_cast(sizeof(record))); + std::vector out; + REQUIRE(receiver.read_record(out)); + CHECK(out == std::vector{ 'a', 'b', 'c' }); + + const uint8_t ping[] = { 7 }; + REQUIRE(receiver.send_control(ping, 1)); + uint8_t got = 0; + REQUIRE(::recv(sender.control, &got, 1, 0) == 1); + CHECK(got == 7); + receiver.shutdown_session(); + CHECK_FALSE(receiver.read_record(out)); // a shut-down session ends cleanly +} + +TEST_CASE("stop_listening ends a blocked accept_session", "[tcp_receiver]") +{ + TcpReceiver receiver(kStopPort); + auto accepted = std::async(std::launch::async, + [&] + { + try + { + receiver.accept_session(); + return false; + } + catch (const std::exception&) + { + return true; + } + }); + std::this_thread::sleep_for(200ms); + REQUIRE(accepted.wait_for(0ms) == std::future_status::timeout); // blocked, with no sender + receiver.stop_listening(); + REQUIRE(accepted.wait_for(3s) == std::future_status::ready); + CHECK(accepted.get()); // it threw instead of returning a session +} + +TEST_CASE("stop_listening ends an accept blocked on a silent connection", "[tcp_receiver]") +{ + TcpReceiver receiver(kStopPort + 10); + const int silent = connect_to(kStopPort + 10); // connects, never sends its hello + REQUIRE(silent >= 0); + auto accepted = std::async(std::launch::async, + [&] + { + try + { + receiver.accept_session(); + } + catch (const std::exception&) + { + } + }); + std::this_thread::sleep_for(300ms); // the hello read is now pending (bounded at 2 s) + const auto begin = std::chrono::steady_clock::now(); + receiver.stop_listening(); + REQUIRE(accepted.wait_for(1500ms) == std::future_status::ready); + CHECK(std::chrono::steady_clock::now() - begin < 1500ms); + ::close(silent); +} + +TEST_CASE("shutdown_session from another thread while accept_session replaces descriptors", "[tcp_receiver][race]") +{ + TcpReceiver receiver(kRacePort); + + for (int round = 0; round < 200; ++round) + { + std::atomic done{ false }; + + std::thread accepting( + [&] + { + try + { + receiver.accept_session(); + } + catch (const std::exception&) + { + } + done = true; + }); + + // A sender that keeps offering a DATA/CTRL pair until a session is up; a shutdown may eat a hello in flight. + std::thread client( + [&] + { + while (!done) + { + Sender sender(kRacePort); + for (int i = 0; i < 20 && !done; ++i) + std::this_thread::sleep_for(1ms); + } + }); + + std::thread shutting( + [&] + { + while (!done) + { + receiver.shutdown_session(); + std::this_thread::yield(); + } + }); + + // Users of the session on other threads: none may touch a descriptor that is being replaced. + std::thread user( + [&] + { + uint8_t byte = 0; + std::vector record; + while (!done) + { + (void)receiver.pending_bytes(); + (void)receiver.send_control(&byte, 1); + std::this_thread::yield(); + } + }); + + const auto deadline = std::chrono::steady_clock::now() + 15s; + while (!done && std::chrono::steady_clock::now() < deadline) + std::this_thread::sleep_for(1ms); + if (!done) + receiver.stop_listening(); // do not hang the suite if a session never completes + accepting.join(); + client.join(); + shutting.join(); + user.join(); + REQUIRE(std::chrono::steady_clock::now() < deadline); + receiver.shutdown_session(); // end any reader before the next round retires the descriptors + } +} diff --git a/src/video_stage/tests/tools/mock_orin_sender.py b/src/video_stage/tests/tools/mock_orin_sender.py new file mode 100755 index 0000000000..aca8de5689 --- /dev/null +++ b/src/video_stage/tests/tools/mock_orin_sender.py @@ -0,0 +1,471 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""A mock of the Orin producer's direct sender, behaving as the real one does. + +It mirrors DirectSender in src/plugins/sensing/core/direct_sender.cpp and what feeds it: + - the stream: H.264 as the Jetson encoder emits it (no AUD, SPS+PPS before every IDR, no B-frames, constant bitrate, + an IDR every fps*5 frames), made once with ffmpeg if the source file is missing; + - frames paced at the publication rate, with source frame ids that step by 2 (60 fps capture, every second capture + published) and a stamp on the Orin's own monotonic clock (offset from this machine's, optionally drifting), as the + frame-ready time of each unit; + - two TCP connections opened by the Orin side, "DATA" then "CTRL" (4-byte hellos), retried every second; on every + (re)connect the queue is cleared and the stream starts at a keyframe; + - a bounded queue: a unit that does not fit is dropped, and so is everything after it until the next keyframe; + - a control thread blocked in recv: 'K' makes the encoder produce a keyframe at the next unit, 'Q' is answered with 'R' + with t2 stamped when the first byte arrives and t3 just before the reply is sent, on the Orin's clock. +One sender per --port, all in one process on one clock, like the producer's cameras. Runs until stopped or --duration. +""" + +from __future__ import annotations + +import argparse +import collections +import socket +import struct +import subprocess +import sys +import threading +import time +from pathlib import Path + +HEADER = struct.Struct("!4sHHIHHQQQQIIQ") +STREAM = struct.Struct("!HHBBHIIIIQHHI") +SAMPLE = struct.Struct("!HBBHHIIQQQQQQII") +CODEC_H264 = 2 +BITRATE = "8M" # the producer's default target + + +# ---------------------------------------------------------------------------------------------------------------------- +# The stream +# ---------------------------------------------------------------------------------------------------------------------- + + +def make_stream( + path: Path, width: int, height: int, fps: int, seconds: int = 12 +) -> None: + """Encode a test pattern the way the Jetson encoder does: IDR period fps*5, no B-frames, CBR, no AUDs.""" + gop = fps * 5 + cmd = [ + "ffmpeg", "-y", "-loglevel", "error", "-f", "lavfi", "-i", f"testsrc2=size={width}x{height}:rate={fps}", + "-t", str(seconds), "-c:v", "libx264", "-preset", "veryfast", "-profile:v", "high", "-bf", "0", "-g", str(gop), + "-x264opts", f"keyint={gop}:min-keyint={gop}:scenecut=0:aud=0:repeat-headers=1:nal-hrd=cbr", + "-b:v", BITRATE, "-maxrate", BITRATE, "-bufsize", "267k", "-pix_fmt", "yuv420p", "-f", "h264", str(path), + ] # fmt: skip + subprocess.run(cmd, check=True) + + +_TRAILING_NALS = (10, 11, 12) + + +def _nals(data: bytes) -> list[tuple[int, int, int]]: + """(offset of the start code, NAL type, offset of the NAL header byte) for every NAL unit.""" + out = [] + i = 0 + while True: + j = data.find(b"\x00\x00\x01", i) + if j < 0 or j + 3 >= len(data): + break + start = j - 1 if j > 0 and data[j - 1] == 0 else j + out.append((start, data[j + 3] & 0x1F, j + 3)) + i = j + 3 + return out + + +def _first_mb_in_slice(data: bytes, header_at: int) -> int: + """The slice's first_mb_in_slice, an Exp-Golomb code right after the NAL header byte.""" + bits = int.from_bytes(data[header_at + 1 : header_at + 5], "big") + zeros = 0 + while zeros < 15 and not (bits >> (31 - zeros)) & 1: + zeros += 1 + code = (bits >> (32 - (2 * zeros + 1))) & ((1 << (2 * zeros + 1)) - 1) + return code - 1 + + +def split_access_units(data: bytes) -> list[tuple[bytes, bool]]: + """Split Annex-B data into (access unit, is IDR). A unit begins at the first NAL after a picture's slices, or at a + picture whose first slice follows another picture directly; later slices of one picture stay with it.""" + starts = [] + previous_vcl = False + for position, (start, nal_type, header) in enumerate(_nals(data)): + vcl = nal_type in (1, 5) + if position == 0: + begin = True + elif vcl: + begin = previous_vcl and _first_mb_in_slice(data, header) == 0 + else: + begin = previous_vcl and nal_type not in _TRAILING_NALS + if begin: + starts.append(start) + # End-of-sequence, end-of-stream and filler data (constant-bitrate padding) belong to the picture before them. + previous_vcl = vcl or (previous_vcl and nal_type in _TRAILING_NALS) + starts.append(len(data)) + units = [] + for a, b in zip(starts, starts[1:]): + au = data[a:b] + units.append((au, any(t == 5 for _, t, _ in _nals(au)))) + return units + + +def record( + au: bytes, + sequence: int, + frame: int, + stamp_ns: int, + width: int, + height: int, + fps: int, +) -> bytes: + descriptor_bytes = STREAM.size + SAMPLE.size + header = HEADER.pack( + b"AVS1", + 1, + 0, + HEADER.size, + 1, + 1, + sequence, + 1, + 0, + 0, + descriptor_bytes, + len(au), + 0, + ) + stream = STREAM.pack( + 0, 1, CODEC_H264, 1, 0, width, height, fps, 1, sequence, 0, 1, 0 + ) + sample = SAMPLE.pack( + 0, 0, 0, 0x0005, 0, 0, 0, frame, 0, stamp_ns, 0, 0, 0, 0, len(au) + ) + body = header + stream + sample + au + return struct.pack("!I", len(body)) + body + + +# ---------------------------------------------------------------------------------------------------------------------- +# The Orin's clock +# ---------------------------------------------------------------------------------------------------------------------- + + +class OrinClock: + """Monotonic time on a machine with another boot time, optionally running a little fast or slow.""" + + def __init__(self, offset_ns: int, drift_ppm: float) -> None: + self._t0 = time.monotonic_ns() + self._offset = offset_ns + self._drift = drift_ppm * 1e-6 + + def now_ns(self) -> int: + local = time.monotonic_ns() + return local + self._offset + int((local - self._t0) * self._drift) + + +# ---------------------------------------------------------------------------------------------------------------------- +# DirectSender (mirrors direct_sender.cpp) +# ---------------------------------------------------------------------------------------------------------------------- + + +class DirectSender: + def __init__( + self, + host: str, + port: int, + width: int, + height: int, + fps: int, + clock: OrinClock, + max_queued_bytes: int, + ): + self.host, self.port, self.width, self.height, self.fps = ( + host, + port, + width, + height, + fps, + ) + self.clock = clock + self.max_queued_bytes = max_queued_bytes + self._cond = threading.Condition() + self._queue: collections.deque[bytes] = collections.deque() + self._queued_bytes = 0 + self._need_keyframe = True + self._stop = False + self._sequence = 0 + self._keyframe_requested = False + self.sent = 0 + self.dropped = 0 + self.keyframe_requests = 0 + self.clock_answers = 0 + self.connects = 0 + self._live: list[socket.socket] = [] + self._thread = threading.Thread(target=self._run, daemon=True) + self._thread.start() + + def stop(self) -> None: + """Cancels any connect or send in progress and waits for the worker thread; a stuck thread is an error.""" + with self._cond: + self._stop = True + self._cond.notify_all() + live = list(self._live) + for s in live: + try: + s.shutdown(socket.SHUT_RDWR) + except OSError: + pass + self._thread.join(timeout=10) + if self._thread.is_alive(): + raise RuntimeError(f"sender for port {self.port} did not stop") + + def take_keyframe_request(self) -> bool: + with self._cond: + requested, self._keyframe_requested = self._keyframe_requested, False + return requested + + def submit(self, au: bytes, keyframe: bool, frame_id: int, stamp_ns: int) -> None: + with self._cond: + if self._need_keyframe: + if not keyframe: + return # a delta frame cannot start a stream; the keyframe request is pending + self._need_keyframe = False + if self._queued_bytes + len(au) > self.max_queued_bytes: + # Dropping any frame breaks the chain after it: discard up to the next keyframe, and ask for one. + self.dropped += 1 + self._need_keyframe = True + self._keyframe_requested = True + return + self._sequence += 1 + rec = record( + au, + self._sequence, + frame_id, + stamp_ns, + self.width, + self.height, + self.fps, + ) + self._queued_bytes += len(rec) + self._queue.append(rec) + self._cond.notify() + + # --- connection -------------------------------------------------------------------------------------------------- + + def _connect(self) -> tuple[socket.socket, socket.socket] | None: + socks = [] + try: + for hello in (b"DATA", b"CTRL"): + s = socket.create_connection((self.host, self.port), timeout=3) + with self._cond: + if self._stop: + s.close() + raise OSError("stopping") + self._live.append(s) + s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + s.settimeout(None) + s.sendall(hello) + socks.append(s) + except OSError: + for s in socks: + s.close() + with self._cond: + self._live = [] + return None + return socks[0], socks[1] + + def _control(self, ctrl: socket.socket, state: dict) -> None: + """Blocks in recv so t2 is the arrival of a request, not the next time the sender looks at the socket.""" + try: + while True: + first = ctrl.recv(1) + if not first: + break + t2 = self.clock.now_ns() + if first == b"K": + with self._cond: + self.keyframe_requests += 1 + self._keyframe_requested = True + elif first == b"Q": + rest = b"" + while len(rest) < 12: + chunk = ctrl.recv(12 - len(rest)) + if not chunk: + raise OSError("closed inside a request") + rest += chunk + seq, t1 = struct.unpack(" None: + while True: + with self._cond: + if self._stop: + return + pair = self._connect() + if pair is None: + with self._cond: + if self._cond.wait_for(lambda: self._stop, timeout=1.0): + return + continue + data, ctrl = pair + self.connects += 1 + state = {"control_ended": False} + control = threading.Thread( + target=self._control, args=(ctrl, state), daemon=True + ) + control.start() + with self._cond: + # The PC needs a keyframe first: drop anything queued before the connection and ask for one. + self._queue.clear() + self._queued_bytes = 0 + self._need_keyframe = True + self._keyframe_requested = True + try: + while True: + with self._cond: + self._cond.wait_for( + lambda: self._stop or self._queue, timeout=0.2 + ) + if self._stop: + return + if not self._queue: + if state["control_ended"]: + break + continue + rec = self._queue.popleft() + self._queued_bytes -= len(rec) + data.sendall(rec) + self.sent += 1 + except OSError: + pass + finally: + for s in (ctrl, data): + try: + s.shutdown(socket.SHUT_RDWR) + except OSError: + pass + s.close() + with self._cond: + self._live = [] + control.join(timeout=2) + + +# ---------------------------------------------------------------------------------------------------------------------- + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) + ap.add_argument( + "--host", default="127.0.0.1", help="the PC running video_stage_worker" + ) + ap.add_argument( + "--port", + type=int, + action="append", + required=True, + help="one per camera (repeat)", + ) + ap.add_argument("--width", type=int, default=2560) + ap.add_argument("--height", type=int, default=1984) + ap.add_argument("--fps", type=int, default=30, help="publication rate") + ap.add_argument( + "--source", + default="orin_like.h264", + help="H.264 file; made with ffmpeg if it does not exist", + ) + ap.add_argument( + "--frame-id-step", + type=int, + default=2, + help="source frame ids advance by this per unit", + ) + ap.add_argument( + "--clock-offset-ms", + type=float, + default=5000.0, + help="this clock's boot-time offset from the PC's", + ) + ap.add_argument("--clock-drift-ppm", type=float, default=0.0) + ap.add_argument("--max-queued-mb", type=float, default=16.0) + ap.add_argument( + "--duration", + type=float, + default=0.0, + help="seconds to run; 0 means until stopped", + ) + args = ap.parse_args() + + source = Path(args.source) + if not source.exists(): + print( + f"making {source} ({args.width}x{args.height}@{args.fps}, Jetson-style H.264)", + flush=True, + ) + make_stream(source, args.width, args.height, args.fps) + units = split_access_units(source.read_bytes()) + keyframes = [i for i, (_, key) in enumerate(units) if key] + if not keyframes: + sys.exit(f"{source} has no IDR access units") + + clock = OrinClock(int(args.clock_offset_ms * 1e6), args.clock_drift_ppm) + senders = [ + DirectSender( + args.host, + p, + args.width, + args.height, + args.fps, + clock, + int(args.max_queued_mb * (1 << 20)), + ) + for p in args.port + ] + period = 1.0 / args.fps + cursors = [0] * len(senders) + frame_id = 0 + started = time.monotonic() + next_tick = started + next_report = started + 5 + try: + while args.duration <= 0 or time.monotonic() - started < args.duration: + delay = next_tick - time.monotonic() + if delay > 0: + time.sleep(delay) + next_tick += period + frame_id += args.frame_id_step + stamp = clock.now_ns() + for n, s in enumerate(senders): + # A keyframe request makes that camera's encoder emit an IDR; the file's next keyframe stands in for it. + if s.take_keyframe_request(): + cursors[n] = next( + (k for k in keyframes if k >= cursors[n]), keyframes[0] + ) + au, key = units[cursors[n] % len(units)] + cursors[n] = (cursors[n] + 1) % len(units) + s.submit(au, key, frame_id, stamp) + if time.monotonic() >= next_report: + next_report += 5 + print( + " | ".join( + f"port {s.port}: sent {s.sent} dropped {s.dropped} K {s.keyframe_requests} Q {s.clock_answers}" + f" connects {s.connects}" + for s in senders + ), + flush=True, + ) + except KeyboardInterrupt: + pass + finally: + for s in senders: + s.stop() + for s in senders: + print( + f"port {s.port}: sent {s.sent}, dropped {s.dropped}, keyframe requests {s.keyframe_requests}, " + f"clock answers {s.clock_answers}, connections {s.connects}" + ) + + +if __name__ == "__main__": + main() diff --git a/src/video_stage/tests/tools/test_mock_orin_sender.py b/src/video_stage/tests/tools/test_mock_orin_sender.py new file mode 100644 index 0000000000..408a162233 --- /dev/null +++ b/src/video_stage/tests/tools/test_mock_orin_sender.py @@ -0,0 +1,503 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Tests for mock_orin_sender.py. No GPU needed; the stream tests need ffmpeg. + +Run: python3 -m unittest discover -s src/video_stage/tests/tools -p 'test_*.py' +""" + +from __future__ import annotations + +import shutil +import socket +import struct +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import mock_orin_sender as m # noqa: E402 + +START = b"\x00\x00\x00\x01" + + +def nal(nal_type: int, payload: bytes = b"\x80") -> bytes: + return START + bytes([0x60 | nal_type]) + payload + + +# first_mb_in_slice ue(v) codes: 0 -> '1', 1 -> '010', 2 -> '011', 3 -> '00100' +def slice_nal(nal_type: int, first_mb: int) -> bytes: + code = {0: 0b1, 1: 0b010, 2: 0b011, 3: 0b00100}[first_mb] + width = {0: 1, 1: 3, 2: 3, 3: 5}[first_mb] + value = (code << (8 - width)) & 0xFF + return START + bytes([0x60 | nal_type, value, 0xAA, 0xBB, 0xCC, 0xDD]) + + +class SplitterTest(unittest.TestCase): + def test_first_mb_in_slice_decoding(self) -> None: + for mb in (0, 1, 2, 3): + data = slice_nal(1, mb) + self.assertEqual(m._first_mb_in_slice(data, 4), mb) + + def test_aud_less_stream_splits_at_parameter_sets_and_slices(self) -> None: + idr = nal(7) + nal(8) + slice_nal(5, 0) + p1 = slice_nal(1, 0) + p2 = slice_nal(1, 0) + data = idr + p1 + p2 + idr + units = m.split_access_units(data) + self.assertEqual([u for u, _ in units], [idr, p1, p2, idr]) + self.assertEqual([k for _, k in units], [True, False, False, True]) + self.assertEqual(b"".join(u for u, _ in units), data) + + def test_filler_attaches_to_the_preceding_unit(self) -> None: + p1 = slice_nal(1, 0) + nal(12, b"\xff\xff") + p2 = slice_nal(1, 0) + units = m.split_access_units(p1 + p2) + self.assertEqual([u for u, _ in units], [p1, p2]) + + def test_multi_slice_picture_is_one_unit(self) -> None: + pic = slice_nal(1, 0) + slice_nal(1, 1) + slice_nal(1, 2) + nxt = slice_nal(1, 0) + units = m.split_access_units(pic + nxt) + self.assertEqual([u for u, _ in units], [pic, nxt]) + + def test_sei_before_a_picture_starts_a_new_unit(self) -> None: + a = slice_nal(1, 0) + b = nal(6) + slice_nal(1, 0) + units = m.split_access_units(a + b) + self.assertEqual([u for u, _ in units], [a, b]) + + @unittest.skipUnless( + shutil.which("ffmpeg") and shutil.which("ffprobe"), "needs ffmpeg and ffprobe" + ) + def test_unit_count_matches_ffprobe_and_units_concatenate_to_the_file(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "s.h264" + m.make_stream(path, 320, 240, 30, seconds=7) + data = path.read_bytes() + units = m.split_access_units(data) + self.assertEqual(b"".join(u for u, _ in units), data) + out = subprocess.run( + ["ffprobe", "-v", "error", "-count_packets", "-select_streams", "v", "-show_entries", + "stream=nb_read_packets", "-of", "csv=p=0", str(path)], + capture_output=True, text=True, check=True, + ).stdout # fmt: skip + self.assertEqual(len(units), int(out.strip())) + keys = [i for i, (_, k) in enumerate(units) if k] + self.assertEqual(keys[0], 0) + self.assertTrue(all(b == a + 150 for a, b in zip(keys, keys[1:]))) + self.assertTrue( + units[0][0].startswith(START) and units[0][0][4] & 0x1F == 7 + ) # SPS first + self.assertFalse( + any(t == 9 for u, _ in units for _, t, _ in m._nals(u)) + ) # no AUDs + + +class RecordTest(unittest.TestCase): + def test_layout(self) -> None: + au = b"\x00\x00\x00\x01\x65abc" + rec = m.record( + au, + sequence=7, + frame=14, + stamp_ns=123456789, + width=2560, + height=1984, + fps=30, + ) + (length,) = struct.unpack("!I", rec[:4]) + self.assertEqual(length, len(rec) - 4) + body = rec[4:] + header = m.HEADER.unpack(body[: m.HEADER.size]) + self.assertEqual(header[0], b"AVS1") + self.assertEqual(header[3], m.HEADER.size) + self.assertEqual(header[6], 7) # sequence + descriptors = body[ + m.HEADER.size : m.HEADER.size + m.STREAM.size + m.SAMPLE.size + ] + self.assertEqual(header[10], len(descriptors)) + self.assertEqual(header[11], len(au)) # payload bytes + stream = m.STREAM.unpack(descriptors[: m.STREAM.size]) + self.assertEqual( + (stream[2], stream[5], stream[6], stream[7]), (m.CODEC_H264, 2560, 1984, 30) + ) + sample = m.SAMPLE.unpack(descriptors[m.STREAM.size :]) + self.assertEqual(sample[7], 14) # frame number + self.assertEqual(sample[9], 123456789) # stamp + self.assertEqual(sample[-1], len(au)) + self.assertEqual(body[-len(au) :], au) + + +class FakeWorker: + """A TCP server accepting DATA then CTRL connections, like the worker's receiver.""" + + def __init__(self) -> None: + self.server = socket.socket() + self.server.bind(("127.0.0.1", 0)) + self.server.listen(8) + self.port = self.server.getsockname()[1] + self.hellos: list[bytes] = [] + self.sessions: list[tuple[socket.socket, socket.socket]] = [] + self.records: list[list[bytes]] = [] + self._threads: list[threading.Thread] = [] + self._stop = False + t = threading.Thread(target=self._accept, daemon=True) + t.start() + + def _accept(self) -> None: + while not self._stop: + try: + a, _ = self.server.accept() + b, _ = self.server.accept() + except OSError: + return + ha, hb = a.recv(4), b.recv(4) + self.hellos += [ha, hb] + data, ctrl = (a, b) if ha == b"DATA" else (b, a) + self.sessions.append((data, ctrl)) + self.records.append([]) + threading.Thread( + target=self._read, args=(data, len(self.records) - 1), daemon=True + ).start() + + def _read(self, data: socket.socket, index: int) -> None: + buf = b"" + try: + while True: + chunk = data.recv(65536) + if not chunk: + return + buf += chunk + while len(buf) >= 4: + (n,) = struct.unpack("!I", buf[:4]) + if len(buf) < 4 + n: + break + self.records[index].append(buf[4 : 4 + n]) + buf = buf[4 + n :] + except OSError: + return + + def close_session(self, index: int) -> None: + for s in self.sessions[index]: + s.close() + + def close(self) -> None: + self._stop = True + self.server.close() + for pair in self.sessions: + for s in pair: + s.close() + + +def wait_for(cond, timeout: float = 5.0) -> bool: + end = time.monotonic() + timeout + while time.monotonic() < end: + if cond(): + return True + time.sleep(0.02) + return False + + +def frame_numbers(records: list[bytes]) -> list[int]: + off = m.HEADER.size + m.STREAM.size + return [m.SAMPLE.unpack(r[off : off + m.SAMPLE.size])[7] for r in records] + + +IDR = b"\x00\x00\x00\x01\x65" + b"I" * 20 +DELTA = b"\x00\x00\x00\x01\x41" + b"P" * 20 + + +class DirectSenderTest(unittest.TestCase): + def setUp(self) -> None: + self.worker = FakeWorker() + self.clock = m.OrinClock(5_000_000_000, 0.0) + self.senders: list[m.DirectSender] = [] + + def tearDown(self) -> None: + for s in self.senders: + s.stop() + self.worker.close() + + def sender(self, max_queued: int = 1 << 20) -> m.DirectSender: + s = m.DirectSender( + "127.0.0.1", self.worker.port, 64, 48, 30, self.clock, max_queued + ) + self.senders.append(s) + return s + + def test_hello_order_is_data_then_ctrl(self) -> None: + self.sender() + self.assertTrue(wait_for(lambda: len(self.worker.hellos) == 2)) + self.assertEqual(self.worker.hellos, [b"DATA", b"CTRL"]) + + def test_deltas_are_ignored_until_a_keyframe_after_connect(self) -> None: + s = self.sender() + self.assertTrue( + wait_for( + lambda: s.connects == 1 and s._need_keyframe and s._keyframe_requested + ) + ) + s.submit(DELTA, False, 2, self.clock.now_ns()) + s.submit(DELTA, False, 4, self.clock.now_ns()) + s.submit(IDR, True, 6, self.clock.now_ns()) + s.submit(DELTA, False, 8, self.clock.now_ns()) + self.assertTrue( + wait_for( + lambda: ( + len(self.worker.records) == 1 and len(self.worker.records[0]) == 2 + ) + ) + ) + self.assertEqual(frame_numbers(self.worker.records[0]), [6, 8]) + self.assertTrue(s.take_keyframe_request()) + self.assertFalse(s.take_keyframe_request()) + + def test_reconnects_after_the_server_closes_and_restarts_at_a_keyframe( + self, + ) -> None: + s = self.sender() + self.assertTrue(wait_for(lambda: s.connects == 1)) + s.submit(IDR, True, 2, self.clock.now_ns()) + self.assertTrue( + wait_for(lambda: self.worker.records and len(self.worker.records[0]) == 1) + ) + self.worker.close_session(0) + self.assertTrue(wait_for(lambda: s.connects == 2, timeout=8)) + # The sender counts a connect once its hellos are written; the fake worker records them on its own thread. + self.assertTrue(wait_for(lambda: len(self.worker.hellos) == 4)) + self.assertEqual(self.worker.hellos, [b"DATA", b"CTRL", b"DATA", b"CTRL"]) + self.assertTrue(wait_for(lambda: s._need_keyframe and s._keyframe_requested)) + s.submit( + DELTA, False, 4, self.clock.now_ns() + ) # ignored: the new connection starts at a keyframe + s.submit(IDR, True, 6, self.clock.now_ns()) + self.assertTrue( + wait_for( + lambda: ( + len(self.worker.records) == 2 and len(self.worker.records[1]) == 1 + ) + ) + ) + self.assertEqual(frame_numbers(self.worker.records[1]), [6]) + + def test_queue_is_cleared_on_reconnect(self) -> None: + s = self.sender() + self.assertTrue(wait_for(lambda: s.connects == 1)) + self.worker.close_session(0) + self.assertTrue(wait_for(lambda: s.connects == 2, timeout=8)) + self.assertTrue(wait_for(lambda: s._need_keyframe)) + self.assertEqual(len(s._queue), 0) + self.assertEqual(s._queued_bytes, 0) + + def test_overflow_drops_then_ignores_deltas_until_the_next_keyframe(self) -> None: + s = self.sender() + self.assertTrue(wait_for(lambda: s.connects == 1)) + # Stop the sender thread from draining: hold the lock while submitting. + rec_size = len(m.record(IDR, 1, 2, 0, 64, 48, 30)) + s.max_queued_bytes = rec_size * 2 + 1 + with s._cond: + pass + s._queue.clear() + s._need_keyframe = False + s._queued_bytes = 0 + # Submit while holding the condition's lock so the sender thread cannot pop between submits. + with s._cond: + for i, (au, key) in enumerate( + [(IDR, True), (DELTA, False), (DELTA, False), (DELTA, False)] + ): + self._submit_locked(s, au, key, 2 * (i + 1)) + self.assertEqual( + s.dropped, 1 + ) # the third unit did not fit, the fourth was ignored + self.assertTrue(s._need_keyframe) + self.assertTrue(s._keyframe_requested) + self.assertEqual(len(s._queue), 2) + # Until the next keyframe, deltas are ignored; the keyframe is taken. + s.submit(DELTA, False, 20, self.clock.now_ns()) + self.assertEqual(s.dropped, 1) + self.assertTrue(wait_for(lambda: len(s._queue) == 0)) + s.submit(IDR, True, 22, self.clock.now_ns()) + self.assertFalse(s._need_keyframe) + + @staticmethod + def _submit_locked(s: m.DirectSender, au: bytes, key: bool, frame: int) -> None: + # Condition wraps an RLock, so submit() can re-enter it while the test holds it. + s.submit(au, key, frame, 0) + + def test_k_sets_keyframe_requested(self) -> None: + s = self.sender() + self.assertTrue(wait_for(lambda: s.connects == 1)) + self.assertTrue( + wait_for(lambda: s.take_keyframe_request()) + ) # the connect's own request + _, ctrl = self.worker.sessions[0] + ctrl.sendall(b"K") + self.assertTrue(wait_for(lambda: s.keyframe_requests == 1)) + self.assertTrue(s.take_keyframe_request()) + + def test_q_is_answered_with_r_echoing_seq_and_t1_on_the_offset_clock(self) -> None: + s = self.sender() + self.assertTrue(wait_for(lambda: s.connects == 1)) + _, ctrl = self.worker.sessions[0] + t1 = time.monotonic_ns() + ctrl.settimeout(5) + ctrl.sendall(struct.pack(" None: + s = self.sender() + self.assertTrue(wait_for(lambda: s.connects == 1)) + _, ctrl = self.worker.sessions[0] + ctrl.settimeout(5) + ctrl.sendall(b"K" + struct.pack(" None: + clock = m.OrinClock(0, 1000.0) # 1000 ppm fast + a_local, a = time.monotonic_ns(), clock.now_ns() + time.sleep(0.3) + b_local, b = time.monotonic_ns(), clock.now_ns() + extra = (b - a) - (b_local - a_local) + self.assertGreater(extra, 0.3e9 * 1000e-6 * 0.5) + flat = m.OrinClock(0, 0.0) + c = flat.now_ns() + time.sleep(0.1) + d = flat.now_ns() + self.assertAlmostEqual((d - c) / 1e9, 0.1, delta=0.03) + + +class StalledPeer: + """Accepts the DATA and CTRL connections, reads the hellos and then never reads again.""" + + def __init__(self) -> None: + self.server = socket.socket() + self.server.bind(("127.0.0.1", 0)) + self.server.listen(8) + self.port = self.server.getsockname()[1] + self.accepted: list[socket.socket] = [] + threading.Thread(target=self._accept, daemon=True).start() + + def _accept(self) -> None: + try: + while True: + c, _ = self.server.accept() + c.recv(4) + self.accepted.append(c) + except OSError: + return + + def close(self) -> None: + self.server.close() + for c in self.accepted: + c.close() + + +class StopTest(unittest.TestCase): + def test_stop_returns_promptly_while_a_send_is_blocked_on_a_peer_that_does_not_read( + self, + ) -> None: + peer = StalledPeer() + self.addCleanup(peer.close) + clock = m.OrinClock(0, 0.0) + s = m.DirectSender("127.0.0.1", peer.port, 64, 48, 30, clock, 512 << 20) + self.assertTrue(wait_for(lambda: s.connects == 1)) + big = b"\x00\x00\x00\x01\x65" + b"I" * (1 << 20) + s.submit(big, True, 2, 0) + for i in range(3, 80): + s.submit(big, False, 2 * i, 0) + # Loopback buffers hold a few MB; with ~80 MB queued the sender thread is stuck in sendall. + time.sleep(0.5) + self.assertGreater(len(s._queue), 0) + started = time.monotonic() + s.stop() + self.assertLess(time.monotonic() - started, 2.0) + self.assertFalse(s._thread.is_alive()) + + def test_stop_returns_promptly_when_connected_and_idle(self) -> None: + peer = StalledPeer() + self.addCleanup(peer.close) + s = m.DirectSender( + "127.0.0.1", peer.port, 64, 48, 30, m.OrinClock(0, 0.0), 1 << 20 + ) + self.assertTrue(wait_for(lambda: s.connects == 1)) + started = time.monotonic() + s.stop() + self.assertLess(time.monotonic() - started, 2.0) + + +class TwoCameraKeyframeTest(unittest.TestCase): + """Runs the mock's main() against two fake workers: a keyframe request from one camera moves only its cursor.""" + + UNITS = 200 + GOP = 40 + + def _source(self, path: Path) -> None: + data = b"" + for i in range(self.UNITS): + key = i % self.GOP == 0 + data += START + bytes([0x65 if key else 0x41, 0x80, 0xAA, i]) + path.write_bytes(data) + + @staticmethod + def _indices(records: list[bytes]) -> list[int]: + return [r[-1] for r in records] + + def test_a_keyframe_request_on_one_camera_leaves_the_other_cursor_alone( + self, + ) -> None: + a, b = FakeWorker(), FakeWorker() + self.addCleanup(a.close) + self.addCleanup(b.close) + with tempfile.TemporaryDirectory() as tmp: + src = Path(tmp) / "s.h264" + self._source(src) + proc = subprocess.Popen( + [sys.executable, str(Path(m.__file__)), "--port", str(a.port), + "--port", str(b.port), "--width", "64", "--height", "48", "--fps", "30", + "--source", str(src), "--duration", "5"], + stdout=subprocess.DEVNULL, + ) # fmt: skip + try: + self.assertTrue( + wait_for( + lambda: all( + w.records and len(w.records[0]) >= 8 for w in (a, b) + ) + ) + ) + before_a = len(a.records[0]) + a.sessions[0][1].sendall(b"K") + self.assertTrue(wait_for(lambda: len(a.records[0]) >= before_a + 10)) + proc.wait(timeout=15) + finally: + if proc.poll() is None: + proc.kill() + proc.wait() + idx_a, idx_b = self._indices(a.records[0]), self._indices(b.records[0]) + steps_a = [y - x for x, y in zip(idx_a, idx_a[1:])] + steps_b = [y - x for x, y in zip(idx_b, idx_b[1:])] + self.assertTrue(any(s > 1 for s in steps_a), f"camera A never jumped: {idx_a}") + jump = next(i for i, s in enumerate(steps_a) if s > 1) + self.assertEqual(idx_a[jump + 1] % self.GOP, 0) # it resumed at a keyframe + self.assertEqual(set(steps_b), {1}, f"camera B was disturbed: {idx_b}") + + +if __name__ == "__main__": + unittest.main() diff --git a/src/video_stage/wire/inc/video_stage/packet.hpp b/src/video_stage/wire/inc/video_stage/packet.hpp new file mode 100644 index 0000000000..c7dade373f --- /dev/null +++ b/src/video_stage/wire/inc/video_stage/packet.hpp @@ -0,0 +1,59 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include +#include + +namespace video_stage +{ + +/// Picture size in luma pixels. +struct Geometry +{ + uint32_t width = 0; + uint32_t height = 0; + + bool operator==(const Geometry& other) const + { + return width == other.width && height == other.height; + } +}; + +/// One H.264 access unit leaving the stage and the key of the input it came from. +struct EncodedPacket +{ + std::vector data; + uint64_t key = 0; + bool keyframe = false; +}; + +/// Source identity carried with a picture from the access unit that produced it to the packet that leaves. +struct FrameMeta +{ + uint64_t frame_number = 0; + uint64_t source_timestamp_ns = 0; + int64_t received_ns = 0; ///< CLOCK_MONOTONIC when the intake finished reading the record + uint64_t key = ~uint64_t{ 0 }; ///< the key this entry was written for; a mismatch means the entry was overwritten + int64_t submitted_ns = 0; ///< when the access unit was handed to a decoder, if there is one +}; + +/// Where finished access units go. Called on the stage's processing thread; it must not block. +class PacketSink +{ +public: + virtual ~PacketSink() = default; + virtual void write(const EncodedPacket& packet, const FrameMeta& meta, const Geometry& geometry) = 0; +}; + +/// CLOCK_MONOTONIC in nanoseconds. +inline int64_t monotonic_ns() +{ + timespec ts{}; + clock_gettime(CLOCK_MONOTONIC, &ts); + return static_cast(ts.tv_sec) * 1'000'000'000 + ts.tv_nsec; +} + +} // namespace video_stage