diff --git a/bz2zstd/src/main.rs b/bz2zstd/src/main.rs index c7a4ed7..5a806df 100644 --- a/bz2zstd/src/main.rs +++ b/bz2zstd/src/main.rs @@ -61,9 +61,10 @@ struct Args { #[arg(short, long)] output: Option, - /// Zstd compression level (1-22, default = 9) - /// Higher values provide better compression but are slower - #[arg(short = 'z', long, default_value_t = 9)] + /// Zstd compression level (-7 to 22, default = 9) + /// Higher values provide better compression but are slower. + /// Negative values provide faster but less efficient compression. + #[arg(short = 'z', long, default_value_t = 9, value_parser = clap::value_parser!(i32).range(-7..=22))] zstd_level: i32, /// Number of threads to use (default = number of logical cores) @@ -79,6 +80,10 @@ struct Args { fn main() -> Result<()> { let args = Args::parse(); + // Pre-validate zstd compression level to ensure it's supported + // This prevents a panic in the worker threads later + zstd::bulk::Compressor::new(args.zstd_level).context("Invalid zstd compression level")?; + // Configure global thread pool if user specified thread count // This affects all Rayon parallel iterators in the application if let Some(jobs) = args.jobs { @@ -309,7 +314,13 @@ fn main() -> Result<()> { .try_for_each_init( // Per-thread initialization: create buffers and compressor once per thread // This avoids lock contention and repeated allocations - || (Vec::new(), Compressor::new(args.zstd_level).unwrap()), + || { + ( + Vec::new(), + Compressor::new(args.zstd_level) + .expect("Compression level was pre-validated"), + ) + }, |(decomp_buf, compressor), (idx, (start_bit, end_bit))| -> Result<()> { // Security Check: Verify that the compressed block size is within limits. // This protects against resource exhaustion from maliciously large compressed blocks. diff --git a/parallel_bzip2_decoder/src/decoder.rs b/parallel_bzip2_decoder/src/decoder.rs index b60b005..3288712 100644 --- a/parallel_bzip2_decoder/src/decoder.rs +++ b/parallel_bzip2_decoder/src/decoder.rs @@ -169,8 +169,12 @@ impl Bz2Decoder { let mut decomp_buf = Vec::new(); // Decompress this block decompress_block_into(slice, start_bit, end_bit, &mut decomp_buf, scratch)?; - // Send result with index for reordering - result_sender.send((idx, decomp_buf)).unwrap(); + // Send result with index for reordering. + // We use map_err to stop the parallel worker pool gracefully + // if the receiver has been dropped. + result_sender.send((idx, decomp_buf)).map_err(|_| { + Bz2Error::Io(io::Error::new(io::ErrorKind::Other, "Receiver dropped")) + })?; Ok(()) }, );