feat: optional reader semaphore

libraries/droplet/src/main.rs

@@ -17,6 +17,7 @@ pub async fn main() {
         |message| {
             println!("{}", message);
         },
+        &None,
     )
     .await
     .unwrap();

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<T: Fn(String), V: Fn(f32)>(
     dir: &Path,
     progress_sfn: V,
     log_sfn: T,
+    reader_semaphore: &Option<Semaphore>,
 ) -> anyhow::Result<Manifest> {
     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<T: Fn(String), V: Fn(f32)>(
     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<(VersionFile, u64, u64)>> = Vec::new();
     let mut current_chunk: Vec<(VersionFile, u64, u64)> = Vec::new();
@@ -109,7 +109,7 @@ pub async fn generate_manifest_rusty<T: Fn(String), V: Fn(f32)>(
             // 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<T: Fn(String), V: Fn(f32)>(
             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<T: Fn(String), V: Fn(f32)>(
                     length: length.try_into().unwrap(),
                     permissions: file.permission,
                 });
+
+                drop(permit);
             }
 
             send_log

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);