From 54e46d0873896ca76993f67d3118401684084e48 Mon Sep 17 00:00:00 2001 From: Harshit Jain Date: Mon, 22 Jun 2026 21:50:43 +0530 Subject: [PATCH 1/3] irondrop: optimize search, upload, and streaming hot paths Reduce directory listing memory by keeping only the requested page window in a bounded heap. Replace chunked request front-drain copies with a cursor-based pending buffer and add a rename fast path for file-backed uploads. Cut search overhead with generation-based cache recency, bulk radix bucket construction, carried subdirectory entry ids, and slice-only pagination cloning. --- src/fs.rs | 119 ++++++++++++++++---- src/http.rs | 112 +++++++++++++++++-- src/search.rs | 302 ++++++++++++++++++++++++++++++++++---------------- src/upload.rs | 81 +++++++++++++- 4 files changed, 487 insertions(+), 127 deletions(-) diff --git a/src/fs.rs b/src/fs.rs index 4bd30d3..0633d6e 100644 --- a/src/fs.rs +++ b/src/fs.rs @@ -5,10 +5,31 @@ use crate::error::AppError; use crate::templates::TemplateEngine; use crate::utils::is_hidden_file; use log::{debug, trace}; +use std::cmp::Ordering; +use std::collections::BinaryHeap; use std::fs; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::time::SystemTime; +#[derive(Eq, PartialEq)] +struct ListingEntry { + path: PathBuf, + file_name: String, + is_dir: bool, +} + +impl Ord for ListingEntry { + fn cmp(&self, other: &Self) -> Ordering { + compare_listing_entries(self, other) + } +} + +impl PartialOrd for ListingEntry { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + /// Enhanced directory listing using modular templates - dark mode only pub fn generate_directory_listing( path: &Path, @@ -19,9 +40,13 @@ pub fn generate_directory_listing( debug!("Generating directory listing for: '{}'", path.display()); trace!("Request path: '{}'", request_path); - let mut entries = Vec::new(); + let limit = 1000; + let page_size = page.max(1); + let selection_size = page_size.saturating_mul(limit); + let mut selected_entries = BinaryHeap::with_capacity(selection_size); + let mut total_count = 0usize; - // Collect and sort entries using lazy metadata lookup (super fast, low memory) + // Keep only the best entries for the requested page instead of sorting the entire directory. trace!("Reading directory entries from: {}", path.display()); for entry in fs::read_dir(path)? { let entry = entry?; @@ -32,21 +57,30 @@ pub fn generate_directory_listing( continue; } - entries.push((entry.path(), file_name, file_type.is_dir())); - } - - debug!("Found {} entries, sorting...", entries.len()); - // Sort: directories first, then alphabetically - entries.sort_by(|a, b| { - let a_is_dir = a.2; - let b_is_dir = b.2; + total_count += 1; + let listing_entry = ListingEntry { + path: entry.path(), + file_name, + is_dir: file_type.is_dir(), + }; - match (a_is_dir, b_is_dir) { - (true, false) => std::cmp::Ordering::Less, - (false, true) => std::cmp::Ordering::Greater, - _ => cmp_case_insensitive_ascii(&a.1, &b.1), + if selection_size == 0 { + continue; + } + if selected_entries.len() < selection_size { + selected_entries.push(listing_entry); + continue; + } + if let Some(current_max) = selected_entries.peek() + && compare_listing_entries(&listing_entry, current_max) == Ordering::Less + { + selected_entries.pop(); + selected_entries.push(listing_entry); } - }); + } + + let mut entries = selected_entries.into_vec(); + entries.sort_unstable_by(compare_listing_entries); let display_path = if request_path.is_empty() || request_path == "/" { "/" @@ -54,9 +88,7 @@ pub fn generate_directory_listing( request_path }; - debug!("Preparing {} entries for template rendering", entries.len()); - let total_count = entries.len(); - let limit = 1000; + debug!("Preparing {} selected entries for template rendering", entries.len()); let total_pages = total_count.div_ceil(limit); let safe_page = page.max(1).min(total_pages.max(1)); let offset = (safe_page - 1) * limit; @@ -64,7 +96,12 @@ pub fn generate_directory_listing( let page_entries: Vec<_> = entries.into_iter().skip(offset).take(limit).collect(); let mut template_entries = Vec::with_capacity(page_entries.len()); - for (entry_path, file_name, is_dir) in page_entries { + for entry in page_entries { + let ListingEntry { + path: entry_path, + file_name, + is_dir, + } = entry; let link_name = if is_dir { format!("{file_name}/") } else { @@ -152,6 +189,48 @@ fn cmp_case_insensitive_ascii(a: &str, b: &str) -> std::cmp::Ordering { } } +fn compare_listing_entries(a: &ListingEntry, b: &ListingEntry) -> Ordering { + match (a.is_dir, b.is_dir) { + (true, false) => Ordering::Less, + (false, true) => Ordering::Greater, + _ => cmp_case_insensitive_ascii(&a.file_name, &b.file_name), + } +} + +#[cfg(test)] +mod perf_tests { + use super::*; + use tempfile::tempdir; + + #[test] + #[ignore] + fn perf_directory_listing_first_page() { + let temp_dir = tempdir().unwrap(); + + for dir_idx in 0..200 { + std::fs::create_dir_all(temp_dir.path().join(format!("dir_{dir_idx:03}"))).unwrap(); + } + for file_idx in 0..9_800 { + std::fs::write( + temp_dir.path().join(format!("file_{file_idx:05}.txt")), + b"irondrop", + ) + .unwrap(); + } + + let start = std::time::Instant::now(); + let html = generate_directory_listing(temp_dir.path(), "/", None, 1).unwrap(); + let elapsed_ms = start.elapsed().as_millis(); + + assert!(html.contains("file_")); + println!( + "PERF directory_listing entries=10000 page=1 render_ms={} html_bytes={}", + elapsed_ms, + html.len() + ); + } +} + /// Format Unix timestamp to human-readable date fn format_timestamp(timestamp: u64) -> String { // Simple date formatting without external dependencies diff --git a/src/http.rs b/src/http.rs index 338fa15..9645b79 100644 --- a/src/http.rs +++ b/src/http.rs @@ -82,6 +82,59 @@ pub enum ResponseBody { AsyncStream(tokio::sync::mpsc::Receiver>), } +struct PendingBuffer { + data: Vec, + start: usize, +} + +impl PendingBuffer { + fn new(data: Vec) -> Self { + Self { data, start: 0 } + } + + fn len(&self) -> usize { + self.data.len().saturating_sub(self.start) + } + + fn as_slice(&self) -> &[u8] { + &self.data[self.start..] + } + + fn extend_from_slice(&mut self, bytes: &[u8]) { + if self.start > 0 && (self.start >= 8192 || self.start * 2 >= self.data.len().max(1)) { + self.compact(); + } + self.data.extend_from_slice(bytes); + } + + fn take(&mut self, count: usize) -> Vec { + let end = self.start + count; + let chunk = self.data[self.start..end].to_vec(); + self.start = end; + if self.start == self.data.len() { + self.data.clear(); + self.start = 0; + } + chunk + } + + fn discard(&mut self, count: usize) { + self.start += count; + if self.start == self.data.len() { + self.data.clear(); + self.start = 0; + } + } + + fn compact(&mut self) { + if self.start == 0 { + return; + } + self.data.drain(..self.start); + self.start = 0; + } +} + impl Request { /// Validates if the given method is a valid HTTP method fn is_valid_http_method(method: &str) -> bool { @@ -659,13 +712,14 @@ where async fn read_chunked_body_async( stream: &mut S, - mut pending: Vec, + remaining_bytes: Vec, ) -> Result where S: tokio::io::AsyncRead + Unpin, { const CHUNK_LINE_LIMIT: usize = 8 * 1024; + let mut pending = PendingBuffer::new(remaining_bytes); let mut total_size: usize = 0; let mut memory_body: Vec = Vec::new(); let mut file_sink: Option<(PathBuf, tokio::fs::File)> = None; @@ -765,16 +819,16 @@ async fn create_temp_body_file_async() -> Result<(PathBuf, tokio::fs::File), App async fn read_crlf_line_async( stream: &mut S, - pending: &mut Vec, + pending: &mut PendingBuffer, max_line_len: usize, ) -> Result, AppError> where S: tokio::io::AsyncRead + Unpin, { loop { - if let Some(pos) = pending.windows(2).position(|w| w == b"\r\n") { - let line = pending[..pos].to_vec(); - pending.drain(0..pos + 2); + if let Some(pos) = pending.as_slice().windows(2).position(|w| w == b"\r\n") { + let line = pending.take(pos); + pending.discard(2); return Ok(line); } @@ -793,7 +847,7 @@ where async fn read_exact_from_buffer_async( stream: &mut S, - pending: &mut Vec, + pending: &mut PendingBuffer, count: usize, ) -> Result, AppError> where @@ -807,12 +861,12 @@ where } pending.extend_from_slice(&buffer[..n]); } - Ok(pending.drain(0..count).collect()) + Ok(pending.take(count)) } async fn consume_expected_crlf_async( stream: &mut S, - pending: &mut Vec, + pending: &mut PendingBuffer, ) -> Result<(), AppError> where S: tokio::io::AsyncRead + Unpin, @@ -826,7 +880,7 @@ where async fn consume_chunked_trailers_async( stream: &mut S, - pending: &mut Vec, + pending: &mut PendingBuffer, ) -> Result<(), AppError> where S: tokio::io::AsyncRead + Unpin, @@ -954,3 +1008,43 @@ where Ok((temp_path, total_written as u64)) } + +#[cfg(test)] +mod perf_tests { + use super::*; + use tokio::io::{AsyncWriteExt, duplex}; + + #[tokio::test] + #[ignore] + async fn perf_chunked_body_parsing() { + let chunk = vec![b'a'; 64 * 1024]; + let mut payload = Vec::new(); + for _ in 0..128 { + payload.extend_from_slice(format!("{:X}\r\n", chunk.len()).as_bytes()); + payload.extend_from_slice(&chunk); + payload.extend_from_slice(b"\r\n"); + } + payload.extend_from_slice(b"0\r\n\r\n"); + + let (mut writer, mut reader) = duplex(payload.len() + 1024); + writer.write_all(&payload).await.unwrap(); + drop(writer); + + let start = Instant::now(); + let body = read_chunked_body_async(&mut reader, Vec::new()).await.unwrap(); + let elapsed_ms = start.elapsed().as_millis(); + + match body { + RequestBody::File { size, .. } => { + assert_eq!(size, (64 * 1024 * 128) as u64); + } + RequestBody::Memory(data) => panic!("expected file-backed body, got {}", data.len()), + } + + println!( + "PERF chunked_body_parse size_bytes={} elapsed_ms={}", + 64 * 1024 * 128, + elapsed_ms + ); + } +} diff --git a/src/search.rs b/src/search.rs index d66ff55..c847c1f 100644 --- a/src/search.rs +++ b/src/search.rs @@ -46,9 +46,10 @@ pub struct SearchParams { /// LRU Cache for search results pub struct SearchCache { cache: HashMap, - order: VecDeque, + order: VecDeque<(u64, String)>, max_size: usize, stats: CacheStats, + next_generation: u64, } #[derive(Clone)] @@ -56,6 +57,7 @@ struct CachedSearchResult { results: Arc>, timestamp: Instant, hit_count: u32, + generation: u64, } #[derive(Default)] @@ -72,9 +74,31 @@ impl SearchCache { order: VecDeque::new(), max_size, stats: CacheStats::default(), + next_generation: 0, } } + fn fresh_generation(&mut self) -> u64 { + self.next_generation = self.next_generation.wrapping_add(1); + self.next_generation + } + + fn evict_oldest(&mut self) -> bool { + while let Some((generation, key)) = self.order.pop_front() { + let should_remove = self + .cache + .get(&key) + .map(|cached| cached.generation == generation) + .unwrap_or(false); + if should_remove { + self.cache.remove(&key); + self.stats.evictions += 1; + return true; + } + } + false + } + /// Force shrink cache when memory usage exceeds threshold pub fn shrink_if_needed(&mut self, force: bool) { let memory_threshold = self.max_size / 2; // Shrink when over 50% capacity @@ -88,9 +112,8 @@ impl SearchCache { }; while self.cache.len() > target_size { - if let Some(lru_key) = self.order.pop_back() { - self.cache.remove(&lru_key); - self.stats.evictions += 1; + if !self.evict_oldest() { + break; } } @@ -104,26 +127,20 @@ impl SearchCache { } pub fn get(&mut self, key: &str) -> Option>> { + let generation = self.fresh_generation(); if let Some(cached) = self.cache.get_mut(key) { // Check if cache is still valid (10 seconds TTL for better performance) if cached.timestamp.elapsed().as_secs() < 10 { cached.hit_count += 1; self.stats.hits += 1; - - // Move to front of LRU queue - if let Some(pos) = self.order.iter().position(|k| k == key) { - self.order.remove(pos); - } - self.order.push_front(key.to_string()); + cached.generation = generation; + self.order.push_back((generation, key.to_string())); debug!("Cache hit for query: {} (hits: {})", key, cached.hit_count); return Some(cached.results.clone()); } // Cache expired, remove it self.cache.remove(key); - if let Some(pos) = self.order.iter().position(|k| k == key) { - self.order.remove(pos); - } debug!("Cache expired for query: {key}"); } self.stats.misses += 1; @@ -132,24 +149,25 @@ impl SearchCache { pub fn put(&mut self, key: String, results: Arc>) { // Evict LRU item if cache is full - while self.cache.len() >= self.max_size { - if let Some(lru_key) = self.order.pop_back() { - self.cache.remove(&lru_key); - self.stats.evictions += 1; - debug!("Evicted cache entry: {lru_key}"); + let replacing_existing = self.cache.contains_key(&key); + while !replacing_existing && self.cache.len() >= self.max_size { + if !self.evict_oldest() { + break; } } let result_count = results.len(); + let generation = self.fresh_generation(); self.cache.insert( key.clone(), CachedSearchResult { results, timestamp: Instant::now(), hit_count: 0, + generation, }, ); - self.order.push_front(key.clone()); + self.order.push_back((generation, key.clone())); debug!("Cached {result_count} results for query: {key}"); } @@ -264,8 +282,16 @@ struct StringPoolEntry { /// Radix index bucket for first-byte acceleration #[derive(Default)] struct RadixBucket { - /// Sorted array of entry indices for binary search - entries: Vec, // Variable size, sorted for O(log n) search + /// Entry indices grouped by first byte for fast prefix narrowing. + entries: Vec, +} + +struct IndexedDirEntry { + path: PathBuf, + name: String, + size: u64, + is_dir: bool, + modified: SystemTime, } /// Cache-aligned memory pool for ultra-efficient string storage @@ -576,13 +602,26 @@ fn murmur3_hash(data: &[u8]) -> u32 { hash } +fn contains_case_insensitive_ascii(haystack: &str, needle_lower: &str) -> bool { + let needle = needle_lower.as_bytes(); + if needle.is_empty() { + return true; + } + haystack + .as_bytes() + .windows(needle.len()) + .any(|window| { + window + .iter() + .zip(needle.iter()) + .all(|(&lhs, &rhs)| lhs.to_ascii_lowercase() == rhs) + }) +} + impl RadixBucket { - /// Add entry to bucket, maintaining sorted order for binary search + /// Add entry to bucket during bulk index construction. fn add_entry(&mut self, entry_id: u32) { - match self.entries.binary_search(&entry_id) { - Ok(_) => {} // Already exists - Err(pos) => self.entries.insert(pos, entry_id), - } + self.entries.push(entry_id); } /// Search bucket for entries matching criteria @@ -841,37 +880,30 @@ impl UltraLowMemoryIndex { continue; } - batch_entries.push(( - file_name, - file_path.clone(), - metadata.len(), - metadata.is_dir(), + batch_entries.push(IndexedDirEntry { + path: file_path, + name: file_name, + size: metadata.len(), + is_dir: metadata.is_dir(), modified, - )); - - // Collect subdirectories for recursive processing - if metadata.is_dir() { - subdirs.push(file_path); - } + }); // Process batch when it's full if batch_entries.len() >= 1000 { - self.process_entry_batch_hierarchical(&mut batch_entries, parent_entry_id)?; + subdirs.extend( + self.process_entry_batch_hierarchical(&mut batch_entries, parent_entry_id)?, + ); } } // Process remaining entries if !batch_entries.is_empty() { - self.process_entry_batch_hierarchical(&mut batch_entries, parent_entry_id)?; + subdirs.extend(self.process_entry_batch_hierarchical(&mut batch_entries, parent_entry_id)?); } - // Recursively process subdirectories with their entry IDs as parents - for subdir in subdirs { - // Find the entry ID for this subdirectory - let subdir_name = subdir.file_name().unwrap().to_string_lossy(); - if let Some(subdir_entry_id) = self.find_entry_by_name(&subdir_name, parent_entry_id) { - self.walk_directory_hierarchical(&subdir, subdir_entry_id, depth + 1)?; - } + // Recursively process subdirectories with the entry IDs generated during batching. + for (subdir, subdir_entry_id) in subdirs { + self.walk_directory_hierarchical(&subdir, subdir_entry_id, depth + 1)?; } Ok(()) @@ -880,18 +912,26 @@ impl UltraLowMemoryIndex { /// Process batch of entries with hierarchical parent references (ultra-memory efficient) fn process_entry_batch_hierarchical( &mut self, - batch: &mut Vec<(String, PathBuf, u64, bool, SystemTime)>, + batch: &mut Vec, parent_entry_id: u32, - ) -> Result<(), AppError> { - for (name, _path, size, is_dir, modified) in batch.drain(..) { + ) -> Result, AppError> { + let mut subdirs = Vec::new(); + + for entry in batch.drain(..) { let entry_id = self.entries.len() as u32; // Add filename to unified string pool - let name_offset = self.add_string(&name); + let name_offset = self.add_string(&entry.name); // Create ultra-compact entry with parent reference let ultra_compact_entry = - UltraCompactEntry::new(name_offset, parent_entry_id, size, modified, is_dir); + UltraCompactEntry::new( + name_offset, + parent_entry_id, + entry.size, + entry.modified, + entry.is_dir, + ); self.entries.push(ultra_compact_entry); @@ -905,34 +945,20 @@ impl UltraLowMemoryIndex { } // Add to radix index for fast searching - if !name.is_empty() { - let first_byte = name.as_bytes()[0]; + if !entry.name.is_empty() { + let first_byte = entry.name.as_bytes()[0]; self.radix_index[first_byte as usize].add_entry(entry_id); } // Update entry count self.entry_count.fetch_add(1, Ordering::Relaxed); - } - - Ok(()) - } - - /// Find entry ID by name within a parent directory - fn find_entry_by_name(&self, name: &str, parent_id: u32) -> Option { - if parent_id as usize >= self.directory_children.len() { - return None; - } - for &child_id in &self.directory_children[parent_id as usize] { - if let Some(entry) = self.entries.get(child_id as usize) - && let Some(entry_name) = self.get_string(entry.get_name_offset()) - && entry_name == name - { - return Some(child_id); + if entry.is_dir { + subdirs.push((entry.path, entry_id)); } } - None + Ok(subdirs) } /// Rebuild the entire index from scratch (ultra-low memory optimized) @@ -965,7 +991,8 @@ impl UltraLowMemoryIndex { self.directory_children.push(Vec::new()); // Walk directory hierarchy starting from root - self.walk_directory_hierarchical(&self.base_dir.clone(), self.root_entry_id, 0)?; + let base_dir = self.base_dir.clone(); + self.walk_directory_hierarchical(&base_dir, self.root_entry_id, 0)?; // Build radix index for fast searching self.build_radix_index(); @@ -1014,10 +1041,19 @@ impl UltraLowMemoryIndex { info!("Building radix index for {} entries", self.entries.len()); let start = Instant::now(); - // Clear existing radix index - self.radix_index = std::array::from_fn(|_| RadixBucket::default()); + let mut bucket_sizes = [0usize; 256]; + for entry in &self.entries { + if let Some(name) = self.get_string(entry.get_name_offset()) { + let first_byte = name.as_bytes().first().copied().unwrap_or(0); + bucket_sizes[first_byte as usize] += 1; + } + } + + // Clear existing radix index and reserve once per bucket. + self.radix_index = std::array::from_fn(|idx| RadixBucket { + entries: Vec::with_capacity(bucket_sizes[idx]), + }); - // Populate radix buckets based on first character of filename for (entry_id, entry) in self.entries.iter().enumerate() { if let Some(name) = self.get_string(entry.get_name_offset()) { let first_byte = name.as_bytes().first().copied().unwrap_or(0); @@ -1041,7 +1077,8 @@ impl UltraLowMemoryIndex { ); let start = Instant::now(); let query_lower = query.to_lowercase(); - let mut candidate_ids = Vec::new(); + let query_is_ascii = query.is_ascii(); + let mut candidate_ids = Vec::with_capacity(limit.saturating_mul(2)); // Strategy 1: Radix-accelerated search using first character if !query_lower.is_empty() { @@ -1057,8 +1094,12 @@ impl UltraLowMemoryIndex { if let Some(entry) = self.entries.get(entry_id as usize) && let Some(name) = self.get_string(entry.get_name_offset()) { - let name_lower = name.to_lowercase(); - if name_lower.contains(&query_lower) { + let matches = if query_is_ascii { + contains_case_insensitive_ascii(name, &query_lower) + } else { + name.to_lowercase().contains(&query_lower) + }; + if matches { candidate_ids.push(entry_id); } } @@ -1080,8 +1121,12 @@ impl UltraLowMemoryIndex { if let Some(entry) = self.entries.get(entry_id as usize) && let Some(name) = self.get_string(entry.get_name_offset()) { - let name_lower = name.to_lowercase(); - if name_lower.contains(&query_lower) { + let matches = if query_is_ascii { + contains_case_insensitive_ascii(name, &query_lower) + } else { + name.to_lowercase().contains(&query_lower) + }; + if matches { candidate_ids.push(entry_id); } } @@ -1102,6 +1147,13 @@ impl UltraLowMemoryIndex { } } + results.sort_by(|a, b| { + b.score + .partial_cmp(&a.score) + .unwrap_or(std::cmp::Ordering::Equal) + }); + results.truncate(limit); + debug!( "Ultra-fast search completed in {:.2}ms, {} candidates -> {} results", start.elapsed().as_millis(), @@ -1664,37 +1716,44 @@ pub fn perform_search( }; // Perform ultra-fast radix-accelerated search + let expanded_limit = params + .offset + .saturating_add(params.limit) + .saturating_mul(2) + .max(params.limit.max(1)); trace!( "Starting radix-accelerated search with expanded limit: {}", - params.limit * 2 + expanded_limit ); - let mut results = concurrent_index.search(¶ms.query, params.limit * 2)?; - debug!("Index search returned {} initial results", results.len()); + let shared_results = concurrent_index.search_shared(¶ms.query, expanded_limit)?; + debug!("Index search returned {} initial results", shared_results.len()); // If index search returns no results, fall back to filesystem search - if results.is_empty() { + if shared_results.is_empty() { info!( "Ultra-low memory index search returned no results, falling back to filesystem search" ); debug!("Initiating parallel filesystem search as fallback"); - results = perform_parallel_search(base_dir, params)?; + let mut results = perform_parallel_search(base_dir, params)?; trace!( "Filesystem search fallback returned {} results", results.len() ); + let start_idx = params.offset.min(results.len()); + let end_idx = (params.offset + params.limit).min(results.len()); + results = results[start_idx..end_idx].to_vec(); + let search_time = start.elapsed(); + info!( + "Ultra-fast search completed for '{}': {} results in {:.2}ms (ultra-low memory)", + params.query, + results.len(), + search_time.as_millis() + ); + return Ok(results); } - - // Sort by relevance score (stable sort to maintain order for equal scores) - results.sort_by(|a, b| { - b.score - .partial_cmp(&a.score) - .unwrap_or(std::cmp::Ordering::Equal) - }); - - // Apply offset and limit - let start_idx = params.offset.min(results.len()); - let end_idx = (params.offset + params.limit).min(results.len()); - results = results[start_idx..end_idx].to_vec(); + let start_idx = params.offset.min(shared_results.len()); + let end_idx = (params.offset + params.limit).min(shared_results.len()); + let results = shared_results[start_idx..end_idx].to_vec(); let search_time = start.elapsed(); info!( @@ -2106,3 +2165,54 @@ pub fn force_memory_cleanup() -> Result<(), AppError> { )) } } + +#[cfg(test)] +mod perf_tests { + use super::*; + use tempfile::tempdir; + + #[test] + #[ignore] + fn perf_index_build_and_cached_search() { + let temp_dir = tempdir().unwrap(); + + for dir_idx in 0..40 { + let dir = temp_dir.path().join(format!("dir_{dir_idx:02}")); + std::fs::create_dir_all(&dir).unwrap(); + for file_idx in 0..200 { + std::fs::write( + dir.join(format!("document_{dir_idx:02}_{file_idx:03}.txt")), + b"irondrop", + ) + .unwrap(); + } + } + + let concurrent = ConcurrentUltraLowMemoryIndex::new(temp_dir.path().to_path_buf()); + + let build_start = Instant::now(); + concurrent.update_if_needed(true).unwrap(); + let build_ms = build_start.elapsed().as_millis(); + + let search_start = Instant::now(); + let first = concurrent.search_shared("document_12_", 64).unwrap(); + let first_search_us = search_start.elapsed().as_micros(); + + let cache_start = Instant::now(); + let second = concurrent.search_shared("document_12_", 64).unwrap(); + let cache_hit_us = cache_start.elapsed().as_micros(); + + let stats = concurrent.get_stats().unwrap(); + assert!(!first.is_empty()); + assert!(Arc::ptr_eq(&first, &second)); + + println!( + "PERF search_index build_ms={} first_search_us={} cache_hit_us={} entries={} memory_bytes={}", + build_ms, + first_search_us, + cache_hit_us, + stats.0, + stats.1 + ); + } +} diff --git a/src/upload.rs b/src/upload.rs index 1d1b554..fcdc820 100644 --- a/src/upload.rs +++ b/src/upload.rs @@ -462,7 +462,7 @@ impl DirectUploadHandler { RequestBody::File { path, .. } => { // If body is in file but small enough for memory processing, // read it into memory for simpler handling - return self.handle_file_based_upload(path, filename); + return self.handle_file_based_upload(path, filename, None); } }; @@ -561,7 +561,9 @@ impl DirectUploadHandler { // This shouldn't happen due to size checks, but handle gracefully return self.handle_memory_upload(body, filename); } - RequestBody::File { path, size: _ } => self.handle_file_based_upload(path, filename), + RequestBody::File { path, size } => { + self.handle_file_based_upload(path, filename, Some(*size)) + } } } @@ -570,6 +572,7 @@ impl DirectUploadHandler { &mut self, source_path: &PathBuf, filename: &str, + known_size: Option, ) -> Result { debug!( "Processing file-based upload: {} -> {}", @@ -586,6 +589,43 @@ impl DirectUploadHandler { let target_path = self.target_dir.join(&final_filename); trace!("Target path: {}", target_path.display()); + // Fast path: reuse the request-body temp file when the target lives on the same filesystem. + match fs::rename(source_path, &target_path) { + Ok(()) => { + let file_size = known_size.unwrap_or_else(|| { + fs::metadata(&target_path).map(|m| m.len()).unwrap_or_default() + }); + let mime_type = get_mime_type(&target_path).to_string(); + info!( + "Successfully uploaded file via rename fast path: {} ({} bytes) to {}", + final_filename, + file_size, + target_path.display() + ); + return Ok(UploadedFile { + original_name: filename.to_string(), + saved_name: final_filename, + saved_path: target_path, + size: file_size, + mime_type, + renamed: was_renamed, + }); + } + Err(err) => { + let is_cross_device = err.raw_os_error() == Some(18); + if !is_cross_device { + error!( + "Failed to move source file {source_path:?} to {target_path:?}: {err}" + ); + return Err(AppError::from(err)); + } + debug!( + "Rename fast path crossed filesystems for {:?}, falling back to buffered copy", + source_path + ); + } + } + // Create temporary file for atomic operation let temp_filename = format!( "{}{}_{}_{:x}_{}.tmp", @@ -997,6 +1037,7 @@ mod tests { use super::*; use std::io::Write; use tempfile::TempDir; + use crate::http::Request; fn create_test_cli(upload_dir: PathBuf) -> Cli { Cli { @@ -1132,6 +1173,42 @@ mod tests { assert!(handler.validate_file_extension("document.jpg").is_err()); } + #[test] + #[ignore] + fn perf_file_backed_upload_fast_path() { + let temp_dir = TempDir::new().unwrap(); + let mut cli = create_test_cli(temp_dir.path().to_path_buf()); + cli.allowed_extensions = Some("*".to_string()); + cli.max_upload_size = Some(256); + let mut handler = DirectUploadHandler::new(&cli).unwrap(); + + let request_temp = temp_dir.path().join("request-body.tmp"); + let file = File::create(&request_temp).unwrap(); + file.set_len(16 * 1024 * 1024).unwrap(); + + let request = Request { + method: "POST".to_string(), + path: "/perf-upload.bin".to_string(), + headers: HashMap::new(), + body: Some(RequestBody::File { + path: request_temp, + size: 16 * 1024 * 1024, + }), + }; + + let start = std::time::Instant::now(); + let response = handler.handle_upload(&request, None).unwrap(); + let elapsed_us = start.elapsed().as_micros(); + + assert_eq!(response.status_code, 200); + assert!(temp_dir.path().join("perf-upload.bin").exists()); + println!( + "PERF upload_fast_path size_bytes={} elapsed_us={}", + 16 * 1024 * 1024, + elapsed_us + ); + } + #[test] fn test_filename_extraction_from_disposition() { // Test various Content-Disposition formats From 722071124bf9e2452b35c16de732eb4579e50b2d Mon Sep 17 00:00:00 2001 From: Harshit Jain Date: Mon, 22 Jun 2026 21:57:59 +0530 Subject: [PATCH 2/3] irondrop: shard rate limiter locks Replace the single global rate limiter mutex with 64 IP-hashed shards to reduce request-path lock contention. Keep cleanup and memory accounting behavior across shards and add a release-mode multithreaded perf probe for unique-IP traffic. --- src/server.rs | 169 ++++++++++++++++++++++++++++++++++++-------------- 1 file changed, 123 insertions(+), 46 deletions(-) diff --git a/src/server.rs b/src/server.rs index 5b68e3d..470ef24 100644 --- a/src/server.rs +++ b/src/server.rs @@ -9,6 +9,7 @@ use crate::router::Router; use glob::Pattern; use log::{debug, info, trace, warn}; use std::collections::HashMap; +use std::hash::{Hash, Hasher}; use std::net::{IpAddr, SocketAddr}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex, mpsc}; @@ -26,7 +27,7 @@ use std::mem; /// Rate limiter for basic DoS protection #[derive(Clone)] pub struct RateLimiter { - connections: Arc>>, + connections: Arc>>>, max_requests_per_minute: u32, max_concurrent_per_ip: u32, max_connections_per_ip: u32, @@ -41,26 +42,41 @@ struct ConnectionInfo { total_connections: u32, } +const RATE_LIMITER_SHARDS: usize = 64; +const MAX_RATE_LIMITER_ENTRIES: usize = 100_000; +const MAX_RATE_LIMITER_ENTRIES_PER_SHARD: usize = + MAX_RATE_LIMITER_ENTRIES.div_ceil(RATE_LIMITER_SHARDS); + impl RateLimiter { pub fn new(max_requests_per_minute: u32, max_concurrent_per_ip: u32) -> Self { + let mut shards = Vec::with_capacity(RATE_LIMITER_SHARDS); + for _ in 0..RATE_LIMITER_SHARDS { + shards.push(Mutex::new(HashMap::new())); + } Self { - connections: Arc::new(Mutex::new(HashMap::new())), + connections: Arc::new(shards), max_requests_per_minute, max_concurrent_per_ip, max_connections_per_ip: 1000, // Limit stored connections per IP } } + fn shard_index(ip: IpAddr) -> usize { + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + ip.hash(&mut hasher); + (hasher.finish() as usize) % RATE_LIMITER_SHARDS + } + pub fn check_rate_limit(&self, ip: IpAddr) -> bool { trace!("Checking rate limit for IP: {}", ip); - let mut connections = self.connections.lock().unwrap(); + let shard_index = Self::shard_index(ip); + let mut connections = self.connections[shard_index].lock().unwrap(); let now = Instant::now(); - // Hard limit on the number of entries to prevent unbounded memory growth - const MAX_RATE_LIMITER_ENTRIES: usize = 100_000; - // Check if we need to evict entries before inserting a new one - if !connections.contains_key(&ip) && connections.len() >= MAX_RATE_LIMITER_ENTRIES { + if !connections.contains_key(&ip) + && connections.len() >= MAX_RATE_LIMITER_ENTRIES_PER_SHARD + { const EVICTION_SAMPLE: usize = 64; let mut best_ip: Option = None; let mut best_last_activity: Option = None; @@ -142,7 +158,8 @@ impl RateLimiter { pub fn release_connection(&self, ip: IpAddr) { trace!("Releasing connection for IP: {}", ip); - if let Ok(mut connections) = self.connections.lock() { + let shard_index = Self::shard_index(ip); + if let Ok(mut connections) = self.connections[shard_index].lock() { if let Some(conn_info) = connections.get_mut(&ip) { let old_count = conn_info.active_connections; conn_info.active_connections = conn_info.active_connections.saturating_sub(1); @@ -159,55 +176,64 @@ impl RateLimiter { pub fn cleanup_old_entries(&self) { trace!("Starting manual cleanup of old rate limiter entries"); - let mut connections = self.connections.lock().unwrap(); let now = Instant::now(); - let initial_count = connections.len(); + let mut cleaned_count = 0usize; - trace!("Rate limiter has {} entries before cleanup", initial_count); + for shard in self.connections.iter() { + let mut connections = shard.lock().unwrap(); + let initial_count = connections.len(); - // Reduced retention time from 5 minutes to 2 minutes - connections - .retain(|_, info| now.duration_since(info.last_activity) < Duration::from_secs(120)); + trace!("Rate limiter shard has {} entries before cleanup", initial_count); - let cleaned_count = initial_count - connections.len(); - if cleaned_count > 0 { - debug!("Cleaned up {} old rate limiter entries", cleaned_count); - // Reduce underlying capacity if we removed many entries - if connections.capacity() > connections.len() * 2 { + // Reduced retention time from 5 minutes to 2 minutes + connections + .retain(|_, info| now.duration_since(info.last_activity) < Duration::from_secs(120)); + + cleaned_count += initial_count - connections.len(); + if initial_count > connections.len() && connections.capacity() > connections.len() * 2 { connections.shrink_to_fit(); } - } else { + } + + if cleaned_count == 0 { trace!("No old entries to clean up"); + } else { + debug!("Cleaned up {} old rate limiter entries", cleaned_count); } } /// Perform aggressive cleanup when memory pressure is detected pub fn cleanup_on_memory_pressure(&self) { debug!("Starting aggressive cleanup due to memory pressure"); - let mut connections = self.connections.lock().unwrap(); let now = Instant::now(); - let initial_count = connections.len(); + let mut cleaned_count = 0usize; - trace!( - "Memory pressure cleanup: checking {} entries", - initial_count - ); + for shard in self.connections.iter() { + let mut connections = shard.lock().unwrap(); + let initial_count = connections.len(); - // More aggressive cleanup - remove entries older than 30 seconds - connections.retain(|_, info| { - info.active_connections > 0 - || now.duration_since(info.last_activity) < Duration::from_secs(30) - }); + trace!( + "Memory pressure cleanup: checking {} entries in shard", + initial_count + ); + + // More aggressive cleanup - remove entries older than 30 seconds + connections.retain(|_, info| { + info.active_connections > 0 + || now.duration_since(info.last_activity) < Duration::from_secs(30) + }); + + cleaned_count += initial_count - connections.len(); + if initial_count > connections.len() && connections.capacity() > connections.len() * 2 { + connections.shrink_to_fit(); + } + } - let cleaned_count = initial_count - connections.len(); if cleaned_count > 0 { warn!( "Memory pressure cleanup removed {} rate limiter entries", cleaned_count ); - if connections.capacity() > connections.len() * 2 { - connections.shrink_to_fit(); - } } else { trace!("Memory pressure cleanup: no entries removed"); } @@ -215,18 +241,30 @@ impl RateLimiter { /// Get rate limiter memory statistics pub fn get_memory_stats(&self) -> (usize, usize) { - if let Ok(connections) = self.connections.lock() { - let entry_count = connections.len(); - let estimated_memory = entry_count * std::mem::size_of::<(IpAddr, ConnectionInfo)>(); - trace!( - "Rate limiter stats: {} entries, ~{} bytes", - entry_count, estimated_memory - ); - (entry_count, estimated_memory) - } else { - trace!("Failed to acquire rate limiter lock for stats"); - (0, 0) + let mut entry_count = 0usize; + let mut estimated_memory = 0usize; + + for shard in self.connections.iter() { + match shard.lock() { + Ok(connections) => { + entry_count += connections.len(); + estimated_memory += + connections.len() * std::mem::size_of::<(IpAddr, ConnectionInfo)>(); + } + Err(_) => { + trace!("Failed to acquire rate limiter shard lock for stats"); + return (0, 0); + } + } } + + trace!( + "Rate limiter stats: {} entries, ~{} bytes across {} shards", + entry_count, + estimated_memory, + RATE_LIMITER_SHARDS + ); + (entry_count, estimated_memory) } } @@ -800,6 +838,8 @@ fn get_process_memory_bytes() -> Option { #[cfg(test)] mod tests { use super::*; + use std::net::Ipv4Addr; + use std::thread; #[test] fn test_upload_statistics_tracking() { @@ -954,6 +994,43 @@ mod tests { assert!(peak2.is_none()); } } + + #[test] + #[ignore] + fn perf_rate_limiter_sharded_unique_ips() { + let limiter = RateLimiter::new(10_000, 32); + let workers = 8u32; + let ips_per_worker = 5_000u32; + let start = Instant::now(); + + thread::scope(|scope| { + for worker in 0..workers { + let limiter = limiter.clone(); + scope.spawn(move || { + for i in 1..=ips_per_worker { + let ordinal = worker * ips_per_worker + (i - 1); + let second = ((ordinal >> 16) & 0xff) as u8; + let third = ((ordinal >> 8) & 0xff) as u8; + let fourth = (ordinal & 0xff) as u8; + let ip = IpAddr::V4(Ipv4Addr::new(10, second, third, fourth)); + assert!(limiter.check_rate_limit(ip)); + } + }); + } + }); + + let elapsed_ms = start.elapsed().as_millis(); + let (entries, memory_bytes) = limiter.get_memory_stats(); + println!( + "PERF rate_limiter_sharded workers={} total_ops={} elapsed_ms={} entries={} memory_bytes={}", + workers, + workers as u64 * ips_per_worker as u64, + elapsed_ms, + entries, + memory_bytes + ); + assert!(entries > 0); + } } /// Upload statistics structure for reporting From 26f01e8f9af99c2532ec7a8d7479e4a26eaa4291 Mon Sep 17 00:00:00 2001 From: Harshit Jain Date: Mon, 22 Jun 2026 22:02:06 +0530 Subject: [PATCH 3/3] irondrop: show upload speed in Mbps Add a live upload summary stat that displays aggregate current upload throughput in megabits per second on the upload UI. Include the resulting cargo fmt formatting updates in the touched Rust modules and upload page assets. --- src/fs.rs | 5 ++- src/http.rs | 4 +- src/search.rs | 45 +++++++++++----------- src/server.rs | 17 +++++---- src/upload.rs | 10 ++--- templates/upload/content.html | 6 ++- templates/upload/script.js | 70 +++++++++++++++++++++++++++++++---- 7 files changed, 109 insertions(+), 48 deletions(-) diff --git a/src/fs.rs b/src/fs.rs index 0633d6e..90c4b3e 100644 --- a/src/fs.rs +++ b/src/fs.rs @@ -88,7 +88,10 @@ pub fn generate_directory_listing( request_path }; - debug!("Preparing {} selected entries for template rendering", entries.len()); + debug!( + "Preparing {} selected entries for template rendering", + entries.len() + ); let total_pages = total_count.div_ceil(limit); let safe_page = page.max(1).min(total_pages.max(1)); let offset = (safe_page - 1) * limit; diff --git a/src/http.rs b/src/http.rs index 9645b79..4931e5a 100644 --- a/src/http.rs +++ b/src/http.rs @@ -1031,7 +1031,9 @@ mod perf_tests { drop(writer); let start = Instant::now(); - let body = read_chunked_body_async(&mut reader, Vec::new()).await.unwrap(); + let body = read_chunked_body_async(&mut reader, Vec::new()) + .await + .unwrap(); let elapsed_ms = start.elapsed().as_millis(); match body { diff --git a/src/search.rs b/src/search.rs index c847c1f..6d0c6b3 100644 --- a/src/search.rs +++ b/src/search.rs @@ -607,15 +607,12 @@ fn contains_case_insensitive_ascii(haystack: &str, needle_lower: &str) -> bool { if needle.is_empty() { return true; } - haystack - .as_bytes() - .windows(needle.len()) - .any(|window| { - window - .iter() - .zip(needle.iter()) - .all(|(&lhs, &rhs)| lhs.to_ascii_lowercase() == rhs) - }) + haystack.as_bytes().windows(needle.len()).any(|window| { + window + .iter() + .zip(needle.iter()) + .all(|(&lhs, &rhs)| lhs.to_ascii_lowercase() == rhs) + }) } impl RadixBucket { @@ -898,7 +895,9 @@ impl UltraLowMemoryIndex { // Process remaining entries if !batch_entries.is_empty() { - subdirs.extend(self.process_entry_batch_hierarchical(&mut batch_entries, parent_entry_id)?); + subdirs.extend( + self.process_entry_batch_hierarchical(&mut batch_entries, parent_entry_id)?, + ); } // Recursively process subdirectories with the entry IDs generated during batching. @@ -924,14 +923,13 @@ impl UltraLowMemoryIndex { let name_offset = self.add_string(&entry.name); // Create ultra-compact entry with parent reference - let ultra_compact_entry = - UltraCompactEntry::new( - name_offset, - parent_entry_id, - entry.size, - entry.modified, - entry.is_dir, - ); + let ultra_compact_entry = UltraCompactEntry::new( + name_offset, + parent_entry_id, + entry.size, + entry.modified, + entry.is_dir, + ); self.entries.push(ultra_compact_entry); @@ -1726,7 +1724,10 @@ pub fn perform_search( expanded_limit ); let shared_results = concurrent_index.search_shared(¶ms.query, expanded_limit)?; - debug!("Index search returned {} initial results", shared_results.len()); + debug!( + "Index search returned {} initial results", + shared_results.len() + ); // If index search returns no results, fall back to filesystem search if shared_results.is_empty() { @@ -2208,11 +2209,7 @@ mod perf_tests { println!( "PERF search_index build_ms={} first_search_us={} cache_hit_us={} entries={} memory_bytes={}", - build_ms, - first_search_us, - cache_hit_us, - stats.0, - stats.1 + build_ms, first_search_us, cache_hit_us, stats.0, stats.1 ); } } diff --git a/src/server.rs b/src/server.rs index 470ef24..a832fcd 100644 --- a/src/server.rs +++ b/src/server.rs @@ -74,8 +74,7 @@ impl RateLimiter { let now = Instant::now(); // Check if we need to evict entries before inserting a new one - if !connections.contains_key(&ip) - && connections.len() >= MAX_RATE_LIMITER_ENTRIES_PER_SHARD + if !connections.contains_key(&ip) && connections.len() >= MAX_RATE_LIMITER_ENTRIES_PER_SHARD { const EVICTION_SAMPLE: usize = 64; let mut best_ip: Option = None; @@ -183,11 +182,15 @@ impl RateLimiter { let mut connections = shard.lock().unwrap(); let initial_count = connections.len(); - trace!("Rate limiter shard has {} entries before cleanup", initial_count); + trace!( + "Rate limiter shard has {} entries before cleanup", + initial_count + ); // Reduced retention time from 5 minutes to 2 minutes - connections - .retain(|_, info| now.duration_since(info.last_activity) < Duration::from_secs(120)); + connections.retain(|_, info| { + now.duration_since(info.last_activity) < Duration::from_secs(120) + }); cleaned_count += initial_count - connections.len(); if initial_count > connections.len() && connections.capacity() > connections.len() * 2 { @@ -260,9 +263,7 @@ impl RateLimiter { trace!( "Rate limiter stats: {} entries, ~{} bytes across {} shards", - entry_count, - estimated_memory, - RATE_LIMITER_SHARDS + entry_count, estimated_memory, RATE_LIMITER_SHARDS ); (entry_count, estimated_memory) } diff --git a/src/upload.rs b/src/upload.rs index fcdc820..cedd7bd 100644 --- a/src/upload.rs +++ b/src/upload.rs @@ -593,7 +593,9 @@ impl DirectUploadHandler { match fs::rename(source_path, &target_path) { Ok(()) => { let file_size = known_size.unwrap_or_else(|| { - fs::metadata(&target_path).map(|m| m.len()).unwrap_or_default() + fs::metadata(&target_path) + .map(|m| m.len()) + .unwrap_or_default() }); let mime_type = get_mime_type(&target_path).to_string(); info!( @@ -614,9 +616,7 @@ impl DirectUploadHandler { Err(err) => { let is_cross_device = err.raw_os_error() == Some(18); if !is_cross_device { - error!( - "Failed to move source file {source_path:?} to {target_path:?}: {err}" - ); + error!("Failed to move source file {source_path:?} to {target_path:?}: {err}"); return Err(AppError::from(err)); } debug!( @@ -1035,9 +1035,9 @@ fn format_bytes(bytes: u64) -> String { #[cfg(test)] mod tests { use super::*; + use crate::http::Request; use std::io::Write; use tempfile::TempDir; - use crate::http::Request; fn create_test_cli(upload_dir: PathBuf) -> Cli { Cli { diff --git a/templates/upload/content.html b/templates/upload/content.html index d8951c9..4d13bde 100644 --- a/templates/upload/content.html +++ b/templates/upload/content.html @@ -52,6 +52,10 @@

Upload Queue

0 uploaded +
+ 0.00 Mbps + current speed +
@@ -82,4 +86,4 @@

Upload Queue

a.setAttribute('href', p); } catch (e) { } })(); - \ No newline at end of file + diff --git a/templates/upload/script.js b/templates/upload/script.js index cba1bae..0ae90d1 100644 --- a/templates/upload/script.js +++ b/templates/upload/script.js @@ -12,6 +12,8 @@ class UploadManager { this.totalBytes = 0; this.uploadedBytes = 0; this.basePath = window.__BASE_PATH || ''; + this.speedSamples = []; + this.speedWindowMs = 2000; this.init(); } @@ -37,6 +39,7 @@ class UploadManager { this.totalFilesEl = document.getElementById('totalFiles'); this.totalSizeEl = document.getElementById('totalSize'); this.completedFilesEl = document.getElementById('completedFiles'); + this.uploadSpeedEl = document.getElementById('uploadSpeed'); this.totalProgressEl = document.getElementById('totalProgress'); this.progressTextEl = document.getElementById('progressText'); } @@ -103,7 +106,7 @@ class UploadManager { } // Touch event handlers for mobile devices - handleTouchStart(e) { + handleTouchStart() { // Provide visual feedback on touch this.dropZone.classList.add('touch-active'); } @@ -151,7 +154,7 @@ class UploadManager { } } - validateFile(file) { + validateFile() { // No size limit - direct streaming handles any file size efficiently return { valid: true }; } @@ -322,9 +325,6 @@ class UploadManager { fileInfo.status = 'uploading'; this.updateQueueItem(fileInfo); - // Get current path from URL or default to / - const currentPath = this.getCurrentPath(); - // Create XMLHttpRequest for progress tracking const xhr = new XMLHttpRequest(); this.uploads.set(id, xhr); @@ -412,19 +412,24 @@ class UploadManager { const totalFiles = this.files.size; const completedFiles = Array.from(this.files.values()) .filter(file => file.status === 'completed').length; + const activeUploads = Array.from(this.files.values()) + .filter(file => file.status === 'uploading').length; + const currentUploadedBytes = Array.from(this.files.values()) + .reduce((sum, file) => sum + file.uploadedBytes, 0); // Calculate total progress let totalProgress = 0; if (this.totalBytes > 0) { - const currentUploadedBytes = Array.from(this.files.values()) - .reduce((sum, file) => sum + file.uploadedBytes, 0); totalProgress = (currentUploadedBytes / this.totalBytes) * 100; } + const speedMbps = this.updateSpeedEstimate(currentUploadedBytes, activeUploads > 0); + // Update DOM this.totalFilesEl.textContent = totalFiles; this.totalSizeEl.textContent = this.formatBytes(this.totalBytes); this.completedFilesEl.textContent = completedFiles; + this.uploadSpeedEl.textContent = this.formatMbps(speedMbps); this.totalProgressEl.style.width = `${totalProgress}%`; this.progressTextEl.textContent = `${Math.round(totalProgress)}% complete`; @@ -484,6 +489,55 @@ class UploadManager { return parseFloat((bytes / Math.pow(k, i)).toFixed(1)) + ' ' + sizes[i]; } + updateSpeedEstimate(currentUploadedBytes, hasActiveUploads) { + if (!hasActiveUploads) { + this.speedSamples = []; + return 0; + } + + const now = performance.now(); + const lastSample = this.speedSamples[this.speedSamples.length - 1]; + if (!lastSample || currentUploadedBytes < lastSample.bytes) { + this.speedSamples = [{ time: now, bytes: currentUploadedBytes }]; + return 0; + } + + if (currentUploadedBytes !== lastSample.bytes) { + this.speedSamples.push({ time: now, bytes: currentUploadedBytes }); + } else if (now - lastSample.time > this.speedWindowMs) { + this.speedSamples = [{ time: now, bytes: currentUploadedBytes }]; + return 0; + } + + const cutoff = now - this.speedWindowMs; + while (this.speedSamples.length > 1 && this.speedSamples[1].time < cutoff) { + this.speedSamples.shift(); + } + + if (this.speedSamples.length < 2) { + return 0; + } + + const firstSample = this.speedSamples[0]; + const bytesDelta = currentUploadedBytes - firstSample.bytes; + const timeDeltaMs = now - firstSample.time; + if (bytesDelta <= 0 || timeDeltaMs <= 0) { + return 0; + } + + return (bytesDelta * 8 * 1000) / (timeDeltaMs * 1000 * 1000); + } + + formatMbps(mbps) { + if (mbps >= 100) { + return `${mbps.toFixed(0)} Mbps`; + } + if (mbps >= 10) { + return `${mbps.toFixed(1)} Mbps`; + } + return `${mbps.toFixed(2)} Mbps`; + } + escapeHtml(text) { const div = document.createElement('div'); div.textContent = text; @@ -571,4 +625,4 @@ document.addEventListener('DOMContentLoaded', () => { // Initialize inline upload form new InlineUploadForm(); -}); \ No newline at end of file +});