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()); +}