devices: virtio: factor out common async code

Create an async_utils module and deduplicate the common async helper
functions there.

In particular, this moves wait_kill (renamed to await_and_exit) and
handle_irq_resample into the new module and replaces all uses throughout
the async device implementations.

BUG=None
TEST=tools/presubmit --quick

Change-Id: If3b5b9f71c8553afa5ccd8c57933ca748fc990e7
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/crosvm/+/3292635
Tested-by: kokoro <noreply+kokoro@google.com>
Commit-Queue: Daniel Verkamp <dverkamp@chromium.org>
Reviewed-by: Chirantan Ekbote <chirantan@chromium.org>

devices/src/virtio/async_utils.rs

@@ -0,0 +1,45 @@
+// Copyright 2021 The Chromium OS Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+//! Virtio device async helper functions.
+
+use std::cell::RefCell;
+use std::rc::Rc;
+
+use anyhow::{Context, Result};
+use base::Event;
+use cros_async::{EventAsync, Executor};
+
+use super::{Interrupt, SignalableInterrupt};
+
+/// Async task that waits for a signal from `event`.  Once this event is readable, exit. Exiting
+/// this future will cause the main loop to break and the worker thread to exit.
+pub async fn await_and_exit(ex: &Executor, event: Event) -> Result<()> {
+    let event_async = EventAsync::new(event.0, ex).context("failed to create EventAsync")?;
+    let _ = event_async.next_val().await;
+    Ok(())
+}
+
+/// Async task that resamples the status of the interrupt when the guest sends a request by
+/// signalling the resample event associated with the interrupt.
+pub async fn handle_irq_resample(ex: &Executor, interrupt: Rc<RefCell<Interrupt>>) -> Result<()> {
+    if let Some(resample_evt) = interrupt.borrow().get_resample_evt() {
+        let resample_evt = resample_evt
+            .try_clone()
+            .context("resample_evt.try_clone() failed")?;
+        let resample_evt =
+            EventAsync::new(resample_evt.0, ex).context("failed to create async resample event")?;
+        loop {
+            let _ = resample_evt
+                .next_val()
+                .await
+                .context("failed to read resample event")?;
+            interrupt.borrow().do_interrupt_resample();
+        }
+    } else {
+        // No resample event; park the future.
+        let () = futures::future::pending().await;
+    }
+    Ok(())
+}

devices/src/virtio/balloon.rs

@@ -19,8 +19,8 @@ use vm_control::{BalloonStats, BalloonTubeCommand, BalloonTubeResult};
 use vm_memory::{GuestAddress, GuestMemory};
 
 use super::{
-    copy_config, descriptor_utils, DescriptorChain, Interrupt, Queue, Reader, SignalableInterrupt,
-    VirtioDevice, TYPE_BALLOON,
+    async_utils, copy_config, descriptor_utils, DescriptorChain, Interrupt, Queue, Reader,
+    SignalableInterrupt, VirtioDevice, TYPE_BALLOON,
 };
 
 #[sorted]
@@ -289,33 +289,6 @@ async fn handle_command_tube(
     }
 }
 
-// Async task that resamples the status of the interrupt when the guest sends a request by
-// signalling the resample event associated with the interrupt.
-async fn handle_irq_resample(ex: &Executor, interrupt: Rc<RefCell<Interrupt>>) {
-    let resample_evt = if let Some(resample_evt) = interrupt.borrow_mut().get_resample_evt() {
-        let resample_evt = resample_evt.try_clone().unwrap();
-        let resample_evt = EventAsync::new(resample_evt.0, ex).unwrap();
-        Some(resample_evt)
-    } else {
-        None
-    };
-    if let Some(resample_evt) = resample_evt {
-        while resample_evt.next_val().await.is_ok() {
-            interrupt.borrow_mut().do_interrupt_resample();
-        }
-    } else {
-        // no resample event, park the future.
-        let () = futures::future::pending().await;
-    }
-}
-
-// Async task that waits for a signal from the kill event given to the device at startup.  Once this event is
-// readable, exit. Exiting this future will cause the main loop to break and the worker thread to
-// exit.
-async fn wait_kill(kill_evt: EventAsync) {
-    let _ = kill_evt.next_val().await;
-}
-
 // The main worker thread. Initialized the asynchronous worker tasks and passes them to the executor
 // to be processed.
 fn run_worker(
@@ -385,12 +358,11 @@ fn run_worker(
         pin_mut!(command);
 
         // Process any requests to resample the irq value.
-        let resample = handle_irq_resample(&ex, interrupt);
+        let resample = async_utils::handle_irq_resample(&ex, interrupt);
         pin_mut!(resample);
 
         // Exit if the kill event is triggered.
-        let kill_evt = EventAsync::new(kill_evt.0, &ex).expect("failed to set up the kill event");
-        let kill = wait_kill(kill_evt);
+        let kill = async_utils::await_and_exit(&ex, kill_evt);
         pin_mut!(kill);
 
         if let Err(e) = ex.run_until(select6(inflate, deflate, stats, command, resample, kill)) {

devices/src/virtio/block/asynchronous.rs

@@ -34,8 +34,8 @@ use vm_memory::GuestMemory;
 
 use super::common::*;
 use crate::virtio::{
-    copy_config, DescriptorChain, DescriptorError, Interrupt, Queue, Reader, SignalableInterrupt,
-    VirtioDevice, Writer, TYPE_BLOCK,
+    async_utils, copy_config, DescriptorChain, DescriptorError, Interrupt, Queue, Reader,
+    SignalableInterrupt, VirtioDevice, Writer, TYPE_BLOCK,
 };
 
 const QUEUE_SIZE: u16 = 256;
@@ -271,41 +271,6 @@ async fn handle_queue(
     }
 }
 
-async fn handle_irq_resample(
-    ex: &Executor,
-    interrupt: Rc<RefCell<Interrupt>>,
-) -> result::Result<(), ControlError> {
-    let resample_evt = if let Some(resample_evt) = interrupt.borrow().get_resample_evt() {
-        let resample_evt = resample_evt
-            .try_clone()
-            .map_err(ControlError::CloneResampleEvent)?;
-        let resample_evt =
-            EventAsync::new(resample_evt.0, ex).map_err(ControlError::AsyncResampleCreate)?;
-        Some(resample_evt)
-    } else {
-        None
-    };
-    if let Some(resample_evt) = resample_evt {
-        loop {
-            let _ = resample_evt
-                .next_val()
-                .await
-                .map_err(ControlError::ReadResampleEvent)?;
-            interrupt.borrow().do_interrupt_resample();
-        }
-    } else {
-        // no resample event, park the future.
-        let () = futures::future::pending().await;
-        Ok(())
-    }
-}
-
-async fn wait_kill(kill_evt: EventAsync) {
-    // Once this event is readable, exit. Exiting this future will cause the main loop to
-    // break and the device process to exit.
-    let _ = kill_evt.next_val().await;
-}
-
 async fn handle_command_tube(
     command_tube: &Option<AsyncTube>,
     interrupt: Rc<RefCell<Interrupt>>,
@@ -423,7 +388,7 @@ fn run_worker(
     let flush_timer_armed = Rc::new(RefCell::new(false));
 
     // Process any requests to resample the irq value.
-    let resample = handle_irq_resample(&ex, Rc::clone(&interrupt));
+    let resample = async_utils::handle_irq_resample(&ex, Rc::clone(&interrupt));
     pin_mut!(resample);
 
     // Handles control requests.
@@ -472,8 +437,7 @@ fn run_worker(
     pin_mut!(disk_flush);
 
     // Exit if the kill event is triggered.
-    let kill_evt = EventAsync::new(kill_evt.0, &ex).expect("Failed to create async kill event fd");
-    let kill = wait_kill(kill_evt);
+    let kill = async_utils::await_and_exit(&ex, kill_evt);
     pin_mut!(kill);
 
     match ex.run_until(select5(queue_handlers, disk_flush, control, resample, kill)) {

devices/src/virtio/mod.rs

@@ -4,6 +4,7 @@
 
 //! Implements virtio devices, queues, and transport mechanisms.
 
+mod async_utils;
 mod balloon;
 mod descriptor_utils;
 mod input;

devices/src/virtio/snd/cras_backend/async_funcs.rs

@@ -671,10 +671,3 @@ pub async fn handle_event_queue<I: SignalableInterrupt>(
         queue.trigger_interrupt(&mem, interrupt);
     }
 }
-
-// Async task that waits for a signal from the kill event given to the device at startup.  Once this event is
-// readable, exit. Exiting this future will cause the main loop to break and the worker thread to
-// exit.
-pub async fn wait_kill(kill_evt: EventAsync) {
-    let _ = kill_evt.next_val().await;
-}

devices/src/virtio/snd/cras_backend/mod.rs

@@ -29,8 +29,8 @@ use crate::virtio::snd::common::*;
 use crate::virtio::snd::constants::*;
 use crate::virtio::snd::layout::*;
 use crate::virtio::{
-    copy_config, DescriptorChain, DescriptorError, Interrupt, Queue, VirtioDevice, Writer,
-    TYPE_SOUND,
+    async_utils, copy_config, DescriptorChain, DescriptorError, Interrupt, Queue, VirtioDevice,
+    Writer, TYPE_SOUND,
 };
 
 pub mod async_funcs;
@@ -690,8 +690,7 @@ fn run_worker(
     let f_rx_response = send_pcm_response_worker(&mem, &rx_queue, interrupt.as_ref(), &mut rx_recv);
 
     // Exit if the kill event is triggered.
-    let kill_evt = EventAsync::new(kill_evt.0, &ex).expect("failed to set up the kill event");
-    let f_kill = wait_kill(kill_evt);
+    let f_kill = async_utils::await_and_exit(&ex, kill_evt);
 
     pin_mut!(f_ctrl, f_tx, f_tx_response, f_rx, f_rx_response, f_kill);
 

devices/src/virtio/vhost/user/vmm/worker.rs

@@ -2,23 +2,15 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-use base::{error, Event};
-use cros_async::{select2, AsyncError, EventAsync, Executor, SelectResult};
+use std::cell::RefCell;
+use std::rc::Rc;
+
+use base::Event;
+use cros_async::{select2, Executor, SelectResult};
 use futures::pin_mut;
-use remain::sorted;
-use thiserror::Error as ThisError;
 use vm_memory::GuestMemory;
 
-use crate::virtio::interrupt::SignalableInterrupt;
-use crate::virtio::{Interrupt, Queue};
-
-#[sorted]
-#[derive(ThisError, Debug)]
-enum Error {
-    /// Failed to read the resample event.
-    #[error("failed to read the resample event: {0}")]
-    ReadResampleEvent(AsyncError),
-}
+use crate::virtio::{async_utils, Interrupt, Queue};
 
 pub struct Worker {
     pub queues: Vec<Queue>,
@@ -27,49 +19,16 @@ pub struct Worker {
 }
 
 impl Worker {
-    // Processes any requests to resample the irq value.
-    async fn handle_irq_resample(
-        resample_evt: EventAsync,
-        interrupt: Interrupt,
-    ) -> Result<(), Error> {
-        loop {
-            let _ = resample_evt
-                .next_val()
-                .await
-                .map_err(Error::ReadResampleEvent)?;
-            interrupt.do_interrupt_resample();
-        }
-    }
-
-    // Waits until the kill event is triggered.
-    async fn wait_kill(kill_evt: EventAsync) {
-        // Once this event is readable, exit. Exiting this future will cause the main loop to
-        // break and the device process to exit.
-        let _ = kill_evt.next_val().await;
-    }
-
     // Runs asynchronous tasks.
     pub fn run(&mut self, interrupt: Interrupt) -> Result<(), String> {
         let ex = Executor::new().expect("failed to create an executor");
-        let resample_evt = interrupt
-            .get_resample_evt()
-            .expect("resample event required")
-            .try_clone()
-            .expect("failed to clone resample event");
-        let async_resample_evt =
-            EventAsync::new(resample_evt.0, &ex).expect("failed to create async resample event");
-        let resample = Self::handle_irq_resample(async_resample_evt, interrupt);
+
+        let interrupt = Rc::new(RefCell::new(interrupt));
+        let resample = async_utils::handle_irq_resample(&ex, interrupt);
         pin_mut!(resample);
 
-        let kill_evt = EventAsync::new(
-            self.kill_evt
-                .try_clone()
-                .expect("failed to clone kill_evt")
-                .0,
-            &ex,
-        )
-        .expect("failed to create async kill event fd");
-        let kill = Self::wait_kill(kill_evt);
+        let kill_evt = self.kill_evt.try_clone().expect("failed to clone kill_evt");
+        let kill = async_utils::await_and_exit(&ex, kill_evt);
         pin_mut!(kill);
 
         match ex.run_until(select2(resample, kill)) {