aleleba/lightcraft
1
0
Fork 0
mirror of https://github.com/storytold/lightcraft.git synced 2026-10-29 12:37:51 -06:00
Code Issues Packages Projects Releases Wiki Activity
Files
bc1f138ee3e8beade33bd60e039a1f8f418d453b
lightcraft/crates/preview/src/tests.rs
T

471 lines
18 KiB
Rust
Raw Blame History

use std::sync::Arc;
use lightcraft_raster::Rgba8;
use super::*;
#[test]
fn cleared_generation_cannot_read_or_repopulate_memory_or_disk() {
let dir = std::env::temp_dir().join(format!("lc-thumbnail-generation-{}-{}", std::process::id(), next_tick()));
let cache = Arc::new(PreviewCache::with_disk(1 << 20, &dir, 1 << 20));
let key = hash_bytes(b"same-key");
let old = cache.generation();
let wrong = Arc::new(gradient(8, 8, 255));
let correct = Arc::new(gradient(8, 8, 0));
cache.put_at(old, key, wrong.clone());
assert!(cache.get_at(old, key).is_some());
cache.clear();
assert!(cache.get(key).is_none());
assert!(cache.get_at(old, key).is_none());
cache.put_at(old, key, wrong.clone());
cache.put_deferred_at(old, key, wrong);
assert!(cache.get(key).is_none());
assert!(cache.disk().unwrap().get(key).is_none());
cache.put_at(cache.generation(), key, correct.clone());
assert_eq!(cache.get(key).unwrap().as_bytes(), correct.as_bytes());
cache.clear();
}
#[test]
fn retired_cache_keeps_its_files_but_refuses_every_read_and_write() {
let dir = temp_dir("retire");
let old = Arc::new(PreviewCache::with_disk(1 << 20, &dir, 1 << 20));
let (kept, late) = (hash_bytes(b"kept"), hash_bytes(b"late"));
let img = Arc::new(gradient(8, 8, 7));
let generation = old.generation();
old.put_at(generation, kept, img.clone());
old.retire();
assert_ne!(old.generation(), generation);
assert!(old.get_at(generation, kept).is_none());
assert!(old.get(kept).is_none());
old.put_at(generation, late, img.clone());
old.put_deferred_at(generation, late, img.clone());
old.put(late, img.clone());
old.put_deferred(late, img.clone());
old.clear();
// a replacement for the same directory still finds the valid file, and only that
let new = PreviewCache::with_disk(1 << 20, &dir, 1 << 20);
assert_eq!(new.get(kept).map(|i| (i.width, i.height)), Some((8, 8)), "the disk file stays (JPEG: lossy)");
assert!(new.get(late).is_none());
new.clear();
}
fn temp_dir(tag: &str) -> std::path::PathBuf {
let d = std::env::temp_dir().join(format!("lc-preview-{tag}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&d);
d
}
fn gradient(w: usize, h: usize, seed: u8) -> Rgba8 {
let data = (0..w * h).map(|i| [((i % w) * 255 / w) as u8, ((i / w) * 255 / h) as u8, seed, 255]).collect();
Rgba8 { width: w, height: h, data }
}
#[test]
fn hashes_are_stable_and_distinct() {
let a = hash_bytes(b"hello");
assert_eq!(a, hash_bytes(b"hello"));
assert_ne!(a, hash_bytes(b"hellp"));
assert_eq!(Hash128::parse(&a.to_string()), Some(a));
assert_ne!(Hasher128::new().str("ab").str("c").finish(), Hasher128::new().str("a").str("bc").finish());
// streaming == one shot
assert_eq!(Hasher128::new().update(b"hel").update(b"lo").finish(), a);
}
#[test]
fn lru_evicts_least_recent_by_cost() {
let mut l = Lru::new(10);
l.insert(1, "a", 4);
l.insert(2, "b", 4);
assert_eq!(l.get(&1), Some(&"a")); // 1 is now most recent
l.insert(3, "c", 4); // over budget: evicts 2
assert!(l.contains(&1) && !l.contains(&2) && l.contains(&3));
assert_eq!(l.cost(), 8);
l.insert(4, "huge", 50); // alone over budget: kept, everything else evicted
assert_eq!((l.len(), l.cost()), (1, 50));
l.insert(4, "small", 1);
assert_eq!((l.len(), l.cost()), (1, 1));
l.retain(|k| *k != 4);
assert!(l.is_empty());
}
#[test]
fn disk_cache_roundtrip_and_prune() {
let dir = temp_dir("disk");
let d = DiskCache::new(&dir, 40_000);
let img = gradient(160, 100, 7);
let k = hash_bytes(b"k1");
assert!(d.get(k).is_none());
d.put(k, &img);
let back = d.get(k).expect("hit");
assert_eq!((back.width, back.height), (160, 100));
let err: i64 = img.data.iter().zip(&back.data).map(|(a, b)| (a[0] as i64 - b[0] as i64).abs() + (a[1] as i64 - b[1] as i64).abs()).max().unwrap();
assert!(err < 24, "JPEG error {err}");
assert!(back.data.iter().all(|p| p[3] == 255));
// fill past the budget: the oldest entries go, the newest stay
let keys: Vec<_> = (0..40u8).map(|i| hash_bytes(&[i, 1])).collect();
for (i, k) in keys.iter().enumerate() {
d.put(*k, &gradient(160, 100, i as u8));
}
assert!(d.size() <= 40_000, "{}", d.size());
assert!(d.get(*keys.last().unwrap()).is_some());
assert!(d.get(keys[0]).is_none());
// a corrupt file is a miss and gets removed
let k2 = hash_bytes(b"corrupt");
d.put(k2, &img);
let hex = k2.to_string();
let p = dir.join(&hex[..2]).join(format!("{hex}.jpg"));
std::fs::write(&p, b"not a jpeg").unwrap();
assert!(d.get(k2).is_none());
assert!(!p.exists());
d.clear();
assert_eq!(d.size(), 0);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn disk_cache_overwrite_accounts_for_replaced_jpeg_only_once() {
let dir = temp_dir("overwrite");
let cache = DiskCache::new(&dir, 1 << 20);
let (a, b) = (hash_bytes(b"overwrite-a"), hash_bytes(b"overwrite-b"));
let first = gradient(80, 60, 4);
cache.put(a, &first);
cache.put(b, &gradient(80, 60, 7));
let initial = cache.size();
// Re-rendering the same key used to double-count bytes although the
// replacement occupied exactly the same pathname on disk.
cache.put(a, &first);
assert_eq!(cache.size(), initial);
assert!(cache.get(b).is_some());
// A replacement with a different encoded size must also update the total.
cache.put(a, &gradient(120, 90, 9));
let path = |key: Hash128| {
let hex = key.to_string();
dir.join(&hex[..2]).join(format!("{hex}.jpg"))
};
let actual = std::fs::metadata(path(a)).unwrap().len() + std::fs::metadata(path(b)).unwrap().len();
assert_eq!(cache.size(), actual);
cache.clear();
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn preview_cache_memory_then_disk() {
let dir = temp_dir("pc");
let k = hash_bytes(b"x");
{
let c = PreviewCache::with_disk(1 << 20, &dir, 1 << 24);
c.put(k, Arc::new(gradient(64, 48, 1)));
assert_eq!(c.mem_usage().0, 1);
assert!(c.get(k).is_some());
}
// a new process: memory is empty, disk serves it
let c = PreviewCache::with_disk(1 << 20, &dir, 1 << 24);
assert_eq!(c.mem_usage().0, 0);
assert_eq!(c.get(k).map(|i| i.width), Some(64));
assert_eq!(c.mem_usage().0, 1);
assert_eq!(c.disk().unwrap().hits.load(std::sync::atomic::Ordering::Relaxed), 1);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn pool_priorities_dedupe_and_drop() {
let mut p: JobPool<u32, u32> = JobPool::new(1);
// One worker, held busy by a gate job while the others queue up.
let gate = Arc::new((std::sync::Mutex::new(false), std::sync::Condvar::new()));
let g = gate.clone();
p.submit(
999,
0,
1000,
Box::new(move || {
let (m, cv) = &*g;
let mut open = m.lock().unwrap();
while !*open {
open = cv.wait(open).unwrap();
}
999
}),
);
// wait until the worker took the gate job
while p.queued() > 0 {
std::thread::yield_now();
}
p.submit(1, 1, 5, Box::new(|| 1)); // off-screen thumb
p.submit(2, 1, 10, Box::new(|| 2)); // visible thumb
p.submit(3, 1, 100, Box::new(|| 3)); // loupe
p.submit(2, 2, 10, Box::new(|| 22)); // newer request for slot 2 replaces the queued one
p.submit(4, 1, 5, Box::new(|| 4));
assert_eq!(p.queued(), 4);
let dropped = p.reprioritize(|s, pr| if *s == 4 { None } else { Some(pr) });
assert_eq!(dropped, vec![4]);
{
let (m, cv) = &*gate;
*m.lock().unwrap() = true;
cv.notify_all();
}
let mut order = Vec::new();
while order.len() < 4 {
if let Some(d) = p.try_recv() {
order.push((d.slot, d.key, d.result));
} else {
std::thread::yield_now();
}
}
assert_eq!(order, vec![(999, 0, 999), (3, 1, 3), (2, 2, 22), (1, 1, 1)]);
}
#[test]
fn pool_runs_inline() {
// No workers: run_inline drains the queue on the caller's thread (the wasm path).
let mut p: JobPool<u8, u8> = JobPool::new(0);
p.submit(1, 0, 1, Box::new(|| 1));
p.submit(2, 0, 9, Box::new(|| 2));
assert_eq!(p.run_inline(1), 1);
assert_eq!(p.try_recv().map(|d| d.result), Some(2));
assert_eq!(p.run_inline(10), 1);
assert_eq!(p.try_recv().map(|d| d.result), Some(1));
}
/// A pool with one worker that is inside a job until the returned sender is used or dropped.
fn pool_with_a_running_job() -> (JobPool<u8, ()>, std::sync::mpsc::Sender<()>) {
let mut p: JobPool<u8, ()> = JobPool::new(1);
let (started, has_started) = std::sync::mpsc::channel();
let (release, released) = std::sync::mpsc::channel::<()>();
p.submit(
1,
0,
1,
Box::new(move || {
let _ = started.send(());
let _ = released.recv_timeout(std::time::Duration::from_secs(60));
}),
);
has_started.recv().unwrap();
(p, release)
}
#[test]
fn pool_drop_waits_for_running_job() {
// issue #620: a job that outlives its pool can still be inside the GPU driver when the process exits
use std::sync::atomic::{AtomicBool, Ordering};
let mut p: JobPool<u8, ()> = JobPool::new(1);
let (started, has_started) = std::sync::mpsc::channel();
let (finished, queued_ran) = (Arc::new(AtomicBool::new(false)), Arc::new(AtomicBool::new(false)));
let f = finished.clone();
p.submit(
1,
0,
9,
Box::new(move || {
let _ = started.send(());
std::thread::sleep(std::time::Duration::from_millis(200));
f.store(true, Ordering::SeqCst);
}),
);
has_started.recv().unwrap();
let q = queued_ran.clone();
p.submit(2, 0, 1, Box::new(move || q.store(true, Ordering::SeqCst)));
drop(p);
assert!(finished.load(Ordering::SeqCst), "drop returned while the job was still running");
assert!(!queued_ran.load(Ordering::SeqCst), "a job that had not started is discarded");
}
#[test]
fn pool_shutdown_reports_workers_and_drop_does_not_wait_again() {
let (mut p, release) = pool_with_a_running_job();
assert_eq!(p.live_workers(), 1);
let t0 = std::time::Instant::now();
assert!(!p.shutdown(std::time::Duration::from_millis(100)), "the job is still running");
let waited = t0.elapsed();
assert!(waited >= std::time::Duration::from_millis(100) && waited < std::time::Duration::from_secs(30), "{waited:?}");
assert_eq!(p.live_workers(), 1);
// stopped: nothing new is queued, and the job that ends is noticed
p.submit(2, 0, 1, Box::new(|| ()));
assert_eq!(p.queued(), 0);
drop(release);
assert!(p.shutdown(std::time::Duration::from_secs(30)));
assert_eq!(p.live_workers(), 0);
let (mut p, _release) = pool_with_a_running_job();
assert!(!p.shutdown(std::time::Duration::ZERO));
let t0 = std::time::Instant::now();
drop(p);
assert!(t0.elapsed() < std::time::Duration::from_secs(1), "dropped after a shutdown: no second wait");
}
#[test]
fn pool_drop_is_bounded_when_a_job_never_ends() {
// a job blocked on something its owner holds must not hang the owner (it would hang quitting)
let (p, _release) = pool_with_a_running_job();
let t0 = std::time::Instant::now();
drop(p);
let waited = t0.elapsed();
assert!(waited >= std::time::Duration::from_secs(1) && waited < std::time::Duration::from_secs(30), "{waited:?}");
}
#[test]
fn pool_stops_after_a_job_panicked_and_when_dropped_by_its_own_job() {
// a worker that unwound is not waited for
let mut p: JobPool<u8, ()> = JobPool::new(1);
p.submit(1, 0, 1, Box::new(|| panic!("a job panics (expected in this test)")));
while p.live_workers() > 0 {
std::thread::yield_now();
}
assert!(p.shutdown(std::time::Duration::from_secs(30)));
// the last owner of a pool is one of its jobs: the worker must not wait for (or join) itself
let slot: Arc<std::sync::Mutex<Option<JobPool<u8, ()>>>> = Arc::new(std::sync::Mutex::new(None));
let (dropped, has_dropped) = std::sync::mpsc::channel();
let mut p: JobPool<u8, ()> = JobPool::new(1);
let s = slot.clone();
p.submit(
1,
0,
1,
Box::new(move || {
let own = loop {
if let Some(p) = s.lock().unwrap().take() {
break p;
}
std::thread::yield_now();
};
let t0 = std::time::Instant::now();
drop(own);
let _ = dropped.send(t0.elapsed());
}),
);
*slot.lock().unwrap() = Some(p);
let waited = has_dropped.recv_timeout(std::time::Duration::from_secs(30)).unwrap();
assert!(waited < std::time::Duration::from_secs(1), "{waited:?}");
}
#[test]
fn disk_cache_never_touches_foreign_files() {
// issue #98: a library opened on a folder that already has a `thumbs/` folder
let dir = temp_dir("foreign");
let foreign = [
dir.join("readme.txt"),
dir.join("ab").join("holiday.jpg"),
dir.join("ab").join("0123456789abcdef0123456789abcdef.jpg"), // wrong shard
dir.join("small").join("0123456789abcdef0123456789abcdef.jpg"),
dir.join("01").join("0123456789ABCDEF0123456789ABCDEF.jpg"), // upper case
dir.join("01").join("0123456789abcdef0123456789abcdef.jpeg"),
dir.join("01").join("0123456789abcdef0123456789abcdef.tmp"),
];
for p in &foreign {
std::fs::create_dir_all(p.parent().unwrap()).unwrap();
std::fs::write(p, vec![7u8; 50_000]).unwrap();
}
// a stale temp file of our own is cleaned up
let own_tmp = dir.join("01").join("0123456789abcdef0123456789abcdef.tmpThreadId9");
std::fs::write(&own_tmp, b"partial").unwrap();
let d = DiskCache::new(&dir, 40_000);
assert_eq!(d.size(), 7, "only the cache's own temp file counts");
// fill past the budget (forces prunes), then clear
for i in 0..40u8 {
d.put(hash_bytes(&[i, 2]), &gradient(160, 100, i));
}
assert!(d.size() <= 40_000);
d.clear();
assert_eq!(d.size(), 0);
assert!(!own_tmp.exists());
for p in &foreign {
assert!(p.exists(), "foreign file deleted: {}", p.display());
}
assert!(disk::is_cache_file("01", "0123456789abcdef0123456789abcdef.jpg"));
assert!(!disk::is_cache_file("01", "0123456789abcdef0123456789abcdef.jpg.bak"));
let _ = std::fs::remove_dir_all(&dir);
}
/// [`JobPool::wait`] returns once a job finishes, which [`JobPool::try_recv`] then hands out, or
/// after the timeout.
#[test]
fn pool_wait_keeps_the_finished_job_for_try_recv() {
let mut p: JobPool<u8, u8> = JobPool::new(1);
assert!(!p.wait(std::time::Duration::from_millis(10)), "nothing runs");
p.submit(1, 0, 1, Box::new(|| 7));
assert!(p.wait(std::time::Duration::from_secs(10)));
assert!(p.wait(std::time::Duration::ZERO), "kept until taken");
assert_eq!(p.try_recv().map(|d| d.result), Some(7));
assert!(p.try_recv().is_none());
}
/// Wait up to `secs` for `f` to hold.
fn eventually(secs: u64, mut f: impl FnMut() -> bool) -> bool {
let t0 = std::time::Instant::now();
while t0.elapsed() < std::time::Duration::from_secs(secs) {
if f() {
return true;
}
std::thread::sleep(std::time::Duration::from_millis(5));
}
false
}
/// A slider drag submits for a view faster than its renders finish. A newer job for a slot used
/// to start beside the running one on another worker: one view's renders piled up on the GPU and
/// most were thrown away. Now a slot runs one job at a time: the newer one waits (replacing an
/// older one still queued), other slots keep going, and it starts when the slot's job ends.
#[test]
fn a_slot_runs_one_job_at_a_time() {
use std::sync::atomic::{AtomicUsize, Ordering};
let mut p: JobPool<u8, u32> = JobPool::new(4);
let (release, released) = std::sync::mpsc::channel::<()>();
let (started, has_started) = std::sync::mpsc::channel();
let running = Arc::new(AtomicUsize::new(0));
let most = Arc::new(AtomicUsize::new(0));
let job = |n: u32, wait: Option<std::sync::mpsc::Receiver<()>>| -> crate::pool::Job<u32> {
let (running, most, started) = (running.clone(), most.clone(), started.clone());
Box::new(move || {
let now = running.fetch_add(1, Ordering::SeqCst) + 1;
most.fetch_max(now, Ordering::SeqCst);
let _ = started.send(n);
if let Some(w) = wait {
let _ = w.recv_timeout(std::time::Duration::from_secs(30));
}
running.fetch_sub(1, Ordering::SeqCst);
n
})
};
p.submit(1, 1, 9, job(1, Some(released)));
assert_eq!(has_started.recv().unwrap(), 1);
// two newer requests for the busy slot: the second replaces the first, neither starts
p.submit(1, 2, 9, job(2, None));
p.submit(1, 3, 9, job(3, None));
// another slot is not held up
p.submit(2, 9, 1, Box::new(|| 9));
let mut done = Vec::new();
assert!(eventually(10, || {
done.extend(std::iter::from_fn(|| p.try_recv()).map(|d| d.result));
done.contains(&9)
}));
std::thread::sleep(std::time::Duration::from_millis(50));
assert!(has_started.try_recv().is_err(), "a second job of slot 1 started beside the first");
assert!(p.is_queued(1));
drop(release);
assert!(eventually(10, || {
done.extend(std::iter::from_fn(|| p.try_recv()).map(|d| d.result));
done.contains(&3)
}));
assert!(!done.contains(&2), "the replaced job never runs");
assert_eq!(most.load(Ordering::SeqCst), 1, "slot 1 ran two jobs at once");
}
/// A job that panics still frees its slot: the slot's next job runs (on another worker).
#[test]
fn a_panicking_job_frees_its_slot() {
let mut p: JobPool<u8, u32> = JobPool::new(2);
p.submit(1, 1, 9, Box::new(|| panic!("simulated render panic")));
assert!(eventually(10, || !p.is_queued(1) && p.live_workers() == 1));
p.submit(1, 2, 9, Box::new(|| 2));
let mut got = None;
assert!(eventually(10, || {
got = got.or_else(|| p.try_recv().map(|d| d.result));
got.is_some()
}));
assert_eq!(got, Some(2));
}
Reference in New Issue View Git Blame Copy Permalink