Review feedback

illumos-utils/src/zfs.rs

@@ -256,7 +256,9 @@ impl DatasetProperties {
     /// Parses dataset properties, assuming that the caller is providing the
     /// output of the following command as stdout:
     ///
-    /// zfs get -rpo name,property,value,source $ZFS_GET_PROPS $DATASETS
+    /// zfs get \
+    ///     [maybe depth arguments] \
+    ///     -Hpo name,property,value,source $ZFS_GET_PROPS $DATASETS
     fn parse_many(
         stdout: &str,
     ) -> Result<Vec<DatasetProperties>, anyhow::Error> {
@@ -392,43 +394,42 @@ pub enum WhichDatasets {
 // set". This helps callers minimize the number of calls to "zfs set" they need
 // to make.
 struct PropertySetter {
-    props: BTreeMap<&'static str, String>,
+    props: Vec<(&'static str, String)>,
 }
 
 impl PropertySetter {
     fn new() -> Self {
-        PropertySetter { props: BTreeMap::new() }
+        PropertySetter { props: Vec::new() }
     }
 
     fn add_size_details(&mut self, details: SizeDetails) -> &mut Self {
         let SizeDetails { quota, reservation, compression } = details;
         let quota = quota
             .map(|q| q.to_bytes().to_string())
             .unwrap_or_else(|| String::from("none"));
-        self.props.insert("quota", quota);
+        self.props.push(("quota", quota));
 
         let reservation = reservation
             .map(|r| r.to_bytes().to_string())
             .unwrap_or_else(|| String::from("none"));
-        self.props.insert("reservation", reservation);
+        self.props.push(("reservation", reservation));
 
         let compression = compression.to_string();
-        self.props.insert("compression", compression);
+        self.props.push(("compression", compression));
 
         self
     }
 
     fn add_id(&mut self, id: DatasetUuid) -> &mut Self {
-        self.props.insert("oxide:uuid", id.to_string());
+        self.props.push(("oxide:uuid", id.to_string()));
         self
     }
 
-    fn as_vec(&self) -> Vec<(&str, &str)> {
-        self.props.iter().map(|(k, v)| (*k, v.as_str())).collect()
+    fn as_vec(&self) -> &Vec<(&str, String)> {
+        &self.props
     }
 }
 
-#[cfg_attr(any(test, feature = "testing"), mockall::automock, allow(dead_code))]
 impl Zfs {
     /// Lists all datasets within a pool or existing dataset.
     ///
@@ -452,7 +453,9 @@ impl Zfs {
     }
 
     /// Get information about datasets within a list of zpools / datasets.
-    /// Returns properties for all input datasets and their direct children.
+    /// Returns properties for all input datasets, and optionally, for
+    /// their children (depending on the value of [WhichDatasets] is provided
+    /// as input).
     ///
     /// This function is similar to [Zfs::list_datasets], but provides a more
     /// substantial results about the datasets found.
@@ -721,9 +724,9 @@ impl Zfs {
         Self::set_values(filesystem_name, &[(name, value)])
     }
 
-    fn set_values<'a>(
-        filesystem_name: &'a str,
-        name_values: &'a [(&'a str, &'a str)],
+    fn set_values<K: std::fmt::Display, V: std::fmt::Display>(
+        filesystem_name: &str,
+        name_values: &[(K, V)],
     ) -> Result<(), SetValueError> {
         if name_values.is_empty() {
             return Ok(());
@@ -828,10 +831,7 @@ impl Zfs {
             err,
         })
     }
-}
 
-// These methods don't work with mockall, so they exist in a separate impl block
-impl Zfs {
     /// Calls "zfs get" to acquire multiple values
     ///
     /// - `names`: The properties being acquired

sled-storage/Cargo.toml

@@ -28,6 +28,7 @@ sled-hardware.workspace = true
 slog.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
+tokio-stream.workspace = true
 uuid.workspace = true
 omicron-workspace-hack.workspace = true
 

sled-storage/src/manager.rs

@@ -14,6 +14,8 @@ use camino::Utf8Path;
 use camino::Utf8PathBuf;
 use debug_ignore::DebugIgnore;
 use futures::future::FutureExt;
+use futures::Stream;
+use futures::StreamExt;
 use illumos_utils::zfs::{DatasetProperties, Mountpoint, WhichDatasets, Zfs};
 use illumos_utils::zpool::{ZpoolName, ZPOOL_MOUNTPOINT_ROOT};
 use key_manager::StorageKeyRequester;
@@ -930,7 +932,7 @@ impl StorageManager {
     // includes details about all possible errors that may occur on
     // a per-dataset granularity.
     async fn datasets_ensure_internal(
-        &mut self,
+        &self,
         log: &Logger,
         config: &DatasetsConfig,
     ) -> DatasetsManagementResult {
@@ -951,7 +953,6 @@ impl StorageManager {
         .map(|props| (props.name.clone(), props))
         .collect::<BTreeMap<String, _>>();
 
-        // Ensure each dataset concurrently
         let futures = config.datasets.values().map(|dataset| async {
             self.dataset_ensure_internal(
                 log,
@@ -961,7 +962,24 @@ impl StorageManager {
             .await
         });
 
-        let status = futures::future::join_all(futures).await;
+        // This "Box::pin" is a workaround for: https://github.com/rust-lang/rust/issues/64552
+        //
+        // Ideally, we would just use:
+        //
+        // ```
+        // let status: Vec<_> = futures::stream::iter(futures)
+        //      .buffered(...)
+        //      .collect()
+        //      .await;
+        // ```
+        const DATASET_ENSURE_CONCURRENCY_LIMIT: usize = 16;
+        let results: std::pin::Pin<Box<dyn Stream<Item = _> + Send>> = Box::pin(
+            futures::stream::iter(futures)
+                .buffered(DATASET_ENSURE_CONCURRENCY_LIMIT),
+        );
+
+        let status: Vec<DatasetManagementStatus> = results.collect().await;
+
         DatasetsManagementResult { status }
     }