Make the Guest -> Upstairs queue fully async (#1086)

The upstairs `struct Guest` is currently used in at least two different tasks:
- The guest uses the `trait BlockIO` to push requests into a queue (protected by
  a `tokio::sync::Mutex`).  There can be multiple tasks sharing an `Arc<Guest>`
  here, e.g. 8x in Propolis.
- The `struct Upstairs` pulls requests from that queue in the `up_main` event
  loop

This PR splits the `Guest` into two separate `structs`, which are **owned** (not
shared) by their respective tasks:
- `Guest` (same name as before) implements `BlockIO` and is used by the guest
  task to send new jobs
- `GuestIoHandle` (new!) is used by the `struct Upstairs` and receives those
  jobs

The two halves communicate via a `tokio::sync::mpsc` queue, removing the
previous combination of `tokio::sync::Mutex<VecDeque<BlockReq>> +
tokio::sync::Notify`

It's still possible to share a `Guest` by wrapping it in an `Arc`, just like
before.

As a side effect of removing the `tokio::sync::Mutex`, many functions become
synchronous; don't be alarmed by the LOC  in this PR, since they're mostly
mechanical.

I don't see any performance changes here (alas), but I think it's still
worthwhile: it simplifies our code and helps set up a future consolidation of
backpressure logic.

Author: mkeeter

crudd/src/main.rs

@@ -571,10 +571,10 @@ async fn main() -> Result<()> {
     };
 
     // TODO: volumes?
-    let guest = Arc::new(Guest::new(None));
+    let (guest, io) = Guest::new(None);
+    let guest = Arc::new(guest);
 
-    let _join_handle =
-        up_main(crucible_opts, opt.gen, None, guest.clone(), None)?;
+    let _join_handle = up_main(crucible_opts, opt.gen, None, io, None)?;
     eprintln!("Crucible runtime is spawned");
 
     // IO time

crutest/src/main.rs

@@ -668,7 +668,8 @@ async fn main() -> Result<()> {
      * We create this here instead of inside up_main() so we can use
      * the methods provided by guest to interact with Crucible.
      */
-    let guest = Arc::new(Guest::new(None));
+    let (guest, io) = Guest::new(None);
+    let guest = Arc::new(guest);
 
     let pr;
     if opt.metrics {
@@ -698,8 +699,7 @@ async fn main() -> Result<()> {
         pr = None;
     }
 
-    let _join_handle =
-        up_main(crucible_opts, opt.gen, None, guest.clone(), pr)?;
+    let _join_handle = up_main(crucible_opts, opt.gen, None, io, pr)?;
     println!("Crucible runtime is spawned");
 
     if let Workload::CliServer { listen, port } = opt.workload {

hammer/src/main.rs

@@ -1,7 +1,6 @@
 // Copyright 2023 Oxide Computer Company
 
 use std::net::SocketAddr;
-use std::sync::Arc;
 
 use anyhow::{bail, Result};
 use clap::Parser;
@@ -130,11 +129,10 @@ async fn main() -> Result<()> {
          * We create this here instead of inside up_main() so we can use
          * the methods provided by guest to interact with Crucible.
          */
-        let guest = Arc::new(Guest::new(None));
+        let (guest, io) = Guest::new(None);
 
         let gen: u64 = i as u64 + opt.gen;
-        let _join_handle =
-            up_main(crucible_opts.clone(), gen, None, guest.clone(), None)?;
+        let _join_handle = up_main(crucible_opts.clone(), gen, None, io, None)?;
         println!("Crucible runtime is spawned");
 
         cpfs.push(crucible::CruciblePseudoFile::from(guest)?);

integration_tests/src/lib.rs

@@ -3058,10 +3058,9 @@ mod test {
         let tds = TestDownstairsSet::small(false).await?;
         let opts = tds.opts();
 
-        let guest = Arc::new(Guest::new(None));
-        let gc = guest.clone();
+        let (guest, io) = Guest::new(None);
 
-        let _join_handle = up_main(opts, 1, None, gc, None)?;
+        let _join_handle = up_main(opts, 1, None, io, None)?;
 
         guest.activate().await?;
         guest.query_work_queue().await?;
@@ -3093,6 +3092,70 @@ mod test {
         Ok(())
     }
 
+    /// Drop the guest right away and confirm that the worker thread stops
+    #[tokio::test]
+    async fn integration_test_guest_drop_early() -> Result<()> {
+        // Spin off three downstairs, build our Crucible struct.
+        let tds = TestDownstairsSet::small(false).await?;
+        let opts = tds.opts();
+
+        let (guest, io) = Guest::new(None);
+
+        let join_handle = up_main(opts, 1, None, io, None)?;
+
+        drop(guest);
+
+        let r = join_handle.await;
+        assert!(r.is_ok());
+        Ok(())
+    }
+
+    /// Same as `integration_test_guest_downstairs`, but dropping the Guest at
+    /// the end and confirming that the worker thread stops
+    #[tokio::test]
+    async fn integration_test_guest_drop() -> Result<()> {
+        const BLOCK_SIZE: usize = 512;
+        // Spin off three downstairs, build our Crucible struct.
+        let tds = TestDownstairsSet::small(false).await?;
+        let opts = tds.opts();
+
+        let (guest, io) = Guest::new(None);
+
+        let join_handle = up_main(opts, 1, None, io, None)?;
+
+        guest.activate().await?;
+        guest.query_work_queue().await?;
+
+        // Verify contents are zero on init
+        let mut buffer = Buffer::new(10, BLOCK_SIZE);
+        guest
+            .read(Block::new(0, BLOCK_SIZE.trailing_zeros()), &mut buffer)
+            .await?;
+
+        assert_eq!(vec![0x00_u8; BLOCK_SIZE * 10], buffer.to_vec());
+
+        // Write data in
+        guest
+            .write(
+                Block::new(0, BLOCK_SIZE.trailing_zeros()),
+                Bytes::from(vec![0x55; BLOCK_SIZE * 10]),
+            )
+            .await?;
+
+        // Read parent, verify contents
+        let mut buffer = Buffer::new(10, BLOCK_SIZE);
+        guest
+            .read(Block::new(0, BLOCK_SIZE.trailing_zeros()), &mut buffer)
+            .await?;
+
+        assert_eq!(vec![0x55_u8; BLOCK_SIZE * 10], buffer.to_vec());
+        drop(guest);
+
+        let r = join_handle.await;
+        assert!(r.is_ok());
+        Ok(())
+    }
+
     #[tokio::test]
     async fn integration_test_guest_zero_length_io() -> Result<()> {
         // Test the guest layer with a write and read of zero length
@@ -3102,10 +3165,9 @@ mod test {
         let tds = TestDownstairsSet::small(false).await?;
         let opts = tds.opts();
 
-        let guest = Arc::new(Guest::new(None));
-        let gc = guest.clone();
+        let (guest, io) = Guest::new(None);
 
-        let _join_handle = up_main(opts, 1, None, gc, None)?;
+        let _join_handle = up_main(opts, 1, None, io, None)?;
 
         guest.activate().await?;
         guest.query_work_queue().await?;
@@ -3138,9 +3200,8 @@ mod test {
         let opts = tds.opts();
 
         let log = csl();
-        let guest = Arc::new(Guest::new(Some(log.clone())));
-        let gc = guest.clone();
-        let _jh = up_main(opts, 1, None, gc, None)?;
+        let (guest, io) = Guest::new(Some(log.clone()));
+        let _jh = up_main(opts, 1, None, io, None)?;
 
         guest.activate().await?;
 
@@ -3212,11 +3273,10 @@ mod test {
         let tds = TestDownstairsSet::small(true).await?;
         let opts = tds.opts();
 
-        let guest = Arc::new(Guest::new(None));
-        let gc = guest.clone();
+        let (guest, io) = Guest::new(None);
 
         // Read-only Upstairs should return errors if writes are attempted.
-        let _join_handle = up_main(opts, 1, None, gc, None)?;
+        let _join_handle = up_main(opts, 1, None, io, None)?;
 
         guest.activate().await?;
 
@@ -3246,9 +3306,8 @@ mod test {
         let opts = tds.opts();
 
         let log = csl();
-        let guest = Arc::new(Guest::new(Some(log.clone())));
-        let gc = guest.clone();
-        let _jh = up_main(opts, 1, None, gc, None)?;
+        let (guest, io) = Guest::new(Some(log.clone()));
+        let _jh = up_main(opts, 1, None, io, None)?;
 
         guest.activate().await?;
 
@@ -3293,10 +3352,9 @@ mod test {
         let tds = TestDownstairsSet::small(false).await?;
         let opts = tds.opts();
 
-        let guest = Arc::new(Guest::new(None));
-        let gc = guest.clone();
+        let (guest, io) = Guest::new(None);
 
-        let _join_handle = up_main(opts, 1, None, gc, None)?;
+        let _join_handle = up_main(opts, 1, None, io, None)?;
 
         guest.activate().await?;
         guest.query_work_queue().await?;
@@ -3366,10 +3424,9 @@ mod test {
         let tds = TestDownstairsSet::small(false).await?;
         let opts = tds.opts();
 
-        let guest = Arc::new(Guest::new(None));
-        let gc = guest.clone();
+        let (guest, io) = Guest::new(None);
 
-        let _join_handle = up_main(opts, 1, None, gc, None)?;
+        let _join_handle = up_main(opts, 1, None, io, None)?;
 
         guest.activate().await?;
         guest.query_work_queue().await?;
@@ -3424,10 +3481,9 @@ mod test {
         let tds = TestDownstairsSet::small(false).await?;
         let opts = tds.opts();
 
-        let guest = Arc::new(Guest::new(None));
-        let gc = guest.clone();
+        let (guest, io) = Guest::new(None);
 
-        let _join_handle = up_main(opts, 1, None, gc, None)?;
+        let _join_handle = up_main(opts, 1, None, io, None)?;
 
         guest.activate().await?;
         guest.query_work_queue().await?;
@@ -3483,10 +3539,9 @@ mod test {
         let tds = TestDownstairsSet::small(false).await?;
         let opts = tds.opts();
 
-        let guest = Arc::new(Guest::new(None));
-        let gc = guest.clone();
+        let (guest, io) = Guest::new(None);
 
-        let _join_handle = up_main(opts, 1, None, gc, None)?;
+        let _join_handle = up_main(opts, 1, None, io, None)?;
 
         guest.activate().await?;
         guest.query_work_queue().await?;
@@ -3537,10 +3592,9 @@ mod test {
         let tds = TestDownstairsSet::small(false).await?;
         let opts = tds.opts();
 
-        let guest = Arc::new(Guest::new(None));
-        let gc = guest.clone();
+        let (guest, io) = Guest::new(None);
 
-        let _join_handle = up_main(opts, 1, None, gc, None)?;
+        let _join_handle = up_main(opts, 1, None, io, None)?;
 
         guest.activate().await?;
         guest.query_work_queue().await?;
@@ -3591,10 +3645,9 @@ mod test {
         let tds = TestDownstairsSet::small(false).await?;
         let opts = tds.opts();
 
-        let guest = Arc::new(Guest::new(None));
-        let gc = guest.clone();
+        let (guest, io) = Guest::new(None);
 
-        let _join_handle = up_main(opts, 1, None, gc, None)?;
+        let _join_handle = up_main(opts, 1, None, io, None)?;
 
         guest.activate().await?;
         guest.query_work_queue().await?;
@@ -3643,10 +3696,9 @@ mod test {
         let tds = TestDownstairsSet::small(false).await.unwrap();
         let opts = tds.opts();
 
-        let guest = Arc::new(Guest::new(None));
-        let gc = guest.clone();
+        let (guest, io) = Guest::new(None);
 
-        let _join_handle = up_main(opts, 1, None, gc, None).unwrap();
+        let _join_handle = up_main(opts, 1, None, io, None).unwrap();
 
         guest.activate().await.unwrap();
 
@@ -3679,10 +3731,9 @@ mod test {
         let tds = TestDownstairsSet::small(false).await.unwrap();
         let opts = tds.opts();
 
-        let guest = Arc::new(Guest::new(None));
-        let gc = guest.clone();
+        let (guest, io) = Guest::new(None);
 
-        let _join_handle = up_main(opts, 1, None, gc, None).unwrap();
+        let _join_handle = up_main(opts, 1, None, io, None).unwrap();
 
         guest.activate().await.unwrap();
 

measure_iops/src/main.rs

@@ -101,19 +101,19 @@ async fn main() -> Result<()> {
         std::process::exit(1);
     }));
 
-    let mut guest = Guest::new(None);
+    let (guest, mut io) = Guest::new(None);
+    let guest = Arc::new(guest);
 
     if let Some(iop_limit) = opt.iop_limit {
-        guest.set_iop_limit(16 * 1024 * 1024, iop_limit);
+        io.set_iop_limit(16 * 1024 * 1024, iop_limit);
     }
 
     if let Some(bw_limit) = opt.bw_limit_in_bytes {
-        guest.set_bw_limit(bw_limit);
+        io.set_bw_limit(bw_limit);
     }
 
     let guest = Arc::new(guest);
-    let _join_handle =
-        up_main(crucible_opts, opt.gen, None, guest.clone(), None)?;
+    let _join_handle = up_main(crucible_opts, opt.gen, None, io, None)?;
     println!("Crucible runtime is spawned");
 
     guest.activate().await?;

nbd_server/src/main.rs

@@ -1,6 +1,5 @@
 // Copyright 2021 Oxide Computer Company
 use std::net::SocketAddr;
-use std::sync::Arc;
 
 use anyhow::{bail, Result};
 use clap::Parser;
@@ -91,10 +90,9 @@ async fn main() -> Result<()> {
      * We create this here instead of inside up_main() so we can use
      * the methods provided by guest to interact with Crucible.
      */
-    let guest = Arc::new(Guest::new(None));
+    let (guest, io) = Guest::new(None);
 
-    let _join_handle =
-        up_main(crucible_opts, opt.gen, None, guest.clone(), None)?;
+    let _join_handle = up_main(crucible_opts, opt.gen, None, io, None)?;
     println!("Crucible runtime is spawned");
 
     // NBD server

upstairs/src/client.rs

@@ -2311,7 +2311,12 @@ async fn client_run(
     .await;
 
     warn!(log, "client task is sending Done({r:?})");
-    tx.send(ClientResponse::Done(r)).await.unwrap();
+    if tx.send(ClientResponse::Done(r)).await.is_err() {
+        warn!(
+            log,
+            "client task could not reply to main task; shutting down?"
+        );
+    }
     while let Some(v) = rx.recv().await {
         warn!(log, "exiting client task is ignoring message {v:?}");
     }

upstairs/src/dummy_downstairs_tests.rs

@@ -522,8 +522,8 @@ pub(crate) mod protocol_test {
 
             // Configure our guest without queue backpressure, to speed up tests
             // which require triggering a timeout
-            let mut g = Guest::new(Some(log.clone()));
-            g.backpressure_config.queue_max_delay = Duration::ZERO;
+            let (g, mut io) = Guest::new(Some(log.clone()));
+            io.backpressure_config.queue_max_delay = Duration::ZERO;
             let guest = Arc::new(g);
 
             let crucible_opts = CrucibleOpts {
@@ -536,7 +536,7 @@ pub(crate) mod protocol_test {
             };
 
             let join_handle =
-                up_main(crucible_opts, 1, None, guest.clone(), None).unwrap();
+                up_main(crucible_opts, 1, None, io, None).unwrap();
 
             let mut handles: Vec<JoinHandle<Result<()>>> = vec![];