diff --git a/crates/backend/src/opendal.rs b/crates/backend/src/opendal.rs index c830c8cea..7ff11f711 100644 --- a/crates/backend/src/opendal.rs +++ b/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 { + 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 { - 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, + ) -> Result<()> { + #[derive(Deserialize)] + struct TestCase { + path: String, + options: BTreeMap, + } + + 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, diff --git a/crates/core/src/backend/decrypt.rs b/crates/core/src/backend/decrypt.rs index c36d74ac5..bdc1ec25c 100644 --- a/crates/core/src/backend/decrypt.rs +++ b/crates/core/src/backend/decrypt.rs @@ -190,17 +190,26 @@ pub trait DecryptReadBackend: ReadBackend + Clone + 'static { /// If the files could not be read. fn stream_list(&self, list: Vec, p: &Progress) -> StreamResult { p.set_length(list.len() as u64); - // we use a zero-capacity channel; the loading is typically the bottleneck, not the processing. - let (tx, rx) = bounded(0); + // Index/snapshot files are small; on B2 this is RTT-bound (one GET each). + // Restic uses `connections + GOMAXPROCS`. Keep extra workers so some can + // GET while others decrypt/parse, and buffer so send does not stall IO. + let workers = (rayon::current_num_threads() + 16).clamp(16, 32); + let (tx, rx) = bounded(workers.saturating_mul(2)); let be = self.clone(); let p = p.clone(); spawn(move || { - _ = list.into_par_iter().try_for_each(|id| { - let file = be.get_file::(&id).map(|file| (id, file)); - p.inc(1); - tx.send(file).ok() // abort as soon as possible if sending fails, i.e. if the receiver is dropped - }); + let work = || { + _ = list.into_par_iter().try_for_each(|id| { + let file = be.get_file::(&id).map(|file| (id, file)); + p.inc(1); + tx.send(file).ok() + }); + }; + match rayon::ThreadPoolBuilder::new().num_threads(workers).build() { + Ok(pool) => pool.install(work), + Err(_) => work(), + } }); Ok(rx) } diff --git a/crates/core/src/blob.rs b/crates/core/src/blob.rs index 4ba312f22..8b410c698 100644 --- a/crates/core/src/blob.rs +++ b/crates/core/src/blob.rs @@ -13,8 +13,13 @@ pub(super) mod constants { /// The maximum size of pack-part which is read at once from the backend. /// (needed to limit the memory size used for large backends) pub(crate) const LIMIT_PACK_READ: u32 = 40 * 1024 * 1024; // 40 MiB - /// The maximum size of holes which are still read when repacking - pub(crate) const MAX_HOLESIZE: u32 = 256 * 1024; // 256 kiB + /// Maximum unused gap that is still fetched with the surrounding blobs. + /// + /// 256 KiB was too small for high-latency object stores (B2): every larger + /// hole became another HTTP range GET, and prune/restore issued those + /// sequentially. 4 MiB is about one RTT of extra download on a ~100 Mbps + /// link, which is cheaper than an extra request. + pub(crate) const MAX_HOLESIZE: u32 = 4 * 1024 * 1024; // 4 MiB } /// All [`BlobType`]s which are supported by the repository diff --git a/crates/core/src/blob/tree.rs b/crates/core/src/blob/tree.rs index d2fbe4083..19a7c42c8 100644 --- a/crates/core/src/blob/tree.rs +++ b/crates/core/src/blob/tree.rs @@ -5,7 +5,7 @@ pub mod rewrite; use std::{ borrow::Cow, cmp::Ordering, - collections::{BTreeMap, BTreeSet, BinaryHeap}, + collections::{BTreeMap, BinaryHeap, HashSet}, ffi::OsStr, mem, path::{Component, Path, PathBuf, Prefix}, @@ -16,6 +16,7 @@ use crossbeam_channel::{Receiver, Sender, bounded, unbounded}; use derive_setters::Setters; use ignore::Match; use ignore::overrides::Override; +use rayon::current_num_threads; use serde::{Deserialize, Deserializer}; use serde_derive::Serialize; @@ -53,8 +54,20 @@ pub enum TreeErrorKind { pub(crate) type TreeResult = Result; pub(super) mod constants { - /// The maximum number of trees that are loaded in parallel - pub(super) const MAX_TREE_LOADER: usize = 4; + /// Minimum / maximum tree-loader threads for `TreeStreamerOnce`. + /// + /// Four was too few on high-latency backends (B2): prune's "finding used + /// blobs..." walks every unique tree with a pack range GET. Restic uses + /// `connections + GOMAXPROCS` workers. We scale with Rayon (2× CPUs, + /// clamped) so a 4-core box gets 8 loaders, not 4. + pub(super) const MIN_TREE_LOADER: usize = 8; + pub(super) const MAX_TREE_LOADER: usize = 32; +} + +fn tree_loader_count() -> usize { + current_num_threads() + .saturating_mul(2) + .clamp(constants::MIN_TREE_LOADER, constants::MAX_TREE_LOADER) } pub(crate) type TreeStreamItem = RusticResult<(PathBuf, Tree)>; @@ -623,7 +636,7 @@ where #[derive(Debug)] pub struct TreeStreamerOnce { /// The visited tree IDs - visited: BTreeSet, + visited: HashSet, /// The queue to send tree IDs to queue_in: Option>, /// The queue to receive trees from @@ -661,10 +674,11 @@ impl TreeStreamerOnce { ) -> RusticResult { p.set_length(ids.len() as u64); - let (out_tx, out_rx) = bounded(constants::MAX_TREE_LOADER); + let loaders = tree_loader_count(); + let (out_tx, out_rx) = bounded(loaders.saturating_mul(4).max(32)); let (in_tx, in_rx) = unbounded(); - for _ in 0..constants::MAX_TREE_LOADER { + for _ in 0..loaders { let be = be.clone(); let index = index.clone(); let in_rx = in_rx.clone(); @@ -683,7 +697,7 @@ impl TreeStreamerOnce { let counter = vec![0; ids.len()]; let mut streamer = Self { - visited: BTreeSet::new(), + visited: HashSet::new(), queue_in: Some(in_tx), queue_out: out_rx, p, diff --git a/crates/core/src/commands/prune.rs b/crates/core/src/commands/prune.rs index 306a5ab42..3229a76b3 100644 --- a/crates/core/src/commands/prune.rs +++ b/crates/core/src/commands/prune.rs @@ -4,7 +4,7 @@ /// accessors along with logging macros. Customize as you see fit. use std::{ cmp::Ordering, - collections::{BTreeMap, BTreeSet}, + collections::{BTreeMap, BTreeSet, HashMap}, str::FromStr, }; @@ -584,7 +584,7 @@ pub struct PrunePlan { /// The time the plan was created time: Zoned, /// The ids of the blobs which are used - used_ids: BTreeMap, + used_ids: HashMap, /// The ids of the existing packs existing_packs: BTreeMap, /// The packs which should be repacked @@ -604,7 +604,7 @@ impl PrunePlan { /// * `existing_packs` - The ids of the existing packs /// * `index_files` - The index files fn new( - used_ids: BTreeMap, + used_ids: HashMap, existing_packs: BTreeMap, index_files: Vec<(IndexId, IndexFile)>, ) -> Self { @@ -1416,14 +1416,15 @@ pub(crate) fn prune_repository( }) .collect(); - // TODO: repack in parallel - for blobs in blob_chunks { + // Range-GETs for holes in the same pack run in parallel. The + // packer already serializes writes via its channel / lock. + blob_chunks.into_par_iter().try_for_each(|blobs| { if opts.fast_repack { - repacker.copy_fast(blobs, &p)?; + repacker.copy_fast(blobs, &p) } else { - repacker.copy(blobs, &p)?; + repacker.copy(blobs, &p) } - } + })?; Ok(()) })?; _ = tree_repacker.finalize()?; @@ -1491,8 +1492,8 @@ impl PackInfo { /// # Arguments /// /// * `pack` - The `PrunePack` to create the `PackInfo` from - /// * `used_ids` - The `BTreeMap` of used ids - fn from_pack(pack: &PrunePack, used_ids: &mut BTreeMap) -> Self { + /// * `used_ids` - The map of used ids + fn from_pack(pack: &PrunePack, used_ids: &mut HashMap) -> Self { let mut pi = Self { blob_type: pack.blob_type, used_blobs: 0, @@ -1584,7 +1585,7 @@ fn find_used_blobs( be: &impl DecryptReadBackend, index: &impl ReadGlobalIndex, ignore_snaps: &[SnapshotId], -) -> RusticResult> { +) -> RusticResult> { let ignore_snaps: BTreeSet<_> = ignore_snaps.iter().collect(); let p = repo.progress_counter("reading snapshots..."); @@ -1601,7 +1602,7 @@ fn find_used_blobs( .try_collect()?; p.finish(); - let mut ids: BTreeMap<_, _> = snap_trees + let mut ids: HashMap<_, _> = snap_trees .iter() .map(|id| (BlobId::from(**id), 0)) .collect(); diff --git a/crates/core/src/index/binarysorted.rs b/crates/core/src/index/binarysorted.rs index f399da24c..4e9e41a8d 100644 --- a/crates/core/src/index/binarysorted.rs +++ b/crates/core/src/index/binarysorted.rs @@ -1,3 +1,5 @@ +use std::cmp::Ordering; + use rayon::prelude::*; use crate::{ @@ -20,6 +22,77 @@ pub(crate) struct SortedEntry { location: BlobLocation, } +/// Max entries in one collector chunk. +/// +/// Growing a single `Vec` of blob ids doubles it. On a large repo that request +/// is hundreds of MiB while the old buffer is still live, and musl aborts: +/// `memory allocation of N bytes failed`. Chunks cap each allocation. +const ENTRY_CHUNK_LEN: usize = if cfg!(test) { 4 } else { 1 << 20 }; + +/// Append-only vec of bounded chunks. Lookups binary-search every chunk. +#[derive(Debug)] +pub(crate) struct Chunked { + chunks: Vec>, +} + +impl Default for Chunked { + fn default() -> Self { + Self { chunks: Vec::new() } + } +} + +impl Chunked { + fn push(&mut self, item: T) { + if self + .chunks + .last() + .is_none_or(|chunk| chunk.len() >= ENTRY_CHUNK_LEN) + { + self.chunks.push(Vec::with_capacity(ENTRY_CHUNK_LEN)); + } + self.chunks + .last_mut() + .expect("chunk is created above") + .push(item); + } + + fn shrink_last(&mut self) { + if let Some(last) = self.chunks.last_mut() { + last.shrink_to_fit(); + } + } + + fn par_sort_unstable(&mut self) + where + T: Ord + Send, + { + self.chunks.par_iter_mut().for_each(|chunk| { + chunk.sort_unstable(); + }); + } + + fn par_sort_unstable_by(&mut self, compare: F) + where + T: Send, + F: Fn(&T, &T) -> Ordering + Sync, + { + self.chunks.par_iter_mut().for_each(|chunk| { + chunk.sort_unstable_by(&compare); + }); + } + + fn par_sort_unstable_by_key(&mut self, f: F) + where + T: Send, + K: Ord, + F: Fn(&T) -> K + Sync, + { + self.chunks.par_iter_mut().for_each(|chunk| { + chunk.sort_unstable_by_key(&f); + }); + } +} + /// `IndexType` determines which information is stored in the index. #[derive(Debug, Clone, Copy)] pub enum IndexType { @@ -36,8 +109,8 @@ pub enum IndexType { pub(crate) enum EntriesVariants { #[default] None, - Ids(Vec), - FullEntries(Vec), + Ids(Chunked), + FullEntries(Chunked), } #[derive(Default, Debug)] @@ -54,7 +127,8 @@ pub struct IndexCollector(BlobTypeMap); pub struct PackIndexes { c: Index, tpe: BlobType, - idx: BlobTypeMap<(u32, usize)>, + pack_idx: BlobTypeMap, + cursors: BlobTypeMap>, } #[derive(Debug)] @@ -89,11 +163,11 @@ impl IndexCollector { pub fn new(tpe: IndexType) -> Self { let mut collector = Self::default(); - collector.0[BlobType::Tree].entries = EntriesVariants::FullEntries(Vec::new()); + collector.0[BlobType::Tree].entries = EntriesVariants::FullEntries(Chunked::default()); collector.0[BlobType::Data].entries = match tpe { IndexType::OnlyTrees => EntriesVariants::None, - IndexType::DataIds => EntriesVariants::Ids(Vec::new()), - IndexType::Full => EntriesVariants::FullEntries(Vec::new()), + IndexType::DataIds => EntriesVariants::Ids(Chunked::default()), + IndexType::Full => EntriesVariants::FullEntries(Chunked::default()), }; collector @@ -110,8 +184,14 @@ impl IndexCollector { Index(self.0.map(|_, mut tc| { match &mut tc.entries { EntriesVariants::None => {} - EntriesVariants::Ids(ids) => ids.par_sort_unstable(), - EntriesVariants::FullEntries(entries) => entries.par_sort_unstable_by_key(|e| e.id), + EntriesVariants::Ids(ids) => { + ids.shrink_last(); + ids.par_sort_unstable(); + } + EntriesVariants::FullEntries(entries) => { + entries.shrink_last(); + entries.par_sort_unstable_by_key(|e| e.id); + } } let packs = tc.packs.into_iter().map(|(id, _)| id).collect(); @@ -130,7 +210,6 @@ impl Extend for IndexCollector { T: IntoIterator, { for p in iter { - let len = p.blobs.len(); let blob_type = p.blob_type(); let size = p.pack_size(); @@ -140,12 +219,6 @@ impl Extend for IndexCollector { self.0[blob_type].total_size += u64::from(size); - match &mut self.0[blob_type].entries { - EntriesVariants::None => {} - EntriesVariants::Ids(idents) => idents.reserve(len), - EntriesVariants::FullEntries(entries) => entries.reserve(len), - } - for blob in &p.blobs { let be = SortedEntry { id: blob.id, @@ -166,39 +239,45 @@ impl Iterator for PackIndexes { type Item = IndexPack; fn next(&mut self) -> Option { - let (pack_idx, idx) = loop { - let (pack_idx, idx) = &mut self.idx[self.tpe]; - let pack_count = u32::try_from(self.c.0[self.tpe].packs.len()) - .expect("pack count should fit into u32"); - if *pack_idx >= pack_count { - if self.tpe == BlobType::Data { + loop { + let tpe = self.tpe; + let pack_count = + u32::try_from(self.c.0[tpe].packs.len()).expect("pack count should fit into u32"); + if self.pack_idx[tpe] >= pack_count { + if tpe == BlobType::Data { return None; } self.tpe = BlobType::Data; - } else { - break (pack_idx, idx); + continue; } - }; - let mut pack = IndexPack { - id: self.c.0[self.tpe].packs[*pack_idx as usize], - ..Default::default() - }; + let pack_idx = self.pack_idx[tpe]; + let mut pack = IndexPack { + id: self.c.0[tpe].packs[pack_idx as usize], + ..Default::default() + }; - if let EntriesVariants::FullEntries(entries) = &self.c.0[self.tpe].entries { - while *idx < entries.len() && entries[*idx].pack_idx == *pack_idx { - let entry = &entries[*idx]; - pack.blobs.push(IndexBlob { - id: entry.id, - tpe: self.tpe, - location: entry.location, - }); - *idx += 1; + if let EntriesVariants::FullEntries(entries) = &self.c.0[tpe].entries { + let cursors = &mut self.cursors[tpe]; + if cursors.len() != entries.chunks.len() { + cursors.resize(entries.chunks.len(), 0); + } + for (chunk, cursor) in entries.chunks.iter().zip(cursors.iter_mut()) { + while *cursor < chunk.len() && chunk[*cursor].pack_idx == pack_idx { + let entry = &chunk[*cursor]; + pack.blobs.push(IndexBlob { + id: entry.id, + tpe, + location: entry.location, + }); + *cursor += 1; + } + } } - } - *pack_idx += 1; - Some(pack) + self.pack_idx[tpe] += 1; + return Some(pack); + } } } @@ -214,34 +293,32 @@ impl IntoIterator for Index { } } PackIndexes { - c: Self(self.0.map(|_, mut tc| { - if let EntriesVariants::FullEntries(entries) = &mut tc.entries { - entries.par_sort_unstable_by(|e1, e2| e1.pack_idx.cmp(&e2.pack_idx)); - } - - tc - })), + c: self, tpe: BlobType::Tree, - idx: BlobTypeMap::default(), + pack_idx: BlobTypeMap::default(), + cursors: BlobTypeMap::default(), } } } impl ReadIndex for Index { fn get_id(&self, blob_type: BlobType, id: &BlobId) -> Option { - let EntriesVariants::FullEntries(vec) = &self.0[blob_type].entries else { + let EntriesVariants::FullEntries(entries) = &self.0[blob_type].entries else { // get_id() only gives results if index contains full entries return None; }; - vec.binary_search_by_key(id, |e| e.id).ok().map(|index| { - let be = &vec[index]; - IndexEntry::new( - blob_type, - self.0[blob_type].packs[be.pack_idx as usize], - be.location, - ) - }) + for chunk in &entries.chunks { + if let Ok(index) = chunk.binary_search_by_key(id, |e| e.id) { + let be = &chunk[index]; + return Some(IndexEntry::new( + blob_type, + self.0[blob_type].packs[be.pack_idx as usize], + be.location, + )); + } + } + None } fn total_size(&self, blob_type: BlobType) -> u64 { @@ -250,10 +327,14 @@ impl ReadIndex for Index { fn has(&self, blob_type: BlobType, id: &BlobId) -> bool { match &self.0[blob_type].entries { - EntriesVariants::FullEntries(entries) => { - entries.binary_search_by_key(id, |e| e.id).is_ok() - } - EntriesVariants::Ids(ids) => ids.binary_search(id).is_ok(), + EntriesVariants::FullEntries(entries) => entries + .chunks + .iter() + .any(|chunk| chunk.binary_search_by_key(id, |e| e.id).is_ok()), + EntriesVariants::Ids(ids) => ids + .chunks + .iter() + .any(|chunk| chunk.binary_search(id).is_ok()), // has() only gives results if index contains full entries or ids EntriesVariants::None => false, } @@ -439,6 +520,53 @@ mod tests { ); assert!(!index.has(BlobType::Tree, &id)); assert!(index.get_id(BlobType::Tree, &id).is_none()); + + // This id is in the second test-sized chunk (ENTRY_CHUNK_LEN is 4 under cfg(test)). + let id = "ee67585c7c53324e74537ab7aa44f889c0767c1b67e7e336fae6204aef2d4c73".parse()?; + assert!(index.has(BlobType::Data, &id)); + assert_eq!( + index.get_id(BlobType::Data, &id), + Some(IndexEntry { + blob_type: BlobType::Data, + pack: "3b25ec6d16401c31099c259311562160b1b5efbcf70bd69d0463104d3b8148fc".parse()?, + location: BlobLocation { + offset: 7737, + length: 7686, + uncompressed_length: Some(NonZeroU32::new(29928).unwrap()), + } + }), + ); + Ok(()) + } + + #[test] + fn into_iter_groups_blobs_by_pack() -> RusticResult<()> { + let packs: Vec<_> = index(IndexType::Full).into_iter().collect(); + assert_eq!(packs.len(), 3); + assert_eq!( + packs[0].id, + "8431a27d38dd7d192dc37abd43a85d6dc4298de72fc8f583c5d7cdd09fa47274".parse()? + ); + assert_eq!(packs[0].blobs.len(), 2); + assert_eq!( + packs[1].id, + "217f145b63fbc10267f5a686186689ea3389bed0d6a54b50ffc84d71f99eb7fa".parse()? + ); + assert_eq!(packs[1].blobs.len(), 3); + assert_eq!( + packs[2].id, + "3b25ec6d16401c31099c259311562160b1b5efbcf70bd69d0463104d3b8148fc".parse()? + ); + assert_eq!(packs[2].blobs.len(), 4); + Ok(()) + } + + #[test] + fn data_ids_has_across_chunks() -> RusticResult<()> { + let index = index(IndexType::DataIds); + let id = "f2ca1bb6c7e907d06dafe4687e579fce76b37e4e93b7605022da52e6ccc26fd2".parse()?; + assert!(index.has(BlobType::Data, &id)); + assert!(index.get_id(BlobType::Data, &id).is_none()); Ok(()) } }