feat: use bolero-generator for rand backend

Cargo.toml

@@ -1,10 +1,14 @@
 [workspace]
 members = [
     "bach",
-    #"bach-*",
+    "bach-tests",
 ]
 resolver = "2"
 
+[workspace.dependencies]
+bolero-generator = { version = "0.12", features = ["any"] }
+bolero = { version = "0.12" }
+
 [profile.bench]
 lto = true
 codegen-units = 1
@@ -18,4 +22,4 @@ codegen-units = 1
 
 [profile.release-debug]
 inherits = "dev"
-opt-level = 3
+opt-level = 3

bach-tests/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "bach-tests"
+version = "0.0.0"
+edition = "2021"
+license = "MIT"
+publish = false
+
+[dependencies]
+mimalloc = { version = "0.1", default-features = false }
+
+[dev-dependencies]
+bach = { path = "../bach", features = ["coop"] }
+bolero.workspace = true
+insta = "1"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }

bach-tests/src/coop.rs

@@ -0,0 +1,52 @@
+use std::sync::Mutex;
+
+use bach::{environment::default::Runtime, ext::*, sync::queue::vec_deque::Queue};
+
+fn sim(f: impl Fn()) -> impl Fn() {
+    crate::testing::init_tracing();
+    move || {
+        let mut rt = Runtime::new().with_coop(true).with_rand(None);
+        rt.run(&f);
+    }
+}
+
+#[derive(Debug)]
+#[allow(dead_code)]
+enum Event {
+    Start,
+    Message { group: u8, actor: u8 },
+}
+
+#[test]
+fn interleavings() {
+    static LOG: Mutex<Vec<Event>> = Mutex::new(vec![]);
+
+    bolero::check!().exhaustive().run(sim(|| {
+        LOG.lock().unwrap().push(Event::Start);
+
+        for group in 0..2 {
+            let (sender, receiver) = Queue::builder().with_capacity(Some(20)).build().channel();
+
+            async move {
+                while let Ok(actor) = receiver.pop().await {
+                    LOG.lock().unwrap().push(Event::Message { group, actor });
+                }
+            }
+            .primary()
+            .spawn_named(format!("[{group}] server"));
+
+            for id in 0..2 {
+                let sender = sender.clone();
+                async move {
+                    for _ in 0..1 {
+                        sender.push(id).await.unwrap();
+                    }
+                }
+                .primary()
+                .spawn_named(format!("[{group}] client{id}"));
+            }
+        }
+    }));
+
+    insta::assert_debug_snapshot!(LOG.lock().unwrap());
+}

bach-tests/src/lib.rs

@@ -0,0 +1,9 @@
+#[global_allocator]
+static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
+#[cfg(test)]
+mod coop;
+#[cfg(test)]
+mod queue;
+#[cfg(test)]
+mod testing;

bach/src/tests.rs → bach-tests/src/queue.rs

@@ -1,4 +1,4 @@
-use crate::{
+use bach::{
     environment::default::Runtime,
     ext::*,
     sync::{duplex::Duplex, queue::vec_deque},

bach-tests/src/snapshots/bach_tests__coop__interleavings.snap

@@ -0,0 +1,40 @@
+---
+source: bach-tests/src/coop.rs
+expression: LOG.lock().unwrap()
+---
+[
+    Start,
+    Message {
+        group: 0,
+        actor: 0,
+    },
+    Message {
+        group: 1,
+        actor: 0,
+    },
+    Message {
+        group: 0,
+        actor: 1,
+    },
+    Message {
+        group: 1,
+        actor: 1,
+    },
+    Start,
+    Message {
+        group: 0,
+        actor: 1,
+    },
+    Message {
+        group: 1,
+        actor: 1,
+    },
+    Message {
+        group: 0,
+        actor: 0,
+    },
+    Message {
+        group: 1,
+        actor: 0,
+    },
+]

bach/src/testing.rs → bach-tests/src/testing.rs

@@ -1,10 +1,4 @@
 pub fn init_tracing() {
-    #[cfg(feature = "tracing")]
-    init_tracing_impl();
-}
-
-#[cfg(feature = "tracing")]
-fn init_tracing_impl() {
     use std::sync::Once;
 
     static TRACING: Once = Once::new();
@@ -26,7 +20,7 @@ fn init_tracing_impl() {
                 w: &mut tracing_subscriber::fmt::format::Writer<'_>,
             ) -> std::fmt::Result {
                 let now =
-                    crate::time::scheduler::scope::try_borrow_mut_with(|s| Some(s.as_ref()?.now()));
+                    bach::time::scheduler::scope::try_borrow_mut_with(|s| Some(s.as_ref()?.now()));
                 if let Some(now) = now {
                     write!(w, "{now}")
                 } else {

bach/Cargo.toml

@@ -9,6 +9,7 @@ edition = "2021"
 
 [features]
 coop = []
+full = ["coop", "metrics", "net", "tracing"]
 metrics = ["dep:metrics"]
 net = []
 tracing = ["dep:tracing"]
@@ -17,6 +18,7 @@ tracing = ["dep:tracing"]
 arr_macro = "0.2"
 async-task = { version = "4", default-features = false }
 atomic-waker = "1"
+bolero-generator.workspace = true
 event-listener-strategy = { version = "0.5.2", default-features = false }
 futures-core = { version = "0.3", default-features = false }
 intrusive-collections = "0.9"
@@ -27,17 +29,10 @@ rand_xoshiro = "0.6"
 tracing = { version = "0.1", optional = true }
 
 [dev-dependencies]
-bolero = "0.11"
-tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+bolero.workspace = true
 
 [lints.rust.unexpected_cfgs]
 level = "warn"
 check-cfg = [
     'cfg(always_disabled)', # used to always disable something
 ]
-
-[package.metadata.cargo-udeps.ignore]
-development = [
-     # keep this so it's easy to debug issues with tracing
-    "tracing-subscriber"
-]

bach/src/coop.rs

@@ -1,37 +1,102 @@
-use crate::{define, task::Info, time::Instant};
+use crate::{define, ext::*};
 use std::{
     collections::{BTreeMap, VecDeque},
     future::Future,
     pin::Pin,
-    sync::Arc,
+    sync::{Arc, Mutex},
     task::{Context, Poll, Waker},
 };
 
 define!(scope, Coop);
 
-pub struct Coop {
+#[derive(Clone, Default)]
+pub struct Coop(Arc<Mutex<State>>);
+
+#[derive(Default)]
+struct State {
     id: u64,
     operations: BTreeMap<Operation, VecDeque<Task>>,
+    moves: Vec<usize>,
+}
+
+impl State {
+    fn schedule(&mut self) -> usize {
+        let mut woken_tasks = 0;
+        let mut max_len = 0;
+
+        // First look at all of the pending tasks and find the `max_len`
+        for tasks in self.operations.values() {
+            woken_tasks += tasks.len();
+            max_len = max_len.max(tasks.len());
+        }
+
+        // Generate a set of interleavings from the `max_len` value
+        //
+        // We generate this once with the assumption that each operation
+        // interleaving is independent from one another. Doing so can drastically
+        // cut down on the required search space.
+        // See: https://en.wikipedia.org/wiki/Partial_order_reduction
+        self.moves.clear();
+        let max_dst = max_len.saturating_sub(1);
+        for src in 0..max_dst {
+            let dst = (src..=max_dst).any();
+            self.moves.push(dst);
+        }
+
+        self.operations.retain(|_operation, tasks| {
+            for (src, dst) in self.moves.iter().copied().enumerate() {
+                // make sure the src applies to this set of tasks
+                if src == tasks.len() {
+                    break;
+                }
+
+                // if dst is in-bounds, then swap it with src. otherwise, leave it in place
+                if dst < tasks.len() {
+                    tasks.swap(src, dst);
+                }
+            }
+
+            for task in tasks.drain(..) {
+                // dropping it wakes it up
+                drop(task)
+            }
+
+            // clear out everything
+            false
+        });
+
+        woken_tasks
+    }
 }
 
 impl Coop {
-    pub fn resource(&mut self) -> Operation {
-        let id = self.id;
-        self.id += 1;
+    pub fn enter<F: FnOnce() -> R, R>(&self, f: F) -> R {
+        scope::with(self.clone(), f)
+    }
+
+    pub fn schedule(&self) -> usize {
+        self.0.lock().unwrap().schedule()
+    }
+
+    fn resource(&mut self) -> Operation {
+        let mut state = self.0.lock().unwrap();
+        let id = state.id;
+        state.id += 1;
         Operation(id)
     }
 
-    pub fn acquire(&mut self, cx: &mut Context<'_>, resource: &Operation) -> Waiting {
+    fn acquire(&mut self, cx: &mut Context<'_>, resource: &Operation) -> Waiting {
         let handle = Arc::new(());
 
         let task = Task {
-            info: Info::current(),
-            instant: Instant::now(),
             waker: cx.waker().clone(),
             handle: handle.clone(),
         };
 
-        self.operations
+        self.0
+            .lock()
+            .unwrap()
+            .operations
             .entry(*resource)
             .or_default()
             .push_back(task);
@@ -73,10 +138,7 @@ impl Operation {
 }
 
 pub struct Task {
-    pub info: Info,
-    pub instant: Instant,
-    pub waker: Waker,
-
+    waker: Waker,
     #[allow(dead_code)] // this just holds the `Waiting` future open
     handle: Arc<()>,
 }

bach/src/environment.rs

@@ -5,7 +5,7 @@ mod macrostep;
 pub use macrostep::Macrostep;
 
 pub trait Environment {
-    fn enter<F: FnOnce() -> O, O>(&self, f: F) -> O;
+    fn enter<F: FnOnce() -> O, O>(&mut self, f: F) -> O;
 
     fn run<Tasks, R>(&mut self, tasks: Tasks) -> Poll<()>
     where