perf!: Optimize file streaming

crates/core/src/backend/decrypt.rs

@@ -1,8 +1,8 @@
 use std::{num::NonZeroU32, sync::Arc};
 
 use bytes::Bytes;
-use crossbeam_channel::{Receiver, unbounded};
-use rayon::prelude::*;
+use crossbeam_channel::{Receiver, bounded};
+use rayon::{prelude::*, spawn};
 use zstd::stream::{copy_encode, decode_all, encode_all};
 
 pub use zstd::compression_level_range;
@@ -173,7 +173,7 @@ pub trait DecryptReadBackend: ReadBackend + Clone + 'static {
     fn stream_all<F: RepoFile>(&self, p: &Progress) -> StreamResult<F::Id, F> {
         let list = self.list(F::TYPE)?;
         let list: Vec<_> = list.into_iter().map(F::Id::from).collect();
-        self.stream_list(&list, p)
+        self.stream_list(list, p)
     }
 
     /// Streams a list of files.
@@ -188,16 +188,20 @@ pub trait DecryptReadBackend: ReadBackend + Clone + 'static {
     /// # Errors
     ///
     /// If the files could not be read.
-    fn stream_list<F: RepoFile>(&self, list: &[F::Id], p: &Progress) -> StreamResult<F::Id, F> {
+    fn stream_list<F: RepoFile>(&self, list: Vec<F::Id>, p: &Progress) -> StreamResult<F::Id, F> {
         p.set_length(list.len() as u64);
-        let (tx, rx) = unbounded();
-
-        list.into_par_iter()
-            .for_each_with((self, p, tx), |(be, p, tx), id| {
-                let file = be.get_file::<F>(id).map(|file| (*id, file));
+        // we use a zero-capacity channel; the loading is typically the bottleneck, not the processing.
+        let (tx, rx) = bounded(0);
+        let be = self.clone();
+        let p = p.clone();
+
+        spawn(move || {
+            _ = list.into_par_iter().try_for_each(|id| {
+                let file = be.get_file::<F>(&id).map(|file| (id, file));
                 p.inc(1);
-                tx.send(file).unwrap();
+                tx.send(file).ok() // abort as soon as possible if sending fails, i.e. if the receiver is dropped
             });
+        });
         Ok(rx)
     }
 }

crates/core/src/commands/prune.rs

@@ -1595,7 +1595,7 @@ fn find_used_blobs<S>(
         .filter(|id| !ignore_snaps.contains(&id))
         .collect();
     let snap_trees: Vec<_> = be
-        .stream_list::<SnapshotFile>(&list, &p)?
+        .stream_list::<SnapshotFile>(list, &p)?
         .into_iter()
         .map_ok(|(_, snap)| snap.tree)
         .try_collect()?;

crates/core/src/progress.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use log::info;
 
 /// A progress used to indicate/update the status of something which is being processed
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Progress(Arc<dyn RusticProgress>);
 
 impl Progress {

crates/core/src/repofile/snapshotfile.rs

@@ -880,7 +880,7 @@ impl SnapshotFile {
             .map(|id| SnapshotId::from(*id))
             .filter(|id| !snaps.contains_key(id))
             .collect();
-        for res in be.stream_list::<Self>(&missing_ids, p)? {
+        for res in be.stream_list::<Self>(missing_ids, p)? {
             let (id, snap) = res?;
             if filter(&snap) {
                 let _ = snaps.insert(id, snap);

crates/core/src/repository.rs

@@ -1272,7 +1272,7 @@ impl<S: Open> Repository<S> {
     /// The result is not sorted and may come in random order!
     pub fn stream_files_list<F: RepoFile>(
         &self,
-        list: &[F::Id],
+        list: Vec<F::Id>,
     ) -> RusticResult<impl Iterator<Item = RusticResult<(F::Id, F)>>> {
         Ok(self
             .dbe()

crates/core/tests/integration/key.rs

@@ -46,7 +46,7 @@ fn test_key_commands(set_up_repo: Result<RepoOpen>) -> Result<()> {
     let found_ids: Vec<KeyId> = repo.find_ids(&[key_id2.to_string()])?.collect();
     assert_eq!(found_ids, &[key_id2]);
     let found_keys: Vec<KeyFile> = repo
-        .stream_files_list::<KeyFile>(&found_ids)?
+        .stream_files_list::<KeyFile>(found_ids)?
         .map(|item| item.unwrap().1)
         .collect();
     assert_eq!(found_keys.len(), 1);