From 943362937b07de37c192afd14a02e5eaa3a50b9f Mon Sep 17 00:00:00 2001 From: sladro Date: Mon, 12 Jan 2026 15:49:01 +0800 Subject: [PATCH] =?UTF-8?q?=E6=80=A7=E8=83=BD=E4=BC=98=E5=8C=961?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- include/graph_manager.h | 10 +- include/utils/spsc_queue.h | 117 +++++++++--- src/graph_manager.cpp | 371 +++++++++++++++++++++++++++++++++---- 3 files changed, 433 insertions(+), 65 deletions(-) diff --git a/include/graph_manager.h b/include/graph_manager.h index 7457a87..139969e 100644 --- a/include/graph_manager.h +++ b/include/graph_manager.h @@ -93,6 +93,8 @@ public: QueueDropStrategy default_strategy, std::string& err); private: + class Executor; + struct NodeMetrics { std::atomic ok_total{0}; std::atomic drop_total{0}; @@ -108,7 +110,12 @@ private: SimpleJson config; NodeContext context; std::unique_ptr node; - std::thread worker; + std::vector cpu_affinity; + + // Scheduling state for Executor. + // Bit0: queued, Bit1: running. + std::atomic sched_state{0}; + uint32_t pool_index = 0; std::shared_ptr metrics; }; @@ -125,6 +132,7 @@ private: QueueDropStrategy built_default_strategy_ = QueueDropStrategy::DropOldest; std::vector nodes_; std::vector edges_; + std::unique_ptr executor_; std::atomic running_{false}; std::atomic stop_requested_{false}; diff --git a/include/utils/spsc_queue.h b/include/utils/spsc_queue.h index 2347e3c..2599bff 100644 --- a/include/utils/spsc_queue.h +++ b/include/utils/spsc_queue.h @@ -3,6 +3,7 @@ #include #include #include +#include #include #include #include @@ -32,30 +33,55 @@ public: buf_.resize(capacity_); } + // 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 cb) { + std::lock_guard lock(mu_); + on_data_available_ = std::move(cb); + } + bool Push(T item) { - std::unique_lock 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; + std::function on_data; + bool notify_data = false; + { + std::unique_lock 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; + } + } + + 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_; } } - buf_[tail_] = std::move(item); - tail_ = (tail_ + 1) % capacity_; - ++size_; - ++pushed_; - data_cv_.notify_one(); + + if (on_data) { + on_data(); + } + if (notify_data) { + data_cv_.notify_one(); + } return true; } @@ -65,11 +91,15 @@ public: return false; } if (size_ == 0) return false; + + const bool was_full = (size_ >= capacity_); out = std::move(buf_[head_]); head_ = (head_ + 1) % capacity_; --size_; ++popped_; - space_cv_.notify_one(); + if (was_full) { + space_cv_.notify_one(); + } return true; } @@ -78,11 +108,51 @@ public: std::unique_lock 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_; - space_cv_.notify_one(); + if (was_full) { + space_cv_.notify_one(); + } + return true; + } + + // Non-blocking pop. + bool TryPop(T& out) { + std::unique_lock 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) { + space_cv_.notify_one(); + } + return true; + } + + // Pop up to max_items currently available (non-blocking). Returns true if any item was popped. + bool TryPopBatch(std::vector& out, size_t max_items) { + if (max_items == 0) return false; + std::unique_lock 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_; + } + size_ -= n; + popped_ += n; + if (was_full) { + space_cv_.notify_one(); + } return true; } @@ -138,6 +208,7 @@ private: mutable std::mutex mu_; std::condition_variable data_cv_; std::condition_variable space_cv_; + std::function on_data_available_; std::vector buf_; size_t head_ = 0; size_t tail_ = 0; diff --git a/src/graph_manager.cpp b/src/graph_manager.cpp index 7d6030c..d5a85d1 100644 --- a/src/graph_manager.cpp +++ b/src/graph_manager.cpp @@ -2,11 +2,15 @@ #include #include +#include +#include #include #include #include #include #include +#include +#include #if __has_include() #include @@ -23,6 +27,52 @@ namespace rk3588 { namespace { +int GetEnvInt(const char* name, int def) { + if (!name) return def; + const char* v = std::getenv(name); + if (!v || !*v) return def; + try { + return std::stoi(v); + } catch (...) { + return def; + } +} + +int DefaultGraphPoolThreads() { + // Override via env: RK3588_GRAPH_POOL_THREADS + const int v = GetEnvInt("RK3588_GRAPH_POOL_THREADS", 0); + if (v > 0 && v <= 64) return v; + const unsigned hc = std::thread::hardware_concurrency(); + const int d = (hc == 0) ? 2 : static_cast(hc); + return std::max(1, std::min(4, d)); +} + +size_t DefaultGraphBatchSize() { + // Override via env: RK3588_GRAPH_BATCH_SIZE + const int v = GetEnvInt("RK3588_GRAPH_BATCH_SIZE", 0); + if (v > 0 && v <= 1024) return static_cast(v); + return 8; +} + +size_t DefaultGraphRunBudget() { + // Max frames processed per scheduled run before yielding. + const int v = GetEnvInt("RK3588_GRAPH_RUN_BUDGET", 0); + if (v > 0 && v <= 4096) return static_cast(v); + return 64; +} + +std::string AffinityKey(std::vector cpus) { + if (cpus.empty()) return ""; + std::sort(cpus.begin(), cpus.end()); + cpus.erase(std::unique(cpus.begin(), cpus.end()), cpus.end()); + std::ostringstream oss; + for (size_t i = 0; i < cpus.size(); ++i) { + if (i) oss << ','; + oss << cpus[i]; + } + return oss.str(); +} + bool WriteTextFile(const std::string& path, const std::string& content, std::string& err) { std::ofstream ofs(path, std::ios::binary | std::ios::trunc); if (!ofs.is_open()) { @@ -145,6 +195,248 @@ bool EffectiveQueueForEdge(const ParsedEdge& e, const SimpleJson* from_node_queu } // namespace +class Graph::Executor { +public: + explicit Executor(Graph& g) + : g_(g), batch_size_(DefaultGraphBatchSize()), run_budget_(DefaultGraphRunBudget()) {} + + bool Start() { + Stop(); + stop_.store(false); + + // Build pools: default pool (no affinity) + pools for explicit cpu_affinity. + pools_.clear(); + key_to_pool_.clear(); + + auto add_pool = [&](const std::string& key, std::vector cpus, int threads) -> uint32_t { + Pool p; + p.key = key; + p.cpus = std::move(cpus); + p.thread_count = std::max(1, threads); + pools_.push_back(std::move(p)); + const uint32_t idx = static_cast(pools_.size() - 1); + key_to_pool_[key] = idx; + return idx; + }; + + default_pool_index_ = add_pool("", {}, DefaultGraphPoolThreads()); + + for (auto& n : g_.nodes_) { + if (!n.enabled || !n.node) continue; + if (n.role == "source" || !n.context.input_queue) continue; + + n.cpu_affinity = ParseCpuAffinity(n.config); + const std::string key = AffinityKey(n.cpu_affinity); + if (!key.empty() && key_to_pool_.find(key) == key_to_pool_.end()) { + add_pool(key, n.cpu_affinity, 1); + } + } + + // Assign nodes to pools. + for (auto& n : g_.nodes_) { + if (!n.enabled || !n.node) continue; + if (n.role == "source" || !n.context.input_queue) continue; + + const std::string key = AffinityKey(n.cpu_affinity); + auto it = key_to_pool_.find(key); + n.pool_index = (it != key_to_pool_.end()) ? it->second : default_pool_index_; + n.sched_state.store(0); + } + + // Start pool threads. + for (auto& p : pools_) { + for (int i = 0; i < p.thread_count; ++i) { + p.threads.emplace_back([this, &p]() { WorkerLoop(p); }); + } + } + + started_ = true; + return true; + } + + void Stop() { + if (!started_) return; + stop_.store(true); + + for (auto& p : pools_) { + { + std::lock_guard lock(p.mu); + p.ready.clear(); + } + p.cv.notify_all(); + } + + for (auto& p : pools_) { + for (auto& t : p.threads) { + if (t.joinable()) t.join(); + } + p.threads.clear(); + { + std::lock_guard lock(p.mu); + p.ready.clear(); + } + } + pools_.clear(); + key_to_pool_.clear(); + started_ = false; + } + + void AttachQueueCallbacks() { + for (auto& n : g_.nodes_) { + if (!n.enabled || !n.node) continue; + if (n.role == "source" || !n.context.input_queue) continue; + NodeEntry* ptr = &n; + n.context.input_queue->SetOnDataAvailable([this, ptr]() { Schedule(ptr); }); + } + } + + void DetachQueueCallbacks() { + for (auto& n : g_.nodes_) { + if (!n.enabled || !n.node) continue; + if (n.role == "source" || !n.context.input_queue) continue; + n.context.input_queue->SetOnDataAvailable(nullptr); + } + } + + void Schedule(NodeEntry* n) { + if (!n) return; + if (!n->enabled || !n->node || !n->context.input_queue) return; + + constexpr uint32_t kQueued = 1u; + constexpr uint32_t kRunning = 2u; + + const uint32_t prev = n->sched_state.fetch_or(kQueued, std::memory_order_acq_rel); + if (prev & kQueued) return; + if (prev & kRunning) return; + + const uint32_t idx = (n->pool_index < pools_.size()) ? n->pool_index : default_pool_index_; + Pool& p = pools_[idx]; + { + std::lock_guard lock(p.mu); + p.ready.push_back(n); + } + p.cv.notify_one(); + } + +private: + struct Pool { + std::string key; + std::vector cpus; + int thread_count = 1; + std::mutex mu; + std::condition_variable cv; + std::deque ready; + std::vector threads; + }; + + void WorkerLoop(Pool& p) { + if (!p.cpus.empty()) { + std::string aerr; + if (!SetCurrentThreadAffinity(p.cpus, aerr)) { + LogWarn("[Graph] SetCurrentThreadAffinity failed for pool: " + aerr); + } + } + + while (true) { + NodeEntry* n = nullptr; + { + std::unique_lock lock(p.mu); + p.cv.wait(lock, [&]() { return stop_.load() || !p.ready.empty(); }); + if (stop_.load() && p.ready.empty()) { + return; + } + if (!p.ready.empty()) { + n = p.ready.front(); + p.ready.pop_front(); + } + } + + if (n) { + RunNode(*n, p); + } + } + } + + void RunNode(NodeEntry& entry, Pool& p) { + if (!entry.enabled || !entry.node || !entry.context.input_queue) return; + + constexpr uint32_t kQueued = 1u; + constexpr uint32_t kRunning = 2u; + + // Mark running. + entry.sched_state.fetch_or(kRunning, std::memory_order_acq_rel); + + std::vector batch; + batch.reserve(batch_size_); + + while (true) { + size_t processed = 0; + + // Drain up to run_budget_ frames to avoid starving other nodes. + while (processed < run_budget_ && entry.context.input_queue->TryPopBatch(batch, batch_size_)) { + for (auto& frame : batch) { + if (!frame) continue; + + const auto t0 = std::chrono::steady_clock::now(); + NodeStatus st = entry.node->Process(frame); + const auto t1 = std::chrono::steady_clock::now(); + const auto ns = std::chrono::duration_cast(t1 - t0).count(); + if (entry.metrics) { + if (ns > 0) { + entry.metrics->process_time_ns_total.fetch_add(static_cast(ns), + std::memory_order_relaxed); + } + if (st == NodeStatus::OK) { + entry.metrics->ok_total.fetch_add(1, std::memory_order_relaxed); + } else if (st == NodeStatus::DROP) { + entry.metrics->drop_total.fetch_add(1, std::memory_order_relaxed); + } else { + entry.metrics->error_total.fetch_add(1, std::memory_order_relaxed); + } + } + + ++processed; + if (processed >= run_budget_) break; + } + } + + // If we hit budget and there's still work, yield by re-enqueueing. + if (processed >= run_budget_ && entry.context.input_queue->Size() > 0) { + entry.sched_state.fetch_or(kQueued, std::memory_order_acq_rel); + entry.sched_state.fetch_and(~kRunning, std::memory_order_acq_rel); + { + std::lock_guard lock(p.mu); + p.ready.push_back(&entry); + } + p.cv.notify_one(); + return; + } + + // No immediate items; if someone scheduled while we were running, continue. + uint32_t st = entry.sched_state.load(std::memory_order_acquire); + if (st & kQueued) { + entry.sched_state.fetch_and(~kQueued, std::memory_order_acq_rel); + continue; + } + + // Try to transition RUNNING -> 0. If it fails, a concurrent schedule happened; loop again. + uint32_t expected = kRunning; + if (entry.sched_state.compare_exchange_strong(expected, 0u, std::memory_order_acq_rel)) { + return; + } + } + } + + Graph& g_; + size_t batch_size_ = 8; + size_t run_budget_ = 64; + std::atomic stop_{false}; + bool started_ = false; + uint32_t default_pool_index_ = 0; + std::vector pools_; + std::unordered_map key_to_pool_; +}; + Graph::Graph(std::string name) : name_(std::move(name)) {} Graph::~Graph() { Stop(); } @@ -157,6 +449,7 @@ bool Graph::Build(const SimpleJson& graph_cfg, PluginLoader& loader, size_t defa nodes_.clear(); edges_.clear(); + executor_.reset(); { std::lock_guard lock(rate_mu_); last_rate_tp_ = {}; @@ -506,52 +799,41 @@ bool Graph::Start() { stop_requested_.store(false); + auto is_source_like = [](const NodeEntry& n) { + return n.role == "source" || !n.context.input_queue; + }; + + // 1) Start non-source nodes first (they should only allocate resources). for (auto& entry : nodes_) { if (!entry.enabled || !entry.node) continue; - + if (is_source_like(entry)) continue; if (!entry.node->Start()) { std::cerr << "[Graph] failed to start node: " << entry.id << "\n"; + Stop(); return false; } + } - // For non-Source nodes, start framework thread - if (entry.role != "source") { - if (entry.context.input_queue) { - entry.worker = std::thread([this, &entry]() { - { - const auto cpus = ParseCpuAffinity(entry.config); - std::string aerr; - if (!cpus.empty() && !SetCurrentThreadAffinity(cpus, aerr)) { - LogWarn("[Graph] SetCurrentThreadAffinity failed for node " + entry.id + ": " + aerr); - } - } - FramePtr frame; - while (entry.context.input_queue->Pop(frame)) { - if (!frame) continue; + // 2) Start executor + attach callbacks before any source can push frames. + executor_ = std::make_unique(*this); + if (!executor_->Start()) { + std::cerr << "[Graph] failed to start executor\n"; + Stop(); + return false; + } + executor_->AttachQueueCallbacks(); - const auto t0 = std::chrono::steady_clock::now(); - NodeStatus st = entry.node->Process(frame); - const auto t1 = std::chrono::steady_clock::now(); - const auto ns = std::chrono::duration_cast(t1 - t0).count(); - if (entry.metrics) { - if (ns > 0) { - entry.metrics->process_time_ns_total.fetch_add(static_cast(ns), - std::memory_order_relaxed); - } - if (st == NodeStatus::OK) { - entry.metrics->ok_total.fetch_add(1, std::memory_order_relaxed); - } else if (st == NodeStatus::DROP) { - entry.metrics->drop_total.fetch_add(1, std::memory_order_relaxed); - } else { - entry.metrics->error_total.fetch_add(1, std::memory_order_relaxed); - } - } - } - for (auto& q : entry.context.output_queues) q->Stop(); - }); - } + // 3) Start sources last. + for (auto& entry : nodes_) { + if (!entry.enabled || !entry.node) continue; + if (!is_source_like(entry)) continue; + if (!entry.node->Start()) { + std::cerr << "[Graph] failed to start node: " << entry.id << "\n"; + Stop(); + return false; } } + std::cout << "[Graph] started graph " << name_ << " with " << nodes_.size() << " nodes\n"; return true; } @@ -577,19 +859,26 @@ void Graph::Stop() { for (auto& q : n.context.output_queues) q->Stop(); } - // 2) Join framework workers after upstream queues are stopped and drained. - for (auto& n : nodes_) { - if (n.worker.joinable()) n.worker.join(); + // 2) Stop all edge queues to unblock downstream nodes. + for (auto& e : edges_) { + if (e.queue) e.queue->Stop(); } - // 3) Drain non-source nodes (no concurrent Process() at this point). + // 3) Stop executor (no concurrent Process() beyond this point). + if (executor_) { + executor_->DetachQueueCallbacks(); + executor_->Stop(); + executor_.reset(); + } + + // 4) Drain non-source nodes. for (auto& n : nodes_) { if (!n.enabled || !n.node) continue; if (is_source_like(n)) continue; n.node->Drain(); } - // 4) Stop remaining nodes (reverse order). + // 5) Stop remaining nodes (reverse order). for (auto it = nodes_.rbegin(); it != nodes_.rend(); ++it) { if (!it->enabled || !it->node) continue; if (is_source_like(*it)) continue;