vmcore/

vm_task.rs

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

//! Infrastructure for spawning tasks and issuing async IO related to VM
//! activity.

// UNSAFETY: Needed to implement the unsafe new_dyn_overlapped_file method on
// Windows and to implement the unsafe io_uring_submit method on Linux.
#![cfg_attr(any(windows, target_os = "linux"), expect(unsafe_code))]

use inspect::Inspect;
use pal_async::driver::Driver;
use pal_async::task::Spawn;
use pal_async::task::TaskMetadata;
use std::fmt::Debug;
use std::pin::Pin;
use std::sync::Arc;

/// A source for [`VmTaskDriver`]s.
///
/// This is used to create device-specific drivers that implement [`Driver`] and
/// [`Spawn`]. These drivers' behavior can be customized based on the needs of
/// the device.
///
/// The backend for these drivers is customizable for different environments.
#[derive(Clone)]
pub struct VmTaskDriverSource {
    backend: Arc<dyn DynVmBackend>,
}

impl VmTaskDriverSource {
    /// Returns a new task driver source backed by `backend`.
    pub fn new(backend: impl 'static + BuildVmTaskDriver) -> Self {
        Self {
            backend: Arc::new(backend),
        }
    }

    /// Returns a VM task driver with default parameters.
    ///
    /// Use this when you don't care where your task runs.
    pub fn simple(&self) -> VmTaskDriver {
        // Don't provide a name, since backends won't do anything with it for
        // default settings.
        self.builder().build("")
    }

    /// Returns a driver that dispatches to the current thread's executor.
    ///
    /// Use this when a shared resource (e.g., a disk) should use whatever
    /// executor its caller is running on, rather than a fixed executor
    /// captured at construction time.
    pub fn current(&self) -> VmTaskDriver {
        VmTaskDriver {
            inner: self.backend.build_current(),
        }
    }

    /// Returns a builder for a custom VM task driver.
    pub fn builder(&self) -> VmTaskDriverBuilder<'_> {
        VmTaskDriverBuilder {
            backend: self.backend.as_ref(),
            run_on_target: false,
            target_vp: None,
        }
    }
}

/// Trait implemented by backends for [`VmTaskDriverSource`].
pub trait BuildVmTaskDriver: Send + Sync {
    /// The associated driver type.
    type Driver: TargetedDriver;
    /// The driver type for `build_current`.
    type CurrentDriver: TargetedDriver;

    /// Builds a new driver that can drive IO and spawn tasks.
    fn build(&self, name: String, target_vp: Option<u32>, run_on_target: bool) -> Self::Driver;

    /// Builds a driver that dispatches to the current thread's executor.
    fn build_current(&self) -> Self::CurrentDriver;
}

/// Trait implemented by drivers built with [`BuildVmTaskDriver`].
pub trait TargetedDriver: 'static + Send + Sync + Inspect {
    /// Returns the implementation to use for spawning tasks.
    fn spawner(&self) -> &dyn Spawn;
    /// Returns the implementation to use for driving IO.
    fn driver(&self) -> &dyn Driver;
    /// Retargets the driver to the specified virtual processor.
    fn retarget_vp(&self, target_vp: u32);
    /// Returns whether a driver's target VP is ready for tasks and IO.
    ///
    /// A driver must be operable even if this is false, but the tasks and IO
    /// may run on a different target VP.
    fn is_target_vp_ready(&self) -> bool {
        true
    }
    /// Waits for this driver's target VP to be ready for tasks and IO.
    fn wait_target_vp_ready(&self) -> impl Future<Output = ()> + Send {
        std::future::ready(())
    }
}

trait DynTargetedDriver: 'static + Send + Sync + Inspect {
    fn spawner(&self) -> &dyn Spawn;
    fn driver(&self) -> &dyn Driver;
    fn retarget_vp(&self, target_vp: u32);
    fn is_ready(&self) -> bool;
    fn wait_ready(&self) -> Pin<Box<dyn '_ + Future<Output = ()> + Send>>;
}

impl<T: TargetedDriver> DynTargetedDriver for T {
    fn spawner(&self) -> &dyn Spawn {
        self.spawner()
    }

    fn driver(&self) -> &dyn Driver {
        self.driver()
    }

    fn retarget_vp(&self, target_vp: u32) {
        self.retarget_vp(target_vp)
    }

    fn is_ready(&self) -> bool {
        self.is_target_vp_ready()
    }

    fn wait_ready(&self) -> Pin<Box<dyn '_ + Future<Output = ()> + Send>> {
        Box::pin(self.wait_target_vp_ready())
    }
}

trait DynVmBackend: Send + Sync {
    fn build(
        &self,
        name: String,
        target_vp: Option<u32>,
        run_on_target: bool,
    ) -> Arc<dyn DynTargetedDriver>;

    fn build_current(&self) -> Arc<dyn DynTargetedDriver>;
}

impl<T: BuildVmTaskDriver> DynVmBackend for T {
    fn build(
        &self,
        name: String,
        target_vp: Option<u32>,
        run_on_target: bool,
    ) -> Arc<dyn DynTargetedDriver> {
        Arc::new(self.build(name, target_vp, run_on_target))
    }

    fn build_current(&self) -> Arc<dyn DynTargetedDriver> {
        Arc::new(BuildVmTaskDriver::build_current(self))
    }
}

/// A builder returned by [`VmTaskDriverSource::builder`].
pub struct VmTaskDriverBuilder<'a> {
    backend: &'a dyn DynVmBackend,
    run_on_target: bool,
    target_vp: Option<u32>,
}

impl VmTaskDriverBuilder<'_> {
    /// A hint to the backend specifying whether spawned tasks should always
    /// run on the thread handling the target VP.
    ///
    /// If `false` (the default), then when spawned tasks are awoken, they may
    /// run on any executor (such as the current one). If `true`, the backend
    /// will endeavor to run them on the same thread that would drive async IO.
    ///
    /// Some devices will want to override the default to reduce jitter or
    /// ensure that IO is issued from the correct processor. For example,
    /// StorVSP sets this to true for its channel workers, so that all of a
    /// channel's IO and tasks ideally run on the same VP's thread.
    pub fn run_on_target(&mut self, run_on_target: bool) -> &mut Self {
        self.run_on_target = run_on_target;
        self
    }

    /// A hint to the backend specifying the guest VP associated with spawned
    /// tasks and IO.
    ///
    /// Backends can use this to ensure that spawned tasks and async IO will run
    /// near or on the target VP. For example, StorVSP sets this to the VP
    /// from the VMBus channel open request's `target_vp` field, so that
    /// each channel's worker runs on the VP the guest specified.
    ///
    /// If not set, the backend uses its default scheduling (typically the
    /// calling thread or a shared pool).
    pub fn target_vp(&mut self, target_vp: u32) -> &mut Self {
        self.target_vp = Some(target_vp);
        self
    }

    /// Builds a VM task driver.
    ///
    /// `name` is used by some backends to identify a spawned thread. It is
    /// ignored by other backends.
    pub fn build(&self, name: impl Into<String>) -> VmTaskDriver {
        VmTaskDriver {
            inner: self
                .backend
                .build(name.into(), self.target_vp, self.run_on_target),
        }
    }
}

/// A driver returned by [`VmTaskDriverSource`].
///
/// This can be used to spawn tasks (via [`Spawn`]) and issue async IO (via [`Driver`]).
#[derive(Clone, Inspect)]
pub struct VmTaskDriver {
    #[inspect(flatten)]
    inner: Arc<dyn DynTargetedDriver>,
}

impl VmTaskDriver {
    /// Updates the target VP for the task.
    ///
    /// The effectiveness of this call, and when it would take effect,
    /// depends on the backend. For example, in the OpenHCL threadpool backend,
    /// this will cause new IO to target the new VP's thread, but
    /// existing IO will continue to run on the original VP's thread.
    pub fn retarget_vp(&self, target_vp: u32) {
        self.inner.retarget_vp(target_vp)
    }

    /// Returns whether the target VP is ready for tasks and IO.
    ///
    /// A driver must be operable even if this is false, but the tasks and IO
    /// may run on a different target VP.
    pub fn is_target_vp_ready(&self) -> bool {
        self.inner.is_ready()
    }

    /// Waits for the target VP to be ready for tasks and IO.
    pub async fn wait_target_vp_ready(&self) {
        self.inner.wait_ready().await
    }
}

impl Driver for VmTaskDriver {
    fn new_dyn_timer(&self) -> pal_async::driver::PollImpl<dyn pal_async::timer::PollTimer> {
        self.inner.driver().new_dyn_timer()
    }

    #[cfg(unix)]
    fn new_dyn_fd_ready(
        &self,
        fd: std::os::fd::RawFd,
    ) -> std::io::Result<pal_async::driver::PollImpl<dyn pal_async::fd::PollFdReady>> {
        self.inner.driver().new_dyn_fd_ready(fd)
    }

    #[cfg(unix)]
    fn new_dyn_socket_ready(
        &self,
        socket: std::os::fd::RawFd,
    ) -> std::io::Result<pal_async::driver::PollImpl<dyn pal_async::socket::PollSocketReady>> {
        self.inner.driver().new_dyn_socket_ready(socket)
    }

    #[cfg(windows)]
    fn new_dyn_socket_ready(
        &self,
        socket: std::os::windows::io::RawSocket,
    ) -> std::io::Result<pal_async::driver::PollImpl<dyn pal_async::socket::PollSocketReady>> {
        self.inner.driver().new_dyn_socket_ready(socket)
    }

    #[cfg(unix)]
    fn new_dyn_wait(
        &self,
        fd: std::os::fd::RawFd,
        read_size: usize,
    ) -> std::io::Result<pal_async::driver::PollImpl<dyn pal_async::wait::PollWait>> {
        self.inner.driver().new_dyn_wait(fd, read_size)
    }

    #[cfg(windows)]
    fn new_dyn_wait(
        &self,
        handle: std::os::windows::io::RawHandle,
    ) -> std::io::Result<pal_async::driver::PollImpl<dyn pal_async::wait::PollWait>> {
        self.inner.driver().new_dyn_wait(handle)
    }

    #[cfg(windows)]
    unsafe fn new_dyn_overlapped_file(
        &self,
        handle: std::os::windows::io::RawHandle,
    ) -> std::io::Result<
        pal_async::driver::PollImpl<dyn pal_async::windows::overlapped::IoOverlapped>,
    > {
        // SAFETY: passthru from caller
        unsafe { self.inner.driver().new_dyn_overlapped_file(handle) }
    }

    #[cfg(target_os = "linux")]
    fn io_uring_probe(&self, opcode: u8) -> bool {
        self.inner.driver().io_uring_probe(opcode)
    }

    #[cfg(target_os = "linux")]
    unsafe fn io_uring_submit(
        &self,
        sqe: pal_async::io_uring::Entry,
    ) -> Pin<Box<dyn Future<Output = std::io::Result<i32>> + Send + '_>> {
        // SAFETY: passthru from caller
        unsafe { self.inner.driver().io_uring_submit(sqe) }
    }

    #[cfg(target_os = "macos")]
    fn new_dyn_process_wait(
        &self,
        pid: i32,
    ) -> std::io::Result<pal_async::driver::PollImpl<dyn pal_async::process::macos::PollProcessWait>>
    {
        self.inner.driver().new_dyn_process_wait(pid)
    }
}

impl Spawn for VmTaskDriver {
    fn scheduler(&self, metadata: &TaskMetadata) -> Arc<dyn pal_async::task::Schedule> {
        self.inner.spawner().scheduler(metadata)
    }
}

/// A backend that spawns all tasks and IO on a single driver.
#[derive(Debug)]
pub struct SingleDriverBackend<T>(T);

impl<T: Driver + Spawn + Clone> SingleDriverBackend<T> {
    /// Returns a new driver backend that spawns all tasks and IO on `driver`,
    /// regardless of policy.
    pub fn new(driver: T) -> Self {
        Self(driver)
    }
}

/// The driver for [`SingleDriverBackend`].
#[derive(Debug)]
pub struct SingleDriver<T>(T);

impl<T> Inspect for SingleDriver<T> {
    fn inspect(&self, req: inspect::Request<'_>) {
        req.ignore();
    }
}

impl<T: Driver + Spawn + Clone> BuildVmTaskDriver for SingleDriverBackend<T> {
    type Driver = SingleDriver<T>;
    type CurrentDriver = SingleDriver<T>;

    fn build(&self, _name: String, _target_vp: Option<u32>, _run_on_target: bool) -> Self::Driver {
        SingleDriver(self.0.clone())
    }

    fn build_current(&self) -> Self::CurrentDriver {
        SingleDriver(self.0.clone())
    }
}

impl<T: Driver + Spawn> TargetedDriver for SingleDriver<T> {
    fn spawner(&self) -> &dyn Spawn {
        &self.0
    }

    fn driver(&self) -> &dyn Driver {
        &self.0
    }

    fn retarget_vp(&self, _target_vp: u32) {}
}

pub mod thread {
    //! Provides a thread-based task VM task driver backend
    //! [`ThreadDriverBackend`].

    use super::BuildVmTaskDriver;
    use super::TargetedDriver;
    use inspect::Inspect;
    use loan_cell::LoanCell;
    use pal_async::DefaultDriver;
    use pal_async::DefaultPool;
    use pal_async::WeakDefaultDriver;
    use pal_async::driver::Driver;
    use pal_async::task::Spawn;
    use pal_async::task::TaskMetadata;
    use std::sync::Arc;

    thread_local! {
        // Holds a WEAK reference on purpose: this is lent for the entire lifetime
        // of a dedicated pool thread's `run()`, and a strong reference would keep
        // the pool's task queue (and its IO backend fds) alive forever, so the
        // thread could never exit once its device drops the driver. See `build`.
        static CURRENT_DRIVER: LoanCell<WeakDefaultDriver> = const { LoanCell::new() };
    }

    /// A backend for [`VmTaskDriverSource`](super::VmTaskDriverSource) based on
    /// individual threads.
    ///
    /// If no target VP is specified, this backend will spawn tasks and IO a
    /// default single-threaded IO driver. If a target VP is specified, the
    /// backend will spawn a separate thread and spawn tasks and IOs there.
    #[derive(Debug)]
    pub struct ThreadDriverBackend {
        default_driver: DefaultDriver,
    }

    impl ThreadDriverBackend {
        /// Returns a new backend, using `default_driver` to back task drivers
        /// that did not specify a target VP.
        pub fn new(default_driver: DefaultDriver) -> Self {
            Self { default_driver }
        }
    }

    impl BuildVmTaskDriver for ThreadDriverBackend {
        type Driver = ThreadDriver;
        type CurrentDriver = CurrentThreadDriver;

        fn build(
            &self,
            name: String,
            target_vp: Option<u32>,
            _run_on_target: bool,
        ) -> Self::Driver {
            // Build a standalone thread for this device if a target VP was specified.
            if target_vp.is_some() {
                let pool = DefaultPool::new();
                let driver = pool.driver();
                // Publish a *weak* reference to this pool's driver into the thread
                // local so that resources built with `VmTaskDriverSource::current()`
                // (e.g. block-device disks) dispatch their IO onto this thread while
                // they run here. It must be weak: `pool.run()` returns only once the
                // last strong driver is dropped, and this reference is lent for the
                // whole of `run()`. A strong clone would be a self-reference that
                // keeps the queue open forever, so the thread (and its epoll/eventfd)
                // would never be reclaimed after the device closes its channel --
                // leaking a thread and fds on every guest reset.
                let tls_driver = driver.downgrade();
                std::thread::Builder::new()
                    .name(name)
                    .spawn(move || {
                        CURRENT_DRIVER.with(|cell| cell.lend(&tls_driver, || pool.run()));
                    })
                    .unwrap();
                ThreadDriver {
                    inner: driver,
                    has_dedicated_thread: true,
                }
            } else {
                ThreadDriver {
                    inner: self.default_driver.clone(),
                    has_dedicated_thread: false,
                }
            }
        }

        fn build_current(&self) -> Self::CurrentDriver {
            CurrentThreadDriver {
                default: self.default_driver.clone(),
            }
        }
    }

    /// A driver targeting a fixed executor thread.
    #[derive(Debug, Inspect)]
    pub struct ThreadDriver {
        #[inspect(skip)]
        inner: DefaultDriver,
        has_dedicated_thread: bool,
    }

    impl TargetedDriver for ThreadDriver {
        fn spawner(&self) -> &dyn Spawn {
            &self.inner
        }

        fn driver(&self) -> &dyn Driver {
            &self.inner
        }

        fn retarget_vp(&self, _target_vp: u32) {}
    }

    /// A driver that dispatches to the current thread's registered executor.
    ///
    /// If the current thread has no registered executor (i.e., it is not a
    /// thread spawned by [`ThreadDriverBackend`]), falls back to a default
    /// driver.
    #[derive(Inspect)]
    pub struct CurrentThreadDriver {
        #[inspect(skip)]
        default: DefaultDriver,
    }

    impl CurrentThreadDriver {
        fn with_driver<R>(&self, f: impl FnOnce(&DefaultDriver) -> R) -> R {
            CURRENT_DRIVER.with(|cell| {
                cell.borrow(|weak| {
                    // The thread local only holds a weak reference (see `build`),
                    // so upgrade it transiently for the duration of `f`. A borrow
                    // only ever runs from a task on the pool's own thread, and
                    // `TaskQueue::run` holds that task's scheduler until the task
                    // is fully torn down, so the upgrade succeeds even when the
                    // device has already dropped its driver. Fall back to the
                    // default when not on a dedicated pool thread.
                    match weak.and_then(|w| w.upgrade()) {
                        Some(driver) => f(&driver),
                        None => f(&self.default),
                    }
                })
            })
        }
    }

    impl Driver for CurrentThreadDriver {
        fn new_dyn_timer(&self) -> pal_async::driver::PollImpl<dyn pal_async::timer::PollTimer> {
            self.with_driver(|d| d.new_dyn_timer())
        }

        #[cfg(unix)]
        fn new_dyn_fd_ready(
            &self,
            fd: std::os::fd::RawFd,
        ) -> std::io::Result<pal_async::driver::PollImpl<dyn pal_async::fd::PollFdReady>> {
            self.with_driver(|d| d.new_dyn_fd_ready(fd))
        }

        #[cfg(unix)]
        fn new_dyn_socket_ready(
            &self,
            socket: std::os::fd::RawFd,
        ) -> std::io::Result<pal_async::driver::PollImpl<dyn pal_async::socket::PollSocketReady>>
        {
            self.with_driver(|d| d.new_dyn_socket_ready(socket))
        }

        #[cfg(windows)]
        fn new_dyn_socket_ready(
            &self,
            socket: std::os::windows::io::RawSocket,
        ) -> std::io::Result<pal_async::driver::PollImpl<dyn pal_async::socket::PollSocketReady>>
        {
            self.with_driver(|d| d.new_dyn_socket_ready(socket))
        }

        #[cfg(unix)]
        fn new_dyn_wait(
            &self,
            fd: std::os::fd::RawFd,
            read_size: usize,
        ) -> std::io::Result<pal_async::driver::PollImpl<dyn pal_async::wait::PollWait>> {
            self.with_driver(|d| d.new_dyn_wait(fd, read_size))
        }

        #[cfg(windows)]
        fn new_dyn_wait(
            &self,
            handle: std::os::windows::io::RawHandle,
        ) -> std::io::Result<pal_async::driver::PollImpl<dyn pal_async::wait::PollWait>> {
            self.with_driver(|d| d.new_dyn_wait(handle))
        }

        #[cfg(windows)]
        unsafe fn new_dyn_overlapped_file(
            &self,
            handle: std::os::windows::io::RawHandle,
        ) -> std::io::Result<
            pal_async::driver::PollImpl<dyn pal_async::windows::overlapped::IoOverlapped>,
        > {
            self.with_driver(|d| {
                // SAFETY: passthru from caller
                unsafe { d.new_dyn_overlapped_file(handle) }
            })
        }

        #[cfg(target_os = "linux")]
        fn io_uring_probe(&self, opcode: u8) -> bool {
            self.with_driver(|d| d.io_uring_probe(opcode))
        }

        #[cfg(target_os = "linux")]
        unsafe fn io_uring_submit(
            &self,
            sqe: pal_async::io_uring::Entry,
        ) -> std::pin::Pin<Box<dyn Future<Output = std::io::Result<i32>> + Send + '_>> {
            use pal_async::io_uring::{IoUringDriver, IoUringSubmit};
            Box::pin(async move {
                // Upgrade the thread local's weak driver to an owned strong one,
                // since the current driver may change across the lifetime of the
                // async block. This is not very expensive, though--in practice
                // this is an Arc refcount increment, and the driver is per-thread
                // so there is no cache contention. Holding the strong reference
                // across the in-flight IO is correct: the pool must stay alive
                // until the IO completes.
                let driver =
                    CURRENT_DRIVER.with(|cell| cell.borrow(|weak| weak.and_then(|w| w.upgrade())));
                let driver = driver.as_ref().unwrap_or(&self.default);
                // SAFETY: passthru from caller
                unsafe {
                    driver
                        .io_uring_submitter()
                        .ok_or(std::io::ErrorKind::Unsupported)?
                        .submit(sqe)
                        .await
                }
            })
        }

        #[cfg(target_os = "macos")]
        fn new_dyn_process_wait(
            &self,
            pid: i32,
        ) -> std::io::Result<
            pal_async::driver::PollImpl<dyn pal_async::process::macos::PollProcessWait>,
        > {
            self.with_driver(|d| d.new_dyn_process_wait(pid))
        }
    }

    impl Spawn for CurrentThreadDriver {
        fn scheduler(&self, metadata: &TaskMetadata) -> Arc<dyn pal_async::task::Schedule> {
            self.with_driver(|d| d.scheduler(metadata))
        }
    }

    impl TargetedDriver for CurrentThreadDriver {
        fn spawner(&self) -> &dyn Spawn {
            self
        }

        fn driver(&self) -> &dyn Driver {
            self
        }

        fn retarget_vp(&self, _target_vp: u32) {}
    }

    #[cfg(test)]
    mod tests {
        use super::BuildVmTaskDriver;
        use super::CURRENT_DRIVER;
        use super::ThreadDriver;
        use super::ThreadDriverBackend;
        use futures::task::ArcWake;
        use pal_async::DefaultPool;
        use pal_async::task::Spawn;
        use std::cell::RefCell;
        use std::future::Future;
        use std::pin::Pin;
        use std::sync::Arc;
        use std::sync::mpsc;
        use std::task::Context;
        use std::task::Poll;

        const POOL_NAME: &str = "test-vp-pool";

        thread_local! {
            static EXIT_SIGNAL: RefCell<Option<SignalOnDrop>> = const { RefCell::new(None) };
        }

        /// Sends on its channel when dropped.
        struct SignalOnDrop(mpsc::Sender<()>);

        impl Drop for SignalOnDrop {
            fn drop(&mut self) {
                let _ = self.0.send(());
            }
        }

        /// Builds a driver with its own dedicated pool thread, plus a receiver
        /// that fires when that thread exits.
        fn dedicated_driver() -> (ThreadDriver, mpsc::Receiver<()>) {
            let default_pool = DefaultPool::new();
            let backend = ThreadDriverBackend::new(default_pool.driver());
            let driver = backend.build(POOL_NAME.into(), Some(0), false);
            assert!(
                driver.has_dedicated_thread,
                "a target VP must get its own pool thread, else this test asserts nothing"
            );
            // Park a sender in the pool thread's thread-local storage. Thread
            // locals are dropped when the thread exits, which happens only after
            // `run()` has returned and the pool has released its IO backend.
            let (send, exited) = mpsc::channel();
            driver
                .inner
                .spawn("exit-signal", async move {
                    EXIT_SIGNAL.with(|slot| *slot.borrow_mut() = Some(SignalOnDrop(send)));
                })
                .detach();
            (driver, exited)
        }

        /// Blocks until the dedicated pool thread exits. If a regression keeps the
        /// pool alive, this never returns and the test runner's timeout fails it.
        fn assert_thread_exits(exited: &mpsc::Receiver<()>) {
            exited
                .recv()
                .expect("the exit signal was dropped without being sent");
        }

        /// What code running on the pool thread sees of `CURRENT_DRIVER`.
        #[derive(Debug, PartialEq)]
        struct DriverView {
            thread: Option<String>,
            lent: bool,
            upgraded: bool,
        }

        /// Takes the same weak upgrade `CurrentThreadDriver::with_driver` makes,
        /// so a failed upgrade here is exactly the case where `with_driver`
        /// falls back to the default driver.
        fn driver_view() -> DriverView {
            CURRENT_DRIVER.with(|cell| {
                cell.borrow(|weak| DriverView {
                    thread: std::thread::current().name().map(String::from),
                    lent: weak.is_some(),
                    upgraded: weak.and_then(|w| w.upgrade()).is_some(),
                })
            })
        }

        /// Reports `driver_view()` when dropped.
        struct ViewOnDrop(mpsc::Sender<DriverView>);

        impl Drop for ViewOnDrop {
            fn drop(&mut self) {
                let _ = self.0.send(driver_view());
            }
        }

        /// Reports `driver_view()` when woken.
        struct ViewOnWake(mpsc::Sender<DriverView>);

        impl ArcWake for ViewOnWake {
            fn wake_by_ref(arc_self: &Arc<Self>) {
                let _ = arc_self.0.send(driver_view());
            }
        }

        /// A future that completes once `go` fires, and is not dropped until
        /// after it has completed.
        struct CompleteOnSignal {
            go: mpsc::Receiver<()>,
            _view: ViewOnDrop,
        }

        impl Future for CompleteOnSignal {
            type Output = ();

            fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> {
                // Blocking the pool thread is fine here: nothing else runs on it.
                self.go.recv().unwrap();
                Poll::Ready(())
            }
        }

        /// Asserts that the view was taken on the dedicated pool thread and
        /// that the pool's driver was still reachable there.
        fn assert_dedicated_driver(views: &mpsc::Receiver<DriverView>, path: &str) {
            let view = views
                .recv()
                .unwrap_or_else(|_| panic!("{path}: the probe was dropped without reporting"));
            assert_eq!(
                view,
                DriverView {
                    thread: Some(POOL_NAME.into()),
                    lent: true,
                    upgraded: true,
                },
                "{path}: the pool's driver must stay reachable while the task is \
                 torn down on its thread, else work started there falls back to \
                 the default driver"
            );
        }

        /// Regression test for the guest-reset thread/fd leak, driving the
        /// production path: `build` with a target VP spawns the dedicated pool
        /// thread and lends that pool's own driver into `CURRENT_DRIVER` for the
        /// whole of `run()`. That reference must be weak. While any strong driver
        /// survives, `IoPool::run` cannot return, so the thread and its IO backend
        /// fds are never reclaimed and every guest reset leaks another set.
        #[test]
        fn dedicated_pool_thread_exits_when_driver_dropped() {
            let (driver, exited) = dedicated_driver();

            // Drop the only strong driver the caller holds, as a device does when
            // the guest resets and its channel closes.
            drop(driver);

            assert_thread_exits(&exited);
        }

        /// A completed task's future is dropped by `run` after it completes. Its
        /// destructor runs on the pool thread and must still see the pool's driver,
        /// even when the task held the last reference to the scheduler.
        #[test]
        fn driver_reachable_from_completed_future_destructor() {
            let (driver, exited) = dedicated_driver();
            let (view_send, views) = mpsc::channel();
            let (go, go_recv) = mpsc::channel();
            driver
                .inner
                .spawn(
                    "probe",
                    CompleteOnSignal {
                        go: go_recv,
                        _view: ViewOnDrop(view_send),
                    },
                )
                .detach();
            drop(driver);
            go.send(()).unwrap();

            assert_dedicated_driver(&views, "completed future destructor");
            assert_thread_exits(&exited);
        }

        /// Dropping a pending task's handle cancels it, and `run` drops the
        /// future on the pool thread.
        #[test]
        fn driver_reachable_from_cancelled_future_destructor() {
            let (driver, exited) = dedicated_driver();
            let (view_send, views) = mpsc::channel();
            let view = ViewOnDrop(view_send);
            let task = driver.inner.spawn("probe", async move {
                let _view = view;
                std::future::pending::<()>().await
            });
            drop(driver);
            drop(task);

            assert_dedicated_driver(&views, "cancelled future destructor");
            assert_thread_exits(&exited);
        }

        /// The output of a detached task is dropped by `run` after the future
        /// itself is gone.
        #[test]
        fn driver_reachable_from_detached_output_destructor() {
            let (driver, exited) = dedicated_driver();
            let (view_send, views) = mpsc::channel();
            let (go, go_recv) = mpsc::channel::<()>();
            driver
                .inner
                .spawn("probe", async move {
                    go_recv.recv().unwrap();
                    ViewOnDrop(view_send)
                })
                .detach();
            drop(driver);
            go.send(()).unwrap();

            assert_dedicated_driver(&views, "detached output destructor");
            assert_thread_exits(&exited);
        }

        /// A task awaiting another task's handle is woken by `run` after the
        /// awaited task's future is gone.
        #[test]
        fn driver_reachable_from_completion_wake() {
            let (driver, exited) = dedicated_driver();
            let (view_send, views) = mpsc::channel();
            let (go, go_recv) = mpsc::channel::<()>();
            let mut task = driver.inner.spawn("probe", async move {
                go_recv.recv().unwrap();
            });
            let waker = futures::task::waker(Arc::new(ViewOnWake(view_send)));
            assert!(
                Pin::new(&mut task)
                    .poll(&mut Context::from_waker(&waker))
                    .is_pending(),
                "the task cannot complete before `go` fires"
            );
            drop(driver);
            go.send(()).unwrap();

            assert_dedicated_driver(&views, "completion wake");
            drop(task);
            assert_thread_exits(&exited);
        }
    }
}