diff --git a/plugins/input_rtsp/input_rtsp_node.cpp b/plugins/input_rtsp/input_rtsp_node.cpp index 84130e6..e50da20 100644 --- a/plugins/input_rtsp/input_rtsp_node.cpp +++ b/plugins/input_rtsp/input_rtsp_node.cpp @@ -2,7 +2,10 @@ #include #include #include +#include #include +#include +#include #include #include @@ -25,6 +28,7 @@ extern "C" { #include #include #include +#include #include } #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(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(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(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(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(filt_pkt->size), + (filt_pkt->flags & AV_PKT_FLAG_KEY) != 0, pts_us); av_packet_unref(filt_pkt); } }