diff --git a/.github/buildomat/jobs/bench.sh b/.github/buildomat/jobs/bench.sh index 019e6a39..3a6ef6ef 100644 --- a/.github/buildomat/jobs/bench.sh +++ b/.github/buildomat/jobs/bench.sh @@ -130,6 +130,7 @@ pfexec cp /input/xde/work/release/xde /kernel/drv/amd64 pfexec add_drv xde banner "bench" +CI=1 cargo kbench local cargo ubench diff --git a/bench/benches/userland.rs b/bench/benches/userland.rs index e38ee92f..965ed510 100644 --- a/bench/benches/userland.rs +++ b/bench/benches/userland.rs @@ -2,7 +2,7 @@ // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at https://mozilla.org/MPL/2.0/. -// Copyright 2024 Oxide Computer Company +// Copyright 2026 Oxide Computer Company //! Userland packet parsing and processing microbenchmarks. @@ -21,6 +21,7 @@ use opte_bench::packet::Dhcp6; use opte_bench::packet::Icmp4; use opte_bench::packet::Icmp6; use opte_bench::packet::ParserKind; +use opte_bench::packet::SlowpathEvict; use opte_bench::packet::TestCase; use opte_bench::packet::ULP_FAST_PATH; use opte_bench::packet::ULP_SLOW_PATH; @@ -44,11 +45,18 @@ pub fn block(c: &mut Criterion, do_parse: bool) { Box::new(Icmp6), Box::new(ULP_FAST_PATH), Box::new(ULP_SLOW_PATH), + Box::new(SlowpathEvict { + capacities: [1 << 10, 1 << 15, 1 << 19, 1 << 20] + .into_iter() + .filter_map(NonZeroU32::new) + .collect(), + p_expires: vec![0.0], + }), ]; for experiment in all_tests { for case in experiment.test_cases() { - if do_parse { + if experiment.do_parse_benchmark() && do_parse { test_parse(c, &**experiment, &*case); } test_handle(c, &**experiment, &*case); @@ -151,6 +159,7 @@ pub fn test_handle( )); let parser = case.parse_with(); + let can_fail = experiment.allow_failure(); c.bench_with_input( BenchmarkId::from_parameter(case.instance_name()), &case, @@ -174,7 +183,7 @@ pub fn test_handle( GenericUlp {}, ) .unwrap(); - port.port.process(dir, black_box(pkt)).unwrap() + port.port.process(dir, black_box(pkt)) } Out => { let pkt = Packet::parse_outbound( @@ -182,11 +191,15 @@ pub fn test_handle( GenericUlp {}, ) .unwrap(); - port.port.process(dir, black_box(pkt)).unwrap() + port.port.process(dir, black_box(pkt)) } }; - assert!(!matches!(res, ProcessResult::Drop { .. })); - if let Modified(spec) = res { + + if !can_fail { + assert!(res.is_ok()); + } + + if let Ok(Modified(spec)) = res { black_box(spec.apply(pkt_m)); } } @@ -198,7 +211,7 @@ pub fn test_handle( VpcParser {}, ) .unwrap(); - port.port.process(dir, black_box(pkt)).unwrap() + port.port.process(dir, black_box(pkt)) } Out => { let pkt = Packet::parse_outbound( @@ -206,11 +219,15 @@ pub fn test_handle( VpcParser {}, ) .unwrap(); - port.port.process(dir, black_box(pkt)).unwrap() + port.port.process(dir, black_box(pkt)) } }; - assert!(!matches!(res, ProcessResult::Drop { .. })); - if let Modified(spec) = res { + + if !can_fail { + assert!(res.is_ok()); + } + + if let Ok(Modified(spec)) = res { black_box(spec.apply(pkt_m)); } } @@ -325,7 +342,44 @@ fn source_filter_allows(c: &mut Criterion) { group.finish(); } -criterion_group!(wall, parse_and_process, source_filter_allows); +fn periodic_cleanup(c: &mut Criterion) { + let (capacities, p_expires) = if std::env::var("CI").is_ok() { + (&[1 << 10, 1 << 15][..], vec![0.0, 0.25]) + } else { + (&[1 << 10, 1 << 15, 1 << 19, 1 << 20][..], vec![0.0, 0.1, 0.25, 0.5]) + }; + let expt = SlowpathEvict { + capacities: capacities + .iter() + .copied() + .filter_map(NonZeroU32::new) + .collect(), + p_expires, + }; + for case in expt.test_cases() { + let port = case.create_port().unwrap(); + let mut c = c.benchmark_group(format!("cleanup/{}", M::label())); + + c.bench_with_input( + BenchmarkId::from_parameter(case.instance_name()), + &case, + |b, _i| { + b.iter_batched( + || case.pre_handle(&port), + |_| black_box(port.port.expire_flows()), + criterion::BatchSize::LargeInput, + ) + }, + ); + } +} + +criterion_group!( + wall, + parse_and_process, + source_filter_allows, + periodic_cleanup +); criterion_group!( name = alloc; config = new_crit(Allocs); diff --git a/bench/src/kbench/workload.rs b/bench/src/kbench/workload.rs index 6d15d2a9..7be672cf 100644 --- a/bench/src/kbench/workload.rs +++ b/bench/src/kbench/workload.rs @@ -2,14 +2,15 @@ // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at https://mozilla.org/MPL/2.0/. -// Copyright 2024 Oxide Computer Company +// Copyright 2026 Oxide Computer Company use super::*; use measurement::Instrumentation; #[allow(dead_code)] -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Default)] pub enum IperfMode { + #[default] ClientSend, ServerSend, // TODO: need an updated illumos package. @@ -28,15 +29,10 @@ impl std::fmt::Display for IperfMode { } } -impl Default for IperfMode { - fn default() -> Self { - Self::ClientSend - } -} - #[allow(dead_code)] -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Default)] pub enum IperfProto { + #[default] Tcp, Udp { /// Target bandwidth in MiB/s. @@ -59,12 +55,6 @@ impl std::fmt::Display for IperfProto { } } -impl Default for IperfProto { - fn default() -> Self { - Self::Tcp - } -} - #[derive(Debug, Clone)] pub struct IperfConfig { pub instrumentation: Instrumentation, diff --git a/bench/src/packet.rs b/bench/src/packet.rs index d5ce5422..6664b3e7 100644 --- a/bench/src/packet.rs +++ b/bench/src/packet.rs @@ -26,7 +26,13 @@ use opte_test_utils::icmp::gen_icmp_echo; use opte_test_utils::icmp::gen_icmpv6_echo; use opte_test_utils::icmp::generate_ndisc; use opte_test_utils::*; +use rand::SeedableRng; +use rand::distr::Bernoulli; +use rand::distr::Distribution; +use rand::rngs::StdRng; use std::collections::BTreeMap; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; pub type TestCase = (MsgBlk, Direction); @@ -42,6 +48,16 @@ pub trait BenchPacket { /// Return a list of discrete scenarios fn test_cases(&self) -> Vec>; + + /// Are `generate`d packets worth benchmarking for parser performance? + fn do_parse_benchmark(&self) -> bool { + true + } + + /// Is packet processing allowed to fail due to table size constraints? + fn allow_failure(&self) -> bool { + false + } } /// An individual packet to time the parse/process timing of. @@ -73,13 +89,9 @@ pub struct UlpProcess { pub const ULP_FAST_PATH: UlpProcess = UlpProcess { fast_path: true }; pub const ULP_SLOW_PATH: UlpProcess = UlpProcess { fast_path: false }; -impl BenchPacket for UlpProcess { - fn packet_label(&self) -> &'static str { - if self.fast_path { "ULP-FastPath" } else { "ULP-SlowPath" } - } - - fn test_cases(&self) -> Vec> { - let ip_cfg = IpCfg::DualStack { +impl UlpProcess { + fn cfg() -> IpCfg { + IpCfg::DualStack { ipv4: Ipv4Cfg { vpc_subnet: "172.30.0.0/22".parse().unwrap(), private_ip: "172.30.0.5".parse().unwrap(), @@ -110,9 +122,17 @@ impl BenchPacket for UlpProcess { attached_subnets: BTreeMap::default(), transit_ips: BTreeMap::default(), }, - }; + } + } +} + +impl BenchPacket for UlpProcess { + fn packet_label(&self) -> &'static str { + if self.fast_path { "ULP-FastPath" } else { "ULP-SlowPath" } + } - let cfg = g1_cfg2(ip_cfg); + fn test_cases(&self) -> Vec> { + let cfg = g1_cfg2(UlpProcess::cfg()); itertools::iproduct!( [IpVariant::V4, IpVariant::V6], @@ -334,6 +354,145 @@ impl BenchPacketInstance for UlpProcessInstance { } } +pub struct SlowpathEvict { + pub capacities: Vec, + pub p_expires: Vec, +} + +impl BenchPacket for SlowpathEvict { + fn packet_label(&self) -> &'static str { + "Eviction" + } + + fn test_cases(&self) -> Vec> { + let cfg = g1_cfg2(UlpProcess::cfg()); + itertools::iproduct!(&self.capacities, &self.p_expires) + .map(|(capacity, p_expire)| { + Box::new(AllSynInstance { + index: 0.into(), + capacity: *capacity, + p_expire: *p_expire, + cfg: cfg.clone(), + }) as Box + }) + .collect() + } + + fn do_parse_benchmark(&self) -> bool { + false + } + + fn allow_failure(&self) -> bool { + true + } +} + +#[derive(Debug)] +pub struct AllSynInstance { + index: AtomicU64, + capacity: NonZeroU32, + p_expire: f64, + cfg: VpcCfg, +} + +impl BenchPacketInstance for AllSynInstance { + fn create_port(&self) -> Option { + let mut g1 = + oxide_net_setup("g1_port", &self.cfg, None, Some(self.capacity)); + g1.port.start(); + set!(g1, "port_state=running"); + + Some(g1) + } + + fn parse_with(&self) -> ParserKind { + ParserKind::OxideVpc + } + + fn pre_handle(&self, port: &PortAndVps) { + while port.port.num_flows("firewall", Direction::In) + < self.capacity.get() - 1 + { + let (mut pkt, dir) = self.generate(); + let pkt = parse_inbound(&mut pkt, VpcParser {}).unwrap(); + match port.port.process(dir, pkt) { + Ok(_) => {} + Err(opte::engine::port::ProcessError::Layer( + opte::engine::layer::LayerError::FlowTableFull { .. }, + )) + | Err(opte::engine::port::ProcessError::FlowTableFull { + .. + }) => break, + e => panic!("unexpected err condition {e:?}"), + } + } + + let mut rng = StdRng::seed_from_u64(0x01de_097e_7e57_0712); + let dist = Bernoulli::new(self.p_expire).unwrap(); + + port.port.inject_expiry(|| dist.sample(&mut rng)); + } + + fn instance_name(&self) -> String { + format!("{}/P{}", self.capacity, self.p_expire) + } + + fn generate(&self) -> (MsgBlk, Direction) { + let my_index = self.index.fetch_add(1, Ordering::Relaxed); + + // SYN packets (or small UDP) are the easiest way to prod at + // UFT expiry behaviour. + let src_port = (my_index / u64::from(u16::MAX)) as u16; + let dst_port = (my_index % u64::from(u16::MAX)) as u16; + + let body = &[][..]; + + let eth = Ethernet { + destination: self.cfg.guest_mac, + source: BS_MAC_ADDR, + ethertype: Ethertype::IPV4, + }; + + let tcp = UlpRepr::Tcp(Tcp { + source: src_port, + destination: dst_port, + flags: TcpFlags::SYN, + sequence: 1234, + acknowledgement: 3456, + window_size: 1, + ..Default::default() + }); + + let ip = L3Repr::Ipv4(Ipv4 { + source: Ipv4Addr::from_const([172, 30, 0, 6]), + destination: self.cfg.ipv4().private_ip, + protocol: IngotIpProto::TCP, + total_len: (Ipv4::MINIMUM_LENGTH + (&tcp, &body).packet_length()) + as u16, + ..Default::default() + }); + + let guest_phys = TestIpPhys { + ip: self.cfg.phys_ip, + mac: self.cfg.guest_mac, + vni: self.cfg.vni, + }; + + let partner_phys = TestIpPhys { + ip: Ipv6Addr::from([ + 0xFD00, 0x0000, 0x00F7, 0x0116, 0x0000, 0x0000, 0x0000, 0x0001, + ]), + mac: ox_vpc_mac([0xF0, 0x00, 0x66]), + vni: self.cfg.vni, + }; + + ( + encap(ulp_pkt(eth, ip, tcp, body), partner_phys, guest_phys), + Direction::In, + ) + } +} + pub struct Dhcp4; impl BenchPacket for Dhcp4 { diff --git a/lib/opte-test-utils/src/lib.rs b/lib/opte-test-utils/src/lib.rs index d7d78dfe..9b35fc2f 100644 --- a/lib/opte-test-utils/src/lib.rs +++ b/lib/opte-test-utils/src/lib.rs @@ -270,6 +270,7 @@ fn oxide_net_builder( v2p: Arc, m2p: Arc, v2b: Arc, + flow_table_limits: Option, ) -> PortBuilder { #[allow(clippy::arc_with_non_send_sync)] let ectx = Arc::new(ExecCtx { log: Box::new(opte::PrintlnLog {}) }); @@ -282,12 +283,13 @@ fn oxide_net_builder( NonZeroU32::new(cfg.mtu), ); - let fw_limit = NonZeroU32::new(8096).unwrap(); - let snat_limit = NonZeroU32::new(8096).unwrap(); + let fw_limit = flow_table_limits.unwrap_or(NonZeroU32::new(8096).unwrap()); + let snat_limit = + flow_table_limits.unwrap_or(NonZeroU32::new(8096).unwrap()); let one_limit = NonZeroU32::new(1).unwrap(); firewall::setup(&mut pb, fw_limit).expect("failed to add firewall layer"); - gateway::setup(&pb, cfg, vpc_map, fw_limit) + gateway::setup(&pb, cfg, vpc_map, one_limit) .expect("failed to setup gateway layer"); router::setup(&pb, cfg, one_limit).expect("failed to add router layer"); nat::setup(&mut pb, cfg, snat_limit).expect("failed to add nat layer"); @@ -392,6 +394,7 @@ pub fn oxide_net_setup2( port_v2p, m2p.clone(), v2b, + flow_table_limits, ) .create(vpc_net, uft_limit, tcp_limit) .unwrap(); diff --git a/lib/opte/src/ddi/time.rs b/lib/opte/src/ddi/time.rs index 27fc1b1a..9a04b3b6 100644 --- a/lib/opte/src/ddi/time.rs +++ b/lib/opte/src/ddi/time.rs @@ -58,6 +58,16 @@ impl Add for Moment { } } +#[cfg(any(feature = "std", test))] +impl core::ops::Sub for Moment { + type Output = Self; + + fn sub(self, rhs: Duration) -> Self::Output { + let new = self.inner.saturating_sub(rhs); + Moment { inner: new } + } +} + impl Moment { /// Compute the delta between `now - self` and return as /// milliseconds. @@ -79,9 +89,21 @@ impl Moment { if #[cfg(all(not(feature = "std"), not(test)))] { Self { inner: unsafe { ddi::gethrtime() } } } else { + // This is a pretty gross workaround for the lack of ways to get + // a raw numeric valus associated with an `Instant`. In order to + // enable `Self::raw()` and related functions on std, which allow + // us to manipulate flow timestamps using `Atomic` integers, + // we need to pick an arbitrary zero point and use our duration + // from there as the timestamp. + // + // Currently this is set a few minutes before program start, to + // enable benchmarks and tests which want one or more flows to + // begin in an expired state. static FIRST_TS: OnceLock = OnceLock::new(); - let first_ts = *FIRST_TS.get_or_init(Instant::now); + let first_ts = *FIRST_TS.get_or_init(|| + Instant::now() - Duration::from_mins(5) + ); Self { inner: Instant::now().saturating_duration_since(first_ts) } } } diff --git a/lib/opte/src/engine/flow_table.rs b/lib/opte/src/engine/flow_table.rs index 452883cb..1074d7e8 100644 --- a/lib/opte/src/engine/flow_table.rs +++ b/lib/opte/src/engine/flow_table.rs @@ -134,6 +134,11 @@ pub trait FlowEntryInfo: fmt::Debug + Send + Sync { /// than the stored value. fn inherit_last_hit(&self, new_time: Moment); + /// Forcibly set the last hit time on this entry to `new_time`, when + /// required by some tests/benchmarks. + #[cfg(any(feature = "std", test))] + fn inherit_last_hit_force(&self, new_time: Moment); + /// Determine whether this flow entry can be evicted to make room for /// another, recursively checking all children when needed. fn eviction_priority(&self, now: Moment) -> Option; @@ -162,6 +167,11 @@ impl FlowEntryInfo for FlowEntry { ); } + #[cfg(any(feature = "std", test))] + fn inherit_last_hit_force(&self, new_time: Moment) { + self.lifetime.last_hit.store(new_time.raw(), Ordering::Relaxed); + } + fn eviction_priority(&self, now: Moment) -> Option { let own_prio = self.policy.eviction_priority(self, now); diff --git a/lib/opte/src/engine/port/mod.rs b/lib/opte/src/engine/port/mod.rs index 220f83bf..f686583e 100644 --- a/lib/opte/src/engine/port/mod.rs +++ b/lib/opte/src/engine/port/mod.rs @@ -93,6 +93,8 @@ use core::result; use core::str::FromStr; use core::sync::atomic::AtomicU64; use core::sync::atomic::Ordering::SeqCst; +#[cfg(any(feature = "std", test))] +use core::time::Duration; use illumos_sys_hdrs::uintptr_t; use ingot::ethernet::Ethertype; use ingot::tcp::TcpRef; @@ -1186,6 +1188,44 @@ impl Port { Ok(()) } + /// Use the function `f` to probabilistically mark flows as being ready + /// for expiry by offsetting their timestamps into the past. + /// + /// `f` should return true for a flow which we want to mark (and mark its + /// children as) expirable. + #[cfg(any(feature = "std", test))] + pub fn inject_expiry(&self, mut f: impl FnMut() -> bool) { + let now = Moment::now(); + let before = now - Duration::from_secs(61); + let further_still = before - Duration::from_secs(61); + + let data = self.data.write(); + for dir in [Direction::In, Direction::Out] { + let map = match dir { + Direction::In => data.uft_in.iter(), + Direction::Out => data.uft_out.iter(), + }; + for (_, entry) in map { + let tcp = + entry.state().tcp_flow.as_ref().and_then(|v| v.upgrade()); + if !f() { + entry.hit_at(now); + if let Some(tcp) = tcp { + tcp.hit_at(now); + } + continue; + } + entry.hit_at(before); + if let Some(tcp) = tcp { + tcp.hit_at(before); + } + for parent in &entry.state().parents { + parent.inherit_last_hit_force(further_still); + } + } + } + } + /// Find a rule in the specified layer and return its id. /// /// Search for a matching rule in the specified layer that has the