diff --git a/libraries/droplet/src/main.rs b/libraries/droplet/src/main.rs index baff5528..9f163ab4 100644 --- a/libraries/droplet/src/main.rs +++ b/libraries/droplet/src/main.rs @@ -17,6 +17,7 @@ pub async fn main() { |message| { println!("{}", message); }, + &None, ) .await .unwrap(); diff --git a/libraries/droplet/src/manifest.rs b/libraries/droplet/src/manifest.rs index d8f38cde..9274ad2c 100644 --- a/libraries/droplet/src/manifest.rs +++ b/libraries/droplet/src/manifest.rs @@ -1,11 +1,7 @@ use std::{ - collections::HashMap, - future::Future, - path::Path, - sync::{ - atomic::{AtomicU64, Ordering}, - Arc, - }, + collections::HashMap, future::Future, path::Path, sync::{ + Arc, atomic::{AtomicU64, Ordering} + } }; use anyhow::{anyhow, Error}; @@ -13,9 +9,12 @@ use futures::{stream::FuturesUnordered, StreamExt}; use hex::ToHex as _; use humansize::{format_size, BINARY}; use serde::{Deserialize, Serialize}; -use serde_json::json; use sha2::{Digest as _, Sha256}; -use tokio::{io::AsyncReadExt as _, join, sync::Mutex}; +use tokio::{ + io::AsyncReadExt as _, + join, + sync::{Mutex, Semaphore}, +}; #[derive(Serialize, Deserialize, Clone)] pub struct FileEntry { @@ -49,6 +48,7 @@ pub async fn generate_manifest_rusty( dir: &Path, progress_sfn: V, log_sfn: T, + reader_semaphore: &Option, ) -> anyhow::Result { let mut backend = create_backend_constructor(dir).ok_or(anyhow!("Could not create backend for path."))?()?; @@ -56,7 +56,7 @@ pub async fn generate_manifest_rusty( let required_single_file = backend.require_whole_files(); let mut files = backend.list_files().await?; - files.sort_by(|a, b| b.size.cmp(&a.size)); + files.sort_by_key(|b| std::cmp::Reverse(b.size)); // Filepath to chunk data let mut chunks: Vec> = Vec::new(); let mut current_chunk: Vec<(VersionFile, u64, u64)> = Vec::new(); @@ -109,7 +109,7 @@ pub async fn generate_manifest_rusty( // Fill up current chunk let remaining = CHUNK_SIZE - current_size; current_chunk.push((version_file.clone(), 0, remaining)); - chunks.push(std::mem::replace(&mut current_chunk, Vec::new())); + chunks.push(std::mem::take(&mut current_chunk)); // This is our offset in our current file let mut offset = remaining; @@ -165,6 +165,12 @@ pub async fn generate_manifest_rusty( let mut chunk_length = 0; for (file, start, length) in chunk { + let permit = if let Some(reader_semaphore) = &reader_semaphore { + Some(reader_semaphore.acquire().await?) + } else { + None + }; + let mut reader = { let mut backend_lock = backend.lock().await; let reader = backend_lock.reader(&file, start, start + length).await?; @@ -187,6 +193,8 @@ pub async fn generate_manifest_rusty( length: length.try_into().unwrap(), permissions: file.permission, }); + + drop(permit); } send_log diff --git a/libraries/droplet/src/versions/backends.rs b/libraries/droplet/src/versions/backends.rs index 3bef9dcd..ecd6876b 100644 --- a/libraries/droplet/src/versions/backends.rs +++ b/libraries/droplet/src/versions/backends.rs @@ -185,7 +185,7 @@ impl VersionBackend for ZipVersionBackend { let version_file = VersionFile { relative_filename: name.to_string(), permission: 0o744, - size: size, + size, }; results.push(version_file);