diff --git a/include/graph_manager.h b/include/graph_manager.h index 6fe36ba..f057798 100644 --- a/include/graph_manager.h +++ b/include/graph_manager.h @@ -5,6 +5,8 @@ #include #include #include +#include +#include #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 node; + std::thread worker; }; std::string name_; std::vector nodes_; + std::atomic running_{false}; }; class GraphManager { diff --git a/include/node.h b/include/node.h index 98d19a6..fe81ff2 100644 --- a/include/node.h +++ b/include/node.h @@ -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. diff --git a/include/utils/spsc_queue.h b/include/utils/spsc_queue.h index 630bffa..0436b90 100644 --- a/include/utils/spsc_queue.h +++ b/include/utils/spsc_queue.h @@ -62,7 +62,10 @@ public: size_t Capacity() const { return capacity_; } - size_t DroppedCount() const { return dropped_; } + size_t DroppedCount() const { + std::lock_guard lock(mu_); + return dropped_; + } private: size_t capacity_ = 0; diff --git a/plugins/ai_yolo/ai_yolo_node.cpp b/plugins/ai_yolo/ai_yolo_node.cpp index 39d4f36..21d365e 100644 --- a/plugins/ai_yolo/ai_yolo_node.cpp +++ b/plugins/ai_yolo/ai_yolo_node.cpp @@ -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 class_filter_; - std::atomic running_{false}; std::shared_ptr> input_queue_; std::vector>> output_queues_; - std::thread worker_; uint64_t processed_ = 0; #if defined(RK3588_ENABLE_RKNN) diff --git a/plugins/alarm/alarm_node.cpp b/plugins/alarm/alarm_node.cpp index 76f3266..084de22 100644 --- a/plugins/alarm/alarm_node.cpp +++ b/plugins/alarm/alarm_node.cpp @@ -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> actions_; ClipAction* clip_action_ = nullptr; - std::atomic running_{false}; std::shared_ptr> input_queue_; - std::thread worker_; uint64_t processed_frames_ = 0; uint64_t alarm_count_ = 0; }; diff --git a/plugins/input_rtsp/input_rtsp_node.cpp b/plugins/input_rtsp/input_rtsp_node.cpp index ab53a1e..8bf6832 100644 --- a/plugins/input_rtsp/input_rtsp_node.cpp +++ b/plugins/input_rtsp/input_rtsp_node.cpp @@ -93,6 +93,10 @@ public: if (worker_.joinable()) worker_.join(); } + void Drain() override { + running_.store(false); + } + private: void LoopStub() { using namespace std::chrono; diff --git a/plugins/osd/osd_node.cpp b/plugins/osd/osd_node.cpp index dfbdee1..960c073 100644 --- a/plugins/osd/osd_node.cpp +++ b/plugins/osd/osd_node.cpp @@ -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(labels_.size())) { @@ -408,10 +396,8 @@ private: int font_scale_ = 1; std::vector labels_; - std::atomic running_{false}; std::shared_ptr> input_queue_; std::vector>> output_queues_; - std::thread worker_; uint64_t processed_ = 0; }; diff --git a/plugins/preprocess/preprocess_node.cpp b/plugins/preprocess/preprocess_node.cpp index 87c8c11..d53f204 100644 --- a/plugins/preprocess/preprocess_node.cpp +++ b/plugins/preprocess/preprocess_node.cpp @@ -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(dst_w_) / frame->width, + static_cast(dst_h_) / frame->height); + out_w = static_cast(frame->width * scale); + out_h = static_cast(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(dst_w_) / frame->width, - static_cast(dst_h_) / frame->height); - out_w = static_cast(frame->width * scale); - out_h = static_cast(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(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(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(); - 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(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(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(); + 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(dst_w_) / frame->width, - static_cast(dst_h_) / frame->height); - out_w = static_cast(frame->width * scale); - out_h = static_cast(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>(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(); - 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(dst_w_) / frame->width, + static_cast(dst_h_) / frame->height); + out_w = static_cast(frame->width * scale); + out_h = static_cast(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>(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(); + 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 running_{false}; std::shared_ptr> input_queue_; std::vector>> 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"); diff --git a/plugins/publish/publish_node.cpp b/plugins/publish/publish_node.cpp index d4570b6..38a2423 100644 --- a/plugins/publish/publish_node.cpp +++ b/plugins/publish/publish_node.cpp @@ -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(); + } + + 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(); + mux_mgr_->Init(ff_outputs_, cid, frame->width, frame->height, fps_, mpp_encoder_->Header()); + } +#endif - std::vector ff_outputs; #if defined(RK3588_ENABLE_ZLMEDIAKIT) - std::vector zlm_outputs; - std::vector> zlm_pubs; + for (const auto& o : zlm_outputs_) { + auto pub = std::make_unique(); + 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(); - 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 outputs_; - std::atomic running_{false}; + std::vector ff_outputs_; + std::vector zlm_outputs_; + std::shared_ptr> input_queue_; - std::thread worker_; uint64_t encoded_frames_ = 0; + +#if defined(RK3588_ENABLE_MPP) + std::unique_ptr mpp_encoder_; + bool encoder_ready_ = false; +#endif + +#if defined(RK3588_ENABLE_FFMPEG) + std::unique_ptr mux_mgr_; +#endif + +#if defined(RK3588_ENABLE_ZLMEDIAKIT) + std::vector> zlm_pubs_; +#endif }; REGISTER_NODE(PublishNode, "publish"); diff --git a/plugins/storage/storage_node.cpp b/plugins/storage/storage_node.cpp index 326ba20..53fe39c 100644 --- a/plugins/storage/storage_node.cpp +++ b/plugins/storage/storage_node.cpp @@ -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 running_{false}; std::shared_ptr> input_queue_; - std::thread worker_; uint64_t total_frames_ = 0; // Current file state diff --git a/src/graph_manager.cpp b/src/graph_manager.cpp index 4cff3ea..e6491e7 100644 --- a/src/graph_manager.cpp +++ b/src/graph_manager.cpp @@ -4,6 +4,8 @@ #include #include #include +#include +#include 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("id", ""); entry.type = node_val.ValueOr("type", ""); + entry.role = node_val.ValueOr("role", ""); entry.enabled = node_val.ValueOr("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 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(qcfg->ValueOr("size", static_cast(default_queue_size))); - std::string strat = qcfg->ValueOr("strategy", "drop_oldest"); - if (strat == "drop_oldest") strategy = QueueDropStrategy::DropOldest; - else strategy = QueueDropStrategy::Block; + std::string policy = qcfg->ValueOr("policy", ""); + std::string strat = qcfg->ValueOr("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>(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(); } }