From 98cb2d95a7aa1841ff7a9dc7a7680cb1fcfa2a28 Mon Sep 17 00:00:00 2001 From: Nic Boet Date: Wed, 28 Oct 2026 17:22:44 -0500 Subject: [PATCH] fix: pair decoded frames with their packet by pts in Monitor::Decode() The codec can consume packets without returning a frame for them; H.264 does after lost or corrupt slices (RTSP over UDP). Decode() paired each frame with the oldest decoder_queue entry, so those entries stayed queued for good: every later frame went to an older packet, and once the queue passed 2 x the keyframe interval the throttle stopped sending while the codec waited for input, so decoding stopped until zmwatch restarted zmc. Match the frame to the entry with its pts and release the entries before it with an earlier pts (decoded, no image), as the flush paths do. A frame older than every entry is dropped; without a usable pts the oldest entry is used, as before. refs #4860 Co-Authored-By: Claude Opus 5.5 --- src/zm_monitor.cpp | 58 +++++++++++- src/zm_monitor.h | 3 + tests/CMakeLists.txt | 1 + tests/zm_monitor_decoder_queue.cpp | 137 +++++++++++++++++++++++++++++ 4 files changed, 195 insertions(+), 4 deletions(-) create mode 100644 tests/zm_monitor_decoder_queue.cpp diff --git a/src/zm_monitor.cpp b/src/zm_monitor.cpp index bab7781f4..3be726136 100644 --- a/src/zm_monitor.cpp +++ b/src/zm_monitor.cpp @@ -3534,6 +3534,49 @@ void Monitor::flushDecoderQueue() { packetqueue.notify_all(); // wake the analysis thread if it's waiting } +// The codec can consume packets without ever returning a frame for them: after +// lost or corrupt slices (RTP over UDP) H.264 drops pictures until it has one it +// can reference again. Pairing each frame with the oldest entry would leave those +// entries in decoder_queue for good, pair every later frame with an older packet, +// and once the queue passes the throttle in Decode() stop decoding altogether. +// Returns the entry whose packet has frame_pts, after releasing the entries before +// it with an earlier pts (decoded, no image). end() when the frame is older than +// every entry; begin() when there is no pts to go by or no entry has it. +std::list::iterator Monitor::MatchDecodedFrame(int64_t frame_pts) { + if (frame_pts == AV_NOPTS_VALUE) return decoder_queue.begin(); + + auto match = std::find_if(decoder_queue.begin(), decoder_queue.end(), + [frame_pts](const ZMPacketLock &lock) { + return lock.packet_ && lock.packet_->packet->pts == frame_pts; + }); + if (match == decoder_queue.end()) { + bool older_than_all = std::all_of(decoder_queue.begin(), decoder_queue.end(), + [frame_pts](const ZMPacketLock &lock) { + return lock.packet_ && lock.packet_->packet->pts != AV_NOPTS_VALUE && + lock.packet_->packet->pts > frame_pts; + }); + return older_than_all ? decoder_queue.end() : decoder_queue.begin(); + } + + size_t released = 0; + for (auto it = decoder_queue.begin(); it != match;) { + if (it->packet_->packet->pts != AV_NOPTS_VALUE && it->packet_->packet->pts < frame_pts) { + it->packet_->decoded = true; + it->packet_->notify_all(); + it = decoder_queue.erase(it); + released++; + } else { + ++it; // reordered (B-frame) reference still to come out + } + } + if (released) { + Debug(1, "Codec returned no frame for %zu packets before pts %" PRId64 ", released them, decoder queue size=%zu", + released, frame_pts, decoder_queue.size()); + packetqueue.notify_all(); + } + return match; +} + bool Monitor::Decode() { AVCodecContext *context = GetVideoCodecContext(); ZMPacketLock packet_lock; @@ -3575,10 +3618,17 @@ bool Monitor::Decode() { int ret = front_packet->receive_frame(context); if (ret > 0) { - // Success - got a decoded frame, take ownership and process it - packet_lock = std::move(decoder_queue.front()); - decoder_queue.pop_front(); - packet = front_packet; + // Success - got a decoded frame; hand it to the packet it came from + av_frame_ptr frame = std::move(front_packet->in_frame); + auto match = MatchDecodedFrame(frame->pts); + if (match == decoder_queue.end()) { + Debug(1, "Dropping decoded frame pts %" PRId64 ", older than every packet in the decoder queue", frame->pts); + return true; + } + packet_lock = std::move(*match); + decoder_queue.erase(match); + packet = packet_lock.packet_; + packet->in_frame = std::move(frame); Debug(2, "Received frame for packet %d, decoder queue pop size=%zu", packet->image_index, decoder_queue.size()); // Continue to PHASE 3 (frame processing) } else if (ret < 0) { diff --git a/src/zm_monitor.h b/src/zm_monitor.h index f0ab408e7..a18df0ba5 100644 --- a/src/zm_monitor.h +++ b/src/zm_monitor.h @@ -808,6 +808,9 @@ class Monitor : public std::enable_shared_from_this { int64_t KeepDtsContinuous(AVPacket *packet, AVRational time_base); std::unique_ptr Poller; std::list decoder_queue; + // The decoder_queue entry a decoded frame with this pts came from; releases + // the entries before it that the codec consumed without returning a frame. + std::list::iterator MatchDecodedFrame(int64_t frame_pts); packetqueue_iterator *analysis_it; std::unique_ptr analysis_thread; packetqueue_iterator *decoder_it; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index a2c2933cd..db3873433 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -18,6 +18,7 @@ set(TEST_SOURCES zm_db_logging_live.cpp zm_db_schema.cpp zm_monitor_dts.cpp + zm_monitor_decoder_queue.cpp zm_box.cpp zm_buffer.cpp zm_audio_detector.cpp diff --git a/tests/zm_monitor_decoder_queue.cpp b/tests/zm_monitor_decoder_queue.cpp new file mode 100644 index 000000000..63388c98e --- /dev/null +++ b/tests/zm_monitor_decoder_queue.cpp @@ -0,0 +1,137 @@ +/* + * This file is part of the ZoneMinder Project. See AUTHORS file for Copyright information + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for + * more details. + * + * You should have received a copy of the GNU General Public License along + * with this program. If not, see . + */ + +#include "zm_catch2.h" + +#include "zm_monitor.h" + +#include + +// Monitor::MatchDecodedFrame(): pairing a decoded frame with the packet it came +// from when the codec consumed some packets without returning a frame (RTP loss). + +namespace { + +void EnsureConfig() { + if (!config.font_file_location) config.font_file_location = ""; + if (!config.event_close_mode) config.event_close_mode = "idle"; +} + +class DecoderQueue : public Monitor { + public: + DecoderQueue() : Monitor() {} + + // Packets sent to the codec, in decode order, by pts. + std::vector> Send(const std::vector &pts) { + std::vector> sent; + for (int64_t p : pts) { + auto packet = std::make_shared(); + packet->packet->pts = p; + packet->codec_type = AVMEDIA_TYPE_VIDEO; + decoder_queue.emplace_back(packet); + sent.push_back(packet); + } + return sent; + } + + // The pts of the matched entry (or -1 for end(), -2 for begin() on an empty + // queue) and what stays queued after it is taken out, as Decode() does. + int64_t Receive(int64_t frame_pts) { + auto match = MatchDecodedFrame(frame_pts); + if (match == decoder_queue.end()) return -1; + int64_t pts = match->packet_->packet->pts; + decoder_queue.erase(match); + return pts; + } + + std::vector Queued() const { + std::vector pts; + for (const auto &lock : decoder_queue) pts.push_back(lock.packet_->packet->pts); + return pts; + } +}; + +} // namespace + +TEST_CASE("Monitor decoder queue: one frame per packet pairs in order", "[Monitor][decoder]") { + EnsureConfig(); + DecoderQueue q; + q.Send({0, 3600, 7200}); + CHECK(q.Receive(0) == 0); + CHECK(q.Receive(3600) == 3600); + CHECK(q.Queued() == std::vector{7200}); +} + +TEST_CASE("Monitor decoder queue: packets the codec dropped are released", "[Monitor][decoder]") { + EnsureConfig(); + DecoderQueue q; + auto sent = q.Send({0, 3600, 7200, 10800, 14400}); + // Corrupt slices: no frame for 0, 3600 and 7200, the next one is 10800. + CHECK(q.Receive(10800) == 10800); + CHECK(q.Queued() == std::vector{14400}); + for (int i = 0; i < 3; i++) { + CHECK(sent[i]->decoded); + CHECK(sent[i]->image == nullptr); + } + CHECK_FALSE(sent[4]->decoded); +} + +TEST_CASE("Monitor decoder queue: the queue does not grow over many losses", "[Monitor][decoder]") { + EnsureConfig(); + DecoderQueue q; + int64_t pts = 0; + for (int burst = 0; burst < 50; burst++) { + // 15 packets in, the codec returns frames for the last 11 only + std::vector in; + for (int i = 0; i < 15; i++) in.push_back(pts + i * 3600); + q.Send(in); + for (int i = 4; i < 15; i++) CHECK(q.Receive(in[i]) == in[i]); + pts += 15 * 3600; + } + CHECK(q.Queued().empty()); +} + +TEST_CASE("Monitor decoder queue: reordered frames keep the references still to come", "[Monitor][decoder]") { + EnsureConfig(); + DecoderQueue q; + // Decode order I P B B (pts 0, 3, 1, 2): output order 0 1 2 3. + q.Send({0, 3, 1, 2}); + CHECK(q.Receive(0) == 0); + CHECK(q.Receive(1) == 1); + CHECK(q.Queued() == std::vector{3, 2}); + CHECK(q.Receive(2) == 2); + CHECK(q.Receive(3) == 3); + CHECK(q.Queued().empty()); +} + +TEST_CASE("Monitor decoder queue: a frame older than every packet is dropped", "[Monitor][decoder]") { + EnsureConfig(); + DecoderQueue q; + q.Send({7200, 10800}); + CHECK(q.Receive(3600) == -1); + CHECK(q.Queued() == std::vector{7200, 10800}); +} + +TEST_CASE("Monitor decoder queue: without a pts the oldest entry is used", "[Monitor][decoder]") { + EnsureConfig(); + DecoderQueue q; + q.Send({0, 3600}); + CHECK(q.Receive(AV_NOPTS_VALUE) == 0); + // A pts no packet has, newer than one of them: also the oldest entry. + CHECK(q.Receive(99999) == 3600); + CHECK(q.Queued().empty()); +}