实现零拷贝

plugins/CMakeLists.txt

@@ -1,12 +1,32 @@
 set(RK_PLUGIN_OUTPUT_DIR ${CMAKE_BINARY_DIR}/plugins)
 
 option(RK3588_ENABLE_FFMPEG "Enable FFmpeg-based RTSP input" OFF)
+option(RK3588_ENABLE_MPP "Enable Rockchip MPP decode/encode" OFF)
 
 if(RK3588_ENABLE_FFMPEG)
     find_package(PkgConfig REQUIRED)
     pkg_check_modules(FFMPEG REQUIRED IMPORTED_TARGET libavformat libavcodec libavutil)
 endif()
 
+if(RK3588_ENABLE_MPP)
+    # Prefer user-provided lib path; otherwise search common locations including rknpu2 prebuilts.
+    find_library(RK_MPP_LIB rockchip_mpp
+        HINTS
+            ${RK_MPP_LIB_PATH}
+            ${RK_MPP_ROOT}/build
+            ${RK_MPP_ROOT}/lib
+            ${RK_RKNN_ROOT}/examples/3rdparty/mpp/Linux/aarch64
+            ${RK_RKNN_ROOT}/examples/3rdparty/mpp/Linux/armhf
+        NO_DEFAULT_PATH
+    )
+    if(NOT RK_MPP_LIB)
+        find_library(RK_MPP_LIB rockchip_mpp)
+    endif()
+    if(NOT RK_MPP_LIB)
+        message(WARNING "MPP enabled but rockchip_mpp library not found; disable RK3588_ENABLE_MPP or set RK_MPP_LIB_PATH")
+    endif()
+endif()
+
 add_library(input_rtsp SHARED input_rtsp/input_rtsp_node.cpp)
 target_include_directories(input_rtsp PRIVATE ${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/third_party)
 target_link_libraries(input_rtsp PRIVATE project_options Threads::Threads)
@@ -15,6 +35,14 @@ if(RK3588_ENABLE_FFMPEG)
     target_include_directories(input_rtsp PRIVATE ${FFMPEG_INCLUDE_DIRS})
     target_link_libraries(input_rtsp PRIVATE PkgConfig::FFMPEG)
 endif()
+if(RK3588_ENABLE_MPP AND RK_MPP_LIB)
+    target_compile_definitions(input_rtsp PRIVATE RK3588_ENABLE_MPP)
+    target_include_directories(input_rtsp PRIVATE
+        ${RK_MPP_ROOT}/inc
+        ${RK_RKNN_ROOT}/examples/3rdparty/mpp/include
+    )
+    target_link_libraries(input_rtsp PRIVATE ${RK_MPP_LIB})
+endif()
 set_target_properties(input_rtsp PROPERTIES
     OUTPUT_NAME "input_rtsp"
     LIBRARY_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR}

plugins/input_rtsp/input_rtsp_node.cpp

@@ -1,11 +1,22 @@
 #include <atomic>
 #include <chrono>
+#include <functional>
 #include <iostream>
 #include <thread>
 #include <vector>
 
 #include "node.h"
 
+#if defined(RK3588_ENABLE_MPP)
+extern "C" {
+#include <rockchip/mpp_buffer.h>
+#include <rockchip/mpp_frame.h>
+#include <rockchip/mpp_packet.h>
+#include <rockchip/rk_mpi.h>
+}
+#include <unistd.h>
+#endif
+
 #ifdef RK3588_ENABLE_FFMPEG
 extern "C" {
 #include <libavcodec/avcodec.h>
@@ -29,6 +40,7 @@ public:
         width_ = config.ValueOr<int>("width", 1920);
         height_ = config.ValueOr<int>("height", 1080);
         use_ffmpeg_ = config.ValueOr<bool>("use_ffmpeg", false);
+        use_mpp_ = config.ValueOr<bool>("use_mpp", true);
         if (ctx.output_queues.empty()) {
             std::cerr << "[input_rtsp] no downstream queue configured for node " << id_ << "\n";
             return false;
@@ -40,20 +52,37 @@ public:
     bool Start() override {
         if (!out_queue_) return false;
         running_.store(true);
-#if defined(RK3588_ENABLE_FFMPEG)
+#if defined(RK3588_ENABLE_FFMPEG) && defined(RK3588_ENABLE_MPP)
+        if (use_mpp_) {
+            worker_ = std::thread(&InputRtspNode::LoopFfmpegMpp, this);
+        } else if (use_ffmpeg_) {
+            worker_ = std::thread(&InputRtspNode::LoopFfmpegCpu, this);
+        } else {
+            worker_ = std::thread(&InputRtspNode::LoopStub, this);
+        }
+#elif defined(RK3588_ENABLE_FFMPEG)
         if (use_ffmpeg_) {
-            worker_ = std::thread(&InputRtspNode::LoopFfmpeg, this);
+            worker_ = std::thread(&InputRtspNode::LoopFfmpegCpu, this);
         } else {
             worker_ = std::thread(&InputRtspNode::LoopStub, this);
         }
 #else
-        if (use_ffmpeg_) {
-            std::cerr << "[input_rtsp] use_ffmpeg requested but not enabled at build time" << "\n";
+        if (use_ffmpeg_ || use_mpp_) {
+            std::cerr << "[input_rtsp] requested ffmpeg/mpp but not enabled at build time" << "\n";
         }
         worker_ = std::thread(&InputRtspNode::LoopStub, this);
 #endif
-        std::cout << "[input_rtsp] start url=" << url_ << " fps=" << fps_
-                  << (use_ffmpeg_ ? " (ffmpeg)" : " (stub)") << "\n";
+
+        std::cout << "[input_rtsp] start url=" << url_ << " fps=" << fps_;
+#if defined(RK3588_ENABLE_MPP)
+        if (use_mpp_) std::cout << " (ffmpeg demux + mpp decode)";
+        else if (use_ffmpeg_) std::cout << " (ffmpeg cpu decode)";
+        else std::cout << " (stub)";
+#else
+        if (use_ffmpeg_) std::cout << " (ffmpeg cpu decode)";
+        else std::cout << " (stub)";
+#endif
+        std::cout << "\n";
         return true;
     }
 
@@ -86,9 +115,8 @@ private:
             std::this_thread::sleep_for(frame_interval);
         }
     }
-
 #if defined(RK3588_ENABLE_FFMPEG)
-    void LoopFfmpeg() {
+    void LoopFfmpegCpu() {
         using namespace std::chrono;
         AVFormatContext* fmt_ctx = nullptr;
         AVCodecContext* codec_ctx = nullptr;
@@ -227,6 +255,251 @@ private:
     }
 #endif
 
+#if defined(RK3588_ENABLE_FFMPEG) && defined(RK3588_ENABLE_MPP)
+    struct MppDecoderWrapper {
+        ~MppDecoderWrapper() { Shutdown(); }
+
+        bool Init(MppCodingType type) {
+            MPP_RET ret = mpp_create(&ctx, &mpi);
+            if (ret != MPP_OK) return false;
+            ret = mpp_init(ctx, MPP_CTX_DEC, type);
+            if (ret != MPP_OK) return false;
+            MppDecCfg cfg = nullptr;
+            mpp_dec_cfg_init(&cfg);
+            ret = mpi->control(ctx, MPP_DEC_GET_CFG, cfg);
+            if (ret != MPP_OK) return false;
+            ret = mpp_dec_cfg_set_u32(cfg, "base:split_parse", 1);
+            if (ret != MPP_OK) return false;
+            ret = mpi->control(ctx, MPP_DEC_SET_CFG, cfg);
+            mpp_dec_cfg_deinit(cfg);
+            return ret == MPP_OK;
+        }
+
+        void Shutdown() {
+            if (packet) {
+                mpp_packet_deinit(&packet);
+                packet = nullptr;
+            }
+            if (frame_group) {
+                mpp_buffer_group_put(frame_group);
+                frame_group = nullptr;
+            }
+            if (ctx) {
+                mpp_destroy(ctx);
+                ctx = nullptr;
+                mpi = nullptr;
+            }
+        }
+
+        bool Decode(const uint8_t* data, size_t size, bool eos, int64_t pts_ms,
+                    const std::function<void(MppFrame)>& on_frame) {
+            if (!ctx || !mpi || !data || size == 0) return false;
+
+            if (!packet) {
+                if (mpp_packet_init(&packet, nullptr, 0) != MPP_OK) return false;
+            }
+            void* pkt_data = const_cast<uint8_t*>(data);
+            mpp_packet_set_data(packet, pkt_data);
+            mpp_packet_set_size(packet, size);
+            mpp_packet_set_pos(packet, pkt_data);
+            mpp_packet_set_length(packet, size);
+            mpp_packet_set_pts(packet, pts_ms);
+            if (eos) mpp_packet_set_eos(packet);
+
+            bool pkt_done = false;
+            while (true) {
+                if (!pkt_done) {
+                    if (mpi->decode_put_packet(ctx, packet) == MPP_OK) {
+                        pkt_done = true;
+                    } else {
+                        usleep(2000);
+                    }
+                }
+
+                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) break;
+
+                if (frame) {
+                    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;
+            }
+
+            mpp_packet_deinit(&packet);
+            packet = nullptr;
+            return true;
+        }
+
+        void HandleInfoChange() {
+            if (!ctx || !mpi || !frame) return;
+            MppBufferGroup group = frame_group;
+            size_t buf_size = mpp_frame_get_buf_size(frame);
+            if (!group) {
+                if (mpp_buffer_group_get_internal(&group, MPP_BUFFER_TYPE_DRM) != MPP_OK) return;
+                frame_group = group;
+            } else {
+                mpp_buffer_group_clear(group);
+            }
+            mpp_buffer_group_limit_config(group, buf_size, 24);
+            mpi->control(ctx, MPP_DEC_SET_EXT_BUF_GROUP, group);
+            mpi->control(ctx, MPP_DEC_SET_INFO_CHANGE_READY, nullptr);
+        }
+
+        MppCtx ctx = nullptr;
+        MppApi* mpi = nullptr;
+        MppBufferGroup frame_group = nullptr;
+        MppPacket packet = nullptr;
+        MppFrame frame = nullptr;
+    };
+
+    void LoopFfmpegMpp() {
+        using namespace std::chrono;
+        AVFormatContext* fmt_ctx = nullptr;
+        AVPacket* pkt = av_packet_alloc();
+        int video_stream = -1;
+        AVRational time_base{1, 1000};
+
+        if (avformat_open_input(&fmt_ctx, url_.c_str(), nullptr, nullptr) < 0) {
+            std::cerr << "[input_rtsp] avformat_open_input failed: " << url_ << "\n";
+            Cleanup(fmt_ctx, nullptr, pkt, nullptr);
+            LoopStub();
+            return;
+        }
+        if (avformat_find_stream_info(fmt_ctx, nullptr) < 0) {
+            std::cerr << "[input_rtsp] avformat_find_stream_info failed\n";
+            Cleanup(fmt_ctx, nullptr, pkt, nullptr);
+            LoopStub();
+            return;
+        }
+        for (unsigned i = 0; i < fmt_ctx->nb_streams; ++i) {
+            if (fmt_ctx->streams[i]->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) {
+                video_stream = static_cast<int>(i);
+                time_base = fmt_ctx->streams[i]->time_base;
+                break;
+            }
+        }
+        if (video_stream < 0) {
+            std::cerr << "[input_rtsp] no video stream\n";
+            Cleanup(fmt_ctx, nullptr, pkt, nullptr);
+            LoopStub();
+            return;
+        }
+
+        MppCodingType coding = MPP_VIDEO_CodingAVC;
+        auto codec_id = fmt_ctx->streams[video_stream]->codecpar->codec_id;
+        if (codec_id == AV_CODEC_ID_H264) coding = MPP_VIDEO_CodingAVC;
+        else if (codec_id == AV_CODEC_ID_HEVC) coding = MPP_VIDEO_CodingHEVC;
+        else {
+            std::cerr << "[input_rtsp] unsupported codec for mpp\n";
+            Cleanup(fmt_ctx, nullptr, pkt, nullptr);
+            LoopStub();
+            return;
+        }
+
+        MppDecoderWrapper dec;
+        if (!dec.Init(coding)) {
+            std::cerr << "[input_rtsp] mpp init failed\n";
+            Cleanup(fmt_ctx, nullptr, pkt, nullptr);
+            LoopStub();
+            return;
+        }
+
+        while (running_.load()) {
+            if (av_read_frame(fmt_ctx, pkt) < 0) {
+                std::this_thread::sleep_for(milliseconds(10));
+                continue;
+            }
+            if (pkt->stream_index != video_stream) {
+                av_packet_unref(pkt);
+                continue;
+            }
+            int64_t pts_ms = pkt->pts == AV_NOPTS_VALUE ? 0
+                              : av_rescale_q(pkt->pts, time_base, {1, 1000});
+            dec.Decode(pkt->data, pkt->size, false, pts_ms,
+                       [&](MppFrame frm) { PushFrameFromMpp(frm); });
+            av_packet_unref(pkt);
+        }
+
+        Cleanup(fmt_ctx, nullptr, pkt, nullptr);
+    }
+
+    void PushFrameFromMpp(MppFrame frm) {
+        MppBuffer buf = mpp_frame_get_buffer(frm);
+        if (!buf) return;
+        mpp_buffer_inc_ref(buf);
+        auto owner = std::shared_ptr<void>(buf, [](void* p) { mpp_buffer_put(reinterpret_cast<MppBuffer>(p)); });
+
+        int width = mpp_frame_get_width(frm);
+        int height = mpp_frame_get_height(frm);
+        int hor_stride = mpp_frame_get_hor_stride(frm);
+        int ver_stride = mpp_frame_get_ver_stride(frm);
+        MppFrameFormat mpp_fmt = mpp_frame_get_fmt(frm);
+
+        PixelFormat fmt = PixelFormat::UNKNOWN;
+        int plane_count = 0;
+        if (mpp_fmt == MPP_FMT_YUV420SP || mpp_fmt == MPP_FMT_YUV420SP_10BIT) {
+            fmt = PixelFormat::NV12;
+            plane_count = 2;
+        } else if (mpp_fmt == MPP_FMT_YUV420P || mpp_fmt == MPP_FMT_YUV420P10) {
+            fmt = PixelFormat::YUV420;
+            plane_count = 3;
+        }
+
+        if (plane_count == 0) {
+            return;
+        }
+
+        auto frame = std::make_shared<Frame>();
+        frame->width = width;
+        frame->height = height;
+        frame->format = fmt;
+        frame->stride = hor_stride;
+        frame->plane_count = plane_count;
+        frame->dma_fd = mpp_buffer_get_fd(buf);
+        frame->data = static_cast<uint8_t*>(mpp_buffer_get_ptr(buf));
+        frame->data_size = mpp_buffer_get_size(buf);
+        frame->data_owner = owner;
+        frame->frame_id = ++frame_id_;
+        frame->pts = mpp_frame_get_pts(frm);
+
+        if (fmt == PixelFormat::NV12 && plane_count == 2) {
+            int y_size = hor_stride * ver_stride;
+            frame->planes[0] = {frame->data, hor_stride, y_size, 0};
+            frame->planes[1] = {frame->data + y_size, hor_stride, y_size / 2, y_size};
+        } else if (fmt == PixelFormat::YUV420 && plane_count == 3) {
+            int y_size = hor_stride * ver_stride;
+            int uv_stride = hor_stride / 2;
+            int uv_h = ver_stride / 2;
+            int u_size = uv_stride * uv_h;
+            frame->planes[0] = {frame->data, hor_stride, y_size, 0};
+            frame->planes[1] = {frame->data + y_size, uv_stride, u_size, y_size};
+            frame->planes[2] = {frame->data + y_size + u_size, uv_stride, u_size, y_size + u_size};
+        }
+
+        out_queue_->Push(frame);
+
+        if (frame_id_ % 100 == 0) {
+            std::cout << "[input_rtsp] mpp frame " << frame->frame_id
+                      << " queue=" << out_queue_->Size()
+                      << " drops=" << out_queue_->DroppedCount() << "\n";
+        }
+    }
+#endif
+
     std::string id_;
     std::string url_;
     int fps_ = 25;
@@ -237,6 +510,7 @@ private:
     std::thread worker_;
     uint64_t frame_id_ = 0;
     bool use_ffmpeg_ = false;
+    bool use_mpp_ = true;
 };
 
 REGISTER_NODE(InputRtspNode, "input_rtsp");