Improve par-mode by changing pipeline to FIFO.

This commit also moves unused once_cell dependency to dev-dependencies.

Cargo.toml

@@ -16,17 +16,16 @@ debug = 1
 [features]
 default = ["par"]
 experimental = ["dep:nalgebra"]
-par = ["dep:rayon"]
+par = ["dep:crossbeam-channel"]
 fakesimd = []
 
 [dependencies]
 crc = "2.1"
+crossbeam-channel = { version = "0.5.8", optional = true }
 heapless = "0.7.10"
 md5 = "0.7.0"
 nalgebra = { version = "0.31", optional = true }
 num-traits = "0.2"
-once_cell = "1.10.0"
-rayon = { version = "1.8", optional = true }
 seq-macro = "0.3"
 serde = { version = "1.0", features = ["derive"] }
 toml = "0.5"
@@ -35,6 +34,7 @@ toml = "0.5"
 bitvec = "1.0.0"
 claxon = "0.4.3"
 hound = "3.5.0"
+once_cell = "1.10.0"
 rand = "0.8.5"
 rstest = "0.13.0"
 tempfile = "3"

report/report.md

@@ -16,21 +16,19 @@ Sources used: wikimedia.i_love_you_california, wikimedia.winter_kiss, wikimedia.
     - default: 0.5443052357800366
     - st: 0.5443052357800366
     - dmse: 0.5418910269181569
-    - bsbs: 0.5435087256676912
     - mae: 0.5374571118440491
 
 
 ### Average compression speed (inverse RTF)
   - Reference
-    - opt8lax: 257.2307394912905
-    - opt8: 259.9417855241402
-    - opt5: 498.18069606854715
+    - opt8lax: 255.3836084284399
+    - opt8: 256.6389900206891
+    - opt5: 493.1953626017115
 
   - Ours
-    - default: 222.1443707999417
-    - st: 95.26297207652442
-    - dmse: 128.94722929725017
-    - bsbs: 13.508132367761204
-    - mae: 26.8191475592669
+    - default: 269.09221608610744
+    - st: 94.99698177817307
+    - dmse: 163.21980486345106
+    - mae: 26.88273747979565
 
 

src/coding.rs

@@ -16,11 +16,6 @@
 
 use std::cell::RefCell;
 
-#[cfg(feature = "par")]
-use rayon::iter::IntoParallelIterator;
-#[cfg(feature = "par")]
-use rayon::iter::ParallelIterator;
-
 use super::component::BitRepr;
 use super::component::ChannelAssignment;
 use super::component::Constant;
@@ -36,6 +31,8 @@ use super::config;
 use super::constant::MAX_LPC_ORDER;
 use super::error::SourceError;
 use super::lpc;
+#[cfg(feature = "par")]
+use super::par;
 use super::rice;
 use super::source::Context;
 use super::source::FrameBuf;
@@ -452,53 +449,6 @@ pub fn encode_fixed_size_frame(
     ret
 }
 
-/// Parallel version of `encode_with_fixed_block_size`.
-///
-/// This function is internally called by `encode_with_fixed_block_size`
-/// when `config.multithread == true`. However, one can explicitly call this
-/// function and disable single-threaded mode.
-///
-/// # Errors
-///
-/// This function returns `SourceError` when it failed to read samples from `src`.
-#[cfg(feature = "par")]
-pub fn parallel_encode_with_fixed_block_size<T: Source>(
-    config: &config::Encoder,
-    mut src: T,
-    block_size: usize,
-) -> Result<Stream, SourceError> {
-    let mut stream = Stream::new(src.sample_rate(), src.channels(), src.bits_per_sample());
-    let channels = src.channels();
-    let mut context = Context::new(src.bits_per_sample(), channels);
-    let mut framebufs = vec![];
-    loop {
-        let mut framebuf = FrameBuf::with_size(channels, block_size);
-        let read_samples = src.read_samples(&mut framebuf, &mut context)?;
-        if read_samples == 0 {
-            break;
-        }
-        framebufs.push((context.current_frame_number(), framebuf));
-    }
-    let stream_info = stream.stream_info();
-    let frames = framebufs.into_par_iter().map(|(frame_number, framebuf)| {
-        let mut f = encode_fixed_size_frame(config, &framebuf, frame_number, stream_info);
-        // It's safe to ignore the error here.
-        f.precompute_bitstream().unwrap_or_default();
-        f
-    });
-
-    for frame in &frames.collect::<Vec<_>>() {
-        stream.add_frame(frame.clone());
-    }
-    stream
-        .stream_info_mut()
-        .set_total_samples(src.len_hint().unwrap_or_else(|| context.total_samples()));
-    stream
-        .stream_info_mut()
-        .set_md5_digest(&context.md5_digest());
-    Ok(stream)
-}
-
 /// Encoder entry function for fixed block-size encoding.
 ///
 /// # Errors
@@ -509,8 +459,11 @@ pub fn encode_with_fixed_block_size<T: Source>(
     mut src: T,
     block_size: usize,
 ) -> Result<Stream, SourceError> {
-    if cfg!(feature = "par") && config.multithread {
-        return parallel_encode_with_fixed_block_size(config, src, block_size);
+    #[cfg(feature = "par")]
+    {
+        if config.multithread {
+            return par::encode_with_fixed_block_size(config, src, block_size);
+        }
     }
     let mut stream = Stream::new(src.sample_rate(), src.channels(), src.bits_per_sample());
     let mut framebuf = FrameBuf::with_size(src.channels(), block_size);