From d15896bdbffe24b15292d135761240f0ed41780c Mon Sep 17 00:00:00 2001 From: Kyle Simpson Date: Thu, 27 Aug 2026 12:54:58 +0100 Subject: [PATCH 1/3] Add microbenchmarks for table exipry & eviction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This PR adds the above benchmarks to `cargo ubench` for the above functions at different levels of table occupancy to get a bead on what the actual costs involved are. For expiry/cleanup tests, we set the maximum table size and then mark some P∊[0,1] flows as having timestamps such that the flow should be removed entirely. For eviction benchmarks, we ensure that we have at least one full LFT in addition to a full UFT before processing a packet on a new 5-tuple. --- bench/benches/userland.rs | 73 +++++++++++-- bench/src/kbench/workload.rs | 20 +--- bench/src/packet.rs | 165 ++++++++++++++++++++++++++++-- lib/opte-test-utils/src/lib.rs | 9 +- lib/opte/src/ddi/time.rs | 22 +++- lib/opte/src/engine/flow_table.rs | 10 ++ lib/opte/src/engine/port/mod.rs | 40 ++++++++ 7 files changed, 300 insertions(+), 39 deletions(-) diff --git a/bench/benches/userland.rs b/bench/benches/userland.rs index e38ee92f..110e8beb 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; @@ -29,6 +30,9 @@ use oxide_vpc::api::IpAddr; use oxide_vpc::api::Ipv4Addr; use oxide_vpc::api::Ipv6Addr; use oxide_vpc::api::SourceFilter; +use rand::SeedableRng; +use rand::distr::Bernoulli; +use rand::distr::Distribution; use std::collections::BTreeSet; use std::hint::black_box; @@ -44,11 +48,12 @@ 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), ]; 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 +156,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 +180,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 +188,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 +208,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 +216,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 +339,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 expt = SlowpathEvict; + for case in expt.test_cases() { + let port = case.create_port().unwrap(); + for p_expire in [0.0, 0.1, 0.25, 0.5] { + let mut c = c.benchmark_group(format!( + "cleanup/{}/P{}", + M::label(), + p_expire + )); + let mut rng = + rand::rngs::StdRng::seed_from_u64(0x01de_097e_7e57_0712); + let dist = Bernoulli::new(p_expire).unwrap(); + + c.bench_with_input( + BenchmarkId::from_parameter(case.instance_name()), + &case, + |b, _i| { + b.iter_batched( + || { + case.pre_handle(&port); + port.port.inject_expiry(|| dist.sample(&mut rng)); + }, + |_| 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..330b4554 100644 --- a/bench/src/packet.rs +++ b/bench/src/packet.rs @@ -27,6 +27,8 @@ use opte_test_utils::icmp::gen_icmpv6_echo; use opte_test_utils::icmp::generate_ndisc; use opte_test_utils::*; use std::collections::BTreeMap; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; pub type TestCase = (MsgBlk, Direction); @@ -42,6 +44,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 +85,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 +118,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 +350,137 @@ impl BenchPacketInstance for UlpProcessInstance { } } +pub struct SlowpathEvict; + +impl BenchPacket for SlowpathEvict { + fn packet_label(&self) -> &'static str { + "Eviction" + } + + fn test_cases(&self) -> Vec> { + let cfg = g1_cfg2(UlpProcess::cfg()); + [1 << 10, 1 << 15, 1 << 19, 1 << 20] + .into_iter() + .map(|n| { + Box::new(AllSynInstance { + index: 0.into(), + capacity: n.try_into().unwrap(), + 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, + + 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:?}"), + } + } + } + + fn instance_name(&self) -> String { + format!("{}", self.capacity) + } + + 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..8cc195c3 100644 --- a/lib/opte/src/ddi/time.rs +++ b/lib/opte/src/ddi/time.rs @@ -6,6 +6,7 @@ //! Moments, periodics, etc. use core::ops::Add; +use core::ops::Sub; use core::time::Duration; cfg_if! { @@ -58,6 +59,23 @@ impl Add for Moment { } } +impl Sub for Moment { + type Output = Self; + + fn sub(self, rhs: Duration) -> Self::Output { + cfg_if! { + if #[cfg(all(not(feature = "std"), not(test)))] { + let new = self.inner - ((rhs.as_secs() * NANOS) as i64 + + rhs.subsec_nanos() as i64); + Moment { inner: new } + } else { + let new = self.inner - rhs; + Moment { inner: new } + } + } + } +} + impl Moment { /// Compute the delta between `now - self` and return as /// milliseconds. @@ -81,7 +99,9 @@ impl Moment { } else { 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 From 19b65b6b7ccfd8dcdbb304338009d1a5f5dbce20 Mon Sep 17 00:00:00 2001 From: Kyle Simpson Date: Wed, 2 Sep 2026 12:17:21 +0100 Subject: [PATCH 2/3] Limit CI to smaller cleanup jobs --- .github/buildomat/jobs/bench.sh | 1 + bench/benches/userland.rs | 61 +++++++++++++++++---------------- bench/src/packet.rs | 26 ++++++++++---- 3 files changed, 52 insertions(+), 36 deletions(-) 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 110e8beb..965ed510 100644 --- a/bench/benches/userland.rs +++ b/bench/benches/userland.rs @@ -30,9 +30,6 @@ use oxide_vpc::api::IpAddr; use oxide_vpc::api::Ipv4Addr; use oxide_vpc::api::Ipv6Addr; use oxide_vpc::api::SourceFilter; -use rand::SeedableRng; -use rand::distr::Bernoulli; -use rand::distr::Distribution; use std::collections::BTreeSet; use std::hint::black_box; @@ -48,7 +45,13 @@ 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), + 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 { @@ -340,34 +343,34 @@ fn source_filter_allows(c: &mut Criterion) { } fn periodic_cleanup(c: &mut Criterion) { - let expt = SlowpathEvict; + 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(); - for p_expire in [0.0, 0.1, 0.25, 0.5] { - let mut c = c.benchmark_group(format!( - "cleanup/{}/P{}", - M::label(), - p_expire - )); - let mut rng = - rand::rngs::StdRng::seed_from_u64(0x01de_097e_7e57_0712); - let dist = Bernoulli::new(p_expire).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); - port.port.inject_expiry(|| dist.sample(&mut rng)); - }, - |_| black_box(port.port.expire_flows()), - criterion::BatchSize::LargeInput, - ) - }, - ); - } + 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, + ) + }, + ); } } diff --git a/bench/src/packet.rs b/bench/src/packet.rs index 330b4554..6664b3e7 100644 --- a/bench/src/packet.rs +++ b/bench/src/packet.rs @@ -26,6 +26,10 @@ 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; @@ -350,7 +354,10 @@ impl BenchPacketInstance for UlpProcessInstance { } } -pub struct SlowpathEvict; +pub struct SlowpathEvict { + pub capacities: Vec, + pub p_expires: Vec, +} impl BenchPacket for SlowpathEvict { fn packet_label(&self) -> &'static str { @@ -359,12 +366,12 @@ impl BenchPacket for SlowpathEvict { fn test_cases(&self) -> Vec> { let cfg = g1_cfg2(UlpProcess::cfg()); - [1 << 10, 1 << 15, 1 << 19, 1 << 20] - .into_iter() - .map(|n| { + itertools::iproduct!(&self.capacities, &self.p_expires) + .map(|(capacity, p_expire)| { Box::new(AllSynInstance { index: 0.into(), - capacity: n.try_into().unwrap(), + capacity: *capacity, + p_expire: *p_expire, cfg: cfg.clone(), }) as Box }) @@ -384,7 +391,7 @@ impl BenchPacket for SlowpathEvict { pub struct AllSynInstance { index: AtomicU64, capacity: NonZeroU32, - + p_expire: f64, cfg: VpcCfg, } @@ -419,10 +426,15 @@ impl BenchPacketInstance for AllSynInstance { 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!("{}", self.capacity) + format!("{}/P{}", self.capacity, self.p_expire) } fn generate(&self) -> (MsgBlk, Direction) { From 304ecfd4c07d7670bd67c4da70c5d89e21e0535e Mon Sep 17 00:00:00 2001 From: Kyle Simpson Date: Thu, 3 Sep 2026 13:45:45 +0100 Subject: [PATCH 3/3] Review feedback --- lib/opte/src/ddi/time.rs | 26 ++++++++++++++------------ 1 file changed, 14 insertions(+), 12 deletions(-) diff --git a/lib/opte/src/ddi/time.rs b/lib/opte/src/ddi/time.rs index 8cc195c3..9a04b3b6 100644 --- a/lib/opte/src/ddi/time.rs +++ b/lib/opte/src/ddi/time.rs @@ -6,7 +6,6 @@ //! Moments, periodics, etc. use core::ops::Add; -use core::ops::Sub; use core::time::Duration; cfg_if! { @@ -59,20 +58,13 @@ impl Add for Moment { } } -impl Sub for Moment { +#[cfg(any(feature = "std", test))] +impl core::ops::Sub for Moment { type Output = Self; fn sub(self, rhs: Duration) -> Self::Output { - cfg_if! { - if #[cfg(all(not(feature = "std"), not(test)))] { - let new = self.inner - ((rhs.as_secs() * NANOS) as i64 + - rhs.subsec_nanos() as i64); - Moment { inner: new } - } else { - let new = self.inner - rhs; - Moment { inner: new } - } - } + let new = self.inner.saturating_sub(rhs); + Moment { inner: new } } } @@ -97,6 +89,16 @@ 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(||