rust: load a disk's volumes concurrently

Opening a volume is dominated by reading its .idx into the needle map,
and the loader did them one at a time, so a disk holding thousands of
volumes needed thousands of serial index reads before the server came
up. Go's concurrentLoadingVolumes spreads the same work over
max(cores, 10) workers; do the same, keeping the directory pre-pass and
the insert serial so only the open is parallel.

seaweed-volume/src/storage/disk_location.rs

@@ -7,8 +7,8 @@
 use std::collections::{HashMap, HashSet};
 use std::fs;
 use std::io;
-use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU64, Ordering};
-use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
 
 use tracing::warn;
 
@@ -122,6 +122,8 @@ impl DiskLocation {
         // Scan for .dat files
         let entries = fs::read_dir(&self.directory)?;
         let mut dat_files: Vec<(String, VolumeId)> = Vec::new();
+        let mut to_load: Vec<(String, VolumeId)> = Vec::new();
+        let mut queued: HashSet<VolumeId> = HashSet::new();
         let mut seen = HashSet::new();
 
         for entry in entries {
@@ -201,6 +203,12 @@ impl DiskLocation {
                 continue;
             }
 
+            // Skip volumes already queued under a different collection name for
+            // the same id -- the loader below keys results by id.
+            if !queued.insert(vid) {
+                continue;
+            }
+
             // Load existing data only; never create a phantom `.dat`. A lone
             // `.vif`/`.idx` (e.g. an EC sidecar whose `.ecx` is on a sibling
             // disk) would otherwise have Volume::new write an 8-byte stub that
@@ -221,28 +229,14 @@ impl DiskLocation {
                 continue;
             }
 
-            match Volume::new(
-                &self.directory,
-                &self.idx_directory,
-                &collection,
-                vid,
-                needle_map_kind,
-                None, // replica placement read from superblock
-                None, // TTL read from superblock
-                0,    // no preallocate on load
-                Version::current(),
-            ) {
-                Ok(mut v) => {
-                    v.location_disk_space_low = self.is_disk_space_low.clone();
-                    crate::metrics::VOLUME_GAUGE
-                        .with_label_values(&[&collection, "volume"])
-                        .inc();
-                    self.volumes.insert(vid, v);
-                }
-                Err(e) => {
-                    warn!(volume_id = vid.0, error = %e, "failed to load volume");
-                }
-            }
+            to_load.push((collection, vid));
+        }
+
+        for (collection, vid, v) in self.open_volumes(to_load, needle_map_kind) {
+            crate::metrics::VOLUME_GAUGE
+                .with_label_values(&[&collection, "volume"])
+                .inc();
+            self.volumes.insert(vid, v);
         }
 
         // After regular volumes, auto-discover EC shards on disk so a
@@ -257,6 +251,60 @@ impl DiskLocation {
         Ok(())
     }
 
+    /// Open the volumes the directory scan selected, on a pool of worker
+    /// threads. Opening a volume is dominated by reading its .idx into the
+    /// needle map, so a disk holding thousands of them takes thousands of
+    /// serial index reads to come up. Mirrors Go's concurrentLoadingVolumes,
+    /// including its max(cores, 10) worker count -- the work is IO-bound, so
+    /// the floor keeps a small-core box from loading one volume at a time.
+    fn open_volumes(
+        &self,
+        to_load: Vec<(String, VolumeId)>,
+        needle_map_kind: NeedleMapKind,
+    ) -> Vec<(String, VolumeId, Volume)> {
+        if to_load.is_empty() {
+            return Vec::new();
+        }
+        let workers = std::thread::available_parallelism()
+            .map(|n| n.get())
+            .unwrap_or(1)
+            .max(10)
+            .min(to_load.len());
+
+        let next = AtomicUsize::new(0);
+        let opened = Mutex::new(Vec::with_capacity(to_load.len()));
+        std::thread::scope(|scope| {
+            for _ in 0..workers {
+                scope.spawn(|| loop {
+                    let i = next.fetch_add(1, Ordering::Relaxed);
+                    let Some((collection, vid)) = to_load.get(i) else {
+                        return;
+                    };
+                    match Volume::new(
+                        &self.directory,
+                        &self.idx_directory,
+                        collection,
+                        *vid,
+                        needle_map_kind,
+                        None, // replica placement read from superblock
+                        None, // TTL read from superblock
+                        0,    // no preallocate on load
+                        Version::current(),
+                    ) {
+                        Ok(mut v) => {
+                            v.location_disk_space_low = self.is_disk_space_low.clone();
+                            opened.lock().unwrap().push((collection.clone(), *vid, v));
+                        }
+                        Err(e) => {
+                            warn!(volume_id = vid.0, error = %e, "failed to load volume");
+                        }
+                    }
+                });
+            }
+        });
+        opened.into_inner().unwrap_or_else(|e| e.into_inner())
+    }
+
     /// Directory pre-pass that recovers interrupted compaction commits. Collects
     /// every volume id that still has a .cpc commit marker or a leftover
     /// .cpd/.cpx temp file across the data and idx directories, then rolls the