diff --git a/src/archiver/archiver_impl.rs b/src/archiver/archiver_impl.rs index 35980f9..80a6671 100644 --- a/src/archiver/archiver_impl.rs +++ b/src/archiver/archiver_impl.rs @@ -5,13 +5,16 @@ use std::rc::Rc; use anyhow::{anyhow, Result}; use bytesize::ByteSize; +use futures::{stream::FuturesOrdered, StreamExt}; use indicatif::ProgressBar; +use tokio::spawn; use vlog::*; use crate::backend::DecryptWriteBackend; use crate::blob::{BlobType, Metadata, Node, NodeType, Packer, Tree}; use crate::chunker::ChunkIter; use crate::crypto::hash; +use crate::id::Id; use crate::index::{IndexedBackend, Indexer}; use crate::repo::SnapshotFile; @@ -174,26 +177,51 @@ impl Archiver { let mut content = Vec::new(); let mut filesize: u64 = 0; + let mut queue = FuturesOrdered::new(); + for chunk in chunk_iter { let chunk = chunk?; let size = chunk.len() as u64; filesize += size; - let id = hash(&chunk); - if !self.index.has_data(&id) - && self.data_packer.add(&chunk, &id, BlobType::Data).await? - { - self.data_blobs_written += 1; - self.data_added += size; + + queue.push(spawn(async move { + let id = hash(&chunk); + (id, chunk, size) + })); + + if queue.len() > 8 { + let (id, chunk, size) = queue.next().await.unwrap()?; + self.process_data_junk(id, &chunk, size, &p).await?; + content.push(id); } - content.push(id); - p.inc(size) } + + while let Some(Ok((id, chunk, size))) = queue.next().await { + self.process_data_junk(id, &chunk, size, &p).await?; + content.push(id); + } + let mut node = node; node.set_content(content); self.add_node(node, filesize); Ok(()) } + async fn process_data_junk( + &mut self, + id: Id, + chunk: &[u8], + size: u64, + p: &ProgressBar, + ) -> Result<()> { + if !self.index.has_data(&id) && self.data_packer.add(&chunk, &id, BlobType::Data).await? { + self.data_blobs_written += 1; + self.data_added += size; + } + p.inc(size); + Ok(()) + } + pub async fn finalize_snapshot(&mut self, mut snap: SnapshotFile) -> Result<()> { self.finish_trees(&PathBuf::from("/")).await?;