Initial commit

.gitignore

@@ -0,0 +1,3 @@
+/build/
+/CMakeUserPresets.json
+/compile_commands.json

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
+        $<$<BOOL:${ROCKSTREAM_WARNINGS_AS_ERRORS}>:-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 <rk_mpi.h>
+        #include <mpp_buffer.h>
+        #include <mpp_frame.h>
+        #include <mpp_packet.h>
+        #include <rk_venc_cfg.h>
+        #include <rk_venc_rc.h>
+        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}")

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 } }
+  ]
+}

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()

cmake/build_config.hpp.in

@@ -0,0 +1,4 @@
+#pragma once
+
+#define ROCKSTREAM_VERSION "@PROJECT_VERSION@"
+#cmakedefine01 ROCKSTREAM_WITH_MPP

include/rockstream/build_info.hpp

@@ -0,0 +1,8 @@
+#pragma once
+
+#include <string_view>
+
+namespace rockstream {
+[[nodiscard]] std::string_view version() noexcept;
+[[nodiscard]] bool mpp_sdk_enabled() noexcept;
+} // namespace rockstream

include/rockstream/latency.hpp

@@ -0,0 +1,45 @@
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <iosfwd>
+#include <span>
+#include <vector>
+
+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<const FrameTiming> frames() const { return frames_; }
+private:
+    FrameTiming& checked(std::uint64_t id);
+    std::vector<FrameTiming> frames_;
+};
+
+// Nearest-rank percentile; sorts its own copy and rejects empty/invalid inputs.
+double percentile(std::vector<double> 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

include/rockstream/probe.hpp

@@ -0,0 +1,33 @@
+#pragma once
+
+#include <cstdint>
+#include <span>
+#include <string>
+#include <string_view>
+
+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<const std::string_view> 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

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

src/latency.cpp

@@ -0,0 +1,128 @@
+#include "rockstream/latency.hpp"
+
+#include <algorithm>
+#include <chrono>
+#include <cmath>
+#include <iomanip>
+#include <ostream>
+#include <stdexcept>
+#include <string_view>
+
+namespace rockstream {
+Nanoseconds monotonic_ns() {
+    return std::chrono::duration_cast<std::chrono::nanoseconds>(
+        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<std::size_t>(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<std::size_t>(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<double> 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::size_t>(std::ceil(percent * static_cast<double>(values.size()) / 100.0));
+    return values[rank ? rank - 1 : 0];
+}
+
+namespace {
+double milliseconds(Nanoseconds ns) { return static_cast<double>(ns) / 1'000'000.0; }
+void distribution(std::ostream& out, std::string_view name, const std::vector<double>& 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<double> 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<double>(end - start) / 1e9;
+        out << "Window throughput=" << static_cast<double>(count) / seconds << " fps, "
+            << static_cast<double>(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

src/main.cpp

@@ -0,0 +1,43 @@
+#include "rockstream/build_info.hpp"
+#include "rockstream/probe.hpp"
+
+#include <exception>
+#include <iostream>
+#include <string_view>
+#include <vector>
+
+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<std::string_view> 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;
+}

src/mpp_probe.cpp

@@ -0,0 +1,376 @@
+#include "rockstream/probe.hpp"
+#include "rockstream/latency.hpp"
+
+#include <rk_mpi.h>
+#include <mpp_buffer.h>
+#include <mpp_frame.h>
+#include <mpp_meta.h>
+#include <mpp_packet.h>
+#include <rk_venc_cfg.h>
+#include <rk_venc_rc.h>
+
+#include <algorithm>
+#include <cerrno>
+#include <chrono>
+#include <csignal>
+#include <cstdio>
+#include <cstring>
+#include <iostream>
+#include <sstream>
+#include <stdexcept>
+#include <string>
+#include <thread>
+#include <vector>
+
+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<std::size_t>(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<std::uint64_t>(o_.width) * o_.height;
+        const auto peak_bitrate = static_cast<std::uint64_t>(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<RK_S32>(n); };
+        const std::vector<Setting> 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<RK_S32>(static_cast<std::uint64_t>(o_.bitrate) * 15 / 16), false},
+            {"rc:bps_max", static_cast<RK_S32>(static_cast<std::uint64_t>(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<RK_U32>(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<RK_S32>(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<std::size_t>(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<unsigned char*>(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<std::size_t>(y) * stride_ + x] =
+                        static_cast<unsigned char>(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<std::uint64_t>(o_.frames) + o_.warmup;
+        const Nanoseconds period = 1'000'000'000LL / o_.fps;
+        const Nanoseconds timeout = static_cast<Nanoseconds>(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<std::uint64_t>(skipped);
+                            due += skipped * period; // Latest source tick; never replay a backlog.
+                        }
+                        const auto id = submitted + 1;
+                        mpp_frame_set_pts(free->frame, static_cast<RK_S64>(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<std::uint64_t>(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<std::uint64_t>(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<Slot> 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

src/probe.cpp

@@ -0,0 +1,104 @@
+#include "rockstream/probe.hpp"
+#include "rockstream/build_config.hpp"
+
+#include <charconv>
+#include <iostream>
+#include <stdexcept>
+
+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<const std::string_view> 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<std::uint64_t>(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<std::uint64_t>(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<std::uint64_t>(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

tests/mpp_headers_check.cpp

@@ -0,0 +1,20 @@
+// Compile-only checks for the public API needed by the planned encoder backend.
+#include <rk_mpi.h>
+#include <mpp_buffer.h>
+#include <mpp_frame.h>
+#include <mpp_meta.h>
+#include <mpp_packet.h>
+#include <rk_venc_cfg.h>
+#include <rk_venc_rc.h>
+
+#include <type_traits>
+
+static_assert(std::is_invocable_r_v<MPP_RET, decltype(&mpp_create), MppCtx*, MppApi**>);
+static_assert(std::is_invocable_r_v<MPP_RET, decltype(&mpp_enc_cfg_init), MppEncCfg*>);
+static_assert(std::is_invocable_r_v<MPP_RET, decltype(&mpp_enc_cfg_set_u32),
+              MppEncCfg, const char*, RK_U32>);
+static_assert(std::is_invocable_r_v<void, decltype(&mpp_frame_set_pts), MppFrame, RK_S64>);
+static_assert(std::is_invocable_r_v<RK_U32, decltype(&mpp_packet_is_eoi), MppPacket>);
+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);

tests/probe_tests.cpp

@@ -0,0 +1,87 @@
+#include "rockstream/latency.hpp"
+#include "rockstream/probe.hpp"
+
+#include <array>
+#include <iostream>
+#include <sstream>
+#include <stdexcept>
+#include <string_view>
+
+namespace {
+void expect(bool condition, const char* message) {
+    if (!condition) throw std::runtime_error(message);
+}
+template<class F> 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;
+    }
+}