diff --git a/include/utils/spsc_queue.h b/include/utils/spsc_queue.h index 066ceb7..62bfa01 100644 --- a/include/utils/spsc_queue.h +++ b/include/utils/spsc_queue.h @@ -10,6 +10,7 @@ namespace rk3588 { enum class QueueDropStrategy { DropOldest, + DropNewest, Block }; @@ -26,6 +27,10 @@ public: if (strategy_ == QueueDropStrategy::DropOldest) { queue_.pop_front(); ++dropped_; + } else if (strategy_ == QueueDropStrategy::DropNewest) { + // Drop the incoming item. + ++dropped_; + return true; } else { // Block until space is available space_cv_.wait(lock, [&] { return queue_.size() < capacity_ || stop_; }); diff --git a/src/graph_manager.cpp b/src/graph_manager.cpp index 1812d5d..d964b45 100644 --- a/src/graph_manager.cpp +++ b/src/graph_manager.cpp @@ -9,6 +9,33 @@ namespace rk3588 { +namespace { + +QueueDropStrategy ParseDropStrategy(const std::string& s, QueueDropStrategy def) { + if (s == "drop_oldest") return QueueDropStrategy::DropOldest; + if (s == "drop_newest") return QueueDropStrategy::DropNewest; + if (s == "block") return QueueDropStrategy::Block; + return def; +} + +void ApplyQueueConfig(const SimpleJson* queue_cfg, size_t default_queue_size, + QueueDropStrategy default_strategy, size_t& out_size, + QueueDropStrategy& out_strategy) { + out_size = default_queue_size; + out_strategy = default_strategy; + if (!queue_cfg || !queue_cfg->IsObject()) return; + + out_size = static_cast(queue_cfg->ValueOr("size", static_cast(default_queue_size))); + std::string policy = queue_cfg->ValueOr("policy", ""); + std::string strat = queue_cfg->ValueOr("strategy", ""); + std::string final_policy = policy.empty() ? strat : policy; + if (!final_policy.empty()) { + out_strategy = ParseDropStrategy(final_policy, default_strategy); + } +} + +} // namespace + Graph::Graph(std::string name) : name_(std::move(name)) {} Graph::~Graph() { Stop(); } @@ -62,13 +89,36 @@ bool Graph::Build(const SimpleJson& graph_cfg, PluginLoader& loader, size_t defa } for (const auto& edge_val : edges_it->second.AsArray()) { - if (!edge_val.IsArray() || edge_val.AsArray().size() != 2) { - err = "Edge must be [from, to]"; + std::string from; + std::string to; + const SimpleJson* queue_cfg = nullptr; + + // Supported forms: + // 1) ["from", "to"] + // 2) ["from", "to", {"queue": {...}}] + // 3) {"from": "a", "to": "b", "queue": {...}} + if (edge_val.IsArray()) { + const auto& edge_arr = edge_val.AsArray(); + if (edge_arr.size() < 2) { + err = "Edge array must be [from, to] or [from, to, {...}]"; + return false; + } + from = edge_arr[0].AsString(""); + to = edge_arr[1].AsString(""); + if (edge_arr.size() >= 3 && edge_arr[2].IsObject()) { + // Third element can be {queue:{...}} or the queue object itself. + queue_cfg = edge_arr[2].Find("queue"); + if (!queue_cfg) queue_cfg = &edge_arr[2]; + } + } else if (edge_val.IsObject()) { + from = edge_val.ValueOr("from", ""); + to = edge_val.ValueOr("to", ""); + queue_cfg = edge_val.Find("queue"); + } else { + err = "Edge must be an array or object"; return false; } - const auto& edge_arr = edge_val.AsArray(); - std::string from = edge_arr[0].AsString(""); - std::string to = edge_arr[1].AsString(""); + if (from.empty() || to.empty()) { err = "Edge has empty endpoint"; return false; @@ -96,21 +146,7 @@ bool Graph::Build(const SimpleJson& graph_cfg, PluginLoader& loader, size_t defa 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 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; - } - } - } + ApplyQueueConfig(queue_cfg, default_queue_size, default_strategy, qsize, strategy); auto queue = std::make_shared>(qsize, strategy); from_it->second->context.output_queues.push_back(queue); @@ -261,7 +297,7 @@ bool GraphManager::Build(const SimpleJson& root_cfg, std::string& err) { if (queue_cfg->IsObject()) { default_queue_size = static_cast(queue_cfg->ValueOr("size", 8)); std::string strategy = queue_cfg->ValueOr("strategy", "drop_oldest"); - if (strategy == "block") default_strategy = QueueDropStrategy::Block; + default_strategy = ParseDropStrategy(strategy, default_strategy); } }