diff --git a/src/fs.rs b/src/fs.rs index 4bd30d3..90c4b3e 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,10 @@ 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 +99,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 +192,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..4931e5a 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,45 @@ 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..6d0c6b3 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,23 @@ 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 +877,32 @@ 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 +911,25 @@ 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); + let ultra_compact_entry = UltraCompactEntry::new( + name_offset, + parent_entry_id, + entry.size, + entry.modified, + entry.is_dir, + ); self.entries.push(ultra_compact_entry); @@ -905,34 +943,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 +989,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 +1039,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 +1075,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 +1092,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 +1119,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 +1145,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 +1714,47 @@ 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 shared_results = concurrent_index.search_shared(¶ms.query, expanded_limit)?; + debug!( + "Index search returned {} initial results", + shared_results.len() ); - let mut results = concurrent_index.search(¶ms.query, params.limit * 2)?; - debug!("Index search returned {} initial results", 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 +2166,50 @@ 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/server.rs b/src/server.rs index 5b68e3d..a832fcd 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,40 @@ 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 +157,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 +175,68 @@ 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 +244,28 @@ 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 +839,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 +995,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 diff --git a/src/upload.rs b/src/upload.rs index 1d1b554..cedd7bd 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", @@ -995,6 +1035,7 @@ fn format_bytes(bytes: u64) -> String { #[cfg(test)] mod tests { use super::*; + use crate::http::Request; use std::io::Write; use tempfile::TempDir; @@ -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 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 +});