From d0fcb0593fedce7af5e5e7a434d29d84cbd32913 Mon Sep 17 00:00:00 2001 From: sladro Date: Tue, 6 Jan 2026 15:38:32 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AEbug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- plugins/preprocess/preprocess_node.cpp | 176 +++++++++++++++---------- 1 file changed, 110 insertions(+), 66 deletions(-) diff --git a/plugins/preprocess/preprocess_node.cpp b/plugins/preprocess/preprocess_node.cpp index 7001761..2a84e86 100644 --- a/plugins/preprocess/preprocess_node.cpp +++ b/plugins/preprocess/preprocess_node.cpp @@ -1,6 +1,7 @@ #include #include #include +#include #include #include #include @@ -41,9 +42,45 @@ int ToRgaFormat(PixelFormat fmt) { } } -std::mutex& GlobalRgaMutex() { - static std::mutex* mu = new std::mutex(); - return *mu; +constexpr int GlobalRgaMaxInflight() { + // Fixed concurrency cap for RGA submissions. + return 2; +} + +class RgaGate { +public: + void Acquire() { + const int max_inflight = GlobalRgaMaxInflight(); + std::unique_lock lock(mu_); + cv_.wait(lock, [&]() { return in_flight_ < max_inflight; }); + ++in_flight_; + } + + void Release() { + std::lock_guard lock(mu_); + if (in_flight_ > 0) { + --in_flight_; + } + cv_.notify_one(); + } + +private: + std::mutex mu_; + std::condition_variable cv_; + int in_flight_ = 0; +}; + +RgaGate& GlobalRgaGate() { + static RgaGate* g = new RgaGate(); + return *g; +} + +class ScopedRgaGate { +public: + ScopedRgaGate() { GlobalRgaGate().Acquire(); } + ~ScopedRgaGate() { GlobalRgaGate().Release(); } + ScopedRgaGate(const ScopedRgaGate&) = delete; + ScopedRgaGate& operator=(const ScopedRgaGate&) = delete; } void EnsureRgaInitializedOnce() { @@ -390,24 +427,27 @@ private: rga_buffer_t src_buf{}; rga_buffer_t dst_buf{}; - DmaBufferPtr src_dma_buf; // Keep alive if we allocate + DmaBufferPtr src_dma_buf; // keep alive if we allocate/copy + + // Prepare tmp buffer outside the RGA critical section. + DmaBufferPtr tmp_dma; + if (need_resize && need_cvt) { + tmp_dma = DmaAlloc(CalcImageSizeStrided(dst_wstride, dst_hstride, frame->format)); + if (!tmp_dma || !tmp_dma->valid()) { + std::cerr << "[preprocess] DMA alloc for tmp failed\n"; + return false; + } + } - // Allocate/copy outside the global RGA lock; librga calls are protected below. - // Important: some RGA drivers cannot map DMABUF whose physical address is >4GB. - // MPP decode outputs often come as dma_fd; to avoid kernel-side mapping failures - // (and downstream heap corruption), prefer copying into our own <4GB dma_heap buffer - // when CPU mapping is available. const bool can_cpu_read_src = (frame->data != nullptr && frame->data_size > 0); - const bool should_copy_src_for_rga = (frame->dma_fd >= 0 && can_cpu_read_src); - - if (should_copy_src_for_rga || (frame->dma_fd < 0 && frame->data)) { + auto CopySrcToDmaIfPossible = [&]() -> bool { + if (!can_cpu_read_src) return false; // Use tight/aligned strides for the copied source buffer. const int copy_wstride = Align16(frame->width); const int copy_hstride = Align16(frame->height); const size_t src_size = CalcImageSizeStrided(copy_wstride, copy_hstride, frame->format); src_dma_buf = DmaAlloc(src_size); if (!src_dma_buf || !src_dma_buf->valid()) { - std::cerr << "[preprocess] DMA alloc for src failed\n"; return false; } @@ -418,43 +458,43 @@ private: // CPU writes to a DMA-BUF must be flushed before RGA reads it. DmaSyncStartFd(src_dma_buf->fd); - if (!CopyToStridedBuffer(*frame, src_dma_buf->data(), src_dma_buf->size, copy_wstride, copy_hstride)) { - DmaSyncEndFd(src_dma_buf->fd); - if (frame->dma_fd >= 0) { - DmaSyncEndFd(frame->dma_fd); - } - std::cerr << "[preprocess] copy src to DMA failed\n"; - return false; - } + const bool ok = CopyToStridedBuffer(*frame, src_dma_buf->data(), src_dma_buf->size, + copy_wstride, copy_hstride); DmaSyncEndFd(src_dma_buf->fd); if (frame->dma_fd >= 0) { DmaSyncEndFd(frame->dma_fd); } + if (!ok) { + src_dma_buf.reset(); + return false; + } + // Update strides used for RGA to match the copied buffer. src_wstride = copy_wstride; src_hstride = copy_hstride; + return true; + }; + + // If there's no DMA fd, we must allocate/copy into a DMA-BUF for RGA. + if (frame->dma_fd < 0) { + if (!CopySrcToDmaIfPossible()) { + std::cerr << "[preprocess] no dma_fd and src copy failed\n"; + return false; + } } - // Serialize librga/im2d usage; multiple pipelines call RGA concurrently. - std::lock_guard lock(GlobalRgaMutex()); + auto RunRgaOnce = [&](int src_fd, std::string& err) -> bool { + // Serialize/limit librga/im2d usage; multiple pipelines call RGA concurrently. + ScopedRgaGate guard; - if (src_dma_buf && src_dma_buf->valid()) { - src_buf = wrapbuffer_fd_t(src_dma_buf->fd, frame->width, frame->height, + src_buf = wrapbuffer_fd_t(src_fd, frame->width, frame->height, src_wstride, src_hstride, src_fmt_rga); - } else if (frame->dma_fd >= 0) { - src_buf = wrapbuffer_fd_t(frame->dma_fd, frame->width, frame->height, - src_wstride, src_hstride, src_fmt_rga); - } else { - return false; - } + dst_buf = wrapbuffer_fd_t(dma_buf->fd, out_w, out_h, + dst_wstride, dst_hstride, dst_fmt_rga); - // Use DMA fd for destination buffer - dst_buf = wrapbuffer_fd_t(dma_buf->fd, out_w, out_h, - dst_wstride, dst_hstride, dst_fmt_rga); - - auto Check = [&](const rga_buffer_t& s, const rga_buffer_t& d) -> bool { + auto Check = [&](const rga_buffer_t& s, const rga_buffer_t& d) -> bool { const im_rect sr{0, 0, s.width, s.height}; const im_rect dr{0, 0, d.width, d.height}; // Do NOT call the imcheck(...) variadic macro with 0 extra args: @@ -463,44 +503,48 @@ private: const im_rect pr{0, 0, 0, 0}; const IM_STATUS chk = imcheck_t(s, d, pat, sr, dr, pr, 0); if (chk != IM_STATUS_NOERROR && chk != IM_STATUS_SUCCESS) { - std::cerr << "[preprocess] RGA imcheck failed: " << imStrError(chk) << "\n"; + err = std::string("RGA imcheck failed: ") + imStrError(chk); return false; } return true; }; - IM_STATUS status = IM_STATUS_SUCCESS; + IM_STATUS status = IM_STATUS_SUCCESS; - if (need_resize && need_cvt) { - // Allocate DMA buffer for intermediate result - auto tmp_dma = DmaAlloc(CalcImageSizeStrided(dst_wstride, dst_hstride, frame->format)); - if (!tmp_dma || !tmp_dma->valid()) { - std::cerr << "[preprocess] DMA alloc for tmp failed\n"; - return false; + if (need_resize && need_cvt) { + rga_buffer_t tmp = wrapbuffer_fd_t(tmp_dma->fd, out_w, out_h, + dst_wstride, dst_hstride, src_fmt_rga); + if (!Check(src_buf, tmp) || !Check(tmp, dst_buf)) { + return false; + } + status = imresize(src_buf, tmp, 0, 0, 0, 1, nullptr); + if (status == IM_STATUS_SUCCESS) { + status = imcvtcolor(tmp, dst_buf, src_fmt_rga, dst_fmt_rga, IM_COLOR_SPACE_DEFAULT, 1, nullptr); + } + } else if (need_resize) { + if (!Check(src_buf, dst_buf)) { + return false; + } + status = imresize(src_buf, dst_buf, 0, 0, 0, 1, nullptr); + } else if (need_cvt) { + if (!Check(src_buf, dst_buf)) { + return false; + } + status = imcvtcolor(src_buf, dst_buf, src_fmt_rga, dst_fmt_rga, IM_COLOR_SPACE_DEFAULT, 1, nullptr); } - rga_buffer_t tmp = wrapbuffer_fd_t(tmp_dma->fd, out_w, out_h, - dst_wstride, dst_hstride, src_fmt_rga); - if (!Check(src_buf, tmp) || !Check(tmp, dst_buf)) { - return false; - } - status = imresize(src_buf, tmp, 0, 0, 0, 1, nullptr); - if (status == IM_STATUS_SUCCESS) { - status = imcvtcolor(tmp, dst_buf, src_fmt_rga, dst_fmt_rga, IM_COLOR_SPACE_DEFAULT, 1, nullptr); - } - } else if (need_resize) { - if (!Check(src_buf, dst_buf)) { - return false; - } - status = imresize(src_buf, dst_buf, 0, 0, 0, 1, nullptr); - } else if (need_cvt) { - if (!Check(src_buf, dst_buf)) { - return false; - } - status = imcvtcolor(src_buf, dst_buf, src_fmt_rga, dst_fmt_rga, IM_COLOR_SPACE_DEFAULT, 1, nullptr); - } - if (status != IM_STATUS_SUCCESS) { - std::cerr << "[preprocess] RGA failed: " << imStrError(status) << "\n"; + if (status != IM_STATUS_SUCCESS) { + err = std::string("RGA failed: ") + imStrError(status); + return false; + } + return true; + }; + + std::string rga_err; + + const int src_fd = (src_dma_buf && src_dma_buf->valid()) ? src_dma_buf->fd : frame->dma_fd; + if (src_fd < 0 || !RunRgaOnce(src_fd, rga_err)) { + std::cerr << "[preprocess] " << (rga_err.empty() ? "RGA failed" : rga_err) << "\n"; return false; }