cros_async: Move `concurrency` to Overlapped enum

`Executor::with_kind_and_concurrency` exists only for the benefit of the
`ExecutorKindSys::Overlapped` enum variant, it's not used outside of
windows code and not used for `ExecutorKindSys::Handle`.

`DiskOption`'s `io_concurrency` value is used to override the
concurrency setting in the `ExecutorKind`.

Test: tools/presubmit
Change-Id: Idd59d78283038ab6f986953f846cd70ef25d4324
Reviewed-on: https://chromium-review.googlesource.com/c/crosvm/crosvm/+/5315960
Reviewed-by: Frederick Mayle <fmayle@google.com>
Commit-Queue: Cody Schuffelen <schuffelen@google.com>
Reviewed-by: Noah Gold <nkgold@google.com>

6 files changed, 157 insertions, 40 deletions

diff --git a/cros_async/src/executor.rs b/cros_async/src/executor.rs
index 2f1e7166b..cb6f726a1 100644
--- a/cros_async/src/executor.rs
+++ b/cros_async/src/executor.rs
@@ -13,6 +13,10 @@ use base::AsRawDescriptors;
 #[cfg(any(target_os = "android", target_os = "linux"))]
 use base::RawDescriptor;
 use once_cell::sync::OnceCell;
+use serde::Deserialize;
+use serde_keyvalue::argh::FromArgValue;
+use serde_keyvalue::ErrorKind;
+use serde_keyvalue::KeyValueDeserializer;
 
 use crate::common_executor;
 use crate::common_executor::RawExecutor;
@@ -25,17 +29,7 @@ use crate::AsyncResult;
 use crate::IntoAsync;
 use crate::IoSource;
 
-#[derive(
-    Clone,
-    Copy,
-    Debug,
-    PartialEq,
-    Eq,
-    serde::Serialize,
-    serde::Deserialize,
-    serde_keyvalue::FromKeyValues,
-)]
-#[serde(deny_unknown_fields, rename_all = "kebab-case", untagged)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum ExecutorKind {
     SysVariants(ExecutorKindSys),
 }
@@ -82,6 +76,72 @@ pub enum SetDefaultExecutorKindError {
     UringUnavailable(linux::uring_executor::Error),
 }
 
+impl FromArgValue for ExecutorKind {
+    fn from_arg_value(value: &str) -> std::result::Result<ExecutorKind, String> {
+        // `from_arg_value` returns a `String` as error, but our deserializer API defines its own
+        // error type. Perform parsing from a closure so we can easily map returned errors.
+        let builder = move || {
+            let mut des = KeyValueDeserializer::from(value);
+
+            let kind: ExecutorKind = match (des.parse_identifier()?, des.next_char()) {
+                #[cfg(any(target_os = "android", target_os = "linux"))]
+                ("epoll", None) => ExecutorKindSys::Fd.into(),
+                #[cfg(any(target_os = "android", target_os = "linux"))]
+                ("uring", None) => ExecutorKindSys::Uring.into(),
+                #[cfg(windows)]
+                ("handle", None) => ExecutorKindSys::Handle.into(),
+                #[cfg(windows)]
+                ("overlapped", None) => ExecutorKindSys::Overlapped { concurrency: None }.into(),
+                #[cfg(windows)]
+                ("overlapped", Some(',')) => {
+                    if des.parse_identifier()? != "concurrency" {
+                        let kind = ErrorKind::SerdeError("expected `concurrency`".to_string());
+                        return Err(des.error_here(kind));
+                    }
+                    if des.next_char() != Some('=') {
+                        return Err(des.error_here(ErrorKind::ExpectedEqual));
+                    }
+                    let concurrency = des.parse_number()?;
+                    ExecutorKindSys::Overlapped {
+                        concurrency: Some(concurrency),
+                    }
+                    .into()
+                }
+                (_identifier, _next) => {
+                    let kind = ErrorKind::SerdeError("unexpected kind".to_string());
+                    return Err(des.error_here(kind));
+                }
+            };
+            des.finish()?;
+            Ok(kind)
+        };
+
+        builder().map_err(|e| e.to_string())
+    }
+}
+
+impl serde::Serialize for ExecutorKind {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        match self {
+            ExecutorKind::SysVariants(sv) => sv.serialize(serializer),
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for ExecutorKind {
+    fn deserialize<D>(deserializer: D) -> Result<ExecutorKind, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        base::error!("ExecutorKind::deserialize");
+        let string = String::deserialize(deserializer)?;
+        ExecutorKind::from_arg_value(&string).map_err(serde::de::Error::custom)
+    }
+}
+
 /// Reference to a task managed by the executor.
 ///
 /// Dropping a `TaskHandle` attempts to cancel the associated task. Call `detach` to allow it to
@@ -287,21 +347,13 @@ impl Executor {
                 Executor::Handle(RawExecutor::new()?)
             }
             #[cfg(windows)]
-            ExecutorKind::SysVariants(ExecutorKindSys::Overlapped) => {
-                Executor::Overlapped(RawExecutor::new()?)
-            }
-        })
-    }
-
-    /// Create a new `Executor` of the given `ExecutorKind`.
-    pub fn with_kind_and_concurrency(kind: ExecutorKind, _concurrency: u32) -> AsyncResult<Self> {
-        Ok(match kind {
-            #[cfg(windows)]
-            ExecutorKind::SysVariants(ExecutorKindSys::Overlapped) => {
-                let reactor = windows::HandleReactor::new_with(_concurrency)?;
+            ExecutorKind::SysVariants(ExecutorKindSys::Overlapped { concurrency }) => {
+                let reactor = match concurrency {
+                    Some(concurrency) => windows::HandleReactor::new_with(concurrency)?,
+                    None => windows::HandleReactor::new()?,
+                };
                 Executor::Overlapped(RawExecutor::new_with(reactor)?)
             }
-            _ => Executor::with_executor_kind(kind)?,
         })
     }
 
diff --git a/cros_async/src/sys/windows/executor.rs b/cros_async/src/sys/windows/executor.rs
index 3d9053b22..31ba4bfa5 100644
--- a/cros_async/src/sys/windows/executor.rs
+++ b/cros_async/src/sys/windows/executor.rs
@@ -2,15 +2,25 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-use serde::Deserialize;
-use serde::Serialize;
-
 /// An enum to express the kind of the backend of `Executor`
-#[derive(
-    Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, serde_keyvalue::FromKeyValues,
-)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Deserialize, serde_keyvalue::FromKeyValues)]
 #[serde(deny_unknown_fields, rename_all = "kebab-case")]
 pub enum ExecutorKindSys {
     Handle,
-    Overlapped,
+    Overlapped { concurrency: Option<u32> },
+}
+
+impl serde::Serialize for ExecutorKindSys {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        serializer.serialize_str(&match self {
+            ExecutorKindSys::Handle => "handle".to_string(),
+            ExecutorKindSys::Overlapped { concurrency: None } => "overlapped".to_string(),
+            ExecutorKindSys::Overlapped {
+                concurrency: Some(n),
+            } => format!("overlapped,concurrency={}", n),
+        })
+    }
 }
diff --git a/cros_async/src/sys/windows/handle_executor.rs b/cros_async/src/sys/windows/handle_executor.rs
index 614730801..c5c4fe522 100644
--- a/cros_async/src/sys/windows/handle_executor.rs
+++ b/cros_async/src/sys/windows/handle_executor.rs
@@ -95,7 +95,7 @@ impl HandleReactor {
         })
     }
 
-    fn new() -> Result<Self> {
+    pub fn new() -> Result<Self> {
         Self::new_with(DEFAULT_IO_CONCURRENCY)
     }
 
diff --git a/devices/src/virtio/block/mod.rs b/devices/src/virtio/block/mod.rs
index d52c4fe2b..b06f31115 100644
--- a/devices/src/virtio/block/mod.rs
+++ b/devices/src/virtio/block/mod.rs
@@ -102,6 +102,7 @@ pub struct DiskOption {
         deserialize_with = "deserialize_disk_id"
     )]
     pub id: Option<[u8; DISK_ID_LEN]>,
+    // Deprecated: Use async_executor=overlapped[concurrency=N]"
     // camel_case variant allowed for backward compatibility.
     #[cfg(windows)]
     #[serde(
@@ -451,6 +452,51 @@ mod tests {
                     pci_address: None,
                 }
             );
+            let params = from_block_arg("/some/path.img,async-executor=overlapped").unwrap();
+            assert_eq!(
+                params,
+                DiskOption {
+                    path: "/some/path.img".into(),
+                    read_only: false,
+                    root: false,
+                    sparse: true,
+                    direct: false,
+                    block_size: 512,
+                    id: None,
+                    io_concurrency: NonZeroU32::new(1).unwrap(),
+                    multiple_workers: false,
+                    async_executor: Some(ExecutorKindSys::Overlapped { concurrency: None }.into()),
+                    packed_queue: false,
+                    bootindex: None,
+                    pci_address: None,
+                }
+            );
+            let params =
+                from_block_arg("/some/path.img,async-executor=\"overlapped,concurrency=4\"")
+                    .unwrap();
+            assert_eq!(
+                params,
+                DiskOption {
+                    path: "/some/path.img".into(),
+                    read_only: false,
+                    root: false,
+                    sparse: true,
+                    direct: false,
+                    block_size: 512,
+                    id: None,
+                    io_concurrency: NonZeroU32::new(1).unwrap(),
+                    multiple_workers: false,
+                    async_executor: Some(
+                        ExecutorKindSys::Overlapped {
+                            concurrency: Some(4)
+                        }
+                        .into()
+                    ),
+                    packed_queue: false,
+                    bootindex: None,
+                    pci_address: None,
+                }
+            );
         }
 
         // id
diff --git a/devices/src/virtio/block/sys/windows.rs b/devices/src/virtio/block/sys/windows.rs
index 4c215290f..7b66a1ab6 100644
--- a/devices/src/virtio/block/sys/windows.rs
+++ b/devices/src/virtio/block/sys/windows.rs
@@ -10,6 +10,7 @@ use anyhow::Context;
 use base::warn;
 use cros_async::sys::windows::ExecutorKindSys;
 use cros_async::Executor;
+use cros_async::ExecutorKind;
 use winapi::um::winbase::FILE_FLAG_NO_BUFFERING;
 use winapi::um::winbase::FILE_FLAG_OVERLAPPED;
 use winapi::um::winnt::FILE_SHARE_READ;
@@ -38,7 +39,13 @@ impl DiskOption {
             flags |= FILE_FLAG_NO_BUFFERING;
         }
 
-        if self.async_executor == Some(ExecutorKindSys::Overlapped.into()) {
+        let is_overlapped = matches!(
+            self.async_executor,
+            Some(ExecutorKind::SysVariants(
+                ExecutorKindSys::Overlapped { .. }
+            ))
+        );
+        if is_overlapped {
             warn!("Opening disk file for overlapped IO");
             flags |= FILE_FLAG_OVERLAPPED;
         }
@@ -50,10 +57,7 @@ impl DiskOption {
         let file = open_option
             .open(&self.path)
             .context("Failed to open disk file")?;
-        let image_type = disk::detect_image_type(
-            &file,
-            self.async_executor == Some(ExecutorKindSys::Overlapped.into()),
-        )?;
+        let image_type = disk::detect_image_type(&file, is_overlapped)?;
         Ok(disk::create_disk_file_of_type(
             file,
             self.sparse,
@@ -66,7 +70,12 @@ impl DiskOption {
 
 impl BlockAsync {
     pub fn create_executor(&self) -> Executor {
-        Executor::with_kind_and_concurrency(self.executor_kind, self.io_concurrency)
-            .expect("Failed to create an executor")
+        let mut kind = self.executor_kind;
+        if let ExecutorKind::SysVariants(ExecutorKindSys::Overlapped { concurrency }) = &mut kind {
+            if concurrency.is_none() {
+                *concurrency = Some(self.io_concurrency);
+            }
+        }
+        Executor::with_executor_kind(kind).expect("Failed to create an executor")
     }
 }
diff --git a/devices/src/virtio/vhost/user/device/block/sys/windows.rs b/devices/src/virtio/vhost/user/device/block/sys/windows.rs
index 058c823a7..0aa56707f 100644
--- a/devices/src/virtio/vhost/user/device/block/sys/windows.rs
+++ b/devices/src/virtio/vhost/user/device/block/sys/windows.rs
@@ -65,7 +65,7 @@ pub fn start_device(opts: Options) -> anyhow::Result<()> {
     let _raise_timer_resolution =
         enable_high_res_timers().context("failed to set timer resolution")?;
 
-    info!("using {} IO handles.", disk_option.io_concurrency.get());
+    info!("using {:?} executor.", disk_option.async_executor);
 
     let kind = disk_option
         .async_executor