1use memmap2::{Mmap, MmapOptions};
2use rayon::prelude::*;
3use std::cmp::min;
4use std::fs::File;
5use std::io::{BufReader, Read, Seek, SeekFrom};
6use std::path::{Path, PathBuf};
7use std::sync::Arc;
8use tracing::{debug, info, trace, warn};
9
10use crate::errors::{Result, WinxError};
11
12pub const DIRECT_READ_THRESHOLD: u64 = 10_000_000;
14
15pub const MAX_MMAP_SIZE: u64 = 1_000_000_000;
17
18pub const MAX_SEGMENTED_MMAP_SIZE: u64 = 4_000_000_000;
20
21pub const SEGMENT_SIZE: u64 = 256_000_000;
23
24const DIRECT_READ_CHUNK_SIZE: usize = 1_048_576;
25const MMAP_PARALLEL_CHUNK_SIZE: usize = 1_048_576;
26const STREAMING_CHUNK_SIZE: usize = 4_194_304;
27
28pub fn read_file_optimized(path: &Path, max_file_size: u64) -> Result<Vec<u8>> {
49 let file = File::open(path).map_err(|e| WinxError::FileAccessError {
51 path: path.to_path_buf(),
52 message: format!("Error opening file: {e}"),
53 })?;
54
55 let metadata = file.metadata().map_err(|e| WinxError::FileAccessError {
56 path: path.to_path_buf(),
57 message: format!("Failed to get file metadata: {e}"),
58 })?;
59
60 let file_size = metadata.len();
62 if file_size > max_file_size {
63 return Err(WinxError::FileTooLarge {
64 path: path.to_path_buf(),
65 size: file_size,
66 max_size: max_file_size,
67 });
68 }
69
70 if file_size < DIRECT_READ_THRESHOLD {
72 debug!("Using direct read for file: {}", path.display());
73 read_direct(&file, file_size, path)
74 } else if file_size < MAX_MMAP_SIZE {
75 debug!("Using memory-mapped read for file: {}", path.display());
76 read_mmap(&file, path)
77 } else if file_size < MAX_SEGMENTED_MMAP_SIZE {
78 debug!("Using segmented memory-mapped read for file: {}", path.display());
79 read_segmented_mmap(&file, file_size, path)
80 } else {
81 debug!("Using streaming read for extremely large file: {}", path.display());
82 read_streaming(&file, file_size, path)
83 }
84}
85
86fn read_direct(file: &File, file_size: u64, path: &Path) -> Result<Vec<u8>> {
104 if file_size < 1_000_000 {
106 let mut buffer = Vec::with_capacity(file_size as usize);
108
109 let mut file_handle = file.try_clone().map_err(|e| WinxError::FileAccessError {
111 path: path.to_path_buf(),
112 message: format!("Error cloning file handle: {e}"),
113 })?;
114
115 file_handle.seek(SeekFrom::Start(0)).map_err(|e| WinxError::FileAccessError {
116 path: path.to_path_buf(),
117 message: format!("Error seeking to start of file: {e}"),
118 })?;
119
120 let mut reader = BufReader::with_capacity(min(file_size as usize, 64 * 1024), file_handle);
122
123 reader.read_to_end(&mut buffer).map_err(|e| WinxError::FileAccessError {
125 path: path.to_path_buf(),
126 message: format!("Error reading file: {e}"),
127 })?;
128
129 return Ok(buffer);
130 }
131
132 let mut buffer = Vec::with_capacity(file_size as usize);
134
135 let mut file_handle = file.try_clone().map_err(|e| WinxError::FileAccessError {
137 path: path.to_path_buf(),
138 message: format!("Error cloning file handle: {e}"),
139 })?;
140
141 file_handle.seek(SeekFrom::Start(0)).map_err(|e| WinxError::FileAccessError {
142 path: path.to_path_buf(),
143 message: format!("Error seeking to start of file: {e}"),
144 })?;
145
146 let mut reader = BufReader::with_capacity(262_144, file_handle); let mut chunk = vec![0; DIRECT_READ_CHUNK_SIZE];
149 let mut bytes_read = 0;
150
151 loop {
152 match reader.read(&mut chunk) {
153 Ok(0) => break, Ok(n) => {
155 buffer.extend_from_slice(&chunk[..n]);
156 bytes_read += n as u64;
157
158 if file_size > 5_000_000 && bytes_read % 5_000_000 < DIRECT_READ_CHUNK_SIZE as u64 {
160 trace!(
161 "Read progress for {}: {}MB/{}MB ({}%)",
162 path.display(),
163 bytes_read / 1_000_000,
164 file_size / 1_000_000,
165 bytes_read * 100 / file_size
166 );
167 }
168 }
169 Err(e) => {
170 return Err(WinxError::FileAccessError {
171 path: path.to_path_buf(),
172 message: format!("Error reading file chunk: {e}"),
173 });
174 }
175 }
176 }
177
178 Ok(buffer)
179}
180
181fn read_mmap(file: &File, path: &Path) -> Result<Vec<u8>> {
199 if file.metadata().map_or(0, |m| m.len()) == 0 {
201 return Ok(Vec::new());
202 }
203
204 let mmap = unsafe { MmapOptions::new().map(file) }.map_err(|e| WinxError::FileAccessError {
212 path: path.to_path_buf(),
213 message: format!("Failed to memory-map file: {e}"),
214 })?;
215
216 if mmap.len() > 10_000_000 {
218 debug!("Using parallel processing for large mmap file: {}", path.display());
220
221 let chunk_count = mmap.len().div_ceil(MMAP_PARALLEL_CHUNK_SIZE);
223 let mut result = vec![0; mmap.len()];
224
225 let chunks: Vec<_> = (0..chunk_count)
227 .into_par_iter()
228 .map(|i| {
229 let start = i * MMAP_PARALLEL_CHUNK_SIZE;
230 let end = min((i + 1) * MMAP_PARALLEL_CHUNK_SIZE, mmap.len());
231
232 if start < mmap.len() {
233 let src = &mmap[start..end];
235 (start, end, src.to_vec())
236 } else {
237 (start, start, Vec::new())
238 }
239 })
240 .collect();
241
242 for (start, end, chunk) in chunks {
244 if start < end {
245 result[start..end].copy_from_slice(&chunk);
246 }
247 }
248
249 Ok(result)
250 } else {
251 Ok(mmap.to_vec())
253 }
254}
255
256fn read_segmented_mmap(file: &File, file_size: u64, path: &Path) -> Result<Vec<u8>> {
275 let segment_count = file_size.div_ceil(SEGMENT_SIZE);
277 debug!(
278 "Reading file {} in {} segments of {}MB each",
279 path.display(),
280 segment_count,
281 SEGMENT_SIZE / 1_000_000
282 );
283
284 let mut result = Vec::with_capacity(file_size as usize);
286
287 for i in 0..segment_count {
289 let segment_start = i * SEGMENT_SIZE;
290 let segment_size = min(SEGMENT_SIZE, file_size - segment_start);
291
292 info!(
293 "Processing segment {}/{} of file {} ({:.1}%)",
294 i + 1,
295 segment_count,
296 path.display(),
297 (segment_start as f64 / file_size as f64) * 100.0
298 );
299
300 let segment_mmap = unsafe {
305 MmapOptions::new().offset(segment_start).len(segment_size as usize).map(file)
306 }
307 .map_err(|e| WinxError::FileAccessError {
308 path: path.to_path_buf(),
309 message: format!("Failed to memory-map file segment {i}: {e}"),
310 })?;
311
312 result.extend_from_slice(&segment_mmap);
314 }
315
316 Ok(result)
317}
318
319fn read_streaming(file: &File, file_size: u64, path: &Path) -> Result<Vec<u8>> {
338 warn!(
339 "Reading extremely large file ({}GB) with streaming approach: {}",
340 file_size / 1_000_000_000,
341 path.display()
342 );
343
344 let initial_capacity = min(file_size as usize, 1_000_000_000); let mut buffer = Vec::with_capacity(initial_capacity);
347
348 let mut reader = BufReader::with_capacity(STREAMING_CHUNK_SIZE, file);
349 let mut chunk = vec![0; STREAMING_CHUNK_SIZE];
350 let mut bytes_read = 0;
351
352 loop {
353 match reader.read(&mut chunk) {
354 Ok(0) => break, Ok(n) => {
356 buffer.extend_from_slice(&chunk[..n]);
357 bytes_read += n as u64;
358
359 if bytes_read % 100_000_000 < STREAMING_CHUNK_SIZE as u64 {
361 info!(
362 "Read progress for large file {}: {:.2}GB/{:.2}GB ({:.1}%)",
363 path.display(),
364 bytes_read as f64 / 1_000_000_000.0,
365 file_size as f64 / 1_000_000_000.0,
366 bytes_read as f64 * 100.0 / file_size as f64
367 );
368 }
369 }
370 Err(e) => {
371 return Err(WinxError::FileAccessError {
372 path: path.to_path_buf(),
373 message: format!("Error reading file chunk at position {bytes_read}: {e}"),
374 });
375 }
376 }
377 }
378
379 Ok(buffer)
380}
381
382pub fn read_file_segment(
402 path: &Path,
403 offset: u64,
404 length: u64,
405 max_file_size: u64,
406) -> Result<Vec<u8>> {
407 let file = File::open(path).map_err(|e| WinxError::FileAccessError {
409 path: path.to_path_buf(),
410 message: format!("Error opening file: {e}"),
411 })?;
412
413 let metadata = file.metadata().map_err(|e| WinxError::FileAccessError {
414 path: path.to_path_buf(),
415 message: format!("Failed to get file metadata: {e}"),
416 })?;
417
418 let file_size = metadata.len();
420 if file_size > max_file_size {
421 return Err(WinxError::FileTooLarge {
422 path: path.to_path_buf(),
423 size: file_size,
424 max_size: max_file_size,
425 });
426 }
427
428 if offset >= file_size {
430 return Err(WinxError::FileAccessError {
431 path: path.to_path_buf(),
432 message: format!("Offset {offset} exceeds file size {file_size}"),
433 });
434 }
435
436 let length = min(length, file_size - offset);
438
439 if length < DIRECT_READ_THRESHOLD {
441 debug!("Using direct read for file segment: {}", path.display());
442 read_segment_direct(&file, offset, length, path)
443 } else {
444 debug!("Using memory-mapped read for file segment: {}", path.display());
445 read_segment_mmap(&file, offset, length, path)
446 }
447}
448
449fn read_segment_direct(file: &File, offset: u64, length: u64, path: &Path) -> Result<Vec<u8>> {
466 let mut seekable_file = file.try_clone().map_err(|e| WinxError::FileAccessError {
468 path: path.to_path_buf(),
469 message: format!("Failed to clone file handle: {e}"),
470 })?;
471
472 seekable_file.seek(SeekFrom::Start(offset)).map_err(|e| WinxError::FileAccessError {
474 path: path.to_path_buf(),
475 message: format!("Failed to seek to offset {offset}: {e}"),
476 })?;
477
478 let mut buffer = Vec::with_capacity(length as usize);
480 let reader = BufReader::with_capacity(min(length as usize, 64 * 1024), seekable_file);
481
482 reader.take(length).read_to_end(&mut buffer).map_err(|e| WinxError::FileAccessError {
484 path: path.to_path_buf(),
485 message: format!("Error reading file segment: {e}"),
486 })?;
487
488 Ok(buffer)
489}
490
491fn read_segment_mmap(file: &File, offset: u64, length: u64, path: &Path) -> Result<Vec<u8>> {
508 let segment_mmap = unsafe { MmapOptions::new().offset(offset).len(length as usize).map(file) }
513 .map_err(|e| WinxError::FileAccessError {
514 path: path.to_path_buf(),
515 message: format!("Failed to memory-map file segment: {e}"),
516 })?;
517
518 Ok(segment_mmap.to_vec())
520}
521
522pub fn read_file_to_string(path: &Path, max_file_size: u64) -> Result<String> {
540 let bytes = read_file_optimized(path, max_file_size)?;
541
542 String::from_utf8(bytes).map_err(|e| WinxError::FileAccessError {
543 path: path.to_path_buf(),
544 message: format!("Failed to decode file as UTF-8: {e}"),
545 })
546}
547
548pub fn process_text_file_parallel<F>(
567 path: &Path,
568 max_file_size: u64,
569 line_processor: F,
570) -> Result<()>
571where
572 F: Fn(&str) + Sync,
573{
574 let content = read_file_to_string(path, max_file_size)?;
575
576 if content.len() > 1_000_000 {
578 content.par_lines().for_each(|line| {
580 line_processor(line);
581 });
582 } else {
583 content.lines().for_each(|line| {
585 line_processor(line);
586 });
587 }
588
589 Ok(())
590}
591
592pub fn read_file_segment_to_string(
609 path: &Path,
610 offset: u64,
611 length: u64,
612 max_file_size: u64,
613) -> Result<String> {
614 let bytes = read_file_segment(path, offset, length, max_file_size)?;
615
616 String::from_utf8(bytes).map_err(|e| WinxError::FileAccessError {
617 path: path.to_path_buf(),
618 message: format!("Failed to decode file segment as UTF-8: {e}"),
619 })
620}
621
622#[derive(Clone)]
627pub struct ShareableMap {
628 data: Arc<Mmap>,
630 path: PathBuf,
632}
633
634impl ShareableMap {
635 pub fn new(path: &Path) -> Result<Self> {
649 let file = File::open(path).map_err(|e| WinxError::FileAccessError {
650 path: path.to_path_buf(),
651 message: format!("Error opening file: {e}"),
652 })?;
653
654 if file
656 .metadata()
657 .map_err(|e| WinxError::FileAccessError {
658 path: path.to_path_buf(),
659 message: format!("Failed to get metadata: {e}"),
660 })?
661 .len()
662 == 0
663 {
664 return Err(WinxError::FileAccessError {
665 path: path.to_path_buf(),
666 message: "Cannot memory map empty file".to_string(),
667 });
668 }
669
670 let mmap =
675 unsafe { MmapOptions::new().map(&file) }.map_err(|e| WinxError::FileAccessError {
676 path: path.to_path_buf(),
677 message: format!("Failed to memory-map file: {e}"),
678 })?;
679
680 Ok(Self { data: Arc::new(mmap), path: path.to_path_buf() })
681 }
682
683 pub fn new_segment(path: &Path, offset: u64, length: u64) -> Result<Self> {
699 if length == 0 {
700 return Err(WinxError::FileAccessError {
701 path: path.to_path_buf(),
702 message: "Cannot memory map segment of length 0".to_string(),
703 });
704 }
705
706 let file = File::open(path).map_err(|e| WinxError::FileAccessError {
707 path: path.to_path_buf(),
708 message: format!("Error opening file: {e}"),
709 })?;
710
711 let mmap = unsafe { MmapOptions::new().offset(offset).len(length as usize).map(&file) }
715 .map_err(|e| WinxError::FileAccessError {
716 path: path.to_path_buf(),
717 message: format!("Failed to memory-map file segment: {e}"),
718 })?;
719
720 Ok(Self { data: Arc::new(mmap), path: path.to_path_buf() })
721 }
722
723 pub fn as_slice(&self) -> &[u8] {
725 &self.data
726 }
727
728 pub fn path(&self) -> &Path {
730 &self.path
731 }
732
733 pub fn len(&self) -> usize {
735 self.data.len()
736 }
737
738 pub fn is_empty(&self) -> bool {
740 self.data.is_empty()
741 }
742}
743
744#[cfg(test)]
745mod tests {
746 use super::*;
747 use std::io::Write;
748 use tempfile::NamedTempFile;
749
750 fn create_test_file(size: usize) -> Result<(NamedTempFile, Vec<u8>)> {
751 let mut file = NamedTempFile::new()?;
752 let mut data = Vec::with_capacity(size);
753
754 for i in 0..size {
756 data.push((i % 256) as u8);
757 }
758
759 file.write_all(&data)?;
760 file.flush()?;
761
762 Ok((file, data))
763 }
764
765 #[test]
766 fn test_direct_read_small_file() -> Result<()> {
767 let size = 10 * 1024; let (file, expected_data) = create_test_file(size)?;
769
770 let result = read_direct(file.as_file(), size as u64, file.path())?;
771 assert_eq!(result, expected_data);
772 Ok(())
773 }
774
775 #[test]
776 fn test_mmap_read() -> Result<()> {
777 let size = 1024 * 1024; let (file, expected_data) = create_test_file(size)?;
779
780 let result = read_mmap(file.as_file(), file.path())?;
781 assert_eq!(result, expected_data);
782 Ok(())
783 }
784
785 #[test]
786 fn test_file_segment_read() -> Result<()> {
787 let size = 1024 * 1024; let (file, data) = create_test_file(size)?;
789
790 let offset = 100 * 1024; let length = 200 * 1024; let expected_segment = &data[offset as usize..(offset + length) as usize];
794
795 let result = read_segment_direct(file.as_file(), offset, length, file.path())?;
796 assert_eq!(result, expected_segment);
797
798 let result = read_segment_mmap(file.as_file(), offset, length, file.path())?;
799 assert_eq!(result, expected_segment);
800 Ok(())
801 }
802
803 #[test]
804 fn test_shareable_map() -> Result<()> {
805 let size = 100 * 1024; let (file, data) = create_test_file(size)?;
807
808 let map = ShareableMap::new(file.path())?;
809 assert_eq!(map.as_slice(), &data);
810
811 let offset = 10 * 1024; let length = 20 * 1024; let segment_map = ShareableMap::new_segment(file.path(), offset, length)?;
815 assert_eq!(segment_map.as_slice(), &data[offset as usize..(offset + length) as usize]);
816 Ok(())
817 }
818
819 #[test]
820 fn test_parallel_processing() -> Result<()> {
821 let mut file = NamedTempFile::new()?;
823 let mut lines = Vec::new();
824
825 for i in 0..1000 {
826 let line = format!("Line {i}\n");
827 file.write_all(line.as_bytes())?;
828 lines.push(format!("Line {i}"));
829 }
830 file.flush()?;
831
832 let processed_lines = std::sync::Mutex::new(Vec::new());
834
835 process_text_file_parallel(file.path(), 1_000_000, |line| {
836 if let Ok(mut lines) = processed_lines.lock() {
837 lines.push(line.to_string());
838 }
839 })?;
840
841 let result =
843 processed_lines.lock().map_err(|error| WinxError::ResourceAllocationError {
844 message: format!("Failed to lock processed lines: {error}"),
845 })?;
846 assert_eq!(result.len(), lines.len());
847
848 for line in &lines {
850 assert!(result.contains(line));
851 }
852 Ok(())
853 }
854}