mirror of
https://github.com/storytold/effectcraft.git
synced 2026-10-14 19:27:05 -04:00
Web: a replaced file goes to reused workers again, even at the same size (#218)
Job workers remembered the files they were sent by path and byte length, so footage replaced by different bytes of the same length at the same path was never resent: a Render Queue export through a reused worker rendered the old image while the page showed the new one. Frame workers remembered paths only, so they never got a replacement at all. Every write of a stored file now gets a version, and both kinds of worker remember the version they were sent. Tests: versions_change_with_every_write (store) and apps/effectcraft-web/tests/workers.mjs (Node, fake Worker: the replacement is sent once, an unchanged file not again). Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_018gnTnH3Jmk5zr4tMq1Mc9f
This commit is contained in:
6 files changed
+114
-19
No files matched your search
@@ -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 });
|
||||
|
||||
@@ -35,6 +35,12 @@ pub fn get(path: &str) -> Option<Arc<[u8]>> {
|
||||
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()
|
||||
|
||||
@@ -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<String>,
|
||||
/// The footage files it was sent, with their versions (a replaced file goes again).
|
||||
files: HashMap<String, u64>,
|
||||
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;
|
||||
|
||||
@@ -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<String, Entry>,
|
||||
/// Keys changed since the last [`Mirror::take_pending`], in first-change order.
|
||||
dirty: Vec<String>,
|
||||
/// 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");
|
||||
|
||||
@@ -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())?;
|
||||
|
||||
@@ -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");
|
||||
Reference in new issue
Block a user