diff --git a/.semgrep/rules/no-clock-read-in-drop.yaml b/.semgrep/rules/no-clock-read-in-drop.yaml new file mode 100644 index 0000000000..eb2e38fb2c --- /dev/null +++ b/.semgrep/rules/no-clock-read-in-drop.yaml @@ -0,0 +1,27 @@ +rules: + - id: rust-no-clock-read-in-drop + languages: [rust] + severity: ERROR + message: | + Do not read the clock from a `Drop` implementation. + + Once anything in the process has paused the virtual clock, tokio routes + every read through the calling thread's runtime context. A `Drop` that + runs during thread-local teardown may find that context already + destroyed, and tokio's response is a panic inside a destructor -- which + aborts the process rather than failing the test. + + Take the reading before the value is dropped and pass it in, or record + the instant when the value is created. + paths: + exclude: + - .codeql/tests/ + - clock/src/ + patterns: + - pattern-inside: | + fn drop(&mut self) { + ... + } + - pattern-either: + - pattern: clock::now() + - pattern: clock::system_now() diff --git a/Cargo.lock b/Cargo.lock index eb3f3da814..ffd3dfc0f4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1316,6 +1316,7 @@ dependencies = [ name = "dataplane-clock" version = "0.25.2" dependencies = [ + "dataplane-concurrency", "tokio", ] diff --git a/Cargo.toml b/Cargo.toml index f0783857a8..3ab7535f2f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -256,7 +256,7 @@ overflow-checks = false codegen-units = 1 rpath = true -[profile.fuzz] +[profile.checked] inherits = "release" opt-level = 2 debug-assertions = true diff --git a/acl/tests/property_predicate.rs b/acl/tests/property_predicate.rs index 3e245db8d4..007462100d 100644 --- a/acl/tests/property_predicate.rs +++ b/acl/tests/property_predicate.rs @@ -211,100 +211,102 @@ where } const MIN_ASSERTED_HITS: u64 = 20; const MIN_ASSERTED_MISSES: u64 = 20; -fn run_property( - name_prefix: &str, - install_dpdk: impl Fn(String, &FiveTupleRule) -> T + core::panic::RefUnwindSafe, -) where - A: KeyAddr, - PrefixSpec: FieldHit + FieldMiss + IsUniversal, - T: Lookup, Verdict>, - RawRule: TypeGenerator, -{ - let asserted_hits = AtomicU64::new(0); - let asserted_misses = AtomicU64::new(0); +macro_rules! run_property { + ($a:ty, $name_prefix:expr, $install_dpdk:expr) => {{ + let asserted_hits = AtomicU64::new(0); + let asserted_misses = AtomicU64::new(0); - bolero::check!() - .with_type::<(RawRule, Box<[u8]>, Box<[u8]>)>() - .for_each(|(raw, hit_bytes, miss_bytes)| { - let rule = build_rule(raw); - let dpdk = install_dpdk(unique_name(name_prefix), &rule); - let reference = ReferenceTable::, Verdict>::new(vec![RefRule::new( - rule.into_backend_fields::(), - Verdict::Drop, - )]); - - let hits = HitsGen { rule }; - let n_hits = sweep(&hits, hit_bytes, |k| { - assert!(rule.accepts(k), "hits gen produced a rejected key: {k:?}"); - assert_eq!(reference.lookup(k), Some(&Verdict::Drop)); - assert_eq!(dpdk.lookup(k), Some(&Verdict::Drop)); - }); - asserted_hits.fetch_add(n_hits, Ordering::Relaxed); + bolero::check!() + .with_type::<(RawRule<$a>, Box<[u8]>, Box<[u8]>)>() + .for_each(|(raw, hit_bytes, miss_bytes)| { + let rule = build_rule(raw); + let dpdk = $install_dpdk(unique_name($name_prefix), &rule); + let reference = ReferenceTable::, Verdict>::new(vec![RefRule::new( + rule.into_backend_fields::(), + Verdict::Drop, + )]); - if !rule.is_universal() { - let misses = MissesGen { rule }; - let n_misses = sweep(&misses, miss_bytes, |k| { - assert!( - !rule.accepts(k), - "misses gen produced an accepted key: {k:?}", - ); - assert_eq!(reference.lookup(k), None); - assert_eq!(dpdk.lookup(k), None); + let hits = HitsGen { rule }; + let n_hits = sweep(&hits, hit_bytes, |k| { + assert!(rule.accepts(k), "hits gen produced a rejected key: {k:?}"); + assert_eq!(reference.lookup(k), Some(&Verdict::Drop)); + assert_eq!(dpdk.lookup(k), Some(&Verdict::Drop)); }); - asserted_misses.fetch_add(n_misses, Ordering::Relaxed); - } - }); + asserted_hits.fetch_add(n_hits, Ordering::Relaxed); - let h = asserted_hits.load(Ordering::Relaxed); - let m = asserted_misses.load(Ordering::Relaxed); - assert!( - h >= MIN_ASSERTED_HITS, - "asserted only {h} hits (< {MIN_ASSERTED_HITS}); generator may have gone inert", - ); - assert!( - m >= MIN_ASSERTED_MISSES, - "asserted only {m} misses (< {MIN_ASSERTED_MISSES}); generator may have gone inert", - ); + if !rule.is_universal() { + let misses = MissesGen { rule }; + let n_misses = sweep(&misses, miss_bytes, |k| { + assert!( + !rule.accepts(k), + "misses gen produced an accepted key: {k:?}", + ); + assert_eq!(reference.lookup(k), None); + assert_eq!(dpdk.lookup(k), None); + }); + asserted_misses.fetch_add(n_misses, Ordering::Relaxed); + } + }); + + let h = asserted_hits.load(Ordering::Relaxed); + let m = asserted_misses.load(Ordering::Relaxed); + assert!( + h >= MIN_ASSERTED_HITS, + "asserted only {h} hits (< {MIN_ASSERTED_HITS}); generator may have gone inert", + ); + assert!( + m >= MIN_ASSERTED_MISSES, + "asserted only {m} misses (< {MIN_ASSERTED_MISSES}); generator may have gone inert", + ); + }}; } #[test] #[dpdk::with_eal] fn property_v4() { - run_property::>("prop_v4", |name, rule| { - install_table( - &name, - NonZero::new(2).expect("nonzero"), - vec![ - RuleSpec::, Verdict>::new( - Priority::new(1).expect("nonzero priority"), - CategoryMask::new(1).expect("nonzero mask"), - rule.into_backend_fields::(), - Verdict::Drop, - ) - .expect("RuleSpec"), - ], - ) - .expect("install_table") - }); + run_property!( + Ipv4Addr, + "prop_v4", + |name: String, rule: &FiveTupleRule| { + install_table( + &name, + NonZero::new(2).expect("nonzero"), + vec![ + RuleSpec::, Verdict>::new( + Priority::new(1).expect("nonzero priority"), + CategoryMask::new(1).expect("nonzero mask"), + rule.into_backend_fields::(), + Verdict::Drop, + ) + .expect("RuleSpec"), + ], + ) + .expect("install_table") + } + ); } #[test] #[dpdk::with_eal] fn property_v6() { - run_property::>("prop_v6", |name, rule| { - install_table( - &name, - NonZero::new(2).expect("nonzero"), - vec![ - RuleSpec::, Verdict>::new( - Priority::new(1).expect("nonzero priority"), - CategoryMask::new(1).expect("nonzero mask"), - rule.into_backend_fields::(), - Verdict::Drop, - ) - .expect("RuleSpec"), - ], - ) - .expect("install_table") - }); + run_property!( + Ipv6Addr, + "prop_v6", + |name: String, rule: &FiveTupleRule| { + install_table( + &name, + NonZero::new(2).expect("nonzero"), + vec![ + RuleSpec::, Verdict>::new( + Priority::new(1).expect("nonzero priority"), + CategoryMask::new(1).expect("nonzero mask"), + rule.into_backend_fields::(), + Verdict::Drop, + ) + .expect("RuleSpec"), + ], + ) + .expect("install_table") + } + ); } diff --git a/clock/Cargo.toml b/clock/Cargo.toml index 17644aa53f..75776e69e1 100644 --- a/clock/Cargo.toml +++ b/clock/Cargo.toml @@ -11,3 +11,9 @@ virtual = ["dep:tokio"] [dependencies] tokio = { workspace = true, optional = true, features = ["test-util", "time"] } + +[dev-dependencies] +concurrency = { workspace = true } + +[lints.rust] +unexpected_cfgs = { level = "warn", check-cfg = ['cfg(wall_clock)'] } diff --git a/clock/build.rs b/clock/build.rs new file mode 100644 index 0000000000..4a4a7fc660 --- /dev/null +++ b/clock/build.rs @@ -0,0 +1,39 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright Open Network Fabric Authors + +use std::process::Command; +use std::{env, fs, path::PathBuf}; + +fn main() { + println!("cargo::rerun-if-env-changed=RUSTC_BOOTSTRAP"); + println!("cargo::rustc-check-cfg=cfg(has_spawn_hook)"); + + let out = PathBuf::from(env::var_os("OUT_DIR").expect("cargo sets OUT_DIR")); + let probe = out.join("spawn_hook_probe.rs"); + if fs::write( + &probe, + "#![feature(thread_spawn_hook)]\n\ + pub fn probe() { std::thread::add_spawn_hook(|_| || {}); }\n", + ) + .is_err() + { + return; + } + + let rustc = env::var_os("RUSTC").unwrap_or_else(|| "rustc".into()); + let accepted = Command::new(rustc) + .args(["--crate-type=lib", "--emit=metadata", "-o"]) + .arg(out.join("spawn_hook_probe.rmeta")) + .arg(&probe) + .status() + .is_ok_and(|status| status.success()); + + if accepted { + println!("cargo::rustc-cfg=has_spawn_hook"); + } else { + println!( + "cargo::warning=thread_spawn_hook is unavailable, so a test that drives the clock \ + cannot check the threads it spawns. Set RUSTC_BOOTSTRAP=1 (the dev shell does)." + ); + } +} diff --git a/clock/src/lib.rs b/clock/src/lib.rs index 9493f60e41..1bd29872d4 100644 --- a/clock/src/lib.rs +++ b/clock/src/lib.rs @@ -1,35 +1,83 @@ // SPDX-License-Identifier: Apache-2.0 // Copyright Open Network Fabric Authors +#![cfg_attr( + all(has_spawn_hook, feature = "virtual", not(wall_clock)), + feature(thread_spawn_hook) +)] #![deny(clippy::all, clippy::pedantic)] #![deny(rustdoc::all)] #![deny(unsafe_code)] pub use std::time::{Duration, Instant, SystemTime, SystemTimeError, TryFromFloatSecsError}; +#[cfg(feature = "virtual")] +pub mod virtual_time; + #[must_use] pub fn now() -> Instant { - #[cfg(feature = "virtual")] + #[cfg(all(feature = "virtual", not(wall_clock)))] { - tokio::time::Instant::now().into_std() + checked_now().unwrap_or_else(|| virtual_time::refuse()) } - #[cfg(not(feature = "virtual"))] + #[cfg(not(all(feature = "virtual", not(wall_clock))))] { Instant::now() } } +#[must_use] +pub fn checked_now() -> Option { + #[cfg(all(feature = "virtual", not(wall_clock)))] + { + if virtual_time::armed() && tokio::runtime::Handle::try_current().is_err() { + return None; + } + Some(tokio::time::Instant::now().into_std()) + } + #[cfg(not(all(feature = "virtual", not(wall_clock))))] + { + Some(Instant::now()) + } +} + +#[must_use] +pub const fn is_routed() -> bool { + cfg!(all(feature = "virtual", not(wall_clock))) +} + +#[must_use] +pub fn elapsed_since_first_reading() -> Option<(bool, Duration)> { + // nosemgrep: rust-no-direct-std-sync-import + static ORIGIN: std::sync::OnceLock = std::sync::OnceLock::new(); + let reading = checked_now()?; + let origin = *ORIGIN.get_or_init(|| reading); + Some(if reading >= origin { + (false, reading.saturating_duration_since(origin)) + } else { + (true, origin.saturating_duration_since(reading)) + }) +} + #[must_use] pub fn system_now() -> SystemTime { SystemTime::now() } +#[cfg(test)] +pub(crate) fn serially() -> concurrency::sync::MutexGuard<'static, ()> { + static SERIAL: concurrency::sync::Mutex<()> = concurrency::sync::Mutex::new(()); + SERIAL.lock() +} + #[cfg(test)] mod tests { + use super::serially; use super::{Duration, now, system_now}; #[test] fn now_is_monotonic() { + let _serial = serially(); let first = now(); let second = now(); assert!(second >= first, "the monotonic clock went backwards"); @@ -37,6 +85,7 @@ mod tests { #[test] fn now_works_with_no_runtime() { + let _serial = serially(); let _ = now(); let _ = system_now(); } diff --git a/clock/src/virtual_time.rs b/clock/src/virtual_time.rs new file mode 100644 index 0000000000..8e79725c5e --- /dev/null +++ b/clock/src/virtual_time.rs @@ -0,0 +1,274 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright Open Network Fabric Authors + +use crate::Duration; +use std::cell::Cell; +// nosemgrep: rust-no-direct-std-sync-import +use std::sync::atomic::{AtomicUsize, Ordering}; + +static LIVE: AtomicUsize = AtomicUsize::new(0); + +const YIELDS: usize = 4; + +thread_local! { + static IN_WORLD: Cell = const { Cell::new(0) }; +} + +#[cfg(not(wall_clock))] +#[inline] +#[must_use] +pub(crate) fn armed() -> bool { + LIVE.load(Ordering::Acquire) != 0 && IN_WORLD.with(Cell::get) != 0 +} + +thread_local! { + #[cfg(all(has_spawn_hook, not(wall_clock)))] + static HOOKED: Cell = const { Cell::new(false) }; +} + +#[cfg(all(has_spawn_hook, not(wall_clock)))] +fn inherit_across_spawns() { + if !HOOKED.replace(true) { + std::thread::add_spawn_hook(|_parent| { + let handle = tokio::runtime::Handle::try_current().ok(); + let in_world = IN_WORLD.with(Cell::get); + move || { + IN_WORLD.with(|depth| depth.set(in_world)); + if let Some(handle) = handle { + std::mem::forget(Box::leak(Box::new(handle)).enter()); + } + } + }); + } +} + +#[cfg(not(all(has_spawn_hook, not(wall_clock))))] +fn inherit_across_spawns() {} + +#[cfg(not(wall_clock))] +#[cold] +#[inline(never)] +pub(crate) fn refuse() -> ! { + panic!( + "clock::now() on a thread with no tokio runtime while the virtual clock is paused.\n\ + \n\ + This read would have answered from the wall clock, which is a different timeline from the \ + paused one -- they disagree by however far the test has advanced -- so comparing it \ + against a deadline taken on the other side is silently wrong, in either direction.\n\ + \n\ + A thread spawned with `std::thread` inherits its parent's clock automatically, so reaching \ + this means this one did not come from there -- DPDK's EAL, or a C library calling \ + `pthread_create`. Enter `clock::virtual_time::Paused::handle()` on the thread itself; a \ + guard held by whoever created it does nothing, because tokio's context is thread-local." + ); +} + +#[derive(Debug)] +pub struct Paused { + runtime: tokio::runtime::Runtime, +} + +impl Paused { + #[must_use] + pub fn new() -> Self { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_time() + .start_paused(!cfg!(wall_clock)) + .build() + .unwrap_or_else(|e| panic!("a current-thread runtime with timers does not build: {e}")); + + if !cfg!(wall_clock) { + inherit_across_spawns(); + IN_WORLD.with(|depth| depth.set(depth.get() + 1)); + LIVE.fetch_add(1, Ordering::AcqRel); + } + + Self { runtime } + } + + pub fn block_on(&self, future: F) -> F::Output { + self.runtime.block_on(future) + } + + #[must_use] + pub fn handle(&self) -> tokio::runtime::Handle { + self.runtime.handle().clone() + } +} + +impl Default for Paused { + fn default() -> Self { + Self::new() + } +} + +impl Drop for Paused { + fn drop(&mut self) { + if !cfg!(wall_clock) { + IN_WORLD.with(|depth| depth.set(depth.get().saturating_sub(1))); + LIVE.fetch_sub(1, Ordering::Release); + } + } +} + +pub async fn advance(by: Duration) { + #[cfg(not(wall_clock))] + { + tokio::time::advance(by).await; + for _ in 0..YIELDS { + tokio::task::yield_now().await; + } + } + #[cfg(wall_clock)] + { + let _ = YIELDS; + tokio::time::sleep(by).await; + } +} + +#[cfg(test)] +mod tests { + use super::{Paused, advance}; + use crate::serially; + use crate::{Duration, now}; + use std::thread; + + const LONG: Duration = if cfg!(wall_clock) { + Duration::from_millis(50) + } else { + Duration::from_hours(1) + }; + + const NEARLY_LONG: Duration = if cfg!(wall_clock) { + Duration::from_millis(20) + } else { + Duration::from_mins(2) + }; + + #[test] + fn the_clock_moves_when_a_test_says_so() { + let _serial = serially(); + let clock = Paused::new(); + clock.block_on(async { + let before = now(); + advance(LONG).await; + assert!( + now().duration_since(before) >= LONG, + "the clock was advanced by {LONG:?} and did not follow" + ); + }); + } + + #[test] + fn a_timer_fires_when_the_clock_passes_it() { + let _serial = serially(); + let clock = Paused::new(); + clock.block_on(async { + let deadline = now() + NEARLY_LONG; + let waiting = tokio::spawn(async move { + while now() < deadline { + tokio::task::yield_now().await; + } + }); + advance(LONG).await; + waiting.await.expect("the waiter panicked"); + }); + } + + #[test] + #[cfg(all(has_spawn_hook, not(wall_clock)))] + fn a_spawned_thread_inherits_the_clock_without_being_told() { + let _serial = serially(); + let clock = Paused::new(); + let (driver, worker) = clock.block_on(async { + advance(LONG).await; + let worker = thread::spawn(now) + .join() + .expect("an inherited read was refused"); + (now(), worker) + }); + assert_eq!( + driver, + worker, + "a spawned thread read {:?} away from the thread that advanced the clock", + driver.saturating_duration_since(worker) + ); + } + + #[test] + #[cfg(not(wall_clock))] + fn a_thread_in_the_world_with_no_clock_is_refused() { + let _serial = serially(); + let clock = Paused::new(); + clock.block_on(async { advance(LONG).await }); + + let refused = std::panic::catch_unwind(now); + let panic = refused.expect_err("a read from the wrong timeline was allowed"); + let message = panic + .downcast_ref::<&'static str>() + .copied() + .or_else(|| panic.downcast_ref::().map(String::as_str)) + .unwrap_or(""); + assert!( + message.contains("no tokio runtime"), + "the refusal did not explain itself: {message}" + ); + } + + #[test] + #[cfg(not(wall_clock))] + fn a_thread_outside_the_tree_is_left_alone() { + let _serial = serially(); + let (started, wait_for_start) = std::sync::mpsc::channel(); + let (finish, wait_to_finish) = std::sync::mpsc::channel(); + + let driving = thread::spawn(move || { + let clock = Paused::new(); + clock.block_on(async { advance(LONG).await }); + started.send(()).expect("the test is waiting"); + wait_to_finish.recv().expect("the test releases this"); + }); + wait_for_start.recv().expect("the driver starts"); + + let outsider = thread::spawn(now).join(); + finish.send(()).expect("the driver is waiting"); + driving.join().expect("the driver panicked"); + outsider.expect("a thread outside the world was refused a clock read"); + } + + #[test] + fn a_thread_that_entered_reads_the_same_clock() { + let _serial = serially(); + let clock = Paused::new(); + let driver = clock.block_on(async { + advance(LONG).await; + now() + }); + + let handle = clock.handle(); + let worker = thread::spawn(move || { + let _guard = handle.enter(); + now() + }) + .join() + .expect("an entered read was refused"); + + assert!( + worker >= driver, + "an entered worker read {:?} behind the thread that advanced the clock", + driver.saturating_duration_since(worker) + ); + } + + #[test] + fn dropping_the_clock_disarms_the_check() { + let _serial = serially(); + { + let clock = Paused::new(); + clock.block_on(async { advance(LONG).await }); + } + thread::spawn(now) + .join() + .expect("an ordinary read was refused after the clock was dropped"); + } +} diff --git a/concurrency/tests/quiescent_shuttle.rs b/concurrency/tests/quiescent_shuttle.rs index ab1647514c..31d252e0b3 100644 --- a/concurrency/tests/quiescent_shuttle.rs +++ b/concurrency/tests/quiescent_shuttle.rs @@ -22,8 +22,6 @@ #![cfg(not(feature = "loom"))] -use std::panic::RefUnwindSafe; - use bolero::TypeGenerator; use dataplane_concurrency::sync::Arc; use dataplane_concurrency::sync::atomic::{AtomicUsize, Ordering}; @@ -155,20 +153,20 @@ fn run_plan(plan: &Plan) { const TEST_TIME: std::time::Duration = std::time::Duration::from_secs(10); -fn fuzz_test( - test: impl Fn(Arg) + RefUnwindSafe, -) { - bolero::check!() - .with_type() - .cloned() - .with_test_time(TEST_TIME) - .for_each(test); +macro_rules! fuzz_test { + ($test:expr) => {{ + bolero::check!() + .with_type() + .cloned() + .with_test_time(TEST_TIME) + .for_each($test); + }}; } #[test] #[cfg(feature = "shuttle")] fn protocol_under_shuttle() { - fuzz_test(|plan: Plan| { + fuzz_test!(|plan: Plan| { let runner = shuttle::Runner::new( shuttle::scheduler::RandomScheduler::new(1), dataplane_concurrency::shuttle_config(), @@ -180,7 +178,7 @@ fn protocol_under_shuttle() { #[test] #[cfg(feature = "shuttle")] fn protocol_under_shuttle_pct() { - fuzz_test(|plan: Plan| { + fuzz_test!(|plan: Plan| { // PCT requires both threads to actually do atomic ops; if // either side is effectively empty, shuttle's PCT scheduler // panics with "test closure did not exercise any concurrency". @@ -208,5 +206,5 @@ fn protocol_under_shuttle_pct() { #[test] #[cfg(not(feature = "shuttle"))] fn protocol_under_std() { - fuzz_test(|plan: Plan| run_plan(&plan)); + fuzz_test!(|plan: Plan| run_plan(&plan)); } diff --git a/concurrency/tests/scope_property.rs b/concurrency/tests/scope_property.rs index 9e323e3808..3b9bba5ac6 100644 --- a/concurrency/tests/scope_property.rs +++ b/concurrency/tests/scope_property.rs @@ -28,8 +28,6 @@ #![cfg(not(feature = "loom"))] -use std::panic::RefUnwindSafe; - use bolero::TypeGenerator; use dataplane_concurrency::sync::Arc; use dataplane_concurrency::sync::atomic::{AtomicUsize, Ordering}; @@ -98,26 +96,26 @@ fn run_plan(plan: &Plan) { const TEST_TIME: std::time::Duration = std::time::Duration::from_secs(10); -fn fuzz_test( - test: impl Fn(Arg) + RefUnwindSafe, -) { - bolero::check!() - .with_type() - .cloned() - .with_test_time(TEST_TIME) - .for_each(test); +macro_rules! fuzz_test { + ($test:expr) => {{ + bolero::check!() + .with_type() + .cloned() + .with_test_time(TEST_TIME) + .for_each($test); + }}; } #[test] #[cfg(feature = "shuttle")] fn scope_conservation_under_shuttle() { - fuzz_test(|plan: Plan| shuttle::check_random(move || run_plan(&plan), 1)); + fuzz_test!(|plan: Plan| shuttle::check_random(move || run_plan(&plan), 1)); } #[test] #[cfg(feature = "shuttle")] fn scope_conservation_under_shuttle_pct() { - fuzz_test(|plan: Plan| { + fuzz_test!(|plan: Plan| { // PCT requires every thread to do at least one atomic op; // skip degenerate shapes that wouldn't exercise concurrency. let nontrivial = plan @@ -136,5 +134,5 @@ fn scope_conservation_under_shuttle_pct() { #[test] #[cfg(not(feature = "shuttle"))] fn scope_conservation_under_std() { - fuzz_test(|plan: Plan| run_plan(&plan)); + fuzz_test!(|plan: Plan| run_plan(&plan)); } diff --git a/config/src/external/overlay/completeness.rs b/config/src/external/overlay/completeness.rs index c74744d0b9..4837da6d9e 100644 --- a/config/src/external/overlay/completeness.rs +++ b/config/src/external/overlay/completeness.rs @@ -362,28 +362,27 @@ fn survey_nat(nat: &VpcExposeNat, seen: &mut Observed) { const CASES: usize = 512; -fn survey_drawn(seen: &RefCell) { - let seen = std::panic::AssertUnwindSafe(seen); - bolero::check!() - .with_generator(Sequence::default()) - .with_iterations(CASES) - .for_each(|ops| { - let overlay = Sequence::fold(ops) - .overlay() - .unwrap_or_else(|e| panic!("{ops:?} does not assemble: {e}")); - survey(&overlay, &mut seen.borrow_mut()); - }); -} - -fn census() -> Observed { - let seen = RefCell::new(Observed::default()); - survey_drawn(&seen); - seen.into_inner() +macro_rules! survey_drawn { + ($seen:expr) => {{ + let seen = $seen; + let seen = std::panic::AssertUnwindSafe(seen); + bolero::check!() + .with_generator(Sequence::default()) + .with_iterations(CASES) + .for_each(|ops| { + let overlay = Sequence::fold(ops) + .overlay() + .unwrap_or_else(|e| panic!("{ops:?} does not assemble: {e}")); + survey(&overlay, &mut seen.borrow_mut()); + }); + }}; } #[test] fn every_surveyed_field_is_classified() { - let seen = census(); + let seen = RefCell::new(Observed::default()); + survey_drawn!(&seen); + let seen = seen.into_inner(); let surveyed: BTreeSet<&str> = seen.0.keys().copied().collect(); let classified: BTreeSet<&str> = REACH.iter().map(|(field, _)| *field).collect(); @@ -417,7 +416,9 @@ fn every_surveyed_field_is_classified() { #[test] fn the_algebra_reaches_what_it_is_recorded_to_reach() { - let seen = census(); + let seen = RefCell::new(Observed::default()); + survey_drawn!(&seen); + let seen = seen.into_inner(); for (field, reach) in REACH { let Some(values) = seen.0.get(field) else { diff --git a/dataplane/src/packet_processor/fuzz.rs b/dataplane/src/packet_processor/fuzz.rs index f5090b877e..095c91c641 100644 --- a/dataplane/src/packet_processor/fuzz.rs +++ b/dataplane/src/packet_processor/fuzz.rs @@ -648,6 +648,24 @@ pub(crate) struct Pick { #[cfg(test)] pub(crate) type Poll = Vec; +#[cfg(test)] +pub(crate) fn settled(body: impl FnOnce()) { + static RUNTIME: concurrency::sync::LazyLock = + concurrency::sync::LazyLock::new(|| { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime") + }); + + RUNTIME.block_on(async { + body(); + for _ in 0..4 { + tokio::task::yield_now().await; + } + }); +} + #[cfg(test)] pub(crate) fn run_schedule( worker: &mut Worker, @@ -686,6 +704,47 @@ pub(crate) fn run_schedule( bursts } +#[cfg(test)] +pub(crate) async fn run_schedule_over_time( + worker: &mut Worker, + loads: &mut [Box], + schedule: &[Poll], + waits: &[Duration], +) -> Vec> { + let mut bursts = Vec::new(); + for (nth, poll) in schedule.iter().enumerate() { + let mut burst = Vec::new(); + let mut origin = Vec::new(); + for pick in poll { + if loads.is_empty() { + break; + } + let which = usize::from(pick.load) % loads.len(); + for _ in 0..pick.take { + let Some(packet) = loads[which].next() else { + break; + }; + burst.push(packet); + origin.push(which); + } + } + if !burst.is_empty() { + for (answer, which) in worker.send_batch(burst).iter().zip(&origin) { + loads[*which].observe(answer); + } + bursts.push(origin); + } + if let Some(wait) = waits.get(nth) { + clock::virtual_time::advance(*wait).await; + } + } + + for load in loads { + drive(worker, load.as_mut()); + } + bursts +} + #[cfg(test)] pub(crate) mod derive { use super::routed::{Blast, Conversation, Inbound}; @@ -1400,56 +1459,59 @@ mod shapes { super::assert_within_budget("shapes::Batch", &Batch); } - #[tokio::test] - #[dpdk::with_eal] - async fn every_shape_leaves_the_pipeline_with_a_verdict() { + #[test] + fn every_shape_leaves_the_pipeline_with_a_verdict() { static FORWARDED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static DROPPED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static BY_SHAPE: LazyLock<[AtomicU64; Shape::ALL.len()]> = LazyLock::new(|| std::array::from_fn(|_| AtomicU64::new(0))); + let _eal = dpdk::test_support::start_eal(); + bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Batch) .for_each(|(exposes, stacks)| { - let Some(mut fabric) = Fabric::build(exposes) else { - return; - }; - let private = exposes.first().and_then(|e| { - e.ips - .first() - .map(lpm::prefix::PrefixWithOptionalPorts::prefix) - }); - - for (shape, headers) in stacks { - let mut headers = headers.clone(); - aim(&mut headers, private); - let Some(mut packet) = wire(&headers) else { - continue; + settled(|| { + let Some(mut fabric) = Fabric::build(exposes) else { + return; }; - BY_SHAPE[*shape as usize].fetch_add(1, Ordering::Relaxed); + let private = exposes.first().and_then(|e| { + e.ips + .first() + .map(lpm::prefix::PrefixWithOptionalPorts::prefix) + }); - arrive(&mut packet, local()); - let out = fabric.send(packet); + for (shape, headers) in stacks { + let mut headers = headers.clone(); + aim(&mut headers, private); + let Some(mut packet) = wire(&headers) else { + continue; + }; + BY_SHAPE[*shape as usize].fetch_add(1, Ordering::Relaxed); - match verdict(&out) { - Verdict::Forwarded { dst_vpcd, .. } => { - assert_eq!( - dst_vpcd, - Some(remote()), - "forwarded without a destination VPC, on a {shape:?} stack: \ + arrive(&mut packet, local()); + let out = fabric.send(packet); + + match verdict(&out) { + Verdict::Forwarded { dst_vpcd, .. } => { + assert_eq!( + dst_vpcd, + Some(remote()), + "forwarded without a destination VPC, on a {shape:?} stack: \ nothing chose where this packet goes" - ); - FORWARDED.fetch_add(1, Ordering::Relaxed); - } - Verdict::Dropped(_) => { - DROPPED.fetch_add(1, Ordering::Relaxed); - } - Verdict::Delivered { .. } => { - unreachable!("the overlay slice has no egress stage") + ); + FORWARDED.fetch_add(1, Ordering::Relaxed); + } + Verdict::Dropped(_) => { + DROPPED.fetch_add(1, Ordering::Relaxed); + } + Verdict::Delivered { .. } => { + unreachable!("the overlay slice has no egress stage") + } } } - } + }); }); let forwarded = FORWARDED.load(Ordering::Relaxed); @@ -1598,93 +1660,97 @@ mod round_trip { super::assert_within_budget("round_trip::Batch", &Batch); } - #[tokio::test] - #[dpdk::with_eal] - async fn a_translated_flow_comes_back_to_where_it_started() { + #[test] + fn a_translated_flow_comes_back_to_where_it_started() { static ROUND_TRIPPED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static NOT_FORWARDED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + let _eal = dpdk::test_support::start_eal(); + bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Batch) .for_each(|(exposes, flows)| { - let Some(mut fabric) = Fabric::build(exposes) else { - return; - }; - let privates = private_addresses(exposes); - if privates.is_empty() { - return; - } - - for flow in flows { - let prefix = privates[usize::from(flow.prefix) % privates.len()]; - let src = match prefix.as_address() { - IpAddr::V4(a) => { - let mut o = a.octets(); - o[3] = o[3].wrapping_add(flow.host % 8); - IpAddr::V4(Ipv4Addr::from(o)) - } - IpAddr::V6(a) => { - let mut o = a.octets(); - o[15] = o[15].wrapping_add(flow.host % 8); - IpAddr::V6(Ipv6Addr::from(o)) - } + settled(|| { + let Some(mut fabric) = Fabric::build(exposes) else { + return; }; - let dst = peer(src); + let privates = private_addresses(exposes); + if privates.is_empty() { + return; + } - let Some(mut request) = udp(src, dst, flow.sport, flow.dport) else { - continue; - }; - arrive(&mut request, local()); - let out = fabric.send(request); - - let Verdict::Forwarded { - src: public_src, - dst: reached, - .. - } = verdict(&out) - else { - NOT_FORWARDED.fetch_add(1, Ordering::Relaxed); - continue; - }; - let (Some(public_src), Some(reached)) = (public_src, reached) else { - continue; - }; - let public_port = out - .transport_src_port() - .unwrap_or_else(|| unreachable!("a udp packet has a source port")) - .get(); + for flow in flows { + let prefix = privates[usize::from(flow.prefix) % privates.len()]; + let src = match prefix.as_address() { + IpAddr::V4(a) => { + let mut o = a.octets(); + o[3] = o[3].wrapping_add(flow.host % 8); + IpAddr::V4(Ipv4Addr::from(o)) + } + IpAddr::V6(a) => { + let mut o = a.octets(); + o[15] = o[15].wrapping_add(flow.host % 8); + IpAddr::V6(Ipv6Addr::from(o)) + } + }; + let dst = peer(src); - let Some(mut reply) = udp(reached, public_src, flow.dport, public_port) else { - continue; - }; - arrive(&mut reply, remote()); - let back = fabric.send(reply); + let Some(mut request) = udp(src, dst, flow.sport, flow.dport) else { + continue; + }; + arrive(&mut request, local()); + let out = fabric.send(request); + + let Verdict::Forwarded { + src: public_src, + dst: reached, + .. + } = verdict(&out) + else { + NOT_FORWARDED.fetch_add(1, Ordering::Relaxed); + continue; + }; + let (Some(public_src), Some(reached)) = (public_src, reached) else { + continue; + }; + let public_port = out + .transport_src_port() + .unwrap_or_else(|| unreachable!("a udp packet has a source port")) + .get(); - match verdict(&back) { - Verdict::Forwarded { src: s, dst: d, .. } => { - assert_eq!( - d, - Some(src), - "the reply did not come back to the host that sent the request" - ); - assert_eq!(s, Some(dst), "the reply's source was rewritten"); - assert_eq!( - back.transport_dst_port().map(std::num::NonZero::get), - Some(flow.sport), - "the reply did not get the original source port back" - ); - ROUND_TRIPPED.fetch_add(1, Ordering::Relaxed); - } - Verdict::Dropped(reason) => panic!( - "the reply of a forwarded flow was dropped: {reason:?} \ + let Some(mut reply) = udp(reached, public_src, flow.dport, public_port) + else { + continue; + }; + arrive(&mut reply, remote()); + let back = fabric.send(reply); + + match verdict(&back) { + Verdict::Forwarded { src: s, dst: d, .. } => { + assert_eq!( + d, + Some(src), + "the reply did not come back to the host that sent the request" + ); + assert_eq!(s, Some(dst), "the reply's source was rewritten"); + assert_eq!( + back.transport_dst_port().map(std::num::NonZero::get), + Some(flow.sport), + "the reply did not get the original source port back" + ); + ROUND_TRIPPED.fetch_add(1, Ordering::Relaxed); + } + Verdict::Dropped(reason) => panic!( + "the reply of a forwarded flow was dropped: {reason:?} \ (request {src} -> {dst} became {public_src}:{public_port})" - ), - Verdict::Delivered { .. } => { - unreachable!("the overlay slice has no egress stage") + ), + Verdict::Delivered { .. } => { + unreachable!("the overlay slice has no egress stage") + } } } - } + }); }); let round_tripped = ROUND_TRIPPED.load(Ordering::Relaxed); @@ -1894,80 +1960,84 @@ mod acl { super::assert_within_budget("acl::Batch", &Batch); } - #[tokio::test] - #[dpdk::with_eal] - async fn the_acl_verdict_follows_the_protocol_the_packet_carries() { + #[test] + fn the_acl_verdict_follows_the_protocol_the_packet_carries() { static DENIED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static PERMITTED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static BEHIND_EXT: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static PERMITTED_OUT: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static DENIED_BY_ACL: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + let _eal = dpdk::test_support::start_eal(); + bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Batch) .for_each(|(exposes, default_allow, rule_proto, packets)| { - let default = if *default_allow { - AclAction::Allow - } else { - AclAction::Deny - }; - let rule = rule_proto.as_match(); - let Some(mut fabric) = - Fabric::build_with_acl(exposes, Some(&peering_acl(default, rule))) - else { - return; - }; - let Some(private) = exposes - .iter() - .flat_map(|e| e.ips.iter().map(PrefixWithOptionalPorts::prefix)) - .next() - .map(|p: Prefix| p.as_address()) - else { - return; - }; - let v6 = private.is_ipv6(); - let dst = peer(private); - - for (spec, headers) in packets { - let Some(mut packet) = wire(headers, *spec, private, dst) else { - continue; + settled(|| { + let default = if *default_allow { + AclAction::Allow + } else { + AclAction::Deny + }; + let rule = rule_proto.as_match(); + let Some(mut fabric) = + Fabric::build_with_acl(exposes, Some(&peering_acl(default, rule))) + else { + return; + }; + let Some(private) = exposes + .iter() + .flat_map(|e| e.ips.iter().map(PrefixWithOptionalPorts::prefix)) + .next() + .map(|p: Prefix| p.as_address()) + else { + return; }; - arrive(&mut packet, local()); - let out = fabric.send(packet); + let v6 = private.is_ipv6(); + let dst = peer(private); - let permitted = rule_matches(rule, carried(spec.proto, v6)) != *default_allow; - if spec.behind_extension { - BEHIND_EXT.fetch_add(1, Ordering::Relaxed); - } + for (spec, headers) in packets { + let Some(mut packet) = wire(headers, *spec, private, dst) else { + continue; + }; + arrive(&mut packet, local()); + let out = fabric.send(packet); - let seen = verdict(&out); - let acl_dropped = seen == Verdict::Dropped(DoneReason::AclDropped); - let forwarded = matches!(seen, Verdict::Forwarded { .. }); - if permitted { - assert!( - !acl_dropped, - "the acl dropped a {:?} packet it permits (rule={rule:?} \ - default={default:?} behind_extension={})", - spec.proto, spec.behind_extension - ); - PERMITTED.fetch_add(1, Ordering::Relaxed); - if forwarded { - PERMITTED_OUT.fetch_add(1, Ordering::Relaxed); + let permitted = + rule_matches(rule, carried(spec.proto, v6)) != *default_allow; + if spec.behind_extension { + BEHIND_EXT.fetch_add(1, Ordering::Relaxed); } - } else { - assert!( - !forwarded, - "a {:?} packet the acl denies was forwarded (rule={rule:?} \ + + let seen = verdict(&out); + let acl_dropped = seen == Verdict::Dropped(DoneReason::AclDropped); + let forwarded = matches!(seen, Verdict::Forwarded { .. }); + if permitted { + assert!( + !acl_dropped, + "the acl dropped a {:?} packet it permits (rule={rule:?} \ default={default:?} behind_extension={})", - spec.proto, spec.behind_extension - ); - DENIED.fetch_add(1, Ordering::Relaxed); - if acl_dropped { - DENIED_BY_ACL.fetch_add(1, Ordering::Relaxed); + spec.proto, spec.behind_extension + ); + PERMITTED.fetch_add(1, Ordering::Relaxed); + if forwarded { + PERMITTED_OUT.fetch_add(1, Ordering::Relaxed); + } + } else { + assert!( + !forwarded, + "a {:?} packet the acl denies was forwarded (rule={rule:?} \ + default={default:?} behind_extension={})", + spec.proto, spec.behind_extension + ); + DENIED.fetch_add(1, Ordering::Relaxed); + if acl_dropped { + DENIED_BY_ACL.fetch_add(1, Ordering::Relaxed); + } } } - } + }); }); let (permitted, permitted_out, denied, denied_by_acl, behind) = ( @@ -2227,71 +2297,74 @@ mod port_forward { true } - #[tokio::test] - #[dpdk::with_eal] - async fn a_forwarded_port_reaches_the_host_behind_it() { + #[test] + fn a_forwarded_port_reaches_the_host_behind_it() { static FORWARDED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static ANSWERED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static REFUSED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + let _eal = dpdk::test_support::start_eal(); + bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Reaches) .for_each(|reaches| { - let Some(mut fabric) = Fabric::routed(&[expose()], None) else { - unreachable!("the port-forwarding fixture does not configure") - }; - - for reach in reaches { - let external: IpAddr = format!("172.16.5.{}", reach.host) - .parse() - .unwrap_or_else(|_| unreachable!()); - let dport = if reach.past_the_range { - EXTERNAL_PORT + PORTS + (reach.port % PORTS) - } else { - EXTERNAL_PORT + reach.port + settled(|| { + let Some(mut fabric) = Fabric::routed(&[expose()], None) else { + unreachable!("the port-forwarding fixture does not configure") }; - let Some(inbound) = udp(outside(), external, reach.src_port, dport) else { - continue; - }; - let out = fabric.send(tunnelled_from(vni(REMOTE_VNI), &inbound)); + for reach in reaches { + let external: IpAddr = format!("172.16.5.{}", reach.host) + .parse() + .unwrap_or_else(|_| unreachable!()); + let dport = if reach.past_the_range { + EXTERNAL_PORT + PORTS + (reach.port % PORTS) + } else { + EXTERNAL_PORT + reach.port + }; - if reach.past_the_range { - assert!( - !matches!(verdict(&out), Verdict::Delivered { .. }), - "a packet to {external}:{dport}, past the declared range, was \ + let Some(inbound) = udp(outside(), external, reach.src_port, dport) else { + continue; + }; + let out = fabric.send(tunnelled_from(vni(REMOTE_VNI), &inbound)); + + if reach.past_the_range { + assert!( + !matches!(verdict(&out), Verdict::Delivered { .. }), + "a packet to {external}:{dport}, past the declared range, was \ forwarded anyway" - ); - REFUSED.fetch_add(1, Ordering::Relaxed); - continue; - } + ); + REFUSED.fetch_add(1, Ordering::Relaxed); + continue; + } - assert!( - matches!(verdict(&out), Verdict::Delivered { .. }), - "a packet to the declared {external}:{dport} was not forwarded: {:?}", - verdict(&out) - ); - let arrived = inside(&out).expect("a forwarded packet was not tunnelled"); - let expected_host: IpAddr = format!("10.0.5.{}", reach.host) - .parse() - .unwrap_or_else(|_| unreachable!()); - assert_eq!( - arrived.ip_destination(), - Some(expected_host), - "{external}:{dport} reached the wrong host" - ); - assert_eq!( - arrived.transport_dst_port().map(std::num::NonZero::get), - Some(INTERNAL_PORT + reach.port), - "{external}:{dport} reached the right host on the wrong port" - ); - FORWARDED.fetch_add(1, Ordering::Relaxed); + assert!( + matches!(verdict(&out), Verdict::Delivered { .. }), + "a packet to the declared {external}:{dport} was not forwarded: {:?}", + verdict(&out) + ); + let arrived = inside(&out).expect("a forwarded packet was not tunnelled"); + let expected_host: IpAddr = format!("10.0.5.{}", reach.host) + .parse() + .unwrap_or_else(|_| unreachable!()); + assert_eq!( + arrived.ip_destination(), + Some(expected_host), + "{external}:{dport} reached the wrong host" + ); + assert_eq!( + arrived.transport_dst_port().map(std::num::NonZero::get), + Some(INTERNAL_PORT + reach.port), + "{external}:{dport} reached the right host on the wrong port" + ); + FORWARDED.fetch_add(1, Ordering::Relaxed); - if answers(&mut fabric, expected_host, *reach, external, dport) { - ANSWERED.fetch_add(1, Ordering::Relaxed); + if answers(&mut fabric, expected_host, *reach, external, dport) { + ANSWERED.fetch_add(1, Ordering::Relaxed); + } } - } + }); }); let (forwarded, answered, refused) = ( @@ -2380,71 +2453,74 @@ mod interleaved { super::assert_within_budget("interleaved::Interleaving", &Interleaving); } - #[tokio::test] - #[dpdk::with_eal] - async fn interleaved_traffic_is_each_satisfied() { + #[test] + fn interleaved_traffic_is_each_satisfied() { static CHECKED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static ABANDONED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static MIXED_LOADS: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static MIXED_KINDS: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + let _eal = dpdk::test_support::start_eal(); + bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Interleaving) .for_each(|(senders, schedule)| { - let Some(mut fabric) = Fabric::routed(&exposes(), None) else { - return; - }; - - let dst: IpAddr = "3.3.3.1".parse().unwrap_or_else(|_| unreachable!()); - let mut kinds = Vec::new(); - let mut loads: Vec> = Vec::new(); - for (i, sender) in senders.iter().enumerate() { - let Ok(src) = format!("1.1.{i}.{}", sender.host).parse::() else { - continue; + settled(|| { + let Some(mut fabric) = Fabric::routed(&exposes(), None) else { + return; }; - kinds.push(sender.kind); - loads.push(match sender.kind { - Kind::Conversation => Box::new(Conversation::new( - Path::fixture(), - src, - dst, - sender.sport, - sender.dport, - )), - Kind::Blast => Box::new(Blast::new( - Path::fixture(), - src, - dst, - sender.sport, - sender.dport, - sender.count, - )) as Box, - }); - } - for burst in run_schedule(fabric.worker(), &mut loads, schedule) { - let mut loads_in: Vec = burst.clone(); - loads_in.sort_unstable(); - loads_in.dedup(); - if loads_in.len() > 1 { - MIXED_LOADS.fetch_add(1, Ordering::Relaxed); + let dst: IpAddr = "3.3.3.1".parse().unwrap_or_else(|_| unreachable!()); + let mut kinds = Vec::new(); + let mut loads: Vec> = Vec::new(); + for (i, sender) in senders.iter().enumerate() { + let Ok(src) = format!("1.1.{i}.{}", sender.host).parse::() else { + continue; + }; + kinds.push(sender.kind); + loads.push(match sender.kind { + Kind::Conversation => Box::new(Conversation::new( + Path::fixture(), + src, + dst, + sender.sport, + sender.dport, + )), + Kind::Blast => Box::new(Blast::new( + Path::fixture(), + src, + dst, + sender.sport, + sender.dport, + sender.count, + )) as Box, + }); } - let mut kinds_in: Vec = burst.iter().map(|i| kinds[*i]).collect(); - kinds_in.sort_unstable_by_key(|k| format!("{k:?}")); - kinds_in.dedup(); - if kinds_in.len() > 1 { - MIXED_KINDS.fetch_add(1, Ordering::Relaxed); + + for burst in run_schedule(fabric.worker(), &mut loads, schedule) { + let mut loads_in: Vec = burst.clone(); + loads_in.sort_unstable(); + loads_in.dedup(); + if loads_in.len() > 1 { + MIXED_LOADS.fetch_add(1, Ordering::Relaxed); + } + let mut kinds_in: Vec = burst.iter().map(|i| kinds[*i]).collect(); + kinds_in.sort_unstable_by_key(|k| format!("{k:?}")); + kinds_in.dedup(); + if kinds_in.len() > 1 { + MIXED_KINDS.fetch_add(1, Ordering::Relaxed); + } } - } - for load in &loads { - if load.checked() { - CHECKED.fetch_add(1, Ordering::Relaxed); - } else { - ABANDONED.fetch_add(1, Ordering::Relaxed); + for load in &loads { + if load.checked() { + CHECKED.fetch_add(1, Ordering::Relaxed); + } else { + ABANDONED.fetch_add(1, Ordering::Relaxed); + } } - } + }); }); let (checked, abandoned, mixed_loads, mixed_kinds) = ( @@ -2564,9 +2640,8 @@ mod offers { super::assert_within_budget("offers::Offered", &Offered); } - #[tokio::test] - #[dpdk::with_eal] - async fn a_configuration_carries_everything_it_offers() { + #[test] + fn a_configuration_carries_everything_it_offers() { static CHECKED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static ABANDONED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static DERIVED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); @@ -2574,43 +2649,47 @@ mod offers { static INBOUND: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static OUTBOUND: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + let _eal = dpdk::test_support::start_eal(); + let overlay = overlay(); bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Offered) .for_each(|(vary, schedule)| { - let mut fabric = Fabric::routed_over_validated( - &overlay, - topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]), - ); + settled(|| { + let mut fabric = Fabric::routed_over_validated( + &overlay, + topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]), + ); - let mut loads = loads_for(&overlay, vary); - DERIVED.fetch_add(loads.len() as u64, Ordering::Relaxed); - for load in &loads { - if load.describe().starts_with("[inbound") { - INBOUND.fetch_add(1, Ordering::Relaxed); - } else { - OUTBOUND.fetch_add(1, Ordering::Relaxed); + let mut loads = loads_for(&overlay, vary); + DERIVED.fetch_add(loads.len() as u64, Ordering::Relaxed); + for load in &loads { + if load.describe().starts_with("[inbound") { + INBOUND.fetch_add(1, Ordering::Relaxed); + } else { + OUTBOUND.fetch_add(1, Ordering::Relaxed); + } } - } - for burst in run_schedule(fabric.worker(), &mut loads, schedule) { - let mut seen = burst.clone(); - seen.sort_unstable(); - seen.dedup(); - if seen.len() > 1 { - MIXED.fetch_add(1, Ordering::Relaxed); + for burst in run_schedule(fabric.worker(), &mut loads, schedule) { + let mut seen = burst.clone(); + seen.sort_unstable(); + seen.dedup(); + if seen.len() > 1 { + MIXED.fetch_add(1, Ordering::Relaxed); + } } - } - for load in &loads { - if load.checked() { - CHECKED.fetch_add(1, Ordering::Relaxed); - } else { - ABANDONED.fetch_add(1, Ordering::Relaxed); + for load in &loads { + if load.checked() { + CHECKED.fetch_add(1, Ordering::Relaxed); + } else { + ABANDONED.fetch_add(1, Ordering::Relaxed); + } } - } + }); }); let (checked, abandoned, derived, mixed) = ( @@ -2669,6 +2748,10 @@ mod generated { static BY_FLOW: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static EXCEPTING: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static AGED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static AGED_MILLIS: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static SURVIVED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + fn report_and_assert_coverage() { let (checked, derived, mixed) = ( CHECKED.load(Ordering::Relaxed), @@ -2794,94 +2877,116 @@ mod generated { } } - #[tokio::test] - #[dpdk::with_eal] - async fn a_generated_configuration_carries_its_own_traffic() { + #[test] + fn a_generated_configuration_carries_its_own_traffic() { + let _eal = dpdk::test_support::start_eal(); bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Generated) .for_each(|(ops, vary, schedule)| { - let draft = Sequence::fold(ops); - let overlay = draft - .overlay() - .unwrap_or_else(|e| panic!("{ops:?} does not assemble: {e}")); - let validated = overlay - .validate() - .unwrap_or_else(|e| panic!("{ops:?} does not validate: {e}")); + settled(|| { + let draft = Sequence::fold(ops); + let overlay = draft + .overlay() + .unwrap_or_else(|e| panic!("{ops:?} does not assemble: {e}")); + let validated = overlay + .validate() + .unwrap_or_else(|e| panic!("{ops:?} does not validate: {e}")); + + let vnis: Vec = validated + .vpc_table() + .values() + .map(config::external::overlay::vpc::ValidatedVpc::vni) + .collect(); + if vnis.is_empty() { + return; + } + if validated.vpc_table().peerings().next().is_some() { + PEERED.fetch_add(1, Ordering::Relaxed); + } + if vnis.len() > 2 { + MULTI.fetch_add(1, Ordering::Relaxed); + } - let vnis: Vec = validated - .vpc_table() - .values() - .map(config::external::overlay::vpc::ValidatedVpc::vni) - .collect(); - if vnis.is_empty() { - return; - } - if validated.vpc_table().peerings().next().is_some() { - PEERED.fetch_add(1, Ordering::Relaxed); - } - if vnis.len() > 2 { - MULTI.fetch_add(1, Ordering::Relaxed); - } + let mut fabric = Fabric::routed_over_validated(&validated, topology(&vnis)); - let mut fabric = Fabric::routed_over_validated(&validated, topology(&vnis)); - - let (permitting, by_flow, excepting) = (Cell::new(0), Cell::new(0), Cell::new(0)); - let mut loads = loads_where( - &validated, - vary, - &carried_counting(&draft, &permitting, &by_flow, &excepting), - ); - PERMITTING.fetch_add(permitting.get(), Ordering::Relaxed); - BY_FLOW.fetch_add(by_flow.get(), Ordering::Relaxed); - EXCEPTING.fetch_add(excepting.get(), Ordering::Relaxed); - DERIVED.fetch_add(loads.len() as u64, Ordering::Relaxed); - for load in &loads { - if load.describe().starts_with("[inbound") { - INBOUND.fetch_add(1, Ordering::Relaxed); + let (permitting, by_flow, excepting) = + (Cell::new(0), Cell::new(0), Cell::new(0)); + let mut loads = loads_where( + &validated, + vary, + &carried_counting(&draft, &permitting, &by_flow, &excepting), + ); + PERMITTING.fetch_add(permitting.get(), Ordering::Relaxed); + BY_FLOW.fetch_add(by_flow.get(), Ordering::Relaxed); + EXCEPTING.fetch_add(excepting.get(), Ordering::Relaxed); + DERIVED.fetch_add(loads.len() as u64, Ordering::Relaxed); + for load in &loads { + if load.describe().starts_with("[inbound") { + INBOUND.fetch_add(1, Ordering::Relaxed); + } } - } - for burst in run_schedule(fabric.worker(), &mut loads, schedule) { - let mut seen = burst.clone(); - seen.sort_unstable(); - seen.dedup(); - if seen.len() > 1 { - MIXED.fetch_add(1, Ordering::Relaxed); + for burst in run_schedule(fabric.worker(), &mut loads, schedule) { + let mut seen = burst.clone(); + seen.sort_unstable(); + seen.dedup(); + if seen.len() > 1 { + MIXED.fetch_add(1, Ordering::Relaxed); + } } - } - for load in &loads { - assert!( - load.checked(), - "a load derived from the configuration did not complete: {}", - load.describe() - ); - CHECKED.fetch_add(1, Ordering::Relaxed); - } + for load in &loads { + assert!( + load.checked(), + "a load derived from the configuration did not complete: {}", + load.describe() + ); + CHECKED.fetch_add(1, Ordering::Relaxed); + } + }); }); report_and_assert_coverage(); } - #[tokio::test] - #[dpdk::with_eal] - async fn a_configuration_carries_nothing_it_denies() { - static SENT: LazyLock = LazyLock::new(|| AtomicU64::new(0)); - static BY_ACL: LazyLock = LazyLock::new(|| AtomicU64::new(0)); - static NARROWED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); - static CONFIGS: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + pub(super) struct OverTime; + + impl ValueGenerator for OverTime { + type Output = (Vec, Vec, Vec, Vec); + + fn generate(&self, driver: &mut D) -> Option { + let (ops, vary, schedule) = Generated.generate(driver)?; + let cap = + (Masquerade::MASQUERADE_ONEWAY_TIMEOUT / 2).as_millis() / (POLLS as u128).max(1); + let cap = u64::try_from(cap).unwrap_or(u64::MAX).max(1); + let waits = (0..POLLS) + .map(|_| { + Some(Duration::from_millis( + driver.gen_u64(Included(&0), Included(&cap))?, + )) + }) + .collect::>>()?; + Some((ops, vary, schedule, waits)) + } + } + + #[test] + fn time_passing_does_not_disturb_a_flow_inside_its_lifetime() { + let _eal = dpdk::test_support::start_eal(); + let clock = clock::virtual_time::Paused::new(); bolero::check!() .with_max_len(MAX_INPUT_LEN) - .with_generator(Generated) - .for_each(|(ops, vary, _schedule)| { + .with_generator(OverTime) + .for_each(|(ops, vary, schedule, waits)| { let draft = Sequence::fold(ops); - let validated = draft - .overlay() - .unwrap_or_else(|e| panic!("{ops:?} does not assemble: {e}")) - .validate() - .unwrap_or_else(|e| panic!("{ops:?} does not validate: {e}")); + let Ok(overlay) = draft.overlay() else { + return; + }; + let Ok(validated) = overlay.validate() else { + return; + }; let vnis: Vec = validated .vpc_table() @@ -2891,41 +2996,112 @@ mod generated { if vnis.is_empty() { return; } - let carried = super::derive::carried_by(&draft); - let narrowed = Cell::new(0); - let mut loads = loads_where(&validated, vary, &|named| { - if carried(named) { - return false; - } - if draft.guard_named(named.peering) != Some(Guard::Deny) { - narrowed.set(narrowed.get() + 1); - } - true - }); + + let mut fabric = Fabric::routed_over_validated(&validated, topology(&vnis)); + let mut loads = loads_where(&validated, vary, &derive::carried_by(&draft)); if loads.is_empty() { return; } - NARROWED.fetch_add(narrowed.get(), Ordering::Relaxed); - CONFIGS.fetch_add(1, Ordering::Relaxed); - let mut fabric = Fabric::routed_over_validated(&validated, topology(&vnis)); - for load in &mut loads { - let Some(packet) = load.next() else { - continue; - }; - let seen = verdict(&fabric.worker().send(packet)); - SENT.fetch_add(1, Ordering::Relaxed); - assert!( - matches!(seen, Verdict::Dropped(_)), - "an acl that refuses this traffic produced {seen:?} for {}", - load.describe() + let moved: Duration = waits.iter().sum(); + clock.block_on(async { + run_schedule_over_time(fabric.worker(), &mut loads, schedule, waits).await; + }); + + if moved > Duration::ZERO { + AGED.fetch_add(1, Ordering::Relaxed); + AGED_MILLIS.fetch_add( + u64::try_from(moved.as_millis()).unwrap_or(u64::MAX), + Ordering::Relaxed, ); - if seen == Verdict::Dropped(DoneReason::AclDropped) { - BY_ACL.fetch_add(1, Ordering::Relaxed); + for load in &loads { + if load.checked() { + SURVIVED.fetch_add(1, Ordering::Relaxed); + } } } }); + let (aged, millis, survived) = ( + AGED.load(Ordering::Relaxed), + AGED_MILLIS.load(Ordering::Relaxed), + SURVIVED.load(Ordering::Relaxed), + ); + println!("aged={aged} cases, {millis}ms of virtual time, survived={survived} loads"); + super::assert_covered( + aged > 0, + "the clock never moved, so this property checked the same thing as the one above it", + ); + super::assert_covered( + survived > 0, + "no load ever made its claim with the clock moving under it, so nothing was aged", + ); + } + + #[test] + fn a_configuration_carries_nothing_it_denies() { + static SENT: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static BY_ACL: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static NARROWED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static CONFIGS: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + let _eal = dpdk::test_support::start_eal(); + + bolero::check!() + .with_max_len(MAX_INPUT_LEN) + .with_generator(Generated) + .for_each(|(ops, vary, _schedule)| { + settled(|| { + let draft = Sequence::fold(ops); + let validated = draft + .overlay() + .unwrap_or_else(|e| panic!("{ops:?} does not assemble: {e}")) + .validate() + .unwrap_or_else(|e| panic!("{ops:?} does not validate: {e}")); + + let vnis: Vec = validated + .vpc_table() + .values() + .map(config::external::overlay::vpc::ValidatedVpc::vni) + .collect(); + if vnis.is_empty() { + return; + } + let carried = super::derive::carried_by(&draft); + let narrowed = Cell::new(0); + let mut loads = loads_where(&validated, vary, &|named| { + if carried(named) { + return false; + } + if draft.guard_named(named.peering) != Some(Guard::Deny) { + narrowed.set(narrowed.get() + 1); + } + true + }); + if loads.is_empty() { + return; + } + NARROWED.fetch_add(narrowed.get(), Ordering::Relaxed); + CONFIGS.fetch_add(1, Ordering::Relaxed); + + let mut fabric = Fabric::routed_over_validated(&validated, topology(&vnis)); + for load in &mut loads { + let Some(packet) = load.next() else { + continue; + }; + let seen = verdict(&fabric.worker().send(packet)); + SENT.fetch_add(1, Ordering::Relaxed); + assert!( + matches!(seen, Verdict::Dropped(_)), + "an acl that refuses this traffic produced {seen:?} for {}", + load.describe() + ); + if seen == Verdict::Dropped(DoneReason::AclDropped) { + BY_ACL.fetch_add(1, Ordering::Relaxed); + } + } + }); + }); + let (configs, sent, by_acl) = ( CONFIGS.load(Ordering::Relaxed), SENT.load(Ordering::Relaxed), @@ -2952,52 +3128,52 @@ mod generated { ); } - #[tokio::test] - #[dpdk::with_eal] - async fn an_excluded_address_is_not_reachable() { + #[test] + fn an_excluded_address_is_not_reachable() { static AIMED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static REFUSED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + let _eal = dpdk::test_support::start_eal(); bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Generated) - .for_each(|(ops, vary, _schedule)| { - let draft = Sequence::fold(ops); - let validated = draft - .overlay() - .unwrap_or_else(|e| panic!("{ops:?} does not assemble: {e}")) - .validate() - .unwrap_or_else(|e| panic!("{ops:?} does not validate: {e}")); - - let vnis: Vec = validated - .vpc_table() - .values() - .map(config::external::overlay::vpc::ValidatedVpc::vni) - .collect(); - if vnis.is_empty() { - return; - } - let probes = super::derive::probes_for(&validated, vary, &draft); - if probes.is_empty() { - return; - } + .for_each(|(ops, vary, _schedule)| settled(|| { + let draft = Sequence::fold(ops); + let validated = draft + .overlay() + .unwrap_or_else(|e| panic!("{ops:?} does not assemble: {e}")) + .validate() + .unwrap_or_else(|e| panic!("{ops:?} does not validate: {e}")); + + let vnis: Vec = validated + .vpc_table() + .values() + .map(config::external::overlay::vpc::ValidatedVpc::vni) + .collect(); + if vnis.is_empty() { + return; + } + let probes = super::derive::probes_for(&validated, vary, &draft); + if probes.is_empty() { + return; + } - let mut fabric = Fabric::routed_over_validated(&validated, topology(&vnis)); - for probe in probes { - let Some(packet) = probe.packet() else { - continue; - }; - let seen = verdict(&fabric.worker().send(packet)); - AIMED.fetch_add(1, Ordering::Relaxed); - assert!( - matches!(seen, Verdict::Dropped(_)), - "an address the configuration excludes was reached: {seen:?} for {probe:?}" - ); - if seen == Verdict::Dropped(DoneReason::Filtered) { - REFUSED.fetch_add(1, Ordering::Relaxed); + let mut fabric = Fabric::routed_over_validated(&validated, topology(&vnis)); + for probe in probes { + let Some(packet) = probe.packet() else { + continue; + }; + let seen = verdict(&fabric.worker().send(packet)); + AIMED.fetch_add(1, Ordering::Relaxed); + assert!( + matches!(seen, Verdict::Dropped(_)), + "an address the configuration excludes was reached: {seen:?} for {probe:?}" + ); + if seen == Verdict::Dropped(DoneReason::Filtered) { + REFUSED.fetch_add(1, Ordering::Relaxed); + } } - } - }); + })); let (aimed, refused) = ( AIMED.load(Ordering::Relaxed), @@ -3080,15 +3256,16 @@ mod burst { } } - #[tokio::test] - #[dpdk::with_eal] - async fn a_burst_of_one_flow_allocates_once() { + #[test] + fn a_burst_of_one_flow_allocates_once() { static CHECKED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + let _eal = dpdk::test_support::start_eal(); + bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Burst) - .for_each(|members| { + .for_each(|members| settled(|| { let m = members[0]; let src: IpAddr = format!("1.1.0.{}", m.host) .parse() @@ -3138,16 +3315,15 @@ mod burst { one packet of it did" ); CHECKED.fetch_add(1, Ordering::Relaxed); - }); + })); let checked = CHECKED.load(Ordering::Relaxed); eprintln!("single-flow-bursts={checked}"); super::assert_covered(checked > 0, "no burst of a single flow was ever delivered"); } - #[tokio::test] - #[dpdk::with_eal] - async fn a_burst_of_one_translated_flow_allocates_once() { + #[test] + fn a_burst_of_one_translated_flow_allocates_once() { static CHECKED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); fn two_sided() -> Option { @@ -3172,10 +3348,12 @@ mod burst { let overlay = two_sided().expect("a valid two-sided configuration"); + let _eal = dpdk::test_support::start_eal(); + bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Burst) - .for_each(|members| { + .for_each(|members| settled(|| { let m = members[0]; let src: IpAddr = format!("1.1.0.{}", m.host) .parse() @@ -3236,7 +3414,7 @@ mod burst { flow-table entries than one packet of it did" ); CHECKED.fetch_add(1, Ordering::Relaxed); - }); + })); let checked = CHECKED.load(Ordering::Relaxed); eprintln!("translated-single-flow-bursts={checked}"); @@ -3246,62 +3424,67 @@ mod burst { ); } - #[tokio::test] - #[dpdk::with_eal] - async fn a_burst_is_treated_the_same_as_one_packet_at_a_time() { + #[test] + fn a_burst_is_treated_the_same_as_one_packet_at_a_time() { static COMPARED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static DELIVERED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + let _eal = dpdk::test_support::start_eal(); + bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Burst) .for_each(|members| { - let packets = || { - members - .iter() - .enumerate() - .map(|(i, m)| { - let src: IpAddr = format!("1.1.{i}.{}", m.host) - .parse() - .unwrap_or_else(|_| unreachable!()); - let dst: IpAddr = "3.3.3.1".parse().unwrap_or_else(|_| unreachable!()); - udp(src, dst, 1024 + u16::try_from(i).unwrap_or(0), m.dport) - .map(|p| tunnelled(&p)) - }) - .collect::>>() - }; - let (Some(singly), Some(together)) = (packets(), packets()) else { - return; - }; + settled(|| { + let packets = || { + members + .iter() + .enumerate() + .map(|(i, m)| { + let src: IpAddr = format!("1.1.{i}.{}", m.host) + .parse() + .unwrap_or_else(|_| unreachable!()); + let dst: IpAddr = + "3.3.3.1".parse().unwrap_or_else(|_| unreachable!()); + udp(src, dst, 1024 + u16::try_from(i).unwrap_or(0), m.dport) + .map(|p| tunnelled(&p)) + }) + .collect::>>() + }; + let (Some(singly), Some(together)) = (packets(), packets()) else { + return; + }; - let (Some(mut a), Some(mut b)) = ( - Fabric::routed(&exposes(), None), - Fabric::routed(&exposes(), None), - ) else { - return; - }; + let (Some(mut a), Some(mut b)) = ( + Fabric::routed(&exposes(), None), + Fabric::routed(&exposes(), None), + ) else { + return; + }; - let one_at_a_time: Vec<_> = - singly.into_iter().map(|p| treatment(&a.send(p))).collect(); - let in_a_burst: Vec<_> = b.send_batch(together).iter().map(treatment).collect(); + let one_at_a_time: Vec<_> = + singly.into_iter().map(|p| treatment(&a.send(p))).collect(); + let in_a_burst: Vec<_> = b.send_batch(together).iter().map(treatment).collect(); - assert_eq!( - one_at_a_time.len(), - in_a_burst.len(), - "a burst did not return as many packets as it was given" - ); - for (i, (alone, batched)) in one_at_a_time.iter().zip(in_a_burst.iter()).enumerate() - { assert_eq!( - alone, batched, - "packet {i} of the burst was treated differently from the same packet \ - sent on its own" + one_at_a_time.len(), + in_a_burst.len(), + "a burst did not return as many packets as it was given" ); - COMPARED.fetch_add(1, Ordering::Relaxed); - if matches!(alone.verdict, Verdict::Delivered { .. }) { - DELIVERED.fetch_add(1, Ordering::Relaxed); + for (i, (alone, batched)) in + one_at_a_time.iter().zip(in_a_burst.iter()).enumerate() + { + assert_eq!( + alone, batched, + "packet {i} of the burst was treated differently from the same packet \ + sent on its own" + ); + COMPARED.fetch_add(1, Ordering::Relaxed); + if matches!(alone.verdict, Verdict::Delivered { .. }) { + DELIVERED.fetch_add(1, Ordering::Relaxed); + } } - } + }); }); let (compared, delivered) = ( @@ -3373,71 +3556,74 @@ mod destination { super::assert_within_budget("destination::Aims", &Aims); } - #[tokio::test] - #[dpdk::with_eal] - async fn a_packet_leaves_for_the_vpc_that_exposes_its_destination() { + #[test] + fn a_packet_leaves_for_the_vpc_that_exposes_its_destination() { static REACHED: LazyLock<[AtomicU64; PEERS as usize]> = LazyLock::new(|| std::array::from_fn(|_| AtomicU64::new(0))); static REFUSED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + let _eal = dpdk::test_support::start_eal(); bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Aims) .for_each(|aims| { - let vnis: Vec<_> = std::iter::once(vni(LOCAL_VNI)) - .chain((0..PEERS).map(|n| vni(peer_vni(n)))) - .collect(); - let overlay = overlay_with_peers(local_prefix(), PEERS).unwrap_or_else(|e| { - unreachable!("the multi-peer contract does not build: {e}") - }); - let Some(mut fabric) = Fabric::routed_over(&overlay, topology(&vnis)) else { - unreachable!("the multi-peer contract does not validate") - }; + settled(|| { + let vnis: Vec<_> = std::iter::once(vni(LOCAL_VNI)) + .chain((0..PEERS).map(|n| vni(peer_vni(n)))) + .collect(); + let overlay = overlay_with_peers(local_prefix(), PEERS).unwrap_or_else(|e| { + unreachable!("the multi-peer contract does not build: {e}") + }); + let Some(mut fabric) = Fabric::routed_over(&overlay, topology(&vnis)) else { + unreachable!("the multi-peer contract does not validate") + }; - for aim in aims { - let src: IpAddr = format!("1.1.0.{}", aim.host) + for aim in aims { + let src: IpAddr = format!("1.1.0.{}", aim.host) + .parse() + .unwrap_or_else(|_| unreachable!()); + let dst: IpAddr = match aim.peer { + Some(n) => format!("10.{}.{}.{}", n + 1, aim.third, aim.host), + None => format!("172.16.{}.{}", aim.third, aim.host), + } .parse() .unwrap_or_else(|_| unreachable!()); - let dst: IpAddr = match aim.peer { - Some(n) => format!("10.{}.{}.{}", n + 1, aim.third, aim.host), - None => format!("172.16.{}.{}", aim.third, aim.host), - } - .parse() - .unwrap_or_else(|_| unreachable!()); - - let Some(packet) = udp(src, dst, aim.sport, aim.dport) else { - continue; - }; - let out = fabric.send(tunnelled(&packet)); - let left = matches!(verdict(&out), Verdict::Delivered { .. }); - if let Some(n) = aim.peer { - assert!( - left, - "a packet to {dst}, which peer {n} exposes, did not leave: {:?}", - verdict(&out) - ); - assert_eq!( - out.try_vxlan().map(net::vxlan::Vxlan::vni), - Some(vni(peer_vni(n))), - "a packet to {dst} left for the wrong vpc" - ); - let carried = inside(&out).expect("a delivered packet was not tunnelled"); - assert_eq!( - carried.ip_destination(), - Some(dst), - "the destination was rewritten on the way out" - ); - REACHED[n as usize].fetch_add(1, Ordering::Relaxed); - } else { - assert!( - !left, - "a packet to {dst}, which no peering covers, was sent to {:?}", - out.try_vxlan().map(net::vxlan::Vxlan::vni) - ); - REFUSED.fetch_add(1, Ordering::Relaxed); + let Some(packet) = udp(src, dst, aim.sport, aim.dport) else { + continue; + }; + let out = fabric.send(tunnelled(&packet)); + let left = matches!(verdict(&out), Verdict::Delivered { .. }); + + if let Some(n) = aim.peer { + assert!( + left, + "a packet to {dst}, which peer {n} exposes, did not leave: {:?}", + verdict(&out) + ); + assert_eq!( + out.try_vxlan().map(net::vxlan::Vxlan::vni), + Some(vni(peer_vni(n))), + "a packet to {dst} left for the wrong vpc" + ); + let carried = + inside(&out).expect("a delivered packet was not tunnelled"); + assert_eq!( + carried.ip_destination(), + Some(dst), + "the destination was rewritten on the way out" + ); + REACHED[n as usize].fetch_add(1, Ordering::Relaxed); + } else { + assert!( + !left, + "a packet to {dst}, which no peering covers, was sent to {:?}", + out.try_vxlan().map(net::vxlan::Vxlan::vni) + ); + REFUSED.fetch_add(1, Ordering::Relaxed); + } } - } + }); }); let reached: Vec = REACHED.iter().map(|c| c.load(Ordering::Relaxed)).collect(); @@ -3661,87 +3847,95 @@ mod routed { reparsed } - #[tokio::test] - #[dpdk::with_eal] - async fn a_tagged_shape_never_reaches_the_wire() { + #[test] + fn a_tagged_shape_never_reaches_the_wire() { static TAGGED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + let _eal = dpdk::test_support::start_eal(); + bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Batch) .for_each(|(exposes, stacks)| { - let Some(mut fabric) = Fabric::routed(exposes, None) else { - return; - }; - let private = exposes.first().and_then(|e| { - e.ips - .first() - .map(lpm::prefix::PrefixWithOptionalPorts::prefix) - }); - - for (shape, headers) in stacks { - let mut headers = headers.clone(); - aim(&mut headers, private); - let Some(frame) = wire(&headers) else { - continue; + settled(|| { + let Some(mut fabric) = Fabric::routed(exposes, None) else { + return; }; - let tagged = *shape == Shape::VlanV4Tcp; - if tagged { - TAGGED.fetch_add(1, Ordering::Relaxed); - } + let private = exposes.first().and_then(|e| { + e.ips + .first() + .map(lpm::prefix::PrefixWithOptionalPorts::prefix) + }); - let out = fabric.send(tunnelled(&frame)); - assert!( - !(tagged && matches!(verdict(&out), Verdict::Delivered { .. })), - "a tagged frame was sent out onto the wire" - ); - } + for (shape, headers) in stacks { + let mut headers = headers.clone(); + aim(&mut headers, private); + let Some(frame) = wire(&headers) else { + continue; + }; + let tagged = *shape == Shape::VlanV4Tcp; + if tagged { + TAGGED.fetch_add(1, Ordering::Relaxed); + } + + let out = fabric.send(tunnelled(&frame)); + assert!( + !(tagged && matches!(verdict(&out), Verdict::Delivered { .. })), + "a tagged frame was sent out onto the wire" + ); + } + }); }); let tagged = TAGGED.load(Ordering::Relaxed); eprintln!("tagged={tagged}"); - let mut control = Fabric::routed(&exposes(), None).expect("a valid configuration"); - assert!( - matches!( - verdict(&control.send(tunnelled(&inner()))), - Verdict::Delivered { .. } - ), - "the untagged control did not reach the wire, so no delivery was observable here" - ); + settled(|| { + let mut control = Fabric::routed(&exposes(), None).expect("a valid configuration"); + assert!( + matches!( + verdict(&control.send(tunnelled(&inner()))), + Verdict::Delivered { .. } + ), + "the untagged control did not reach the wire, so no delivery was observable here" + ); + }); super::assert_covered(tagged > 0, "no tagged shape was ever generated"); } - #[tokio::test] - #[dpdk::with_eal] - async fn a_tunnelled_flow_comes_back_through_the_tunnel() { + #[test] + fn a_tunnelled_flow_comes_back_through_the_tunnel() { static ROUND_TRIPPED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static ABANDONED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + let _eal = dpdk::test_support::start_eal(); + bolero::check!() .with_max_len(MAX_INPUT_LEN) .with_generator(Flows) .for_each(|flows| { - let Some(mut fabric) = Fabric::routed(&exposes(), None) else { - return; - }; + settled(|| { + let Some(mut fabric) = Fabric::routed(&exposes(), None) else { + return; + }; - for flow in flows { - let src: IpAddr = format!("1.1.0.{}", flow.host) - .parse() - .unwrap_or_else(|_| unreachable!()); - let dst: IpAddr = "3.3.3.1".parse().unwrap_or_else(|_| unreachable!()); - let mut load = - Conversation::new(Path::fixture(), src, dst, flow.sport, flow.dport); + for flow in flows { + let src: IpAddr = format!("1.1.0.{}", flow.host) + .parse() + .unwrap_or_else(|_| unreachable!()); + let dst: IpAddr = "3.3.3.1".parse().unwrap_or_else(|_| unreachable!()); + let mut load = + Conversation::new(Path::fixture(), src, dst, flow.sport, flow.dport); - drive(fabric.worker(), &mut load); + drive(fabric.worker(), &mut load); - if load.checked() { - ROUND_TRIPPED.fetch_add(1, Ordering::Relaxed); - } else { - ABANDONED.fetch_add(1, Ordering::Relaxed); + if load.checked() { + ROUND_TRIPPED.fetch_add(1, Ordering::Relaxed); + } else { + ABANDONED.fetch_add(1, Ordering::Relaxed); + } } - } + }); }); let round_tripped = ROUND_TRIPPED.load(Ordering::Relaxed); diff --git a/default.nix b/default.nix index c153120f62..f02de46771 100644 --- a/default.nix +++ b/default.nix @@ -31,12 +31,14 @@ let kernel ; }; - sanitizers = split-str ",+" sanitize; + as-set = str: lib.sort (a: b: a < b) (lib.unique (split-str ",+" str)); + sanitizers = as-set sanitize; + instrumentations = as-set instrumentation; cargo-features = split-str ",+" features; profile' = import ./nix/profiles.nix { inherit sanitizers - instrumentation + instrumentations profile cargo-features host-arch @@ -49,7 +51,7 @@ let profile-tests' = import ./nix/profiles.nix { inherit sanitizers - instrumentation + instrumentations profile cargo-features host-arch @@ -61,7 +63,7 @@ let { "debug" = "dev"; "release" = "release"; - "fuzz" = "fuzz"; + "checked" = "checked"; } .${profile}; overlays = import ./nix/overlays { @@ -69,6 +71,7 @@ let libc nightly sanitizers + instrumentations sources ; profile = profile'; @@ -88,8 +91,8 @@ let in if platform != "wasm32-wasip1" then over.pkgsCross.${platform'.info.nixarch} else over; sysroot-stamp = '' - printf '%s' '${sanitize}' > "$out/.sanitize" - printf '%s' '${instrumentation}' > "$out/.instrumentation" + printf '%s' '${builtins.concatStringsSep "," sanitizers}' > "$out/.sanitize" + printf '%s' '${builtins.concatStringsSep "," instrumentations}' > "$out/.instrumentation" ''; sysroot = if platform != "wasm32-wasip1" then @@ -651,7 +654,12 @@ let ++ cargo-cmd-prefix-tests )) # Record the remapped source root without changing normal archives. - + (if instrumentation == "coverage" then "; echo -n '${src-prefix}' > $out/source-prefix" else ""); + + ( + if builtins.elem "coverage" instrumentations then + "; echo -n '${src-prefix}' > $out/source-prefix" + else + "" + ); }; }; diff --git a/hardware/Cargo.toml b/hardware/Cargo.toml index edd376c090..f396747971 100644 --- a/hardware/Cargo.toml +++ b/hardware/Cargo.toml @@ -43,7 +43,7 @@ n-vm = { workspace = true } test-utils = { workspace = true, features = [] } # external -bolero = { workspace = true, features = ["alloc"] } +bolero = { workspace = true, features = ["std"] } hwlocality = { workspace = true, features = ["hwloc-latest"] } pci-ids = { workspace = true, features = [] } serde = { workspace = true, features = ["std"] } diff --git a/justfile b/justfile index 2f481f291f..ed49d266d3 100644 --- a/justfile +++ b/justfile @@ -205,10 +205,20 @@ fuzz target time="60s" *args="": # asan does not need that, and skipping the std rebuild keeps it far quicker. # `sanitize=NONE` drops instrumentation altogether, which buys roughly four times # the executions per second in exchange for only catching what the test asserts. + case "{{ sanitize }}" in + "") want=address ;; + NONE) want=none ;; + *) want="{{ sanitize }}" ;; + esac sysroot="${DATAPLANE_SYSROOT:-}" if [ -n "${sysroot}" ] && [ -r "${sysroot}/.sanitize" ]; then built_with="$(cat "${sysroot}/.sanitize")" - if [ "${built_with}" != "{{ sanitize }}" ]; then + built_with="${built_with:-none}" + if [ "${want}" != "${built_with}" ] && [ -z "{{ sanitize }}" ]; then + printf 'warning: rust is built with %s and this sysroot with %s, so the C dependencies -- dpdk above all -- are not instrumented.\n' \ + "${want}" "${built_with}" >&2 + printf ' `just sanitize=NONE fuzz ...` instruments neither and runs about four times quicker.\n' >&2 + elif [ "${want}" != "${built_with}" ]; then printf 'refusing to fuzz: sanitize=%s was asked for, but this sysroot was built with sanitize=%s.\n' \ "{{ sanitize }}" "${built_with:-}" >&2 printf 'the C dependencies would not be instrumented. Re-enter the shell with:\n' >&2 @@ -217,9 +227,20 @@ fuzz target time="60s" *args="": exit 1 fi fi + inherited="$(cargo config get -Zunstable-options --format json-value build.rustflags 2>/dev/null | jq -r 'join(" ")')" + export RUSTFLAGS="${inherited} ${RUSTFLAGS:-}" + + sancov_rt="$(clang -print-file-name=libclang_rt.fuzzer_no_main-$(uname -m).a 2>/dev/null || true)" + if [ -f "${sancov_rt}" ]; then + export RUSTFLAGS="${RUSTFLAGS} -Clink-arg=${sancov_rt} -Clink-arg=-lstdc++" + else + printf 'warning: no libFuzzer runtime beside clang; packages with a non-bolero test binary will not link.\n' >&2 + fi + corpus_dir="{{ fuzz_corpus_root }}/$(printf '%s' '{{ target }}' | tr -c 'A-Za-z0-9_.-' '_')" mkdir -p "${corpus_dir}" cargo bolero test '{{ target }}' --rustc-bootstrap -T '{{ time }}' \ + --profile checked \ --corpus-dir "${corpus_dir}" \ -l '{{ fuzz_max_input_length }}' \ -E='-len_control={{ fuzz_len_control }}' \ @@ -339,9 +360,19 @@ setup-roots *args: {{ args }} done +[private] +[script] +_refuse-instrumented-artifact: + if [ -n '{{ instrument }}' ] && [ '{{ instrument }}' != "none" ]; then + printf 'refusing to build a container at instrument=%s: an instrumented build is a diagnostic,\n' '{{ instrument }}' >&2 + printf 'not an artifact, and instrumentation is not part of the version -- so this image would\n' >&2 + printf 'take a clean image tag and replace it.\n' >&2 + exit 1 + fi + # Build the dataplane container image [script] -build-container target="dataplane" *args: (build (if target == "dataplane" { "dataplane.tar" } else if target == "validator" { "workspace.validator" } else { "containers." + target }) args) +build-container target="dataplane" *args: _refuse-instrumented-artifact (build (if target == "dataplane" { "dataplane.tar" } else if target == "validator" { "workspace.validator" } else { "containers." + target }) args) {{ _just_debuggable_ }} declare -xr DOCKER_HOST="${DOCKER_HOST:-unix://{{docker_sock}}}" case "{{target}}" in diff --git a/lpm/Cargo.toml b/lpm/Cargo.toml index 07ba003fed..ff58b5978d 100644 --- a/lpm/Cargo.toml +++ b/lpm/Cargo.toml @@ -21,5 +21,5 @@ thiserror = { workspace = true } tracing = { workspace = true } [dev-dependencies] -bolero = { workspace = true, default-features = false } +bolero = { workspace = true, features = ["std"] } serde_yaml_ng = { workspace = true } diff --git a/mgmt/src/tests/mgmt.rs b/mgmt/src/tests/mgmt.rs index 98e39671a0..800ebd5bb2 100644 --- a/mgmt/src/tests/mgmt.rs +++ b/mgmt/src/tests/mgmt.rs @@ -733,55 +733,58 @@ mod dataplane_tables { }); } - fn drive(flavour: NatFlavour) { - use concurrency::sync::atomic::{AtomicUsize, Ordering}; - let seen = AtomicUsize::new(0); - let built = AtomicUsize::new(0); - - let generator = GatewayAgentBuilder::new().flavours(vec![flavour]).build(); - - bolero::check!() - .with_generator(generator) - .cloned() - .for_each(|agent| { - seen.fetch_add(1, Ordering::Relaxed); - let external = ExternalConfig::try_from(&agent) - .unwrap_or_else(|e| panic!("a schema-legal CRD did not convert: {e}")); - let Ok(validated) = external.validate() else { - return; - }; - built.fetch_add(1, Ordering::Relaxed); - build_tables(&validated, flavour); - }); + macro_rules! drive { + ($flavour:expr) => {{ + let flavour = $flavour; + use concurrency::sync::atomic::{AtomicUsize, Ordering}; + let seen = AtomicUsize::new(0); + let built = AtomicUsize::new(0); + + let generator = GatewayAgentBuilder::new().flavours(vec![flavour]).build(); + + bolero::check!() + .with_generator(generator) + .cloned() + .for_each(|agent| { + seen.fetch_add(1, Ordering::Relaxed); + let external = ExternalConfig::try_from(&agent) + .unwrap_or_else(|e| panic!("a schema-legal CRD did not convert: {e}")); + let Ok(validated) = external.validate() else { + return; + }; + built.fetch_add(1, Ordering::Relaxed); + build_tables(&validated, flavour); + }); - let seen = seen.load(Ordering::Relaxed); - let built = built.load(Ordering::Relaxed); - println!("{flavour:?}: {built}/{seen} configurations validated and built their tables"); - assert!( - built * 2 >= seen, - "only {built} of {seen} {flavour:?} configurations validated, so this checked much \ + let seen = seen.load(Ordering::Relaxed); + let built = built.load(Ordering::Relaxed); + println!("{flavour:?}: {built}/{seen} configurations validated and built their tables"); + assert!( + built * 2 >= seen, + "only {built} of {seen} {flavour:?} configurations validated, so this checked much \ less than it looks like it did" - ); + ); + }}; } #[test] fn a_static_nat_configuration_builds_its_tables() { - drive(NatFlavour::Static); + drive!(NatFlavour::Static); } #[test] fn a_masquerade_configuration_builds_its_tables() { - drive(NatFlavour::Masquerade); + drive!(NatFlavour::Masquerade); } #[test] fn a_port_forwarding_configuration_builds_its_tables() { - drive(NatFlavour::PortForward); + drive!(NatFlavour::PortForward); } #[test] fn a_configuration_with_no_nat_builds_its_tables() { - drive(NatFlavour::None); + drive!(NatFlavour::None); } } diff --git a/nat/src/masquerade/expiry.rs b/nat/src/masquerade/expiry.rs index e1b20ac50a..bb9f99b113 100644 --- a/nat/src/masquerade/expiry.rs +++ b/nat/src/masquerade/expiry.rs @@ -7,6 +7,7 @@ use crate::Masquerade; use crate::masquerade::probe::{Arrival, Fabric, run}; use crate::static_nat::probe::build; use clock::Duration; +use clock::virtual_time::advance; use config::external::overlay::vpcpeering::VpcExpose; use config::external::overlay::vpcpeering::contract::{LOCAL_VNI, REMOTE_VNI}; use flow_entry::flow_table::FlowLookup; @@ -26,19 +27,7 @@ fn vni(raw: u32) -> Vni { } fn with_paused_clock>(body: impl FnOnce() -> F) { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_time() - .start_paused(true) - .build() - .unwrap_or_else(|e| unreachable!("{e}")); - runtime.block_on(body()); -} - -async fn advance(by: Duration) { - tokio::time::advance(by).await; - for _ in 0..4 { - tokio::task::yield_now().await; - } + clock::virtual_time::Paused::new().block_on(body()); } fn fabric() -> (Fabric, Vec) { diff --git a/nat/src/masquerade/fuzz.rs b/nat/src/masquerade/fuzz.rs index 08888ccf4f..51f5ce4fc0 100644 --- a/nat/src/masquerade/fuzz.rs +++ b/nat/src/masquerade/fuzz.rs @@ -41,13 +41,15 @@ impl ValueGenerator for Scenario { } } -fn with_runtime(body: impl FnOnce()) { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_time() - .build() - .unwrap_or_else(|e| unreachable!("{e}")); - let _guard = runtime.enter(); - body(); +fn settled(body: impl FnOnce()) { + const PAST_ANY_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(30); + // nosemgrep: rust-no-direct-std-sync-import + static CLOCK: std::sync::LazyLock = + std::sync::LazyLock::new(clock::virtual_time::Paused::new); // nosemgrep: rust-no-direct-std-sync-import + CLOCK.block_on(async { + body(); + clock::virtual_time::advance(PAST_ANY_TIMEOUT).await; + }); } fn fabric(exposes: &[VpcExpose]) -> Option { @@ -108,11 +110,11 @@ impl Tally { fn a_masqueraded_flow_comes_back() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: false }) - .cloned() - .for_each(|(exposes, probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: false }) + .cloned() + .for_each(|(exposes, probes): (Vec, Vec)| { + settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -150,8 +152,7 @@ fn a_masqueraded_flow_comes_back() { tally.reached.fetch_add(1, Ordering::Relaxed); } }); - }); - + }); tally.report("reversibility"); } @@ -159,11 +160,10 @@ fn a_masqueraded_flow_comes_back() { fn a_flow_keeps_its_translation() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: false }) - .cloned() - .for_each(|(exposes, probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: false }) + .cloned() + .for_each(|(exposes, probes): (Vec, Vec)| settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -190,9 +190,7 @@ fn a_flow_keeps_its_translation() { ); tally.reached.fetch_add(1, Ordering::Relaxed); } - }); - }); - + })); tally.report("stability"); } @@ -208,11 +206,10 @@ fn out_unchanged(out: &[Packet], before: (IpAddr, u16)) -> bool { fn an_internal_endpoint_keeps_one_public_address() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: false }) - .cloned() - .for_each(|(exposes, probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: false }) + .cloned() + .for_each(|(exposes, probes): (Vec, Vec)| settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -259,9 +256,7 @@ fn an_internal_endpoint_keeps_one_public_address() { ); tally.reached.fetch_add(1, Ordering::Relaxed); } - }); - }); - + })); tally.report("address pairing"); } @@ -277,11 +272,10 @@ fn an_internal_endpoint_keeps_one_public_address() { fn distinct_flows_do_not_share_a_translation() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: false }) - .cloned() - .for_each(|(exposes, probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: false }) + .cloned() + .for_each(|(exposes, probes): (Vec, Vec)| settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -309,9 +303,7 @@ fn distinct_flows_do_not_share_a_translation() { } tally.reached.fetch_add(1, Ordering::Relaxed); } - }); - }); - + })); tally.report("exclusivity"); } @@ -319,11 +311,11 @@ fn distinct_flows_do_not_share_a_translation() { fn a_translation_stays_inside_the_public_range() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: false }) - .cloned() - .for_each(|(exposes, probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: false }) + .cloned() + .for_each(|(exposes, probes): (Vec, Vec)| { + settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -353,8 +345,7 @@ fn a_translation_stays_inside_the_public_range() { tally.reached.fetch_add(1, Ordering::Relaxed); } }); - }); - + }); tally.report("containment"); } @@ -362,11 +353,10 @@ fn a_translation_stays_inside_the_public_range() { fn nothing_is_masqueraded_without_permission() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: true }) - .cloned() - .for_each(|(exposes, probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: true }) + .cloned() + .for_each(|(exposes, probes): (Vec, Vec)| settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -395,9 +385,7 @@ fn nothing_is_masqueraded_without_permission() { ); tally.reached.fetch_add(1, Ordering::Relaxed); } - }); - }); - + })); tally.report("permission"); } @@ -405,11 +393,10 @@ fn nothing_is_masqueraded_without_permission() { fn a_flow_that_cannot_be_masqueraded_says_so() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: true }) - .cloned() - .for_each(|(exposes, probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: true }) + .cloned() + .for_each(|(exposes, probes): (Vec, Vec)| settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -438,8 +425,6 @@ fn a_flow_that_cannot_be_masqueraded_says_so() { ); tally.reached.fetch_add(1, Ordering::Relaxed); } - }); - }); - + })); tally.report("attribution"); } diff --git a/nat/src/portfw/expiry.rs b/nat/src/portfw/expiry.rs index 0c7275e9c6..7293081296 100644 --- a/nat/src/portfw/expiry.rs +++ b/nat/src/portfw/expiry.rs @@ -7,6 +7,7 @@ use crate::portfw::PortForwarder; use crate::portfw::probe::{Arrival, Fabric, PAST_ANY_TIMEOUT, run}; use crate::static_nat::probe::build; use clock::Duration; +use clock::virtual_time::advance; use config::external::overlay::vpcpeering::VpcExpose; use flow_entry::flow_table::FlowLookup; use lpm::prefix::{L4Protocol, PrefixWithOptionalPorts}; @@ -15,19 +16,7 @@ use std::net::IpAddr; const WITHIN_LIFETIME: Duration = Duration::from_secs(1); fn with_paused_clock>(body: impl FnOnce() -> F) { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_time() - .start_paused(true) - .build() - .unwrap_or_else(|e| unreachable!("{e}")); - runtime.block_on(body()); -} - -async fn advance(by: Duration) { - tokio::time::advance(by).await; - for _ in 0..4 { - tokio::task::yield_now().await; - } + clock::virtual_time::Paused::new().block_on(body()); } fn fabric() -> Fabric { diff --git a/nat/src/portfw/fuzz.rs b/nat/src/portfw/fuzz.rs index a58d6390bb..d6d5768332 100644 --- a/nat/src/portfw/fuzz.rs +++ b/nat/src/portfw/fuzz.rs @@ -40,13 +40,15 @@ impl ValueGenerator for Scenario { } } -fn with_runtime(body: impl FnOnce()) { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_time() - .build() - .unwrap_or_else(|e| unreachable!("{e}")); - let _guard = runtime.enter(); - body(); +fn settled(body: impl FnOnce()) { + const PAST_ANY_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(30); + // nosemgrep: rust-no-direct-std-sync-import + static CLOCK: std::sync::LazyLock = + std::sync::LazyLock::new(clock::virtual_time::Paused::new); // nosemgrep: rust-no-direct-std-sync-import + CLOCK.block_on(async { + body(); + clock::virtual_time::advance(PAST_ANY_TIMEOUT).await; + }); } fn fabric(exposes: &[VpcExpose]) -> Option { @@ -123,11 +125,11 @@ fn forward( fn a_forwarded_packet_answers_as_the_published_tuple() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: false }) - .cloned() - .for_each(|(exposes, probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: false }) + .cloned() + .for_each(|(exposes, probes): (Vec, Vec)| { + settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -158,8 +160,7 @@ fn a_forwarded_packet_answers_as_the_published_tuple() { tally.reached.fetch_add(1, Ordering::Relaxed); } }); - }); - + }); tally.report("reversibility"); } @@ -167,11 +168,11 @@ fn a_forwarded_packet_answers_as_the_published_tuple() { fn a_forwarded_packet_lands_inside_the_published_target() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: false }) - .cloned() - .for_each(|(exposes, probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: false }) + .cloned() + .for_each(|(exposes, probes): (Vec, Vec)| { + settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -194,8 +195,7 @@ fn a_forwarded_packet_lands_inside_the_published_target() { tally.reached.fetch_add(1, Ordering::Relaxed); } }); - }); - + }); tally.report("containment"); } @@ -203,11 +203,11 @@ fn a_forwarded_packet_lands_inside_the_published_target() { fn distinct_published_tuples_reach_distinct_targets() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: false }) - .cloned() - .for_each(|(exposes, _probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: false }) + .cloned() + .for_each(|(exposes, _probes): (Vec, Vec)| { + settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -248,8 +248,7 @@ fn distinct_published_tuples_reach_distinct_targets() { } } }); - }); - + }); tally.report("injectivity"); } @@ -257,11 +256,11 @@ fn distinct_published_tuples_reach_distinct_targets() { fn nothing_is_forwarded_that_was_not_published() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: true }) - .cloned() - .for_each(|(exposes, probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: true }) + .cloned() + .for_each(|(exposes, probes): (Vec, Vec)| { + settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -293,8 +292,7 @@ fn nothing_is_forwarded_that_was_not_published() { tally.reached.fetch_add(1, Ordering::Relaxed); } }); - }); - + }); tally.report("permission"); } @@ -302,11 +300,11 @@ fn nothing_is_forwarded_that_was_not_published() { fn forwarding_touches_only_the_destination() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: false }) - .cloned() - .for_each(|(exposes, probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: false }) + .cloned() + .for_each(|(exposes, probes): (Vec, Vec)| { + settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -335,8 +333,7 @@ fn forwarding_touches_only_the_destination() { tally.reached.fetch_add(1, Ordering::Relaxed); } }); - }); - + }); tally.report("frame"); } @@ -344,11 +341,11 @@ fn forwarding_touches_only_the_destination() { fn a_forwarded_flow_keeps_its_target() { let tally = Tally::default(); - with_runtime(|| { - bolero::check!() - .with_generator(Scenario { strays: false }) - .cloned() - .for_each(|(exposes, probes): (Vec, Vec)| { + bolero::check!() + .with_generator(Scenario { strays: false }) + .cloned() + .for_each(|(exposes, probes): (Vec, Vec)| { + settled(|| { tally.seen.fetch_add(1, Ordering::Relaxed); let Some(fabric) = fabric(&exposes) else { return; @@ -378,7 +375,6 @@ fn a_forwarded_flow_keeps_its_target() { tally.reached.fetch_add(1, Ordering::Relaxed); } }); - }); - + }); tally.report("stability"); } diff --git a/net/src/flows/flow_info_fuzz.rs b/net/src/flows/flow_info_fuzz.rs index 657866e53e..9c68a5e861 100644 --- a/net/src/flows/flow_info_fuzz.rs +++ b/net/src/flows/flow_info_fuzz.rs @@ -72,21 +72,19 @@ fn flow() -> FlowInfo { info } -fn with_paused_clock>(body: impl FnOnce() -> F) { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_time() - .start_paused(true) - .build() - .unwrap_or_else(|e| unreachable!("{e}")); - runtime.block_on(body()); +fn paused(body: impl FnOnce()) { + // nosemgrep: rust-no-direct-std-sync-import + static CLOCK: std::sync::LazyLock = + std::sync::LazyLock::new(clock::virtual_time::Paused::new); // nosemgrep: rust-no-direct-std-sync-import + CLOCK.block_on(async { body() }); } #[test] fn expiry_never_moves_backwards() { - with_paused_clock(|| async { - bolero::check!() - .with_type::>() - .for_each(|ops: &Vec| { + bolero::check!() + .with_type::>() + .for_each(|ops: &Vec| { + paused(|| { let entry = flow(); let mut high_water = entry.expires_at(); @@ -100,7 +98,7 @@ fn expiry_never_moves_backwards() { high_water = now; } }); - }); + }); } fn apply(flow: &FlowInfo, op: Op) { @@ -117,9 +115,9 @@ fn apply(flow: &FlowInfo, op: Op) { #[test] fn a_refused_refresh_leaves_the_deadline_alone() { - with_paused_clock(|| async { - bolero::check!().with_type::<(Status, Millis)>().for_each( - |(status, millis): &(Status, Millis)| { + bolero::check!().with_type::<(Status, Millis)>().for_each( + |(status, millis): &(Status, Millis)| { + paused(|| { let entry = flow(); entry.update_status((*status).into()); let before = entry.expires_at(); @@ -138,16 +136,16 @@ fn a_refused_refresh_leaves_the_deadline_alone() { "extend_expiry refused for status {status:?} but moved the deadline anyway" ); } - }, - ); - }); + }); + }, + ); } #[test] fn a_refresh_is_permitted_exactly_when_the_status_allows() { - with_paused_clock(|| async { - bolero::check!().with_type::<(Status, Millis)>().for_each( - |(status, millis): &(Status, Millis)| { + bolero::check!().with_type::<(Status, Millis)>().for_each( + |(status, millis): &(Status, Millis)| { + paused(|| { let status = FlowStatus::from(*status); let entry = flow(); @@ -167,17 +165,17 @@ fn a_refresh_is_permitted_exactly_when_the_status_allows() { status != FlowStatus::Expired, "extend_expiry on a {status} flow returned {extend:?}" ); - }, - ); - }); + }); + }, + ); } #[test] fn invalidating_is_idempotent_and_cancels_the_timer() { - with_paused_clock(|| async { - bolero::check!() - .with_type::() - .for_each(|status: &Status| { + bolero::check!() + .with_type::() + .for_each(|status: &Status| { + paused(|| { let entry = flow(); let started_active = FlowStatus::from(*status) == FlowStatus::Active; entry.update_status((*status).into()); @@ -204,15 +202,15 @@ fn invalidating_is_idempotent_and_cancels_the_timer() { "invalidating twice did not leave the flow cancelled" ); }); - }); + }); } #[test] fn a_related_pair_refers_to_its_partner() { - with_paused_clock(|| async { - bolero::check!() - .with_type::<(u16, u16)>() - .for_each(|(a, b): &(u16, u16)| { + bolero::check!() + .with_type::<(u16, u16)>() + .for_each(|(a, b): &(u16, u16)| { + paused(|| { let (one, two) = (key(*a), key(b.wrapping_add(1))); let built = FlowInfo::related_pair( clock::now() + Duration::from_secs(1), @@ -258,7 +256,7 @@ fn a_related_pair_refers_to_its_partner() { survives with no reverse" ); }); - }); + }); } #[test] @@ -306,10 +304,10 @@ fn every_status_survives_its_byte() { #[test] fn the_unchecked_refreshes_move_the_deadline_exactly() { - with_paused_clock(|| async { - bolero::check!() - .with_type::<(Millis, Millis)>() - .for_each(|(a, b): &(Millis, Millis)| { + bolero::check!() + .with_type::<(Millis, Millis)>() + .for_each(|(a, b): &(Millis, Millis)| { + paused(|| { let (extend, reset) = (a.duration(), b.duration()); let subject = flow(); @@ -350,15 +348,15 @@ fn the_unchecked_refreshes_move_the_deadline_exactly() { ); assert_eq!(subject.expires_at(), held, "and must leave it where it was"); }); - }); + }); } #[test] fn a_flow_is_active_exactly_when_its_status_says_so() { - with_paused_clock(|| async { - bolero::check!() - .with_type::() - .for_each(|status: &Status| { + bolero::check!() + .with_type::() + .for_each(|status: &Status| { + paused(|| { let want = FlowStatus::from(*status); let flow = flow(); flow.update_status(want); @@ -368,15 +366,15 @@ fn a_flow_is_active_exactly_when_its_status_says_so() { "is_active disagreed with the status it was asked about" ); }); - }); + }); } #[test] fn a_flow_built_with_a_status_has_it() { - with_paused_clock(|| async { - bolero::check!() - .with_type::<(u16, Status)>() - .for_each(|(port, status): &(u16, Status)| { + bolero::check!() + .with_type::<(u16, Status)>() + .for_each(|(port, status): &(u16, Status)| { + paused(|| { let want = FlowStatus::from(*status); let flow = FlowInfo::new_with_status( key(*port), @@ -389,14 +387,15 @@ fn a_flow_built_with_a_status_has_it() { "the status asked for was not the one built" ); }); - }); + }); } #[test] fn a_genid_is_remembered_and_reaches_the_partner() { - with_paused_clock(|| async { - bolero::check!().with_type::<(u16, u16, i64)>().for_each( - |(a, b, genid): &(u16, u16, i64)| { + bolero::check!() + .with_type::<(u16, u16, i64)>() + .for_each(|(a, b, genid): &(u16, u16, i64)| { + paused(|| { let (one, two) = (key(*a), key(b.wrapping_add(1))); let Ok((first, second)) = FlowInfo::related_pair( clock::now() + Duration::from_secs(1), @@ -429,17 +428,16 @@ fn a_genid_is_remembered_and_reaches_the_partner() { "set_genid_pair must reach the partner, or the halves disagree about which \ configuration they belong to" ); - }, - ); - }); + }); + }); } #[test] fn each_flag_predicate_answers_for_its_own_bit() { - with_paused_clock(|| async { - bolero::check!() - .with_type::<(u16, u16, u8)>() - .for_each(|(a, b, bits): &(u16, u16, u8)| { + bolero::check!() + .with_type::<(u16, u16, u8)>() + .for_each(|(a, b, bits): &(u16, u16, u8)| { + paused(|| { let flags = FlowInfoFlags::from_bits_truncate(*bits); let (one, two) = (key(*a), key(b.wrapping_add(1))); let Ok((first, _second)) = FlowInfo::related_pair( @@ -470,15 +468,15 @@ fn each_flag_predicate_answers_for_its_own_bit() { "the initiator predicate answered for the wrong bit" ); }); - }); + }); } #[test] fn the_destination_vpc_is_remembered() { - with_paused_clock(|| async { - bolero::check!() - .with_type::>() - .for_each(|vni: &Option| { + bolero::check!() + .with_type::>() + .for_each(|vni: &Option| { + paused(|| { let want = vni .and_then(|v| crate::vxlan::Vni::new_checked(v % 0x00FF_FFFF).ok()) .map(crate::packet::VpcDiscriminant::from_vni); @@ -490,5 +488,5 @@ fn the_destination_vpc_is_remembered() { "the destination vpc read back must be the one stamped" ); }); - }); + }); } diff --git a/net/src/headers/embedded_view.rs b/net/src/headers/embedded_view.rs index d3fffc3472..df0945f6f4 100644 --- a/net/src/headers/embedded_view.rs +++ b/net/src/headers/embedded_view.rs @@ -2093,40 +2093,43 @@ mod embedded_view_properties { ); } - fn exercise_the_embedded_split() { - bolero::check!() - .with_generator(ShapedIcmpError) - .for_each(|h: &Headers| { - let mut owned = h.clone(); - let Some(outer) = owned.as_view_mut::() else { - return; - }; - let Some(ew) = outer.as_embedded_mut::<(&Ipv6, &HopByHop, &TruncatedTcp)>() else { - return; - }; - let (ip, ext, tcp) = ew.look_mut(); - - let want_hops = ip.hop_limit().wrapping_add(1); - ip.set_hop_limit(want_hops); - let seen_ext = ext.next_header(); - let seen_tcp = matches!(tcp, TruncatedTcp::FullHeader(_)); - - assert_eq!( - ip.hop_limit(), - want_hops, - "the write through ipv6 did not stick" - ); - assert_eq!( - ext.next_header(), - seen_ext, - "the extension header changed under a write to ipv6" - ); - assert_eq!( - matches!(tcp, TruncatedTcp::FullHeader(_)), - seen_tcp, - "the quoted transport changed under a write to ipv6" - ); - }); + macro_rules! exercise_the_embedded_split { + () => {{ + bolero::check!() + .with_generator(ShapedIcmpError) + .for_each(|h: &Headers| { + let mut owned = h.clone(); + let Some(outer) = owned.as_view_mut::() else { + return; + }; + let Some(ew) = outer.as_embedded_mut::<(&Ipv6, &HopByHop, &TruncatedTcp)>() + else { + return; + }; + let (ip, ext, tcp) = ew.look_mut(); + + let want_hops = ip.hop_limit().wrapping_add(1); + ip.set_hop_limit(want_hops); + let seen_ext = ext.next_header(); + let seen_tcp = matches!(tcp, TruncatedTcp::FullHeader(_)); + + assert_eq!( + ip.hop_limit(), + want_hops, + "the write through ipv6 did not stick" + ); + assert_eq!( + ext.next_header(), + seen_ext, + "the extension header changed under a write to ipv6" + ); + assert_eq!( + matches!(tcp, TruncatedTcp::FullHeader(_)), + seen_tcp, + "the quoted transport changed under a write to ipv6" + ); + }); + }}; } macro_rules! split_shards { @@ -2134,7 +2137,7 @@ mod embedded_view_properties { $( #[test] fn $name() { - exercise_the_embedded_split(); + exercise_the_embedded_split!(); } )* }; diff --git a/net/src/headers/view.rs b/net/src/headers/view.rs index b0a38e11da..3de063d6cd 100644 --- a/net/src/headers/view.rs +++ b/net/src/headers/view.rs @@ -2966,37 +2966,40 @@ mod view_mut_properties { vxlan ); - fn exercise_the_mutable_split() { - bolero::check!() - .with_generator(ShapedHeaders) - .for_each(|h: &Headers| { - let mut owned = h.clone(); - let Some(view) = owned.as_view_mut::<(&Eth, &Net, &Transport)>() else { - return; - }; - let (eth, net, transport) = view.look_mut(); + macro_rules! exercise_the_mutable_split { + () => {{ + bolero::check!() + .with_generator(ShapedHeaders) + .for_each(|h: &Headers| { + let mut owned = h.clone(); + let Some(view) = owned.as_view_mut::<(&Eth, &Net, &Transport)>() else { + return; + }; + let (eth, net, transport) = view.look_mut(); + + let want_src = crate::eth::mac::SourceMac::try_from(crate::eth::mac::Mac([ + 2, 0, 0, 0, 0, 1, + ])) + .unwrap_or_else(|_| { + unreachable!("a locally-administered unicast mac is a valid source") + }); + eth.set_source(want_src); + let seen_net = net.dst_addr(); + let seen_transport = transport.dst_port(); - let want_src = - crate::eth::mac::SourceMac::try_from(crate::eth::mac::Mac([2, 0, 0, 0, 0, 1])) - .unwrap_or_else(|_| { - unreachable!("a locally-administered unicast mac is a valid source") - }); - eth.set_source(want_src); - let seen_net = net.dst_addr(); - let seen_transport = transport.dst_port(); - - assert_eq!( - eth.source(), - want_src, - "the write through eth did not stick" - ); - assert_eq!(net.dst_addr(), seen_net, "net changed under a write to eth"); - assert_eq!( - transport.dst_port(), - seen_transport, - "transport changed under a write to eth" - ); - }); + assert_eq!( + eth.source(), + want_src, + "the write through eth did not stick" + ); + assert_eq!(net.dst_addr(), seen_net, "net changed under a write to eth"); + assert_eq!( + transport.dst_port(), + seen_transport, + "transport changed under a write to eth" + ); + }); + }}; } macro_rules! split_shards { @@ -3004,7 +3007,7 @@ mod view_mut_properties { $( #[test] fn $name() { - exercise_the_mutable_split(); + exercise_the_mutable_split!(); } )* }; diff --git a/nix/overlays/dataplane.nix b/nix/overlays/dataplane.nix index 76a5fca2ce..6bcecfab4b 100644 --- a/nix/overlays/dataplane.nix +++ b/nix/overlays/dataplane.nix @@ -3,6 +3,7 @@ { sources, sanitizers, + instrumentations, platform, profile, ... @@ -206,6 +207,7 @@ in (builtins.elem "thread" sanitizers) || (builtins.elem "address" sanitizers) || (builtins.elem "safe-stack" sanitizers) + || (builtins.elem "fuzz" instrumentations) ) [ # This allows address / thread sanitizer to build (some sanitizers do not like -Wl,-z,defs or diff --git a/nix/profiles.nix b/nix/profiles.nix index c05f346764..23189eab28 100644 --- a/nix/profiles.nix +++ b/nix/profiles.nix @@ -5,7 +5,7 @@ host-arch, profile, sanitizers, - instrumentation, + instrumentations, cargo-features ? [ ], for-tests ? false, }: @@ -64,7 +64,7 @@ let ] ) ++ (if is-emulated-test then [ "--cfg=emulated" ] else [ ]) - ++ (if instrumentation == "coverage" then [ "--cfg=instrumented" ] else [ ]) + ++ (if builtins.elem "coverage" instrumentations then [ "--cfg=instrumented" ] else [ ]) ++ (map (flag: "-Clink-arg=${flag}") common.NIX_CFLAGS_LINK); optimize-for.debug.NIX_CFLAGS_COMPILE = [ "-fno-inline" @@ -229,6 +229,22 @@ let "-Ctarget-feature=-crt-static" # shadow-stack doesn't work with static libc ] ++ (map (flag: "-Clink-arg=${flag}") sanitize.shadow-stack.NIX_CFLAGS_LINK); + instrument.fuzz.NIX_CFLAGS_COMPILE = [ + "-fsanitize=fuzzer-no-link" + "-fno-lto" + ]; + instrument.fuzz.NIX_CXXFLAGS_COMPILE = instrument.fuzz.NIX_CFLAGS_COMPILE; + instrument.fuzz.NIX_CFLAGS_LINK = [ ]; + instrument.fuzz.RUSTFLAGS = [ + "--cfg=fuzzing" + "-Cpasses=sancov-module" + "-Cllvm-args=-sanitizer-coverage-inline-8bit-counters" + "-Cllvm-args=-sanitizer-coverage-level=4" + "-Cllvm-args=-sanitizer-coverage-pc-table" + "-Cllvm-args=-sanitizer-coverage-trace-compares" + "-Cllvm-args=-sanitizer-coverage-stack-depth" + ] + ++ (map (flag: "-Clink-arg=${flag}") instrument.fuzz.NIX_CFLAGS_LINK); instrument.none.NIX_CFLAGS_COMPILE = [ ]; instrument.none.NIX_CXXFLAGS_COMPILE = instrument.none.NIX_CFLAGS_COMPILE; instrument.none.NIX_CFLAGS_LINK = instrument.none.NIX_CFLAGS_COMPILE; @@ -260,14 +276,14 @@ let optimize-for.performance secure ]; - fuzz = release; + checked = release; }; in combine-profiles ( [ profile-map."${profile}" march."${arch}" - instrument."${instrumentation}" ] + ++ (map (i: instrument.${i}) instrumentations) ++ (map (s: sanitize.${s}) sanitizers) ) diff --git a/routing/Cargo.toml b/routing/Cargo.toml index d7c2376794..545532d4f5 100644 --- a/routing/Cargo.toml +++ b/routing/Cargo.toml @@ -61,7 +61,7 @@ lpm = { workspace = true, features = ["testing"] } clock = { workspace = true, features = ["virtual"] } criterion = { workspace = true } iai-callgrind = { workspace = true } -bolero = { workspace = true, default-features = false } +bolero = { workspace = true, features = ["std"] } concurrency = { workspace = true } net = { workspace = true, features = ["test_buffer"] } rand = { workspace = true, default-features = false, features = ["thread_rng"] } diff --git a/stats/src/rate.rs b/stats/src/rate.rs index e44b61626a..759cc6c334 100644 --- a/stats/src/rate.rs +++ b/stats/src/rate.rs @@ -746,101 +746,102 @@ mod test { use std::time::Duration; - fn arbitrary_polynomial() { - const NANOS_PER_SEC: u128 = 1_000_000_000; - bolero::check!() - .with_type() - .cloned() - .for_each(|(x, c): (Duration, [u64; N])| { - let x = if x < Duration::from_micros(1) { - Duration::from_micros(1) - } else if x > Duration::from_secs(10) { - Duration::from_secs(10) - } else { - x - }; - // we will get overflow errors if we don't clamp the slope - let c = c.map(|x| u128::from(x.clamp(0, 1_000))); - let basic = move |x: Duration| { - let x = x.as_nanos() / NANOS_PER_SEC; - u64::try_from( - c.iter() - .enumerate() - .fold(0u128, |acc, (i, &c)| acc + c * x.pow(i as u32)), - ) - .unwrap() - }; - let basic_prime = move |x: Duration| { - let x = x.as_nanos() / NANOS_PER_SEC; - c.iter().enumerate().fold(0u128, |acc, (i, &c)| { - if i == 0 { - return acc; - } - acc + u128::try_from(i).unwrap() * c * x.pow(i as u32 - 1) - }) as f64 - }; - let comparer = DerivativeComparer { - f: basic, - d: basic_prime, - step: Duration::from_secs(1), - }; - let comparison = comparer.compare(x); - if comparison.relative_error().is_nan() { - assert!(comparison.diff().abs() < 0.001); - return; - } - assert!(comparison.relative_error().abs() < 0.01); - }) + macro_rules! arbitrary_polynomial { + ($n:expr) => {{ + const NANOS_PER_SEC: u128 = 1_000_000_000; + bolero::check!() + .with_type() + .cloned() + .for_each(|(x, c): (Duration, [u64; $n])| { + let x = if x < Duration::from_micros(1) { + Duration::from_micros(1) + } else if x > Duration::from_secs(10) { + Duration::from_secs(10) + } else { + x + }; + let c = c.map(|x| u128::from(x.clamp(0, 1_000))); + let basic = move |x: Duration| { + let x = x.as_nanos() / NANOS_PER_SEC; + u64::try_from( + c.iter() + .enumerate() + .fold(0u128, |acc, (i, &c)| acc + c * x.pow(i as u32)), + ) + .unwrap() + }; + let basic_prime = move |x: Duration| { + let x = x.as_nanos() / NANOS_PER_SEC; + c.iter().enumerate().fold(0u128, |acc, (i, &c)| { + if i == 0 { + return acc; + } + acc + u128::try_from(i).unwrap() * c * x.pow(i as u32 - 1) + }) as f64 + }; + let comparer = DerivativeComparer { + f: basic, + d: basic_prime, + step: Duration::from_secs(1), + }; + let comparison = comparer.compare(x); + if comparison.relative_error().is_nan() { + assert!(comparison.diff().abs() < 0.001); + return; + } + assert!(comparison.relative_error().abs() < 0.01); + }) + }}; } #[test] fn derivative_of_arbitrary_1() { - arbitrary_polynomial::<1>(); + arbitrary_polynomial!(1); } #[test] fn derivative_of_arbitrary_2() { - arbitrary_polynomial::<2>(); + arbitrary_polynomial!(2); } #[test] fn derivative_of_arbitrary_3() { - arbitrary_polynomial::<3>(); + arbitrary_polynomial!(3); } #[test] fn derivative_of_arbitrary_4() { - arbitrary_polynomial::<4>(); + arbitrary_polynomial!(4); } #[test] fn derivative_of_arbitrary_5() { - arbitrary_polynomial::<5>(); + arbitrary_polynomial!(5); } #[test] fn derivative_of_arbitrary_6() { - arbitrary_polynomial::<6>(); + arbitrary_polynomial!(6); } #[test] fn derivative_of_arbitrary_7() { - arbitrary_polynomial::<7>(); + arbitrary_polynomial!(7); } #[test] fn derivative_of_arbitrary_8() { - arbitrary_polynomial::<8>(); + arbitrary_polynomial!(8); } #[test] fn derivative_of_arbitrary_9() { - arbitrary_polynomial::<9>(); + arbitrary_polynomial!(9); } #[test] fn derivative_of_arbitrary_10() { - arbitrary_polynomial::<10>(); + arbitrary_polynomial!(10); } #[test] fn derivative_of_arbitrary_11() { - arbitrary_polynomial::<11>(); + arbitrary_polynomial!(11); } #[test] fn derivative_of_arbitrary_12() { - arbitrary_polynomial::<12>(); + arbitrary_polynomial!(12); } #[test] diff --git a/tracectl/src/control.rs b/tracectl/src/control.rs index 0d83910089..48e47c2303 100644 --- a/tracectl/src/control.rs +++ b/tracectl/src/control.rs @@ -502,6 +502,7 @@ impl TracingControl { S: Subscriber + for<'span> LookupSpan<'span>, { tracing_subscriber::fmt::layer() + .with_timer(crate::stamp::Stamp) .with_line_number(true) .with_target(true) .with_thread_ids(false) diff --git a/tracectl/src/lib.rs b/tracectl/src/lib.rs index 4e18c10b4e..0e19cea8af 100644 --- a/tracectl/src/lib.rs +++ b/tracectl/src/lib.rs @@ -11,6 +11,7 @@ pub mod evidence; pub mod control; pub mod display; +mod stamp; pub mod targets; mod throttle; diff --git a/tracectl/src/stamp.rs b/tracectl/src/stamp.rs new file mode 100644 index 0000000000..99e93729b6 --- /dev/null +++ b/tracectl/src/stamp.rs @@ -0,0 +1,94 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright Open Network Fabric Authors + +use clock::Duration; +use std::fmt; +use tracing_subscriber::fmt::format::Writer; +use tracing_subscriber::fmt::time::{FormatTime, SystemTime}; + +#[derive(Debug, Clone, Copy, Default)] +pub struct Stamp; + +impl FormatTime for Stamp { + fn format_time(&self, writer: &mut Writer<'_>) -> fmt::Result { + if !clock::is_routed() { + return SystemTime.format_time(writer); + } + match clock::elapsed_since_first_reading() { + Some((behind, elapsed)) => write!( + writer, + "T{}{}", + if behind { '-' } else { '+' }, + Rendered(elapsed) + ), + None => writer.write_str("T+?off-clock"), + } + } +} + +struct Rendered(Duration); + +impl fmt::Display for Rendered { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}.{:06}s", self.0.as_secs(), self.0.subsec_micros()) + } +} + +#[cfg(test)] +mod tests { + use super::{Rendered, Stamp}; + use clock::Duration; + use clock::virtual_time::{Paused, advance}; + use std::thread; + use tracing_subscriber::fmt::format::Writer; + use tracing_subscriber::fmt::time::FormatTime; + + fn render() -> String { + let mut out = String::new(); + Stamp + .format_time(&mut Writer::new(&mut out)) + .expect("the stamp did not format"); + out + } + + #[test] + fn a_line_is_stamped_on_the_clock_the_code_is_reading() { + let clock = Paused::new(); + clock.block_on(async { + let first = render(); + advance(Duration::from_hours(1)).await; + let later = render(); + assert!( + first.starts_with("T+0."), + "the first line was stamped {first}" + ); + assert!( + later.starts_with("T+3600."), + "an hour passed and the stamp said {later}" + ); + }); + } + + #[test] + fn a_line_from_off_the_clock_says_so() { + let clock = Paused::new(); + clock.block_on(async { advance(Duration::from_hours(1)).await }); + let stamped = thread::spawn(render) + .join() + .expect("formatting a stamp panicked"); + assert_eq!(stamped, "T+?off-clock"); + } + + #[test] + fn an_offset_reads_at_a_glance() { + assert_eq!( + Rendered(Duration::from_hours(1)).to_string(), + "3600.000000s" + ); + assert_eq!( + Rendered(Duration::from_micros(1)).to_string(), + "0.000001s", + "sub-millisecond detail is what separates two lines in the same burst" + ); + } +}