perf: reuse per-thread zstd output buffer for prune trees

crates/backend/src/opendal.rs

@@ -34,6 +34,14 @@ use crate::reqwest::reqwest_client;
 mod constants {
     /// Default number of retries
     pub(super) const DEFAULT_RETRY: usize = 5;
+
+    /// B2 `b2_list_file_names` page size to request.
+    ///
+    /// `OpenDAL` only sends `maxFileCount` when `ListOptions.limit` is set. If we
+    /// omit it, B2 defaults to 100 names per page (max 10000). Prune (and
+    /// check) list every pack under `data/`, so the default turns a few dozen
+    /// round-trips into thousands and dominates runtime against B2.
+    pub(super) const B2_LIST_PAGE_SIZE: usize = 10_000;
 }
 
 /// `OpenDALBackend` contains a wrapper around an blocking operator of the `OpenDAL` library.
@@ -217,6 +225,30 @@ impl OpenDALBackend {
         Ok(Self { operator })
     }
 
+    /// Listing options used for repository (and source) listings.
+    ///
+    /// For B2 this raises the per-request page size from the API default of 100
+    /// to the documented maximum of 10000.
+    fn list_options(&self, recursive: bool) -> ListOptions {
+        ListOptions {
+            recursive,
+            limit: self.list_page_size(),
+            ..Default::default()
+        }
+    }
+
+    /// Backend-specific list page size, if we should override the service default.
+    fn list_page_size(&self) -> Option<usize> {
+        let info = self.operator.info();
+        if !info.capability().list_with_limit {
+            return None;
+        }
+        match info.scheme() {
+            "b2" => Some(constants::B2_LIST_PAGE_SIZE),
+            _ => None,
+        }
+    }
+
     /// Return a path for the given file type and id.
     ///
     /// # Arguments
@@ -249,10 +281,7 @@ impl OpenDALBackend {
     /// # Errors
     /// If listing fails or exclude patterns cannot be compiled
     pub fn as_source(self, excludes: &Excludes) -> RusticResult<OpenDALReadSource> {
-        let list_options = ListOptions {
-            recursive: true,
-            ..Default::default()
-        };
+        let list_options = self.list_options(true);
         // openDAL lister may entries in random order; hence we collect and sort them here.
         // This also allows to handle listing errors directly
         let mut entries: Vec<_> = self
@@ -348,14 +377,9 @@ impl ReadBackend for OpenDALBackend {
         }
 
         let path = tpe.dirname().to_string() + "/";
-        let list_options = ListOptions {
-            recursive: true,
-            ..Default::default()
-        };
-
         let lister = self
             .operator
-            .lister_options(&path, list_options)
+            .lister_options(&path, self.list_options(true))
             .map_err(|err| {
                 RusticError::with_source(ErrorKind::Backend, "Listing failed for `{type}`", err)
                     .attach_context("type", tpe.to_string())
@@ -405,13 +429,9 @@ impl ReadBackend for OpenDALBackend {
         }
 
         let path = tpe.dirname().to_string() + "/";
-        let list_options = ListOptions {
-            recursive: true,
-            ..Default::default()
-        };
         let lister = self
             .operator
-            .lister_options(&path, list_options)
+            .lister_options(&path, self.list_options(true))
             .map_err(|err| {
                 RusticError::with_source(ErrorKind::Backend, "Listing failed for `{type}`", err)
                     .attach_context("type", tpe.to_string())
@@ -632,6 +652,28 @@ mod tests {
         assert!(Throttle::from_str(input).is_err());
     }
 
+    #[rstest]
+    #[case("b2", Some(constants::B2_LIST_PAGE_SIZE))]
+    #[case("s3_aws", None)]
+    fn list_page_size_matches_scheme(
+        #[case] fixture: &str,
+        #[case] expected: Option<usize>,
+    ) -> Result<()> {
+        #[derive(Deserialize)]
+        struct TestCase {
+            path: String,
+            options: BTreeMap<String, String>,
+        }
+
+        let fixture_path = PathBuf::from(format!("tests/fixtures/opendal/{fixture}.toml"));
+        let test: TestCase = toml::from_str(&fs::read_to_string(fixture_path)?)?;
+        let backend = OpenDALBackend::new(test.path, test.options)?;
+
+        assert_eq!(backend.list_page_size(), expected);
+        assert_eq!(backend.list_options(true).limit, expected);
+        Ok(())
+    }
+
     #[rstest]
     fn new_opendal_backend(
         #[files("tests/fixtures/opendal/*.toml")] test_case: PathBuf,

crates/core/Cargo.toml

@@ -53,6 +53,7 @@ log = { workspace = true }
 crossbeam-channel = "0.5.15"
 pariter = "0.6.0"
 rayon = "1.11.0"
+rustc-hash = "2.1.1"
 
 # crypto
 aes256ctr_poly1305aes = { version = "0.2.1", features = ["std"] } # we need std here for error impls
@@ -75,7 +76,7 @@ cached = { version = "2.0.2", default-features = false, features = ["proc_macro"
 dunce = "1.0.5"
 filetime = "0.2.27"
 ignore = "0.4.25"
-nix = { version = "0.31.1", default-features = false, features = ["user", "fs"] }
+nix = { version = "0.31.1", default-features = false, features = ["user", "fs", "resource"] }
 path-dedot = "4.0.1"
 walkdir = "2.5.0"
 

crates/core/src/backend.rs

@@ -15,6 +15,7 @@ use std::{io::Read, ops::Deref, path::PathBuf, sync::Arc};
 use bytes::{Buf, Bytes, buf::Reader};
 use enum_map::Enum;
 use log::trace;
+use rayon::current_num_threads;
 
 #[cfg(test)]
 use mockall::mock;
@@ -105,6 +106,22 @@ pub trait ReadBackend: Send + Sync + 'static {
     /// * If the files could not be listed.
     fn list_with_size(&self, tpe: FileType) -> RusticResult<Vec<(Id, u32)>>;
 
+    /// How many parallel readers to use for `stream_list` of these files.
+    ///
+    /// Remote backends keep extra workers for B2 RTTs. Cached backends use a
+    /// few workers when the files are already on disk so we do not thrash HDD.
+    fn prefetch_workers(&self, _tpe: FileType, _ids: &[Id]) -> usize {
+        (current_num_threads() + 16).clamp(16, 32)
+    }
+
+    /// How many threads to spawn to load trees (`TreeStreamer`).
+    ///
+    /// Remote backends use `2 × CPUs` (8–32) so prune is not RTT-bound on B2.
+    /// Cached backends use fewer when pack files are already on disk.
+    fn tree_loader_count(&self) -> usize {
+        current_num_threads().saturating_mul(2).clamp(8, 32)
+    }
+
     /// Lists all files of the given type.
     ///
     /// # Arguments
@@ -448,6 +465,12 @@ impl ReadBackend for Arc<dyn WriteBackend> {
     fn list_with_size(&self, tpe: FileType) -> RusticResult<Vec<(Id, u32)>> {
         self.deref().list_with_size(tpe)
     }
+    fn prefetch_workers(&self, tpe: FileType, ids: &[Id]) -> usize {
+        self.deref().prefetch_workers(tpe, ids)
+    }
+    fn tree_loader_count(&self) -> usize {
+        self.deref().tree_loader_count()
+    }
     fn list(&self, tpe: FileType) -> RusticResult<Vec<Id>> {
         self.deref().list(tpe)
     }