diff --git a/crates/preview/src/lib.rs b/crates/preview/src/lib.rs index a571ecda0..14bd50a1b 100644 --- a/crates/preview/src/lib.rs +++ b/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, + 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>, mem: Mutex>>, disk: Option, + /// The writer thread's queue (started with the first deferred write), and the bytes waiting in it. + #[cfg(not(target_arch = "wasm32"))] + writer: Mutex>>, + queued: Arc, } 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) -> 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) -> Option> { + let (tx, rx) = std::sync::mpsc::channel::(); + 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. diff --git a/crates/preview/src/tests.rs b/crates/preview/src/tests.rs index 17e1ba89c..2f07e927f 100644 --- a/crates/preview/src/tests.rs +++ b/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 = (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); +} diff --git a/docs/whats-new.md b/docs/whats-new.md index 16a6003ce..d5d8aa44e 100644 --- a/docs/whats-new.md +++ b/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.