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

性能优化2，有待测试版本
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-12 17:53:55 +08:00
parent 2a14f765c4
commit b62391fa97
8 changed files with 758 additions and 219 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
include/ai_scheduler.h
View File

@ -34,6 +34,11 @@ struct InferInput {
int height = 0;
bool is_nhwc = true; // true: NHWC, false: NCHW
// Optional DMA-BUF input for RKNN zero-copy (best-effort).
// When dma_fd >= 0, AiScheduler will try rknn_create_mem_from_fd + rknn_set_io_mem.
int dma_fd = -1;
int dma_offset = 0;
#if defined(RK3588_ENABLE_RKNN)
// The actual data type of `data` passed to RKNN. Default preserves existing behavior.
rknn_tensor_type type = RKNN_TENSOR_UINT8;

276
include/utils/spsc_queue.h
View File

@ -1,10 +1,13 @@
#pragma once
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <mutex>
#include <thread>
#include <vector>
#include <utility>
@ -30,106 +33,96 @@ public:
SpscQueue(size_t capacity, QueueDropStrategy strategy)
: capacity_(capacity == 0 ? 1 : capacity), strategy_(strategy) {
buf_.resize(capacity_);
buffer_.resize(capacity_);
for (size_t i = 0; i < capacity_; ++i) {
buffer_[i].seq.store(i, std::memory_order_relaxed);
}
}
// Called when the queue transitions from empty -> non-empty (best-effort).
// Must be fast and must not call back into this queue.
void SetOnDataAvailable(std::function<void()> cb) {
std::lock_guard<std::mutex> lock(mu_);
std::lock_guard<std::mutex> lock(cb_mu_);
on_data_available_ = std::move(cb);
}
bool Push(T item) {
std::function<void()> on_data;
bool notify_data = false;
{
std::unique_lock<std::mutex> lock(mu_);
if (stop_) return false;
if (size_ >= capacity_) {
if (strategy_ == QueueDropStrategy::DropOldest) {
// Drop the oldest item.
head_ = (head_ + 1) % capacity_;
--size_;
++dropped_;
} else if (strategy_ == QueueDropStrategy::DropNewest) {
// Drop the incoming item.
++dropped_;
return true;
} else {
// Block until space is available
space_cv_.wait(lock, [&] { return size_ < capacity_ || stop_; });
if (stop_) return false;
while (true) {
if (stop_.load(std::memory_order_acquire)) return false;
if (TryEnqueue(item)) {
pushed_.fetch_add(1, std::memory_order_relaxed);
const size_t prev_size = size_.fetch_add(1, std::memory_order_acq_rel);
if (prev_size == 0) {
std::function<void()> cb;
{
std::lock_guard<std::mutex> lock(cb_mu_);
cb = on_data_available_;
}
if (cb) cb();
data_cv_.notify_one();
}
return true;
}
const bool was_empty = (size_ == 0);
buf_[tail_] = std::move(item);
tail_ = (tail_ + 1) % capacity_;
++size_;
++pushed_;
// Avoid per-item wakeups when consumer is already running.
if (was_empty) {
notify_data = true;
on_data = on_data_available_;
// Queue full.
if (strategy_ == QueueDropStrategy::DropNewest) {
dropped_.fetch_add(1, std::memory_order_relaxed);
return true;
}
}
if (on_data) {
on_data();
if (strategy_ == QueueDropStrategy::DropOldest) {
if (TryDropOneOldest()) {
continue;
}
std::this_thread::yield();
continue;
}
// Block until space is available.
std::unique_lock<std::mutex> lock(wait_mu_);
space_cv_.wait(lock, [&] {
return stop_.load(std::memory_order_acquire) ||
size_.load(std::memory_order_acquire) < capacity_;
});
}
if (notify_data) {
data_cv_.notify_one();
}
return true;
}
bool Pop(T& out, std::chrono::milliseconds timeout) {
std::unique_lock<std::mutex> lock(mu_);
if (!data_cv_.wait_for(lock, timeout, [&] { return size_ > 0 || stop_; })) {
if (TryPop(out)) return true;
std::unique_lock<std::mutex> lock(wait_mu_);
if (!data_cv_.wait_for(lock, timeout, [&] {
return stop_.load(std::memory_order_acquire) ||
size_.load(std::memory_order_acquire) > 0;
})) {
return false;
}
if (size_ == 0) return false;
const bool was_full = (size_ >= capacity_);
out = std::move(buf_[head_]);
head_ = (head_ + 1) % capacity_;
--size_;
++popped_;
if (was_full) {
space_cv_.notify_one();
}
return true;
return TryPop(out);
}
// Blocks until an item is available or the queue is stopped.
bool Pop(T& out) {
std::unique_lock<std::mutex> lock(mu_);
data_cv_.wait(lock, [&] { return size_ > 0 || stop_; });
if (size_ == 0) return false;
const bool was_full = (size_ >= capacity_);
out = std::move(buf_[head_]);
head_ = (head_ + 1) % capacity_;
--size_;
++popped_;
if (was_full) {
space_cv_.notify_one();
while (true) {
if (TryPop(out)) return true;
std::unique_lock<std::mutex> lock(wait_mu_);
data_cv_.wait(lock, [&] {
return stop_.load(std::memory_order_acquire) ||
size_.load(std::memory_order_acquire) > 0;
});
if (stop_.load(std::memory_order_acquire) &&
size_.load(std::memory_order_acquire) == 0) {
return false;
}
}
return true;
}
// Non-blocking pop.
bool TryPop(T& out) {
std::unique_lock<std::mutex> lock(mu_);
if (size_ == 0) return false;
const bool was_full = (size_ >= capacity_);
out = std::move(buf_[head_]);
head_ = (head_ + 1) % capacity_;
--size_;
++popped_;
if (was_full) {
if (!TryDequeue(out)) return false;
popped_.fetch_add(1, std::memory_order_relaxed);
const size_t prev_size = size_.fetch_sub(1, std::memory_order_acq_rel);
if (prev_size == capacity_) {
// Queue was full before this pop.
space_cv_.notify_one();
}
return true;
@ -138,85 +131,144 @@ public:
// Pop up to max_items currently available (non-blocking). Returns true if any item was popped.
bool TryPopBatch(std::vector<T>& out, size_t max_items) {
if (max_items == 0) return false;
std::unique_lock<std::mutex> lock(mu_);
if (size_ == 0) return false;
const bool was_full = (size_ >= capacity_);
const size_t n = (size_ < max_items) ? size_ : max_items;
out.clear();
out.reserve(n);
for (size_t i = 0; i < n; ++i) {
out.push_back(std::move(buf_[head_]));
head_ = (head_ + 1) % capacity_;
out.reserve(max_items);
for (size_t i = 0; i < max_items; ++i) {
T item;
if (!TryPop(item)) break;
out.push_back(std::move(item));
}
size_ -= n;
popped_ += n;
if (was_full) {
space_cv_.notify_one();
}
return true;
return !out.empty();
}
void Stop() {
std::lock_guard<std::mutex> lock(mu_);
stop_ = true;
stop_.store(true, std::memory_order_release);
data_cv_.notify_all();
space_cv_.notify_all();
}
bool IsStopped() const {
std::lock_guard<std::mutex> lock(mu_);
return stop_;
return stop_.load(std::memory_order_acquire);
}
size_t Size() const {
std::lock_guard<std::mutex> lock(mu_);
return size_;
return size_.load(std::memory_order_acquire);
}
size_t Capacity() const { return capacity_; }
size_t DroppedCount() const {
std::lock_guard<std::mutex> lock(mu_);
return dropped_;
return dropped_.load(std::memory_order_acquire);
}
size_t PushedCount() const {
std::lock_guard<std::mutex> lock(mu_);
return pushed_;
return pushed_.load(std::memory_order_acquire);
}
size_t PoppedCount() const {
std::lock_guard<std::mutex> lock(mu_);
return popped_;
return popped_.load(std::memory_order_acquire);
}
Stats GetStats() const {
std::lock_guard<std::mutex> lock(mu_);
Stats s;
s.size = size_;
s.size = size_.load(std::memory_order_acquire);
s.capacity = capacity_;
s.dropped = dropped_;
s.pushed = pushed_;
s.popped = popped_;
s.stopped = stop_;
s.dropped = dropped_.load(std::memory_order_acquire);
s.pushed = pushed_.load(std::memory_order_acquire);
s.popped = popped_.load(std::memory_order_acquire);
s.stopped = stop_.load(std::memory_order_acquire);
return s;
}
private:
struct Cell {
std::atomic<size_t> seq;
T data;
};
// Lock-free bounded MPMC queue (Vyukov). We still expose it as SPSC in this project,
// but DropOldest may dequeue from the producer thread.
bool TryEnqueue(T& item) {
size_t pos = enqueue_pos_.load(std::memory_order_relaxed);
while (true) {
Cell& cell = buffer_[pos % capacity_];
const size_t seq = cell.seq.load(std::memory_order_acquire);
const intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos);
if (dif == 0) {
if (enqueue_pos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) {
cell.data = std::move(item);
cell.seq.store(pos + 1, std::memory_order_release);
return true;
}
} else if (dif < 0) {
return false; // full
} else {
pos = enqueue_pos_.load(std::memory_order_relaxed);
}
}
}
bool TryDequeue(T& out) {
size_t pos = dequeue_pos_.load(std::memory_order_relaxed);
while (true) {
Cell& cell = buffer_[pos % capacity_];
const size_t seq = cell.seq.load(std::memory_order_acquire);
const intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos + 1);
if (dif == 0) {
if (dequeue_pos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) {
out = std::move(cell.data);
cell.seq.store(pos + capacity_, std::memory_order_release);
return true;
}
} else if (dif < 0) {
return false; // empty
} else {
pos = dequeue_pos_.load(std::memory_order_relaxed);
}
}
}
bool TryDropOneOldest() {
size_t pos = dequeue_pos_.load(std::memory_order_relaxed);
while (true) {
Cell& cell = buffer_[pos % capacity_];
const size_t seq = cell.seq.load(std::memory_order_acquire);
const intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos + 1);
if (dif == 0) {
if (dequeue_pos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) {
// Drop without moving data out.
cell.seq.store(pos + capacity_, std::memory_order_release);
size_.fetch_sub(1, std::memory_order_acq_rel);
dropped_.fetch_add(1, std::memory_order_relaxed);
return true;
}
} else if (dif < 0) {
return false;
} else {
pos = dequeue_pos_.load(std::memory_order_relaxed);
}
}
}
size_t capacity_ = 0;
QueueDropStrategy strategy_ = QueueDropStrategy::DropOldest;
mutable std::mutex mu_;
alignas(64) std::atomic<size_t> enqueue_pos_{0};
alignas(64) std::atomic<size_t> dequeue_pos_{0};
std::vector<Cell> buffer_;
alignas(64) std::atomic<size_t> size_{0};
std::atomic<bool> stop_{false};
std::atomic<size_t> dropped_{0};
std::atomic<size_t> pushed_{0};
std::atomic<size_t> popped_{0};
mutable std::mutex wait_mu_;
std::condition_variable data_cv_;
std::condition_variable space_cv_;
mutable std::mutex cb_mu_;
std::function<void()> on_data_available_;
std::vector<T> buf_;
size_t head_ = 0;
size_t tail_ = 0;
size_t size_ = 0;
bool stop_ = false;
size_t dropped_ = 0;
size_t pushed_ = 0;
size_t popped_ = 0;
};
} // namespace rk3588

10
plugins/CMakeLists.txt
View File

@ -290,9 +290,17 @@ set_target_properties(tracker PROPERTIES
)
# osd plugin (on-screen display for detection results)
add_library(osd SHARED osd/osd_node.cpp)
add_library(osd SHARED
osd/osd_node.cpp
${CMAKE_SOURCE_DIR}/src/utils/dma_alloc.cpp
)
target_include_directories(osd PRIVATE ${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/third_party)
target_link_libraries(osd PRIVATE project_options Threads::Threads)
if(RK3588_ENABLE_RGA AND RK_RGA_LIB)
target_compile_definitions(osd PRIVATE RK3588_ENABLE_RGA)
target_include_directories(osd PRIVATE ${RK_RGA_INCLUDE_DIR})
target_link_libraries(osd PRIVATE ${RK_RGA_LIB})
endif()
set_target_properties(osd PROPERTIES
OUTPUT_NAME "osd"
LIBRARY_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR}

10
plugins/ai_yolo/ai_yolo_node.cpp
View File

@ -2,6 +2,7 @@
#include <atomic>
#include <chrono>
#include <cmath>
#include <cstddef>
#include <cstring>
#include <iostream>
#include <memory>
@ -427,6 +428,15 @@ private:
if (static_cast<size_t>(src_stride) == packed_row && frame->data_size >= packed_size) {
input.data = src;
input.size = packed_size;
// Best-effort RKNN DMA-BUF zero-copy path.
if (frame->dma_fd >= 0 && frame->data) {
const ptrdiff_t off = src - frame->data;
if (off >= 0 && static_cast<size_t>(off) + packed_size <= frame->data_size) {
input.dma_fd = frame->dma_fd;
input.dma_offset = static_cast<int>(off);
}
}
} else {
if (frame->data_size < static_cast<size_t>(src_stride) * static_cast<size_t>(h)) {
std::cerr << "[ai_yolo] invalid RGB buffer size/stride (data_size=" << frame->data_size

372
plugins/osd/osd_node.cpp
View File

@ -10,8 +10,15 @@
#include "face/face_result.h"
#include "node.h"
#include "utils/dma_alloc.h"
#include "utils/logger.h"
#if defined(RK3588_ENABLE_RGA)
#include "im2d.hpp"
#include "im2d_buffer.h"
#include "im2d_type.h"
#endif
namespace rk3588 {
namespace {
@ -57,6 +64,62 @@ inline int Clamp(int val, int min_val, int max_val) {
return val < min_val ? min_val : (val > max_val ? max_val : val);
}
#if defined(RK3588_ENABLE_RGA)
inline uint32_t PackColorArgb(const Color& c) {
return (0xFFu << 24) | (static_cast<uint32_t>(c.r) << 16) |
(static_cast<uint32_t>(c.g) << 8) | static_cast<uint32_t>(c.b);
}
inline bool WrapFrameAsRgaDst(const Frame& frame, rga_buffer_t& dst, int& rga_stride_bytes) {
if (frame.dma_fd < 0) return false;
int rga_fmt = 0;
int stride_bytes = 0;
int wstride = 0;
int hstride = frame.height;
if (frame.format == PixelFormat::RGB) {
rga_fmt = RK_FORMAT_RGB_888;
stride_bytes = frame.planes[0].stride > 0 ? frame.planes[0].stride : (frame.stride > 0 ? frame.stride : frame.width * 3);
wstride = stride_bytes / 3;
if (frame.planes[0].size > 0 && stride_bytes > 0) {
const int hs = frame.planes[0].size / stride_bytes;
if (hs >= frame.height) hstride = hs;
}
} else if (frame.format == PixelFormat::BGR) {
rga_fmt = RK_FORMAT_BGR_888;
stride_bytes = frame.planes[0].stride > 0 ? frame.planes[0].stride : (frame.stride > 0 ? frame.stride : frame.width * 3);
wstride = stride_bytes / 3;
if (frame.planes[0].size > 0 && stride_bytes > 0) {
const int hs = frame.planes[0].size / stride_bytes;
if (hs >= frame.height) hstride = hs;
}
} else if (frame.format == PixelFormat::NV12) {
rga_fmt = RK_FORMAT_YCbCr_420_SP;
stride_bytes = frame.planes[0].stride > 0 ? frame.planes[0].stride : (frame.stride > 0 ? frame.stride : frame.width);
wstride = stride_bytes;
if (frame.planes[0].size > 0 && stride_bytes > 0) {
const int hs = frame.planes[0].size / stride_bytes;
if (hs >= frame.height) hstride = hs;
}
} else if (frame.format == PixelFormat::YUV420) {
rga_fmt = RK_FORMAT_YCbCr_420_P;
stride_bytes = frame.planes[0].stride > 0 ? frame.planes[0].stride : (frame.stride > 0 ? frame.stride : frame.width);
wstride = stride_bytes;
if (frame.planes[0].size > 0 && stride_bytes > 0) {
const int hs = frame.planes[0].size / stride_bytes;
if (hs >= frame.height) hstride = hs;
}
} else {
return false;
}
rga_stride_bytes = stride_bytes;
dst = wrapbuffer_fd_t(frame.dma_fd, frame.width, frame.height, wstride, hstride, rga_fmt);
return true;
}
#endif
void DrawHLine(uint8_t* data, int w, int h, int stride, PixelFormat fmt,
int x1, int x2, int y, int thickness, const Color& color) {
x1 = Clamp(x1, 0, w - 1);
@ -290,6 +353,7 @@ public:
draw_face_det_ = config.ValueOr<bool>("draw_face_det", draw_face_det_);
line_width_ = config.ValueOr<int>("line_width", 2);
font_scale_ = config.ValueOr<int>("font_scale", 1);
use_rga_bbox_ = config.ValueOr<bool>("use_rga_bbox", use_rga_bbox_);
if (const SimpleJson* dbg = config.Find("debug"); dbg && dbg->IsObject()) {
stats_log_ = dbg->ValueOr<bool>("stats", stats_log_);
@ -321,7 +385,13 @@ public:
bool Start() override {
LogInfo(std::string("[osd] start id=") + id_ + " draw_bbox=" + (draw_bbox_ ? "true" : "false") +
" draw_text=" + (draw_text_ ? "true" : "false") +
" draw_face_det=" + (draw_face_det_ ? "true" : "false"));
" draw_face_det=" + (draw_face_det_ ? "true" : "false") +
" use_rga_bbox=" + (use_rga_bbox_ ? "true" : "false"));
#if !defined(RK3588_ENABLE_RGA)
if (use_rga_bbox_ && draw_bbox_) {
LogError("[osd] use_rga_bbox=true but RK3588_ENABLE_RGA is disabled at build; bbox will NOT be drawn id=" + id_);
}
#endif
return true;
}
@ -355,6 +425,83 @@ private:
static FramePtr CloneFrameForWrite(const FramePtr& src) {
if (!src || !src->data || src->data_size == 0) return src;
// Prefer DMA clone when src already lives in DMA-BUF.
if (src->dma_fd >= 0) {
if (auto dma = DmaAlloc(src->data_size); dma && dma->valid()) {
#if defined(RK3588_ENABLE_RGA)
// Fast path: RGA copy.
rga_buffer_t rga_src{};
rga_buffer_t rga_dst{};
int dummy_stride = 0;
Frame dst_frame = *src;
dst_frame.dma_fd = dma->fd;
dst_frame.data = dma->data();
dst_frame.data_size = dma->size;
dst_frame.data_owner = dma;
if (WrapFrameAsRgaDst(*src, rga_src, dummy_stride) &&
WrapFrameAsRgaDst(dst_frame, rga_dst, dummy_stride)) {
if (imcopy(rga_src, rga_dst, 1, nullptr) == IM_STATUS_SUCCESS) {
auto out = std::make_shared<Frame>(*src);
out->dma_fd = dma->fd;
out->data = dma->data();
out->data_size = dma->size;
out->data_owner = dma;
for (int i = 0; i < 3; ++i) {
FramePlane p = src->planes[i];
int offset = p.offset;
if (p.data && src->data) {
const ptrdiff_t d = p.data - src->data;
if (d >= 0 && static_cast<size_t>(d) < src->data_size) {
offset = static_cast<int>(d);
}
}
p.offset = offset;
if (offset >= 0 && static_cast<size_t>(offset) < out->data_size) {
p.data = out->data + offset;
} else {
p.data = nullptr;
}
out->planes[i] = p;
}
return out;
}
}
#endif
// Fallback: CPU memcpy (requires cache sync on linux).
DmaSyncStartFd(src->dma_fd);
DmaSyncStart(dma.get());
memcpy(dma->data(), src->data, src->data_size);
DmaSyncEnd(dma.get());
DmaSyncEndFd(src->dma_fd);
auto out = std::make_shared<Frame>(*src);
out->dma_fd = dma->fd;
out->data = dma->data();
out->data_size = dma->size;
out->data_owner = dma;
for (int i = 0; i < 3; ++i) {
FramePlane p = src->planes[i];
int offset = p.offset;
if (p.data && src->data) {
const ptrdiff_t d = p.data - src->data;
if (d >= 0 && static_cast<size_t>(d) < src->data_size) {
offset = static_cast<int>(d);
}
}
p.offset = offset;
if (offset >= 0 && static_cast<size_t>(offset) < out->data_size) {
p.data = out->data + offset;
} else {
p.data = nullptr;
}
out->planes[i] = p;
}
return out;
}
}
auto buf = std::make_shared<std::vector<uint8_t>>(src->data_size);
memcpy(buf->data(), src->data, src->data_size);
@ -409,6 +556,61 @@ private:
uint8_t* data = frame->planes[0].data ? frame->planes[0].data : frame->data;
PixelFormat fmt = frame->format;
const bool want_rga_bbox = use_rga_bbox_ && draw_bbox_;
bool rga_bbox_ok = false;
#if defined(RK3588_ENABLE_RGA)
if (want_rga_bbox) {
if (frame->dma_fd < 0) {
LogRgaBboxErrorThrottled("frame has no dma_fd");
} else if (!(fmt == PixelFormat::RGB || fmt == PixelFormat::BGR ||
fmt == PixelFormat::NV12 || fmt == PixelFormat::YUV420)) {
LogRgaBboxErrorThrottled("unsupported format=" + std::to_string(static_cast<int>(fmt)));
} else {
rga_buffer_t dst{};
int rga_stride_bytes = 0;
if (!WrapFrameAsRgaDst(*frame, dst, rga_stride_bytes)) {
LogRgaBboxErrorThrottled("WrapFrameAsRgaDst failed");
} else {
(void)rga_stride_bytes;
im_job_handle_t job = imbeginJob();
if (!job) {
LogRgaBboxErrorThrottled("imbeginJob failed");
} else {
bool ok = true;
for (const auto& det : frame->det->items) {
int x = Clamp(static_cast<int>(det.bbox.x), 0, w - 1);
int y = Clamp(static_cast<int>(det.bbox.y), 0, h - 1);
int rw = static_cast<int>(det.bbox.w);
int rh = static_cast<int>(det.bbox.h);
rw = Clamp(rw, 1, w - x);
rh = Clamp(rh, 1, h - y);
im_rect rect{x, y, rw, rh};
const uint32_t c = PackColorArgb(GetClassColor(det.cls_id));
if (imrectangleTask(job, dst, rect, c, line_width_) <= 0) {
ok = false;
break;
}
}
if (ok) {
ok = (imendJob(job, IM_SYNC, 0, nullptr) > 0);
} else {
(void)imendJob(job, IM_SYNC, 0, nullptr);
}
rga_bbox_ok = ok;
if (!rga_bbox_ok) {
LogRgaBboxErrorThrottled("imrectangle/imendJob failed");
}
}
}
}
}
#else
if (want_rga_bbox) {
LogRgaBboxErrorThrottled("RK3588_ENABLE_RGA is disabled at build");
}
#endif
int stride;
if (fmt == PixelFormat::RGB || fmt == PixelFormat::BGR) {
stride = frame->planes[0].stride > 0 ? frame->planes[0].stride : (frame->stride > 0 ? frame->stride : w * 3);
@ -418,6 +620,13 @@ private:
return; // Unsupported format
}
const bool do_cpu_bbox = draw_bbox_ && !use_rga_bbox_;
const bool do_cpu_text = draw_text_;
if ((do_cpu_bbox || do_cpu_text) && frame->dma_fd >= 0) {
DmaSyncStartFd(frame->dma_fd);
}
for (const auto& det : frame->det->items) {
int x1 = static_cast<int>(det.bbox.x);
int y1 = static_cast<int>(det.bbox.y);
@ -426,11 +635,11 @@ private:
Color color = GetClassColor(det.cls_id);
if (draw_bbox_) {
if (do_cpu_bbox) {
DrawRect(data, w, h, stride, fmt, x1, y1, x2, y2, line_width_, color);
}
if (draw_text_) {
if (do_cpu_text) {
char label[96];
if (det.track_id >= 0) {
snprintf(label, sizeof(label), "%s %.0f%% id=%d", GetLabel(det.cls_id), det.score * 100, det.track_id);
@ -442,6 +651,10 @@ private:
DrawText(data, w, h, stride, fmt, x1, text_y, label, font_scale_, color);
}
}
if ((do_cpu_bbox || do_cpu_text) && frame->dma_fd >= 0) {
DmaSyncEndFd(frame->dma_fd);
}
}
void DrawFaceRecog(FramePtr frame) {
@ -452,6 +665,61 @@ private:
uint8_t* data = frame->planes[0].data ? frame->planes[0].data : frame->data;
PixelFormat fmt = frame->format;
const bool want_rga_bbox = use_rga_bbox_ && draw_bbox_;
bool rga_bbox_ok = false;
#if defined(RK3588_ENABLE_RGA)
if (want_rga_bbox) {
if (frame->dma_fd < 0) {
LogRgaBboxErrorThrottled("frame has no dma_fd");
} else if (!(fmt == PixelFormat::RGB || fmt == PixelFormat::BGR ||
fmt == PixelFormat::NV12 || fmt == PixelFormat::YUV420)) {
LogRgaBboxErrorThrottled("unsupported format=" + std::to_string(static_cast<int>(fmt)));
} else {
rga_buffer_t dst{};
int rga_stride_bytes = 0;
if (!WrapFrameAsRgaDst(*frame, dst, rga_stride_bytes)) {
LogRgaBboxErrorThrottled("WrapFrameAsRgaDst failed");
} else {
(void)rga_stride_bytes;
im_job_handle_t job = imbeginJob();
if (!job) {
LogRgaBboxErrorThrottled("imbeginJob failed");
} else {
bool ok = true;
for (const auto& it : frame->face_recog->items) {
int x = Clamp(static_cast<int>(it.bbox.x), 0, w - 1);
int y = Clamp(static_cast<int>(it.bbox.y), 0, h - 1);
int rw = static_cast<int>(it.bbox.w);
int rh = static_cast<int>(it.bbox.h);
rw = Clamp(rw, 1, w - x);
rh = Clamp(rh, 1, h - y);
im_rect rect{x, y, rw, rh};
const Color color = it.unknown ? Color{255, 0, 0} : Color{0, 255, 0};
if (imrectangleTask(job, dst, rect, PackColorArgb(color), line_width_) <= 0) {
ok = false;
break;
}
}
if (ok) {
ok = (imendJob(job, IM_SYNC, 0, nullptr) > 0);
} else {
(void)imendJob(job, IM_SYNC, 0, nullptr);
}
rga_bbox_ok = ok;
if (!rga_bbox_ok) {
LogRgaBboxErrorThrottled("imrectangle/imendJob failed");
}
}
}
}
}
#else
if (want_rga_bbox) {
LogRgaBboxErrorThrottled("RK3588_ENABLE_RGA is disabled at build");
}
#endif
int stride;
if (fmt == PixelFormat::RGB || fmt == PixelFormat::BGR) {
stride = frame->planes[0].stride > 0 ? frame->planes[0].stride : (frame->stride > 0 ? frame->stride : w * 3);
@ -461,6 +729,12 @@ private:
return;
}
const bool do_cpu_bbox = draw_bbox_ && !use_rga_bbox_;
const bool do_cpu_text = draw_text_;
if ((do_cpu_bbox || do_cpu_text) && frame->dma_fd >= 0) {
DmaSyncStartFd(frame->dma_fd);
}
for (const auto& it : frame->face_recog->items) {
int x1 = static_cast<int>(it.bbox.x);
int y1 = static_cast<int>(it.bbox.y);
@ -469,11 +743,11 @@ private:
const Color color = it.unknown ? Color{255, 0, 0} : Color{0, 255, 0};
if (draw_bbox_) {
if (do_cpu_bbox) {
DrawRect(data, w, h, stride, fmt, x1, y1, x2, y2, line_width_, color);
}
if (draw_text_) {
if (do_cpu_text) {
char label[96];
const char* name = it.best_name.empty() ? (it.unknown ? "unknown" : "") : it.best_name.c_str();
snprintf(label, sizeof(label), "%s %.2f", name, it.best_sim);
@ -482,6 +756,10 @@ private:
DrawText(data, w, h, stride, fmt, x1, text_y, label, font_scale_, color);
}
}
if ((do_cpu_bbox || do_cpu_text) && frame->dma_fd >= 0) {
DmaSyncEndFd(frame->dma_fd);
}
}
void DrawFaceDet(FramePtr frame) {
@ -492,6 +770,61 @@ private:
uint8_t* data = frame->planes[0].data ? frame->planes[0].data : frame->data;
PixelFormat fmt = frame->format;
const bool want_rga_bbox = use_rga_bbox_ && draw_bbox_;
bool rga_bbox_ok = false;
#if defined(RK3588_ENABLE_RGA)
if (want_rga_bbox) {
if (frame->dma_fd < 0) {
LogRgaBboxErrorThrottled("frame has no dma_fd");
} else if (!(fmt == PixelFormat::RGB || fmt == PixelFormat::BGR ||
fmt == PixelFormat::NV12 || fmt == PixelFormat::YUV420)) {
LogRgaBboxErrorThrottled("unsupported format=" + std::to_string(static_cast<int>(fmt)));
} else {
rga_buffer_t dst{};
int rga_stride_bytes = 0;
if (!WrapFrameAsRgaDst(*frame, dst, rga_stride_bytes)) {
LogRgaBboxErrorThrottled("WrapFrameAsRgaDst failed");
} else {
(void)rga_stride_bytes;
im_job_handle_t job = imbeginJob();
if (!job) {
LogRgaBboxErrorThrottled("imbeginJob failed");
} else {
bool ok = true;
const Color color{0, 255, 255};
for (const auto& it : frame->face_det->faces) {
int x = Clamp(static_cast<int>(it.bbox.x), 0, w - 1);
int y = Clamp(static_cast<int>(it.bbox.y), 0, h - 1);
int rw = static_cast<int>(it.bbox.w);
int rh = static_cast<int>(it.bbox.h);
rw = Clamp(rw, 1, w - x);
rh = Clamp(rh, 1, h - y);
im_rect rect{x, y, rw, rh};
if (imrectangleTask(job, dst, rect, PackColorArgb(color), line_width_) <= 0) {
ok = false;
break;
}
}
if (ok) {
ok = (imendJob(job, IM_SYNC, 0, nullptr) > 0);
} else {
(void)imendJob(job, IM_SYNC, 0, nullptr);
}
rga_bbox_ok = ok;
if (!rga_bbox_ok) {
LogRgaBboxErrorThrottled("imrectangle/imendJob failed");
}
}
}
}
}
#else
if (want_rga_bbox) {
LogRgaBboxErrorThrottled("RK3588_ENABLE_RGA is disabled at build");
}
#endif
int stride;
if (fmt == PixelFormat::RGB || fmt == PixelFormat::BGR) {
stride = frame->planes[0].stride > 0 ? frame->planes[0].stride : (frame->stride > 0 ? frame->stride : w * 3);
@ -501,6 +834,12 @@ private:
return;
}
const bool do_cpu_bbox = draw_bbox_ && !use_rga_bbox_;
const bool do_cpu_text = draw_text_;
if ((do_cpu_bbox || do_cpu_text) && frame->dma_fd >= 0) {
DmaSyncStartFd(frame->dma_fd);
}
const Color color{0, 255, 255};
for (const auto& it : frame->face_det->faces) {
int x1 = static_cast<int>(it.bbox.x);
@ -508,10 +847,10 @@ private:
int x2 = static_cast<int>(it.bbox.x + it.bbox.w);
int y2 = static_cast<int>(it.bbox.y + it.bbox.h);
if (draw_bbox_) {
if (do_cpu_bbox) {
DrawRect(data, w, h, stride, fmt, x1, y1, x2, y2, line_width_, color);
}
if (draw_text_) {
if (do_cpu_text) {
char label[64];
snprintf(label, sizeof(label), "face %.0f%%", it.score * 100);
int text_y = y1 - 8 * font_scale_;
@ -519,16 +858,35 @@ private:
DrawText(data, w, h, stride, fmt, x1, text_y, label, font_scale_, color);
}
}
if ((do_cpu_bbox || do_cpu_text) && frame->dma_fd >= 0) {
DmaSyncEndFd(frame->dma_fd);
}
}
std::string id_;
bool draw_bbox_ = true;
bool draw_text_ = true;
bool draw_face_det_ = true;
bool use_rga_bbox_ = true;
int line_width_ = 2;
int font_scale_ = 1;
std::vector<std::string> labels_;
void LogRgaBboxErrorThrottled(const std::string& reason) {
const auto now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::steady_clock::now().time_since_epoch())
.count();
if (last_rga_bbox_err_ms_ > 0 &&
now_ms - static_cast<int64_t>(last_rga_bbox_err_ms_) < 1000) {
return;
}
last_rga_bbox_err_ms_ = static_cast<uint64_t>(now_ms);
LogError("[osd] RGA bbox failed: " + reason + " id=" + id_);
}
uint64_t last_rga_bbox_err_ms_ = 0;
std::shared_ptr<SpscQueue<FramePtr>> input_queue_;
std::vector<std::shared_ptr<SpscQueue<FramePtr>>> output_queues_;
uint64_t processed_ = 0;

32
plugins/preprocess/preprocess_node.cpp
View File

@ -489,15 +489,9 @@ private:
rga_buffer_t dst_buf{};
DmaBufferPtr src_dma_buf; // keep alive if we allocate/copy
// Prepare tmp buffer outside the RGA critical section.
// Lazily allocate tmp buffer only when we must fall back to a 2-step pipeline.
// (resize + cvtcolor). Prefer a single improcess() call to reduce scheduling overhead.
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;
}
}
const bool can_cpu_read_src = (frame->data != nullptr && frame->data_size > 0);
auto CopySrcToDmaIfPossible = [&]() -> bool {
@ -572,6 +566,28 @@ private:
IM_STATUS status = IM_STATUS_SUCCESS;
if (need_resize && need_cvt) {
// Try to fuse resize + CSC in a single call.
if (Check(src_buf, dst_buf)) {
rga_buffer_t pat{};
const im_rect sr{0, 0, src_buf.width, src_buf.height};
const im_rect dr{0, 0, dst_buf.width, dst_buf.height};
const im_rect pr{0, 0, 0, 0};
status = improcess(src_buf, dst_buf, pat, sr, dr, pr,
0, nullptr, nullptr, IM_SYNC);
if (status == IM_STATUS_SUCCESS) {
return true;
}
}
// Fallback: 2-step (resize + cvtcolor).
if (!tmp_dma || !tmp_dma->valid()) {
tmp_dma = DmaAlloc(CalcImageSizeStrided(dst_wstride, dst_hstride, frame->format));
if (!tmp_dma || !tmp_dma->valid()) {
err = "DMA alloc for tmp failed";
return false;
}
}
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)) {

122
plugins/publish/publish_node.cpp
View File

@ -109,21 +109,68 @@ public:
// Global init (safe to call multiple times)
avformat_network_init();
// Async write queue: bound memory and never block the realtime pipeline.
queue_ = std::make_shared<SpscQueue<EncodedPacket>>(queue_size_, QueueDropStrategy::DropOldest);
running_ = true;
monitor_thread_ = std::thread(&AvMuxer::MonitorLoop, this);
worker_thread_ = std::thread(&AvMuxer::WorkerLoop, this);
LogInfo("[publish] muxer init async for " + url_);
return true;
}
bool WriteFrame(const EncodedPacket& pkt) {
std::lock_guard<std::mutex> lock(mutex_);
if (!ready_ || !fmt_ || !stream_) return false;
if (!running_.load(std::memory_order_acquire) || !queue_) return false;
if (pkt.data.empty()) return false;
EncodedPacket copy = pkt;
return queue_->Push(std::move(copy));
}
void Close() {
running_ = false;
if (queue_) queue_->Stop();
if (worker_thread_.joinable()) worker_thread_.join();
}
private:
void WorkerLoop() {
while (running_.load(std::memory_order_acquire)) {
if (!TryOpen()) {
if (!running_.load(std::memory_order_acquire)) break;
std::this_thread::sleep_for(std::chrono::seconds(1));
continue;
}
ready_ = true;
warned_ = false;
last_pts_ = -1;
LogInfo("[publish] muxer server ready: " + url_);
while (running_.load(std::memory_order_acquire) && ready_) {
EncodedPacket pkt;
if (!queue_ || !queue_->Pop(pkt, std::chrono::milliseconds(100))) {
continue;
}
if (!WriteFrameInternal(pkt)) {
ready_ = false;
}
}
CleanupContext();
ready_ = false;
if (running_.load(std::memory_order_acquire)) {
std::this_thread::sleep_for(std::chrono::seconds(1));
}
}
CleanupContext();
}
bool WriteFrameInternal(const EncodedPacket& pkt) {
if (!fmt_ || !stream_ || pkt.data.empty()) return false;
AVPacket* out = av_packet_alloc();
if (!out) return false;
if (av_new_packet(out, static_cast<int>(pkt.data.size())) < 0) {
av_packet_free(&out);
return false;
@ -141,9 +188,8 @@ public:
out->dts = pts;
out->duration = av_rescale_q(1000 / std::max(1, fps_), AVRational{1, 1000}, stream_->time_base);
int ret = av_interleaved_write_frame(fmt_, out);
const int ret = av_interleaved_write_frame(fmt_, out);
av_packet_free(&out);
if (ret < 0) {
if (!warned_) {
char errbuf[128];
@ -151,58 +197,19 @@ public:
std::cerr << "[publish] write failed for " << url_ << ": " << errbuf << ", resetting...\n";
warned_ = true;
}
// Signal monitor thread to reset
ready_ = false;
cv_.notify_all();
return false;
}
return ret >= 0;
return true;
}
void Close() {
running_ = false;
// Break the av_io_open wait if possible (via CheckInterrupt)
cv_.notify_all();
if (monitor_thread_.joinable()) monitor_thread_.join();
}
private:
void MonitorLoop() {
while (running_) {
if (TryOpen()) {
{
std::lock_guard<std::mutex> lock(mutex_);
ready_ = true;
warned_ = false;
LogInfo("[publish] muxer server ready: " + url_);
}
// Wait until error occurs or stop requested
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock, [this] { return !running_ || !ready_; });
}
// Cleanup context
{
std::lock_guard<std::mutex> lock(mutex_);
if (fmt_) {
// Try to write trailer if logical end (not socket error)
// But usually we are here because of error, so av_write_trailer might hang/fail.
// We skip trailer on error reset to avoid blocking.
if (fmt_->pb) {
avio_closep(&fmt_->pb);
}
avformat_free_context(fmt_);
fmt_ = nullptr;
stream_ = nullptr;
}
ready_ = false;
}
if (running_) {
// Delay before retry
std::this_thread::sleep_for(std::chrono::seconds(1));
}
void CleanupContext() {
if (!fmt_) return;
if (fmt_->pb) {
avio_closep(&fmt_->pb);
}
avformat_free_context(fmt_);
fmt_ = nullptr;
stream_ = nullptr;
}
bool TryOpen() {
@ -265,7 +272,6 @@ private:
av_dict_free(&opts);
{
std::lock_guard<std::mutex> lock(mutex_);
fmt_ = fmt;
stream_ = stream;
}
@ -301,9 +307,9 @@ private:
std::string url_;
std::atomic<bool> running_{false};
std::thread monitor_thread_;
std::mutex mutex_;
std::condition_variable cv_;
std::thread worker_thread_;
std::shared_ptr<SpscQueue<EncodedPacket>> queue_;
size_t queue_size_ = 64;
bool ready_ = false;
bool warned_ = false;

144
src/ai_scheduler.cpp
View File

@ -315,25 +315,68 @@ InferResult AiScheduler::Infer(ModelHandle handle, const InferInput& input) {
return result;
}
// Setup input
rknn_input inputs[1];
memset(inputs, 0, sizeof(inputs));
inputs[0].index = 0;
inputs[0].type = input.type;
inputs[0].size = input.size;
inputs[0].fmt = input.is_nhwc ? RKNN_TENSOR_NHWC : RKNN_TENSOR_NCHW;
inputs[0].buf = const_cast<void*>(input.data);
inputs[0].pass_through = 0;
struct InputMemGuard {
rknn_context ctx = 0;
rknn_tensor_mem* mem = nullptr;
~InputMemGuard() {
if (ctx && mem) {
rknn_destroy_mem(ctx, mem);
mem = nullptr;
}
}
};
InputMemGuard input_mem{ctx->ctx, nullptr};
bool used_io_mem = false;
int ret = rknn_inputs_set(ctx->ctx, ctx->n_input, inputs);
if (ret < 0) {
result.error = "rknn_inputs_set failed: " + std::to_string(ret);
total_errors_.fetch_add(1);
return result;
// Best-effort RKNN zero-copy input via DMA-BUF.
if (input.dma_fd >= 0) {
void* virt_base = const_cast<void*>(input.data);
const int32_t offset = input.dma_offset;
if (offset != 0) {
virt_base = static_cast<uint8_t*>(virt_base) - offset;
}
input_mem.mem = rknn_create_mem_from_fd(ctx->ctx, input.dma_fd,
virt_base,
static_cast<uint32_t>(input.size),
offset);
if (input_mem.mem) {
rknn_tensor_attr attr = ctx->input_attrs.empty() ? rknn_tensor_attr{} : ctx->input_attrs[0];
attr.index = 0;
attr.type = input.type;
attr.fmt = input.is_nhwc ? RKNN_TENSOR_NHWC : RKNN_TENSOR_NCHW;
const int mem_ret = rknn_set_io_mem(ctx->ctx, input_mem.mem, &attr);
if (mem_ret == 0) {
used_io_mem = true;
} else {
// Fallback to normal inputs_set path.
rknn_destroy_mem(ctx->ctx, input_mem.mem);
input_mem.mem = nullptr;
}
}
}
if (!used_io_mem) {
// Setup input (legacy copy path).
rknn_input inputs[1];
memset(inputs, 0, sizeof(inputs));
inputs[0].index = 0;
inputs[0].type = input.type;
inputs[0].size = input.size;
inputs[0].fmt = input.is_nhwc ? RKNN_TENSOR_NHWC : RKNN_TENSOR_NCHW;
inputs[0].buf = const_cast<void*>(input.data);
inputs[0].pass_through = 0;
int ret = rknn_inputs_set(ctx->ctx, ctx->n_input, inputs);
if (ret < 0) {
result.error = "rknn_inputs_set failed: " + std::to_string(ret);
total_errors_.fetch_add(1);
return result;
}
}
// Run inference
ret = rknn_run(ctx->ctx, nullptr);
int ret = rknn_run(ctx->ctx, nullptr);
if (ret < 0) {
result.error = "rknn_run failed: " + std::to_string(ret);
total_errors_.fetch_add(1);
@ -428,24 +471,65 @@ AiScheduler::BorrowedInferResult AiScheduler::InferBorrowed(ModelHandle handle,
result.infer_lock = std::unique_lock<std::mutex>(ctx->infer_mutex);
result.keepalive = ctx;
// Setup input
rknn_input inputs[1];
memset(inputs, 0, sizeof(inputs));
inputs[0].index = 0;
inputs[0].type = input.type;
inputs[0].size = input.size;
inputs[0].fmt = input.is_nhwc ? RKNN_TENSOR_NHWC : RKNN_TENSOR_NCHW;
inputs[0].buf = const_cast<void*>(input.data);
inputs[0].pass_through = 0;
struct InputMemGuard {
rknn_context ctx = 0;
rknn_tensor_mem* mem = nullptr;
~InputMemGuard() {
if (ctx && mem) {
rknn_destroy_mem(ctx, mem);
mem = nullptr;
}
}
};
InputMemGuard input_mem{ctx->ctx, nullptr};
bool used_io_mem = false;
int ret = rknn_inputs_set(ctx->ctx, ctx->n_input, inputs);
if (ret < 0) {
result.error = "rknn_inputs_set failed: " + std::to_string(ret);
total_errors_.fetch_add(1);
return result;
// Best-effort RKNN zero-copy input via DMA-BUF.
if (input.dma_fd >= 0) {
void* virt_base = const_cast<void*>(input.data);
const int32_t offset = input.dma_offset;
if (offset != 0) {
virt_base = static_cast<uint8_t*>(virt_base) - offset;
}
input_mem.mem = rknn_create_mem_from_fd(ctx->ctx, input.dma_fd,
virt_base,
static_cast<uint32_t>(input.size),
offset);
if (input_mem.mem) {
rknn_tensor_attr attr = ctx->input_attrs.empty() ? rknn_tensor_attr{} : ctx->input_attrs[0];
attr.index = 0;
attr.type = input.type;
attr.fmt = input.is_nhwc ? RKNN_TENSOR_NHWC : RKNN_TENSOR_NCHW;
const int mem_ret = rknn_set_io_mem(ctx->ctx, input_mem.mem, &attr);
if (mem_ret == 0) {
used_io_mem = true;
} else {
rknn_destroy_mem(ctx->ctx, input_mem.mem);
input_mem.mem = nullptr;
}
}
}
ret = rknn_run(ctx->ctx, nullptr);
if (!used_io_mem) {
// Setup input (legacy copy path).
rknn_input inputs[1];
memset(inputs, 0, sizeof(inputs));
inputs[0].index = 0;
inputs[0].type = input.type;
inputs[0].size = input.size;
inputs[0].fmt = input.is_nhwc ? RKNN_TENSOR_NHWC : RKNN_TENSOR_NCHW;
inputs[0].buf = const_cast<void*>(input.data);
inputs[0].pass_through = 0;
int ret = rknn_inputs_set(ctx->ctx, ctx->n_input, inputs);
if (ret < 0) {
result.error = "rknn_inputs_set failed: " + std::to_string(ret);
total_errors_.fetch_add(1);
return result;
}
}
int ret = rknn_run(ctx->ctx, nullptr);
if (ret < 0) {
result.error = "rknn_run failed: " + std::to_string(ret);
total_errors_.fetch_add(1);