26.0.3

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",

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"
 

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<String>,
     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) {
+        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);
+            let skip_samples = (offset_skip / 2).min(data.len());
-            if let Err(e) = out.write_all(&data_bytes[offset_skip.min(data_bytes.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 {
+                offset_buffer.clear();
-                    for s in extra.iter_mut() {
+                append_samples(&mut offset_buffer, extra, target_big);
-                        *s = swap16(*s);
+                offset_buffer_used = (sample_offset as usize * 4).min(offset_buffer.len());
-                    }
-                }
-                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);
                 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;
                 }

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: Write>(w: &mut W, format: OutputFormat, bytes: u32) -> io
     }
 }
 
-/// Buffered writer matching the original's 32 KiB write coalescing.
+/// Append host-endian `i16` samples to `out`, converting byte order once if
-pub struct BufferedWriter {
+/// the requested output format differs from the host byte order.
-    file: File,
+pub fn append_samples(out: &mut Vec<u8>, 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::<u8>(),
+                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<Vec<u8>>) -> 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<SyncSender<Vec<u8>>>,
+    handle: Option<JoinHandle<io::Result<()>>>,
     buf: Vec<u8>,
     closed: bool,
 }
 
-impl BufferedWriter {
+impl AsyncWriter {
-    pub fn new(file: File) -> Self {
+    /// Open the writer, write the format header, and start the writer thread.
-        Self {
+    pub fn new(mut file: File, format: OutputFormat, bytes: u32) -> io::Result<Self> {
-            file,
+        write_header(&mut file, format, bytes)?;
-            buf: Vec::with_capacity(OUTBUF_SZ),
+        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 self.buf.len() >= WRITE_CHUNK_SIZE {
-            if space == 0 {
+                self.flush_chunk()?;
-                self.flush_buf()?;
-                continue;
             }
+            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<()> {
+    /// Append samples with one byte-order conversion pass.
-        if !self.buf.is_empty() {
+    pub fn write_samples(&mut self, samples: &[i16], target_big_endian: bool) -> io::Result<()> {
-            self.file.write_all(&self.buf)?;
+        append_samples(&mut self.buf, samples, target_big_endian);
-            self.buf.clear();
+        if self.buf.len() >= WRITE_CHUNK_SIZE {
+            self.flush_chunk()?;
         }
         Ok(())
     }
 
-    pub fn flush(&mut self) -> io::Result<()> {
+    fn flush_chunk(&mut self) -> io::Result<()> {
-        self.flush_buf()?;
+        if self.buf.is_empty() {
-        self.file.flush()
+            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 {
+impl Drop for AsyncWriter {
-    fn write(&mut self, data: &[u8]) -> io::Result<usize> {
-        self.write_all(data)?;
-        Ok(data.len())
-    }
-    fn flush(&mut self) -> io::Result<()> {
-        BufferedWriter::flush(self)
-    }
-}
-
-impl Drop for BufferedWriter {
     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);
+    }
+}

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();