Do not construct a Vec for evicted LFT flow IDs

lib/opte/src/engine/flow_table.rs

@@ -332,26 +332,85 @@ impl<S: FlowState> FlowTable<S> {
         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<F>(&mut self, now: Moment, f: F) -> Vec<InnerFlowId>
-    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<F, T>(
+        &mut self,
+        partner: &mut FlowTable<T>,
+        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<S: FlowState> FlowTable<S> {
                     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<S: FlowState> FlowTable<S> {
                     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<S: FlowState> FlowTable<S> {
         }
 
         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);
     }

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();
         }
     }

lib/opte/src/engine/port/mod.rs

@@ -1181,13 +1181,13 @@ impl<N: NetworkImpl> Port<N> {
         // 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 {