From cc4109e8346921c2876ffdffb9b3cf8043420c9b Mon Sep 17 00:00:00 2001 From: Denis Leshchev Date: Thu, 25 Jun 2026 00:55:06 +0000 Subject: [PATCH 1/2] #205 - Preserve kernel fallback for dynamic RX flows Install the flow-isolation send-to-kernel fallback before programming dynamic RX flows so control-plane traffic (ARP, SSH, application control planes) keeps reaching the Linux kernel. Previously the fallback was installed only for static startup flows, so a queues-only config with flow_isolation: true black-holed all unmatched/kernel traffic once the first dynamic flow was added. - ensure_dynamic_flow_isolation_fallback_locked() installs the ETH jump rule and send-to-kernel fallback under flow_lock_ in add_rx_flow_async and add_rx_flows_async. - Track fallback installation per {port, group} so repeated dynamic flow updates do not duplicate the rule. - Clear fallback tracking when programmed flows are destroyed. - Document the dynamic-flow fallback behavior in configuration.md. Signed-off-by: Denis Leshchev --- docs/api-reference/configuration.md | 8 +++++--- src/engines/dpdk/daqiri_dpdk_engine.cpp | 18 ++++++++++++++++++ src/engines/dpdk/daqiri_dpdk_engine.h | 2 ++ 3 files changed, 25 insertions(+), 3 deletions(-) diff --git a/docs/api-reference/configuration.md b/docs/api-reference/configuration.md index 980db103..f095858f 100644 --- a/docs/api-reference/configuration.md +++ b/docs/api-reference/configuration.md @@ -247,9 +247,11 @@ Flex-item flows cannot be combined with VLAN/tunnel transform actions in v1. to the application. Static startup flows install send-to-kernel fallback rules per flow class (standard or flex-item), so unmatched traffic in those classes is steered back to the Linux kernel. Queues-only configs can set `flow_isolation: true` and then install dynamic RX flows -after `daqiri_init()`; until a dynamic rule is added, application traffic is not delivered to -DAQIRI RX queues. When `false`, unmatched packets go to a default queue. Mixing standard and -flex-item flow classes on one interface is not supported. +after `daqiri_init()`; the first dynamic RX flow installs the same send-to-kernel fallback (so +unmatched control traffic such as ARP keeps reaching the kernel), and until a dynamic rule is +added, application traffic is not delivered to DAQIRI RX queues. When `false`, unmatched packets +go to a default queue. Mixing standard and flex-item flow classes on one interface is not +supported. - type: `boolean` - default: `false` diff --git a/src/engines/dpdk/daqiri_dpdk_engine.cpp b/src/engines/dpdk/daqiri_dpdk_engine.cpp index 5e2f93e6..83d2da9a 100644 --- a/src/engines/dpdk/daqiri_dpdk_engine.cpp +++ b/src/engines/dpdk/daqiri_dpdk_engine.cpp @@ -3136,6 +3136,13 @@ bool DpdkEngine::validate_dynamic_rx_flow(int port, const FlowRuleConfig& flow) return false; } +bool DpdkEngine::ensure_dynamic_flow_isolation_fallback_locked(int port) { + if (port < 0 || port >= static_cast(cfg_.ifs_.size())) { return false; } + if (!cfg_.ifs_[port].rx_.flow_isolation_) { return true; } + if (!ensure_eth_jump_rule(port, kRxFlowGroup)) { return false; } + return add_send_to_kernel_fallback(port, kRxFlowGroup); +} + void DpdkEngine::build_ipv4_udp_flow_pattern(const FlowMatch& match, struct rte_flow_item pattern[MAX_PATTERN_NUM], struct rte_flow_item_ipv4* ip_spec, @@ -3997,6 +4004,9 @@ Status DpdkEngine::add_rx_flow_async(int port, const FlowRuleConfig& flow, FlowO *op_id = 0; std::lock_guard guard(flow_lock_); if (!validate_dynamic_rx_flow(port, flow)) { return Status::INVALID_PARAMETER; } + if (!ensure_dynamic_flow_isolation_fallback_locked(port)) { + return Status::INTERNAL_ERROR; + } const FlowId flow_id = allocate_dynamic_flow_id(); if (flow_id == 0) { return Status::NO_SPACE_AVAILABLE; } @@ -4020,6 +4030,9 @@ Status DpdkEngine::add_rx_flows_async(int port, for (const auto& flow : flows) { if (!validate_dynamic_rx_flow(port, flow)) { return Status::INVALID_PARAMETER; } } + if (!ensure_dynamic_flow_isolation_fallback_locked(port)) { + return Status::INTERNAL_ERROR; + } std::vector flow_ids; flow_ids.reserve(flows.size()); @@ -4143,6 +4156,7 @@ void DpdkEngine::destroy_programmed_flows() { } } programmed_flows_.clear(); + send_to_kernel_fallback_installed_.clear(); for (uint16_t port = 0; port < drop_all_traffic_flow.size(); ++port) { struct rte_flow* drop_flow = drop_all_traffic_flow[port].drop; @@ -4897,6 +4911,9 @@ struct rte_flow* DpdkEngine::add_tx_flow(int port, const FlowConfig& cfg) { } bool DpdkEngine::add_send_to_kernel_fallback(int port, uint32_t group) { + const uint64_t key = eth_jump_key(port, group); + if (send_to_kernel_fallback_installed_.count(key) != 0) { return true; } + struct rte_flow_attr attr; struct rte_flow_item pattern[MAX_PATTERN_NUM]; struct rte_flow_action action[MAX_ACTION_NUM]; @@ -4940,6 +4957,7 @@ bool DpdkEngine::add_send_to_kernel_fallback(int port, uint32_t group) { } track_flow(static_cast(port), flow); + send_to_kernel_fallback_installed_.insert(key); return true; } diff --git a/src/engines/dpdk/daqiri_dpdk_engine.h b/src/engines/dpdk/daqiri_dpdk_engine.h index edf65b00..e5dd3f74 100644 --- a/src/engines/dpdk/daqiri_dpdk_engine.h +++ b/src/engines/dpdk/daqiri_dpdk_engine.h @@ -266,6 +266,7 @@ class DpdkEngine : public Engine { void release_dynamic_flow_id(FlowId flow_id); bool reserve_static_flow_ids(); bool validate_dynamic_rx_flow(int port, const FlowRuleConfig& flow) const; + bool ensure_dynamic_flow_isolation_fallback_locked(int port); bool is_valid_rx_queue(int port, uint16_t queue_id) const; bool is_ipv4_udp_flow_match(const FlowMatch& match) const; bool ipv4_udp_flow_template_index(const FlowMatch& match, uint8_t* template_index) const; @@ -495,6 +496,7 @@ class DpdkEngine : public Engine { std::array flow_template_states_{}; std::vector programmed_flows_; std::unordered_set eth_jump_installed_; + std::unordered_set send_to_kernel_fallback_installed_; std::unordered_map eth_jump_flows_; std::unordered_map tx_rings; std::unordered_map tx_burst_buffers; From e635ee3f7e5c7b2449f946469d37c95f5ec3b118 Mon Sep 17 00:00:00 2001 From: Denis Leshchev Date: Thu, 25 Jun 2026 11:39:28 +0000 Subject: [PATCH 2/2] #205 - Support dynamic flex-item isolation fallback Signed-off-by: Denis Leshchev --- docs/api-reference/configuration.md | 10 ++-- src/engines/dpdk/daqiri_dpdk_engine.cpp | 68 ++++++++++++++++++------- src/engines/dpdk/daqiri_dpdk_engine.h | 2 +- 3 files changed, 56 insertions(+), 24 deletions(-) diff --git a/docs/api-reference/configuration.md b/docs/api-reference/configuration.md index f095858f..66debe41 100644 --- a/docs/api-reference/configuration.md +++ b/docs/api-reference/configuration.md @@ -247,11 +247,11 @@ Flex-item flows cannot be combined with VLAN/tunnel transform actions in v1. to the application. Static startup flows install send-to-kernel fallback rules per flow class (standard or flex-item), so unmatched traffic in those classes is steered back to the Linux kernel. Queues-only configs can set `flow_isolation: true` and then install dynamic RX flows -after `daqiri_init()`; the first dynamic RX flow installs the same send-to-kernel fallback (so -unmatched control traffic such as ARP keeps reaching the kernel), and until a dynamic rule is -added, application traffic is not delivered to DAQIRI RX queues. When `false`, unmatched packets -go to a default queue. Mixing standard and flex-item flow classes on one interface is not -supported. +after `daqiri_init()`; the first dynamic RX flow installs the send-to-kernel fallback for that +flow class (so unmatched control traffic such as ARP keeps reaching the kernel), and until a +dynamic rule is added, application traffic is not delivered to DAQIRI RX queues. When `false`, +unmatched packets go to a default queue. Mixing standard and flex-item flow classes on one +interface is not supported, including across dynamic flow additions or within one dynamic batch. - type: `boolean` - default: `false` diff --git a/src/engines/dpdk/daqiri_dpdk_engine.cpp b/src/engines/dpdk/daqiri_dpdk_engine.cpp index 83d2da9a..1d1bd7b6 100644 --- a/src/engines/dpdk/daqiri_dpdk_engine.cpp +++ b/src/engines/dpdk/daqiri_dpdk_engine.cpp @@ -2864,10 +2864,16 @@ struct FlowTemplateCreateStorage { static constexpr uint32_t kDynamicFlowQueueId = 0; static constexpr FlowId kMaxDynamicFlowMarkId = 0x00ffffffU; -static constexpr uint32_t kRxFlowGroup = 3; +static constexpr uint32_t kFlexItemRxFlowGroup = 1; +static constexpr uint32_t kStandardRxFlowGroup = 3; static constexpr uint32_t kRxFlowPriority = 1; static constexpr size_t kMaxIpv4UdpFlowTemplateFieldsForSingleTable = 7; +static uint32_t dynamic_rx_flow_group(const FlowRuleConfig& flow) { + return flow.match_.type_ == FlowMatchType::FLEX_ITEM ? kFlexItemRxFlowGroup + : kStandardRxFlowGroup; +} + enum class Ipv4UdpFlowTemplateField { IPV4_LEN, IPV4_SRC, @@ -3136,11 +3142,28 @@ bool DpdkEngine::validate_dynamic_rx_flow(int port, const FlowRuleConfig& flow) return false; } -bool DpdkEngine::ensure_dynamic_flow_isolation_fallback_locked(int port) { - if (port < 0 || port >= static_cast(cfg_.ifs_.size())) { return false; } - if (!cfg_.ifs_[port].rx_.flow_isolation_) { return true; } - if (!ensure_eth_jump_rule(port, kRxFlowGroup)) { return false; } - return add_send_to_kernel_fallback(port, kRxFlowGroup); +Status DpdkEngine::ensure_dynamic_flow_isolation_fallback_locked(int port, uint32_t group) { + if (port < 0 || port >= static_cast(cfg_.ifs_.size())) { return Status::INVALID_PARAMETER; } + + for (const auto key : eth_jump_installed_) { + const int installed_port = static_cast(key >> 32); + const uint32_t installed_group = static_cast(key & 0xffffffffU); + if (installed_port != port || installed_group == group) { continue; } + + DAQIRI_LOG_ERROR( + "Dynamic RX flow on port {} requires DPDK flow group {}, but group {} already owns " + "the group-0 ETH jump; standard IPv4/UDP and flex-item flows cannot mix on one " + "interface", + port, + group, + installed_group); + return Status::INVALID_PARAMETER; + } + + if (!cfg_.ifs_[port].rx_.flow_isolation_) { return Status::SUCCESS; } + if (!add_send_to_kernel_fallback(port, group)) { return Status::INTERNAL_ERROR; } + if (!ensure_eth_jump_rule(port, group)) { return Status::INTERNAL_ERROR; } + return Status::SUCCESS; } void DpdkEngine::build_ipv4_udp_flow_pattern(const FlowMatch& match, @@ -3352,7 +3375,7 @@ bool DpdkEngine::ensure_ipv4_udp_flow_template_table(uint16_t port) { struct rte_flow_template_table_attr table_attr; memset(&table_attr, 0, sizeof(table_attr)); table_attr.flow_attr.ingress = 1; - table_attr.flow_attr.group = kRxFlowGroup; + table_attr.flow_attr.group = kStandardRxFlowGroup; table_attr.flow_attr.priority = kRxFlowPriority; table_attr.nb_flows = state.capacity; table_attr.insertion_type = RTE_FLOW_TABLE_INSERTION_TYPE_PATTERN; @@ -3608,7 +3631,7 @@ Status DpdkEngine::add_rx_flow_template_locked(int port, const FlowRuleConfig& flow, FlowId flow_id, FlowOpId op_id) { - if (!ensure_eth_jump_rule(port, kRxFlowGroup) || + if (!ensure_eth_jump_rule(port, kStandardRxFlowGroup) || !ensure_ipv4_udp_flow_template_table(static_cast(port))) { return add_rx_flow_legacy_locked(port, flow, flow_id, op_id); } @@ -3656,7 +3679,7 @@ Status DpdkEngine::add_rx_flows_template_locked(int port, const std::vector& flows, const std::vector& flow_ids, FlowOpId op_id) { - if (!ensure_eth_jump_rule(port, kRxFlowGroup) || + if (!ensure_eth_jump_rule(port, kStandardRxFlowGroup) || !ensure_ipv4_udp_flow_template_table(static_cast(port))) { return add_rx_flows_legacy_locked(port, flows, flow_ids, op_id); } @@ -4004,9 +4027,9 @@ Status DpdkEngine::add_rx_flow_async(int port, const FlowRuleConfig& flow, FlowO *op_id = 0; std::lock_guard guard(flow_lock_); if (!validate_dynamic_rx_flow(port, flow)) { return Status::INVALID_PARAMETER; } - if (!ensure_dynamic_flow_isolation_fallback_locked(port)) { - return Status::INTERNAL_ERROR; - } + const uint32_t flow_group = dynamic_rx_flow_group(flow); + const Status fallback_status = ensure_dynamic_flow_isolation_fallback_locked(port, flow_group); + if (fallback_status != Status::SUCCESS) { return fallback_status; } const FlowId flow_id = allocate_dynamic_flow_id(); if (flow_id == 0) { return Status::NO_SPACE_AVAILABLE; } @@ -4030,9 +4053,18 @@ Status DpdkEngine::add_rx_flows_async(int port, for (const auto& flow : flows) { if (!validate_dynamic_rx_flow(port, flow)) { return Status::INVALID_PARAMETER; } } - if (!ensure_dynamic_flow_isolation_fallback_locked(port)) { - return Status::INTERNAL_ERROR; + const uint32_t flow_group = dynamic_rx_flow_group(flows.front()); + for (const auto& flow : flows) { + const uint32_t current_group = dynamic_rx_flow_group(flow); + if (current_group == flow_group) { continue; } + DAQIRI_LOG_ERROR( + "Dynamic RX flow batch on port {} mixes standard IPv4/UDP and flex-item flows, " + "which cannot share one DPDK group-0 ETH jump", + port); + return Status::INVALID_PARAMETER; } + const Status fallback_status = ensure_dynamic_flow_isolation_fallback_locked(port, flow_group); + if (fallback_status != Status::SUCCESS) { return fallback_status; } std::vector flow_ids; flow_ids.reserve(flows.size()); @@ -4300,7 +4332,7 @@ struct rte_flow* DpdkEngine::add_flex_item_flow( flex_item_handle_key(static_cast(port), match_info.flex_item_id_); auto handle_it = flex_item_handles_.find(handle_key); if (handle_it == flex_item_handles_.end()) { - if (!ensure_eth_jump_rule(port, 1)) { return nullptr; } + if (!ensure_eth_jump_rule(port, kFlexItemRxFlowGroup)) { return nullptr; } struct rte_flow_item_flex_conf flex_conf; flex_conf.tunnel = FLEX_TUNNEL_MODE_SINGLE; @@ -4382,7 +4414,7 @@ struct rte_flow* DpdkEngine::add_flex_item_flow( attr.ingress = 1; attr.priority = 0; - attr.group = 1; + attr.group = kFlexItemRxFlowGroup; res = rte_flow_validate(port, &attr, pattern, action, &error); if (res != 0) { @@ -4754,7 +4786,7 @@ struct rte_flow* DpdkEngine::add_flow(int port, struct rte_flow_item_ipv4 ip_spec {}; struct rte_flow_item_ipv4 ip_mask {}; - if (!ensure_eth_jump_rule(port, 3)) { return nullptr; } + if (!ensure_eth_jump_rule(port, kStandardRxFlowGroup)) { return nullptr; } auto resource = std::make_shared(); resource->port = port; @@ -4811,7 +4843,7 @@ struct rte_flow* DpdkEngine::add_flow(int port, attr.ingress = 1; attr.priority = has_outer_transform ? 0 : 1; - attr.group = 3; + attr.group = kStandardRxFlowGroup; int res = rte_flow_validate(port, &attr, pattern, action, &error); if (res != 0) { diff --git a/src/engines/dpdk/daqiri_dpdk_engine.h b/src/engines/dpdk/daqiri_dpdk_engine.h index e5dd3f74..3dabec26 100644 --- a/src/engines/dpdk/daqiri_dpdk_engine.h +++ b/src/engines/dpdk/daqiri_dpdk_engine.h @@ -266,7 +266,7 @@ class DpdkEngine : public Engine { void release_dynamic_flow_id(FlowId flow_id); bool reserve_static_flow_ids(); bool validate_dynamic_rx_flow(int port, const FlowRuleConfig& flow) const; - bool ensure_dynamic_flow_isolation_fallback_locked(int port); + Status ensure_dynamic_flow_isolation_fallback_locked(int port, uint32_t group); bool is_valid_rx_queue(int port, uint16_t queue_id) const; bool is_ipv4_udp_flow_match(const FlowMatch& match) const; bool ipv4_udp_flow_template_index(const FlowMatch& match, uint8_t* template_index) const;