Doni/safesight
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修bug
Some checks are pending
CI / host-build (push) Waiting to run
Details
CI / rk3588-cross-build (push) Waiting to run
Details

Browse Source
This commit is contained in:
sladro 2026-01-06 15:38:32 +08:00
parent caad46130f
commit d0fcb0593f
1 changed files with 110 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

176
plugins/preprocess/preprocess_node.cpp
View File

@ -1,6 +1,7 @@
#include <algorithm>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstring>
#include <iostream>
#include <memory>
@ -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<std::mutex> lock(mu_);
cv_.wait(lock, [&]() { return in_flight_ < max_inflight; });
++in_flight_;
}
void Release() {
std::lock_guard<std::mutex> 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<std::mutex> 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;
}