diff --git a/apps/effectcraft-web/js/host.js b/apps/effectcraft-web/js/host.js index cdc71f9d..486865ae 100644 --- a/apps/effectcraft-web/js/host.js +++ b/apps/effectcraft-web/js/host.js @@ -278,8 +278,8 @@ export function audioStop() { // ------------------------------------------------------------------ workers -// Idle workers are reused; each keeps the files it was sent (path → size) so a file goes to a -// worker once. +// Idle workers are reused; each keeps the files it was sent (path → version) so a file goes to a +// worker once, and again when it was replaced (even by one of the same size, #218). const idle = []; const running = new Map(); // job id → {worker, done} // Job workers open a WebGPU device of their own (unless `?nogpuworkers`): their jobs' frames @@ -310,7 +310,8 @@ function spawn(base) { return w; } -/// Run a job in a worker. `files`: [[path, Uint8Array]] the job reads; `onMessage(type, json, +/// Run a job in a worker. `files`: [[path, Uint8Array, version]] the job reads (`version` +/// changes with every write of the file); `onMessage(type, json, /// path, bytes)` receives "reply" (json), "file" (path, bytes: a rendered file to download), /// "store" (path, bytes: a file to keep in browser storage) and "error" (json = message). export function workerRun(id, json, files, base, onMessage) { @@ -339,9 +340,10 @@ export function workerRun(id, json, files, base, onMessage) { onMessage("error", String(e.message || "worker error"), null, null); w.worker.terminate(); }; - for (const [path, bytes] of files) { - if (w.files.get(path) === bytes.length) continue; - w.files.set(path, bytes.length); + for (const [path, bytes, version] of files) { + const tag = version ?? bytes.length; + if (w.files.get(path) === tag) continue; + w.files.set(path, tag); w.worker.postMessage({ type: "file", path, bytes }); } w.worker.postMessage({ type: "job", json }); diff --git a/apps/effectcraft-web/src/files.rs b/apps/effectcraft-web/src/files.rs index ef0889ba..ca6699ea 100644 --- a/apps/effectcraft-web/src/files.rs +++ b/apps/effectcraft-web/src/files.rs @@ -35,6 +35,12 @@ pub fn get(path: &str) -> Option> { STORE.file(path) } +/// A file's bytes and version (new with every write: workers resend a file whose version they +/// haven't seen). +pub fn get_versioned(path: &str) -> Option<(Arc<[u8]>, u64)> { + STORE.file_versioned(path) +} + /// `(path, size)` of every file in the table. pub fn list() -> Vec<(String, usize)> { STORE.files().into_iter().map(|(p, n, _, _)| (p, n)).collect() diff --git a/apps/effectcraft-web/src/frames.rs b/apps/effectcraft-web/src/frames.rs index fc703eba..523fd03c 100644 --- a/apps/effectcraft-web/src/frames.rs +++ b/apps/effectcraft-web/src/frames.rs @@ -19,7 +19,7 @@ //! frame times. Without WebGPU, frames render on the worker's CPU. use std::cell::RefCell; -use std::collections::HashSet; +use std::collections::HashMap; use std::sync::Arc; use effectcraft_engine::offload::footage_files; @@ -45,7 +45,8 @@ extern "C" { struct Fw { index: u32, mirror: Mirror, - files: HashSet, + /// The footage files it was sent, with their versions (a replaced file goes again). + files: HashMap, busy: Option<(u64, RemoteDone)>, alive: bool, /// Its GPU adapter (after it reported), or why it has none. @@ -99,7 +100,7 @@ impl WebFrames { let mut s = s.borrow_mut(); for _ in 0..n { let index = frame_worker_start(&crate::page_url(""), on.as_ref().unchecked_ref(), gpu); - s.workers.push(Fw { index, mirror: Mirror::default(), files: HashSet::new(), busy: None, alive: true, gpu: None }); + s.workers.push(Fw { index, mirror: Mirror::default(), files: HashMap::new(), busy: None, alive: true, gpu: None }); } }); HANDLER.with(|h| *h.borrow_mut() = Some(on)); @@ -116,14 +117,13 @@ impl WebFrames { // Footage the worker hasn't got yet. let w = &mut s.workers[k]; for p in footage_files(&job.project) { - if w.files.contains(&p) { + let Some((d, version)) = crate::files::get_versioned(&p) else { continue }; + if w.files.get(&p) == Some(&version) { continue; } - if let Some(d) = crate::files::get(&p) { - let bytes = js_sys::Uint8Array::from(&d[..]); - frame_worker_post(w.index, &obj(&[("type", "file".into()), ("path", p.as_str().into()), ("bytes", bytes.into())]), &js_sys::Array::new()); - w.files.insert(p); - } + let bytes = js_sys::Uint8Array::from(&d[..]); + frame_worker_post(w.index, &obj(&[("type", "file".into()), ("path", p.as_str().into()), ("bytes", bytes.into())]), &js_sys::Array::new()); + w.files.insert(p, version); } let sync = w.mirror.sync(job.revision, &job.project); let index = w.index; diff --git a/apps/effectcraft-web/src/store.rs b/apps/effectcraft-web/src/store.rs index ae02649b..7017b620 100644 --- a/apps/effectcraft-web/src/store.rs +++ b/apps/effectcraft-web/src/store.rs @@ -42,6 +42,9 @@ pub struct Entry { pub modified: f64, /// Written to the browser's storage (render outputs are not). pub persist: bool, + /// Which write this is: new with every write of the key, so a file replaced by another of + /// the same size still reads as changed (workers are sent it again, #218). + pub version: u64, } /// A queued write for the browser storage: `data` = `None` deletes the key. @@ -58,17 +61,21 @@ pub struct Mirror { entries: BTreeMap, /// Keys changed since the last [`Mirror::take_pending`], in first-change order. dirty: Vec, + /// The last [`Entry::version`] given. + version: u64, } impl Mirror { /// An entry read from the browser's storage at startup (nothing to flush). pub fn load(&mut self, key: &str, data: Arc<[u8]>, modified: f64) { - self.entries.insert(key.to_string(), Entry { data, modified, persist: true }); + let version = self.next_version(); + self.entries.insert(key.to_string(), Entry { data, modified, persist: true, version }); } pub fn put(&mut self, key: &str, data: Arc<[u8]>, modified: f64, persist: bool) { let was_persisted = self.entries.get(key).is_some_and(|e| e.persist); - self.entries.insert(key.to_string(), Entry { data, modified, persist }); + let version = self.next_version(); + self.entries.insert(key.to_string(), Entry { data, modified, persist, version }); if persist || was_persisted { self.touch(key); } @@ -86,6 +93,11 @@ impl Mirror { } } + fn next_version(&mut self) -> u64 { + self.version = self.version.wrapping_add(1); + self.version + } + fn touch(&mut self, key: &str) { if !self.dirty.iter().any(|k| k == key) { self.dirty.push(key.to_string()); @@ -172,6 +184,11 @@ impl Store { self.lock().get(&file_key(path)).map(|e| e.data.clone()) } + /// The file under `path` with its [`Entry::version`]. + pub fn file_versioned(&self, path: &str) -> Option<(Arc<[u8]>, u64)> { + self.lock().get(&file_key(path)).map(|e| (e.data.clone(), e.version)) + } + pub fn remove_file(&self, path: &str) -> bool { let r = self.lock().remove(&file_key(path)); self.changed(); @@ -412,6 +429,20 @@ mod tests { assert_eq!(m.list("files").len(), 1); } + /// A file replaced by different bytes of the same size gets a new version, so workers are + /// sent it again (#218). + #[test] + fn versions_change_with_every_write() { + let s = store(); + s.put_file("/collision.png", Arc::from(&b"red!"[..]), true); + let (_, a) = s.file_versioned("/collision.png").unwrap(); + s.put_file("/collision.png", Arc::from(&b"blue"[..]), true); + let (data, b) = s.file_versioned("/collision.png").unwrap(); + assert_ne!(a, b); + assert_eq!(&data[..], b"blue"); + assert_eq!(s.file_versioned("/collision.png").map(|f| f.1), Some(b), "reading doesn't change it"); + } + #[test] fn keys_and_paths_round_trip() { assert_eq!(file_key("/a b/c.mov"), "files/a b/c.mov"); diff --git a/apps/effectcraft-web/src/worker.rs b/apps/effectcraft-web/src/worker.rs index 81c5fb31..bd86fd5d 100644 --- a/apps/effectcraft-web/src/worker.rs +++ b/apps/effectcraft-web/src/worker.rs @@ -60,8 +60,8 @@ impl Offload for WorkerOffload { let id = req.id; let files = js_sys::Array::new(); for p in &req.files { - if let Some(d) = crate::files::get(p) { - files.push(&js_sys::Array::of2(&JsValue::from_str(p), &js_sys::Uint8Array::from(&d[..]))); + if let Some((d, version)) = crate::files::get_versioned(p) { + files.push(&js_sys::Array::of3(&JsValue::from_str(p), &js_sys::Uint8Array::from(&d[..]), &JsValue::from_f64(version as f64))); } } let json = serde_json::to_string(&req).map_err(|e| e.to_string())?; diff --git a/apps/effectcraft-web/tests/workers.mjs b/apps/effectcraft-web/tests/workers.mjs new file mode 100644 index 00000000..96e3f1f0 --- /dev/null +++ b/apps/effectcraft-web/tests/workers.mjs @@ -0,0 +1,56 @@ +// Job-worker plumbing of the web app (js/host.js) under Node with a fake Worker, +// no browser or Wasm build needed: +// +// node apps/effectcraft-web/tests/workers.mjs +// +// - A reused worker is sent a file again when it was replaced, even by one of the same size +// (#218), and not when it is unchanged. +// Exits non-zero on failure. +import assert from "node:assert/strict"; +import { dirname, join } from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; + +const here = dirname(fileURLToPath(import.meta.url)); +globalThis.self = globalThis; + +// A Worker that starts on "init" and finishes each job at once. +let failStart = false; +const workers = []; +class FakeWorker { + constructor() { + this.posted = []; + this.terminated = false; + workers.push(this); + } + postMessage(m) { + this.posted.push(m); + setTimeout(() => { + if (m.type === "init") { + if (failStart) this.onerror?.({ message: "worker failed to start" }); + else this.onmessage?.({ data: { type: "ready" } }); + } else if (m.type === "job") { + this.onmessage?.({ data: { type: "reply", json: "{}", done: true } }); + } + }); + } + terminate() { + this.terminated = true; + } +} +globalThis.Worker = FakeWorker; + +const host = await import(pathToFileURL(join(here, "../js/host.js"))); +const run = (id, files) => + new Promise((resolve) => host.workerRun(id, "{}", files, "http://localhost/", (kind, json) => kind !== "file" && resolve({ kind, json }))); +const sent = (w) => w.posted.filter((m) => m.type === "file").map((m) => m.bytes[0]); + +// #218: equal-length replacement. +const red = new Uint8Array(512).fill(1); +const blue = new Uint8Array(512).fill(2); +assert.equal((await run(1, [["/collision.png", red, 1]])).kind, "reply"); +assert.equal((await run(2, [["/collision.png", blue, 2]])).kind, "reply"); +assert.equal((await run(3, [["/collision.png", blue, 2]])).kind, "reply"); +assert.equal(workers.length, 1, "the worker is reused"); +assert.deepEqual(sent(workers[0]), [1, 2], "the replacement is sent once, the unchanged file not again"); + +console.log("workers: ok");