cros_async: add a command-line option to switch Executor backend

Currently Executor internally decides which backend to use, such as
FdExecutor vs URingExecutor on Linux. This makes it hard to test both
FdExecutor and URingExecutor. In addition, we came across the uring
instability issue (b:238965061), thus we need a way to explicitly switch
the polling backend to mitigate the issue.

This commit adds a command-line option to explicitly configure the
Executor's polling backend.

BUG=b:239154263
TEST=confirmed crosvm --async-executor correcly switchs the Executor for
`crosvm --async-executor epoll (or uring) device gpu --socket
/tmp/gpu.sock` command. Used additional debug log to tell which executor
is running, which is not commited

Change-Id: Ib310d9edc8ab6005f1b7f210b03668b9a45fa73f
Reviewed-on: https://chromium-review.googlesource.com/c/crosvm/crosvm/+/3811014
Reviewed-by: Keiichi Watanabe <keiichiw@chromium.org>
Commit-Queue: Takaya Saeki <takayas@chromium.org>
Reviewed-by: Daniel Verkamp <dverkamp@chromium.org>
Reviewed-by: Noah Gold <nkgold@google.com>

cros_async/src/io_ext.rs

@@ -275,7 +275,7 @@ mod tests {
     use crate::sys::unix::executor::async_poll_from_local;
     use crate::sys::unix::executor::async_uring_from;
     use crate::sys::unix::executor::async_uring_from_local;
-    use crate::sys::unix::uring_executor::use_uring;
+    use crate::sys::unix::uring_executor::is_uring_stable;
     use crate::sys::unix::FdExecutor;
     use crate::sys::unix::PollSource;
     use crate::sys::unix::URingExecutor;
@@ -319,7 +319,7 @@ mod tests {
 
     #[test]
     fn await_uring_from_poll() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
         // Start a uring operation and then await the result from an FdExecutor.
@@ -353,7 +353,7 @@ mod tests {
 
     #[test]
     fn await_poll_from_uring() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
         // Start a poll operation and then await the result from a URingExecutor.
@@ -387,7 +387,7 @@ mod tests {
 
     #[test]
     fn readvec() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
         async fn go<F: AsRawDescriptor>(async_source: Box<dyn IoSourceExt<F>>) {
@@ -423,7 +423,7 @@ mod tests {
 
     #[test]
     fn writevec() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
         async fn go<F: AsRawDescriptor>(async_source: Box<dyn IoSourceExt<F>>) {
@@ -458,7 +458,7 @@ mod tests {
 
     #[test]
     fn readmem() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
         async fn go<F: AsRawDescriptor>(async_source: Box<dyn IoSourceExt<F>>) {
@@ -511,7 +511,7 @@ mod tests {
 
     #[test]
     fn writemem() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
         async fn go<F: AsRawDescriptor>(async_source: Box<dyn IoSourceExt<F>>) {
@@ -550,7 +550,7 @@ mod tests {
 
     #[test]
     fn read_u64s() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
         async fn go(async_source: File, ex: URingExecutor) -> u64 {
@@ -566,7 +566,7 @@ mod tests {
 
     #[test]
     fn read_eventfds() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -605,7 +605,7 @@ mod tests {
 
     #[test]
     fn fsync() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
         async fn go<F: AsRawDescriptor>(source: Box<dyn IoSourceExt<F>>) {

cros_async/src/lib.rs

@@ -70,6 +70,7 @@ mod select;
 pub mod sync;
 pub mod sys;
 pub use sys::Executor;
+pub use sys::ExecutorKind;
 mod timer;
 mod waker;
 

cros_async/src/sys.rs

@@ -5,9 +5,16 @@
 cfg_if::cfg_if! {
     if #[cfg(unix)] {
         pub mod unix;
-        pub use unix::{async_types, event, executor::Executor, run_one};
+        pub use unix as platform;
     } else if #[cfg(windows)] {
         pub mod windows;
-        pub use windows::{async_types, event, executor::Executor, run_one};
+        pub use windows as platform;
     }
 }
+
+pub use platform::async_types;
+pub use platform::event;
+pub use platform::executor::Executor;
+pub use platform::executor::ExecutorKind;
+pub use platform::executor::SetDefaultExecutorKindError;
+pub use platform::run_one;

cros_async/src/sys/unix/event.rs

@@ -37,7 +37,7 @@ impl EventAsync {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::sys::unix::uring_executor::use_uring;
+    use crate::sys::unix::uring_executor::is_uring_stable;
 
     #[test]
     fn next_val_reads_value() {
@@ -55,7 +55,7 @@ mod tests {
 
     #[test]
     fn next_val_reads_value_poll_and_ring() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 

cros_async/src/sys/unix/executor.rs

@@ -3,13 +3,19 @@
 // found in the LICENSE file.
 
 use std::future::Future;
+use std::str::FromStr;
 
 use async_task::Task;
+use base::warn;
 use base::AsRawDescriptors;
 use base::RawDescriptor;
+use once_cell::sync::OnceCell;
+use thiserror::Error as ThisError;
 
 use super::poll_source::Error as PollError;
-use super::uring_executor::use_uring;
+use super::uring_executor::check_uring_availability;
+use super::uring_executor::is_uring_stable;
+use super::uring_executor::Error as UringError;
 use super::FdExecutor;
 use super::PollSource;
 use super::URingExecutor;
@@ -154,18 +160,86 @@ pub enum Executor {
     Fd(FdExecutor),
 }
 
+/// An enum to express the kind of the backend of `Executor`
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ExecutorKind {
+    Uring,
+    Fd,
+}
+
+/// If set, [`ExecutorKind::default()`] returns the value of `DEFAULT_EXECUTOR_KIND`.
+/// If not set, [`ExecutorKind::default()`] returns a statically-chosen default value, and
+/// [`ExecutorKind::default()`] initializes `DEFAULT_EXECUTOR_KIND` with that value.
+static DEFAULT_EXECUTOR_KIND: OnceCell<ExecutorKind> = OnceCell::new();
+
+impl Default for ExecutorKind {
+    fn default() -> Self {
+        *DEFAULT_EXECUTOR_KIND.get_or_init(|| ExecutorKind::Fd)
+    }
+}
+
+impl FromStr for ExecutorKind {
+    type Err = &'static str;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "uring" => Ok(ExecutorKind::Uring),
+            // For command-line parsing, user-friendly "epoll" is chosen instead of fd.
+            "epoll" => Ok(ExecutorKind::Fd),
+            _ => Err("unknown executor kind"),
+        }
+    }
+}
+
+/// The error type for [`Executor::set_default_executor_kind()`].
+#[derive(Debug, ThisError)]
+pub enum SetDefaultExecutorKindError {
+    /// The default executor kind is set more than once.
+    #[error("The default executor kind is already set to {0:?}")]
+    SetMoreThanOnce(ExecutorKind),
+
+    /// io_uring is unavailable. The reason might be the lack of the kernel support,
+    /// but is not limited to that.
+    #[error("io_uring is unavailable: {0}")]
+    UringUnavailable(UringError),
+}
+
 impl Executor {
     /// Create a new `Executor`.
     pub fn new() -> AsyncResult<Self> {
-        if use_uring() {
-            Ok(URingExecutor::new().map(Executor::Uring)?)
-        } else {
-            Ok(FdExecutor::new()
+        match ExecutorKind::default() {
+            ExecutorKind::Uring => Ok(URingExecutor::new().map(Executor::Uring)?),
+            ExecutorKind::Fd => Ok(FdExecutor::new()
                 .map(Executor::Fd)
-                .map_err(PollError::Executor)?)
+                .map_err(PollError::Executor)?),
         }
     }
 
+    /// Set the default ExecutorKind for [`Self::new()`]. This call is effective only once.
+    /// If a call is the first call, it sets the default, and `set_default_executor_kind`
+    /// returns `Ok(())`. Otherwise, it returns `SetDefaultExecutorKindError::SetMoreThanOnce`
+    /// which contains the existing ExecutorKind value configured by the first call.
+    pub fn set_default_executor_kind(
+        executor_kind: ExecutorKind,
+    ) -> Result<(), SetDefaultExecutorKindError> {
+        if executor_kind == ExecutorKind::Uring {
+            check_uring_availability().map_err(SetDefaultExecutorKindError::UringUnavailable)?;
+            if !is_uring_stable() {
+                warn!(
+                    "Enabling io_uring executor on the kernel version where io_uring is unstable"
+                );
+            }
+        }
+
+        DEFAULT_EXECUTOR_KIND.set(executor_kind).map_err(|_|
+            // `expect` succeeds since this closure runs only when DEFAULT_EXECUTOR_KIND is set.
+            SetDefaultExecutorKindError::SetMoreThanOnce(
+                *DEFAULT_EXECUTOR_KIND
+                    .get()
+                    .expect("Failed to get DEFAULT_EXECUTOR_KIND"),
+            ))
+    }
+
     /// Create a new `Box<dyn IoSourceExt<F>>` associated with `self`. Callers may then use the
     /// returned `IoSourceExt` to directly start async operations without needing a separate
     /// reference to the executor.

cros_async/src/sys/unix/timer.rs

@@ -28,7 +28,7 @@ mod tests {
     use std::time::Instant;
 
     use super::*;
-    use crate::sys::unix::uring_executor::use_uring;
+    use crate::sys::unix::uring_executor::is_uring_stable;
     use crate::Executor;
 
     #[test]
@@ -46,7 +46,7 @@ mod tests {
 
     #[test]
     fn one_shot() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 

cros_async/src/sys/unix/uring_executor.rs

@@ -148,7 +148,7 @@ impl From<Error> for io::Error {
     }
 }
 
-static USE_URING: Lazy<bool> = Lazy::new(|| {
+static IS_URING_STABLE: Lazy<bool> = Lazy::new(|| {
     let mut utsname = MaybeUninit::zeroed();
 
     // Safe because this will only modify `utsname` and we check the return value.
@@ -178,11 +178,20 @@ static USE_URING: Lazy<bool> = Lazy::new(|| {
     }
 });
 
-// Checks if the uring executor is available.
+// Checks if the uring executor is stable.
 // Caches the result so that the check is only run once.
 // Useful for falling back to the FD executor on pre-uring kernels.
-pub(crate) fn use_uring() -> bool {
-    *USE_URING
+pub(crate) fn is_uring_stable() -> bool {
+    *IS_URING_STABLE
+}
+
+// Checks the uring availability by checking if the uring creation succeeds.
+// If uring creation succeeds, it returns `Ok(())`. It returns an `URingContextError` otherwise.
+// It fails if the kernel does not support io_uring, but note that the cause is not limited to it.
+pub(crate) fn check_uring_availability() -> Result<()> {
+    URingContext::new(8)
+        .map(drop)
+        .map_err(Error::URingContextError)
 }
 
 pub struct RegisteredSource {
@@ -992,7 +1001,7 @@ mod tests {
 
     #[test]
     fn dont_drop_backing_mem_read() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -1036,7 +1045,7 @@ mod tests {
 
     #[test]
     fn dont_drop_backing_mem_write() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -1081,7 +1090,7 @@ mod tests {
 
     #[test]
     fn canceled_before_completion() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -1125,7 +1134,7 @@ mod tests {
     #[ignore]
     #[test]
     fn drop_before_completion() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -1174,7 +1183,7 @@ mod tests {
 
     #[test]
     fn drop_on_different_thread() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 

cros_async/src/sys/unix/uring_source.rs

@@ -230,7 +230,7 @@ mod tests {
     use std::fs::OpenOptions;
     use std::path::PathBuf;
 
-    use super::super::uring_executor::use_uring;
+    use super::super::uring_executor::is_uring_stable;
     use super::super::UringSource;
     use super::*;
     use crate::io_ext::ReadAsync;
@@ -238,7 +238,7 @@ mod tests {
 
     #[test]
     fn read_to_mem() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -277,7 +277,7 @@ mod tests {
 
     #[test]
     fn readvec() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -299,7 +299,7 @@ mod tests {
 
     #[test]
     fn readmulti() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -334,7 +334,7 @@ mod tests {
 
     #[test]
     fn u64_from_file() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -347,7 +347,7 @@ mod tests {
 
     #[test]
     fn event() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -388,7 +388,7 @@ mod tests {
 
     #[test]
     fn pend_on_pipe() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -418,7 +418,7 @@ mod tests {
 
     #[test]
     fn readmem() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -472,7 +472,7 @@ mod tests {
 
     #[test]
     fn range_error() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -500,7 +500,7 @@ mod tests {
 
     #[test]
     fn fallocate() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -539,7 +539,7 @@ mod tests {
 
     #[test]
     fn fsync() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -555,7 +555,7 @@ mod tests {
 
     #[test]
     fn wait_read() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -571,7 +571,7 @@ mod tests {
 
     #[test]
     fn writemem() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -597,7 +597,7 @@ mod tests {
 
     #[test]
     fn writevec() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 
@@ -622,7 +622,7 @@ mod tests {
 
     #[test]
     fn writemulti() {
-        if !use_uring() {
+        if !is_uring_stable() {
             return;
         }
 

cros_async/src/sys/windows/executor.rs

@@ -3,8 +3,11 @@
 // found in the LICENSE file.
 
 use std::future::Future;
+use std::str::FromStr;
 
 use async_task::Task;
+use once_cell::sync::OnceCell;
+use thiserror::Error as ThisError;
 
 use super::HandleExecutor;
 use super::HandleSource;
@@ -122,10 +125,49 @@ pub enum Executor {
     Handle(HandleExecutor),
 }
 
+/// An enum to express the kind of the backend of `Executor`
+#[derive(Clone, Copy, Debug)]
+pub enum ExecutorKind {
+    Handle,
+}
+
+/// If set, [`Executor::new()`] is created with `ExecutorKind` of `DEFAULT_EXECUTOR_KIND`.
+static DEFAULT_EXECUTOR_KIND: OnceCell<ExecutorKind> = OnceCell::new();
+
+impl FromStr for ExecutorKind {
+    type Err = &'static str;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "handle" => Ok(ExecutorKind::Handle),
+            _ => Err("unknown executor kind"),
+        }
+    }
+}
+
+impl Default for ExecutorKind {
+    fn default() -> Self {
+        DEFAULT_EXECUTOR_KIND
+            .get()
+            .copied()
+            .unwrap_or(ExecutorKind::Handle)
+    }
+}
+
+/// The error type for [`Executor::set_default_executor_kind()`].
+#[derive(ThisError, Debug)]
+pub enum SetDefaultExecutorKindError {
+    /// The default executor kind is set more than once.
+    #[error("The default executor kind is already set to {0:?}")]
+    SetMoreThanOnce(ExecutorKind),
+}
+
 impl Executor {
     /// Create a new `Executor`.
     pub fn new() -> AsyncResult<Self> {
-        Ok(Executor::Handle(HandleExecutor::new()))
+        match ExecutorKind::default() {
+            ExecutorKind::Handle => Ok(Executor::Handle(HandleExecutor::new())),
+        }
     }
 
     /// Create a new `Box<dyn IoSourceExt<F>>` associated with `self`. Callers may then use the
@@ -140,6 +182,22 @@ impl Executor {
         }
     }
 
+    /// Set the default ExecutorKind for [`Self::new()`]. This call is effective only once.
+    /// If a call is the first call, it sets the default, and `set_default_executor_kind`
+    /// returns `Ok(())`. Otherwise, it returns `SetDefaultExecutorKindError::SetMoreThanOnce`
+    /// which contains the existing ExecutorKind value configured by the first call.
+    pub fn set_default_executor_kind(
+        executor_kind: ExecutorKind,
+    ) -> Result<(), SetDefaultExecutorKindError> {
+        DEFAULT_EXECUTOR_KIND.set(executor_kind).map_err(|_|
+            // `expect` succeeds since this closure runs only when DEFAULT_EXECUTOR_KIND is set.
+            SetDefaultExecutorKindError::SetMoreThanOnce(
+                *DEFAULT_EXECUTOR_KIND
+                    .get()
+                    .expect("Failed to get DEFAULT_EXECUTOR_KIND"),
+            ))
+    }
+
     /// Spawn a new future for this executor to run to completion. Callers may use the returned
     /// `Task` to await on the result of `f`. Dropping the returned `Task` will cancel `f`,
     /// preventing it from being polled again. To drop a `Task` without canceling the future

src/crosvm/cmdline.rs

@@ -30,6 +30,7 @@ use arch::Pstore;
 use arch::VcpuAffinity;
 use argh::FromArgs;
 use base::getpid;
+use cros_async::ExecutorKind;
 use devices::virtio::block::block::DiskOption;
 #[cfg(any(feature = "video-decoder", feature = "video-encoder"))]
 use devices::virtio::device_constants::video::VideoDeviceConfig;
@@ -105,6 +106,10 @@ pub struct CrosvmCmdlineArgs {
     #[argh(switch)]
     /// disable output to syslog
     pub no_syslog: bool,
+    /// configure async executor backend; "uring" or "epoll" on Linux, "handle" on Windows.
+    /// If this option is omitted on Linux, "epoll" is used by default.
+    #[argh(option, arg_name = "EXECUTOR")]
+    pub async_executor: Option<ExecutorKind>,
     #[argh(subcommand)]
     pub command: Command,
 }

src/main.rs

@@ -528,6 +528,11 @@ fn crosvm_main() -> Result<CommandStatus> {
         ..Default::default()
     };
 
+    if let Some(async_executor) = args.async_executor {
+        cros_async::Executor::set_default_executor_kind(async_executor)
+            .context("Failed to set the default async executor")?;
+    }
+
     let ret = match args.command {
         Command::CrossPlatform(command) => {
             // Past this point, usage of exit is in danger of leaking zombie processes.