Avoid race condition in crutest rand-read/write (#1261)

Previously, the IO tasks could stop before the stat-gathering main loop,
so you could get nonsensical readings like
```
 713.4 MiB/sec ± 14.2 MiB/sec      (713 IOPS)
 683.6 MiB/sec ± 11.2 MiB/sec      (683 IOPS)
 698.8 MiB/sec ± 409.6 KiB/sec     (698 IOPS)
 722.2 MiB/sec ± 10.2 MiB/sec      (722 IOPS)
 674.4 MiB/sec ± 13.6 MiB/sec      (674 IOPS)
   737 MiB/sec ± 5.8 MiB/sec       (737 IOPS)
 709.6 MiB/sec ± 16.8 MiB/sec      (709 IOPS)
 728.6 MiB/sec ± 10.6 MiB/sec      (728 IOPS)
 723.6 MiB/sec ± 15.2 MiB/sec      (723 IOPS)
 694.6 MiB/sec ± 8.6 MiB/sec       (694 IOPS)
 713.2 MiB/sec ± 7.2 MiB/sec       (713 IOPS)
 stopping guest [0 jobs remaining]
----------------------------------------------
average read performance: 650.1 MiB/sec ± 196.6 MiB/sec (650 IOPS)
```
Notice the average read performance is nowhere near the average values
reported during runtime!

This is because the last two IO samples are captured as the system is
shutting down. Here are the raw samples:
```
[733164339.2, 762943897.6, 728550604.8, 705062502.4, 732325478.4, 733164339.2,
 746586112.0, 767977062.4, 692899020.8, 721420288.0, 778882252.8, 766718771.2,
 726453452.8, 761685606.4, 775107379.2, 752877568.0, 742811238.4, 774687948.8,
 719323136.0, 737358643.2, 755394150.4, 740294656.0, 3774873.6, 0.0]
 ```

Author: mkeeter

crutest/src/main.rs

@@ -2,6 +2,7 @@
 use std::fs::File;
 use std::io::Write;
 use std::net::{IpAddr, SocketAddr};
+use std::num::NonZeroU64;
 use std::path::{Path, PathBuf};
 use std::sync::{
     atomic::{AtomicBool, AtomicUsize, Ordering},
@@ -308,8 +309,8 @@ struct RandReadWriteWorkload {
     #[clap(long, default_value_t = 1.0)]
     sample_time: f64,
     /// Number of subsamples per sample
-    #[clap(long, default_value_t = 10)]
-    subsample_count: u64,
+    #[clap(long, default_value_t = NonZeroU64::new(10).unwrap())]
+    subsample_count: NonZeroU64,
 }
 
 /// Mode flags for `rand_write_read_workload`
@@ -334,7 +335,7 @@ struct RandReadWriteConfig {
     /// Rate at which we should print samples
     sample_time_secs: f64,
     /// Number of subsamples for each `sample_time_secs`, for standard deviation
-    subsample_count: u64,
+    subsample_count: NonZeroU64,
     fill: bool,
 }
 
@@ -2493,43 +2494,34 @@ async fn rand_read_write_workload(
         workers.push(handle);
     }
 
-    // Spawn a task which stops us after the given time
-    {
-        let stop = stop.clone();
-        let time_secs = cfg.time_secs;
-        tokio::task::spawn(async move {
-            tokio::time::sleep(Duration::from_secs(time_secs)).await;
-            stop.store(true, Ordering::Release);
-        });
-    }
-
     // Initial sleep
     let mut prev = 0;
 
     let subsample_delay = Duration::from_secs_f64(
-        cfg.sample_time_secs / cfg.subsample_count as f64,
+        cfg.sample_time_secs / cfg.subsample_count.get() as f64,
     );
     let mut samples = vec![];
     let mut first = true;
-    let mut next_time = std::time::Instant::now() + subsample_delay;
-    while !stop.load(Ordering::Relaxed) {
+    let mut next_time = Instant::now() + subsample_delay;
+    let stop_time = Instant::now() + Duration::from_secs(cfg.time_secs);
+    'outer: loop {
         // Store speeds in bytes/sec, correcting for our sub-second sample time
         let start = samples.len();
-        for _ in 0..cfg.subsample_count {
-            tokio::time::sleep_until(next_time.into()).await;
+        for _ in 0..cfg.subsample_count.get() {
+            tokio::time::sleep_until(next_time).await;
             let bytes = byte_count.load(Ordering::Acquire);
             next_time += subsample_delay;
             samples.push((bytes - prev) as f64 / subsample_delay.as_secs_f64());
             prev = bytes;
+            if Instant::now() >= stop_time {
+                break 'outer;
+            }
         }
         // Ignore the first round of samples, to reduce startup effects
         if std::mem::take(&mut first) {
             samples.clear();
             continue;
         }
-        if stop.load(Ordering::Relaxed) {
-            break;
-        }
         let slice = &samples[start..];
         let mean = slice.iter().sum::<f64>() / slice.len() as f64;
         let stdev = (slice.iter().map(|&d| (d - mean).powi(2)).sum::<f64>()