From e0ca2b48de822aa1883760dcd7089401ab7978c3 Mon Sep 17 00:00:00 2001 From: Kyle Simpson Date: Fri, 28 Aug 2026 19:28:34 +0100 Subject: [PATCH 1/3] Limit eviction search to at most 2048 entries This commit introduces a hard limit to the number of elements that a slowpath traversal of an LFT is willing to check for eviction scores. In practice, walking the entire datastructrure has proven unacceptably expensive, and greatly harms our overall throughput at high load. Each `FlowTable` includes an eviction counter that we use as the starting point for a scan. This is filled in whenever an eviction lookup runs past its budget, if that budget is lower than the total size. This prevents us from rechecking the same entries over and over. There is more work to be done around making sure that we can have better estimates ahead-of-time of likely-evictable flows, but we're not there yet! Flow cleanup is already an expensive operation, such that we need to narrow lock granularity before we can push any pre-prep work into the periodic task. Closes #1041. --- lib/opte/src/engine/flow_table.rs | 94 +++++++++++++++++++++---------- lib/opte/src/engine/layer.rs | 71 ++++++++++++----------- 2 files changed, 103 insertions(+), 62 deletions(-) diff --git a/lib/opte/src/engine/flow_table.rs b/lib/opte/src/engine/flow_table.rs index 0ce5c914..6bd34791 100644 --- a/lib/opte/src/engine/flow_table.rs +++ b/lib/opte/src/engine/flow_table.rs @@ -28,6 +28,7 @@ use core::sync::atomic::AtomicU64; use core::sync::atomic::Ordering; #[cfg(all(not(feature = "std"), not(test)))] use illumos_sys_hdrs::uintptr_t; +use itertools::Either; use opte_api::OpteError; use serde::Serialize; use serde::de::DeserializeOwned; @@ -266,6 +267,15 @@ pub struct FlowTable { limit: NonZeroU32, policy: Arc>, map: BTreeMap>>, + + // When looking up an eviction candidate, we cannot perform a whole table + // scan. Ideally we would have a secondary candidate list maintained by + // the periodic cleanup task. + // + // For now, and in any future case where we have no valid entries in said + // list, we want to examine a subset of `map` but don't want to keep + // rechecking the first `n` entries. + eviction_cursor: Option, } impl FlowTable { @@ -475,47 +485,72 @@ impl FlowTable { /// /// Entries which have been killed due to the loss of a dependency will be /// used where possible. - pub fn find_evictable_entry(&self) -> Option<(InnerFlowId, &FlowEntry)> { + pub fn find_evictable_entry( + &mut self, + ) -> Option<(InnerFlowId, &FlowEntry)> { let now = Moment::now(); // TODO: some form of datastructure to accelerate this? // Who would be responsible for keeping that up to date? - // If that cache is wrong, we're just hitting the O(n) scan anyhow. + // If that cache is wrong, we're just hitting the limited scan anyhow. + const SCAN_BUDGET: usize = 2048; + let len = self.map.len(); + let to_scan = len.min(SCAN_BUDGET); let mut to_evict = None; - for (key, entry) in self.map.iter() { - if entry.is_killed() { - to_evict = Some((EvictionKey::Dead, *key, entry)); - break; - } - // If we have no information, then default to preserving the flow. - let prio = entry.eviction_priority(now).unwrap_or_default(); - if let EvictionPriority::Protected = prio { - continue; - } + let mut visited = 0; + while visited < to_scan { + let map = if let Some(from) = self.eviction_cursor.take() { + Either::Left(self.map.range(from..)) + } else { + Either::Right(self.map.iter()) + }; - let last_hit = entry.last_hit(); + for (key, entry) in map { + // Note: we can't use .take() on map because we need to discover + // the flow ID we'll be starting with on the next scan. + if visited >= to_scan { + self.eviction_cursor = Some(*key); + break; + } + visited += 1; - match to_evict { - None => { - to_evict = Some(( - EvictionKey::Evictable(prio, last_hit), - *key, - entry, - )) + if entry.is_killed() { + to_evict = Some((EvictionKey::Dead, *key, entry)); + break; } - Some((EvictionKey::Evictable(curr_prio, curr_time), ..)) - if prio > curr_prio + + // If we have no information, then default to preserving the flow. + let prio = entry.eviction_priority(now).unwrap_or_default(); + if let EvictionPriority::Protected = prio { + continue; + } + + let last_hit = entry.last_hit(); + + match to_evict { + None => { + to_evict = Some(( + EvictionKey::Evictable(prio, last_hit), + *key, + entry, + )) + } + Some(( + EvictionKey::Evictable(curr_prio, curr_time), + .., + )) if prio > curr_prio || (prio == curr_prio && last_hit < curr_time) => - { - to_evict = Some(( - EvictionKey::Evictable(prio, last_hit), - *key, - entry, - )); + { + to_evict = Some(( + EvictionKey::Evictable(prio, last_hit), + *key, + entry, + )); + } + Some(_) => {} } - Some(_) => {} } } @@ -559,6 +594,7 @@ impl FlowTable { limit, policy, map: BTreeMap::new(), + eviction_cursor: None, } } diff --git a/lib/opte/src/engine/layer.rs b/lib/opte/src/engine/layer.rs index fc542bdc..238ca54b 100644 --- a/lib/opte/src/engine/layer.rs +++ b/lib/opte/src/engine/layer.rs @@ -346,6 +346,36 @@ impl LayerFlowTable { fn num_flows(&self) -> u32 { self.count } + + #[inline(always)] + /// Determine whether there is currently space for a new entry to be + /// inserted. + /// + /// If out of space, this method will attempt to evict an existing entry. + fn check_for_space( + &mut self, + stats: &KStatNamed, + layer_name: &'static str, + dir: Direction, + ) -> result::Result { + if self.count < self.limit.get() { + return Ok(SpaceCreated::AmpleSpace); + } + + // Both in/out share the same `killed` flag, children, and evictability, + // so we only need to check the outbound table. + if let Some((out_key, out_entry)) = self.ft_out.find_evictable_entry() { + let in_key = out_entry.state().in_flow_pair; + Ok(SpaceCreated::Evict { out_key, in_key }) + } else { + let stat = match dir { + Direction::In => &stats.vals.in_lft_full, + Direction::Out => &stats.vals.out_lft_full, + }; + stat.incr(1); + Err(LayerError::FlowTableFull { layer: layer_name, dir }) + } + } } /// The result of a flowtable lookup. @@ -815,35 +845,6 @@ impl Layer { } } - /// Determine whether there is currently space for a new entry to be - /// inserted. - /// - /// If out of space, this method will attempt to evict an existing entry. - fn check_for_space( - &self, - dir: Direction, - ) -> result::Result { - if self.ft.count < self.ft.limit.get() { - return Ok(SpaceCreated::AmpleSpace); - } - - // Both in/out share the same `killed` flag, children, and evictability, - // so we only need to check the outbound table. - if let Some((out_key, out_entry)) = - self.ft.ft_out.find_evictable_entry() - { - let in_key = out_entry.state().in_flow_pair; - Ok(SpaceCreated::Evict { out_key, in_key }) - } else { - let stat = match dir { - Direction::In => &self.stats.vals.in_lft_full, - Direction::Out => &self.stats.vals.out_lft_full, - }; - stat.incr(1); - Err(LayerError::FlowTableFull { layer: self.name, dir }) - } - } - fn complete_eviction(&mut self, entry: SpaceCreated) { if let SpaceCreated::Evict { in_key, out_key } = entry { self.stats.vals.evictions.incr(1); @@ -978,7 +979,8 @@ impl Layer { Action::Allow => Ok(LayerResult::Allow), Action::StatefulAllow => { - let write_to = self.check_for_space(In)?; + let write_to = + self.ft.check_for_space(&self.stats, self.name, In)?; self.complete_eviction(write_to); // The outbound flow ID mirrors the inbound. Remember, @@ -1084,7 +1086,8 @@ impl Layer { // In general, the semantic of a StatefulAction is // that it gets an FT entry. If there are no slots // available, then we must fail until one opens up. - let write_to = self.check_for_space(In)?; + let write_to = + self.ft.check_for_space(&self.stats, self.name, In)?; let desc = match action.gen_desc(pkt.flow(), pkt, ameta) { Ok(aord) => match aord { @@ -1264,7 +1267,8 @@ impl Layer { Action::Allow => Ok(LayerResult::Allow), Action::StatefulAllow => { - let write_to = self.check_for_space(Out)?; + let write_to = + self.ft.check_for_space(&self.stats, self.name, Out)?; self.complete_eviction(write_to); // The inbound flow ID must be calculated _after_ the @@ -1376,7 +1380,8 @@ impl Layer { // In general, the semantic of a StatefulAction is // that it gets an FT entry. If there are no slots // available, then we must fail until one opens up. - let write_to = self.check_for_space(Out)?; + let write_to = + self.ft.check_for_space(&self.stats, self.name, Out)?; let desc = match action.gen_desc(pkt.flow(), pkt, ameta) { Ok(aord) => match aord { From fdad819cebb825b1ea968cb55a1cca508b59cafb Mon Sep 17 00:00:00 2001 From: Kyle Simpson Date: Tue, 1 Sep 2026 14:24:25 +0100 Subject: [PATCH 2/3] Test that checkpointing works as expected. --- lib/opte/src/engine/flow_table.rs | 147 ++++++++++++++++++++++-------- 1 file changed, 110 insertions(+), 37 deletions(-) diff --git a/lib/opte/src/engine/flow_table.rs b/lib/opte/src/engine/flow_table.rs index 6bd34791..c6bca03e 100644 --- a/lib/opte/src/engine/flow_table.rs +++ b/lib/opte/src/engine/flow_table.rs @@ -279,6 +279,10 @@ pub struct FlowTable { } impl FlowTable { + /// The maximum number of distinct entries that a single call to + /// [`Self::find_evictable_entry`] will consider. + const SCAN_BUDGET: usize = 2048; + /// Add a new entry to the flow table, returning a shared refrence to /// the entry. /// @@ -493,15 +497,16 @@ impl FlowTable { // TODO: some form of datastructure to accelerate this? // Who would be responsible for keeping that up to date? // If that cache is wrong, we're just hitting the limited scan anyhow. - const SCAN_BUDGET: usize = 2048; let len = self.map.len(); - let to_scan = len.min(SCAN_BUDGET); + let to_scan = len.min(Self::SCAN_BUDGET); let mut to_evict = None; let mut visited = 0; while visited < to_scan { - let map = if let Some(from) = self.eviction_cursor.take() { + let map = if let Some(from) = self.eviction_cursor.take() + && to_scan < len + { Either::Left(self.map.range(from..)) } else { Either::Right(self.map.iter()) @@ -554,7 +559,10 @@ impl FlowTable { } } - to_evict.map(|(_, k, v)| (k, v.as_ref())) + to_evict.map(|(_, k, v)| { + self.eviction_cursor = Some(k); + (k, v.as_ref()) + }) } /// Get the maximum number of entries this flow table may hold. @@ -957,7 +965,7 @@ mod test { } } - pub const FT_SIZE: Option = NonZeroU32::new(16); + pub const FT_SIZE: NonZeroU32 = NonZeroU32::new(16).unwrap(); #[derive(Debug, Clone)] struct FixedPolicy { @@ -992,8 +1000,7 @@ mod test { proto_info: PortInfo { src_port: 37890, dst_port: 443 }.into(), }; - let mut ft = - FlowTable::new("port", "flow-expired-test", FT_SIZE.unwrap(), None); + let mut ft = FlowTable::new("port", "flow-expired-test", FT_SIZE, None); assert_eq!(ft.num_flows(), 0); ft.add(flowid, ()).unwrap(); let now = Moment::now(); @@ -1015,8 +1022,7 @@ mod test { proto_info: PortInfo { src_port: 37890, dst_port: 443 }.into(), }; - let mut ft = - FlowTable::new("port", "flow-clear-test", FT_SIZE.unwrap(), None); + let mut ft = FlowTable::new("port", "flow-clear-test", FT_SIZE, None); assert_eq!(ft.num_flows(), 0); ft.add(flowid, ()).unwrap(); assert_eq!(ft.num_flows(), 1); @@ -1035,10 +1041,8 @@ mod test { proto_info: PortInfo { src_port: 37890, dst_port: 443 }.into(), }; - let mut ft1 = - FlowTable::new("port", "parent-table", FT_SIZE.unwrap(), None); - let mut ft2 = - FlowTable::new("port", "child-table", FT_SIZE.unwrap(), None); + let mut ft1 = FlowTable::new("port", "parent-table", FT_SIZE, None); + let mut ft2 = FlowTable::new("port", "child-table", FT_SIZE, None); let fe1 = ft1.add(flowid, ()).unwrap(); let fe2 = ft2.add(flowid, ()).unwrap(); @@ -1071,10 +1075,8 @@ mod test { proto_info: PortInfo { src_port: 37890, dst_port: 443 }.into(), }; - let mut ft1 = - FlowTable::new("port", "parent-table", FT_SIZE.unwrap(), None); - let mut ft2 = - FlowTable::new("port", "child-table", FT_SIZE.unwrap(), None); + let mut ft1 = FlowTable::new("port", "parent-table", FT_SIZE, None); + let mut ft2 = FlowTable::new("port", "child-table", FT_SIZE, None); let fe1 = ft1.add(flowid, ()).unwrap(); let fe2 = ft2.add(flowid, ParentSet(vec![fe1.clone() as Arc<_>])).unwrap(); @@ -1108,11 +1110,11 @@ mod test { // Fill up the tables. let mut default_ft = - FlowTable::new("port", "no-prio-table", FT_SIZE.unwrap(), None); + FlowTable::new("port", "no-prio-table", FT_SIZE, None); let mut evict_ft = FlowTable::new( "port", "prio-table", - FT_SIZE.unwrap(), + FT_SIZE, Some(Arc::new(FixedPolicy { time: Duration::from_secs(FLOW_DEF_EXPIRE_SECS), default: Some(EvictionPriority::Evictable(NonZeroU16::MIN)), @@ -1136,18 +1138,18 @@ mod test { // With the default policy and a table full of UDP entries, we can't make // room for anything new. assert!(default_ft.add(flowid, ()).is_err()); - assert_eq!(default_ft.num_flows(), FT_SIZE.unwrap().get()); + assert_eq!(default_ft.num_flows(), FT_SIZE.get()); // On a table where every flow is evictable, we can! assert!(evict_ft.add(flowid, ()).is_ok()); - assert_eq!(evict_ft.num_flows(), FT_SIZE.unwrap().get()); + assert_eq!(evict_ft.num_flows(), FT_SIZE.get()); // If we soft-kill a flow entry (i.e., one of its ancestors was evicted) // then we can make room to insert a new one. default_ft.map.values().next().unwrap().mark_evicted(); - assert_eq!(default_ft.num_flows(), FT_SIZE.unwrap().get()); + assert_eq!(default_ft.num_flows(), FT_SIZE.get()); assert!(default_ft.add(flowid, ()).is_ok()); - assert_eq!(default_ft.num_flows(), FT_SIZE.unwrap().get()); + assert_eq!(default_ft.num_flows(), FT_SIZE.get()); } #[test] @@ -1173,7 +1175,7 @@ mod test { let mut evict_ft = FlowTable::new( "port", "prio-table", - FT_SIZE.unwrap(), + FT_SIZE, Some(Arc::new(FixedPolicy { time: Duration::from_secs(FLOW_DEF_EXPIRE_SECS), default: Some(EvictionPriority::Evictable(NonZeroU16::MIN)), @@ -1205,7 +1207,7 @@ mod test { // This is the entry we will evict, regardless of the age of all others. assert!(evict_ft.map.contains_key(&sacrificial_flow)); assert!(evict_ft.add(flowid, ()).is_ok()); - assert_eq!(evict_ft.num_flows(), FT_SIZE.unwrap().get()); + assert_eq!(evict_ft.num_flows(), FT_SIZE.get()); assert!(!evict_ft.map.contains_key(&sacrificial_flow)); } @@ -1220,14 +1222,11 @@ mod test { proto_info: PortInfo { src_port: 37890, dst_port: 443 }.into(), }; - let mut ft1 = - FlowTable::new("port", "parent-table", FT_SIZE.unwrap(), None); - let mut ft2 = - FlowTable::new("port", "child-table", FT_SIZE.unwrap(), None); + let mut ft1 = FlowTable::new("port", "parent-table", FT_SIZE, None); + let mut ft2 = FlowTable::new("port", "child-table", FT_SIZE, None); let mut ft2_2 = - FlowTable::new("port", "other-child-table", FT_SIZE.unwrap(), None); - let mut ft3 = - FlowTable::new("port", "grandchild-table", FT_SIZE.unwrap(), None); + FlowTable::new("port", "other-child-table", FT_SIZE, None); + let mut ft3 = FlowTable::new("port", "grandchild-table", FT_SIZE, None); let fe1 = ft1.add(flowid, ()).unwrap(); let fe2 = ft2.add(flowid, ()).unwrap(); let fe_out_of_chain = ft2_2.add(flowid, ()).unwrap(); @@ -1259,12 +1258,11 @@ mod test { proto_info: PortInfo { src_port: 37890, dst_port: 443 }.into(), }; - let mut ft1 = - FlowTable::new("port", "parent-table", FT_SIZE.unwrap(), None); + let mut ft1 = FlowTable::new("port", "parent-table", FT_SIZE, None); let mut ft2 = FlowTable::new( "port", "child-table", - FT_SIZE.unwrap(), + FT_SIZE, Some(Arc::new(FixedPolicy { time: Duration::from_secs(FLOW_DEF_EXPIRE_SECS), default: Some(EvictionPriority::Evictable(NonZeroU16::MAX)), @@ -1274,7 +1272,7 @@ mod test { let mut ft2_2 = FlowTable::new( "port", "other-child-table", - FT_SIZE.unwrap(), + FT_SIZE, Some(Arc::new(FixedPolicy { time: Duration::from_secs(FLOW_DEF_EXPIRE_SECS), default: Some(EvictionPriority::Evictable(NonZeroU16::MIN)), @@ -1284,7 +1282,7 @@ mod test { let mut ft2_3 = FlowTable::new( "port", "other-other-child-table", - FT_SIZE.unwrap(), + FT_SIZE, Some(Arc::new(FixedPolicy { time: Duration::from_secs(FLOW_DEF_EXPIRE_SECS), default: Some(EvictionPriority::Protected), @@ -1332,6 +1330,81 @@ mod test { ); } + #[test] + fn eviction_candidate_scan_saves_progress() { + let scan_budget = u32::try_from(FlowTable::<()>::SCAN_BUDGET).unwrap(); + let table_size: NonZeroU32 = 5000.try_into().unwrap(); + let perturb_at = 4500; + assert!(table_size.get() > scan_budget); + assert!((scan_budget..table_size.get()).contains(&perturb_at)); + + let sacrificial_flow = InnerFlowId { + proto: Protocol::UDP.into(), + addrs: AddrPair::V4 { + src: "192.168.2.10".parse().unwrap(), + dst: "76.76.21.21".parse().unwrap(), + }, + proto_info: PortInfo { src_port: perturb_at as u16, dst_port: 443 } + .into(), + }; + + let mut evict_ft = FlowTable::new( + "port", + "prio-table", + table_size, + Some(Arc::new(FixedPolicy { + time: Duration::from_secs(FLOW_DEF_EXPIRE_SECS), + default: Some(EvictionPriority::Protected), + manual: vec![( + sacrificial_flow, + EvictionPriority::Evictable(16.try_into().unwrap()), + )] + .into_iter() + .collect(), + })), + ); + for i in 0..evict_ft.limit.get() { + let new_id = InnerFlowId { + proto: Protocol::UDP.into(), + addrs: AddrPair::V4 { + src: "192.168.2.10".parse().unwrap(), + dst: "76.76.21.21".parse().unwrap(), + }, + proto_info: PortInfo { src_port: i as u16, dst_port: 443 } + .into(), + }; + evict_ft.add(new_id, ()).unwrap(); + + if i == perturb_at { + assert_eq!(new_id, sacrificial_flow); + let entry = evict_ft.map.get(&new_id).unwrap(); + entry.hit_at(entry.last_hit() - Duration::from_secs(61)); + } + } + + assert!(evict_ft.eviction_cursor.is_none()); + + // The usable entry should only be seen after several scans, in this + // case. Each scan, successful or otherwise, will advance the cursor. + assert!(evict_ft.find_evictable_entry().is_none()); + let c1 = evict_ft.eviction_cursor.unwrap(); + assert!(evict_ft.find_evictable_entry().is_none()); + let c2 = evict_ft.eviction_cursor.unwrap(); + assert!(c1 < c2); + assert!(evict_ft.find_evictable_entry().is_some()); + let c3 = evict_ft.eviction_cursor.unwrap(); + assert!(c2 < c3); + + evict_ft.expire(&c3); + + // A scan should wrap around at the end of the map, if we hit the end + // with some remaining budget. Accordingly exepct that the flow ID + // checkpoint is *lower* this time. + assert!(evict_ft.find_evictable_entry().is_none()); + let c4 = evict_ft.eviction_cursor.unwrap(); + assert!(c4 < c3); + } + #[test] fn priority_lerp() { let low = 10.try_into().unwrap(); From 65d3c2ea5b0c913e538b1ad489fa92ef2ffaae61 Mon Sep 17 00:00:00 2001 From: Kyle Simpson Date: Fri, 4 Sep 2026 13:59:24 +0100 Subject: [PATCH 3/3] Rebase fix --- lib/opte/src/engine/flow_table.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/opte/src/engine/flow_table.rs b/lib/opte/src/engine/flow_table.rs index c6bca03e..7078d561 100644 --- a/lib/opte/src/engine/flow_table.rs +++ b/lib/opte/src/engine/flow_table.rs @@ -1395,7 +1395,7 @@ mod test { let c3 = evict_ft.eviction_cursor.unwrap(); assert!(c2 < c3); - evict_ft.expire(&c3); + evict_ft.expire(&c3, true); // A scan should wrap around at the end of the map, if we hit the end // with some remaining budget. Accordingly exepct that the flow ID