From becfbebc78ae335fc158970dff62d3c966c28667 Mon Sep 17 00:00:00 2001 From: Evgeny Demchenko Date: Sun, 24 May 2026 08:47:14 +0100 Subject: [PATCH] cuframes-rtsp-source: + --policy + --ack-timeout-ms CLI flags MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Opt-in для STRICT_WAIT policy (default остаётся DROP_OLDEST). Use case STRICT_WAIT: Frame integrity критичен (e.g. recording, frame-accurate analytics). Producer ждёт ack от всех subscribers перед wrap ring → no torn frames. Trade-off: slow consumer задерживает all (default 200ms timeout затем subscriber dropped from bitmap). Use case DROP_OLDEST (default): Low-latency real-time display (TV grid). Producer wraps freely; v0.3 per-slot CUDA events закрывают race без waiting. Validation: policy=wait + ack-timeout-ms<=0 = infinite hold dead consumer — warning + force к 200ms safe default. Co-Authored-By: Claude Opus 4.7 --- tools/cuframes-rtsp-source/main.cpp | 27 ++++++++++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/tools/cuframes-rtsp-source/main.cpp b/tools/cuframes-rtsp-source/main.cpp index 6a67d02..a5be25b 100644 --- a/tools/cuframes-rtsp-source/main.cpp +++ b/tools/cuframes-rtsp-source/main.cpp @@ -61,6 +61,8 @@ struct Args { bool realtime = false; // emulate -re у ffmpeg CLI: sleep по pts bool loop = false; // loop input на eof (для file://) bool enable_packet_ring = false; // v0.2 — публиковать encoded packets + std::string policy = "drop"; // "drop" = DROP_OLDEST, "wait" = STRICT_WAIT + int ack_timeout_ms = 200; // only used при policy=wait; <=0 = infinite (unsafe) }; static void print_usage() { @@ -78,6 +80,14 @@ static void print_usage() { " --loop loop input на EOF (только для file://)\n" " --enable-packet-ring v0.2: дополнительно публиковать encoded packets\n" " (для consumer'ов с -c:v copy, Frigate record path)\n" + " --policy MODE drop (default) = DROP_OLDEST — producer wrap'ает ring\n" + " без ожидания consumer ack. Подходит для multi-consumer.\n" + " wait = STRICT_WAIT — producer ждёт ack от всех subscribers\n" + " перед overwrite. Безопаснее для frame integrity, но slow\n" + " consumer задерживает all (default ack-timeout 200ms).\n" + " --ack-timeout-ms N только при --policy wait. Max wait для ack (default 200).\n" + " <=0 = infinite — НЕ РЕКОМЕНДУЕТСЯ (dead consumer вешает\n" + " producer навсегда).\n" " --verbose debug logs\n" " -h, --help this help\n"; } @@ -96,11 +106,23 @@ static int parse_args(int argc, char **argv, Args &a) { else if (s == "--realtime") a.realtime = true; else if (s == "--loop") a.loop = true; else if (s == "--enable-packet-ring") a.enable_packet_ring = true; + else if (s == "--policy") a.policy = next(); + else if (s == "--ack-timeout-ms") a.ack_timeout_ms = std::stoi(next()); else if (s == "--verbose") a.verbose = true; else if (s == "-h" || s == "--help") { print_usage(); std::exit(0); } else { std::cerr << "Unknown arg: " << s << "\n"; print_usage(); std::exit(1); } } if (a.rtsp_url.empty() || a.key.empty()) { print_usage(); return 1; } + if (a.policy != "drop" && a.policy != "wait") { + std::cerr << "Invalid --policy '" << a.policy << "' (use drop|wait)\n"; + return 1; + } + if (a.policy == "wait" && a.ack_timeout_ms <= 0) { + std::cerr << "WARNING: --policy wait + --ack-timeout-ms<=0 = infinite wait.\n" + << " Dead consumer повесит producer навсегда. Forcing к 200ms.\n" + << " Set явно --ack-timeout-ms 200 (или больше) чтобы убрать warning.\n"; + a.ack_timeout_ms = 200; + } return 0; } @@ -230,7 +252,10 @@ int main(int argc, char **argv) { po.width = width; po.height = height; po.format = CUFRAMES_FORMAT_NV12; - po.policy = CUFRAMES_POLICY_DROP_OLDEST; + po.policy = (a.policy == "wait") + ? CUFRAMES_POLICY_STRICT_WAIT + : CUFRAMES_POLICY_DROP_OLDEST; + po.consumer_ack_timeout_ms = a.ack_timeout_ms; po.cuda_device = a.cuda_device; po.ring_size = a.ring_size; /* для logging */