diff --git a/lib/opte/src/engine/flow_table.rs b/lib/opte/src/engine/flow_table.rs index f7094e1b..0ce5c914 100644 --- a/lib/opte/src/engine/flow_table.rs +++ b/lib/opte/src/engine/flow_table.rs @@ -332,26 +332,85 @@ impl FlowTable { flows } - pub fn expire(&mut self, flowid: &InnerFlowId) { + pub(crate) fn expire(&mut self, flowid: &InnerFlowId, mark_evicted: bool) { flow_expired_probe(&self.port_c, &self.name_c, flowid, None, None); if let Some(entry) = self.map.remove(flowid) { entry.expiry_cleanup(); - entry.mark_evicted(); + if mark_evicted { + entry.mark_evicted(); + } } } - pub fn expire_flows(&mut self, now: Moment, f: F) -> Vec - where + /// Remove all flows from `self` which are past their expiry time. + pub fn expire_flows(&mut self, now: Moment) { + let name_c = &self.name_c; + let port_c = &self.port_c; + + self.map.retain(|flowid, entry| { + // A flow cannot be expired by the timer while it still has children + // relying upon its existence. Check whether any remain, and remove + // dangling references to child entries which have expired. + // + // The dangling entries here will have been left by `expire_flows` + // called on other layers. + { + // We have a write lock on the port, so there shouldn't be + // contention here. + let mut children = entry.lifetime.children.write(); + children.retain(|el| el.0.upgrade().is_some()); + if !children.is_empty() { + return true; + } + } + // If we move to per-layer lock granularity, then we may need to extend + // the lifetime of the above writelock and/or poison `entry` such that + // `port::associate_lfts_upstack` fails. See that function for + // commentary on the guarantees provided by the port-wide lock. + + if entry.is_expired(now) { + let my_time = entry.last_hit(); + flow_expired_probe( + port_c, + name_c, + flowid, + Some(my_time.raw_millis()), + Some(now.raw_millis()), + ); + entry.expiry_cleanup(); + + return false; + } + + !entry.is_killed() + }); + } + + /// Remove all flows from `self` which are past their expiry time, + /// identifying the partner flow `extractor(&flow_state)` of each and + /// removing it from `partner`. + /// + /// Flows identified by `extractor` in `partner` *MUST* share the same + /// `FlowLifetime` as the entry removed from `self`. + pub fn expire_flows_partner( + &mut self, + partner: &mut FlowTable, + extractor: F, + now: Moment, + ) where F: Fn(&S) -> InnerFlowId, + T: FlowState, { let name_c = &self.name_c; let port_c = &self.port_c; - let mut expired = vec![]; self.map.retain(|flowid, entry| { // A flow cannot be expired by the timer while it still has children // relying upon its existence. Check whether any remain, and remove // dangling references to child entries which have expired. + // + // The dangling entries here will have been left by `expire_flows` + // called on other layers. { // We have a write lock on the port, so there shouldn't be // contention here. @@ -361,6 +420,8 @@ impl FlowTable { return true; } } + // The same lock commentary from `expire_flows` applies here. + if entry.is_expired(now) { let my_time = entry.last_hit(); flow_expired_probe( @@ -371,14 +432,25 @@ impl FlowTable { Some(now.raw_millis()), ); entry.expiry_cleanup(); - expired.push(f(entry.state())); + + // We don't need to call into `mark_evicted` here or when when + // removing the partner flow in this case because we know that the + // partner flow has the same lifetime, and so the same (empty) child + // set. + let partner_flow = extractor(entry.state()); + #[cfg(debug_assertions)] + { + if let Some(other) = partner.get(&partner_flow) { + assert!(Arc::ptr_eq(&entry.lifetime, &other.lifetime)) + } + } + partner.expire(&partner_flow, false); + return false; } !entry.is_killed() }); - - expired } /// Determine whether there is currently space for a new entry to be @@ -391,7 +463,7 @@ impl FlowTable { } if let Some((key, _)) = self.find_evictable_entry() { - self.expire(&key); + self.expire(&key, true); Ok(()) } else { Err(OpteError::MaxCapacity(self.limit.get() as u64)) @@ -820,7 +892,6 @@ mod test { use crate::api::PortInfo; use crate::engine::ip::v4::Protocol; use crate::engine::packet::AddrPair; - use crate::engine::packet::FLOW_ID_DEFAULT; use core::time::Duration; impl Dump for () { @@ -891,11 +962,9 @@ mod test { ft.add(flowid, ()).unwrap(); let now = Moment::now(); assert_eq!(ft.num_flows(), 1); - ft.expire_flows(now, |_| FLOW_ID_DEFAULT); + ft.expire_flows(now); assert_eq!(ft.num_flows(), 1); - ft.expire_flows(now + Duration::new(FLOW_DEF_EXPIRE_SECS, 0), |_| { - FLOW_ID_DEFAULT - }); + ft.expire_flows(now + Duration::new(FLOW_DEF_EXPIRE_SECS, 0)); assert_eq!(ft.num_flows(), 0); } @@ -943,15 +1012,15 @@ mod test { // A flow entry cannot be removed by the timer until all its children // have been evicted or expired. let t2 = now + Duration::new(FLOW_DEF_EXPIRE_SECS, 0); - ft1.expire_flows(t2, |_| FLOW_ID_DEFAULT); + ft1.expire_flows(t2); assert_eq!(ft1.num_flows(), 1); // If we go via ft2 first, we will be able to remove its entries (which // have no children), which in turn makes ft1's entries available for // eviction. - ft2.expire_flows(t2, |_| FLOW_ID_DEFAULT); + ft2.expire_flows(t2); assert_eq!(ft2.num_flows(), 0); - ft1.expire_flows(t2, |_| FLOW_ID_DEFAULT); + ft1.expire_flows(t2); assert_eq!(ft1.num_flows(), 0); } @@ -985,7 +1054,7 @@ mod test { // When fe2 is expired, it should pass on its expiry time to fe1. let t3 = t2 + Duration::new(FLOW_DEF_EXPIRE_SECS, 0); - ft2.expire_flows(t3, |_| FLOW_ID_DEFAULT); + ft2.expire_flows(t3); assert_eq!(ft2.num_flows(), 0); assert_eq!(fe1.last_hit(), t2); } diff --git a/lib/opte/src/engine/layer.rs b/lib/opte/src/engine/layer.rs index 4974d245..fc542bdc 100644 --- a/lib/opte/src/engine/layer.rs +++ b/lib/opte/src/engine/layer.rs @@ -243,11 +243,11 @@ impl LayerFlowTable { fn expire_flows(&mut self, now: Moment) { // Flow table in/out entries share a lifetime struct, so it's irrelevant // which of these tables we check first. - let to_expire = - self.ft_out.expire_flows(now, LftOutEntry::extract_pair); - for flow in to_expire { - self.ft_in.expire(&flow); - } + self.ft_out.expire_flows_partner( + &mut self.ft_in, + LftOutEntry::extract_pair, + now, + ); self.count = self.ft_out.num_flows(); } @@ -847,8 +847,13 @@ impl Layer { fn complete_eviction(&mut self, entry: SpaceCreated) { if let SpaceCreated::Evict { in_key, out_key } = entry { self.stats.vals.evictions.incr(1); - self.ft.ft_out.expire(&out_key); - self.ft.ft_in.expire(&in_key); + + // These two entries share the same `FlowLifetime`, so + // we can avoid wasting work by marking all children as + // `killed` only for the first entry. + self.ft.ft_out.expire(&out_key, true); + self.ft.ft_in.expire(&in_key, false); + self.ft.count = self.ft.ft_out.num_flows(); } } diff --git a/lib/opte/src/engine/port/mod.rs b/lib/opte/src/engine/port/mod.rs index ea8f2e04..a4fa6c56 100644 --- a/lib/opte/src/engine/port/mod.rs +++ b/lib/opte/src/engine/port/mod.rs @@ -1181,13 +1181,13 @@ impl Port { // A TCP state entry or UFT may in turn reference any number of LFT // hits, so we visit those first to maximise the likelihood that we can // clear up as many entries as possible. - let _ = data.tcp_flows.expire_flows(now, |_| FLOW_ID_DEFAULT); + data.tcp_flows.expire_flows(now); self.stats.vals.tcp_flows.set(u64::from(data.tcp_flows.num_flows())); - let _ = data.uft_in.expire_flows(now, |_| FLOW_ID_DEFAULT); + data.uft_in.expire_flows(now); self.stats.vals.in_uft_flows.set(u64::from(data.uft_in.num_flows())); - let _ = data.uft_out.expire_flows(now, |_| FLOW_ID_DEFAULT); + data.uft_out.expire_flows(now); self.stats.vals.out_uft_flows.set(u64::from(data.uft_out.num_flows())); for l in &mut data.layers {