release (optional): take shared tables once per burst

acl-filter/src/lib.rs

@@ -34,6 +34,7 @@ mod tests;
 pub use access::{
     AclFilterContext, AclFilterContextReader, AclFilterContextReaderFactory, AclFilterContextWriter,
 };
+use context::AclTables;
 
 pub struct AclFilter {
     name: String,
@@ -51,7 +52,11 @@ impl AclFilter {
         }
     }
 
-    fn process_packet<Buf: PacketBufferMut>(&mut self, packet: &mut Packet<Buf>) {
+    fn process_packet<Buf: PacketBufferMut>(
+        &mut self,
+        packet: &mut Packet<Buf>,
+        tables: &AclTables,
+    ) {
         let nfi = &self.name;
         let genid = self.pipeline_data.genid();
 
@@ -65,7 +70,7 @@ impl AclFilter {
         let valid_flow = self.packet_has_valid_flow(packet.meta(), genid);
         let has_valid_flow = valid_flow.is_some();
 
-        let (verdict, flow_scoped_allow) = self.lookup(&summary, valid_flow);
+        let (verdict, flow_scoped_allow) = self.lookup(tables, &summary, valid_flow);
         if verdict == AclAction::Deny {
             debug!("{nfi}: Packet rejected by ACLs, dropping packet");
             packet.invalidate_flows();
@@ -116,12 +121,10 @@ impl AclFilter {
     // matching the packet itself.
     fn lookup(
         &self,
+        tables: &AclTables,
         summary: &PacketSummary,
         flow_info: Option<&Arc<FlowInfo>>,
     ) -> (AclAction, bool) {
-        let guard = self.tablesr.load();
-        let tables = &guard.acls;
-
         // Every peering with an ACL has a default action registered for both directions, and only
         // such a peering has rules. Without one, nothing can match either way: allow the packet
         // without the lookups, and without reaching the reverse flow (a cold cache line, and an
@@ -171,9 +174,12 @@ impl AclFilter {
 
 impl<Buf: PacketBufferMut> NetworkFunction<Buf> for AclFilter {
     fn process_burst(&mut self, burst: &mut Vec<Packet<Buf>>) {
+        // Load the tables once per burst, not once per packet: a load takes a reference to them,
+        // an atomic read-modify-write each way.
+        let guard = self.tablesr.load();
         for packet in burst.iter_mut() {
             if !packet.is_done() && packet.meta().is_overlay() {
-                self.process_packet(packet);
+                self.process_packet(packet, &guard.acls);
             }
         }
     }

dataplane/src/packet_processor/egress.rs

@@ -15,7 +15,7 @@ use net::interface::InterfaceIndex;
 use net::packet::{DoneReason, Packet};
 use pipeline::NetworkFunction;
 
-use routing::{AtableReader, IfState, IfTable, IfTableReader, IfType, Interface};
+use routing::{AdjacencyTable, AtableReader, IfState, IfTable, IfTableReader, IfType, Interface};
 
 use tracectl::trace_target;
 trace_target!("egress", LevelFilter::WARN, &["pipeline"]);
@@ -110,12 +110,13 @@ impl Egress {
     fn get_adj_mac<Buf: PacketBufferMut>(
         &self,
         packet: &mut Packet<Buf>,
+        atable: Option<&AdjacencyTable>,
         addr: IpAddr,
         ifindex: InterfaceIndex,
     ) -> Option<DestinationMac> {
         let nfi = &self.name;
 
-        if let Some(atable) = self.atabler.enter() {
+        if let Some(atable) = atable {
             /* do lookup on the adjacency table */
             let Some(adj) = atable.get_adjacency(addr, ifindex) else {
                 warn!("{nfi}: missing L2 info for {addr}");
@@ -144,6 +145,7 @@ impl Egress {
         &self,
         ifindex: InterfaceIndex,
         packet: &mut Packet<Buf>,
+        atable: Option<&AdjacencyTable>,
     ) -> Option<DestinationMac> {
         let nfi = &self.name;
         // if packet was annotated with a next-hop address, try to resolve it using the
@@ -152,9 +154,9 @@ impl Egress {
         // resolve it with the adjacency table as well. If that fails, that's where the
         // ARP/ND would need to be triggered.
         if let Some(nh_addr) = packet.meta().nh_addr {
-            self.get_adj_mac(packet, nh_addr, ifindex)
+            self.get_adj_mac(packet, atable, nh_addr, ifindex)
         } else if let Some(destination) = packet.ip_destination() {
-            self.get_adj_mac(packet, destination, ifindex)
+            self.get_adj_mac(packet, atable, destination, ifindex)
         } else {
             warn!("{nfi}: could not determine packet destination IP address");
             // This should be a bug because if a packet was not IP, we'd have
@@ -165,15 +167,20 @@ impl Egress {
     }
 
     #[inline]
-    fn egress_process<Buf: PacketBufferMut>(&self, packet: &mut Packet<Buf>, iftable: &IfTable) {
+    fn egress_process<Buf: PacketBufferMut>(
+        &self,
+        packet: &mut Packet<Buf>,
+        iftable: &IfTable,
+        atable: Option<&AdjacencyTable>,
+    ) {
         let Some(oif) = packet.meta().oif else {
             warn!("{}: Missing oif metadata!", &self.name);
             packet.done(DoneReason::RouteFailure);
             return;
         };
 
         /* resolve destination mac */
-        let Some(dst_mac) = self.resolve_next_mac(oif, packet) else {
+        let Some(dst_mac) = self.resolve_next_mac(oif, packet, atable) else {
             // we could not figure out the destination MAC.
             // resolve_next_mac() already calls packet.done()
             return;
@@ -191,10 +198,14 @@ impl Egress {
 
 impl<Buf: PacketBufferMut> NetworkFunction<Buf> for Egress {
     fn process_burst(&mut self, burst: &mut Vec<Packet<Buf>>) {
+        // Enter the tables once per burst, not once per packet: each entry and exit is an atomic
+        // read-modify-write, a full barrier on x86. A writer waits for the burst at most.
+        let iftable = self.iftr.enter();
+        let atable = self.atabler.enter();
         for packet in burst.iter_mut() {
             if !packet.is_done() {
-                if let Some(iftable) = self.iftr.enter() {
-                    self.egress_process(packet, &iftable);
+                if let Some(iftable) = iftable.as_deref() {
+                    self.egress_process(packet, iftable, atable.as_deref());
                 } else {
                     warn!("{}: Fib iftable no longer readable!", &self.name);
                     packet.done(DoneReason::InternalFailure);

dataplane/src/packet_processor/ingress.rs

@@ -222,10 +222,12 @@ impl Ingress {
 impl<Buf: PacketBufferMut> NetworkFunction<Buf> for Ingress {
     #[tracing::instrument(level = "trace", skip(self, burst))]
     fn process_burst(&mut self, burst: &mut Vec<Packet<Buf>>) {
+        // Enter the interface table once per burst, not once per packet (see Egress).
+        let iftable = self.iftr.enter();
         for packet in burst.iter_mut() {
             let nfi = self.name();
             if !packet.is_done() {
-                if let Some(iftable) = self.iftr.enter() {
+                if let Some(iftable) = iftable.as_deref() {
                     match packet.meta().iif {
                         None => {
                             warn!("no iif set in packet metadata (driver bug)");

dataplane/src/packet_processor/ipforward.rs

@@ -25,8 +25,8 @@ use std::rc::Rc;
 use tracing::{debug, error, warn};
 
 use routing::{
-    EgressObject, Encapsulation, FibEntry, FibKey, FibReader, FibTableReader, PktInstruction, Vtep,
-    VxlanEncapsulation,
+    EgressObject, Encapsulation, FibEntry, FibKey, FibPin, FibReader, FibTableReader,
+    PktInstruction, Vtep, VxlanEncapsulation,
 };
 
 use tracectl::{custom_target, tdebug, trace_target};
@@ -45,10 +45,11 @@ pub struct IpForwarder {
 /// Cache up to two FIB readers per burst to avoid repeated table lookups.
 ///
 /// Two slots cover the routing FIB and the VNI FIB used after VXLAN decapsulation.
-/// FIB replacements are picked up on eviction or at the next burst.
+/// FIB replacements are picked up on eviction or at the next burst. Each cached reader is held
+/// entered (pinned) while it is cached, so that entering it for each packet is cheap.
 #[derive(Default)]
 struct FibMemo {
-    slots: [Option<(FibKey, Rc<FibReader>)>; 2],
+    slots: [Option<(FibKey, FibPin)>; 2],
 }
 
 impl FibMemo {
@@ -59,9 +60,11 @@ impl FibMemo {
         } else if self.slots[0].as_ref().is_none_or(|(k, _)| *k != key) {
             let reader = fibtr.get_fib_reader(key).ok()?;
             self.slots[1] = self.slots[0].take();
-            self.slots[0] = Some((key, reader));
+            self.slots[0] = Some((key, FibPin::new(reader)));
         }
-        self.slots[0].as_ref().map(|(_, reader)| Rc::clone(reader))
+        self.slots[0]
+            .as_ref()
+            .map(|(_, pin)| Rc::clone(pin.reader()))
     }
 }
 

flow-filter/src/lib.rs

@@ -7,7 +7,7 @@ use concurrency::sync::Arc;
 use config::external::overlay::vpcpeering::{ValidatedExpose, VpcExposeNatConfig};
 use net::FlowKey;
 use net::buffer::PacketBufferMut;
-use net::flows::{FlowInfo, FlowStatus};
+use net::flows::{FlowInfo, FlowInfoFlags, FlowStatus};
 use net::headers::{TryHeaders, TryIp, UpperLayerProto};
 use net::ip::NextHeader;
 use net::packet::{DoneReason, Packet, PacketMeta, VpcDiscriminant};
@@ -260,10 +260,10 @@ impl FlowFilter {
         if flow_summary.needs_tracking {
             meta.set_forced_flow_tracking(true);
         }
-        if flow_summary.flow_info.get_flags().requires_static_nat_src() {
+        if flow_summary.flags.requires_static_nat_src() {
             meta.set_static_nat_src(true);
         }
-        if flow_summary.flow_info.get_flags().requires_static_nat_dst() {
+        if flow_summary.flags.requires_static_nat_dst() {
             meta.set_static_nat_dst(true);
         }
     }
@@ -306,7 +306,9 @@ impl FlowFilter {
         if flow_summary.genid == genid {
             return false;
         }
-        let (nfi, flowkey) = (&self.name, flow_summary.flow_info.flowkey());
+        let nfi = &self.name;
+        let flowkey = meta.flow_info.as_deref().map(FlowInfo::flowkey);
+        let flowkey = Displayed(flowkey);
         if flow_summary.dst_vpcd != new_dst_vpcd {
             debug!("{nfi}: Outdated flow {flowkey} (new dst: {new_dst_vpcd}) will be invalidated.");
             return true;
@@ -321,7 +323,7 @@ impl FlowFilter {
         }
         // The reverse key of a flow depends on static NAT, whether or not the flow also uses
         // masquerade or port forwarding, and these NFs do not check it.
-        let flags = flow_summary.flow_info.get_flags();
+        let flags = flow_summary.flags;
         if meta.requires_static_nat_src() != flags.requires_static_nat_src()
             || meta.requires_static_nat_dst() != flags.requires_static_nat_dst()
         {
@@ -351,9 +353,7 @@ impl FlowFilter {
     ) -> (Option<VpcDiscriminant>, SourceGate) {
         // The only case when we need re-validation is when we have an active flow with an outdated
         // genid. If this is not the case, return None.
-        if flow_summary.flow_info.status() != FlowStatus::Active
-            || flow_summary.flow_info.genid() >= genid
-        {
+        if flow_summary.status != FlowStatus::Active || flow_summary.genid >= genid {
             return (None, SourceGate::Ungated);
         }
         // We only need to pass this information when re-validating a flow in the return direction.
@@ -381,11 +381,11 @@ impl FlowFilter {
         genid: i64,
     ) -> Option<VpcDiscriminant> {
         let nfi = &self.name;
-        if flow_summary.flow_info.status() != FlowStatus::Active {
+        if flow_summary.status != FlowStatus::Active {
             debug!("{nfi}: Packet has inactive flow information");
             return None;
         }
-        let flow_genid = flow_summary.flow_info.genid();
+        let flow_genid = flow_summary.genid;
         if flow_genid < genid {
             debug!(
                 "{nfi}: Packet has outdated flow information from a prior configuration ({flow_genid} < {genid})"
@@ -410,14 +410,17 @@ impl<Buf: PacketBufferMut> NetworkFunction<Buf> for FlowFilter {
     }
 }
 
+/// What the filter needs to know about the flow a packet refers to: copied out, so that it does not
+/// hold the flow (an atomic increment and decrement on the flow's line, per packet).
 #[derive(Debug, Clone)]
 struct FlowSummary {
     genid: i64,
+    status: FlowStatus,
+    flags: FlowInfoFlags,
     dst_vpcd: VpcDiscriminant,
     needs_masquerade: bool,
     needs_port_forwarding: bool,
     needs_tracking: bool,
-    flow_info: Arc<FlowInfo>,
 }
 
 impl FlowSummary {
@@ -431,17 +434,30 @@ impl FlowSummary {
         };
         Some(Self {
             genid: flow_info.genid(),
+            status: flow_info.status(),
+            flags: flow_info.get_flags(),
             dst_vpcd,
             needs_masquerade: locked_info.masquerade_data.is_some(),
             needs_port_forwarding: locked_info.port_fw_data.is_some(),
             needs_tracking: locked_info.tracked_data.is_some()
                 || locked_info.static_nat_data.is_some(),
-            flow_info: flow_info.clone(),
         })
     }
 
     fn is_initiator(&self) -> bool {
-        self.flow_info.get_flags().is_initiator()
+        self.flags.is_initiator()
+    }
+}
+
+/// An optional flow key, for log messages.
+struct Displayed<'a>(Option<&'a FlowKey>);
+
+impl std::fmt::Display for Displayed<'_> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self.0 {
+            Some(key) => write!(f, "{key}"),
+            None => f.write_str("(unknown)"),
+        }
     }
 }
 

flow-filter/src/tests.rs

@@ -1344,7 +1344,8 @@ fn invalidation_decision_matches_spec() {
                 needs_masquerade: case.flow_masquerade,
                 needs_port_forwarding: case.flow_port_forwarding,
                 needs_tracking: case.flow_tracking,
-                flow_info,
+                status: flow_info.status(),
+                flags: flow_info.get_flags(),
             });
 
             assert_eq!(

routing/src/fib/fibtype.rs

@@ -512,6 +512,45 @@ impl AsRef<FibReader> for ReadHandle<Fib> {
         unsafe { &*(std::ptr::from_ref::<ReadHandle<Fib>>(self).cast::<FibReader>()) }
     }
 }
+/// A [`FibReader`] held entered, for a burst of packets.
+///
+/// Entering a reader takes two atomic read-modify-writes and a full fence (one of each on the way
+/// in, one RMW on the way out). While a pin lives, entering the same reader again takes a
+/// thread-local counter instead. A pin delays the FIB's writer by as long as it is held, so hold
+/// one for a burst, not longer.
+pub struct FibPin {
+    // Declared first, so dropped first: it borrows from `reader`.
+    _guard: Option<ReadGuard<'static, Fib>>,
+    reader: Rc<FibReader>,
+}
+
+impl FibPin {
+    /// Pin `reader` (entered, or found invalid: then the pin holds nothing).
+    #[must_use]
+    pub fn new(reader: Rc<FibReader>) -> Self {
+        let guard = reader.0.enter();
+        // SAFETY: the guard borrows the `ReadHandle` inside the `Rc`'s allocation, which does not
+        // move and which this pin keeps alive: the pin owns the `Rc`, and drops the guard first
+        // (field order). The guard is never handed out, so nothing can outlive the pin with it.
+        #[allow(unsafe_code)]
+        let guard = unsafe {
+            std::mem::transmute::<Option<ReadGuard<'_, Fib>>, Option<ReadGuard<'static, Fib>>>(
+                guard,
+            )
+        };
+        Self {
+            _guard: guard,
+            reader,
+        }
+    }
+
+    /// The pinned reader.
+    #[must_use]
+    pub fn reader(&self) -> &Rc<FibReader> {
+        &self.reader
+    }
+}
+
 #[derive(Debug, Clone)]
 #[repr(transparent)]
 pub struct FibReaderFactory(pub(crate) ReadHandleFactory<Fib>);

routing/src/lib.rs

@@ -27,14 +27,15 @@ mod router;
 mod routingdb;
 
 // re-exports
+pub use atable::adjacency::AdjacencyTable;
 pub use atable::atablerw::{AtableReader, AtableReaderFactory};
 pub use config::RouterConfig;
 pub use errors::RouterError;
 pub use evpn::Vtep;
 pub use fib::fibgroupstore::FibRoute;
 pub use fib::fibobjects::{EgressObject, FibEntry, FibGroup, PktInstruction};
 pub use fib::fibtable::{FibTableReader, FibTableReaderFactory};
-pub use fib::fibtype::{Fib, FibKey, FibReader};
+pub use fib::fibtype::{Fib, FibKey, FibPin, FibReader};
 pub use frr::frrmi::FrrAppliedConfig;
 pub use frr::renderer::builder::Render;
 pub use interfaces::iftable::IfTable;