Parallelize video frame rendering and extraction

Three changes to improve throughput on multi-core systems:

1. Encoder (video/encoder.rs): render frames in parallel batches using
   rayon's par_iter. Batch size matches the rayon thread pool so all cores
   stay busy. Frames are still written to ffmpeg stdin sequentially to
   preserve order. Memory overhead is bounded to threads * frame_size.

2. Decoder (video/decoder.rs): buffer incoming frames in batches and
   run DCT extraction across the batch in parallel. Reads from ffmpeg
   stdout remain sequential (required), but the CPU-bound extract_frame
   work scales with core count. Same bounded memory overhead.

3. Encoder (video/encoder.rs): scale the FFV1 -slices argument to
   rayon::current_num_threads() (capped at 16) so ffmpeg's own
   intra-frame parallelism also benefits from available cores.

All 23 unit tests pass.

https://claude.ai/code/session_01Wqvuiq921BpoEXMH2eozMG

Author: claude

src/video/decoder.rs

@@ -1,5 +1,6 @@
 use anyhow::{Context, Result};
 use log::info;
+use rayon::prelude::*;
 
 use crate::config::{self, Yts3Config};
 use crate::video::dct::DctTables;
@@ -63,21 +64,48 @@ impl VideoDecoder {
         let stdout = child.stdout.as_mut().unwrap();
         let frame_size = self.width as usize * self.height as usize;
         let mut all_data = Vec::new();
-        let mut frame_buf = vec![0u8; frame_size];
         let mut frame_count = 0u64;
 
+        // Read frames in batches from ffmpeg (I/O must be sequential) and extract
+        // bits from each batch in parallel. Batch size matches the rayon thread pool
+        // so all cores stay busy while we keep memory bounded to `threads * frame_size`.
+        let batch_size = rayon::current_num_threads();
+        let mut batch: Vec<Vec<u8>> = Vec::with_capacity(batch_size);
+
         loop {
+            let mut frame_buf = vec![0u8; frame_size];
             match read_exact_or_eof(stdout, &mut frame_buf) {
                 Ok(true) => {
-                    let frame_data = self.extract_frame(&frame_buf);
-                    all_data.extend_from_slice(&frame_data);
+                    batch.push(frame_buf);
                     frame_count += 1;
+
+                    if batch.len() >= batch_size {
+                        let extracted: Vec<Vec<u8>> = batch
+                            .par_iter()
+                            .map(|f| self.extract_frame(f))
+                            .collect();
+                        for frame_data in extracted {
+                            all_data.extend_from_slice(&frame_data);
+                        }
+                        batch.clear();
+                    }
                 }
                 Ok(false) => break, // EOF
                 Err(e) => return Err(e.into()),
             }
         }
 
+        // Process any remaining frames in the last (partial) batch
+        if !batch.is_empty() {
+            let extracted: Vec<Vec<u8>> = batch
+                .par_iter()
+                .map(|f| self.extract_frame(f))
+                .collect();
+            for frame_data in extracted {
+                all_data.extend_from_slice(&frame_data);
+            }
+        }
+
         let status = child.wait().context("ffmpeg decode process failed")?;
         if !status.success() {
             anyhow::bail!("ffmpeg decode exited with status: {}", status);

src/video/encoder.rs

@@ -3,6 +3,7 @@ use std::process::{Command, Stdio};
 
 use anyhow::{Context, Result};
 use log::info;
+use rayon::prelude::*;
 
 use crate::config::{self, Yts3Config};
 use crate::video::dct::DctTables;
@@ -57,6 +58,10 @@ impl VideoEncoder {
             self.fps
         );
 
+        // Scale FFV1 slice count to available threads for better intra-frame parallelism
+        // inside ffmpeg. Clamped to 16 (a reasonable FFV1 upper bound).
+        let ffv1_slices = rayon::current_num_threads().min(16).to_string();
+
         let mut child = Command::new("ffmpeg")
             .args([
                 "-y",
@@ -75,7 +80,7 @@ impl VideoEncoder {
                 "-level",
                 "3",
                 "-slices",
-                "4",
+                &ffv1_slices,
                 "-slicecrc",
                 "1",
                 output_path,
@@ -88,19 +93,33 @@ impl VideoEncoder {
 
         let stdin = child.stdin.as_mut().unwrap();
 
-        for frame_idx in 0..num_frames {
-            let data_offset = frame_idx * self.bytes_per_frame;
-            let data_end = (data_offset + self.bytes_per_frame).min(packet_data.len());
-            let frame_data = if data_offset < packet_data.len() {
-                &packet_data[data_offset..data_end]
-            } else {
-                &[]
-            };
-
-            let frame_pixels = self.render_frame(frame_data);
-            stdin
-                .write_all(&frame_pixels)
-                .context("failed to write frame data to ffmpeg")?;
+        // Render frames in parallel batches, then write each batch to ffmpeg in order.
+        // Batch size matches the rayon thread pool so we keep all cores busy without
+        // holding more than `threads * frame_size` bytes of rendered pixel data at once.
+        let batch_size = rayon::current_num_threads();
+        let mut frame_idx = 0;
+        while frame_idx < num_frames {
+            let batch_end = (frame_idx + batch_size).min(num_frames);
+            let frames: Vec<Vec<u8>> = (frame_idx..batch_end)
+                .into_par_iter()
+                .map(|idx| {
+                    let data_offset = idx * self.bytes_per_frame;
+                    let data_end = (data_offset + self.bytes_per_frame).min(packet_data.len());
+                    let frame_data = if data_offset < packet_data.len() {
+                        &packet_data[data_offset..data_end]
+                    } else {
+                        &[]
+                    };
+                    self.render_frame(frame_data)
+                })
+                .collect();
+
+            for frame_pixels in &frames {
+                stdin
+                    .write_all(frame_pixels)
+                    .context("failed to write frame data to ffmpeg")?;
+            }
+            frame_idx = batch_end;
         }
 
         drop(child.stdin.take());