修bug

plugins/input_rtsp/input_rtsp_node.cpp

@@ -2,7 +2,10 @@
 #include <chrono>
 #include <functional>
 #include <iostream>
+#include <cstdio>
 #include <cstring>
+#include <mutex>
+#include <string>
 #include <thread>
 #include <vector>
 
@@ -25,6 +28,7 @@ extern "C" {
 #include <libavcodec/avcodec.h>
 #include <libavformat/avformat.h>
 #include <libavutil/imgutils.h>
+#include <libavutil/log.h>
 #include <libavutil/opt.h>
 }
 #endif
@@ -105,6 +109,40 @@ public:
     }
 
 private:
+#if defined(RK3588_ENABLE_FFMPEG)
+    static void InstallFfmpegLogFilterOnce() {
+        static std::once_flag once;
+        std::call_once(once, []() {
+            // Filter known benign RTSP probe noise while keeping real errors.
+            av_log_set_callback([](void* avcl, int level, const char* fmt, va_list vl) {
+                (void)level;
+                va_list vl_format;
+                va_copy(vl_format, vl);
+
+                char buf[1024];
+                vsnprintf(buf, sizeof(buf), fmt, vl_format);
+                va_end(vl_format);
+                buf[sizeof(buf) - 1] = '\0';
+
+                const char* item = avcl ? av_default_item_name(avcl) : "";
+                if (item && std::strstr(item, "rtsp") != nullptr) {
+                    // Seen during startup/probing on some cameras; not fatal if we later decode fine.
+                    if (std::strstr(buf, "decoding for stream") && std::strstr(buf, "failed")) {
+                        return;  // drop
+                    }
+                }
+
+                va_list vl_default;
+                va_copy(vl_default, vl);
+                av_log_default_callback(avcl, level, fmt, vl_default);
+                va_end(vl_default);
+            });
+
+            // Do not change global log level here; keep defaults and only filter specific noise.
+        });
+    }
+#endif
+
     void ApplyAffinity() {
         if (cpu_affinity_.empty()) return;
         std::string aerr;
@@ -146,6 +184,9 @@ private:
     }
 #if defined(RK3588_ENABLE_FFMPEG)
     void LoopFfmpegCpu() {
+#if defined(RK3588_ENABLE_FFMPEG)
+        InstallFfmpegLogFilterOnce();
+#endif
         ApplyAffinity();
         using namespace std::chrono;
 
@@ -162,6 +203,10 @@ private:
 
             AVDictionary* opts = nullptr;
             if (ffmpeg_force_tcp_) av_dict_set(&opts, "rtsp_transport", "tcp", 0);
+            av_dict_set(&opts, "analyzeduration", "0", 0);
+            av_dict_set(&opts, "probesize", "32768", 0);
+            av_dict_set(&opts, "fflags", "nobuffer", 0);
+            av_dict_set(&opts, "flags", "low_delay", 0);
             if (avformat_open_input(&fmt_ctx, url_.c_str(), nullptr, &opts) < 0) {
                 std::cerr << "[input_rtsp] avformat_open_input failed: " << url_ << "\n";
                 av_dict_free(&opts);
@@ -176,11 +221,8 @@ private:
             }
             av_dict_free(&opts);
             if (avformat_find_stream_info(fmt_ctx, nullptr) < 0) {
-                std::cerr << "[input_rtsp] avformat_find_stream_info failed\n";
-                Cleanup(fmt_ctx, codec_ctx, pkt, frm);
-                std::this_thread::sleep_for(seconds(backoff));
-                backoff = std::min(backoff_max, backoff * 2);
-                continue;
+                // For RTSP, codec parameters are often available from SDP already; keep going.
+                std::cerr << "[input_rtsp] avformat_find_stream_info failed (continuing)\n";
             }
             for (unsigned i = 0; i < fmt_ctx->nb_streams; ++i) {
                 if (fmt_ctx->streams[i]->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) {
@@ -327,6 +369,50 @@ private:
         return 0;
     }
 
+    struct NalSummary {
+        bool has_idr = false;
+        bool has_param = false;  // SPS/PPS (or VPS for H265)
+    };
+
+    static NalSummary SummarizeAnnexB(AVCodecID codec_id, const uint8_t* data, size_t size) {
+        NalSummary s{};
+        if (!data || size < 5) return s;
+
+        size_t i = 0;
+        while (i + 4 <= size) {
+            int sc = GetAnnexBStartCodeSize(data + i, size - i);
+            if (sc == 0) {
+                ++i;
+                continue;
+            }
+            size_t nal = i + static_cast<size_t>(sc);
+            if (nal >= size) break;
+            uint8_t b0 = data[nal];
+
+            if (codec_id == AV_CODEC_ID_H264) {
+                int t = b0 & 0x1F;
+                if (t == 5) s.has_idr = true;
+                if (t == 7 || t == 8) s.has_param = true;
+            } else if (codec_id == AV_CODEC_ID_HEVC) {
+                int t = (b0 >> 1) & 0x3F;
+                if (t == 19 || t == 20) s.has_idr = true;
+                if (t == 32 || t == 33 || t == 34) s.has_param = true;
+            }
+
+            // Jump to next start code by searching forward.
+            size_t j = nal + 1;
+            while (j + 3 < size) {
+                if ((data[j] == 0 && data[j + 1] == 0 && data[j + 2] == 1) ||
+                    (j + 4 < size && data[j] == 0 && data[j + 1] == 0 && data[j + 2] == 0 && data[j + 3] == 1)) {
+                    break;
+                }
+                ++j;
+            }
+            i = j;
+        }
+        return s;
+    }
+
     struct MppDecoderWrapper {
         ~MppDecoderWrapper() { Shutdown(); }
 
@@ -389,59 +475,55 @@ private:
             mpp_packet_clr_eos(packet);
             if (eos) mpp_packet_set_eos(packet);
 
-            bool pkt_done = false;
-            int put_tries = 0;
-            while (true) {
-                if (!pkt_done) {
-                    MPP_RET put_ret = mpi->decode_put_packet(ctx, packet);
-                    if (put_ret == MPP_OK) {
-                        pkt_done = true;
-                    } else {
-                        if (++fail_put_packet_ret_ <= 3 || fail_put_packet_ret_ % 200 == 0) {
-                            std::cerr << "[input_rtsp] decode_put_packet ret=" << put_ret
-                                      << " tries=" << put_tries << "\n";
-                        }
-                        usleep(2000);
-                        if (++put_tries >= 500) {
-                            if (++fail_put_packet_ <= 3 || fail_put_packet_ % 200 == 0) {
-                                std::cerr << "[input_rtsp] decode_put_packet timeout ret=" << put_ret
-                                          << " tries=" << put_tries << "\n";
-                            }
-                            mpp_packet_deinit(&packet);
-                            return false;
+            auto drain_frames = [&]() {
+                for (int n = 0; n < 8; ++n) {
+                    MPP_RET gr = mpi->decode_get_frame(ctx, &frame);
+                    if (gr == MPP_ERR_TIMEOUT) return;
+                    if (gr != MPP_OK) {
+                        if (++fail_get_frame_ <= 3 || fail_get_frame_ % 200 == 0) {
+                            std::cerr << "[input_rtsp] decode_get_frame ret=" << gr << "\n";
                         }
+                        return;
                     }
-                }
-
-                int get_frm = 0;
-                MPP_RET ret = mpi->decode_get_frame(ctx, &frame);
-                if (ret == MPP_ERR_TIMEOUT) {
-                    usleep(2000);
-                    continue;
-                }
-                if (ret != MPP_OK) {
-                    if (++fail_get_frame_ <= 3 || fail_get_frame_ % 200 == 0) {
-                        std::cerr << "[input_rtsp] decode_get_frame ret=" << ret << "\n";
-                    }
-                    break;
-                }
-
-                if (frame) {
+                    if (!frame) return;
                     if (mpp_frame_get_info_change(frame)) {
                         HandleInfoChange();
                     } else {
                         if (on_frame) on_frame(frame);
                     }
-                    get_frm = 1;
                     mpp_frame_deinit(&frame);
                     frame = nullptr;
                 }
+            };
 
-                if (get_frm) continue;
-                break;
+            bool pkt_done = false;
+            MPP_RET last_put = MPP_NOK;
+            int put_tries = 0;
+            while (!pkt_done && put_tries++ < 200) {
+                MPP_RET put_ret = mpi->decode_put_packet(ctx, packet);
+                if (put_ret == MPP_OK) {
+                    pkt_done = true;
+                    break;
+                }
+
+                last_put = put_ret;
+                // Buffer full is normal backpressure; try draining and retry.
+                drain_frames();
+                usleep(2000);
             }
 
+            // Always try to drain after put attempt(s).
+            drain_frames();
             mpp_packet_deinit(&packet);
+
+            if (!pkt_done) {
+                // Treat as packet drop to keep pipeline progressing; caller can decide to wait for next IDR.
+                if (last_put != MPP_ERR_BUFFER_FULL) {
+                    if (++fail_put_packet_ <= 3 || fail_put_packet_ % 200 == 0) {
+                        std::cerr << "[input_rtsp] decode_put_packet failed ret=" << last_put << "\n";
+                    }
+                }
+            }
             return pkt_done;
         }
 
@@ -472,6 +554,16 @@ private:
                     mpp_packet_deinit(&pkt);
                     return true;
                 }
+                // Drain output to relieve buffer full.
+                MppFrame tmp = nullptr;
+                if (mpi->decode_get_frame(ctx, &tmp) == MPP_OK && tmp) {
+                    if (mpp_frame_get_info_change(tmp)) {
+                        frame = tmp;
+                        HandleInfoChange();
+                        frame = nullptr;
+                    }
+                    mpp_frame_deinit(&tmp);
+                }
                 usleep(2000);
             }
 
@@ -512,11 +604,13 @@ private:
         MppFrame frame = nullptr;
         uint64_t fail_copy_init_ = 0;
         uint64_t fail_put_packet_ = 0;
-        uint64_t fail_put_packet_ret_ = 0;
         uint64_t fail_get_frame_ = 0;
     };
 
     void LoopFfmpegMpp() {
+#if defined(RK3588_ENABLE_FFMPEG)
+        InstallFfmpegLogFilterOnce();
+#endif
         ApplyAffinity();
         using namespace std::chrono;
 
@@ -533,6 +627,10 @@ private:
 
             AVDictionary* opts = nullptr;
             if (ffmpeg_force_tcp_) av_dict_set(&opts, "rtsp_transport", "tcp", 0);
+            av_dict_set(&opts, "analyzeduration", "0", 0);
+            av_dict_set(&opts, "probesize", "32768", 0);
+            av_dict_set(&opts, "fflags", "nobuffer", 0);
+            av_dict_set(&opts, "flags", "low_delay", 0);
             if (avformat_open_input(&fmt_ctx, url_.c_str(), nullptr, &opts) < 0) {
                 std::cerr << "[input_rtsp] avformat_open_input failed: " << url_ << "\n";
                 av_dict_free(&opts);
@@ -549,13 +647,8 @@ private:
             }
             av_dict_free(&opts);
             if (avformat_find_stream_info(fmt_ctx, nullptr) < 0) {
-                std::cerr << "[input_rtsp] avformat_find_stream_info failed\n";
-                if (bsf_ctx) av_bsf_free(&bsf_ctx);
-                if (filt_pkt) av_packet_free(&filt_pkt);
-                Cleanup(fmt_ctx, nullptr, pkt, nullptr);
-                std::this_thread::sleep_for(seconds(backoff));
-                backoff = std::min(backoff_max, backoff * 2);
-                continue;
+                // For RTSP, codec parameters are often available from SDP already; keep going.
+                std::cerr << "[input_rtsp] avformat_find_stream_info failed (continuing)\n";
             }
             for (unsigned i = 0; i < fmt_ctx->nb_streams; ++i) {
                 if (fmt_ctx->streams[i]->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) {
@@ -650,9 +743,40 @@ private:
             int read_fail = 0;
             uint64_t pkt_count = 0;
             uint64_t decode_fail = 0;
+            bool need_idr = true;
+            uint64_t dropped_until_idr = 0;
+            bool announced_wait_idr = false;
             const int64_t step_us = (fps_ > 0) ? (1000000LL / fps_) : 40000LL;
             int64_t last_pts_us = 0;
             bool have_last_pts = false;
+
+            auto handle_annexb = [&](const uint8_t* data, size_t size, bool key_flag, int64_t pts_us) {
+                if (!data || size == 0) return;
+                NalSummary ns = SummarizeAnnexB(codec_id, data, size);
+                const bool is_idr = key_flag || ns.has_idr;
+                if (need_idr && !is_idr) {
+                    ++dropped_until_idr;
+                    if (!announced_wait_idr || dropped_until_idr % 300 == 0) {
+                        std::cout << "[input_rtsp] waiting for IDR/keyframe; dropped=" << dropped_until_idr << "\n";
+                        announced_wait_idr = true;
+                    }
+                    return;
+                }
+
+                const bool ok = dec.Decode(data, size, false, pts_us,
+                                           [&](MppFrame frm) { PushFrameFromMpp(frm); });
+                if (is_idr && ok) {
+                    need_idr = false;
+                    decode_fail = 0;
+                } else if (!ok) {
+                    // If we are in steady state and decoding starts failing, re-sync on next IDR.
+                    if (!need_idr && ++decode_fail >= 3) {
+                        need_idr = true;
+                        announced_wait_idr = false;
+                        std::cerr << "[input_rtsp] decoder desync/backpressure; re-sync on next IDR\n";
+                    }
+                }
+            };
             while (running_.load()) {
                 if (av_read_frame(fmt_ctx, pkt) < 0) {
                     if (++read_fail >= 50) break;
@@ -666,7 +790,11 @@ private:
                 }
 
                 ++pkt_count;
-                int64_t raw_ts = (pkt->pts != AV_NOPTS_VALUE) ? pkt->pts : pkt->dts;
+                int64_t raw_pts = pkt->pts;
+                int64_t raw_dts = pkt->dts;
+                if (raw_pts != AV_NOPTS_VALUE && raw_pts < 0) raw_pts = AV_NOPTS_VALUE;
+                if (raw_dts != AV_NOPTS_VALUE && raw_dts < 0) raw_dts = AV_NOPTS_VALUE;
+                int64_t raw_ts = (raw_pts != AV_NOPTS_VALUE) ? raw_pts : raw_dts;
                 int64_t pts_us = 0;
                 if (raw_ts == AV_NOPTS_VALUE) {
                     pts_us = have_last_pts ? (last_pts_us + step_us) : 0;
@@ -682,11 +810,11 @@ private:
                     std::cout << "[input_rtsp] recv pkt#" << pkt_count
                               << " size=" << pkt->size
                               << " pts=";
-                    if (pkt->pts == AV_NOPTS_VALUE) std::cout << "NOPTS";
-                    else std::cout << pkt->pts;
+                    if (raw_pts == AV_NOPTS_VALUE) std::cout << "NOPTS";
+                    else std::cout << raw_pts;
                     std::cout << " dts=";
-                    if (pkt->dts == AV_NOPTS_VALUE) std::cout << "NOPTS";
-                    else std::cout << pkt->dts;
+                    if (raw_dts == AV_NOPTS_VALUE) std::cout << "NOPTS";
+                    else std::cout << raw_dts;
                     std::cout << " pts_us=" << pts_us
                               << " key=" << ((pkt->flags & AV_PKT_FLAG_KEY) ? 1 : 0)
                               << "\n";
@@ -706,21 +834,13 @@ private:
                                       << "\n";
                         }
                     }
-                    if (!dec.Decode(pkt->data, static_cast<size_t>(pkt->size), false, pts_us,
-                                    [&](MppFrame frm) { PushFrameFromMpp(frm); })) {
-                        if (++decode_fail <= 3 || decode_fail % 200 == 0) {
-                            std::cerr << "[input_rtsp] mpp Decode() failed (annexb), fail_count=" << decode_fail << "\n";
-                        }
-                    }
+                    handle_annexb(pkt->data, static_cast<size_t>(pkt->size),
+                                  (pkt->flags & AV_PKT_FLAG_KEY) != 0, pts_us);
                 } else if (ensure_bsf()) {
                     if (av_bsf_send_packet(bsf_ctx, pkt) == 0) {
                         while (av_bsf_receive_packet(bsf_ctx, filt_pkt) == 0) {
-                            if (!dec.Decode(filt_pkt->data, static_cast<size_t>(filt_pkt->size), false, pts_us,
-                                            [&](MppFrame frm) { PushFrameFromMpp(frm); })) {
-                                if (++decode_fail <= 3 || decode_fail % 200 == 0) {
-                                    std::cerr << "[input_rtsp] mpp Decode() failed (bsf), fail_count=" << decode_fail << "\n";
-                                }
-                            }
+                            handle_annexb(filt_pkt->data, static_cast<size_t>(filt_pkt->size),
+                                          (filt_pkt->flags & AV_PKT_FLAG_KEY) != 0, pts_us);
                             av_packet_unref(filt_pkt);
                         }
                     }