optimize SSD performance

Author: bea4dev

src/cozip/src/lib.rs

@@ -144,6 +144,7 @@ settings.gpu_slot_count = "GPU slot count"
 settings.gpu_submit_chunks = "GPU submit chunks"
 settings.gpu_min_chunk_kib = "GPU min chunk"
 settings.gpu_decode = "GPU decode"
+settings.parallel_read_threads = "Parallel read threads"
 settings.parallel_write_threads = "Parallel write threads"
 settings.force_gpu = "Force GPU"
 settings.no_pdeflate_options = "The selected archives are ZIP only, so no extra GPU decode settings are available."

src/cozip_deflate/src/lib.rs

@@ -144,6 +144,7 @@ settings.gpu_slot_count = "GPU slot 数"
 settings.gpu_submit_chunks = "GPU submit chunks"
 settings.gpu_min_chunk_kib = "GPU 最小チャンク"
 settings.gpu_decode = "GPU 解凍"
+settings.parallel_read_threads = "並列読み込みスレッド数"
 settings.parallel_write_threads = "並列書き込みスレッド数"
 settings.force_gpu = "GPU 強制"
 settings.no_pdeflate_options = "選択されたアーカイブは ZIP のみなので、追加の GPU 解凍設定はありません。"

src/cozip_desktop/locales/en_US.toml

@@ -635,6 +635,27 @@ impl CozipDesktopApp {
         match plan.format {
             ArchiveFormat::Zip => {
                 form = form
+                    .child(self.settings_row(
+                        self.t("settings.parallel_read_threads"),
+                        self.stepper_control(
+                            "zip-read-threads-dec",
+                            "zip-read-threads-inc",
+                            plan.zip_options.parallel_read_threads.to_string(),
+                            |this, _, _| {
+                                if let Some(plan) = this.compress_plan_mut() {
+                                    plan.zip_options.parallel_read_threads =
+                                        plan.zip_options.parallel_read_threads.saturating_sub(1).max(1);
+                                }
+                            },
+                            |this, _, _| {
+                                if let Some(plan) = this.compress_plan_mut() {
+                                    plan.zip_options.parallel_read_threads =
+                                        plan.zip_options.parallel_read_threads.saturating_add(1).min(128);
+                                }
+                            },
+                            cx,
+                        ),
+                    ))
                     .child(self.settings_row(
                         self.t("settings.compression_level"),
                         self.stepper_control(
@@ -680,6 +701,27 @@ impl CozipDesktopApp {
             ArchiveFormat::Cozip => {
                 let opts = &plan.pdeflate_options;
                 form = form
+                    .child(self.settings_row(
+                        self.t("settings.parallel_read_threads"),
+                        self.stepper_control(
+                            "pdeflate-read-threads-dec",
+                            "pdeflate-read-threads-inc",
+                            opts.parallel_read_threads.to_string(),
+                            |this, _, _| {
+                                if let Some(plan) = this.compress_plan_mut() {
+                                    plan.pdeflate_options.parallel_read_threads =
+                                        plan.pdeflate_options.parallel_read_threads.saturating_sub(1).max(1);
+                                }
+                            },
+                            |this, _, _| {
+                                if let Some(plan) = this.compress_plan_mut() {
+                                    plan.pdeflate_options.parallel_read_threads =
+                                        plan.pdeflate_options.parallel_read_threads.saturating_add(1).min(128);
+                                }
+                            },
+                            cx,
+                        ),
+                    ))
                     .child(self.settings_row(
                         self.t("settings.huffman"),
                         self.toggle_control(

src/cozip_desktop/locales/ja_JP.toml

@@ -17,9 +17,12 @@ fn main() {
             Some(InitialScreen::Compress) | Some(InitialScreen::Decompress) => {
                 size(px(620.0), px(300.0))
             }
-            Some(InitialScreen::CompressSettings) | Some(InitialScreen::DecompressSettings) => {
+            Some(InitialScreen::DecompressSettings) => {
                 size(px(760.0), px(520.0))
             }
+            Some(InitialScreen::CompressSettings) => {
+                size(px(760.0), px(620.0))
+            }
             None => size(px(1360.0), px(920.0)),
         };
         let bounds = Bounds::centered(None, window_size, cx);

src/cozip_desktop/src/app.rs

@@ -1,10 +1,11 @@
 use std::collections::VecDeque;
 use std::fs::{File as StdFile, OpenOptions};
-use std::io::{BufReader, BufWriter, Read, Write};
+use std::io::{self, BufReader, BufWriter, Read, Write};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Instant;
 
+use cozip_util::{ParallelFileReader, ParallelFileReaderOptions, ParallelReadHandle};
 use thiserror::Error;
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
@@ -137,6 +138,124 @@ impl<R: AsyncRead + Unpin> AsyncStream<R> {
     }
 }
 
+struct ParallelPrefetchReader {
+    reader: ParallelFileReader,
+    file_len: u64,
+    next_submit_offset: u64,
+    request_size: usize,
+    max_inflight_ops: usize,
+    max_inflight_bytes: usize,
+    inflight_bytes: usize,
+    inflight: VecDeque<(ParallelReadHandle, usize)>,
+    current: Vec<u8>,
+    current_pos: usize,
+}
+
+impl ParallelPrefetchReader {
+    fn new(
+        file: StdFile,
+        chunk_size: usize,
+        options: ParallelFileReaderOptions,
+    ) -> Result<Self, CozipDeflateError> {
+        let file_len = file.metadata()?.len();
+        let request_size = chunk_size.max(1);
+        let max_inflight_ops = if options.max_inflight_ops > 0 {
+            options.max_inflight_ops
+        } else {
+            let by_bytes = options.max_backlog_bytes.max(request_size) / request_size;
+            by_bytes.clamp(64, 4096)
+        };
+        let max_inflight_bytes = options.max_backlog_bytes.max(request_size);
+        let reader =
+            ParallelFileReader::new(file, options).map_err(|error| io::Error::other(error.to_string()))?;
+        let mut this = Self {
+            reader,
+            file_len,
+            next_submit_offset: 0,
+            request_size,
+            max_inflight_ops,
+            max_inflight_bytes,
+            inflight_bytes: 0,
+            inflight: VecDeque::new(),
+            current: Vec::new(),
+            current_pos: 0,
+        };
+        this.fill_prefetch()?;
+        Ok(this)
+    }
+
+    fn fill_prefetch(&mut self) -> io::Result<()> {
+        while self.inflight.len() < self.max_inflight_ops
+            && self.inflight_bytes < self.max_inflight_bytes
+            && self.next_submit_offset < self.file_len
+        {
+            let remaining = self.file_len.saturating_sub(self.next_submit_offset);
+            let mut len =
+                usize::try_from(remaining.min(self.request_size as u64)).unwrap_or(self.request_size);
+            let available_budget = self.max_inflight_bytes.saturating_sub(self.inflight_bytes);
+            if len > available_budget && available_budget > 0 {
+                len = available_budget.min(len);
+            }
+            if len == 0 {
+                break;
+            }
+            let handle = self
+                .reader
+                .submit(self.next_submit_offset, len)
+                .map_err(|error| io::Error::other(error.to_string()))?;
+            self.inflight.push_back((handle, len));
+            self.inflight_bytes = self.inflight_bytes.saturating_add(len);
+            self.next_submit_offset = self.next_submit_offset.saturating_add(len as u64);
+        }
+        Ok(())
+    }
+
+    fn finish(self) -> Result<(), CozipDeflateError> {
+        self.reader
+            .drain()
+            .map_err(|error| io::Error::other(error.to_string()).into())
+    }
+}
+
+impl Read for ParallelPrefetchReader {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        if buf.is_empty() {
+            return Ok(0);
+        }
+
+        let mut written = 0usize;
+        loop {
+            if self.current_pos >= self.current.len() {
+                let Some((handle, len)) = self.inflight.pop_front() else {
+                    return Ok(written);
+                };
+                self.inflight_bytes = self.inflight_bytes.saturating_sub(len);
+                self.current = handle
+                    .recv()
+                    .map_err(|error| io::Error::other(error.to_string()))?;
+                self.current_pos = 0;
+                self.fill_prefetch()?;
+                if self.current.is_empty() {
+                    if written > 0 {
+                        return Ok(written);
+                    }
+                    continue;
+                }
+            }
+
+            let available = self.current.len().saturating_sub(self.current_pos);
+            let take = available.min(buf.len().saturating_sub(written));
+            buf[written..written + take]
+                .copy_from_slice(&self.current[self.current_pos..self.current_pos + take]);
+            self.current_pos = self.current_pos.saturating_add(take);
+            written = written.saturating_add(take);
+            if written == buf.len() {
+                return Ok(written);
+            }
+        }
+    }
+}
+
 #[derive(Debug, Clone, Copy, Default)]
 pub struct CoZipDeflateInitStats {
     pub gpu_context_init_ms: f64,
@@ -200,6 +319,10 @@ impl CoZipDeflate {
         self.options.parallel_write_threads
     }
 
+    pub fn parallel_read_threads(&self) -> usize {
+        self.options.parallel_read_threads
+    }
+
     pub fn compress_stream<R: Read + Send, W: Write>(
         &self,
         reader: &mut R,
@@ -431,6 +554,22 @@ impl CoZipDeflate {
         .map_err(map_pdeflate_error)
     }
 
+    pub fn compress_file_parallel_read_with_options(
+        &self,
+        input_file: StdFile,
+        output_file: StdFile,
+        stream_options: StreamOptions,
+        reader_options: ParallelFileReaderOptions,
+    ) -> Result<PDeflateStats, CozipDeflateError> {
+        let chunk_size = self.options.chunk_size.max(1);
+        let mut reader = ParallelPrefetchReader::new(input_file, chunk_size, reader_options)?;
+        let mut writer = BufWriter::new(output_file);
+        let stats = self.compress_stream_with_options(&mut reader, &mut writer, stream_options)?;
+        writer.flush()?;
+        reader.finish()?;
+        Ok(stats)
+    }
+
     pub fn decompress_file_from_name<PIn: AsRef<Path>, POut: AsRef<Path>>(
         &self,
         input_path: PIn,

src/cozip_desktop/src/main.rs

@@ -204,6 +204,7 @@ pub struct PDeflateOptions {
     pub gpu_pipelined_submit_chunks: usize,
     pub gpu_min_chunk_size: usize,
     pub gpu_tail_stop_ratio: f32,
+    pub parallel_read_threads: usize,
     pub parallel_write_threads: usize,
     pub huffman_encode_enabled: bool,
     pub compression_mode: PDeflateCompressionMode,
@@ -212,7 +213,7 @@ pub struct PDeflateOptions {
 
 impl Default for PDeflateOptions {
     fn default() -> Self {
-        let parallel_write_threads = thread::available_parallelism()
+        let parallel_threads = thread::available_parallelism()
             .map(|value| value.get())
             .unwrap_or(1)
             .max(1);
@@ -236,7 +237,8 @@ impl Default for PDeflateOptions {
             gpu_pipelined_submit_chunks: 4,
             gpu_min_chunk_size: 64 * 1024,
             gpu_tail_stop_ratio: 1.0,
-            parallel_write_threads,
+            parallel_read_threads: parallel_threads,
+            parallel_write_threads: parallel_threads,
             huffman_encode_enabled: false,
             compression_mode: PDeflateCompressionMode::Speed,
             hybrid_scheduler_policy: PDeflateHybridSchedulerPolicy::GlobalQueue,
@@ -1739,6 +1741,12 @@ fn validate_options(options: &PDeflateOptions) -> Result<(), PDeflateError> {
             "gpu_table_sample_stride must be > 0",
         ));
     }
+    if options.parallel_read_threads == 0 {
+        return Err(PDeflateError::InvalidOptions(
+            "parallel_read_threads must be > 0",
+        ));
+    }
+
     if options.parallel_write_threads == 0 {
         return Err(PDeflateError::InvalidOptions(
             "parallel_write_threads must be > 0",