diff --git a/shuttle/tests/basic/mpsc.rs b/shuttle/tests/basic/mpsc.rs index 7d8795d4..d0c944e2 100644 --- a/shuttle/tests/basic/mpsc.rs +++ b/shuttle/tests/basic/mpsc.rs @@ -502,9 +502,20 @@ fn mpsc_senders_with_blocking_inner(num_senders: usize, channel_size: usize) { } } +// The interleaving space here grows superexponentially in the number of senders: an exhaustive DFS +// of (4, 2) takes 4,339,144 iterations (~30s), versus 35,675 for (3, 1) and 8,117 for (3, 2). +// +// The behavior that matters is the channel's blocked-sender queue, and the maximum number of +// simultaneously blocked senders is `num_senders - channel_size`. So (3, 1) exercises two +// concurrently blocked senders just as (4, 2) did, while (3, 2) covers a buffer holding more than +// one message alongside a blocking sender. Together they keep the search exhaustive and cover both +// dimensions in ~0.3s. #[test] fn mpsc_some_senders_with_blocking() { - check_dfs(|| mpsc_senders_with_blocking_inner(4, 2), None); + // Two senders blocked simultaneously on a full channel. + check_dfs(|| mpsc_senders_with_blocking_inner(3, 1), None); + // Buffer depth > 1 with one blocked sender. + check_dfs(|| mpsc_senders_with_blocking_inner(3, 2), None); } #[test] diff --git a/shuttle/tests/future/batch_semaphore.rs b/shuttle/tests/future/batch_semaphore.rs index 0af29c5f..1e7b78f5 100644 --- a/shuttle/tests/future/batch_semaphore.rs +++ b/shuttle/tests/future/batch_semaphore.rs @@ -279,27 +279,54 @@ fn batch_semaphore_clock_imprecise() { ); } -// Create a semaphore with `num_permits` permits and spawn a bunch of tasks that each -// try to grab a bunch of permits. Task i sets the i'th bit in a shared atomic counter. -// Afterwards, we'll see which combinations were allowable over a full dfs run. +// Create a semaphore with `num_permits` permits and have a bunch of tasks each try to grab a bunch +// of permits. Task i sets the i'th bit in a shared atomic counter while holding its permits, and +// records the counter value it observed *before* setting its own bit — that is, the set of tasks +// that were holding permits at the same time as it. Over a full DFS run this yields the exact set +// of possible co-residencies, which must be exactly those whose permit demands sum to at most +// `num_permits`. +// +// Note that the *last* participant runs on the calling task rather than being spawned. The calling +// task has to exist either way and performs no semaphore operations of its own, so giving every +// participant its own spawned task just adds a schedulable entity that multiplies the interleaving +// space without adding any contention. Folding the last participant into the caller keeps the same +// number of concurrent contenders and produces an identical set of observed states, while cutting +// the exhaustive search roughly 100x: for `(5, [3, 3, 2])` the DFS explores 15,376 interleavings +// instead of 1,554,091, and for `(5, [3, 3, 3])` it explores 4,437 instead of 590,311. +// +// Note also that `future::yield_now` below is load-bearing: it is the window during which a task's +// bit is observable to others. Without it the search is 22x cheaper but every task observes 0, so +// no co-residency is detected at all. async fn semtest(num_permits: usize, counts: Vec, states: &Arc>>, mode: Fairness) { - let s = Arc::new(BatchSemaphore::new(num_permits, mode)); - let r = Arc::new(AtomicUsize::new(0)); - let mut handles = vec![]; - for (i, &c) in counts.iter().enumerate() { - let s = s.clone(); - let r = r.clone(); - let states = states.clone(); + // One participant: acquire `c` permits, publish bit `i` for one scheduling step, then release. + async fn participant( + i: usize, + c: usize, + s: Arc, + r: Arc, + states: Arc>>, + ) { let val = 1usize << i; - handles.push(future::spawn(async move { - s.acquire(c).await.unwrap(); - let v = r.fetch_add(val, Ordering::SeqCst); - future::yield_now().await; - let _ = r.fetch_sub(val, Ordering::SeqCst); - states.lock().unwrap().insert((i, v)); - s.release(c); - })); + s.acquire(c).await.unwrap(); + let v = r.fetch_add(val, Ordering::SeqCst); + future::yield_now().await; + let _ = r.fetch_sub(val, Ordering::SeqCst); + states.lock().unwrap().insert((i, v)); + s.release(c); } + + let s = Arc::new(BatchSemaphore::new(num_permits, mode)); + let r = Arc::new(AtomicUsize::new(0)); + + let (&last, rest) = counts.split_last().expect("need at least one participant"); + let handles = rest + .iter() + .enumerate() + .map(|(i, &c)| future::spawn(participant(i, c, s.clone(), r.clone(), states.clone()))) + .collect::>(); + + participant(counts.len() - 1, last, s.clone(), r.clone(), states.clone()).await; + for h in handles { h.await.unwrap(); } diff --git a/wrappers/tokio/impls/tokio/inner/src/sync/watch.rs b/wrappers/tokio/impls/tokio/inner/src/sync/watch.rs index 73c1119b..0d6d5f86 100644 --- a/wrappers/tokio/impls/tokio/inner/src/sync/watch.rs +++ b/wrappers/tokio/impls/tokio/inner/src/sync/watch.rs @@ -1052,7 +1052,9 @@ mod tests { send_thread.join().unwrap(); }, - 500_000, + // The schedule space for this body is small; 500_000 iterations was well past the point + // of diminishing returns (cost is linear in iterations). + 50_000, ); } @@ -1095,7 +1097,8 @@ mod tests { }); assert!(send.borrow().eq(&2)); }, - 500_000, + // See the note on `watch_spurious_wakeup`. + 50_000, ); } diff --git a/wrappers/tokio/impls/tokio/inner/tests/mpsc.rs b/wrappers/tokio/impls/tokio/inner/tests/mpsc.rs index 3b31a363..f07afc8c 100644 --- a/wrappers/tokio/impls/tokio/inner/tests/mpsc.rs +++ b/wrappers/tokio/impls/tokio/inner/tests/mpsc.rs @@ -314,6 +314,9 @@ fn async_mpsc_some_senders_with_blocking() { ); } +// Each iteration spawns 1000 tasks, so iterations are expensive. The synchronous analogue +// (`shuttle::tests::basic::mpsc::mpsc_many_senders_with_blocking`) uses 10 for the same body; this +// test is about scaling to many senders, not about breadth of schedules, so match that. #[test] fn async_mpsc_many_senders_with_blocking() { shuttle::check_random( @@ -322,7 +325,7 @@ fn async_mpsc_many_senders_with_blocking() { mpsc_senders_with_blocking_inner(1000, 500).await; }); }, - 1000, + 10, ); } diff --git a/wrappers/tokio/impls/tokio/inner/tests/notify.rs b/wrappers/tokio/impls/tokio/inner/tests/notify.rs index 01f219df..0091c4aa 100644 --- a/wrappers/tokio/impls/tokio/inner/tests/notify.rs +++ b/wrappers/tokio/impls/tokio/inner/tests/notify.rs @@ -281,10 +281,18 @@ fn notify_mpmc_channel_2_test(do_enable: bool) { h.push(future::spawn(async move { tx1.send(1); })); - h.push(future::spawn(async move { - tx2.send(2); - })); - futures::future::join_all(h).await; + + // The second send runs on this task rather than in a spawned one. This task has to + // exist regardless and would otherwise do nothing but join, so spawning a fourth task + // for it only adds a schedulable entity that multiplies the interleaving space without + // adding any concurrency to the channel. Joining sequentially rather than with + // `join_all` is likewise cheaper for the same reason. Together these keep the search + // exhaustive while cutting it from 666,570 interleavings to a small fraction. + tx2.send(2); + + for handle in h { + handle.await.unwrap(); + } assert_eq!(counter.load(Ordering::SeqCst), 3); }); }); diff --git a/wrappers/tokio/impls/tokio/inner/tests/runtime.rs b/wrappers/tokio/impls/tokio/inner/tests/runtime.rs index ee31d137..bb8fe8e8 100644 --- a/wrappers/tokio/impls/tokio/inner/tests/runtime.rs +++ b/wrappers/tokio/impls/tokio/inner/tests/runtime.rs @@ -40,6 +40,7 @@ fn runtime_mpsc_many_senders_with_blocking() { mpsc_senders_with_blocking_inner(1000, 500).await; }); }, - 1000, + // 1000 tasks spawned per iteration; see the note on `async_mpsc_many_senders_with_blocking`. + 10, ); } diff --git a/wrappers/tokio/impls/tokio/inner/tests/semaphore.rs b/wrappers/tokio/impls/tokio/inner/tests/semaphore.rs index 0298644c..51f1948d 100644 --- a/wrappers/tokio/impls/tokio/inner/tests/semaphore.rs +++ b/wrappers/tokio/impls/tokio/inner/tests/semaphore.rs @@ -64,24 +64,46 @@ fn semaphore_acquire() { ); } +// Task i acquires `counts[i]` permits, sets bit i in a shared counter while holding them, and +// records which other tasks were holding permits at the same time. A full DFS run therefore yields +// the exact set of possible co-residencies, which must be exactly those whose permit demands sum to +// at most `num_permits`. +// +// The last participant runs on the calling task rather than being spawned: the caller has to exist +// either way and performs no semaphore operations itself, so spawning a task per participant just +// adds a schedulable entity that multiplies the interleaving space without adding contention. This +// keeps the search exhaustive and the observed states identical while cutting `(5, [3, 3, 2])` from +// 1,554,091 interleavings to 15,376. The `yield_now` is load-bearing — it is the window in which a +// task's bit is observable to others. async fn semtest(num_permits: usize, counts: Vec, states: &Arc>>) { - let s = Arc::new(Semaphore::new(num_permits)); - let r = Arc::new(AtomicUsize::new(0)); - let mut handles = vec![]; - for (i, &c) in counts.iter().enumerate() { - let s = s.clone(); - let r = r.clone(); - let states = states.clone(); + async fn participant( + i: usize, + c: usize, + s: Arc, + r: Arc, + states: Arc>>, + ) { let val = 1usize << i; - handles.push(future::spawn(async move { - let permit = s.acquire_many(c as u32).await.unwrap(); - let v = r.fetch_add(val, Ordering::SeqCst); - future::yield_now().await; - let _ = r.fetch_sub(val, Ordering::SeqCst); - states.lock().unwrap().insert((i, v)); - drop(permit); - })); + let permit = s.acquire_many(c as u32).await.unwrap(); + let v = r.fetch_add(val, Ordering::SeqCst); + future::yield_now().await; + let _ = r.fetch_sub(val, Ordering::SeqCst); + states.lock().unwrap().insert((i, v)); + drop(permit); } + + let s = Arc::new(Semaphore::new(num_permits)); + let r = Arc::new(AtomicUsize::new(0)); + + let (&last, rest) = counts.split_last().expect("need at least one participant"); + let handles = rest + .iter() + .enumerate() + .map(|(i, &c)| future::spawn(participant(i, c, s.clone(), r.clone(), states.clone()))) + .collect::>(); + + participant(counts.len() - 1, last, s.clone(), r.clone(), states.clone()).await; + for h in handles { h.await.unwrap(); } diff --git a/wrappers/tokio/impls/tokio/inner/tests/watch.rs b/wrappers/tokio/impls/tokio/inner/tests/watch.rs index 132fc72a..b9ec9a72 100644 --- a/wrappers/tokio/impls/tokio/inner/tests/watch.rs +++ b/wrappers/tokio/impls/tokio/inner/tests/watch.rs @@ -39,7 +39,7 @@ fn watch_basic() { futures::future::join_all([p, h1, h2]).await; }); }, - 200_000, + 20_000, ); } @@ -69,7 +69,9 @@ fn watch_changed() { h.await.unwrap(); }); }, - 200_000, + // The `values` set below reaches its final contents within the first handful of iterations, + // so 200_000 was three orders of magnitude more than the assertion needs. + 20_000, ); let values = values.lock().unwrap(); @@ -120,7 +122,7 @@ fn wait_for_test() { th1.join().unwrap(); th2.join().unwrap(); }, - 50_000, + 10_000, ); }