性能优化1

include/graph_manager.h

@@ -93,6 +93,8 @@ public:
                           QueueDropStrategy default_strategy, std::string& err);
 
 private:
+    class Executor;
+
     struct NodeMetrics {
         std::atomic<uint64_t> ok_total{0};
         std::atomic<uint64_t> drop_total{0};
@@ -108,7 +110,12 @@ private:
         SimpleJson config;
         NodeContext context;
         std::unique_ptr<INode> node;
-        std::thread worker;
+        std::vector<int> cpu_affinity;
+
+        // Scheduling state for Executor.
+        // Bit0: queued, Bit1: running.
+        std::atomic<uint32_t> sched_state{0};
+        uint32_t pool_index = 0;
 
         std::shared_ptr<NodeMetrics> metrics;
     };
@@ -125,6 +132,7 @@ private:
     QueueDropStrategy built_default_strategy_ = QueueDropStrategy::DropOldest;
     std::vector<NodeEntry> nodes_;
     std::vector<EdgeEntry> edges_;
+    std::unique_ptr<Executor> executor_;
     std::atomic<bool> running_{false};
     std::atomic<bool> stop_requested_{false};
 

include/utils/spsc_queue.h

@@ -3,6 +3,7 @@
 #include <chrono>
 #include <condition_variable>
 #include <cstddef>
+#include <functional>
 #include <mutex>
 #include <vector>
 #include <utility>
@@ -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<void()> cb) {
+        std::lock_guard<std::mutex> lock(mu_);
+        on_data_available_ = std::move(cb);
+    }
+
     bool Push(T item) {
-        std::unique_lock<std::mutex> lock(mu_);
-        if (stop_) return false;
-        if (size_ >= capacity_) {
-            if (strategy_ == QueueDropStrategy::DropOldest) {
-                // Drop the oldest item.
-                head_ = (head_ + 1) % capacity_;
-                --size_;
-                ++dropped_;
-            } else if (strategy_ == QueueDropStrategy::DropNewest) {
-                // Drop the incoming item.
-                ++dropped_;
-                return true;
-            } else {
-                // Block until space is available
-                space_cv_.wait(lock, [&] { return size_ < capacity_ || stop_; });
-                if (stop_) return false;
+        std::function<void()> on_data;
+        bool notify_data = false;
+        {
+            std::unique_lock<std::mutex> lock(mu_);
+            if (stop_) return false;
+            if (size_ >= capacity_) {
+                if (strategy_ == QueueDropStrategy::DropOldest) {
+                    // Drop the oldest item.
+                    head_ = (head_ + 1) % capacity_;
+                    --size_;
+                    ++dropped_;
+                } else if (strategy_ == QueueDropStrategy::DropNewest) {
+                    // Drop the incoming item.
+                    ++dropped_;
+                    return true;
+                } else {
+                    // Block until space is available
+                    space_cv_.wait(lock, [&] { return size_ < capacity_ || stop_; });
+                    if (stop_) return false;
+                }
+            }
+
+            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<std::mutex> lock(mu_);
         data_cv_.wait(lock, [&] { return size_ > 0 || stop_; });
         if (size_ == 0) return false;
+
+        const bool was_full = (size_ >= capacity_);
         out = std::move(buf_[head_]);
         head_ = (head_ + 1) % capacity_;
         --size_;
         ++popped_;
-        space_cv_.notify_one();
+        if (was_full) {
+            space_cv_.notify_one();
+        }
+        return true;
+    }
+
+    // Non-blocking pop.
+    bool TryPop(T& out) {
+        std::unique_lock<std::mutex> lock(mu_);
+        if (size_ == 0) return false;
+        const bool was_full = (size_ >= capacity_);
+        out = std::move(buf_[head_]);
+        head_ = (head_ + 1) % capacity_;
+        --size_;
+        ++popped_;
+        if (was_full) {
+            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<T>& out, size_t max_items) {
+        if (max_items == 0) return false;
+        std::unique_lock<std::mutex> lock(mu_);
+        if (size_ == 0) return false;
+        const bool was_full = (size_ >= capacity_);
+        const size_t n = (size_ < max_items) ? size_ : max_items;
+        out.clear();
+        out.reserve(n);
+        for (size_t i = 0; i < n; ++i) {
+            out.push_back(std::move(buf_[head_]));
+            head_ = (head_ + 1) % capacity_;
+        }
+        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<void()> on_data_available_;
     std::vector<T> buf_;
     size_t head_ = 0;
     size_t tail_ = 0;

src/graph_manager.cpp

@@ -2,11 +2,15 @@
 
 #include <fstream>
 #include <iostream>
+#include <algorithm>
+#include <cstdlib>
 #include <map>
 #include <set>
 #include <deque>
 #include <thread>
 #include <chrono>
+#include <unordered_map>
+#include <sstream>
 
 #if __has_include(<filesystem>)
 #include <filesystem>
@@ -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<int>(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<size_t>(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<size_t>(v);
+    return 64;
+}
+
+std::string AffinityKey(std::vector<int> 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<int> 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<uint32_t>(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<std::mutex> 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<std::mutex> 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<std::mutex> lock(p.mu);
+            p.ready.push_back(n);
+        }
+        p.cv.notify_one();
+    }
+
+private:
+    struct Pool {
+        std::string key;
+        std::vector<int> cpus;
+        int thread_count = 1;
+        std::mutex mu;
+        std::condition_variable cv;
+        std::deque<NodeEntry*> ready;
+        std::vector<std::thread> 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<std::mutex> 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<FramePtr> 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<std::chrono::nanoseconds>(t1 - t0).count();
+                    if (entry.metrics) {
+                        if (ns > 0) {
+                            entry.metrics->process_time_ns_total.fetch_add(static_cast<uint64_t>(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<std::mutex> 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<bool> stop_{false};
+    bool started_ = false;
+    uint32_t default_pool_index_ = 0;
+    std::vector<Pool> pools_;
+    std::unordered_map<std::string, uint32_t> 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<std::mutex> 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<Executor>(*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<std::chrono::nanoseconds>(t1 - t0).count();
-                         if (entry.metrics) {
-                             if (ns > 0) {
-                                 entry.metrics->process_time_ns_total.fetch_add(static_cast<uint64_t>(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;