From c03450db0619742c98c7f389de63468e3c22c4f7 Mon Sep 17 00:00:00 2001 From: Martin Wohlert Date: Wed, 4 Nov 2026 09:39:26 +0100 Subject: [PATCH] 26.0.3 --- Cargo.lock | 10 +-- Cargo.toml | 2 +- crates/cli/src/lib.rs | 62 ++++--------- crates/cli/src/output.rs | 184 +++++++++++++++++++++++++++++++-------- crates/cli/tests/e2e.rs | 25 ++++++ 5 files changed, 197 insertions(+), 86 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 77595d2..3639c22 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10,7 +10,7 @@ checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" [[package]] name = "cdda_interface" -version = "26.0.1" +version = "26.0.3" dependencies = [ "cdrustanoia_capi_types", "cdrustanoia_core", @@ -20,7 +20,7 @@ dependencies = [ [[package]] name = "cdda_paranoia" -version = "26.0.1" +version = "26.0.3" dependencies = [ "cdda_interface", "cdrustanoia_capi_types", @@ -30,14 +30,14 @@ dependencies = [ [[package]] name = "cdrustanoia_capi_types" -version = "26.0.1" +version = "26.0.3" dependencies = [ "libc", ] [[package]] name = "cdrustanoia_cli" -version = "26.0.1" +version = "26.0.3" dependencies = [ "cdrustanoia_core", "libc", @@ -46,7 +46,7 @@ dependencies = [ [[package]] name = "cdrustanoia_core" -version = "26.0.1" +version = "26.0.3" dependencies = [ "libc", "tempfile", diff --git a/Cargo.toml b/Cargo.toml index b98bbeb..d044098 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,7 +9,7 @@ members = [ ] [workspace.package] -version = "26.0.2" +version = "26.0.3" edition = "2021" license = "LGPL-2.1-or-later" diff --git a/crates/cli/src/lib.rs b/crates/cli/src/lib.rs index 18bac5e..94b3b6f 100644 --- a/crates/cli/src/lib.rs +++ b/crates/cli/src/lib.rs @@ -8,12 +8,12 @@ mod progress; mod span; use crate::log::Log; -use crate::output::{write_header, BufferedWriter, OutputFormat}; +use crate::output::{append_samples, AsyncWriter, OutputFormat}; use crate::progress::Progress; use crate::span::parse_span; use args::Args; use cdrustanoia_core::device::discover::{find_a_cdrom, identify, identify_cooked, identify_sgio}; -use cdrustanoia_core::device::{host_big_endian, swap16, Drive}; +use cdrustanoia_core::device::{host_big_endian, Drive}; use cdrustanoia_core::paranoia::*; use cdrustanoia_core::{VERSION, VERSIONNUM}; use std::fs::{File, OpenOptions}; @@ -282,15 +282,6 @@ fn display_toc(d: &dyn Drive, log: &mut Log) { log.report(format_args!(" ")); } -fn samples_as_bytes(samples: &[i16]) -> &[u8] { - unsafe { - std::slice::from_raw_parts( - samples.as_ptr() as *const u8, - std::mem::size_of_val(samples), - ) - } -} - fn make_output_name( template: &Option, batch: bool, @@ -623,7 +614,7 @@ of the machine.\n\nMore information about /dev/cdrom:" let host_big = host_big_endian(); let target_big = args.output_format.output_big_endian(host_big); - let mut offset_buffer = [0u8; 2352]; + let mut offset_buffer = Vec::new(); let mut offset_buffer_used = 0usize; let mut offset_skip = (sample_offset * 4) as usize; let mut cursor = first_sector; @@ -679,16 +670,18 @@ of the machine.\n\nMore information about /dev/cdrom:" return 1; } }; - let mut out = BufferedWriter::new(file); let bytes = ((batch_last - batch_first + 1) * 2352) as u32; - if let Err(e) = write_header(&mut out, args.output_format, bytes) { - log.report(format_args!("Error writing output: {}", e)); - return 1; - } + let mut out = match AsyncWriter::new(file, args.output_format, bytes) { + Ok(writer) => writer, + Err(e) => { + log.report(format_args!("Error writing output: {}", e)); + return 1; + } + }; if offset_buffer_used > 0 { cursor += 1; - if let Err(e) = out.write_all(&offset_buffer[offset_buffer_used..]) { + if let Err(e) = out.write_bytes(&offset_buffer[offset_buffer_used..]) { log.report(format_args!("Error writing output: {}", e)); return 1; } @@ -717,26 +710,15 @@ of the machine.\n\nMore information about /dev/cdrom:" progress.skipped_flag = false; cursor += 1; - if target_big != host_big { - for s in data.iter_mut() { - *s = swap16(*s); - } - } progress.callback(cursor * 1176 - 1, -2, &mut log); - let data_bytes = samples_as_bytes(data); - if let Err(e) = out.write_all(&data_bytes[offset_skip.min(data_bytes.len())..]) { + let skip_samples = (offset_skip / 2).min(data.len()); + if let Err(e) = out.write_samples(&data[skip_samples..], target_big) { log.report(format_args!("Error writing output: {}", e)); return 1; } offset_skip = 0; - if target_big != host_big { - for s in data.iter_mut() { - *s = swap16(*s); - } - } - if sample_offset != 0 && cursor > batch_last { let extra = { let mut callback = |pos: i64, func: i32| progress.callback(pos, func, &mut log); @@ -757,21 +739,11 @@ of the machine.\n\nMore information about /dev/cdrom:" break; } progress.skipped_flag = false; - if target_big != host_big { - for s in extra.iter_mut() { - *s = swap16(*s); - } - } - let eb = samples_as_bytes(extra); - offset_buffer[..eb.len().min(2352)].copy_from_slice(&eb[..eb.len().min(2352)]); - if target_big != host_big { - for s in extra.iter_mut() { - *s = swap16(*s); - } - } - offset_buffer_used = (sample_offset as usize * 4).min(2352); + offset_buffer.clear(); + append_samples(&mut offset_buffer, extra, target_big); + offset_buffer_used = (sample_offset as usize * 4).min(offset_buffer.len()); progress.callback(cursor * 1176, -2, &mut log); - if let Err(e) = out.write_all(&offset_buffer[..offset_buffer_used]) { + if let Err(e) = out.write_bytes(&offset_buffer[..offset_buffer_used]) { log.report(format_args!("Error writing output: {}", e)); return 1; } diff --git a/crates/cli/src/output.rs b/crates/cli/src/output.rs index 31545e1..a593798 100644 --- a/crates/cli/src/output.rs +++ b/crates/cli/src/output.rs @@ -3,8 +3,14 @@ use std::fs::File; use std::io::{self, Write}; +use std::sync::mpsc::{sync_channel, Receiver, SyncSender}; +use std::thread::{self, JoinHandle}; -pub const OUTBUF_SZ: usize = 32 * 1024; +/// Size of the chunks handed from the rip loop to the writer thread. +pub const WRITE_CHUNK_SIZE: usize = 256 * 1024; + +/// Maximum number of queued chunks. At 256 KiB each this is 16 MiB. +pub const WRITE_QUEUE_CHUNKS: usize = 64; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum OutputFormat { @@ -126,29 +132,71 @@ pub fn write_header(w: &mut W, format: OutputFormat, bytes: u32) -> io } } -/// Buffered writer matching the original's 32 KiB write coalescing. -pub struct BufferedWriter { - file: File, +/// Append host-endian `i16` samples to `out`, converting byte order once if +/// the requested output format differs from the host byte order. +pub fn append_samples(out: &mut Vec, samples: &[i16], target_big_endian: bool) { + if target_big_endian == cfg!(target_endian = "big") { + // SAFETY: the byte view covers exactly the initialized `i16` elements. + let bytes = unsafe { + std::slice::from_raw_parts( + samples.as_ptr().cast::(), + std::mem::size_of_val(samples), + ) + }; + out.extend_from_slice(bytes); + } else { + out.reserve(samples.len() * 2); + for &sample in samples { + let bytes = if target_big_endian { + sample.to_be_bytes() + } else { + sample.to_le_bytes() + }; + out.extend_from_slice(&bytes); + } + } +} + +fn writer_loop(mut file: File, rx: Receiver>) -> io::Result<()> { + while let Ok(chunk) = rx.recv() { + file.write_all(&chunk)?; + } + file.flush() +} + +/// Asynchronous output writer. +/// +/// The rip loop fills fixed-size chunks and hands them to a writer thread +/// through a bounded channel. The bound provides backpressure when the +/// destination cannot keep up, without growing memory without limit. +pub struct AsyncWriter { + tx: Option>>, + handle: Option>>, buf: Vec, closed: bool, } -impl BufferedWriter { - pub fn new(file: File) -> Self { - Self { - file, - buf: Vec::with_capacity(OUTBUF_SZ), +impl AsyncWriter { + /// Open the writer, write the format header, and start the writer thread. + pub fn new(mut file: File, format: OutputFormat, bytes: u32) -> io::Result { + write_header(&mut file, format, bytes)?; + let (tx, rx) = sync_channel(WRITE_QUEUE_CHUNKS); + let handle = thread::spawn(move || writer_loop(file, rx)); + Ok(Self { + tx: Some(tx), + handle: Some(handle), + buf: Vec::with_capacity(WRITE_CHUNK_SIZE), closed: false, - } + }) } - pub fn write_all(&mut self, mut data: &[u8]) -> io::Result<()> { + /// Append raw output bytes, flushing complete chunks to the writer thread. + pub fn write_bytes(&mut self, mut data: &[u8]) -> io::Result<()> { while !data.is_empty() { - let space = OUTBUF_SZ.saturating_sub(self.buf.len()); - if space == 0 { - self.flush_buf()?; - continue; + if self.buf.len() >= WRITE_CHUNK_SIZE { + self.flush_chunk()?; } + let space = WRITE_CHUNK_SIZE.saturating_sub(self.buf.len()); let n = space.min(data.len()); self.buf.extend_from_slice(&data[..n]); data = &data[n..]; @@ -156,39 +204,60 @@ impl BufferedWriter { Ok(()) } - fn flush_buf(&mut self) -> io::Result<()> { - if !self.buf.is_empty() { - self.file.write_all(&self.buf)?; - self.buf.clear(); + /// Append samples with one byte-order conversion pass. + pub fn write_samples(&mut self, samples: &[i16], target_big_endian: bool) -> io::Result<()> { + append_samples(&mut self.buf, samples, target_big_endian); + if self.buf.len() >= WRITE_CHUNK_SIZE { + self.flush_chunk()?; } Ok(()) } - pub fn flush(&mut self) -> io::Result<()> { - self.flush_buf()?; - self.file.flush() + fn flush_chunk(&mut self) -> io::Result<()> { + if self.buf.is_empty() { + return Ok(()); + } + let chunk = std::mem::replace(&mut self.buf, Vec::with_capacity(WRITE_CHUNK_SIZE)); + self.tx + .as_ref() + .ok_or_else(|| { + io::Error::new(io::ErrorKind::BrokenPipe, "output writer is already closed") + })? + .send(chunk) + .map_err(|_| { + io::Error::new( + io::ErrorKind::BrokenPipe, + "output writer thread stopped unexpectedly", + ) + }) } + fn join_writer(&mut self) -> io::Result<()> { + self.tx.take(); + match self.handle.take() { + Some(handle) => match handle.join() { + Ok(result) => result, + Err(_) => Err(io::Error::other("output writer thread panicked")), + }, + None => Ok(()), + } + } + + /// Flush pending data, stop the writer thread, and propagate its result. pub fn close(mut self) -> io::Result<()> { self.closed = true; - self.flush() + let flush_result = self.flush_chunk(); + let join_result = self.join_writer(); + flush_result?; + join_result } } -impl Write for BufferedWriter { - fn write(&mut self, data: &[u8]) -> io::Result { - self.write_all(data)?; - Ok(data.len()) - } - fn flush(&mut self) -> io::Result<()> { - BufferedWriter::flush(self) - } -} - -impl Drop for BufferedWriter { +impl Drop for AsyncWriter { fn drop(&mut self) { if !self.closed { - let _ = self.flush_buf(); + let _ = self.flush_chunk(); + let _ = self.join_writer(); } } } @@ -224,3 +293,48 @@ mod tests { assert_eq!(&v[8..12], b"AIFC"); } } + +#[cfg(test)] +mod async_tests { + use super::*; + use std::fs::File; + + #[test] + fn async_writer_writes_header_and_data() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("out.wav"); + let file = File::create(&path).unwrap(); + let mut writer = AsyncWriter::new(file, OutputFormat::Wav, 4).unwrap(); + writer.write_bytes(b"abcd").unwrap(); + writer.close().unwrap(); + let data = std::fs::read(&path).unwrap(); + assert_eq!(data.len(), 48); + assert_eq!(&data[..4], b"RIFF"); + assert_eq!(&data[44..], b"abcd"); + } + + #[test] + fn async_writer_converts_endianness_once() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("out.aiff"); + let file = File::create(&path).unwrap(); + let mut writer = AsyncWriter::new(file, OutputFormat::Aiff, 4).unwrap(); + writer.write_samples(&[0x1234, -2], true).unwrap(); + writer.close().unwrap(); + let data = std::fs::read(&path).unwrap(); + assert_eq!(&data[54..], &[0x12, 0x34, 0xff, 0xfe]); + } + + #[test] + fn async_writer_handles_multiple_chunks() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("out.raw"); + let file = File::create(&path).unwrap(); + let payload = vec![0x5a; WRITE_CHUNK_SIZE * 3 + 7]; + let mut writer = + AsyncWriter::new(file, OutputFormat::RawHost, payload.len() as u32).unwrap(); + writer.write_bytes(&payload).unwrap(); + writer.close().unwrap(); + assert_eq!(std::fs::read(&path).unwrap(), payload); + } +} diff --git a/crates/cli/tests/e2e.rs b/crates/cli/tests/e2e.rs index 4d88e43..675ce0f 100644 --- a/crates/cli/tests/e2e.rs +++ b/crates/cli/tests/e2e.rs @@ -98,6 +98,31 @@ fn end_to_end_image_rip() { assert!(analyze_text.contains("Checking drive cache/timing behavior")); std::env::remove_var("CDRUSTANOIA_TEST_IMAGE"); + // A payload larger than the async writer's chunk size exercises the + // queue and chunk-boundary path end to end. + let big_image = dir.path().join("big.raw"); + let big_out = dir.path().join("big.out.raw"); + let big_sectors = 300usize; + let mut big_samples = Vec::with_capacity(big_sectors * 1176); + for i in 0..big_sectors * 1176 { + big_samples.push((((i as i64 * 65537) % 65536) as i32 - 32768) as i16); + } + std::fs::write(&big_image, samples_to_bytes(&big_samples)).unwrap(); + std::env::set_var("CDRUSTANOIA_TEST_IMAGE", &big_image); + let argv = vec![ + "cdparanoia".to_string(), + "-q".to_string(), + "-r".to_string(), + "1".to_string(), + big_out.to_string_lossy().into_owned(), + ]; + assert_eq!(cdrustanoia_cli::run(&argv), 0); + assert_eq!( + std::fs::read(&big_out).unwrap(), + samples_to_bytes(&big_samples) + ); + std::env::set_var("CDRUSTANOIA_TEST_IMAGE", &image); + // Multi-track batch mode. let multi = dir.path().join("multi.raw"); let mut multi_samples = Vec::new();