From 3d53d40cd9dc3c92fead8d504e448e45d91de564 Mon Sep 17 00:00:00 2001 From: RedNicStone Date: Thu, 29 Oct 2026 13:36:44 +0000 Subject: [PATCH] Initial commit --- .gitignore | 3 + CMakeLists.txt | 129 ++++++++++ CMakePresets.json | 44 ++++ cmake/FindRockchipMPP.cmake | 35 +++ cmake/build_config.hpp.in | 4 + include/rockstream/build_info.hpp | 8 + include/rockstream/latency.hpp | 45 ++++ include/rockstream/probe.hpp | 33 +++ src/build_info.cpp | 7 + src/latency.cpp | 128 ++++++++++ src/main.cpp | 43 ++++ src/mpp_probe.cpp | 376 ++++++++++++++++++++++++++++++ src/probe.cpp | 104 +++++++++ tests/mpp_headers_check.cpp | 20 ++ tests/probe_tests.cpp | 87 +++++++ 15 files changed, 1066 insertions(+) create mode 100644 .gitignore create mode 100644 CMakeLists.txt create mode 100644 CMakePresets.json create mode 100644 cmake/FindRockchipMPP.cmake create mode 100644 cmake/build_config.hpp.in create mode 100644 include/rockstream/build_info.hpp create mode 100644 include/rockstream/latency.hpp create mode 100644 include/rockstream/probe.hpp create mode 100644 src/build_info.cpp create mode 100644 src/latency.cpp create mode 100644 src/main.cpp create mode 100644 src/mpp_probe.cpp create mode 100644 src/probe.cpp create mode 100644 tests/mpp_headers_check.cpp create mode 100644 tests/probe_tests.cpp diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..8061587 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +/build/ +/CMakeUserPresets.json +/compile_commands.json diff --git a/CMakeLists.txt b/CMakeLists.txt new file mode 100644 index 0000000..b59fd45 --- /dev/null +++ b/CMakeLists.txt @@ -0,0 +1,129 @@ +cmake_minimum_required(VERSION 3.22) + +project(rockstream VERSION 0.1.0 LANGUAGES CXX) + +if(NOT CMAKE_SYSTEM_NAME STREQUAL "Linux") + message(FATAL_ERROR "rockstream currently supports Linux only") +endif() + +option(ROCKSTREAM_WITH_MPP "Build with the target BSP's Rockchip MPP SDK" OFF) +option(ROCKSTREAM_SANITIZERS "Enable AddressSanitizer and UndefinedBehaviorSanitizer" OFF) +option(ROCKSTREAM_WARNINGS_AS_ERRORS "Treat project compiler warnings as errors" OFF) + +list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake") +set(CMAKE_EXPORT_COMPILE_COMMANDS ON) +include(CTest) +include(GNUInstallDirs) +find_package(Threads REQUIRED) + +add_library(rockstream_options INTERFACE) +target_compile_features(rockstream_options INTERFACE cxx_std_20) +if(CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang") + target_compile_options(rockstream_options INTERFACE + -Wall -Wextra -Wpedantic -Wconversion -Wsign-conversion -Wshadow + $<$:-Werror>) + if(ROCKSTREAM_SANITIZERS) + include(CheckCXXSourceCompiles) + include(CMakePushCheckState) + cmake_push_check_state(RESET) + set(CMAKE_REQUIRED_FLAGS "-fsanitize=address,undefined") + set(CMAKE_REQUIRED_LINK_OPTIONS "-fsanitize=address,undefined") + check_cxx_source_compiles("int main() { return 0; }" ROCKSTREAM_SANITIZERS_LINK) + cmake_pop_check_state() + if(NOT ROCKSTREAM_SANITIZERS_LINK) + message(FATAL_ERROR "Sanitizer runtime compile/link check failed. Install ASan and UBSan runtimes matching your compiler, or disable ROCKSTREAM_SANITIZERS.") + endif() + target_compile_options(rockstream_options INTERFACE + -fsanitize=address,undefined -fno-omit-frame-pointer) + target_link_options(rockstream_options INTERFACE -fsanitize=address,undefined) + endif() +elseif(ROCKSTREAM_SANITIZERS OR ROCKSTREAM_WARNINGS_AS_ERRORS) + message(FATAL_ERROR "Requested compiler options require GCC or Clang") +endif() + +if(ROCKSTREAM_WITH_MPP) + find_package(RockchipMPP REQUIRED) + # Compile and link only: this check never executes hardware code, including + # when cross compiling. Catch missing public headers or an unusable library. + include(CheckCXXSourceCompiles) + include(CMakePushCheckState) + cmake_push_check_state(RESET) + set(CMAKE_REQUIRED_LIBRARIES RockchipMPP::MPP) + unset(ROCKSTREAM_MPP_SDK_LINKS CACHE) + check_cxx_source_compiles(" + #include + #include + #include + #include + #include + #include + int main() { + MppCtx ctx = nullptr; + MppApi* api = nullptr; + MppEncCfg cfg = nullptr; + if (mpp_enc_cfg_init(&cfg) != MPP_OK) return 1; + mpp_enc_cfg_deinit(cfg); + if (mpp_create(&ctx, &api) != MPP_OK) return 1; + return mpp_destroy(ctx); + }" ROCKSTREAM_MPP_SDK_LINKS) + cmake_pop_check_state() + if(NOT ROCKSTREAM_MPP_SDK_LINKS) + message(FATAL_ERROR "MPP SDK compile/link check failed. Use matching BSP headers and libraries; see CMakeFiles/CMakeError.log or CMakeConfigureLog.yaml.") + endif() +endif() + +configure_file(cmake/build_config.hpp.in generated/rockstream/build_config.hpp @ONLY) + +add_library(rockstream_core STATIC src/build_info.cpp src/probe.cpp src/latency.cpp) +target_include_directories(rockstream_core PUBLIC + "${CMAKE_CURRENT_SOURCE_DIR}/include" + "${CMAKE_CURRENT_BINARY_DIR}/generated") +target_link_libraries(rockstream_core PRIVATE rockstream_options PUBLIC Threads::Threads) +set_target_properties(rockstream_core PROPERTIES CXX_EXTENSIONS OFF) +if(ROCKSTREAM_WITH_MPP) + target_sources(rockstream_core PRIVATE src/mpp_probe.cpp) + target_link_libraries(rockstream_core PRIVATE RockchipMPP::MPP) +else() + # Host development gets public API declarations, never a pretend MPP runtime. + add_library(rockstream_mpp_headers INTERFACE) + target_include_directories(rockstream_mpp_headers SYSTEM INTERFACE + "${CMAKE_CURRENT_SOURCE_DIR}/third_party/mpp/include") + target_link_libraries(rockstream_core PRIVATE rockstream_mpp_headers) +endif() + +add_executable(rockstream src/main.cpp) +target_link_libraries(rockstream PRIVATE rockstream_core rockstream_options) +set_target_properties(rockstream PROPERTIES CXX_EXTENSIONS OFF) +install(TARGETS rockstream RUNTIME DESTINATION "${CMAKE_INSTALL_BINDIR}") + +if(BUILD_TESTING) + # Compile only: exercise SDK declarations without requiring device access or + # linking undefined MPP calls. Board builds use their installed SDK instead. + add_library(mpp_headers_check OBJECT tests/mpp_headers_check.cpp) + if(NOT ROCKSTREAM_WITH_MPP) + # Compile the complete native probe against pinned headers on the host. + # Its objects are deliberately never linked into a host executable. + target_sources(mpp_headers_check PRIVATE src/mpp_probe.cpp) + target_include_directories(mpp_headers_check PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/include") + endif() + target_link_libraries(mpp_headers_check PRIVATE rockstream_options) + set_target_properties(mpp_headers_check PROPERTIES CXX_EXTENSIONS OFF) + if(ROCKSTREAM_WITH_MPP) + target_link_libraries(mpp_headers_check PRIVATE RockchipMPP::MPP) + else() + target_link_libraries(mpp_headers_check PRIVATE rockstream_mpp_headers) + endif() + add_test(NAME cli_help COMMAND rockstream --help) + add_test(NAME cli_build_info COMMAND rockstream --build-info) + set_tests_properties(cli_build_info PROPERTIES PASS_REGULAR_EXPRESSION "MPP SDK: (enabled|disabled)") + add_test(NAME cli_reject_unknown COMMAND rockstream --invalid-option) + set_tests_properties(cli_reject_unknown PROPERTIES WILL_FAIL TRUE) + add_test(NAME probe_dry_run COMMAND rockstream encode-probe --dry-run --tiles off --split-ctus 60) + set_tests_properties(probe_dry_run PROPERTIES PASS_REGULAR_EXPRESSION "Dry run: no encoder initialized") + add_executable(probe_tests tests/probe_tests.cpp) + target_link_libraries(probe_tests PRIVATE rockstream_core rockstream_options) + set_target_properties(probe_tests PROPERTIES CXX_EXTENSIONS OFF) + add_test(NAME probe_logic COMMAND probe_tests) +endif() + +message(STATUS "rockstream ${PROJECT_VERSION}: ${CMAKE_SYSTEM_PROCESSOR}, MPP=${ROCKSTREAM_WITH_MPP}, sanitizers=${ROCKSTREAM_SANITIZERS}") diff --git a/CMakePresets.json b/CMakePresets.json new file mode 100644 index 0000000..e41750d --- /dev/null +++ b/CMakePresets.json @@ -0,0 +1,44 @@ +{ + "version": 3, + "cmakeMinimumRequired": { "major": 3, "minor": 22, "patch": 0 }, + "configurePresets": [ + { + "name": "host-debug", + "displayName": "Host development without Rockchip hardware", + "generator": "Ninja", + "binaryDir": "${sourceDir}/build/${presetName}", + "cacheVariables": { + "CMAKE_BUILD_TYPE": "Debug", + "BUILD_TESTING": "ON", + "ROCKSTREAM_WITH_MPP": "OFF", + "ROCKSTREAM_WARNINGS_AS_ERRORS": "ON" + } + }, + { + "name": "host-asan", + "inherits": "host-debug", + "cacheVariables": { "ROCKSTREAM_SANITIZERS": "ON" } + }, + { + "name": "rk3588-release", + "displayName": "Native RK3588 build with BSP MPP SDK", + "generator": "Ninja", + "binaryDir": "${sourceDir}/build/${presetName}", + "cacheVariables": { + "CMAKE_BUILD_TYPE": "Release", + "BUILD_TESTING": "ON", + "ROCKSTREAM_WITH_MPP": "ON" + } + } + ], + "buildPresets": [ + { "name": "host-debug", "configurePreset": "host-debug" }, + { "name": "host-asan", "configurePreset": "host-asan" }, + { "name": "rk3588-release", "configurePreset": "rk3588-release" } + ], + "testPresets": [ + { "name": "host-debug", "configurePreset": "host-debug", "output": { "outputOnFailure": true } }, + { "name": "host-asan", "configurePreset": "host-asan", "output": { "outputOnFailure": true } }, + { "name": "rk3588-release", "configurePreset": "rk3588-release", "output": { "outputOnFailure": true } } + ] +} diff --git a/cmake/FindRockchipMPP.cmake b/cmake/FindRockchipMPP.cmake new file mode 100644 index 0000000..aab8f39 --- /dev/null +++ b/cmake/FindRockchipMPP.cmake @@ -0,0 +1,35 @@ +# Prefer the BSP's pkg-config metadata, preserving its transitive link flags. +# SDKs without a .pc file can use -DRockchipMPP_ROOT=/path/to/prefix. +find_package(PkgConfig QUIET) +if(PkgConfig_FOUND) + pkg_check_modules(PC_RockchipMPP QUIET IMPORTED_TARGET rockchip_mpp) +endif() + +find_path(RockchipMPP_INCLUDE_DIR NAMES rk_mpi.h + HINTS ${PC_RockchipMPP_INCLUDE_DIRS} + PATH_SUFFIXES rockchip mpp include/rockchip include/mpp include) +include(FindPackageHandleStandardArgs) + +if(TARGET PkgConfig::PC_RockchipMPP) + set(RockchipMPP_VERSION "${PC_RockchipMPP_VERSION}") + find_package_handle_standard_args(RockchipMPP + REQUIRED_VARS RockchipMPP_INCLUDE_DIR VERSION_VAR RockchipMPP_VERSION) + if(RockchipMPP_FOUND AND NOT TARGET RockchipMPP::MPP) + add_library(RockchipMPP::MPP INTERFACE IMPORTED) + set_target_properties(RockchipMPP::MPP PROPERTIES + INTERFACE_INCLUDE_DIRECTORIES "${RockchipMPP_INCLUDE_DIR}" + INTERFACE_LINK_LIBRARIES PkgConfig::PC_RockchipMPP) + endif() +else() + find_library(RockchipMPP_LIBRARY NAMES rockchip_mpp mpp) + find_package_handle_standard_args(RockchipMPP + REQUIRED_VARS RockchipMPP_INCLUDE_DIR RockchipMPP_LIBRARY + REASON_FAILURE_MESSAGE "Install the target BSP's MPP development SDK, or set RockchipMPP_ROOT. Host-only builds use ROCKSTREAM_WITH_MPP=OFF.") + if(RockchipMPP_FOUND AND NOT TARGET RockchipMPP::MPP) + add_library(RockchipMPP::MPP UNKNOWN IMPORTED) + set_target_properties(RockchipMPP::MPP PROPERTIES + IMPORTED_LOCATION "${RockchipMPP_LIBRARY}" + INTERFACE_INCLUDE_DIRECTORIES "${RockchipMPP_INCLUDE_DIR}") + endif() + mark_as_advanced(RockchipMPP_INCLUDE_DIR RockchipMPP_LIBRARY) +endif() diff --git a/cmake/build_config.hpp.in b/cmake/build_config.hpp.in new file mode 100644 index 0000000..8434c5a --- /dev/null +++ b/cmake/build_config.hpp.in @@ -0,0 +1,4 @@ +#pragma once + +#define ROCKSTREAM_VERSION "@PROJECT_VERSION@" +#cmakedefine01 ROCKSTREAM_WITH_MPP diff --git a/include/rockstream/build_info.hpp b/include/rockstream/build_info.hpp new file mode 100644 index 0000000..5946df4 --- /dev/null +++ b/include/rockstream/build_info.hpp @@ -0,0 +1,8 @@ +#pragma once + +#include + +namespace rockstream { +[[nodiscard]] std::string_view version() noexcept; +[[nodiscard]] bool mpp_sdk_enabled() noexcept; +} // namespace rockstream diff --git a/include/rockstream/latency.hpp b/include/rockstream/latency.hpp new file mode 100644 index 0000000..1f281f7 --- /dev/null +++ b/include/rockstream/latency.hpp @@ -0,0 +1,45 @@ +#pragma once + +#include +#include +#include +#include +#include + +namespace rockstream { +using Nanoseconds = std::int64_t; +Nanoseconds monotonic_ns(); + +struct FrameTiming { + std::uint64_t id = 0; + Nanoseconds scheduled = 0; + Nanoseconds submit = 0; + Nanoseconds accepted = 0; + Nanoseconds first = 0; + Nanoseconds final = 0; + Nanoseconds reclaimed = 0; + std::size_t bytes = 0; + std::size_t packets = 0; + int intra = -1; // -1 when the SDK did not provide an indicator. +}; + +// Frame identity is carried in MPP PTS; completion and reclamation are independent. +class LatencyLog { +public: + explicit LatencyLog(std::size_t capacity); + void submitted(std::uint64_t id, Nanoseconds scheduled, Nanoseconds begin, Nanoseconds end); + void packet(std::uint64_t id, Nanoseconds observed, std::size_t bytes, bool eoi, int intra); + void reclaimed(std::uint64_t id, Nanoseconds observed); + [[nodiscard]] bool complete(std::uint64_t id) const; + [[nodiscard]] const FrameTiming& at(std::uint64_t id) const; + [[nodiscard]] std::span frames() const { return frames_; } +private: + FrameTiming& checked(std::uint64_t id); + std::vector frames_; +}; + +// Nearest-rank percentile; sorts its own copy and rejects empty/invalid inputs. +double percentile(std::vector values, double percent); +void print_latency_report(std::ostream& out, const LatencyLog& log, std::uint32_t warmup); +void write_latency_csv(std::ostream& out, const LatencyLog& log, std::uint32_t warmup); +} // namespace rockstream diff --git a/include/rockstream/probe.hpp b/include/rockstream/probe.hpp new file mode 100644 index 0000000..8473962 --- /dev/null +++ b/include/rockstream/probe.hpp @@ -0,0 +1,33 @@ +#pragma once + +#include +#include +#include +#include + +namespace rockstream { +struct ProbeOptions { + std::uint32_t width = 3840; + std::uint32_t height = 2160; + std::uint32_t fps = 60; + std::uint32_t bitrate = 25'000'000; + std::uint32_t gop = 60; + std::uint32_t frames = 600; + std::uint32_t warmup = 60; + std::uint32_t inflight = 1; + std::uint32_t split_ctus = 0; + std::uint32_t timeout_ms = 5000; + bool tiles = true; + bool paced = true; + bool dry_run = false; + std::string csv; + std::string output; +}; + +ProbeOptions parse_probe_options(std::span args); +void validate_probe_options(const ProbeOptions& options); +void print_probe_help(); +int run_probe(const ProbeOptions& options); +// Implemented by the real backend only in MPP-enabled builds. +int run_mpp_probe(const ProbeOptions& options); +} // namespace rockstream diff --git a/src/build_info.cpp b/src/build_info.cpp new file mode 100644 index 0000000..75be0c1 --- /dev/null +++ b/src/build_info.cpp @@ -0,0 +1,7 @@ +#include "rockstream/build_info.hpp" +#include "rockstream/build_config.hpp" + +namespace rockstream { +std::string_view version() noexcept { return ROCKSTREAM_VERSION; } +bool mpp_sdk_enabled() noexcept { return ROCKSTREAM_WITH_MPP != 0; } +} // namespace rockstream diff --git a/src/latency.cpp b/src/latency.cpp new file mode 100644 index 0000000..dd3154c --- /dev/null +++ b/src/latency.cpp @@ -0,0 +1,128 @@ +#include "rockstream/latency.hpp" + +#include +#include +#include +#include +#include +#include +#include + +namespace rockstream { +Nanoseconds monotonic_ns() { + return std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); +} + +LatencyLog::LatencyLog(std::size_t capacity) : frames_(capacity) {} + +FrameTiming& LatencyLog::checked(std::uint64_t id) { + if (id == 0 || id > frames_.size()) throw std::runtime_error("Unknown frame PTS/identity"); + return frames_[static_cast(id - 1)]; +} + +const FrameTiming& LatencyLog::at(std::uint64_t id) const { + if (id == 0 || id > frames_.size()) throw std::runtime_error("Unknown frame PTS/identity"); + return frames_[static_cast(id - 1)]; +} + +void LatencyLog::submitted(std::uint64_t id, Nanoseconds scheduled, Nanoseconds begin, Nanoseconds end) { + auto& f = checked(id); + if (f.id || begin <= 0 || end < begin || scheduled > begin) + throw std::runtime_error("Invalid or duplicate submission"); + f.id = id; + f.scheduled = scheduled; + f.submit = begin; + f.accepted = end; +} + +void LatencyLog::packet(std::uint64_t id, Nanoseconds t, std::size_t bytes, bool eoi, int intra) { + auto& f = checked(id); + if (!f.id || f.final || t < f.accepted || (f.first && t < f.first) || intra < -1 || intra > 1) + throw std::runtime_error("Packet has invalid identity/order/time"); + if (bytes && !f.first) f.first = t; + f.bytes += bytes; + ++f.packets; + f.intra = std::max(f.intra, intra); + if (eoi) { + if (!f.bytes) throw std::runtime_error("Encoder returned an empty/dropped frame"); + f.final = t; + } +} + +void LatencyLog::reclaimed(std::uint64_t id, Nanoseconds t) { + auto& f = checked(id); + if (!f.id || f.reclaimed || t < f.accepted) throw std::runtime_error("Invalid or duplicate input reclaim"); + f.reclaimed = t; +} + +bool LatencyLog::complete(std::uint64_t id) const { + const auto& f = at(id); + return f.final != 0 && f.reclaimed != 0; +} + +double percentile(std::vector values, double percent) { + if (values.empty() || !std::isfinite(percent) || percent < 0 || percent > 100) + throw std::invalid_argument("Invalid percentile input"); + std::sort(values.begin(), values.end()); + const auto rank = static_cast(std::ceil(percent * static_cast(values.size()) / 100.0)); + return values[rank ? rank - 1 : 0]; +} + +namespace { +double milliseconds(Nanoseconds ns) { return static_cast(ns) / 1'000'000.0; } +void distribution(std::ostream& out, std::string_view name, const std::vector& values) { + if (values.empty()) return; + out << name << " ms: median=" << percentile(values, 50) << " p95=" << percentile(values, 95) + << " p99=" << percentile(values, 99) << " max=" << percentile(values, 100) << '\n'; +} +} + +void print_latency_report(std::ostream& out, const LatencyLog& log, std::uint32_t warmup) { + std::vector first, final, reclaim, calls, lateness, intra, inter, unknown; + std::size_t count = 0, bytes = 0, packets = 0; + Nanoseconds start = 0, end = 0; + out << std::fixed << std::setprecision(3); + for (const auto& f : log.frames()) { + if (!f.id || !log.complete(f.id)) continue; + if (f.id == 1) out << "Startup first/final output ms: " << milliseconds(f.first - f.submit) + << " / " << milliseconds(f.final - f.submit) << '\n'; + if (f.id <= warmup) continue; + if (!start) start = f.submit; + end = std::max(end, f.final); + ++count; + bytes += f.bytes; + packets += f.packets; + first.push_back(milliseconds(f.first - f.submit)); + final.push_back(milliseconds(f.final - f.submit)); + reclaim.push_back(milliseconds(f.reclaimed - f.submit)); + calls.push_back(milliseconds(f.accepted - f.submit)); + lateness.push_back(milliseconds(f.submit - f.scheduled)); + (f.intra < 0 ? unknown : (f.intra ? intra : inter)).push_back(milliseconds(f.final - f.submit)); + } + out << "Measured completed frames=" << count << " packets=" << packets << " bytes=" << bytes << '\n'; + distribution(out, "Submit -> first output", first); + distribution(out, "Submit -> final output", final); + distribution(out, "Submit -> input reclaimed", reclaim); + distribution(out, "Successful submit call", calls); + distribution(out, "Source schedule lateness", lateness); + distribution(out, "Intra final output", intra); + distribution(out, "Non-intra final output", inter); + distribution(out, "Unclassified final output (no SDK intra indicator)", unknown); + if (end > start) { + const double seconds = static_cast(end - start) / 1e9; + out << "Window throughput=" << static_cast(count) / seconds << " fps, " + << static_cast(bytes) * 8.0 / seconds / 1e6 << " Mbit/s\n"; + } +} + +void write_latency_csv(std::ostream& out, const LatencyLog& log, std::uint32_t warmup) { + out << "frame_id,warmup,intra,scheduled_ns,submit_ns,accepted_ns,first_ns,final_ns,reclaimed_ns,bytes,packets\n"; + for (const auto& f : log.frames()) { + if (!f.id) continue; + out << f.id << ',' << (f.id <= warmup) << ',' << f.intra << ',' << f.scheduled << ',' + << f.submit << ',' << f.accepted << ',' << f.first << ',' << f.final << ',' << f.reclaimed + << ',' << f.bytes << ',' << f.packets << '\n'; + } +} +} // namespace rockstream diff --git a/src/main.cpp b/src/main.cpp new file mode 100644 index 0000000..6b5474a --- /dev/null +++ b/src/main.cpp @@ -0,0 +1,43 @@ +#include "rockstream/build_info.hpp" +#include "rockstream/probe.hpp" + +#include +#include +#include +#include + +int main(int argc, char* argv[]) { + const std::string_view option = argc == 2 ? argv[1] : ""; + if (argc >= 2 && std::string_view(argv[1]) == "encode-probe") { + try { + if (argc == 3 && std::string_view(argv[2]) == "--help") { + rockstream::print_probe_help(); + return 0; + } + std::vector args; + for (int i = 2; i < argc; ++i) args.emplace_back(argv[i]); + return rockstream::run_probe(rockstream::parse_probe_options(args)); + } catch (const std::exception& e) { + std::cerr << "Encoder probe: " << e.what() << '\n'; + return 1; + } + } + if (argc == 1 || option == "--help") { + std::cout << "Usage: rockstream [--help | --version | --build-info]\n" + " rockstream encode-probe [--help | options]\n" + "RK3588 low-latency HEVC encoder probe. Capture and transport are pending.\n"; + return 0; + } + if (option == "--version") { + std::cout << "rockstream " << rockstream::version() << '\n'; + return 0; + } + if (option == "--build-info") { + std::cout << "rockstream " << rockstream::version() << '\n' + << "MPP SDK: " << (rockstream::mpp_sdk_enabled() ? "enabled" : "disabled") + << "\nHardware availability: not validated\n"; + return 0; + } + std::cerr << "Expected --help, --version, --build-info, or encode-probe.\n"; + return 2; +} diff --git a/src/mpp_probe.cpp b/src/mpp_probe.cpp new file mode 100644 index 0000000..6908bd8 --- /dev/null +++ b/src/mpp_probe.cpp @@ -0,0 +1,376 @@ +#include "rockstream/probe.hpp" +#include "rockstream/latency.hpp" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace rockstream { +namespace { +volatile std::sig_atomic_t interrupted = 0; +void interrupt_handler(int) { interrupted = 1; } + +class SignalGuard { +public: + SignalGuard() { + interrupted = 0; + struct sigaction action {}; + action.sa_handler = interrupt_handler; + sigemptyset(&action.sa_mask); + if (sigaction(SIGINT, &action, &previous_) != 0) + throw std::runtime_error("Cannot install SIGINT handler"); + } + ~SignalGuard() { sigaction(SIGINT, &previous_, nullptr); } + SignalGuard(const SignalGuard&) = delete; + SignalGuard& operator=(const SignalGuard&) = delete; +private: + struct sigaction previous_ {}; +}; + +void check(MPP_RET ret, const char* operation) { + if (ret != MPP_OK) + throw std::runtime_error(std::string(operation) + " failed (MPP status " + std::to_string(ret) + ")"); +} + +class NewFile { +public: + explicit NewFile(const std::string& path) { + if (!path.empty()) { + file_ = std::fopen(path.c_str(), "wbx"); + if (!file_) throw std::runtime_error("Cannot create " + path + ": " + std::strerror(errno)); + } + } + ~NewFile() { if (file_) std::fclose(file_); } + NewFile(const NewFile&) = delete; + NewFile& operator=(const NewFile&) = delete; + void write(const void* data, std::size_t size) { + if (file_ && size && std::fwrite(data, 1, size, file_) != size) + throw std::runtime_error("Output write failed"); + } + void flush() { + if (file_ && std::fflush(file_) != 0) throw std::runtime_error("Output flush failed"); + } +private: + std::FILE* file_ = nullptr; +}; + +struct Packet { + MppPacket value = nullptr; + ~Packet() { + if (!value) return; + // On an exception before normal metadata consumption, the output queue + // no longer owns this packet's returned input; release it ourselves. + if (mpp_packet_has_meta(value)) { + MppFrame input = nullptr; + if (mpp_meta_get_frame(mpp_packet_get_meta(value), KEY_INPUT_FRAME, &input) == MPP_OK && input) + mpp_frame_deinit(&input); + } + mpp_packet_deinit(&value); + } +}; + +struct ReturnedFrame { + MppFrame value = nullptr; + ~ReturnedFrame() { if (value) mpp_frame_deinit(&value); } +}; + +struct Slot { + MppBuffer buffer = nullptr; + // Null while MPP owns the frame; restored only through KEY_INPUT_FRAME. + MppFrame frame = nullptr; + std::uint64_t id = 0; +}; + +struct Setting { const char* key; RK_S32 value; bool unsigned_value; }; + +class Probe { +public: + explicit Probe(const ProbeOptions& options) + : o_(options), log_(static_cast(o_.frames) + o_.warmup), + slots_(o_.inflight), csv_(o_.csv), output_(o_.output) {} + + ~Probe() { + // Stop all hardware access before releasing application-owned DMA refs. + // Reset/destroy may themselves block if the kernel/driver is hung. + if (ctx_) { + if (initialized_) { + const auto ret = api_->reset(ctx_); + if (ret != MPP_OK) std::cerr << "MPP reset during cleanup failed: " << ret << '\n'; + } + mpp_destroy(ctx_); + } + for (auto& slot : slots_) { + if (slot.frame) mpp_frame_deinit(&slot.frame); + if (slot.buffer) mpp_buffer_put(slot.buffer); + } + if (group_) mpp_buffer_group_put(group_); + if (cfg_) mpp_enc_cfg_deinit(cfg_); + } + Probe(const Probe&) = delete; + Probe& operator=(const Probe&) = delete; + + void initialize() { + check(mpp_check_support_format(MPP_CTX_ENC, MPP_VIDEO_CodingHEVC), "HEVC encoder support"); + check(mpp_create(&ctx_, &api_), "mpp_create"); + // Nonblocking input selects the asynchronous API with returned frames. + RK_S64 input_timeout = MPP_POLL_NON_BLOCK; + RK_S64 output_timeout = 1; // ms: wake on data, periodically revisit source/deadlines. + check(api_->control(ctx_, MPP_SET_INPUT_TIMEOUT, &input_timeout), "input timeout"); + check(api_->control(ctx_, MPP_SET_OUTPUT_TIMEOUT, &output_timeout), "output timeout"); + check(mpp_init(ctx_, MPP_CTX_ENC, MPP_VIDEO_CodingHEVC), "mpp_init HEVC (requires Rockchip hardware)"); + initialized_ = true; + check(mpp_enc_cfg_init(&cfg_), "mpp_enc_cfg_init"); + check(api_->control(ctx_, MPP_ENC_GET_CFG, cfg_), "MPP_ENC_GET_CFG"); + + stride_ = (o_.width + 63U) & ~63U; + storage_height_ = (o_.height + 63U) & ~63U; + const auto pixels = static_cast(o_.width) * o_.height; + const auto peak_bitrate = static_cast(o_.bitrate) * 17 / 16; + const bool large_level = pixels > 8'912'896 || pixels * o_.fps > 534'773'760 || peak_bitrate > 160'000'000; + const RK_S32 level = large_level ? 186 : 153; // HEVC levels 6.2 / 5.1. + const RK_S32 tier = peak_bitrate > (large_level ? 240'000'000U : 40'000'000U) ? 1 : 0; + const auto s32 = [](std::uint32_t n) { return static_cast(n); }; + const std::vector settings { + {"prep:width", s32(o_.width), false}, {"prep:height", s32(o_.height), false}, + {"prep:hor_stride", s32(stride_), false}, {"prep:ver_stride", s32(storage_height_), false}, + {"prep:format", MPP_FMT_YUV420SP, false}, + {"prep:colorspace", MPP_FRAME_SPC_BT709, false}, + {"prep:colorprim", MPP_FRAME_PRI_BT709, false}, + {"prep:colortrc", MPP_FRAME_TRC_BT709, false}, + {"prep:colorrange", MPP_FRAME_RANGE_MPEG, false}, + {"rc:mode", MPP_ENC_RC_MODE_CBR, false}, + {"rc:fps_in_flex", 0, false}, {"rc:fps_in_num", s32(o_.fps), false}, + {"rc:fps_in_denorm", 1, false}, {"rc:fps_out_flex", 0, false}, + {"rc:fps_out_num", s32(o_.fps), false}, {"rc:fps_out_denorm", 1, false}, + {"rc:gop", s32(o_.gop), false}, {"rc:bps_target", s32(o_.bitrate), false}, + {"rc:bps_min", static_cast(static_cast(o_.bitrate) * 15 / 16), false}, + {"rc:bps_max", static_cast(static_cast(o_.bitrate) * 17 / 16), false}, + {"rc:qp_init", -1, false}, {"rc:qp_min", 10, false}, {"rc:qp_max", 51, false}, + {"rc:qp_min_i", 10, false}, {"rc:qp_max_i", 51, false}, {"rc:qp_ip", 2, false}, + {"rc:max_reenc_times", 0, true}, {"rc:drop_mode", MPP_ENC_RC_DROP_FRM_DISABLED, true}, + {"rc:super_mode", MPP_ENC_RC_SUPER_FRM_NONE, true}, + {"h265:profile", 1, false}, {"h265:level", level, false}, {"h265:tier", tier, false}, + {"h265:auto_tile", o_.tiles ? 1 : 0, false}, + {"split:mode", o_.split_ctus ? MPP_ENC_SPLIT_BY_CTU : MPP_ENC_SPLIT_NONE, true}, + {"split:arg", s32(o_.split_ctus), true}, + {"split:out", o_.split_ctus ? MPP_ENC_SPLIT_OUT_LOWDELAY : 0, true} + }; + for (const auto& s : settings) { + check(s.unsigned_value ? mpp_enc_cfg_set_u32(cfg_, s.key, static_cast(s.value)) + : mpp_enc_cfg_set_s32(cfg_, s.key, s.value), s.key); + } + check(api_->control(ctx_, MPP_ENC_SET_CFG, cfg_), "MPP_ENC_SET_CFG"); + check(api_->control(ctx_, MPP_ENC_GET_CFG, cfg_), "MPP_ENC_GET_CFG readback"); + for (const auto& s : settings) { + RK_S32 value = 0; + if (s.unsigned_value) { + RK_U32 u = 0; + check(mpp_enc_cfg_get_u32(cfg_, s.key, &u), s.key); + value = static_cast(u); + } else check(mpp_enc_cfg_get_s32(cfg_, s.key, &value), s.key); + if (value != s.value) throw std::runtime_error(std::string("SDK changed requested setting: ") + s.key); + } + MppEncHeaderMode headers = MPP_ENC_HEADER_MODE_EACH_IDR; + check(api_->control(ctx_, MPP_ENC_SET_HEADER_MODE, &headers), "repeat headers at IDR"); + std::cout << "SDK settings read back successfully. HEVC level_idc=" << level << " tier=" << tier + << " stride=" << stride_ << " storage_height=" << storage_height_ + << "; default IPPP reference configuration.\n"; + + check(mpp_buffer_group_get_internal(&group_, MPP_BUFFER_TYPE_DRM), "DMA buffer group (DRM)"); + const auto y_size = static_cast(stride_) * storage_height_; + for (auto& slot : slots_) { + check(mpp_buffer_get(group_, &slot.buffer, y_size * 3 / 2), "DMA input allocation"); + auto* data = static_cast(mpp_buffer_get_ptr(slot.buffer)); + if (!data) throw std::runtime_error("Cannot map synthetic input DMA buffer"); + check(mpp_buffer_sync_begin(slot.buffer), "CPU DMA sync begin"); + std::memset(data, 16, y_size); + std::memset(data + y_size, 128, y_size / 2); + for (std::uint32_t y = 0; y < o_.height; ++y) + for (std::uint32_t x = 0; x < o_.width; ++x) + data[static_cast(y) * stride_ + x] = + static_cast(16U + (x * 219U / o_.width + y / 8U) % 220U); + check(mpp_buffer_sync_end(slot.buffer), "CPU DMA sync end"); + check(mpp_frame_init(&slot.frame), "mpp_frame_init"); + mpp_frame_set_width(slot.frame, o_.width); + mpp_frame_set_height(slot.frame, o_.height); + mpp_frame_set_hor_stride(slot.frame, stride_); + mpp_frame_set_ver_stride(slot.frame, storage_height_); + mpp_frame_set_fmt(slot.frame, MPP_FMT_YUV420SP); + mpp_frame_set_buffer(slot.frame, slot.buffer); + } + } + + int run() { + SignalGuard signals; + const auto total = static_cast(o_.frames) + o_.warmup; + const Nanoseconds period = 1'000'000'000LL / o_.fps; + const Nanoseconds timeout = static_cast(o_.timeout_ms) * 1'000'000; + auto due = monotonic_ns(); + Nanoseconds submission_wait_start = 0; + std::uint64_t submitted = 0, completed = 0, skipped_ticks = 0, retries = 0; + try { + while ((!interrupted && submitted < total) || completed < submitted) { + auto now = monotonic_ns(); + if (!interrupted && submitted < total && (!o_.paced || now >= due)) { + auto free = std::find_if(slots_.begin(), slots_.end(), [](const Slot& s) { return s.id == 0; }); + if (free != slots_.end()) { + if (o_.paced && now - due >= period) { + const auto skipped = (now - due) / period; + skipped_ticks += static_cast(skipped); + due += skipped * period; // Latest source tick; never replay a backlog. + } + const auto id = submitted + 1; + mpp_frame_set_pts(free->frame, static_cast(id)); + const auto begin = monotonic_ns(); + const auto ret = api_->encode_put_frame(ctx_, free->frame); + const auto end = monotonic_ns(); + if (ret == MPP_OK) { + // The async API holds the original MppFrame, not a copy. + free->frame = nullptr; + free->id = id; + log_.submitted(id, o_.paced ? due : begin, begin, end); + ++submitted; + due += period; + submission_wait_start = 0; + } else if (ret == MPP_NOK || ret == MPP_ERR_BUFFER_FULL || ret == MPP_ERR_TIMEOUT) { + ++retries; + if (!submission_wait_start) submission_wait_start = begin; + } + else check(ret, "encode_put_frame"); + } + } + + if (completed < submitted) receive(); + else if (!interrupted && submitted < total) { + if (o_.paced && monotonic_ns() < due) + std::this_thread::sleep_until(std::chrono::steady_clock::time_point(std::chrono::nanoseconds(due))); + else std::this_thread::sleep_for(std::chrono::milliseconds(1)); // Submission backpressure. + } + for (auto& slot : slots_) { + if (slot.id && log_.complete(slot.id)) { + slot.id = 0; + ++completed; + } + } + now = monotonic_ns(); + for (const auto& slot : slots_) { + if (slot.id && now - log_.at(slot.id).submit > timeout) + throw std::runtime_error("Frame timeout awaiting output/reclaim; frame=" + std::to_string(slot.id)); + } + if (!interrupted && submission_wait_start && now - submission_wait_start > timeout) + throw std::runtime_error("Timed out submitting frames"); + } + output_.flush(); + save_csv(); + print_latency_report(std::cout, log_, o_.warmup); + std::cout << "Accepted=" << submitted << " fully_completed=" << completed << " source_ticks_skipped=" + << skipped_ticks << " submit_retries=" << retries << '\n'; + if (interrupted) std::cout << "Interrupted: accepted inputs drained; report may be partial.\n"; + return interrupted ? 130 : 0; + } catch (...) { + // Preserve partial records, including zero completion timestamps. + try { save_csv(); } catch (const std::exception& e) { std::cerr << "CSV save failed: " << e.what() << '\n'; } + throw; + } + } + +private: + Slot& slot_for(std::uint64_t id) { + auto it = std::find_if(slots_.begin(), slots_.end(), [id](const Slot& s) { return s.id == id; }); + if (it == slots_.end()) throw std::runtime_error("MPP returned unknown frame PTS/identity"); + return *it; + } + + void receive() { + Packet packet; + const auto ret = api_->encode_get_packet(ctx_, &packet.value); + const auto observed = monotonic_ns(); + if (ret == MPP_ERR_TIMEOUT || ret == MPP_NOK) return; + check(ret, "encode_get_packet"); + if (!packet.value) return; + const auto pts = mpp_packet_get_pts(packet.value); + if (pts <= 0) throw std::runtime_error("MPP packet did not preserve input PTS"); + const auto id = static_cast(pts); + slot_for(id); + int intra = -1; + if (mpp_packet_has_meta(packet.value)) { + const auto meta = mpp_packet_get_meta(packet.value); + RK_S32 flag = 0; + if (mpp_meta_get_s32(meta, KEY_OUTPUT_INTRA, &flag) == MPP_OK) intra = flag != 0 ? 1 : 0; + ReturnedFrame frame; + if (mpp_meta_get_frame(meta, KEY_INPUT_FRAME, &frame.value) == MPP_OK && frame.value) { + // Avoid destroying a handle already owned by a slot if a broken + // SDK repeats the same completion metadata. + for (const auto& slot : slots_) { + if (slot.frame == frame.value) { + frame.value = nullptr; + throw std::runtime_error("SDK returned the same input handle twice"); + } + } + const auto reclaimed_pts = mpp_frame_get_pts(frame.value); + if (reclaimed_pts <= 0) { + throw std::runtime_error("Reclaimed input has invalid PTS"); + } + auto& slot = slot_for(static_cast(reclaimed_pts)); + if (slot.frame || mpp_frame_get_buffer(frame.value) != slot.buffer) + throw std::runtime_error("Invalid or duplicate DMA frame reclaim"); + slot.frame = frame.value; // Application owns this handle again. + frame.value = nullptr; + log_.reclaimed(slot.id, observed); + } + } + const auto bytes = mpp_packet_get_length(packet.value); + const bool eoi = !mpp_packet_is_partition(packet.value) || mpp_packet_is_eoi(packet.value); + log_.packet(id, observed, bytes, eoi, intra); + const auto* data = mpp_packet_get_pos(packet.value); + if (bytes && !data) throw std::runtime_error("Encoded packet has no data"); + output_.write(data, bytes); + } + + void save_csv() { + if (o_.csv.empty() || csv_saved_) return; + csv_saved_ = true; + std::ostringstream text; + write_latency_csv(text, log_, o_.warmup); + const auto data = text.str(); + csv_.write(data.data(), data.size()); + csv_.flush(); + } + + const ProbeOptions& o_; + LatencyLog log_; + std::vector slots_; + NewFile csv_, output_; + MppCtx ctx_ = nullptr; + MppApi* api_ = nullptr; + MppEncCfg cfg_ = nullptr; + MppBufferGroup group_ = nullptr; + std::uint32_t stride_ = 0, storage_height_ = 0; + bool initialized_ = false; + bool csv_saved_ = false; +}; +} // namespace + +int run_mpp_probe(const ProbeOptions& options) { + Probe probe(options); // Full construction before initialization for exception-safe cleanup. + probe.initialize(); + return probe.run(); +} +} // namespace rockstream diff --git a/src/probe.cpp b/src/probe.cpp new file mode 100644 index 0000000..9f41f61 --- /dev/null +++ b/src/probe.cpp @@ -0,0 +1,104 @@ +#include "rockstream/probe.hpp" +#include "rockstream/build_config.hpp" + +#include +#include +#include + +namespace rockstream { +namespace { +std::uint32_t number(std::string_view text, std::string_view key) { + std::uint32_t value = 0; + const auto result = std::from_chars(text.data(), text.data() + text.size(), value); + if (result.ec != std::errc{} || result.ptr != text.data() + text.size()) { + throw std::invalid_argument(std::string(key) + " requires an unsigned integer"); + } + return value; +} +} + +ProbeOptions parse_probe_options(std::span args) { + ProbeOptions o; + for (std::size_t i = 0; i < args.size(); ++i) { + const auto key = args[i]; + if (key == "--dry-run") { o.dry_run = true; continue; } + if (++i == args.size()) throw std::invalid_argument(std::string(key) + " needs a value"); + const auto value = args[i]; + if (key == "--width") o.width = number(value, key); + else if (key == "--height") o.height = number(value, key); + else if (key == "--fps") o.fps = number(value, key); + else if (key == "--bitrate") o.bitrate = number(value, key); + else if (key == "--gop") o.gop = number(value, key); + else if (key == "--frames") o.frames = number(value, key); + else if (key == "--warmup") o.warmup = number(value, key); + else if (key == "--inflight") o.inflight = number(value, key); + else if (key == "--split-ctus") o.split_ctus = number(value, key); + else if (key == "--timeout-ms") o.timeout_ms = number(value, key); + else if (key == "--csv") o.csv = value; + else if (key == "--output") o.output = value; + else if (key == "--tiles" && (value == "on" || value == "off")) o.tiles = value == "on"; + else if (key == "--mode" && (value == "paced" || value == "unpaced")) o.paced = value == "paced"; + else throw std::invalid_argument("Unknown option or invalid value: " + std::string(key)); + } + validate_probe_options(o); + return o; +} + +void validate_probe_options(const ProbeOptions& o) { + if (o.width < 64 || o.height < 64 || o.width > 8192 || o.height > 8192 || + o.width % 2 != 0 || o.height % 2 != 0 || + static_cast(o.width) * o.height > 7680ULL * 4320ULL) + throw std::invalid_argument("Dimensions must be even, 64..8192, and at most 7680*4320 pixels"); + if (o.fps == 0 || o.fps > 240) throw std::invalid_argument("fps must be 1..240"); + if (o.bitrate < 10'000 || o.bitrate > 750'000'000) + throw std::invalid_argument("bitrate must be 10000..750000000 bits/s"); + if (static_cast(o.width) * o.height * o.fps > 4'278'190'080ULL) + throw std::invalid_argument("Requested sample rate exceeds HEVC level 6.2"); + if (o.gop == 0 || o.gop > 10000) throw std::invalid_argument("gop must be 1..10000"); + if (o.frames == 0 || static_cast(o.frames) + o.warmup > 100000) + throw std::invalid_argument("frames must be positive; frames + warmup must be <= 100000"); + if (o.inflight == 0 || o.inflight > 4) throw std::invalid_argument("inflight must be 1..4"); + if (o.split_ctus > 65535) throw std::invalid_argument("split-ctus must be 0..65535 (0 disables splitting)"); + if (o.timeout_ms < 10 || o.timeout_ms > 60000) + throw std::invalid_argument("timeout-ms must be 10..60000"); + if (!o.csv.empty() && o.csv == o.output) throw std::invalid_argument("csv and output must be different files"); +} + +void print_probe_help() { + std::cout << "Usage: rockstream encode-probe [options]\n" + " --width N --height N NV12 dimensions (3840 x 2160)\n" + " --fps N Input/output rate (60)\n" + " --bitrate N --gop N CBR bits/s and GOP (25000000, 60)\n" + " --frames N --warmup N Measured and initial excluded frames (600, 60)\n" + " --mode paced|unpaced Realtime source or throughput run (paced)\n" + " --inflight N Bounded raw buffer pool (1; maximum 4)\n" + " --tiles on|off HEVC automatic tiles (on)\n" + " --split-ctus N Early slice output; 0 disables (0)\n" + " --timeout-ms N Per-frame/submission watchdog (5000)\n" + " --csv PATH Write timings after drain (new file only)\n" + " --output PATH Record Annex B HEVC; perturbs timing (new file only)\n" + " --dry-run Validate/print options without device access\n" + "Synthetic static luma ramp, neutral chroma. No camera or network.\n"; +} + +int run_probe(const ProbeOptions& o) { + validate_probe_options(o); + std::cout << "HEVC Main / NV12 " << o.width << 'x' << o.height << " @ " << o.fps + << " fps, CBR " << o.bitrate << " bits/s, GOP " << o.gop + << "\nRequested tiles=" << (o.tiles ? "on" : "off") << " split_ctus=" << o.split_ctus + << " inflight=" << o.inflight << " mode=" << (o.paced ? "paced" : "unpaced") + << " warmup=" << o.warmup << " measured_frames=" << o.frames + << " timeout_ms=" << o.timeout_ms << '\n' + << "Source: prefilled static NV12 ramp; no camera/network latency measured.\n"; + if (o.dry_run) { + std::cout << "Dry run: no encoder initialized.\n"; + return 0; + } + if (!o.output.empty()) std::cout << "Diagnostic bitstream recording enabled: timings include output IO interference.\n"; +#if ROCKSTREAM_WITH_MPP + return run_mpp_probe(o); +#else + throw std::runtime_error("Encoder probe requires ROCKSTREAM_WITH_MPP=ON and the board's MPP SDK; use --dry-run on this host"); +#endif +} +} // namespace rockstream diff --git a/tests/mpp_headers_check.cpp b/tests/mpp_headers_check.cpp new file mode 100644 index 0000000..02ac9a9 --- /dev/null +++ b/tests/mpp_headers_check.cpp @@ -0,0 +1,20 @@ +// Compile-only checks for the public API needed by the planned encoder backend. +#include +#include +#include +#include +#include +#include +#include + +#include + +static_assert(std::is_invocable_r_v); +static_assert(std::is_invocable_r_v); +static_assert(std::is_invocable_r_v); +static_assert(std::is_invocable_r_v); +static_assert(std::is_invocable_r_v); +static_assert(MPP_ENC_SPLIT_BY_CTU != MPP_ENC_SPLIT_NONE); +static_assert(MPP_ENC_SPLIT_OUT_LOWDELAY != 0); +static_assert(MPP_FMT_YUV420SP != MPP_FMT_BUTT); diff --git a/tests/probe_tests.cpp b/tests/probe_tests.cpp new file mode 100644 index 0000000..f84d311 --- /dev/null +++ b/tests/probe_tests.cpp @@ -0,0 +1,87 @@ +#include "rockstream/latency.hpp" +#include "rockstream/probe.hpp" + +#include +#include +#include +#include +#include + +namespace { +void expect(bool condition, const char* message) { + if (!condition) throw std::runtime_error(message); +} +template void rejects(F action) { + try { action(); } catch (const std::exception&) { return; } + throw std::runtime_error("Expected rejection"); +} +} + +int main() { + try { + using namespace rockstream; + using namespace std::literals; + const auto parsed = parse_probe_options(std::array{"--tiles"sv, "off"sv, "--mode"sv, "unpaced"sv, + "--split-ctus"sv, "60"sv, "--inflight"sv, "2"sv, "--dry-run"sv}); + expect(!parsed.tiles && !parsed.paced && parsed.split_ctus == 60 && parsed.inflight == 2 && parsed.dry_run, + "Probe configuration did not parse correctly"); + rejects([] { parse_probe_options(std::array{"--width"sv, "3839"sv}); }); + rejects([] { parse_probe_options(std::array{"--fps"sv, "0"sv}); }); + rejects([] { parse_probe_options(std::array{"--frames"sv, "4294967296"sv}); }); + rejects([] { parse_probe_options(std::array{"--inflight"sv, "-1"sv}); }); + rejects([] { parse_probe_options(std::array{"--split-ctus"sv, "2x"sv}); }); + rejects([] { parse_probe_options(std::array{"--tiles"sv, "maybe"sv}); }); + rejects([] { parse_probe_options(std::array{"--frames"sv}); }); + rejects([] { parse_probe_options(std::array{"--unknown"sv, "1"sv}); }); + rejects([] { parse_probe_options(std::array{"--output"sv, "same"sv, "--csv"sv, "same"sv}); }); + + LatencyLog log(3); + log.submitted(1, 90, 100, 110); + log.submitted(2, 190, 200, 210); + log.packet(1, 120, 20, false, true); + expect(!log.complete(1), "First partition must not complete a frame"); + log.reclaimed(1, 130); + expect(!log.complete(1), "Reclaim alone must not complete a partitioned frame"); + log.packet(2, 220, 50, true, false); + expect(!log.complete(2), "EOI must not imply buffer reclamation"); + log.packet(1, 240, 30, true, false); + log.reclaimed(2, 250); + expect(log.complete(1) && log.complete(2), "Completion state lost"); + expect(log.at(1).first == 120 && log.at(1).final == 240 && log.at(1).bytes == 50 && + log.at(1).packets == 2 && log.at(1).intra == 1, "Partition aggregation failed"); + rejects([&] { log.reclaimed(1, 260); }); + rejects([&] { log.packet(1, 260, 1, false, false); }); + rejects([&] { log.submitted(1, 1, 2, 3); }); + rejects([&] { log.packet(0, 260, 1, true, false); }); + rejects([&] { log.packet(3, 260, 1, true, false); }); + log.submitted(3, 290, 300, 310); + rejects([&] { log.packet(3, 320, 0, true, false); }); + rejects([&] { log.packet(3, 305, 1, true, false); }); + + expect(percentile({1, 2, 3, 4}, 50) == 2 && percentile({1, 2, 3, 4}, 99) == 4, + "Nearest-rank percentiles incorrect"); + rejects([] { percentile({}, 50); }); + rejects([] { percentile({1}, 101); }); + std::ostringstream csv, report; + write_latency_csv(csv, log, 1); + expect(csv.str().find("1,1,1,90,100,110,120,240,130,50,2") != std::string::npos, + "CSV lost raw timestamps or warmup/intra flags"); + print_latency_report(report, log, 1); + expect(report.str().find("Measured completed frames=1") != std::string::npos, + "Report included warmup or incomplete frames"); + LatencyLog missing_metadata(1); + missing_metadata.submitted(1, 100, 100, 110); + missing_metadata.packet(1, 120, 1, true, -1); + missing_metadata.reclaimed(1, 130); + std::ostringstream unclassified; + print_latency_report(unclassified, missing_metadata, 0); + expect(unclassified.str().find("Unclassified final output") != std::string::npos && + unclassified.str().find("Non-intra final output") == std::string::npos, + "Missing intra metadata was misclassified"); + std::cout << "Probe configuration, frame lifecycle, CSV and percentile checks passed.\n"; + return 0; + } catch (const std::exception& e) { + std::cerr << e.what() << '\n'; + return 1; + } +}