Merge pull request #995 from vangel76/gap-vp9-parallel-loopfilter

#980: VP9 decode loop-filters a frame as a superblock-row wavefront

crates/vp9/src/decoder.rs

@@ -12,7 +12,8 @@ use crate::tables::*;
 use crate::tile::{FrameShared, RefUse, Strip, TileDecoder};
 use crate::{ColorInfo, Picture, Plane};
 use std::collections::VecDeque;
-use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::{Arc, Condvar, Mutex, MutexGuard, RwLock};
 
 /// Counters describing what the decoder has seen (useful to check test coverage).
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
@@ -71,8 +72,9 @@ pub struct Decoder {
     pending: VecDeque<Pending>,
     #[cfg(all(feature = "threads", not(target_arch = "wasm32")))]
     jobs: VecDeque<std::thread::JoinHandle<()>>,
+    /// Tile columns and the loop filter wavefront of post jobs run on it.
     #[cfg(feature = "threads")]
-    pool: Option<rayon::ThreadPool>,
+    pool: Option<Arc<rayon::ThreadPool>>,
 }
 
 impl Default for Decoder {
@@ -109,7 +111,7 @@ impl Decoder {
     pub fn with_threads(threads: usize) -> Self {
         #[cfg(feature = "threads")]
         let pool = if threads > 1 && cfg!(not(target_arch = "wasm32")) {
-            rayon::ThreadPoolBuilder::new().num_threads(threads).thread_name(|i| format!("vp9-{i}")).build().ok()
+            rayon::ThreadPoolBuilder::new().num_threads(threads).thread_name(|i| format!("vp9-{i}")).build().ok().map(Arc::new)
         } else {
             None
         };
@@ -414,7 +416,16 @@ impl Decoder {
         self.stats.draft_frames += draft as u64;
         let lf = (self.st.lf.level > 0 && !draft)
             .then(|| LfFrame::new(&self.st.lf, &seg, mi_rows, mi_cols, h.color.subsampling_x, h.color.subsampling_y, h.color.bit_depth));
-        let job = PostJob { frame: frame.clone(), strips, mi: mi.clone(), lf, out: h.show_frame.then_some((pts, draft)), pools: self.bufs.clone() };
+        let job = PostJob {
+            frame: frame.clone(),
+            strips,
+            mi: mi.clone(),
+            lf,
+            out: h.show_frame.then_some((pts, draft)),
+            pools: self.bufs.clone(),
+            #[cfg(feature = "threads")]
+            pool: self.pool.clone(),
+        };
         self.run_post(job);
         // refresh_probs (6.1.2).
         if counting {
@@ -558,9 +569,19 @@ fn recycle_strips(strips: Vec<Strip>, pools: &Pools) {
     }
 }
 
-/// The post stage of one frame: build each band from the tile column strips, loop filter it
-/// (superblocks in raster order, so the result is the specification's), publish it once the
-/// next band's filtering can no longer change it, and copy it into the output picture.
+/// The post stage of one frame: build each band from the tile column strips, loop filter it,
+/// publish it once no later filtering can change it, and copy the frame into the output picture.
+///
+/// The specification filters superblock by superblock in raster order (8.8.1). Superblock (r, c)
+/// changes samples of (r, c - 1) (the vertical edges on its left) and of (r - 1, c) (the
+/// horizontal edges on its top reach 7 rows up), and the last superblock before it to change the
+/// bottom rows of (r - 1, c) is (r - 1, c + 1). So row r may filter superblock c once row r - 1
+/// has filtered superblock c + 1: rows run as a wavefront, each two superblocks behind the row
+/// above, with the specification's result. Every row filters its own band: before superblock
+/// c it copies the bottom rows of band r - 1 in that superblock column (shared by row r - 1 once
+/// final for it) into its scratch rows, and once done it writes them back into band r - 1, which
+/// is then final and published. The post thread takes rows in order and helpers on the
+/// decoder's pool join in; with one thread the rows run one after the other, as before.
 struct PostJob {
     frame: Arc<Frame>,
     strips: Vec<Strip>,
@@ -569,9 +590,11 @@ struct PostJob {
     /// (pts, draft) when the frame is shown.
     out: Option<(i64, bool)>,
     pools: Arc<Pools>,
+    #[cfg(feature = "threads")]
+    pool: Option<Arc<rayon::ThreadPool>>,
 }
 
-/// Publishes placeholder bands if a post job dies, so that waiting frames cannot hang.
+/// Publishes placeholder bands if a post job dies or fails, so that waiting frames cannot hang.
 struct PublishGuard<'a>(&'a Frame);
 
 impl Drop for PublishGuard<'_> {
@@ -589,64 +612,295 @@ impl PostJob {
         let f = &*self.frame;
         let _guard = PublishGuard(f);
         let mut pic = self.out.map(|(pts, draft)| PictureBuilder::new(f, pts, draft));
-        let sb_rows = f.sb_rows();
-        let mut held: Option<Band> = None;
-        for r in 0..sb_rows {
+        let Some(lf) = self.lf else {
+            // Without the loop filter every band is final as built.
+            for r in 0..f.sb_rows() {
+                let band = build_band(f, &self.strips, r);
+                if let Some(pic) = &mut pic {
+                    pic.add(f, r, &band);
+                }
+                f.publish(r, band);
+            }
+            recycle_strip_planes(self.strips, &self.pools);
+            return pic.map(PictureBuilder::finish);
+        };
+        let wave = Arc::new(Wave::new(self.frame.clone(), self.strips, self.mi, lf, self.pools, pic));
+        #[cfg(feature = "threads")]
+        if let Some(pool) = &self.pool {
+            // Rows that can be filtered at the same time, the post thread being one of them.
+            let rows = wave.sb_rows.min(wave.sb_cols.div_ceil(2)).min(pool.current_num_threads() + 1);
+            for _ in 1..rows {
+                let w = wave.clone();
+                pool.spawn(move || w.work());
+            }
+        }
+        wave.work();
+        if !wave.wait_done() {
+            return None;
+        }
+        // Every row is done, so no helper reads the strips any more.
+        let strips = std::mem::take(&mut *wave.strips.write().unwrap_or_else(std::sync::PoisonError::into_inner));
+        recycle_strip_planes(strips, &wave.pools);
+        lock(&wave.pic).take().map(PictureBuilder::finish)
+    }
+}
+
+fn recycle_strip_planes(strips: Vec<Strip>, pools: &Pools) {
+    for s in strips {
+        for p in s.planes {
+            pools.samples.put(p);
+        }
+    }
+}
+
+/// Band `r` of a frame from the tile column strips, before the loop filter.
+fn build_band(f: &Frame, strips: &[Strip], r: usize) -> Band {
     let mut band = f.new_band();
     for p in 0..3 {
         let (fs, bh) = (f.strides[p], f.band_h[p]);
         let dst = &mut band.planes[p];
-                for s in &self.strips {
+        for s in strips {
             let ss = s.strides[p];
             let x0 = s.x_off[p];
-                    let w = ss.min(fs - x0);
+            let w = ss.min(fs.saturating_sub(x0));
             for k in 0..bh {
                 let y = r * bh + k;
                 let d = (BAND_PAD + k) * fs + x0;
-                        dst[d..d + w].copy_from_slice(&s.planes[p][y * ss..y * ss + w]);
+                if let (Some(d), Some(s)) = (dst.get_mut(d..d + w), s.planes[p].get(y * ss..y * ss + w)) {
+                    d.copy_from_slice(s);
                 }
             }
         }
-            if let Some(lf) = &self.lf {
-                if let Some(prev) = &held {
-                    // The bottom rows of the previous band, which this row's filter may change.
+    }
+    band
+}
+
+/// Progress of one superblock row of the wavefront.
+struct RowProgress {
+    /// Superblocks filtered; `sb_cols` once the row is done (and band r - 1 published).
+    done: AtomicUsize,
+    /// Threads waiting on `cv` (so that progress without a waiter costs no wake-up call).
+    waiters: AtomicUsize,
+    lock: Mutex<()>,
+    cv: Condvar,
+    /// The bottom `BAND_PAD` rows of the band of each plane, shared superblock column by column
+    /// once final for this row, for the row below.
+    bottom: Mutex<[Vec<u16>; 3]>,
+    /// The band once the row is done, until the row below has written its scratch rows back.
+    band: Mutex<Option<Band>>,
+}
+
+/// The shared state of a frame's loop filter wavefront (see [`PostJob`]). No lock is held while
+/// a superblock is filtered: each row filters its own band and shares only copies.
+struct Wave {
+    frame: Arc<Frame>,
+    strips: RwLock<Vec<Strip>>,
+    mi: Arc<MiGrid>,
+    lf: LfFrame,
+    pools: Arc<Pools>,
+    sb_rows: usize,
+    sb_cols: usize,
+    rows: Vec<RowProgress>,
+    /// The next row to take.
+    next: AtomicUsize,
+    failed: AtomicBool,
+    /// The output picture (shown frames), filled band by band as bands are published.
+    pic: Mutex<Option<PictureBuilder>>,
+}
+
+fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
+    m.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
+impl Wave {
+    fn new(frame: Arc<Frame>, strips: Vec<Strip>, mi: Arc<MiGrid>, lf: LfFrame, pools: Arc<Pools>, pic: Option<PictureBuilder>) -> Wave {
+        let sb_rows = frame.sb_rows();
+        let sb_cols = lf.mi_cols.div_ceil(8);
+        let rows = (0..sb_rows)
+            .map(|_| RowProgress {
+                done: AtomicUsize::new(0),
+                waiters: AtomicUsize::new(0),
+                lock: Mutex::new(()),
+                cv: Condvar::new(),
+                bottom: Mutex::new(std::array::from_fn(|p| pools.bands.take_any(BAND_PAD * frame.strides[p], 0))),
+                band: Mutex::new(None),
+            })
+            .collect();
+        Wave {
+            frame,
+            strips: RwLock::new(strips),
+            mi,
+            lf,
+            pools,
+            sb_rows,
+            sb_cols,
+            rows,
+            next: AtomicUsize::new(0),
+            failed: AtomicBool::new(false),
+            pic: Mutex::new(pic),
+        }
+    }
+
+    /// Take rows and filter them until none is left. A row is only taken by a running thread, so
+    /// a row waiting for the one above always waits for a thread that is making progress.
+    fn work(&self) {
+        while !self.failed.load(Ordering::Acquire) {
+            let r = self.next.fetch_add(1, Ordering::Relaxed);
+            if r >= self.sb_rows {
+                return;
+            }
+            let ok = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| self.row(r))).unwrap_or(false);
+            if !ok {
+                self.fail();
+                return;
+            }
+        }
+    }
+
+    /// Wait until every row is done (false if one failed).
+    fn wait_done(&self) -> bool {
+        self.sb_rows.checked_sub(1).is_none_or(|last| self.wait(last, self.sb_cols))
+    }
+
+    /// Wait until row `r` has filtered `need` superblocks (false if a row failed).
+    fn wait(&self, r: usize, need: usize) -> bool {
+        let Some(row) = self.rows.get(r) else { return false };
+        if row.done.load(Ordering::SeqCst) >= need {
+            return true;
+        }
+        let mut g = lock(&row.lock);
+        row.waiters.fetch_add(1, Ordering::SeqCst);
+        let ok = loop {
+            if row.done.load(Ordering::SeqCst) >= need {
+                break true;
+            }
+            if self.failed.load(Ordering::SeqCst) {
+                break false;
+            }
+            g = row.cv.wait(g).unwrap_or_else(std::sync::PoisonError::into_inner);
+        };
+        row.waiters.fetch_sub(1, Ordering::SeqCst);
+        ok
+    }
+
+    fn set_done(&self, r: usize, n: usize) {
+        if let Some(row) = self.rows.get(r) {
+            // A waiter registers under the lock before it checks `done`, so either it sees the
+            // new value or it is counted here and woken.
+            row.done.store(n, Ordering::SeqCst);
+            if row.waiters.load(Ordering::SeqCst) > 0 {
+                let _g = lock(&row.lock);
+                row.cv.notify_all();
+            }
+        }
+    }
+
+    /// Stop the wavefront: waiting rows give up, the post job publishes placeholders.
+    fn fail(&self) {
+        self.failed.store(true, Ordering::SeqCst);
+        for row in &self.rows {
+            let _g = lock(&row.lock);
+            row.cv.notify_all();
+        }
+    }
+
+    /// Publish final band `r` and copy it into the output picture.
+    fn publish(&self, r: usize, band: Band) {
+        if let Some(pic) = lock(&self.pic).as_mut() {
+            pic.add(&self.frame, r, &band);
+        }
+        self.frame.publish(r, band);
+    }
+
+    /// Share the bottom rows of superblock column `c` of row `r`'s band (final for this row).
+    fn share_bottom(&self, r: usize, band: &Band, c: usize) {
+        let Some(row) = self.rows.get(r) else { return };
+        let mut bottom = lock(&row.bottom);
+        for (p, dst) in bottom.iter_mut().enumerate() {
+            copy_column(&self.frame, p, c, self.sb_cols, &band.planes[p], self.frame.band_h[p], dst, 0);
+        }
+    }
+
+    /// Build and filter row `r`; publish band r - 1 (and the last band) when done.
+    fn row(&self, r: usize) -> bool {
+        let f = &*self.frame;
+        let sb_cols = self.sb_cols;
+        let mut band = build_band(f, &self.strips.read().unwrap_or_else(std::sync::PoisonError::into_inner), r);
+        let above = r.checked_sub(1).and_then(|a| self.rows.get(a));
+        for c in 0..sb_cols {
+            if let Some(above) = above {
+                // Superblock c + 1 of the row above is filtered: the bottom rows of its column c
+                // are final for that row and go into our scratch rows.
+                if !self.wait(r - 1, (c + 2).min(sb_cols)) {
+                    return false;
+                }
+                let bottom = lock(&above.bottom);
+                for (p, src) in bottom.iter().enumerate() {
+                    copy_column(f, p, c, sb_cols, src, 0, &mut band.planes[p], 0);
+                }
+            }
+            let [a, b, d] = &mut band.planes;
+            let mut views = [band_view(f, a, 0, r), band_view(f, b, 1, r), band_view(f, d, 2, r)];
+            filter_superblock(&mut views, &self.mi, &self.lf, r * 8, c * 8);
+            // Superblock c filtered the left edge of column c, so column c - 1 is final here.
+            if c > 0 {
+                self.share_bottom(r, &band, c - 1);
+            }
+            if c + 1 < sb_cols {
+                self.set_done(r, c + 1);
+            }
+        }
+        self.share_bottom(r, &band, sb_cols.saturating_sub(1));
+        if let Some(above) = above {
+            // The scratch rows now hold the final bottom rows of band r - 1.
+            let Some(mut pb) = lock(&above.band).take() else { return false };
             for p in 0..3 {
                 let (fs, bh) = (f.strides[p], f.band_h[p]);
-                        band.planes[p][..BAND_PAD * fs].copy_from_slice(&prev.planes[p][bh * fs..(bh + BAND_PAD) * fs]);
+                if let (Some(d), Some(s)) = (pb.planes[p].get_mut(bh * fs..(bh + BAND_PAD) * fs), band.planes[p].get(..BAND_PAD * fs)) {
+                    d.copy_from_slice(s);
                 }
             }
-                let [a, b, c] = &mut band.planes;
-                let mut views = [band_view(f, a, 0, r), band_view(f, b, 1, r), band_view(f, c, 2, r)];
-                for c in 0..lf.mi_cols.div_ceil(8) {
-                    filter_superblock(&mut views, &self.mi, lf, r * 8, c * 8);
+            self.publish(r - 1, pb);
         }
-                if let Some(prev) = &mut held {
-                    for p in 0..3 {
-                        let (fs, bh) = (f.strides[p], f.band_h[p]);
-                        prev.planes[p][bh * fs..(bh + BAND_PAD) * fs].copy_from_slice(&band.planes[p][..BAND_PAD * fs]);
+        if r + 1 == self.sb_rows {
+            self.publish(r, band);
+        } else if let Some(row) = self.rows.get(r) {
+            *lock(&row.band) = Some(band);
+        }
+        self.set_done(r, sb_cols);
+        true
+    }
+}
+
+impl Drop for Wave {
+    fn drop(&mut self) {
+        recycle_strip_planes(std::mem::take(self.strips.get_mut().unwrap_or_else(std::sync::PoisonError::into_inner)), &self.pools);
+        for row in &self.rows {
+            for p in std::mem::take(&mut *lock(&row.bottom)) {
+                self.pools.bands.put(p);
+            }
+            if let Some(b) = lock(&row.band).take() {
+                for p in b.planes {
+                    self.pools.bands.put(p);
                 }
             }
         }
-            if let Some(prev) = held.take() {
-                if let Some(pic) = &mut pic {
-                    pic.add(f, r - 1, &prev);
     }
-                f.publish(r - 1, prev);
+}
+
+/// Copy `BAND_PAD` rows of superblock column `c` of plane `p` (the last column takes the rest of
+/// the stride) from row `src_row` of `src` to row `dst_row` of `dst`, both `strides[p]` wide.
+#[allow(clippy::too_many_arguments)]
+fn copy_column(f: &Frame, p: usize, c: usize, sb_cols: usize, src: &[u16], src_row: usize, dst: &mut [u16], dst_row: usize) {
+    let fs = f.strides[p];
+    let cw = fs / sb_cols.max(1);
+    let x0 = c * cw;
+    let x1 = if c + 1 >= sb_cols { fs } else { x0 + cw };
+    for k in 0..BAND_PAD {
+        let (s, d) = ((src_row + k) * fs, (dst_row + k) * fs);
+        if let (Some(d), Some(s)) = (dst.get_mut(d + x0..d + x1), src.get(s + x0..s + x1)) {
+            d.copy_from_slice(s);
         }
-            held = Some(band);
-        }
-        if let Some(last) = held {
-            if let Some(pic) = &mut pic {
-                pic.add(f, sb_rows - 1, &last);
-            }
-            f.publish(sb_rows - 1, last);
-        }
-        for s in self.strips {
-            for p in s.planes {
-                self.pools.samples.put(p);
-            }
-        }
-        pic.map(PictureBuilder::finish)
     }
 }
 

crates/vp9/tests/common/mod.rs

@@ -91,6 +91,8 @@ pub const FIXTURES: &[Fixture] = &[
     fx!("error_resilient", "testsrc2", 352, 288, 16, NOISE, P420, ["-b:v", "600k", "-error-resilient", "1"]),
     fx!("odd_size", "testsrc2", 347, 251, 10, NOISE, P420, ["-b:v", "600k", "-frame-parallel", "0"]),
     fx!("tiny_odd", "testsrc2", 33, 17, 8, NOISE, P420, ["-b:v", "200k", "-frame-parallel", "0"]),
+    // Two superblock columns and twelve rows: the loop filter wavefront waits for whole rows.
+    fx!("narrow_tall", "testsrc2", 120, 720, 6, NOISE, P420, ["-b:v", "400k", "-frame-parallel", "0"]),
     fx!("row_mt", "testsrc2", 640, 360, 10, NOISE, P420, ["-b:v", "1M", "-row-mt", "1", "-tile-columns", "1", "-frame-parallel", "0"]),
     fx!("speed0", "mandelbrot", 176, 144, 8, NOISE, P420, ["2pass", "-b:v", "300k", "-speed", "0", "-frame-parallel", "0"]),
     fx!("speed1", "testsrc2", 352, 288, 10, NOISE, P420, ["-b:v", "600k", "-speed", "1", "-frame-parallel", "0"]),

crates/vp9/tests/conformance.rs

@@ -40,6 +40,7 @@ fixture_tests!(
     error_resilient,
     odd_size,
     tiny_odd,
+    narrow_tall,
     row_mt,
     speed0,
     speed1,
@@ -101,6 +102,19 @@ fn draft_mode_changes_only_flagged_non_reference_frames() {
     }
 }
 
+/// The loop filter runs as a wavefront over superblock rows on the decoder's pool: repeated
+/// decodes with several thread counts all match ffmpeg (a race shows up as a mismatch).
+#[test]
+fn loop_filter_wavefront_repeatable() {
+    let f = common::fixture("hd_1080p");
+    let Some((ivf, yuv)) = common::ensure(f) else { return };
+    let reference = std::fs::read(yuv).unwrap();
+    for threads in [2, 5, 16, 16, 16] {
+        let pics = common::decode_file_threads(&ivf, threads).map_err(|e| e.1).unwrap();
+        common::compare(&pics, &reference, f.pix_fmt).unwrap_or_else(|e| panic!("threads {threads}: {e}"));
+    }
+}
+
 /// Two streams of different sizes concatenated into one IVF (key frame at the size change).
 #[test]
 fn resize_concatenated() {

docs/gaps.md

@@ -222,7 +222,9 @@ platforms** G4, G9, G17; **performance** G16; **ecosystem** G11; **measurement**
   software decode is ~4 fps at 1080p single-threaded; 8K is not real time.
 - **Done:** Render (Effects) In to Out renders only the frames between In and Out, as partial
   previews named by segment hash and relative frames, instead of every segment In/Out touches
-  (#424); Render Audio still renders whole audio segments.
+  (#424); Render Audio still renders whole audio segments. VP9 software decode loop-filters a
+  frame as a wavefront over superblock rows on the decoder's pool instead of on one thread,
+  bit-identical, 1.2–1.5× faster with 16 threads (#980).
 - **Estimate:** 30–60 h. Doc: [performance.md](performance.md).
 
 ## G17. Audio device and playback reliability

docs/performance.md

@@ -518,6 +518,36 @@ decode, and the UI-thread present (upload and draw, including the backdrop copie
 blended layers) stays at 9–10 ms p50, as for `stack3`. Before, every frame was composited on the
 CPU (≈ 2 cores more at 23.976 fps) and Full resolution could not keep up at this load.
 
+## Results (VP9 loop filter wavefront, #980, before → after)
+
+The VP9 post stage of a frame (band assembly, loop filter, publication) ran on one thread, and
+later frames wait per band for their references, so the loop filter (about 30 % of the decode
+profile) set the pace of every inter frame: with 16 threads the tile columns spent most of their
+time waiting for the previous frame's bands. The loop filter now runs as a wavefront over
+superblock rows (row r two superblocks behind row r - 1, which is what the raster order of 8.8.1
+allows), the post thread and helpers on the decoder's pool each taking a row. No lock is held
+while a superblock is filtered: rows exchange copies of the 8 rows the filter reaches across a
+band boundary. Output is bit-identical (the conformance tests compare 1, 2, 3, 5 and 16 threads
+with ffmpeg).
+
+AMD Ryzen 7 7800X3D (8 cores / 16 threads), shared with other agents' builds, 2026-10-11.
+`vp9dec` example, 16 threads, runs alternating base / after; paired speed-up = base wall time /
+after wall time per pair:
+
+| fixture | load | paired speed-up, median (range) | after faster | CPU ms/frame |
+|---|---|---|---|---|
+| `bench_1080p` (4 tile columns) | 15–18 | 1.39 (1.20–2.09) | 6/6 | 13.5 → 15.3 |
+| `bench_4k` (8 tile columns) | 15–18 | 1.52 (1.49–1.59) | 6/6 | 48.0 → 57.8 |
+| 1080p30 realtime, noisy (6 Mb/s) | 15–18 | 1.56 (1.27–1.72) | 6/6 | 15.3 → 20.8 |
+| `bench_1080p` | 25–33 | 1.17 (0.91–1.65) | 4/6 | 15.7 → 17.9 |
+| `bench_4k` | 25–33 | 1.29 (0.99–2.03) | 5/6 | 56.9 → 63.1 |
+| 1080p30 realtime, noisy | 25–33 | 1.53 (1.40–1.94) | 6/6 | 22.5 → 25.4 |
+
+With one thread the rows run one after the other: 31.12 → 31.28 G user instructions for the noisy
+1080p clip (+0.5 %, `perf stat`). The extra CPU with threads is mostly rayon workers spinning
+before they sleep. With the machine oversubscribed (load above 16) the wavefront gains less, as
+every row waits for the one above.
+
 ## Results (M4.10: 4K VP9, AV1 and HEVC, before → after)
 
 Baseline = commit `a814ff6` (decoders unchanged since M4.8); after = M4.10. Interleaved runs on