Disk cache: frames and layer buffers are compressed on the writer thread

A render thread handed each frame to the disk cache already copied pixel by
pixel and LZ4-compressed (~9 ms per 1080p frame) before it could start the
next one, and layer buffers were encoded even when the queue then dropped
them. The queue now holds raw RGBA (one memcpy via ColorImage::as_raw) and
Arc<Buf>s; the writer thread encodes them. Every frame of a 120-frame bench
run now reaches disk (113–119 before). cache_bench prints the flush time.

crates/render/src/cache.rs

@@ -361,10 +361,11 @@ pub trait LayerStore: Send + Sync {
     /// mid-render). An entry whose files changed, or damaged, is a miss and is dropped, so the
     /// buffer rendered again replaces it. `virtual_id` names files served from memory.
     fn get_layer(&self, key: u128, virtual_id: Option<VirtualFileId<'_>>) -> Option<(Arc<Buf>, ReadFiles)>;
-    /// Keep a buffer that was slow to render, with the source files it was made of. `since`: the
+    /// Keep a buffer that was slow to render, with the source files it was made of (shared, so
+    /// a store can encode it off the render thread). `since`: the
     /// [`LayerStore::drop_generation`] when its render started (a buffer made of a file dropped
     /// after that is refused).
-    fn put_layer(&self, key: u128, buf: &Buf, files: &EntryFiles, since: u64);
+    fn put_layer(&self, key: u128, buf: &Arc<Buf>, files: &EntryFiles, since: u64);
     /// How many times files were dropped from the store (File ▸ Reload Footage,
     /// [`crate::disk_cache::DiskCache::drop_files`]); `0` for a store that never drops.
     fn drop_generation(&self) -> u64 {
@@ -540,7 +541,7 @@ impl LayerStore for PrefetchStore {
     }
 
     /// Queues the layer for writing to the persistent store the first time `key` is seen.
-    fn put_layer(&self, key: u128, buf: &Buf, files: &EntryFiles, _since: u64) {
+    fn put_layer(&self, key: u128, buf: &Arc<Buf>, files: &EntryFiles, _since: u64) {
         let Ok(mut s) = self.st.lock() else { return };
         if s.known.insert(key) {
             s.writes.push((key, crate::disk_cache::layer_entry(buf, files)));
@@ -983,7 +984,7 @@ mod shared_insert_tests {
         assert!(store.get_layer(5, Some(&two)).is_none());
         assert!(!store.has(5));
         assert_eq!(store.take_misses(), [5]);
-        store.put_layer(5, &buf(), &EntryFiles::of(&files, Some(&two)).unwrap(), 0);
+        store.put_layer(5, &Arc::new(buf()), &EntryFiles::of(&files, Some(&two)).unwrap(), 0);
         assert_eq!(store.take_writes().len(), 1, "written over the changed entry");
         // Through the layer cache: a hit reports its files; incomplete lists are not queued.
         let cache = LayerCache::default();
@@ -1065,7 +1066,7 @@ mod shared_insert_tests {
         fn get_layer(&self, key: u128, virtual_id: Option<crate::disk_cache::VirtualFileId<'_>>) -> Option<(Arc<Buf>, crate::reads::ReadFiles)> {
             super::LayerStore::get_layer(&*self.dc, key, virtual_id)
         }
-        fn put_layer(&self, key: u128, buf: &Buf, files: &crate::disk_cache::EntryFiles, since: u64) {
+        fn put_layer(&self, key: u128, buf: &Arc<Buf>, files: &crate::disk_cache::EntryFiles, since: u64) {
             self.arrived.wait();
             self.go_on.wait();
             super::LayerStore::put_layer(&*self.dc, key, buf, files, since);

crates/render/src/disk_cache.rs

@@ -264,11 +264,41 @@ impl LayerPrefetch {
 #[cfg_attr(target_arch = "wasm32", allow(dead_code))]
 enum Job {
     /// Kind, key, header format byte, payload.
-    Write(Kind, u128, u8, Vec<u8>),
+    Write(Kind, u128, u8, Payload),
     /// Delete the frame and layer entries made of these files ([`DiskCache::drop_files`]).
     Drop(Vec<Arc<str>>),
 }
 
+/// What a queued write holds. 8-bit frames and layer buffers are encoded (LZ4) on the writer
+/// thread, not on the render thread that queued them: encoding a 1080p frame took ~9 ms of every
+/// frame's render, and a full queue no longer throws away encoding work.
+enum Payload {
+    Encoded(Vec<u8>),
+    /// 8-bit premultiplied RGBA, `width * height * 4` bytes, and the files it was made of.
+    Frame(u32, u32, Vec<u8>, EntryFiles),
+    Layer(Arc<Buf>, EntryFiles),
+}
+
+impl Payload {
+    /// Bytes it holds (the frame queue's budget).
+    #[cfg_attr(target_arch = "wasm32", allow(dead_code))]
+    fn len(&self) -> usize {
+        match self {
+            Payload::Encoded(p) | Payload::Frame(_, _, p, _) => p.len(),
+            Payload::Layer(b, _) => b.img.data.len().saturating_mul(std::mem::size_of::<effectcraft_raster::Px>()),
+        }
+    }
+
+    /// The entry's payload: its source files, then its pixels.
+    fn encode(self) -> Vec<u8> {
+        match self {
+            Payload::Encoded(p) => p,
+            Payload::Frame(w, h, rgba, files) => with_files(&files, encode_frame(w, h, &rgba)),
+            Payload::Layer(b, files) => with_files(&files, encode_layer(&b)),
+        }
+    }
+}
+
 struct Queue {
     /// Frame writes (taken first) and their payload bytes.
     frames: std::collections::VecDeque<Job>,
@@ -771,20 +801,20 @@ impl DiskCache {
 
     /// Queue a write; `false` when it was dropped (a full queue).
     fn enqueue(&self, kind: Kind, key: u128, payload: Vec<u8>) -> bool {
-        self.enqueue_format(kind, key, FORMAT_RGBA8, payload, None)
+        self.enqueue_format(kind, key, FORMAT_RGBA8, Payload::Encoded(payload), None)
     }
 
     /// [`DiskCache::enqueue`] with the entry's header format byte. `made`: the source files of a
     /// frame or layer entry and the drop generation its reads started at
     /// ([`DiskCache::drop_generation`]); refused (`false`) when one of them was dropped since.
-    fn enqueue_format(&self, kind: Kind, key: u128, format: u8, payload: Vec<u8>, made: Option<(&EntryFiles, u64)>) -> bool {
+    fn enqueue_format(&self, kind: Kind, key: u128, format: u8, payload: Payload, made: Option<(&EntryFiles, u64)>) -> bool {
         let stale = |q: &Queue| made.is_some_and(|(files, since)| q.drops.stale(files.files().iter().map(|f| &f.path), since));
         #[cfg(target_arch = "wasm32")]
         {
             if self.queue.0.lock().is_ok_and(|q| stale(&q)) {
                 return false;
             }
-            self.write_format(kind, key, format, &payload);
+            self.write_format(kind, key, format, &payload.encode());
             true
         }
         #[cfg(not(target_arch = "wasm32"))]
@@ -1033,7 +1063,7 @@ impl DiskCache {
         }
         self.count_encode();
         let payload = with_files(&files, encode_frame_f16(width, height, rgba));
-        self.enqueue_format(Kind::Frame, key, FORMAT_F16_FILES, payload, Some((&files, since)))
+        self.enqueue_format(Kind::Frame, key, FORMAT_F16_FILES, Payload::Encoded(payload), Some((&files, since)))
     }
 
     /// Write a half-float frame synchronously (tests); a list that can't be stored is not.
@@ -1050,6 +1080,22 @@ impl DiskCache {
     /// render started at).
     #[allow(clippy::too_many_arguments)]
     pub fn put_frame(&self, key: u128, width: u32, height: u32, rgba: &[u8], files: &ReadFiles, virtual_id: Option<VirtualFileId<'_>>, since: u64) -> bool {
+        self.put_frame_owned(key, width, height, rgba.to_vec(), files, virtual_id, since)
+    }
+
+    /// [`DiskCache::put_frame`] of a frame the caller hands over (no copy; compressed on the
+    /// writer thread).
+    #[allow(clippy::too_many_arguments)]
+    pub fn put_frame_owned(
+        &self,
+        key: u128,
+        width: u32,
+        height: u32,
+        rgba: Vec<u8>,
+        files: &ReadFiles,
+        virtual_id: Option<VirtualFileId<'_>>,
+        since: u64,
+    ) -> bool {
         if rgba.len() as u64 != width as u64 * height as u64 * 4 {
             return false;
         }
@@ -1057,13 +1103,13 @@ impl DiskCache {
         if self.contains(Kind::Frame, key) {
             return true;
         }
-        // (no room on the disk: refused before the pixels are compressed)
+        // (no room on the disk: refused before the pixels are queued and compressed)
         if !self.accepts_writes() {
             return false;
         }
         self.count_encode();
-        let payload = with_files(&files, encode_frame(width, height, rgba));
-        self.enqueue_format(Kind::Frame, key, FORMAT_RGBA8_FILES, payload, Some((&files, since)))
+        let made = files.clone();
+        self.enqueue_format(Kind::Frame, key, FORMAT_RGBA8_FILES, Payload::Frame(width, height, rgba, files), Some((&made, since)))
     }
 
     /// A layer buffer (key: [`layer_disk_key`]) whose source files are unchanged, with those
@@ -1078,13 +1124,14 @@ impl DiskCache {
     /// [`EntryFiles::of`]: a buffer whose files can't all be named must not be stored). Refused
     /// when one of them was dropped after `since`, the [`DiskCache::drop_generation`] its render
     /// started at.
-    pub fn put_layer(&self, key: u128, buf: &Buf, files: &EntryFiles, since: u64) {
-        // (no room on the disk: refused before the buffer is compressed)
+    /// The buffer is shared, and encoded on the writer thread.
+    pub fn put_layer(&self, key: u128, buf: &Arc<Buf>, files: &EntryFiles, since: u64) {
+        // (no room on the disk: refused before the buffer is queued and compressed)
         if self.contains(Kind::Layer, key) || !self.accepts_writes() {
             return;
         }
         self.count_encode();
-        let _ = self.enqueue_format(Kind::Layer, key, FORMAT_LAYER_FILES, with_files(files, encode_layer(buf)), Some((files, since)));
+        let _ = self.enqueue_format(Kind::Layer, key, FORMAT_LAYER_FILES, Payload::Layer(buf.clone(), files.clone()), Some((files, since)));
     }
 
     /// Write a layer buffer synchronously (tests); a list that can't be stored is not.
@@ -1192,7 +1239,7 @@ fn writer(dc: std::sync::Weak<DiskCache>, q: Arc<(Mutex<Queue>, Condvar)>) {
         let Some(cache) = dc.upgrade() else { return };
         let written = match job {
             Job::Write(kind, key, format, payload) => {
-                cache.write_format(kind, key, format, &payload);
+                cache.write_format(kind, key, format, &payload.encode());
                 (kind == Kind::Frame).then_some(key)
             }
             Job::Drop(paths) => {
@@ -1292,7 +1339,7 @@ impl crate::cache::LayerStore for DiskCache {
         DiskCache::get_layer(self, key, virtual_id).map(|(b, f)| (Arc::new(b), f))
     }
     /// Queues a layer write on the disk cache.
-    fn put_layer(&self, key: u128, buf: &Buf, files: &EntryFiles, since: u64) {
+    fn put_layer(&self, key: u128, buf: &Arc<Buf>, files: &EntryFiles, since: u64) {
         DiskCache::put_layer(self, key, buf, files, since);
     }
     /// The current drop generation of the disk cache.
@@ -1706,7 +1753,7 @@ mod tests {
         let dc = DiskCache::open(&dir, 1 << 30).unwrap();
         assert!(dc.get_layer(42, None).is_none());
         let b = buf(1, 33, 17);
-        dc.put_layer(42, &b, &EntryFiles::default(), 0);
+        dc.put_layer(42, &Arc::new(b.clone()), &EntryFiles::default(), 0);
         dc.flush();
         let (back, files) = dc.get_layer(42, None).unwrap();
         assert!(files.is_empty() && files.is_complete());
@@ -1715,10 +1762,14 @@ mod tests {
         let rgba: Vec<u8> = (0..20 * 10 * 4).map(|i| (i % 251) as u8).collect();
         assert!(dc.put_frame(7, 20, 10, &rgba, &ReadFiles::default(), None, 0));
         assert!(!dc.put_frame(8, 2, 2, &[0; 3], &ReadFiles::default(), None, 0), "a wrong-size frame is refused");
+        // Handed over and compressed on the writer thread: the same pixels come back.
+        assert!(dc.put_frame_owned(9, 20, 10, rgba.clone(), &ReadFiles::default(), None, 0));
+        assert!(!dc.put_frame_owned(10, 2, 2, vec![0; 3], &ReadFiles::default(), None, 0), "a wrong-size frame is refused");
         dc.flush();
         assert_eq!(dc.get_frame(7, None).map(|f| f.0).unwrap().rgba, rgba);
+        assert_eq!(dc.get_frame(9, None).map(|f| f.0).unwrap().rgba, rgba);
         let st = dc.stats();
-        assert_eq!((st.entries, st.hits, st.misses, st.writes), (2, 2, 1, 2));
+        assert_eq!((st.entries, st.hits, st.misses, st.writes), (3, 3, 1, 3));
         // Wrong kind is a miss.
         assert!(dc.get_frame(42, None).map(|f| f.0).is_none());
     }
@@ -1780,7 +1831,7 @@ mod tests {
         let dir = tmpdir("burst");
         let dc = DiskCache::open(&dir, 1 << 30).unwrap();
         for k in 0..300 {
-            dc.put_layer(10_000 + k, &buf(1, 64, 64), &EntryFiles::default(), 0);
+            dc.put_layer(10_000 + k, &Arc::new(buf(1, 64, 64)), &EntryFiles::default(), 0);
         }
         let rgba = vec![7u8; 32 * 32 * 4];
         for k in 0..300 {
@@ -1950,7 +2001,7 @@ mod tests {
         let rgba = vec![7u8; 16 * 16 * 4];
         assert!(!dc.put_frame(1, 16, 16, &rgba, &ReadFiles::default(), None, 0), "refused");
         assert!(!dc.put_frame_f16(2, 16, 16, &[0x3c00u16; 16 * 16 * 4], &ReadFiles::default(), None, 0), "refused");
-        dc.put_layer(3, &buf(3, 16, 16), &EntryFiles::default(), 0);
+        dc.put_layer(3, &Arc::new(buf(3, 16, 16)), &EntryFiles::default(), 0);
         dc.flush();
         assert_eq!(dc.encodes(), 0, "nothing compressed");
         assert_eq!(dc.stats().entries, 0);
@@ -1960,7 +2011,7 @@ mod tests {
         assert_eq!(dc.stats().limit_bytes, 32 << 20);
         assert!(!dc.accepts_writes());
         assert!(!dc.put_frame(1, 16, 16, &rgba, &ReadFiles::default(), None, 0), "refused");
-        dc.put_layer(3, &buf(3, 16, 16), &EntryFiles::default(), 0);
+        dc.put_layer(3, &Arc::new(buf(3, 16, 16)), &EntryFiles::default(), 0);
         assert_eq!(dc.encodes(), 0, "nothing compressed");
         // A small limit the setting asks for (not free space) is used.
         dc.set_disk_space(None);
@@ -1971,7 +2022,7 @@ mod tests {
         dc.set_disk_space(Some((4096 * GB, 500 * GB)));
         assert!(dc.put_frame(1, 16, 16, &rgba, &ReadFiles::default(), None, 0));
         assert!(dc.put_frame_f16(2, 16, 16, &[0x3c00u16; 16 * 16 * 4], &ReadFiles::default(), None, 0));
-        dc.put_layer(3, &buf(3, 16, 16), &EntryFiles::default(), 0);
+        dc.put_layer(3, &Arc::new(buf(3, 16, 16)), &EntryFiles::default(), 0);
         dc.flush();
         assert_eq!(dc.encodes(), 3);
         assert_eq!(dc.stats().entries, 3);
@@ -2076,11 +2127,11 @@ mod tests {
         assert!(dc.get_frame(k, Some(&two)).is_none() && !dc.contains(Kind::Frame, k), "replaced: a miss, deleted");
         // A layer buffer made of it.
         let lk = layer_disk_key(7);
-        dc.put_layer(lk, &buf(1, 4, 4), &EntryFiles::of(&files, Some(&one)).unwrap(), 0);
+        dc.put_layer(lk, &Arc::new(buf(1, 4, 4)), &EntryFiles::of(&files, Some(&one)).unwrap(), 0);
         dc.flush();
         assert!(dc.get_layer(lk, Some(&one)).is_some_and(|(b, f)| b.img == buf(1, 4, 4).img && f == files && f.is_complete()));
         assert!(dc.get_layer(lk, None).is_none() && !dc.contains(Kind::Layer, lk), "no identity: a miss, deleted");
-        dc.put_layer(lk, &buf(1, 4, 4), &EntryFiles::of(&files, Some(&one)).unwrap(), 0);
+        dc.put_layer(lk, &Arc::new(buf(1, 4, 4)), &EntryFiles::of(&files, Some(&one)).unwrap(), 0);
         dc.flush();
         assert!(dc.get_layer(lk, Some(&two)).is_none() && !dc.contains(Kind::Layer, lk), "replaced: a miss, deleted");
         // The browser's sealed layer entries carry the same list.
@@ -2324,7 +2375,7 @@ mod tests {
         assert!(dc.dropped_since(&read(&[&a]), before) && !dc.dropped_since(&read(&[&b]), before));
         assert!(!dc.put_frame(1, 1, 1, &[1; 4], &read(&[&a, &b]), None, before), "a frame from before the drop");
         assert!(!dc.put_frame_f16(2, 1, 1, &[1; 4], &read(&[&a]), None, before));
-        dc.put_layer(3, &buf(3, 2, 2), &EntryFiles::of(&read(&[&a]), None).unwrap(), before);
+        dc.put_layer(3, &Arc::new(buf(3, 2, 2)), &EntryFiles::of(&read(&[&a]), None).unwrap(), before);
         assert!(dc.put_frame(4, 1, 1, &[4; 4], &read(&[&b]), None, before), "b.png wasn't dropped");
         dc.flush();
         assert!(!dc.contains(Kind::Frame, 1) && !dc.contains(Kind::Frame, 2) && !dc.contains(Kind::Layer, 3));
@@ -2334,7 +2385,7 @@ mod tests {
         let after = dc.drop_generation();
         assert!(after > before && !dc.dropped_since(&read(&[&a]), after));
         assert!(dc.put_frame(5, 1, 1, &[5; 4], &read(&[&a]), None, after));
-        dc.put_layer(6, &buf(6, 2, 2), &EntryFiles::of(&read(&[&a]), None).unwrap(), after);
+        dc.put_layer(6, &Arc::new(buf(6, 2, 2)), &EntryFiles::of(&read(&[&a]), None).unwrap(), after);
         dc.flush();
         assert_eq!(dc.get_frame(5, None).map(|f| f.0.rgba), Some(vec![5; 4]));
         assert!(dc.get_layer(6, None).is_some());

crates/ui-egui/examples/cache_bench.rs

@@ -170,8 +170,10 @@ fn main() {
     println!("footage cache     {} hits, {} misses, {} frames ({} MB) kept", p.hits, p.misses, p.frames, p.bytes >> 20);
     if let Some(dc) = &s.disk_cache {
         // (writes finish on the disk cache's own thread)
+        let t = Instant::now();
         dc.flush();
+        let flushed = t.elapsed().as_secs_f64();
         let d = dc.stats();
-        println!("disk cache        {} written, {} dropped, {} entries ({} MB)", d.writes, d.dropped, d.entries, d.bytes >> 20);
+        println!("disk cache        {} written, {} dropped, {} entries ({} MB), flushed in {flushed:.2} s", d.writes, d.dropped, d.entries, d.bytes >> 20);
     }
 }

crates/ui-egui/src/frames.rs

@@ -571,10 +571,8 @@ fn put_disk_frame(dc: &DiskCache, dk: u128, img: &FrameImage, files: &ReadFiles,
         return false;
     }
     match img {
-        FrameImage::Cpu(img) => {
-            let rgba: Vec<u8> = img.pixels.iter().flat_map(|c| c.to_array()).collect();
-            dc.put_frame(dk, img.size[0] as u32, img.size[1] as u32, &rgba, files, virtual_id, since)
-        }
+        // (one copy; the disk cache's writer thread compresses it)
+        FrameImage::Cpu(img) => dc.put_frame_owned(dk, img.size[0] as u32, img.size[1] as u32, img.as_raw().to_vec(), files, virtual_id, since),
         FrameImage::CpuF16(f) => dc.put_frame_f16(dk, f.width, f.height, f.bits(), files, virtual_id, since),
         FrameImage::Gpu(_) => false,
     }