Make test suite finish faster

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]

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<usize>, states: &Arc<Mutex<HashSet<(usize, usize)>>>, 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<BatchSemaphore>,
+        r: Arc<AtomicUsize>,
+        states: Arc<Mutex<HashSet<(usize, usize)>>>,
+    ) {
         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::<Vec<_>>();
+
+    participant(counts.len() - 1, last, s.clone(), r.clone(), states.clone()).await;
+
     for h in handles {
         h.await.unwrap();
     }

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,
         );
     }
 

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,
     );
 }
 

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);
         });
     });

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,
     );
 }

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<usize>, states: &Arc<Mutex<HashSet<(usize, usize)>>>) {
-    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<Semaphore>,
+        r: Arc<AtomicUsize>,
+        states: Arc<Mutex<HashSet<(usize, usize)>>>,
+    ) {
         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::<Vec<_>>();
+
+    participant(counts.len() - 1, last, s.clone(), r.clone(), states.clone()).await;
+
     for h in handles {
         h.await.unwrap();
     }

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,
     );
 }