(10) concurrency

.cargo/config.toml

@@ -13,7 +13,7 @@ CARGO_LLVM_COV_BUILD_DIR = { value = "target/llvm-cov/target", relative = true,
 
 [build]
 # Register `emulated` and `instrumented` so cfg sites do not trip unexpected_cfgs natively.
-rustflags = ["--cfg=tokio_unstable", "--check-cfg=cfg(emulated)", "--check-cfg=cfg(instrumented)"]
+rustflags = ["--cfg=tokio_unstable", "--check-cfg=cfg(emulated)", "--check-cfg=cfg(instrumented)", "--check-cfg=cfg(sanitized)"]
 
 [target.wasm32-wasip1]
 # Trailing `--` separates wasmtime's CLI from the module + module args

.config/nextest.toml

@@ -144,6 +144,14 @@ test-group = 'vm'
 [test-groups]
 vm = { max-threads = 1 }
 
+# Find hung tests by stopping each test after 10 seconds and the run after five minutes. These
+# limits can also stop valid, slow tests. Rerun reported tests with a normal profile.
+[profile.scratch]
+slow-timeout = { period = "10s", terminate-after = 1 }
+global-timeout = "300s"
+fail-fast = false
+final-status-level = "all"
+
 [profile.miri]
 slow-timeout = { period = "500s" }
 

Cargo.toml

@@ -156,8 +156,9 @@ k8s-openapi = { version = "0.28.0", default-features = false, features = [] }
 kanal = { version = "0.1.1", default-features = false, features = [] }
 kube = { version = "4.2.0", default-features = false, features = [] }
 kube-core = { version = "4.2.0", default-features = false, features = [] }
-#left-right = { version = "0.11.7", default-features = false, features = [] }
-left-right = { git = "https://github.com/githedgehog/left-right.git", branch = "fredi/fix-writehandle-drop", default-features = false, features = [] }
+# The hh/shuttle fork makes WriteHandle::wait yield to Shuttle. Without it, the writer's
+# sched_yield loop blocks the scheduler thread, preventing the reader from releasing its epoch.
+left-right = { git = "https://github.com/githedgehog/left-right.git", branch = "hh/shuttle", default-features = false, features = [] }
 libc = { version = "0.2.189", default-features = false, features = [] }
 linkme = { version = "0.3.37", default-features = false, features = [] }
 log = { version = "0.4.34", default-features = false, features = [] } # TODO: try to remove this

acl-filter/src/context.rs

@@ -10,8 +10,6 @@ use acl::dpdk::lookup::DpdkAclLookup;
 use acl::dpdk::rule::{AclFieldChunks, RuleSpec};
 #[cfg(test)]
 use acl::reference::table::{RefRule, ReferenceTable};
-use concurrency::sync::LazyLock;
-use concurrency::sync::atomic::{AtomicU64, Ordering};
 use config::ConfigError;
 use config::external::overlay::ValidatedOverlay;
 use config::external::overlay::acl::{AclAction, AclProtoMatch, AclScope, ValidatedAclRule};
@@ -404,16 +402,20 @@ impl<K: MatchKey, A> fmt::Debug for AnyTable<K, A> {
     }
 }
 
-// Lazily initialized so this compiles under the loom backend, whose AtomicU64::new is not const
-// (each instance registers with the loom executor). The atomic itself is still the backend atomic,
-// so fetch_add() stays instrumented; only construction is deferred. On every other backend LazyLock
-// is a thin wrapper over an otherwise-const atomic.
-static TABLE_SEQ: LazyLock<AtomicU64> = LazyLock::new(|| AtomicU64::new(0));
+// DPDK's ACL registry survives model-checker executions. Its name counter must also persist to
+// avoid reusing a live context's name.
+use concurrency::process_global::atomic::{AtomicU64, Ordering};
+
+static TABLE_SEQ: AtomicU64 = AtomicU64::new(0);
+
+fn next_in_sequence() -> u64 {
+    TABLE_SEQ.fetch_add(1, Ordering::Relaxed)
+}
 
 /// A process-unique rte_acl context name. rte_acl rejects duplicate names, and a hot-swap briefly
 /// keeps the old and new contexts alive at once, so the name must be unique across the process.
 fn table_name(base: &str) -> String {
-    format!("acl_{base}_{}", TABLE_SEQ.fetch_add(1, Ordering::Relaxed))
+    format!("acl_{base}_{}", next_in_sequence())
 }
 
 /// Build one table for the selected backend from rules in precedence (insertion) order.

acl-filter/src/nf_fuzz.rs

@@ -91,7 +91,7 @@ fn filter(built: &crate::fuzz_gen::BuiltOverlay) -> AclFilter {
 fn judged(drawn: usize) -> bool {
     /// Sixteen cases' worth: far under any real run, far over a single replay.
     const ENOUGH_PROBES: usize = PROBES * 16;
-    !cfg!(instrumented) && !cfg!(emulated) && drawn >= ENOUGH_PROBES
+    !cfg!(instrumented) && !cfg!(emulated) && !cfg!(sanitized) && drawn >= ENOUGH_PROBES
 }
 
 #[derive(Default)]

acl-filter/src/tests.rs

@@ -1012,7 +1012,7 @@ mod end_to_end {
         // Port forwarding
         let mut portfw_writer = PortFwTableWriter::new();
         portfw_writer
-            .update_from_vpc_table(overlay.vpc_table())
+            .update_from_vpc_table(overlay.vpc_table(), &flow_table, 1)
             .unwrap();
         pipeline = pipeline.add_stage(PortForwarder::new(
             "port-forwarder",

acl/benches/table_build.rs

@@ -7,8 +7,7 @@
 mod bench {
     use std::hint::black_box;
 
-    use concurrency::sync::LazyLock;
-    use concurrency::sync::atomic::{AtomicU32, Ordering};
+    use concurrency::process_global::atomic::{AtomicU32, Ordering};
     use core::net::{Ipv4Addr, Ipv6Addr};
     use core::num::NonZero;
 
@@ -54,8 +53,7 @@ mod bench {
     const RULE_COUNTS: [usize; 15] = [
         1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384,
     ];
-    // Lazily initialized so this compiles under the loom backend, whose AtomicU32::new is not const
-    static SEQ: LazyLock<AtomicU32> = LazyLock::new(|| AtomicU32::new(0));
+    static SEQ: AtomicU32 = AtomicU32::new(0);
 
     fn unique_name(prefix: &str) -> String {
         format!("{prefix}_{}", SEQ.fetch_add(1, Ordering::Relaxed))

acl/src/dpdk/dyn_table.rs

@@ -431,11 +431,9 @@ mod failing_repros {
             offset,
         }
     }
-    use concurrency::sync::LazyLock;
-    use concurrency::sync::atomic::{AtomicU32, Ordering};
+    use concurrency::process_global::atomic::{AtomicU32, Ordering};
 
-    // Lazily initialized so this compiles under the loom backend, whose AtomicU32::new is not const
-    static SEQ: LazyLock<AtomicU32> = LazyLock::new(|| AtomicU32::new(0));
+    static SEQ: AtomicU32 = AtomicU32::new(0);
     fn uname(p: &str) -> String {
         format!("{p}_{}", SEQ.fetch_add(1, Ordering::Relaxed))
     }
@@ -507,11 +505,9 @@ mod tests {
 
     dpdk_table_alias!(type FiveTupleTable<A> = FiveTuple);
 
-    use concurrency::sync::LazyLock;
-    use concurrency::sync::atomic::{AtomicU32, Ordering};
+    use concurrency::process_global::atomic::{AtomicU32, Ordering};
 
-    // Lazily initialized so this compiles under the loom backend, whose AtomicU32::new is not const
-    static CTX_SEQ: LazyLock<AtomicU32> = LazyLock::new(|| AtomicU32::new(0));
+    static CTX_SEQ: AtomicU32 = AtomicU32::new(0);
     fn unique_name(prefix: &str) -> String {
         format!("{prefix}_{}", CTX_SEQ.fetch_add(1, Ordering::Relaxed))
     }

acl/tests/property_dyn_shape.rs

@@ -4,8 +4,7 @@
 #![cfg(feature = "dpdk")]
 #![allow(clippy::expect_used, clippy::unwrap_used)]
 
-use concurrency::sync::LazyLock;
-use concurrency::sync::atomic::{AtomicU32, AtomicU64, Ordering};
+use concurrency::process_global::atomic::{AtomicU32, AtomicU64, Ordering};
 use core::num::NonZero;
 
 use bolero::TypeGenerator;
@@ -148,8 +147,7 @@ impl ValueGenerator for ShapeMisses {
     }
 }
 
-// Lazily initialized so this compiles under the loom backend, whose AtomicU32::new is not const
-static CTX_SEQ: LazyLock<AtomicU32> = LazyLock::new(|| AtomicU32::new(0));
+static CTX_SEQ: AtomicU32 = AtomicU32::new(0);
 
 fn unique_name(prefix: &str) -> String {
     format!("{prefix}_{}", CTX_SEQ.fetch_add(1, Ordering::Relaxed))
@@ -210,10 +208,9 @@ where
 #[test]
 #[dpdk::with_eal]
 fn dyn_dpdk_and_reference_agree_on_random_shapes() {
-    // Lazily initialized so this compiles under the loom backend, whose AtomicU64::new is not const
-    static ASSERTED_HITS: LazyLock<AtomicU64> = LazyLock::new(|| AtomicU64::new(0));
-    static ASSERTED_MISSES: LazyLock<AtomicU64> = LazyLock::new(|| AtomicU64::new(0));
-    static SHAPES_RUN: LazyLock<AtomicU64> = LazyLock::new(|| AtomicU64::new(0));
+    static ASSERTED_HITS: AtomicU64 = AtomicU64::new(0);
+    static ASSERTED_MISSES: AtomicU64 = AtomicU64::new(0);
+    static SHAPES_RUN: AtomicU64 = AtomicU64::new(0);
 
     bolero::check!()
         .with_type::<(RawShape, Box<[u8]>, Box<[u8]>)>()

acl/tests/property_predicate.rs

@@ -4,8 +4,7 @@
 #![cfg(feature = "dpdk")]
 #![allow(clippy::expect_used, clippy::unwrap_used)]
 
-use concurrency::sync::LazyLock;
-use concurrency::sync::atomic::{AtomicU32, AtomicU64, Ordering};
+use concurrency::process_global::atomic::{AtomicU32, AtomicU64, Ordering};
 use core::net::{Ipv4Addr, Ipv6Addr};
 use core::num::NonZero;
 use core::ops::Bound;
@@ -180,8 +179,7 @@ enum Verdict {
     Drop,
 }
 
-// Lazily initialized so this compiles under the loom backend, whose AtomicU32::new is not const
-static CTX_SEQ: LazyLock<AtomicU32> = LazyLock::new(|| AtomicU32::new(0));
+static CTX_SEQ: AtomicU32 = AtomicU32::new(0);
 
 fn unique_name(prefix: &str) -> String {
     format!("{prefix}_{}", CTX_SEQ.fetch_add(1, Ordering::Relaxed))

concurrency/src/lib.rs

@@ -38,13 +38,6 @@
 //!   schedule that `parking_lot` permits. Tests that hinge on that
 //!   interleaving need `RwLock<T>` with explicit `read()` then
 //!   `write()`, or a richer state machine in the facade.
-//! * **`static FOO: Mutex<T> = Mutex::new(...)` does not compile
-//!   under loom.** `loom::sync::Mutex::new` is plain `fn`, not
-//!   `const fn`, so a static initialiser fails to typecheck. Use
-//!   `OnceLock` for the static (the facade re-exports
-//!   `std::sync::OnceLock` under all backends) or move the
-//!   construction into a runtime initialiser gated by
-//!   `#[concurrency_mode(std)]`.
 //! * **`OnceLock` under loom/shuttle** is the real `std::sync::OnceLock`,
 //!   not a model-aware shim. Loom and shuttle do not see the
 //!   atomics inside `OnceLock::get_or_init`, so tests whose

config/src/external/overlay/acl.rs

@@ -424,8 +424,8 @@ impl ValidatedAclRule {
 
 #[derive(Debug, Clone, PartialEq, Eq, Default)]
 pub struct Acl {
-    default: AclAction,
-    rules: Vec<AclRule>,
+    pub(crate) default: AclAction,
+    pub(crate) rules: Vec<AclRule>,
 }
 
 impl Acl {