Skip to content

Commit 4ab9d49

Browse files
daniel-nolandclaude
andcommitted
perf(dataplane,routing,acl-filter,flow-filter): take shared tables once per burst
A worker ran about 34 locked instructions per packet, each a full barrier on x86. Many were stages taking shared, rarely changing state once per packet rather than once per burst: - egress and ingress entered the interface table, and egress the adjacency table, for every packet. Entering a left-right reader is an atomic RMW and a fence on the way in, and another RMW on the way out. - the FIB memo cached a reader per burst but entered it per packet. - the ACL filter loaded its tables, an `Arc` reference both ways, per packet. - the flow filter cloned the packet's flow into its summary. Egress, ingress and the ACL filter now take their tables once per burst. The FIB memo holds each reader it caches entered (`FibPin`), so entering it per packet takes left-right's nested, thread-local path. The flow filter copies what it needs from the flow instead of holding it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Daniel Noland <daniel@githedgehog.com>
1 parent 6c7ca8b commit 4ab9d49

8 files changed

Lines changed: 115 additions & 36 deletions

File tree

‎acl-filter/src/lib.rs‎

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ mod tests;
3434
pub use access::{
3535
AclFilterContext, AclFilterContextReader, AclFilterContextReaderFactory, AclFilterContextWriter,
3636
};
37+
use context::AclTables;
3738

3839
pub struct AclFilter {
3940
name: String,
@@ -51,7 +52,11 @@ impl AclFilter {
5152
}
5253
}
5354

54-
fn process_packet<Buf: PacketBufferMut>(&mut self, packet: &mut Packet<Buf>) {
55+
fn process_packet<Buf: PacketBufferMut>(
56+
&mut self,
57+
packet: &mut Packet<Buf>,
58+
tables: &AclTables,
59+
) {
5560
let nfi = &self.name;
5661
let genid = self.pipeline_data.genid();
5762

@@ -65,7 +70,7 @@ impl AclFilter {
6570
let valid_flow = self.packet_has_valid_flow(packet.meta(), genid);
6671
let has_valid_flow = valid_flow.is_some();
6772

68-
let (verdict, flow_scoped_allow) = self.lookup(&summary, valid_flow);
73+
let (verdict, flow_scoped_allow) = self.lookup(tables, &summary, valid_flow);
6974
if verdict == AclAction::Deny {
7075
debug!("{nfi}: Packet rejected by ACLs, dropping packet");
7176
packet.invalidate_flows();
@@ -116,12 +121,10 @@ impl AclFilter {
116121
// matching the packet itself.
117122
fn lookup(
118123
&self,
124+
tables: &AclTables,
119125
summary: &PacketSummary,
120126
flow_info: Option<&Arc<FlowInfo>>,
121127
) -> (AclAction, bool) {
122-
let guard = self.tablesr.load();
123-
let tables = &guard.acls;
124-
125128
// Every peering with an ACL has a default action registered for both directions, and only
126129
// such a peering has rules. Without one, nothing can match either way: allow the packet
127130
// without the lookups, and without reaching the reverse flow (a cold cache line, and an
@@ -171,9 +174,12 @@ impl AclFilter {
171174

172175
impl<Buf: PacketBufferMut> NetworkFunction<Buf> for AclFilter {
173176
fn process_burst(&mut self, burst: &mut Vec<Packet<Buf>>) {
177+
// Load the tables once per burst, not once per packet: a load takes a reference to them,
178+
// an atomic read-modify-write each way.
179+
let guard = self.tablesr.load();
174180
for packet in burst.iter_mut() {
175181
if !packet.is_done() && packet.meta().is_overlay() {
176-
self.process_packet(packet);
182+
self.process_packet(packet, &guard.acls);
177183
}
178184
}
179185
}

‎dataplane/src/packet_processor/egress.rs‎

Lines changed: 19 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ use net::interface::InterfaceIndex;
1515
use net::packet::{DoneReason, Packet};
1616
use pipeline::NetworkFunction;
1717

18-
use routing::{AtableReader, IfState, IfTable, IfTableReader, IfType, Interface};
18+
use routing::{AdjacencyTable, AtableReader, IfState, IfTable, IfTableReader, IfType, Interface};
1919

2020
use tracectl::trace_target;
2121
trace_target!("egress", LevelFilter::WARN, &["pipeline"]);
@@ -110,12 +110,13 @@ impl Egress {
110110
fn get_adj_mac<Buf: PacketBufferMut>(
111111
&self,
112112
packet: &mut Packet<Buf>,
113+
atable: Option<&AdjacencyTable>,
113114
addr: IpAddr,
114115
ifindex: InterfaceIndex,
115116
) -> Option<DestinationMac> {
116117
let nfi = &self.name;
117118

118-
if let Some(atable) = self.atabler.enter() {
119+
if let Some(atable) = atable {
119120
/* do lookup on the adjacency table */
120121
let Some(adj) = atable.get_adjacency(addr, ifindex) else {
121122
warn!("{nfi}: missing L2 info for {addr}");
@@ -144,6 +145,7 @@ impl Egress {
144145
&self,
145146
ifindex: InterfaceIndex,
146147
packet: &mut Packet<Buf>,
148+
atable: Option<&AdjacencyTable>,
147149
) -> Option<DestinationMac> {
148150
let nfi = &self.name;
149151
// if packet was annotated with a next-hop address, try to resolve it using the
@@ -152,9 +154,9 @@ impl Egress {
152154
// resolve it with the adjacency table as well. If that fails, that's where the
153155
// ARP/ND would need to be triggered.
154156
if let Some(nh_addr) = packet.meta().nh_addr {
155-
self.get_adj_mac(packet, nh_addr, ifindex)
157+
self.get_adj_mac(packet, atable, nh_addr, ifindex)
156158
} else if let Some(destination) = packet.ip_destination() {
157-
self.get_adj_mac(packet, destination, ifindex)
159+
self.get_adj_mac(packet, atable, destination, ifindex)
158160
} else {
159161
warn!("{nfi}: could not determine packet destination IP address");
160162
// This should be a bug because if a packet was not IP, we'd have
@@ -165,15 +167,20 @@ impl Egress {
165167
}
166168

167169
#[inline]
168-
fn egress_process<Buf: PacketBufferMut>(&self, packet: &mut Packet<Buf>, iftable: &IfTable) {
170+
fn egress_process<Buf: PacketBufferMut>(
171+
&self,
172+
packet: &mut Packet<Buf>,
173+
iftable: &IfTable,
174+
atable: Option<&AdjacencyTable>,
175+
) {
169176
let Some(oif) = packet.meta().oif else {
170177
warn!("{}: Missing oif metadata!", &self.name);
171178
packet.done(DoneReason::RouteFailure);
172179
return;
173180
};
174181

175182
/* resolve destination mac */
176-
let Some(dst_mac) = self.resolve_next_mac(oif, packet) else {
183+
let Some(dst_mac) = self.resolve_next_mac(oif, packet, atable) else {
177184
// we could not figure out the destination MAC.
178185
// resolve_next_mac() already calls packet.done()
179186
return;
@@ -191,10 +198,14 @@ impl Egress {
191198

192199
impl<Buf: PacketBufferMut> NetworkFunction<Buf> for Egress {
193200
fn process_burst(&mut self, burst: &mut Vec<Packet<Buf>>) {
201+
// Enter the tables once per burst, not once per packet: each entry and exit is an atomic
202+
// read-modify-write, a full barrier on x86. A writer waits for the burst at most.
203+
let iftable = self.iftr.enter();
204+
let atable = self.atabler.enter();
194205
for packet in burst.iter_mut() {
195206
if !packet.is_done() {
196-
if let Some(iftable) = self.iftr.enter() {
197-
self.egress_process(packet, &iftable);
207+
if let Some(iftable) = iftable.as_deref() {
208+
self.egress_process(packet, iftable, atable.as_deref());
198209
} else {
199210
warn!("{}: Fib iftable no longer readable!", &self.name);
200211
packet.done(DoneReason::InternalFailure);

‎dataplane/src/packet_processor/ingress.rs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -222,10 +222,12 @@ impl Ingress {
222222
impl<Buf: PacketBufferMut> NetworkFunction<Buf> for Ingress {
223223
#[tracing::instrument(level = "trace", skip(self, burst))]
224224
fn process_burst(&mut self, burst: &mut Vec<Packet<Buf>>) {
225+
// Enter the interface table once per burst, not once per packet (see Egress).
226+
let iftable = self.iftr.enter();
225227
for packet in burst.iter_mut() {
226228
let nfi = self.name();
227229
if !packet.is_done() {
228-
if let Some(iftable) = self.iftr.enter() {
230+
if let Some(iftable) = iftable.as_deref() {
229231
match packet.meta().iif {
230232
None => {
231233
warn!("no iif set in packet metadata (driver bug)");

‎dataplane/src/packet_processor/ipforward.rs‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,8 @@ use std::rc::Rc;
2525
use tracing::{debug, error, warn};
2626

2727
use routing::{
28-
EgressObject, Encapsulation, FibEntry, FibKey, FibReader, FibTableReader, PktInstruction, Vtep,
29-
VxlanEncapsulation,
28+
EgressObject, Encapsulation, FibEntry, FibKey, FibPin, FibReader, FibTableReader,
29+
PktInstruction, Vtep, VxlanEncapsulation,
3030
};
3131

3232
use tracectl::{custom_target, tdebug, trace_target};
@@ -45,10 +45,11 @@ pub struct IpForwarder {
4545
/// Cache up to two FIB readers per burst to avoid repeated table lookups.
4646
///
4747
/// Two slots cover the routing FIB and the VNI FIB used after VXLAN decapsulation.
48-
/// FIB replacements are picked up on eviction or at the next burst.
48+
/// FIB replacements are picked up on eviction or at the next burst. Each cached reader is held
49+
/// entered (pinned) while it is cached, so that entering it for each packet is cheap.
4950
#[derive(Default)]
5051
struct FibMemo {
51-
slots: [Option<(FibKey, Rc<FibReader>)>; 2],
52+
slots: [Option<(FibKey, FibPin)>; 2],
5253
}
5354

5455
impl FibMemo {
@@ -59,9 +60,11 @@ impl FibMemo {
5960
} else if self.slots[0].as_ref().is_none_or(|(k, _)| *k != key) {
6061
let reader = fibtr.get_fib_reader(key).ok()?;
6162
self.slots[1] = self.slots[0].take();
62-
self.slots[0] = Some((key, reader));
63+
self.slots[0] = Some((key, FibPin::new(reader)));
6364
}
64-
self.slots[0].as_ref().map(|(_, reader)| Rc::clone(reader))
65+
self.slots[0]
66+
.as_ref()
67+
.map(|(_, pin)| Rc::clone(pin.reader()))
6568
}
6669
}
6770

‎flow-filter/src/lib.rs‎

Lines changed: 29 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ use concurrency::sync::Arc;
77
use config::external::overlay::vpcpeering::{ValidatedExpose, VpcExposeNatConfig};
88
use net::FlowKey;
99
use net::buffer::PacketBufferMut;
10-
use net::flows::{FlowInfo, FlowStatus};
10+
use net::flows::{FlowInfo, FlowInfoFlags, FlowStatus};
1111
use net::headers::{TryHeaders, TryIp, UpperLayerProto};
1212
use net::ip::NextHeader;
1313
use net::packet::{DoneReason, Packet, PacketMeta, VpcDiscriminant};
@@ -260,10 +260,10 @@ impl FlowFilter {
260260
if flow_summary.needs_tracking {
261261
meta.set_forced_flow_tracking(true);
262262
}
263-
if flow_summary.flow_info.get_flags().requires_static_nat_src() {
263+
if flow_summary.flags.requires_static_nat_src() {
264264
meta.set_static_nat_src(true);
265265
}
266-
if flow_summary.flow_info.get_flags().requires_static_nat_dst() {
266+
if flow_summary.flags.requires_static_nat_dst() {
267267
meta.set_static_nat_dst(true);
268268
}
269269
}
@@ -306,7 +306,9 @@ impl FlowFilter {
306306
if flow_summary.genid == genid {
307307
return false;
308308
}
309-
let (nfi, flowkey) = (&self.name, flow_summary.flow_info.flowkey());
309+
let nfi = &self.name;
310+
let flowkey = meta.flow_info.as_deref().map(FlowInfo::flowkey);
311+
let flowkey = Displayed(flowkey);
310312
if flow_summary.dst_vpcd != new_dst_vpcd {
311313
debug!("{nfi}: Outdated flow {flowkey} (new dst: {new_dst_vpcd}) will be invalidated.");
312314
return true;
@@ -321,7 +323,7 @@ impl FlowFilter {
321323
}
322324
// The reverse key of a flow depends on static NAT, whether or not the flow also uses
323325
// masquerade or port forwarding, and these NFs do not check it.
324-
let flags = flow_summary.flow_info.get_flags();
326+
let flags = flow_summary.flags;
325327
if meta.requires_static_nat_src() != flags.requires_static_nat_src()
326328
|| meta.requires_static_nat_dst() != flags.requires_static_nat_dst()
327329
{
@@ -351,9 +353,7 @@ impl FlowFilter {
351353
) -> (Option<VpcDiscriminant>, SourceGate) {
352354
// The only case when we need re-validation is when we have an active flow with an outdated
353355
// genid. If this is not the case, return None.
354-
if flow_summary.flow_info.status() != FlowStatus::Active
355-
|| flow_summary.flow_info.genid() >= genid
356-
{
356+
if flow_summary.status != FlowStatus::Active || flow_summary.genid >= genid {
357357
return (None, SourceGate::Ungated);
358358
}
359359
// We only need to pass this information when re-validating a flow in the return direction.
@@ -381,11 +381,11 @@ impl FlowFilter {
381381
genid: i64,
382382
) -> Option<VpcDiscriminant> {
383383
let nfi = &self.name;
384-
if flow_summary.flow_info.status() != FlowStatus::Active {
384+
if flow_summary.status != FlowStatus::Active {
385385
debug!("{nfi}: Packet has inactive flow information");
386386
return None;
387387
}
388-
let flow_genid = flow_summary.flow_info.genid();
388+
let flow_genid = flow_summary.genid;
389389
if flow_genid < genid {
390390
debug!(
391391
"{nfi}: Packet has outdated flow information from a prior configuration ({flow_genid} < {genid})"
@@ -410,14 +410,17 @@ impl<Buf: PacketBufferMut> NetworkFunction<Buf> for FlowFilter {
410410
}
411411
}
412412

413+
/// What the filter needs to know about the flow a packet refers to: copied out, so that it does not
414+
/// hold the flow (an atomic increment and decrement on the flow's line, per packet).
413415
#[derive(Debug, Clone)]
414416
struct FlowSummary {
415417
genid: i64,
418+
status: FlowStatus,
419+
flags: FlowInfoFlags,
416420
dst_vpcd: VpcDiscriminant,
417421
needs_masquerade: bool,
418422
needs_port_forwarding: bool,
419423
needs_tracking: bool,
420-
flow_info: Arc<FlowInfo>,
421424
}
422425

423426
impl FlowSummary {
@@ -431,17 +434,30 @@ impl FlowSummary {
431434
};
432435
Some(Self {
433436
genid: flow_info.genid(),
437+
status: flow_info.status(),
438+
flags: flow_info.get_flags(),
434439
dst_vpcd,
435440
needs_masquerade: locked_info.masquerade_data.is_some(),
436441
needs_port_forwarding: locked_info.port_fw_data.is_some(),
437442
needs_tracking: locked_info.tracked_data.is_some()
438443
|| locked_info.static_nat_data.is_some(),
439-
flow_info: flow_info.clone(),
440444
})
441445
}
442446

443447
fn is_initiator(&self) -> bool {
444-
self.flow_info.get_flags().is_initiator()
448+
self.flags.is_initiator()
449+
}
450+
}
451+
452+
/// An optional flow key, for log messages.
453+
struct Displayed<'a>(Option<&'a FlowKey>);
454+
455+
impl std::fmt::Display for Displayed<'_> {
456+
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
457+
match self.0 {
458+
Some(key) => write!(f, "{key}"),
459+
None => f.write_str("(unknown)"),
460+
}
445461
}
446462
}
447463

‎flow-filter/src/tests.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1344,7 +1344,8 @@ fn invalidation_decision_matches_spec() {
13441344
needs_masquerade: case.flow_masquerade,
13451345
needs_port_forwarding: case.flow_port_forwarding,
13461346
needs_tracking: case.flow_tracking,
1347-
flow_info,
1347+
status: flow_info.status(),
1348+
flags: flow_info.get_flags(),
13481349
});
13491350

13501351
assert_eq!(

‎routing/src/fib/fibtype.rs‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -512,6 +512,45 @@ impl AsRef<FibReader> for ReadHandle<Fib> {
512512
unsafe { &*(std::ptr::from_ref::<ReadHandle<Fib>>(self).cast::<FibReader>()) }
513513
}
514514
}
515+
/// A [`FibReader`] held entered, for a burst of packets.
516+
///
517+
/// Entering a reader takes two atomic read-modify-writes and a full fence (one of each on the way
518+
/// in, one RMW on the way out). While a pin lives, entering the same reader again takes a
519+
/// thread-local counter instead. A pin delays the FIB's writer by as long as it is held, so hold
520+
/// one for a burst, not longer.
521+
pub struct FibPin {
522+
// Declared first, so dropped first: it borrows from `reader`.
523+
_guard: Option<ReadGuard<'static, Fib>>,
524+
reader: Rc<FibReader>,
525+
}
526+
527+
impl FibPin {
528+
/// Pin `reader` (entered, or found invalid: then the pin holds nothing).
529+
#[must_use]
530+
pub fn new(reader: Rc<FibReader>) -> Self {
531+
let guard = reader.0.enter();
532+
// SAFETY: the guard borrows the `ReadHandle` inside the `Rc`'s allocation, which does not
533+
// move and which this pin keeps alive: the pin owns the `Rc`, and drops the guard first
534+
// (field order). The guard is never handed out, so nothing can outlive the pin with it.
535+
#[allow(unsafe_code)]
536+
let guard = unsafe {
537+
std::mem::transmute::<Option<ReadGuard<'_, Fib>>, Option<ReadGuard<'static, Fib>>>(
538+
guard,
539+
)
540+
};
541+
Self {
542+
_guard: guard,
543+
reader,
544+
}
545+
}
546+
547+
/// The pinned reader.
548+
#[must_use]
549+
pub fn reader(&self) -> &Rc<FibReader> {
550+
&self.reader
551+
}
552+
}
553+
515554
#[derive(Debug, Clone)]
516555
#[repr(transparent)]
517556
pub struct FibReaderFactory(pub(crate) ReadHandleFactory<Fib>);

‎routing/src/lib.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,14 +27,15 @@ mod router;
2727
mod routingdb;
2828

2929
// re-exports
30+
pub use atable::adjacency::AdjacencyTable;
3031
pub use atable::atablerw::{AtableReader, AtableReaderFactory};
3132
pub use config::RouterConfig;
3233
pub use errors::RouterError;
3334
pub use evpn::Vtep;
3435
pub use fib::fibgroupstore::FibRoute;
3536
pub use fib::fibobjects::{EgressObject, FibEntry, FibGroup, PktInstruction};
3637
pub use fib::fibtable::{FibTableReader, FibTableReaderFactory};
37-
pub use fib::fibtype::{Fib, FibKey, FibReader};
38+
pub use fib::fibtype::{Fib, FibKey, FibPin, FibReader};
3839
pub use frr::frrmi::FrrAppliedConfig;
3940
pub use frr::renderer::builder::Render;
4041
pub use interfaces::iftable::IfTable;

0 commit comments

Comments
 (0)