Merge PR #1777

# Conflicts:
#	docs/whats-new.md

crates/preview/src/lib.rs

@@ -18,6 +18,7 @@ pub mod lru;
 pub mod pool;
 
 use std::path::Path;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex, RwLock};
 
 pub use disk::{DiskCache, decode_jpeg, encode_jpeg};
@@ -26,23 +27,58 @@ use lightcraft_raster::Rgba8;
 pub use lru::{Lru, next_tick};
 pub use pool::JobPool;
 
+/// Images that deferred disk writes ([`PreviewCache::put_deferred`]) may hold waiting for the writer, in bytes. A
+/// write that would go past it is skipped (the image stays in memory; a later render writes it), so renders that
+/// arrive faster than the disk takes them can't pile up threads and memory (issue #1522).
+#[cfg(not(target_arch = "wasm32"))]
+const MAX_QUEUED_BYTES: usize = 128 << 20;
+
+/// A deferred disk write waiting for the cache's writer thread.
+#[cfg(not(target_arch = "wasm32"))]
+struct Write {
+    generation: u64,
+    key: Hash128,
+    img: Arc<Rgba8>,
+    cost: usize,
+}
+
 /// Rendered-thumbnail cache: memory LRU over an optional disk cache.
 pub struct PreviewCache {
     /// Current generation; `None` once [`Self::retire`]d (nothing is read or written any more).
     generation: RwLock<Option<u64>>,
     mem: Mutex<Lru<Hash128, Arc<Rgba8>>>,
     disk: Option<DiskCache>,
+    /// The writer thread's queue (started with the first deferred write), and the bytes waiting in it.
+    #[cfg(not(target_arch = "wasm32"))]
+    writer: Mutex<Option<std::sync::mpsc::Sender<Write>>>,
+    queued: Arc<AtomicUsize>,
 }
 
 impl PreviewCache {
     /// Memory-only cache of at most `mem_bytes`.
     pub fn memory(mem_bytes: usize) -> PreviewCache {
-        PreviewCache { generation: RwLock::new(Some(0)), mem: Mutex::new(Lru::new(mem_bytes)), disk: None }
+        Self::new(mem_bytes, None)
+    }
+
+    fn new(mem_bytes: usize, disk: Option<DiskCache>) -> PreviewCache {
+        PreviewCache {
+            generation: RwLock::new(Some(0)),
+            mem: Mutex::new(Lru::new(mem_bytes)),
+            disk,
+            #[cfg(not(target_arch = "wasm32"))]
+            writer: Mutex::new(None),
+            queued: Arc::new(AtomicUsize::new(0)),
+        }
+    }
+
+    /// Bytes of images waiting for a deferred disk write.
+    pub fn queued_bytes(&self) -> usize {
+        self.queued.load(Ordering::Relaxed)
     }
 
     /// Memory LRU + disk cache in `dir` (at most `disk_bytes`).
     pub fn with_disk(mem_bytes: usize, dir: &Path, disk_bytes: u64) -> PreviewCache {
-        PreviewCache { generation: RwLock::new(Some(0)), mem: Mutex::new(Lru::new(mem_bytes)), disk: Some(DiskCache::new(dir, disk_bytes)) }
+        Self::new(mem_bytes, Some(DiskCache::new(dir, disk_bytes)))
     }
 
     pub fn disk(&self) -> Option<&DiskCache> {
@@ -102,25 +138,56 @@ impl PreviewCache {
         let c = cost(&img);
         self.mem.lock().unwrap_or_else(|e| e.into_inner()).insert(key, img.clone(), c);
         drop(current);
-        if self.disk.is_none() {
-            return;
+        let Some(d) = &self.disk else { return };
+        #[cfg(target_arch = "wasm32")]
+        {
+            d.put(key, &img);
         }
-        let me = self.clone();
-        let write = move || {
-            let current = me.generation.read().unwrap_or_else(std::sync::PoisonError::into_inner);
-            if *current != Some(generation) {
+        #[cfg(not(target_arch = "wasm32"))]
+        {
+            let _ = d;
+            if self.queued.load(Ordering::Relaxed).saturating_add(c) > MAX_QUEUED_BYTES {
+                log::debug!("preview cache: disk writes are behind; this one is skipped (the image stays in memory)");
                 return;
             }
-            if let Some(d) = &me.disk {
-                d.put(key, &img);
+            let mut writer = self.writer.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+            if writer.is_none() {
+                *writer = self.start_writer();
+            }
+            let Some(tx) = writer.as_ref() else { return };
+            self.queued.fetch_add(c, Ordering::Relaxed);
+            if tx.send(Write { generation, key, img, cost: c }).is_err() {
+                self.queued.fetch_sub(c, Ordering::Relaxed);
+                *writer = None;
             }
-        };
-        #[cfg(not(target_arch = "wasm32"))]
-        if std::thread::Builder::new().name("lc-preview-write".into()).spawn(write).is_err() {
-            log::warn!("preview cache: could not start a writer thread");
         }
-        #[cfg(target_arch = "wasm32")]
-        write();
+    }
+
+    /// The thread that makes the deferred disk writes, one after another. It holds the cache weakly, so it ends
+    /// once the cache is dropped (its queue's sender with it).
+    #[cfg(not(target_arch = "wasm32"))]
+    fn start_writer(self: &Arc<Self>) -> Option<std::sync::mpsc::Sender<Write>> {
+        let (tx, rx) = std::sync::mpsc::channel::<Write>();
+        let cache = Arc::downgrade(self);
+        let queued = self.queued.clone();
+        let spawned = std::thread::Builder::new().name("lc-preview-write".into()).spawn(move || {
+            for job in rx {
+                if let Some(me) = cache.upgrade() {
+                    let current = me.generation.read().unwrap_or_else(std::sync::PoisonError::into_inner);
+                    if *current == Some(job.generation)
+                        && let Some(d) = &me.disk
+                    {
+                        d.put(job.key, &job.img);
+                    }
+                }
+                queued.fetch_sub(job.cost, Ordering::Relaxed);
+            }
+        });
+        if spawned.is_err() {
+            log::warn!("preview cache: could not start the writer thread");
+            return None;
+        }
+        Some(tx)
     }
 
     /// Drop everything (memory and disk). Does nothing once [`Self::retire`]d.

crates/preview/src/tests.rs

@@ -769,3 +769,26 @@ fn a_panicking_job_frees_its_slot() {
     }));
     assert_eq!(got, Some(2));
 }
+
+/// Deferred disk writes go through one writer with a byte budget: a burst of big renders can't queue more than the
+/// budget (each used to start a thread of its own, holding its image; issue #1522), and what was queued reaches disk.
+#[test]
+fn deferred_writes_are_bounded_and_drain_to_disk() {
+    let dir = temp_dir("deferred-budget");
+    let cache = Arc::new(PreviewCache::with_disk(1 << 30, &dir, 1 << 31));
+    let big = Arc::new(gradient(4096, 2048, 3)); // 32 MiB each
+    let keys: Vec<Hash128> = (0..24u32).map(|i| hash_bytes(&i.to_le_bytes())).collect();
+    for key in &keys {
+        cache.put_deferred(*key, big.clone());
+        assert!(cache.queued_bytes() <= MAX_QUEUED_BYTES, "{} bytes queued", cache.queued_bytes());
+    }
+    let t = std::time::Instant::now();
+    while cache.queued_bytes() > 0 && t.elapsed() < std::time::Duration::from_secs(120) {
+        std::thread::sleep(std::time::Duration::from_millis(20));
+    }
+    assert_eq!(cache.queued_bytes(), 0, "the queue drains");
+    assert!(cache.disk().unwrap().get(keys[0]).is_some(), "the first write reached disk");
+    // every image is still in memory, written to disk or not
+    assert!(keys.iter().all(|k| cache.get(*k).is_some()));
+    let _ = std::fs::remove_dir_all(&dir);
+}

docs/whats-new.md

@@ -23,6 +23,12 @@
   raw whose tile claims far more samples than it holds, and an ARW6 with a broken tile are refused first. All 74
   files in the test corpus still decode.
 
+### Reliability
+- Thumbnails and previews are written to the cache on disk by one background writer instead of a new thread for
+  each image (issue #1522). When renders come in faster than the disk takes them, at most 128 MB of images wait;
+  past that a write is skipped (the image stays in memory and is written by a later render), so a slow disk can't
+  pile up threads and memory.
+
 ### Hard-edged Heal and Clone spots
 - A spot with Feather at 0 no longer draws black pixels exactly on its radius. This also fixes spots at 0% Opacity
   changing pixels along their edge. The hard edge now leaves pixels on the radius unchanged.