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

修复阶段4之前的问题

Browse Source
This commit is contained in:
sladro 2025-12-29 09:33:51 +08:00
parent 14635c47c9
commit 99dd85293e
11 changed files with 514 additions and 482 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
include/graph_manager.h
View File

@ -5,6 +5,8 @@
#include <mutex>
#include <condition_variable>
#include <vector>
#include <thread>
#include <atomic>
#include "node.h"
#include "plugin_loader.h"
@ -26,14 +28,17 @@ private:
struct NodeEntry {
std::string id;
std::string type;
std::string role;
bool enabled = true;
SimpleJson config;
NodeContext context;
std::unique_ptr<INode> node;
std::thread worker;
};
std::string name_;
std::vector<NodeEntry> nodes_;
std::atomic<bool> running_{false};
};
class GraphManager {

7
include/node.h
View File

@ -31,11 +31,14 @@ public:
virtual std::string Id() const = 0;
virtual std::string Type() const = 0;
virtual bool Init(const SimpleJson& config, const NodeContext& ctx) = 0;
// Initialize resources.
// Note: For Source nodes, this should start the capture thread.
// For Filter/Sink nodes, this should ONLY allocate resources; the framework will drive Process().
virtual bool Start() = 0;
virtual void Stop() = 0;
// Process a single frame (for filter/sink nodes driven by GraphMgr).
// Default implementation does nothing; source nodes typically ignore this.
// Process a single frame (driven by GraphMgr for Filter/Sink nodes).
// Returns status to indicate if frame was processed or dropped.
virtual NodeStatus Process(FramePtr /*frame*/) { return NodeStatus::OK; }
// Dynamic config update without restart. Returns true if update succeeded.

5
include/utils/spsc_queue.h
View File

@ -62,7 +62,10 @@ public:
size_t Capacity() const { return capacity_; }
size_t DroppedCount() const { return dropped_; }
size_t DroppedCount() const {
std::lock_guard<std::mutex> lock(mu_);
return dropped_;
}
private:
size_t capacity_ = 0;

45
plugins/ai_yolo/ai_yolo_node.cpp
View File

@ -336,19 +336,11 @@ public:
}
bool Start() override {
if (!input_queue_) return false;
running_.store(true);
worker_ = std::thread(&AiYoloNode::WorkerLoop, this);
std::cout << "[ai_yolo] started, conf=" << conf_thresh_ << " nms=" << nms_thresh_ << "\n";
return true;
}
void Stop() override {
running_.store(false);
if (input_queue_) input_queue_->Stop();
for (auto& q : output_queues_) q->Stop();
if (worker_.joinable()) worker_.join();
#if defined(RK3588_ENABLE_RKNN)
if (model_handle_ != kInvalidModelHandle) {
AiScheduler::Instance().UnloadModel(model_handle_);
@ -358,6 +350,21 @@ public:
std::cout << "[ai_yolo] stopped\n";
}
NodeStatus Process(FramePtr frame) override {
if (!frame) return NodeStatus::DROP;
#if defined(RK3588_ENABLE_RKNN)
RunInference(frame);
#endif
PushToDownstream(frame);
++processed_;
if (processed_ % 100 == 0) {
std::cout << "[ai_yolo] processed " << processed_ << " frames\n";
}
return NodeStatus::OK;
}
private:
void PushToDownstream(FramePtr frame) {
for (auto& q : output_queues_) {
@ -365,26 +372,6 @@ private:
}
}
void WorkerLoop() {
using namespace std::chrono;
FramePtr frame;
while (running_.load()) {
if (!input_queue_->Pop(frame, milliseconds(200))) continue;
if (!frame) continue;
#if defined(RK3588_ENABLE_RKNN)
RunInference(frame);
#endif
PushToDownstream(frame);
++processed_;
if (processed_ % 100 == 0) {
std::cout << "[ai_yolo] processed " << processed_ << " frames\n";
}
}
}
#if defined(RK3588_ENABLE_RKNN)
void RunInference(FramePtr frame) {
if (!frame->data || frame->data_size == 0) return;
@ -540,10 +527,8 @@ private:
bool auto_detect_version_ = false;
std::set<int> class_filter_;
std::atomic<bool> running_{false};
std::shared_ptr<SpscQueue<FramePtr>> input_queue_;
std::vector<std::shared_ptr<SpscQueue<FramePtr>>> output_queues_;
std::thread worker_;
uint64_t processed_ = 0;
#if defined(RK3588_ENABLE_RKNN)

48
plugins/alarm/alarm_node.cpp
View File

@ -113,18 +113,11 @@ public:
}
bool Start() override {
if (!input_queue_) return false;
running_.store(true);
worker_ = std::thread(&AlarmNode::WorkerLoop, this);
std::cout << "[alarm] started\n";
return true;
}
void Stop() override {
running_.store(false);
if (input_queue_) input_queue_->Stop();
if (worker_.joinable()) worker_.join();
// Drain all actions
for (auto& action : actions_) {
action->Drain();
@ -140,33 +133,28 @@ public:
}
}
private:
void WorkerLoop() {
using namespace std::chrono;
FramePtr frame;
NodeStatus Process(FramePtr frame) override {
if (!frame) return NodeStatus::DROP;
while (running_.load()) {
if (!input_queue_->Pop(frame, milliseconds(200))) continue;
if (!frame) continue;
// Always push to ring buffer for pre-event recording
frame_buffer_->Push(frame);
// Always push to ring buffer for pre-event recording
frame_buffer_->Push(frame);
// Push to clip action for post-event collection if active
if (clip_action_) {
clip_action_->PushPostEventFrame(frame);
}
// Evaluate rules
auto result = rule_engine_.Evaluate(frame);
if (result.matched) {
TriggerAlarm(result, frame);
}
++processed_frames_;
// Push to clip action for post-event collection if active
if (clip_action_) {
clip_action_->PushPostEventFrame(frame);
}
// Evaluate rules
auto result = rule_engine_.Evaluate(frame);
if (result.matched) {
TriggerAlarm(result, frame);
}
++processed_frames_;
return NodeStatus::OK;
}
private:
void TriggerAlarm(const RuleMatchResult& result, FramePtr frame) {
++alarm_count_;
@ -193,9 +181,7 @@ private:
std::vector<std::unique_ptr<IAlarmAction>> actions_;
ClipAction* clip_action_ = nullptr;
std::atomic<bool> running_{false};
std::shared_ptr<SpscQueue<FramePtr>> input_queue_;
std::thread worker_;
uint64_t processed_frames_ = 0;
uint64_t alarm_count_ = 0;
};

4
plugins/input_rtsp/input_rtsp_node.cpp
View File

@ -93,6 +93,10 @@ public:
if (worker_.joinable()) worker_.join();
}
void Drain() override {
running_.store(false);
}
private:
void LoopStub() {
using namespace std::chrono;

46
plugins/osd/osd_node.cpp
View File

@ -308,21 +308,30 @@ public:
}
bool Start() override {
if (!input_queue_) return false;
running_.store(true);
worker_ = std::thread(&OsdNode::WorkerLoop, this);
std::cout << "[osd] started, draw_bbox=" << draw_bbox_ << " draw_text=" << draw_text_ << "\n";
return true;
}
void Stop() override {
running_.store(false);
if (input_queue_) input_queue_->Stop();
for (auto& q : output_queues_) q->Stop();
if (worker_.joinable()) worker_.join();
std::cout << "[osd] stopped\n";
}
NodeStatus Process(FramePtr frame) override {
if (!frame) return NodeStatus::DROP;
if (frame->det && frame->data) {
DrawDetections(frame);
}
PushToDownstream(frame);
++processed_;
if (processed_ % 100 == 0) {
std::cout << "[osd] processed " << processed_ << " frames\n";
}
return NodeStatus::OK;
}
private:
void PushToDownstream(FramePtr frame) {
for (auto& q : output_queues_) {
@ -330,27 +339,6 @@ private:
}
}
void WorkerLoop() {
using namespace std::chrono;
FramePtr frame;
while (running_.load()) {
if (!input_queue_->Pop(frame, milliseconds(200))) continue;
if (!frame) continue;
if (frame->det && frame->data) {
DrawDetections(frame);
}
PushToDownstream(frame);
++processed_;
if (processed_ % 100 == 0) {
std::cout << "[osd] processed " << processed_ << " frames\n";
}
}
}
const char* GetLabel(int cls_id) const {
if (!labels_.empty()) {
if (cls_id >= 0 && cls_id < static_cast<int>(labels_.size())) {
@ -408,10 +396,8 @@ private:
int font_scale_ = 1;
std::vector<std::string> labels_;
std::atomic<bool> running_{false};
std::shared_ptr<SpscQueue<FramePtr>> input_queue_;
std::vector<std::shared_ptr<SpscQueue<FramePtr>>> output_queues_;
std::thread worker_;
uint64_t processed_ = 0;
};

505
plugins/preprocess/preprocess_node.cpp
View File

@ -104,30 +104,35 @@ public:
}
bool Start() override {
if (!input_queue_) return false;
running_.store(true);
#if defined(RK3588_ENABLE_RGA)
if (use_rga_) {
worker_ = std::thread(&PreprocessNode::LoopRga, this);
} else {
worker_ = std::thread(&PreprocessNode::LoopSwscale, this);
}
#elif defined(RK3588_ENABLE_FFMPEG)
worker_ = std::thread(&PreprocessNode::LoopSwscale, this);
#else
worker_ = std::thread(&PreprocessNode::LoopPassthrough, this);
#endif
std::cout << "[preprocess] start dst=" << dst_w_ << "x" << dst_h_
<< (use_rga_ ? " (rga)" : " (swscale)") << "\n";
return true;
}
void Stop() override {
running_.store(false);
if (input_queue_) input_queue_->Stop();
for (auto& q : output_queues_) q->Stop();
if (worker_.joinable()) worker_.join();
#if defined(RK3588_ENABLE_FFMPEG)
if (sws_ctx_) {
sws_freeContext(sws_ctx_);
sws_ctx_ = nullptr;
}
#endif
}
NodeStatus Process(FramePtr frame) override {
if (!frame) return NodeStatus::DROP;
#if defined(RK3588_ENABLE_RGA)
if (use_rga_) {
ProcessRga(frame);
} else {
ProcessSwscale(frame);
}
#elif defined(RK3588_ENABLE_FFMPEG)
ProcessSwscale(frame);
#else
ProcessPassthrough(frame);
#endif
return NodeStatus::OK;
}
private:
@ -137,265 +142,239 @@ private:
}
}
void LoopPassthrough() {
using namespace std::chrono;
FramePtr frame;
while (running_.load()) {
if (!input_queue_->Pop(frame, milliseconds(200))) continue;
if (!frame) continue;
PushToDownstream(frame);
++processed_;
if (processed_ % 100 == 0) {
std::cout << "[preprocess] passthrough frame " << frame->frame_id << "\n";
}
void ProcessPassthrough(FramePtr frame) {
PushToDownstream(frame);
++processed_;
if (processed_ % 100 == 0) {
std::cout << "[preprocess] passthrough frame " << frame->frame_id << "\n";
}
}
#if defined(RK3588_ENABLE_RGA)
void LoopRga() {
using namespace std::chrono;
FramePtr frame;
while (running_.load()) {
if (!input_queue_->Pop(frame, milliseconds(200))) continue;
if (!frame) continue;
void ProcessRga(FramePtr frame) {
PixelFormat out_fmt = (dst_fmt_ != PixelFormat::UNKNOWN) ? dst_fmt_ : frame->format;
int out_w = dst_w_;
int out_h = dst_h_;
PixelFormat out_fmt = (dst_fmt_ != PixelFormat::UNKNOWN) ? dst_fmt_ : frame->format;
int out_w = dst_w_;
int out_h = dst_h_;
if (keep_ratio_ && frame->width > 0 && frame->height > 0) {
float scale = std::min(static_cast<float>(dst_w_) / frame->width,
static_cast<float>(dst_h_) / frame->height);
out_w = static_cast<int>(frame->width * scale);
out_h = static_cast<int>(frame->height * scale);
out_w = (out_w + 1) & ~1;
out_h = (out_h + 1) & ~1;
}
if (keep_ratio_ && frame->width > 0 && frame->height > 0) {
float scale = std::min(static_cast<float>(dst_w_) / frame->width,
static_cast<float>(dst_h_) / frame->height);
out_w = static_cast<int>(frame->width * scale);
out_h = static_cast<int>(frame->height * scale);
out_w = (out_w + 1) & ~1;
out_h = (out_h + 1) & ~1;
}
int src_fmt_rga = ToRgaFormat(frame->format);
int dst_fmt_rga = ToRgaFormat(out_fmt);
bool need_cvt = (src_fmt_rga != dst_fmt_rga);
bool need_resize = (frame->width != out_w || frame->height != out_h);
int src_fmt_rga = ToRgaFormat(frame->format);
int dst_fmt_rga = ToRgaFormat(out_fmt);
bool need_cvt = (src_fmt_rga != dst_fmt_rga);
bool need_resize = (frame->width != out_w || frame->height != out_h);
// If no processing needed, passthrough directly
if (!need_cvt && !need_resize) {
PushToDownstream(frame);
++processed_;
if (processed_ % 100 == 0) {
std::cout << "[preprocess] passthrough frame " << frame->frame_id
<< " " << frame->width << "x" << frame->height << " (no change)\n";
}
continue;
}
size_t out_size = CalcImageSize(out_w, out_h, out_fmt);
if (out_size == 0 || src_fmt_rga == RK_FORMAT_UNKNOWN || dst_fmt_rga == RK_FORMAT_UNKNOWN) {
std::cerr << "[preprocess] unsupported format for RGA\n";
PushToDownstream(frame);
continue;
}
// Use DMA-BUF allocation to avoid >4GB address issue with RGA
auto dma_buf = DmaAlloc(out_size);
if (!dma_buf || !dma_buf->valid()) {
std::cerr << "[preprocess] DMA alloc failed, falling back to std::vector\n";
PushToDownstream(frame);
continue;
}
// Calculate proper strides (RGA requires aligned strides)
// For YUV formats, wstride is the width of Y plane
// For RGB/BGR formats, wstride is width (not width*3)
int src_wstride = Align16(frame->width);
int src_hstride = Align16(frame->height);
int dst_wstride = Align16(out_w);
int dst_hstride = Align16(out_h);
if (processed_ < 3) {
std::cout << "[preprocess] src: " << frame->width << "x" << frame->height
<< " fmt=" << static_cast<int>(frame->format) << " rga_fmt=" << src_fmt_rga
<< " wstride=" << src_wstride << " hstride=" << src_hstride
<< " data_size=" << frame->data_size << "\n";
std::cout << "[preprocess] dst: " << out_w << "x" << out_h
<< " fmt=" << static_cast<int>(out_fmt) << " rga_fmt=" << dst_fmt_rga
<< " wstride=" << dst_wstride << " hstride=" << dst_hstride << "\n";
}
rga_buffer_t src_buf{};
rga_buffer_t dst_buf{};
DmaBufferPtr src_dma_buf; // Keep alive if we allocate
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 if (frame->data) {
// Source doesn't have DMA fd, copy to DMA buffer first to avoid >4GB address issue
size_t src_size = CalcImageSize(frame->width, frame->height, 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";
PushToDownstream(frame);
continue;
}
memcpy(src_dma_buf->data(), frame->data, std::min(src_size, frame->data_size));
src_buf = wrapbuffer_fd_t(src_dma_buf->fd, frame->width, frame->height,
src_wstride, src_hstride, src_fmt_rga);
} else {
PushToDownstream(frame);
continue;
}
// 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);
IM_STATUS status = IM_STATUS_SUCCESS;
if (need_resize && need_cvt) {
// Allocate DMA buffer for intermediate result
auto tmp_dma = DmaAlloc(CalcImageSize(out_w, out_h, frame->format));
if (!tmp_dma || !tmp_dma->valid()) {
std::cerr << "[preprocess] DMA alloc for tmp failed\n";
PushToDownstream(frame);
continue;
}
rga_buffer_t tmp = wrapbuffer_fd_t(tmp_dma->fd, out_w, out_h,
dst_wstride, dst_hstride, src_fmt_rga);
status = imresize(src_buf, tmp);
if (status == IM_STATUS_SUCCESS) {
status = imcvtcolor(tmp, dst_buf, src_fmt_rga, dst_fmt_rga, IM_COLOR_SPACE_DEFAULT);
}
} else if (need_resize) {
status = imresize(src_buf, dst_buf);
} else if (need_cvt) {
status = imcvtcolor(src_buf, dst_buf, src_fmt_rga, dst_fmt_rga, IM_COLOR_SPACE_DEFAULT);
}
if (status != IM_STATUS_SUCCESS) {
std::cerr << "[preprocess] RGA failed: " << imStrError(status) << "\n";
PushToDownstream(frame);
continue;
}
auto out_frame = std::make_shared<Frame>();
out_frame->width = out_w;
out_frame->height = out_h;
out_frame->format = out_fmt;
out_frame->stride = dst_wstride;
out_frame->dma_fd = dma_buf->fd;
out_frame->data = dma_buf->data();
out_frame->data_size = dma_buf->size;
out_frame->data_owner = dma_buf; // DmaBuffer shared_ptr keeps fd alive
out_frame->pts = frame->pts;
out_frame->frame_id = frame->frame_id;
out_frame->det = frame->det;
out_frame->user_meta = frame->user_meta;
SetupPlanes(*out_frame, out_fmt);
PushToDownstream(out_frame);
// If no processing needed, passthrough directly
if (!need_cvt && !need_resize) {
PushToDownstream(frame);
++processed_;
if (processed_ % 100 == 0) {
std::cout << "[preprocess] rga frame " << out_frame->frame_id
<< " " << frame->width << "x" << frame->height
<< " -> " << out_w << "x" << out_h << "\n";
std::cout << "[preprocess] passthrough frame " << frame->frame_id
<< " " << frame->width << "x" << frame->height << " (no change)\n";
}
return;
}
size_t out_size = CalcImageSize(out_w, out_h, out_fmt);
if (out_size == 0 || src_fmt_rga == RK_FORMAT_UNKNOWN || dst_fmt_rga == RK_FORMAT_UNKNOWN) {
std::cerr << "[preprocess] unsupported format for RGA\n";
PushToDownstream(frame);
return;
}
// Use DMA-BUF allocation to avoid >4GB address issue with RGA
auto dma_buf = DmaAlloc(out_size);
if (!dma_buf || !dma_buf->valid()) {
std::cerr << "[preprocess] DMA alloc failed, falling back to std::vector\n";
PushToDownstream(frame);
return;
}
// Calculate proper strides (RGA requires aligned strides)
// For YUV formats, wstride is the width of Y plane
// For RGB/BGR formats, wstride is width (not width*3)
int src_wstride = Align16(frame->width);
int src_hstride = Align16(frame->height);
int dst_wstride = Align16(out_w);
int dst_hstride = Align16(out_h);
if (processed_ < 3) {
std::cout << "[preprocess] src: " << frame->width << "x" << frame->height
<< " fmt=" << static_cast<int>(frame->format) << " rga_fmt=" << src_fmt_rga
<< " wstride=" << src_wstride << " hstride=" << src_hstride
<< " data_size=" << frame->data_size << "\n";
std::cout << "[preprocess] dst: " << out_w << "x" << out_h
<< " fmt=" << static_cast<int>(out_fmt) << " rga_fmt=" << dst_fmt_rga
<< " wstride=" << dst_wstride << " hstride=" << dst_hstride << "\n";
}
rga_buffer_t src_buf{};
rga_buffer_t dst_buf{};
DmaBufferPtr src_dma_buf; // Keep alive if we allocate
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 if (frame->data) {
// Source doesn't have DMA fd, copy to DMA buffer first to avoid >4GB address issue
size_t src_size = CalcImageSize(frame->width, frame->height, 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";
PushToDownstream(frame);
return;
}
memcpy(src_dma_buf->data(), frame->data, std::min(src_size, frame->data_size));
src_buf = wrapbuffer_fd_t(src_dma_buf->fd, frame->width, frame->height,
src_wstride, src_hstride, src_fmt_rga);
} else {
PushToDownstream(frame);
return;
}
// 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);
IM_STATUS status = IM_STATUS_SUCCESS;
if (need_resize && need_cvt) {
// Allocate DMA buffer for intermediate result
auto tmp_dma = DmaAlloc(CalcImageSize(out_w, out_h, frame->format));
if (!tmp_dma || !tmp_dma->valid()) {
std::cerr << "[preprocess] DMA alloc for tmp failed\n";
PushToDownstream(frame);
return;
}
rga_buffer_t tmp = wrapbuffer_fd_t(tmp_dma->fd, out_w, out_h,
dst_wstride, dst_hstride, src_fmt_rga);
status = imresize(src_buf, tmp);
if (status == IM_STATUS_SUCCESS) {
status = imcvtcolor(tmp, dst_buf, src_fmt_rga, dst_fmt_rga, IM_COLOR_SPACE_DEFAULT);
}
} else if (need_resize) {
status = imresize(src_buf, dst_buf);
} else if (need_cvt) {
status = imcvtcolor(src_buf, dst_buf, src_fmt_rga, dst_fmt_rga, IM_COLOR_SPACE_DEFAULT);
}
if (status != IM_STATUS_SUCCESS) {
std::cerr << "[preprocess] RGA failed: " << imStrError(status) << "\n";
PushToDownstream(frame);
return;
}
auto out_frame = std::make_shared<Frame>();
out_frame->width = out_w;
out_frame->height = out_h;
out_frame->format = out_fmt;
out_frame->stride = dst_wstride;
out_frame->dma_fd = dma_buf->fd;
out_frame->data = dma_buf->data();
out_frame->data_size = dma_buf->size;
out_frame->data_owner = dma_buf; // DmaBuffer shared_ptr keeps fd alive
out_frame->pts = frame->pts;
out_frame->frame_id = frame->frame_id;
out_frame->det = frame->det;
out_frame->user_meta = frame->user_meta;
SetupPlanes(*out_frame, out_fmt);
PushToDownstream(out_frame);
++processed_;
if (processed_ % 100 == 0) {
std::cout << "[preprocess] rga frame " << out_frame->frame_id
<< " " << frame->width << "x" << frame->height
<< " -> " << out_w << "x" << out_h << "\n";
}
}
#endif
#if defined(RK3588_ENABLE_FFMPEG)
void LoopSwscale() {
using namespace std::chrono;
FramePtr frame;
SwsContext* sws_ctx = nullptr;
int last_src_w = 0, last_src_h = 0;
AVPixelFormat last_src_fmt = AV_PIX_FMT_NONE;
AVPixelFormat last_dst_fmt = AV_PIX_FMT_NONE;
void ProcessSwscale(FramePtr frame) {
PixelFormat out_fmt = (dst_fmt_ != PixelFormat::UNKNOWN) ? dst_fmt_ : frame->format;
int out_w = dst_w_;
int out_h = dst_h_;
while (running_.load()) {
if (!input_queue_->Pop(frame, milliseconds(200))) continue;
if (!frame) continue;
PixelFormat out_fmt = (dst_fmt_ != PixelFormat::UNKNOWN) ? dst_fmt_ : frame->format;
int out_w = dst_w_;
int out_h = dst_h_;
if (keep_ratio_ && frame->width > 0 && frame->height > 0) {
float scale = std::min(static_cast<float>(dst_w_) / frame->width,
static_cast<float>(dst_h_) / frame->height);
out_w = static_cast<int>(frame->width * scale);
out_h = static_cast<int>(frame->height * scale);
out_w = (out_w + 1) & ~1;
out_h = (out_h + 1) & ~1;
}
AVPixelFormat src_av_fmt = ToAvFormat(frame->format);
AVPixelFormat dst_av_fmt = ToAvFormat(out_fmt);
if (src_av_fmt == AV_PIX_FMT_NONE || dst_av_fmt == AV_PIX_FMT_NONE) {
PushToDownstream(frame);
continue;
}
if (!sws_ctx || frame->width != last_src_w || frame->height != last_src_h ||
src_av_fmt != last_src_fmt || dst_av_fmt != last_dst_fmt) {
if (sws_ctx) sws_freeContext(sws_ctx);
sws_ctx = sws_getContext(frame->width, frame->height, src_av_fmt,
out_w, out_h, dst_av_fmt,
SWS_BILINEAR, nullptr, nullptr, nullptr);
last_src_w = frame->width;
last_src_h = frame->height;
last_src_fmt = src_av_fmt;
last_dst_fmt = dst_av_fmt;
}
if (!sws_ctx) {
PushToDownstream(frame);
continue;
}
size_t out_size = CalcImageSize(out_w, out_h, out_fmt);
auto buffer = std::make_shared<std::vector<uint8_t>>(out_size);
uint8_t* src_data[4] = {nullptr};
int src_linesize[4] = {0};
uint8_t* dst_data[4] = {nullptr};
int dst_linesize[4] = {0};
SetupAvPlanes(frame.get(), src_data, src_linesize);
av_image_fill_arrays(dst_data, dst_linesize, buffer->data(),
dst_av_fmt, out_w, out_h, 1);
sws_scale(sws_ctx, src_data, src_linesize, 0, frame->height,
dst_data, dst_linesize);
auto out_frame = std::make_shared<Frame>();
out_frame->width = out_w;
out_frame->height = out_h;
out_frame->format = out_fmt;
out_frame->stride = out_w;
out_frame->data = buffer->data();
out_frame->data_size = buffer->size();
out_frame->data_owner = buffer;
out_frame->pts = frame->pts;
out_frame->frame_id = frame->frame_id;
out_frame->det = frame->det;
out_frame->user_meta = frame->user_meta;
SetupPlanes(*out_frame, out_fmt);
PushToDownstream(out_frame);
++processed_;
if (processed_ % 100 == 0) {
std::cout << "[preprocess] swscale frame " << out_frame->frame_id
<< " " << frame->width << "x" << frame->height
<< " -> " << out_w << "x" << out_h << "\n";
}
if (keep_ratio_ && frame->width > 0 && frame->height > 0) {
float scale = std::min(static_cast<float>(dst_w_) / frame->width,
static_cast<float>(dst_h_) / frame->height);
out_w = static_cast<int>(frame->width * scale);
out_h = static_cast<int>(frame->height * scale);
out_w = (out_w + 1) & ~1;
out_h = (out_h + 1) & ~1;
}
if (sws_ctx) sws_freeContext(sws_ctx);
AVPixelFormat src_av_fmt = ToAvFormat(frame->format);
AVPixelFormat dst_av_fmt = ToAvFormat(out_fmt);
if (src_av_fmt == AV_PIX_FMT_NONE || dst_av_fmt == AV_PIX_FMT_NONE) {
PushToDownstream(frame);
return;
}
if (!sws_ctx_ || frame->width != last_src_w_ || frame->height != last_src_h_ ||
src_av_fmt != last_src_fmt_ || dst_av_fmt != last_dst_fmt_) {
if (sws_ctx_) sws_freeContext(sws_ctx_);
sws_ctx_ = sws_getContext(frame->width, frame->height, src_av_fmt,
out_w, out_h, dst_av_fmt,
SWS_BILINEAR, nullptr, nullptr, nullptr);
last_src_w_ = frame->width;
last_src_h_ = frame->height;
last_src_fmt_ = src_av_fmt;
last_dst_fmt_ = dst_av_fmt;
}
if (!sws_ctx_) {
PushToDownstream(frame);
return;
}
size_t out_size = CalcImageSize(out_w, out_h, out_fmt);
auto buffer = std::make_shared<std::vector<uint8_t>>(out_size);
uint8_t* src_data[4] = {nullptr};
int src_linesize[4] = {0};
uint8_t* dst_data[4] = {nullptr};
int dst_linesize[4] = {0};
SetupAvPlanes(frame.get(), src_data, src_linesize);
av_image_fill_arrays(dst_data, dst_linesize, buffer->data(),
dst_av_fmt, out_w, out_h, 1);
sws_scale(sws_ctx_, src_data, src_linesize, 0, frame->height,
dst_data, dst_linesize);
auto out_frame = std::make_shared<Frame>();
out_frame->width = out_w;
out_frame->height = out_h;
out_frame->format = out_fmt;
out_frame->stride = out_w;
out_frame->data = buffer->data();
out_frame->data_size = buffer->size();
out_frame->data_owner = buffer;
out_frame->pts = frame->pts;
out_frame->frame_id = frame->frame_id;
out_frame->det = frame->det;
out_frame->user_meta = frame->user_meta;
SetupPlanes(*out_frame, out_fmt);
PushToDownstream(out_frame);
++processed_;
if (processed_ % 100 == 0) {
std::cout << "[preprocess] swscale frame " << out_frame->frame_id
<< " " << frame->width << "x" << frame->height
<< " -> " << out_w << "x" << out_h << "\n";
}
}
#endif
static AVPixelFormat ToAvFormat(PixelFormat fmt) {
switch (fmt) {
@ -456,11 +435,17 @@ private:
PixelFormat dst_fmt_ = PixelFormat::UNKNOWN;
bool use_rga_ = true;
std::atomic<bool> running_{false};
std::shared_ptr<SpscQueue<FramePtr>> input_queue_;
std::vector<std::shared_ptr<SpscQueue<FramePtr>>> output_queues_;
std::thread worker_;
uint64_t processed_ = 0;
#if defined(RK3588_ENABLE_FFMPEG)
SwsContext* sws_ctx_ = nullptr;
int last_src_w_ = 0;
int last_src_h_ = 0;
AVPixelFormat last_src_fmt_ = AV_PIX_FMT_NONE;
AVPixelFormat last_dst_fmt_ = AV_PIX_FMT_NONE;
#endif
};
REGISTER_NODE(PreprocessNode, "preprocess");

201
plugins/publish/publish_node.cpp
View File

@ -639,23 +639,19 @@ public:
std::cerr << "[publish] no input queue for node " << id_ << "\n";
return false;
}
for (const auto& o : outputs_) {
if (o.proto == "rtsp_server") {
zlm_outputs_.push_back(o);
} else {
ff_outputs_.push_back(o);
}
}
return true;
}
bool Start() override {
if (!input_queue_) return false;
running_.store(true);
#if defined(RK3588_ENABLE_MPP)
if (use_mpp_) {
worker_ = std::thread(&PublishNode::LoopMpp, this);
} else {
worker_ = std::thread(&PublishNode::LoopStub, this);
}
#else
worker_ = std::thread(&PublishNode::LoopStub, this);
#endif
std::cout << "[publish] start codec=" << codec_ << " fps=" << fps_ << " gop=" << gop_
<< " bitrate=" << bitrate_kbps_ << "kbps"
<< (use_mpp_ ? " (mpp venc)" : " (stub)") << "\n";
@ -663,9 +659,33 @@ public:
}
void Stop() override {
running_.store(false);
if (input_queue_) input_queue_->Stop();
if (worker_.joinable()) worker_.join();
#if defined(RK3588_ENABLE_FFMPEG)
if (mux_mgr_) mux_mgr_->Close();
#endif
#if defined(RK3588_ENABLE_ZLMEDIAKIT)
for (auto& p : zlm_pubs_) p->Close();
zlm_pubs_.clear();
#endif
#if defined(RK3588_ENABLE_MPP)
if (mpp_encoder_) mpp_encoder_->Shutdown();
#endif
}
NodeStatus Process(FramePtr frame) override {
if (!frame) return NodeStatus::DROP;
#if defined(RK3588_ENABLE_MPP)
if (use_mpp_) {
ProcessMpp(frame);
} else {
ProcessStub(frame);
}
#else
ProcessStub(frame);
#endif
return NodeStatus::OK;
}
private:
@ -1046,104 +1066,69 @@ private:
};
#endif
void LoopStub() {
using namespace std::chrono;
FramePtr frame;
while (running_.load()) {
if (!input_queue_->Pop(frame, milliseconds(200))) continue;
++encoded_frames_;
if (encoded_frames_ % 100 == 0 && frame) {
std::cout << "[publish] stub frame " << frame->frame_id
<< " queue=" << input_queue_->Size()
<< " drops=" << input_queue_->DroppedCount() << "\n";
}
void ProcessStub(FramePtr frame) {
++encoded_frames_;
if (encoded_frames_ % 100 == 0) {
std::cout << "[publish] stub frame " << frame->frame_id
<< " queue=" << input_queue_->Size()
<< " drops=" << input_queue_->DroppedCount() << "\n";
}
}
#if defined(RK3588_ENABLE_MPP)
void LoopMpp() {
using namespace std::chrono;
FramePtr frame;
MppVencEncoder encoder;
bool encoder_ready = false;
void ProcessMpp(FramePtr frame) {
if (!mpp_encoder_) {
mpp_encoder_ = std::make_unique<MppVencEncoder>();
}
if (!encoder_ready_) {
if (!mpp_encoder_->InitFromFrame(*frame, codec_, fps_, gop_, bitrate_kbps_)) {
std::cerr << "[publish] encoder init failed, fallback to stub\n";
use_mpp_ = false;
ProcessStub(frame);
return;
}
#if defined(RK3588_ENABLE_FFMPEG)
if (use_ffmpeg_mux_) {
AVCodecID cid = (codec_ == "h265" || codec_ == "hevc") ? AV_CODEC_ID_HEVC
: AV_CODEC_ID_H264;
if (!mux_mgr_) mux_mgr_ = std::make_unique<AvMuxerManager>();
mux_mgr_->Init(ff_outputs_, cid, frame->width, frame->height, fps_, mpp_encoder_->Header());
}
#endif
std::vector<OutputConfig> ff_outputs;
#if defined(RK3588_ENABLE_ZLMEDIAKIT)
std::vector<OutputConfig> zlm_outputs;
std::vector<std::unique_ptr<ZlmRtspPublisher>> zlm_pubs;
for (const auto& o : zlm_outputs_) {
auto pub = std::make_unique<ZlmRtspPublisher>();
if (pub->Init(o.port, o.path, id_, codec_, frame->width, frame->height, fps_, bitrate_kbps_)) {
zlm_pubs_.push_back(std::move(pub));
}
}
#endif
encoder_ready_ = true;
}
const bool is_h265 = (codec_ == "h265" || codec_ == "hevc");
#endif
for (const auto& o : outputs_) {
if (o.proto == "rtsp_server") {
#if defined(RK3588_ENABLE_ZLMEDIAKIT)
zlm_outputs.push_back(o);
#else
std::cerr << "[publish] output proto=rtsp_server requested but RK3588_ENABLE_ZLMEDIAKIT is off" << "\n";
#endif
} else {
ff_outputs.push_back(o);
mpp_encoder_->Encode(frame, [&](const EncodedPacket& pkt) {
++encoded_frames_;
if (encoded_frames_ % 100 == 0) {
std::cout << "[publish] encoded frame " << encoded_frames_
<< " queue=" << input_queue_->Size()
<< " drops=" << input_queue_->DroppedCount() << "\n";
}
}
#if defined(RK3588_ENABLE_FFMPEG)
AvMuxerManager mux_mgr;
#endif
while (running_.load()) {
if (!input_queue_->Pop(frame, milliseconds(200))) continue;
if (!frame) continue;
if (!encoder_ready) {
if (!encoder.InitFromFrame(*frame, codec_, fps_, gop_, bitrate_kbps_)) {
std::cerr << "[publish] encoder init failed, fallback to stub" << "\n";
LoopStub();
return;
}
#if defined(RK3588_ENABLE_FFMPEG)
if (use_ffmpeg_mux_) {
AVCodecID cid = (codec_ == "h265" || codec_ == "hevc") ? AV_CODEC_ID_HEVC
: AV_CODEC_ID_H264;
mux_mgr.Init(ff_outputs, cid, frame->width, frame->height, fps_, encoder.Header());
}
#endif
#if defined(RK3588_ENABLE_ZLMEDIAKIT)
for (const auto& o : zlm_outputs) {
auto pub = std::make_unique<ZlmRtspPublisher>();
if (pub->Init(o.port, o.path, id_, codec_, frame->width, frame->height, fps_, bitrate_kbps_)) {
zlm_pubs.push_back(std::move(pub));
}
}
#endif
encoder_ready = true;
}
encoder.Encode(frame, [&](const EncodedPacket& pkt) {
++encoded_frames_;
if (encoded_frames_ % 100 == 0) {
std::cout << "[publish] encoded frame " << encoded_frames_
<< " queue=" << input_queue_->Size()
<< " drops=" << input_queue_->DroppedCount() << "\n";
}
#if defined(RK3588_ENABLE_FFMPEG)
if (use_ffmpeg_mux_) mux_mgr.Write(pkt);
if (use_ffmpeg_mux_ && mux_mgr_) mux_mgr_->Write(pkt);
#else
(void)pkt;
(void)pkt;
#endif
#if defined(RK3588_ENABLE_ZLMEDIAKIT)
for (auto& p : zlm_pubs) {
p->Write(pkt, encoder.Header(), is_h265);
}
#endif
});
}
#if defined(RK3588_ENABLE_FFMPEG)
mux_mgr.Close();
for (auto& p : zlm_pubs_) {
p->Write(pkt, mpp_encoder_->Header(), is_h265);
}
#endif
});
}
#endif
@ -1155,10 +1140,24 @@ private:
bool use_mpp_ = false;
bool use_ffmpeg_mux_ = false;
std::vector<OutputConfig> outputs_;
std::atomic<bool> running_{false};
std::vector<OutputConfig> ff_outputs_;
std::vector<OutputConfig> zlm_outputs_;
std::shared_ptr<SpscQueue<FramePtr>> input_queue_;
std::thread worker_;
uint64_t encoded_frames_ = 0;
#if defined(RK3588_ENABLE_MPP)
std::unique_ptr<MppVencEncoder> mpp_encoder_;
bool encoder_ready_ = false;
#endif
#if defined(RK3588_ENABLE_FFMPEG)
std::unique_ptr<AvMuxerManager> mux_mgr_;
#endif
#if defined(RK3588_ENABLE_ZLMEDIAKIT)
std::vector<std::unique_ptr<ZlmRtspPublisher>> zlm_pubs_;
#endif
};
REGISTER_NODE(PublishNode, "publish");

23
plugins/storage/storage_node.cpp
View File

@ -283,16 +283,12 @@ public:
}
bool Start() override {
running_.store(true);
worker_ = std::thread(&StorageNode::WorkerLoop, this);
std::cout << "[storage] started\n";
return true;
}
void Stop() override {
running_.store(false);
if (input_queue_) input_queue_->Stop();
if (worker_.joinable()) worker_.join();
CloseCurrentFile();
std::cout << "[storage] stopped, recorded " << total_frames_ << " frames\n";
}
@ -301,20 +297,15 @@ public:
CloseCurrentFile();
}
private:
void WorkerLoop() {
using namespace std::chrono;
FramePtr frame;
NodeStatus Process(FramePtr frame) override {
if (!frame) return NodeStatus::DROP;
while (running_.load()) {
if (!input_queue_->Pop(frame, milliseconds(200))) continue;
if (!frame) continue;
ProcessFrame(frame);
++total_frames_;
}
ProcessFrame(frame);
++total_frames_;
return NodeStatus::OK;
}
private:
void ProcessFrame(FramePtr frame) {
// Check if we need to start a new segment
auto now = std::chrono::steady_clock::now();
@ -477,9 +468,7 @@ private:
int fps_ = 25;
int bitrate_kbps_ = 2000;
std::atomic<bool> running_{false};
std::shared_ptr<SpscQueue<FramePtr>> input_queue_;
std::thread worker_;
uint64_t total_frames_ = 0;
// Current file state

107
src/graph_manager.cpp
View File

@ -4,6 +4,8 @@
#include <iostream>
#include <map>
#include <set>
#include <thread>
#include <chrono>
namespace rk3588 {
@ -35,7 +37,9 @@ bool Graph::Build(const SimpleJson& graph_cfg, PluginLoader& loader, size_t defa
entry.config = node_val;
entry.id = node_val.ValueOr<std::string>("id", "");
entry.type = node_val.ValueOr<std::string>("type", "");
entry.role = node_val.ValueOr<std::string>("role", "");
entry.enabled = node_val.ValueOr<bool>("enable", true);
if (entry.id.empty() || entry.type.empty()) {
err = "Node missing id or type";
return false;
@ -52,7 +56,9 @@ bool Graph::Build(const SimpleJson& graph_cfg, PluginLoader& loader, size_t defa
std::map<std::string, NodeEntry*> id_to_node;
for (auto& n : nodes_) {
id_to_node[n.id] = &n;
if (n.enabled) {
id_to_node[n.id] = &n;
}
}
for (const auto& edge_val : edges_it->second.AsArray()) {
@ -67,33 +73,65 @@ bool Graph::Build(const SimpleJson& graph_cfg, PluginLoader& loader, size_t defa
err = "Edge has empty endpoint";
return false;
}
auto from_it = id_to_node.find(from);
auto to_it = id_to_node.find(to);
if (from_it == id_to_node.end() || to_it == id_to_node.end()) {
err = "Edge references unknown node";
return false;
// Check if nodes exist but are disabled
bool from_exists = false;
bool to_exists = false;
for(const auto& n : nodes_) {
if(n.id == from) from_exists = true;
if(n.id == to) to_exists = true;
}
if (!from_exists || !to_exists) {
err = "Edge references unknown node: " + from + " -> " + to;
return false;
}
// At least one is disabled, skip edge
continue;
}
size_t qsize = default_queue_size;
QueueDropStrategy strategy = default_strategy;
if (const auto* qcfg = edge_val.Find("queue")) {
if (qcfg->IsObject()) {
qsize = static_cast<size_t>(qcfg->ValueOr<int>("size", static_cast<int>(default_queue_size)));
std::string strat = qcfg->ValueOr<std::string>("strategy", "drop_oldest");
if (strat == "drop_oldest") strategy = QueueDropStrategy::DropOldest;
else strategy = QueueDropStrategy::Block;
std::string policy = qcfg->ValueOr<std::string>("policy", "");
std::string strat = qcfg->ValueOr<std::string>("strategy", "");
std::string final_policy = policy.empty() ? strat : policy;
if (final_policy == "drop_oldest" || final_policy == "drop_newest") {
strategy = QueueDropStrategy::DropOldest;
} else if (final_policy == "block") {
strategy = QueueDropStrategy::Block;
}
}
}
auto queue = std::make_shared<SpscQueue<FramePtr>>(qsize, strategy);
from_it->second->context.output_queues.push_back(queue);
// For now support single input per node; first edge wins.
if (!to_it->second->context.input_queue) {
to_it->second->context.input_queue = queue;
} else {
std::cerr << "[Graph] Warning: Node " << to << " already has input. Ignoring edge from " << from << "\n";
}
}
// Instantiate nodes via plugins
// Role validation & Instantiation
for (auto& entry : nodes_) {
if (!entry.enabled) continue;
if (entry.role == "source") {
if (entry.context.input_queue) {
err = "Source node " + entry.id + " cannot have input";
return false;
}
}
std::string load_err;
entry.node = loader.Create(entry.type, load_err);
if (!entry.node) {
@ -110,24 +148,73 @@ bool Graph::Build(const SimpleJson& graph_cfg, PluginLoader& loader, size_t defa
}
bool Graph::Start() {
bool expected = false;
if (!running_.compare_exchange_strong(expected, true)) {
return true; // Already running
}
for (auto& entry : nodes_) {
if (!entry.enabled || !entry.node) continue;
if (!entry.node->Start()) {
std::cerr << "[Graph] failed to start node: " << entry.id << "\n";
return false;
}
// For non-Source nodes, start framework thread
if (entry.role != "source") {
if (entry.context.input_queue) {
entry.worker = std::thread([this, &entry]() {
FramePtr frame;
while (running_) {
if (entry.context.input_queue->Pop(frame, std::chrono::milliseconds(100))) {
if (frame) {
entry.node->Process(frame);
}
}
}
});
}
}
}
std::cout << "[Graph] started graph " << name_ << " with " << nodes_.size() << " nodes\n";
return true;
}
void Graph::Stop() {
bool expected = true;
if (!running_.compare_exchange_strong(expected, false)) {
// Already stopped or stopping, but we need to ensure threads are joined if called from destructor
// If we are in destructor, running_ might be false but threads joined?
// We should just proceed to cleanup to be safe.
}
// 1. Drain
for (auto& n : nodes_) {
if (n.node) n.node->Drain();
}
// 2. Wait for data to flush
std::this_thread::sleep_for(std::chrono::milliseconds(200));
// 3. Stop queues
for (auto& n : nodes_) {
if (n.context.input_queue) n.context.input_queue->Stop();
for (auto& q : n.context.output_queues) q->Stop();
}
// 4. Join threads
for (auto& n : nodes_) {
if (n.worker.joinable()) {
n.worker.join();
}
}
// 5. Stop nodes
for (auto it = nodes_.rbegin(); it != nodes_.rend(); ++it) {
if (it->node) {
it->node->Stop();
}
if (it->context.input_queue) it->context.input_queue->Stop();
for (auto& q : it->context.output_queues) q->Stop();
}
}