From 5bb676876673dc95f6f6d0e6dcac8e2068385c78 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Sat, 3 Oct 2026 08:36:00 -0400 Subject: [PATCH] M13.10: web frame workers fed by project diffs, offloaded Roto Brush propagation, non-blocking wait jobs, browser Media Browser Viewer frames render in long-lived Web Workers holding a project replica (engine::remote: structural JSON diff/patch, Mirror, FrameServer); the frame cache hands jobs to a RemoteFrames renderer (UI-thread fallback). Roto Brush propagation runs in a job worker and streams segmentations back (FrameSeg binary form), relayed to frame workers. Control-channel wait:true commands whose job is offloaded reply when it ends. The Media Browser browses a pluggable Browser: the file system on the desktop; browser storage and File System Access folders on the web. Co-Authored-By: Claude Opus 5.5 --- apps/effectcraft-web/js/host.js | 155 +++++++++ apps/effectcraft-web/src/app.rs | 11 +- apps/effectcraft-web/src/browse.rs | 277 ++++++++++++++++ apps/effectcraft-web/src/frames.rs | 261 +++++++++++++++ apps/effectcraft-web/src/lib.rs | 7 +- apps/effectcraft-web/src/worker.rs | 11 + apps/effectcraft-web/web/worker.js | 11 +- crates/engine/src/commands/panels_cmds.rs | 68 +++- crates/engine/src/lib.rs | 8 + crates/engine/src/media_browser.rs | 182 +++++++++- crates/engine/src/offload.rs | 130 +++++++- crates/engine/src/remote.rs | 384 ++++++++++++++++++++++ crates/engine/src/roto.rs | 41 ++- crates/engine/src/tests_panels.rs | 61 ++++ crates/engine/src/tests_roto.rs | 49 +++ crates/render/src/lib.rs | 3 +- crates/track/src/roto/mod.rs | 55 ++++ crates/track/src/roto/tests.rs | 6 + crates/ui-egui/src/control.rs | 87 ++++- crates/ui-egui/src/frames.rs | 205 ++++++++++-- crates/ui-egui/src/lib.rs | 34 +- crates/ui-egui/src/panels/media_panels.rs | 26 +- crates/ui-egui/tests/ui_web_offload.rs | 204 ++++++++++++ 23 files changed, 2197 insertions(+), 79 deletions(-) create mode 100644 apps/effectcraft-web/src/browse.rs create mode 100644 apps/effectcraft-web/src/frames.rs create mode 100644 crates/engine/src/remote.rs create mode 100644 crates/ui-egui/tests/ui_web_offload.rs diff --git a/apps/effectcraft-web/js/host.js b/apps/effectcraft-web/js/host.js index 7b6200eb..1fcdc40b 100644 --- a/apps/effectcraft-web/js/host.js +++ b/apps/effectcraft-web/js/host.js @@ -333,3 +333,158 @@ export function workerCancel(id) { export function workerStats() { return { running: running.size, idle: idle.length }; } + +// ------------------------------------------------------------------ frame workers + +// Long-lived workers that render viewer frames from a project replica (src/frames.rs). Unlike job +// workers they are never terminated between requests; the page sends one frame at a time each. +const frameWorkers = []; + +/// Start a frame worker; returns its index. `onMessage(index, json, bytes, error)` receives each +/// `frameReply` (JSON, pixels as a Uint8Array) or, if the worker dies, the error. +export function frameWorkerStart(base, onMessage) { + const index = frameWorkers.length; + const worker = new Worker(new URL("worker.js", base), { type: "module", name: `effectcraft-frames-${index}` }); + worker.onmessage = (e) => { + const m = e.data; + if (m && m.type === "frameReply") onMessage(index, m.json, m.bytes ? new Uint8Array(m.bytes) : null, undefined); + }; + worker.onerror = (e) => { + onMessage(index, undefined, undefined, String(e.message || "frame worker failed")); + worker.terminate(); + }; + worker.postMessage({ type: "init", module: self.effectcraftModule ?? null, base }); + frameWorkers.push(worker); + return index; +} + +/// Post a message ({type: "frame", json} or {type: "file", path, bytes}) to frame worker `index`. +export function frameWorkerPost(index, msg, transfer) { + const w = frameWorkers[index]; + if (w) w.postMessage(msg, transfer ?? []); +} + +// ------------------------------------------------------------------ media browser + +// Folders opened with the File System Access API (src/browse.rs): name → {handle, files: Map of +// "name/sub/file" → FileSystemFileHandle}. Their handles are remembered in IndexedDB so the +// folders come back on the next visit (once the browser grants access again). +const folders = new Map(); +const HANDLES_DB = "effectcraft-handles"; +const MAX_ENTRIES = 20000; + +async function handlesDb() { + const open = indexedDB.open(HANDLES_DB, 1); + open.onupgradeneeded = () => open.result.createObjectStore("folders", { keyPath: "name" }); + return req(open); +} + +async function enumerate(dir, prefix, out, files, depth) { + for await (const [name, h] of dir.entries()) { + if (name.startsWith(".") || out.length >= MAX_ENTRIES) continue; + const rel = `${prefix}/${name}`; + if (h.kind === "directory") { + out.push({ path: rel, dir: true, size: 0, modified: 0 }); + if (depth < 8) await enumerate(h, rel, out, files, depth + 1); + } else { + const f = await h.getFile(); + out.push({ path: rel, dir: false, size: f.size, modified: Math.floor(f.lastModified / 1000) }); + files.set(rel, h); + } + } +} + +async function openFolder(handle, remember) { + const name = handle.name; + const entries = []; + const files = new Map(); + await enumerate(handle, name, entries, files, 0); + folders.set(name, { handle, files }); + if (remember) { + try { + const db = await handlesDb(); + await req(db.transaction("folders", "readwrite").objectStore("folders").put({ name, handle })); + } catch (e) { console.info("media browser: cannot remember the folder", e); } + } + return { name, entries }; +} + +/// Whether the browser can open folders (File System Access API). +export function browseCanPickFolder() { + return typeof self.showDirectoryPicker === "function"; +} + +/// Ask for a folder; resolves with {name, entries: [{path, dir, size, modified}]}. +export async function browsePickFolder() { + const handle = await self.showDirectoryPicker({ id: "effectcraft-media", mode: "read" }); + return openFolder(handle, true); +} + +/// Folders opened on earlier visits: [{name, granted, entries?}] (entries when access is granted). +export async function browseRestore() { + if (!self.indexedDB || !browseCanPickFolder()) return []; + let all = []; + try { + const db = await handlesDb(); + all = await req(db.transaction("folders", "readonly").objectStore("folders").getAll()); + } catch { return []; } + const out = []; + for (const { name, handle } of all) { + let granted = false; + try { granted = (await handle.queryPermission({ mode: "read" })) === "granted"; } catch {} + if (granted) { + try { out.push({ granted, ...(await openFolder(handle, false)) }); continue; } catch {} + } + out.push({ name, granted: false }); + } + return out; +} + +/// Ask for access to a remembered folder again (needs a user gesture); resolves like +/// browsePickFolder. +export async function browseReconnect(name) { + const db = await handlesDb(); + const rec = await req(db.transaction("folders", "readonly").objectStore("folders").get(name)); + if (!rec) throw new Error(`${name} is not remembered`); + if ((await rec.handle.requestPermission({ mode: "read" })) !== "granted") throw new Error(`access to ${name} was not granted`); + return openFolder(rec.handle, false); +} + +/// Forget a remembered folder. +export async function browseForget(name) { + folders.delete(name); + try { + const db = await handlesDb(); + await req(db.transaction("folders", "readwrite").objectStore("folders").delete(name)); + } catch {} +} + +/// A file of an opened folder ("/Folders/name/sub/file"): its bytes. +export async function browseRead(path) { + const rel = path.replace(/^\/Folders\//, ""); + const folder = folders.get(rel.split("/")[0]); + const h = folder && folder.files.get(rel); + if (!h) throw new Error(`${path}: its folder is not open (reconnect it)`); + return new Uint8Array(await (await h.getFile()).arrayBuffer()); +} + +/// The fallback where folders can't be opened: pick files (or a whole folder, `directory`) with a +/// plain file input; resolves with [{path, bytes}] (path relative to the picked folder). +export function browsePickFiles(directory) { + return new Promise((resolve, reject) => { + const input = document.createElement("input"); + input.type = "file"; + input.multiple = true; + if (directory) input.webkitdirectory = true; + input.onchange = async () => { + try { + const out = []; + for (const f of input.files) { + out.push({ path: (directory && f.webkitRelativePath) || f.name, bytes: new Uint8Array(await f.arrayBuffer()) }); + } + resolve(out); + } catch (e) { reject(e); } + }; + input.click(); + }); +} diff --git a/apps/effectcraft-web/src/app.rs b/apps/effectcraft-web/src/app.rs index 0076cd4f..e6712e3d 100644 --- a/apps/effectcraft-web/src/app.rs +++ b/apps/effectcraft-web/src/app.rs @@ -6,7 +6,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use effectcraft_engine::Session; -use crate::{api, audio, files, persist, worker}; +use crate::{api, audio, browse, files, frames, persist, worker}; use effectcraft_ui_egui::EffectcraftApp; use serde_json::json; use wasm_bindgen::prelude::*; @@ -103,6 +103,7 @@ impl eframe::App for WebApp { persist::snapshot(&self.app.session, ctx.input(|i| i.time), 1.0); } api::set_info("gpu", json!({"compositor": self.app.gpu_adapter(), "viewerOnGpu": self.app.viewer_on_gpu()})); + api::set_info("frameWorkers", frames::stats()); } fn ui(&mut self, ui: &mut egui::Ui, frame: &mut eframe::Frame) { @@ -137,6 +138,7 @@ pub fn session() -> Session { exporter: Some(Arc::new(effectcraft_host::FileExporter { sink: Some(files::export_sink()) })), config: Some(Arc::new(persist::config())), offload: (!query_flag("noworkers")).then(|| Arc::new(worker::WorkerOffload) as Arc), + browser: Some(Arc::new(browse::WebBrowser)), ..Default::default() } } @@ -183,6 +185,7 @@ pub async fn start(canvas_id: String) -> Result<(), JsValue> { }; api::set_info("storage", json!({"backend": persist::backend(), "entries": stored, "loadMs": js_sys::Date::now() - t0})); audio::install_unlock(); + browse::restore(); let runner = eframe::WebRunner::new(); runner .start( @@ -206,6 +209,12 @@ pub async fn start(canvas_id: String) -> Result<(), JsValue> { app.offer_recovery(r); } install_hooks(&mut app); + // Viewer frames render in frame workers (`?frameworkers=N`, 0 or `?noworkers`: + // on the page's thread). + let n = if query_flag("noworkers") { 0 } else { query_value("frameworkers").and_then(|v| v.parse().ok()).unwrap_or(2usize).min(8) }; + if n > 0 { + app.frames.set_remote(Some(Arc::new(frames::WebFrames::start(n)))); + } let (tx, rx) = std::sync::mpsc::channel(); api::set_sender(tx); app = app.with_control(rx); diff --git a/apps/effectcraft-web/src/browse.rs b/apps/effectcraft-web/src/browse.rs new file mode 100644 index 00000000..aba2eb22 --- /dev/null +++ b/apps/effectcraft-web/src/browse.rs @@ -0,0 +1,277 @@ +//! The Media Browser in the browser ([`effectcraft_engine::media_browser::Browser`]). +//! +//! It browses a virtual tree: +//! +//! - **Browser Storage** (`/Browser Storage/…`): every file in the persistent file table (OPFS / +//! IndexedDB): imported media, uploaded files and folders, saved projects. Files added with +//! **Add Files…** / **Upload Folder…** (plain ``, so every browser) are copied +//! there and stay across visits: the recent list. +//! - **Folders** (`/Folders//…`): folders opened with **Open Folder…** through the File +//! System Access API (`showDirectoryPicker`, Chromium browsers). Only their listing is read; +//! a file's bytes are read when it is imported ([`Browser::fetch`]) and then kept in the file +//! table. The folder handles are remembered (IndexedDB), so after a reload the folders come +//! back once the browser grants access again (**Reconnect **). + +use std::sync::{LazyLock, Mutex}; + +use effectcraft_engine::media_browser::{Browser, Entry, VirtualTree}; +use serde_json::{Value, json}; +use wasm_bindgen::JsCast; +use wasm_bindgen::prelude::*; + +#[wasm_bindgen(module = "/js/host.js")] +extern "C" { + #[wasm_bindgen(js_name = browseCanPickFolder)] + fn can_pick_folder() -> bool; + #[wasm_bindgen(js_name = browsePickFolder)] + fn pick_folder() -> js_sys::Promise; + #[wasm_bindgen(js_name = browseRestore)] + fn restore_folders() -> js_sys::Promise; + #[wasm_bindgen(js_name = browseReconnect)] + fn reconnect(name: &str) -> js_sys::Promise; + #[wasm_bindgen(js_name = browseForget)] + fn forget(name: &str) -> js_sys::Promise; + #[wasm_bindgen(js_name = browseRead)] + fn read(path: &str) -> js_sys::Promise; + #[wasm_bindgen(js_name = browsePickFiles)] + fn pick_files(directory: bool) -> js_sys::Promise; +} + +const STORAGE: &str = "/Browser Storage"; +const FOLDERS: &str = "/Folders"; + +#[derive(Default)] +struct State { + /// Opened folders' files (`/Folders/…`). + folders: VirtualTree, + /// Opened folder names, and remembered ones waiting for access. + open: Vec, + disconnected: Vec, +} + +static STATE: LazyLock> = LazyLock::new(Mutex::default); + +fn state() -> std::sync::MutexGuard<'static, State> { + STATE.lock().unwrap_or_else(|e| e.into_inner()) +} + +/// The web app's Media Browser. +pub struct WebBrowser; + +/// The stored files as a tree under `/Browser Storage` (rebuilt per listing: storage changes). +fn storage_tree() -> VirtualTree { + let mut t = VirtualTree::default(); + t.add_dir(STORAGE); + for (path, size, modified, persisted) in crate::files::STORE.files() { + if !persisted || path.starts_with(FOLDERS) { + continue; + } + t.add_file(&format!("{STORAGE}{path}"), &path, size as u64, Some((modified / 1000.0) as u64)); + } + t +} + +/// Add an opened folder's listing (`[{path, dir, size, modified}]`, paths `name/…`). +fn add_folder(v: &Value) -> Option { + let name = v["name"].as_str()?.to_string(); + let mut st = state(); + st.folders.remove(&format!("{FOLDERS}/{name}")); + st.folders.add_dir(&format!("{FOLDERS}/{name}")); + for e in v["entries"].as_array().into_iter().flatten() { + let Some(rel) = e["path"].as_str() else { continue }; + let path = format!("{FOLDERS}/{rel}"); + if e["dir"].as_bool() == Some(true) { + st.folders.add_dir(&path); + } else { + st.folders.add_file(&path, &path, e["size"].as_u64().unwrap_or(0), e["modified"].as_u64()); + } + } + st.open.retain(|n| *n != name); + st.open.push(name.clone()); + st.disconnected.retain(|n| *n != name); + Some(name) +} + +fn to_value(v: &JsValue) -> Value { + js_sys::JSON::stringify(v).ok().and_then(|s| s.as_string()).and_then(|s| serde_json::from_str(&s).ok()).unwrap_or(Value::Null) +} + +/// Run a promise; on success, `then` gets its value (as JSON) on the UI thread. +fn spawn(p: js_sys::Promise, what: &'static str, then: impl FnOnce(&mut effectcraft_ui_egui::EffectcraftApp, Value) + 'static) { + wasm_bindgen_futures::spawn_local(async move { + match wasm_bindgen_futures::JsFuture::from(p).await { + Ok(v) => { + let v = to_value(&v); + crate::post(move |app| then(app, v)); + } + Err(e) => { + let msg = e.as_string().or_else(|| js_sys::Reflect::get(&e, &"message".into()).ok().and_then(|m| m.as_string())).unwrap_or_default(); + // Cancelling a picker is not an error. + if !msg.contains("abort") && !msg.contains("AbortError") { + crate::post(move |app| app.ui.status = format!("{what}: {msg}")); + } + } + } + }); +} + +fn go(app: &mut effectcraft_ui_egui::EffectcraftApp, path: &str) { + if let Err(e) = app.session.execute("mediaBrowser.go", json!({"path": path})) { + app.ui.status = e.to_string(); + } +} + +/// Bring back the folders opened on earlier visits (those the browser still grants access to). +pub fn restore() { + spawn(restore_folders(), "Media Browser", |_, v| { + for f in v.as_array().into_iter().flatten() { + if f["granted"].as_bool() == Some(true) { + add_folder(f); + } else if let Some(n) = f["name"].as_str() { + let mut st = state(); + if !st.disconnected.iter().any(|d| d == n) { + st.disconnected.push(n.to_string()); + } + } + } + }); +} + +impl Browser for WebBrowser { + fn home(&self) -> String { + STORAGE.into() + } + + fn places(&self) -> Vec<(String, String)> { + let mut v = vec![("Browser Storage".to_string(), STORAGE.to_string())]; + v.extend(state().open.iter().map(|n| (n.clone(), format!("{FOLDERS}/{n}")))); + v + } + + fn list(&self, dir: &str, importable_only: bool) -> Result, String> { + let dir = if dir.len() > 1 { dir.trim_end_matches('/') } else { dir }; + if dir == STORAGE || dir.starts_with(&format!("{STORAGE}/")) { + return storage_tree().list(dir, importable_only); + } + if dir == "/" { + let mut t = VirtualTree::default(); + t.add_dir(STORAGE); + t.add_dir(FOLDERS); + return t.list("/", false); + } + let st = state(); + if dir == FOLDERS && st.open.is_empty() { + return Ok(vec![]); + } + st.folders.list(dir, importable_only).map_err(|e| { + let name = dir.trim_start_matches(FOLDERS).trim_start_matches('/').split('/').next().unwrap_or(""); + if st.disconnected.iter().any(|d| d == name) { format!("{name}: click Reconnect {name} to allow access again") } else { e } + }) + } + + fn parent(&self, dir: &str) -> Option { + let d = dir.trim_end_matches('/'); + match d.rfind('/') { + Some(0) if d.len() > 1 => Some("/".into()), + Some(i) if i > 0 => Some(d[..i].to_string()), + _ => None, + } + } + + fn fetch(&self, paths: &[String], params: &Value) -> bool { + let missing: Vec = paths.iter().filter(|p| p.starts_with(&format!("{FOLDERS}/")) && crate::files::get(p).is_none()).cloned().collect(); + if missing.is_empty() { + return true; + } + let params = params.clone(); + wasm_bindgen_futures::spawn_local(async move { + for p in &missing { + match wasm_bindgen_futures::JsFuture::from(read(p)).await { + Ok(b) => crate::files::put(p, js_sys::Uint8Array::new(&b).to_vec().into()), + Err(e) => { + let msg = format!("{p}: {}", e.as_string().unwrap_or_else(|| "cannot read the file (reconnect its folder)".into())); + crate::post(move |app| app.ui.status = msg); + return; + } + } + } + crate::post(move |app| match app.session.execute("mediaBrowser.import", params) { + Ok(r) => { + if let Some(errs) = r["errors"].as_array().filter(|e| !e.is_empty()) { + app.ui.status = errs.iter().filter_map(|e| e.as_str()).collect::>().join("; "); + } + } + Err(e) => app.ui.status = e.to_string(), + }); + }); + false + } + + fn actions(&self) -> Vec<(String, String)> { + let mut v = vec![]; + if can_pick_folder() { + v.push(("openFolder".to_string(), "Open Folder…".to_string())); + } + v.push(("addFiles".into(), "Add Files…".into())); + v.push(("uploadFolder".into(), "Upload Folder…".into())); + for n in &state().disconnected { + v.push((format!("reconnect:{n}"), format!("Reconnect {n}"))); + } + v + } + + fn action(&self, id: &str) -> Result { + match id { + "openFolder" => { + if !can_pick_folder() { + return Err("this browser cannot open folders (use Add Files… or Upload Folder…)".into()); + } + spawn(pick_folder(), "Open Folder", |app, v| { + if let Some(n) = add_folder(&v) { + go(app, &format!("{FOLDERS}/{n}")); + } + }); + Ok(json!({"pending": true})) + } + "addFiles" | "uploadFolder" => { + let p = pick_files(id == "uploadFolder"); + wasm_bindgen_futures::spawn_local(async move { + let Ok(list) = wasm_bindgen_futures::JsFuture::from(p).await else { return }; + let mut n = 0; + for f in js_sys::Array::from(&list).iter() { + let path = js_sys::Reflect::get(&f, &"path".into()).ok().and_then(|v| v.as_string()); + let bytes = js_sys::Reflect::get(&f, &"bytes".into()).ok().and_then(|b| b.dyn_into::().ok()); + if let (Some(path), Some(bytes)) = (path, bytes) { + crate::files::put(&format!("/{}", path.trim_start_matches('/')), bytes.to_vec().into()); + n += 1; + } + } + crate::post(move |app| { + app.ui.status = format!("{n} file(s) added to Browser Storage"); + go(app, STORAGE); + }); + }); + Ok(json!({"pending": true})) + } + _ => { + if let Some(name) = id.strip_prefix("reconnect:") { + spawn(reconnect(name), "Reconnect", |app, v| { + if let Some(n) = add_folder(&v) { + go(app, &format!("{FOLDERS}/{n}")); + } + }); + return Ok(json!({"pending": true})); + } + if let Some(name) = id.strip_prefix("forget:") { + let _ = forget(name); + let mut st = state(); + st.open.retain(|n| n != name); + st.disconnected.retain(|n| n != name); + st.folders.remove(&format!("{FOLDERS}/{name}")); + return Ok(json!({"forgotten": name})); + } + Err(format!("unknown Media Browser action `{id}`")) + } + } + } +} diff --git a/apps/effectcraft-web/src/frames.rs b/apps/effectcraft-web/src/frames.rs new file mode 100644 index 00000000..abad352f --- /dev/null +++ b/apps/effectcraft-web/src/frames.rs @@ -0,0 +1,261 @@ +//! Viewer frames rendered in Web Workers ([`effectcraft_engine::remote`]). +//! +//! **Page side** — [`WebFrames`], the frame cache's [`RemoteFrames`]: a few long-lived frame +//! workers (`js/host.js` `frameWorkerStart`), each holding a replica of the project. Before a +//! render, the worker gets the footage files it lacks and a sync (the whole project the first +//! time, then diffs: [`Mirror`]); the frame comes back as premultiplied RGBA8 in a transferred +//! buffer. One frame per worker at a time, so the page always decides what renders next (the +//! viewer's frame before prefetch). A worker that fails is dropped; with none left, frames render +//! on the page's thread again (`Frames::pump`). +//! +//! **Worker side** — [`worker_frame`]: the worker's [`FrameServer`] (footage from the worker's +//! media pool, expressions) handles each message and posts its replies. + +use std::cell::RefCell; +use std::collections::HashSet; + +use effectcraft_engine::offload::footage_files; +use effectcraft_engine::remote::{FrameMsg, FrameReply, FrameServer, Mirror}; +use effectcraft_ui_egui::frames::{RemoteDone, RemoteFrame, RemoteFrames, RemoteJob}; +use serde_json::{Value, json}; +use wasm_bindgen::JsCast; +use wasm_bindgen::prelude::*; + +#[wasm_bindgen(module = "/js/host.js")] +extern "C" { + #[wasm_bindgen(js_name = frameWorkerStart)] + fn frame_worker_start(base: &str, on_message: &js_sys::Function) -> u32; + #[wasm_bindgen(js_name = frameWorkerPost)] + fn frame_worker_post(index: u32, msg: &JsValue, transfer: &js_sys::Array); +} + +/// One frame worker as the page sees it. +struct Fw { + index: u32, + mirror: Mirror, + files: HashSet, + busy: Option<(u64, RemoteDone)>, + alive: bool, +} + +#[derive(Default)] +struct State { + workers: Vec, + next_id: u64, + rendered: u64, + failed: u64, + last_ms: f64, + /// Bytes of project sync messages sent (diffs keep this small). + sync_bytes: u64, + syncs: u64, +} + +type OnMessage = Closure; + +thread_local! { + static STATE: RefCell = RefCell::new(State::default()); + static HANDLER: RefCell> = const { RefCell::new(None) }; + /// The worker's frame server (worker side). + static SERVER: RefCell> = const { RefCell::new(None) }; +} + +fn obj(pairs: &[(&str, JsValue)]) -> JsValue { + let o = js_sys::Object::new(); + for (k, v) in pairs { + let _ = js_sys::Reflect::set(&o, &JsValue::from_str(k), v); + } + o.into() +} + +/// The page's frame workers (see the module docs). +pub struct WebFrames; + +impl WebFrames { + /// Start `n` frame workers. + pub fn start(n: usize) -> WebFrames { + let on: OnMessage = Closure::new(|index: u32, json: JsValue, bytes: JsValue, error: JsValue| on_message(index, json, bytes, error)); + STATE.with(|s| { + let mut s = s.borrow_mut(); + for _ in 0..n { + let index = frame_worker_start(&crate::page_url(""), on.as_ref().unchecked_ref()); + s.workers.push(Fw { index, mirror: Mirror::default(), files: HashSet::new(), busy: None, alive: true }); + } + }); + HANDLER.with(|h| *h.borrow_mut() = Some(on)); + WebFrames + } +} + +impl RemoteFrames for WebFrames { + fn slots(&self) -> usize { + STATE.with(|s| s.borrow().workers.iter().filter(|w| w.alive).count()) + } + + fn purge(&self) { + purge(); + } + + fn start(&self, job: RemoteJob, done: RemoteDone) { + let r = STATE.with(|s| { + let mut s = s.borrow_mut(); + s.next_id += 1; + let id = s.next_id; + let Some(k) = s.workers.iter().position(|w| w.alive && w.busy.is_none()) else { return Err((done, "no idle frame worker".to_string())) }; + // 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) { + 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 sync = w.mirror.sync(job.revision, &job.project); + let index = w.index; + let mut sent = 0; + if let Some(m) = sync { + let t = serde_json::to_string(&m).unwrap_or_default(); + sent = t.len() as u64; + frame_worker_post(index, &obj(&[("type", "frame".into()), ("json", t.into())]), &js_sys::Array::new()); + } + let req = FrameMsg::Render { id, revision: job.revision, comp: job.comp, time: job.t, opts: Box::new(job.opts) }; + frame_worker_post( + index, + &obj(&[("type", "frame".into()), ("json", serde_json::to_string(&req).unwrap_or_default().into())]), + &js_sys::Array::new(), + ); + s.workers[k].busy = Some((id, done)); + if sent > 0 { + s.syncs += 1; + s.sync_bytes += sent; + } + Ok(()) + }); + if let Err((done, e)) = r { + done(Err(e)); + } + } +} + +/// A message from frame worker `index`: a JSON [`FrameReply`] (with the pixels in `bytes`), or +/// `error` when the worker died. +fn on_message(index: u32, json: JsValue, bytes: JsValue, error: JsValue) { + let done = STATE.with(|s| { + let mut s = s.borrow_mut(); + let st = &mut *s; + let w = st.workers.iter_mut().find(|w| w.index == index)?; + if let Some(e) = error.as_string() { + log::warn!("frame worker {index}: {e}; frames render on the page's thread if none is left"); + w.alive = false; + return w.busy.take().map(|(_, d)| (d, Err(e))); + } + let reply: FrameReply = serde_json::from_str(&json.as_string()?).ok()?; + match reply { + FrameReply::Frame { id, width, height, ms } => { + let (_, d) = w.busy.take_if(|b| b.0 == id)?; + let rgba = bytes.dyn_into::().map(|b| b.to_vec()).unwrap_or_default(); + st.rendered += 1; + st.last_ms = ms; + Some((d, Ok(RemoteFrame { width, height, rgba, ms }))) + } + FrameReply::Failed { id, error, .. } => { + st.failed += 1; + let (_, d) = w.busy.take_if(|b| b.0 == id)?; + Some((d, Err(error))) + } + FrameReply::Resync { error } => { + log::info!("frame worker {index}: {error}: sending the whole project again"); + w.mirror.reset(); + None + } + } + }); + if let Some((d, r)) = done { + d(r); + } + crate::repaint(); +} + +/// Relay Roto Brush segmentations (from a propagation worker) to the frame workers, so their +/// renders use them instead of computing them again. +pub fn relay_segs(segs: &[effectcraft_engine::offload::SegData]) { + if segs.is_empty() { + return; + } + let t = serde_json::to_string(&FrameMsg::Segs { segs: segs.to_vec() }).unwrap_or_default(); + STATE.with(|s| { + for w in s.borrow().workers.iter().filter(|w| w.alive) { + frame_worker_post(w.index, &obj(&[("type", "frame".into()), ("json", t.as_str().into())]), &js_sys::Array::new()); + } + }); +} + +/// Ask every frame worker to drop its cached layer buffers (Edit ▸ Purge). +pub fn purge() { + let t = serde_json::to_string(&FrameMsg::Purge).unwrap_or_default(); + STATE.with(|s| { + for w in s.borrow().workers.iter().filter(|w| w.alive) { + frame_worker_post(w.index, &obj(&[("type", "frame".into()), ("json", t.as_str().into())]), &js_sys::Array::new()); + } + }); +} + +/// `effectcraft.info().frameWorkers`. +pub fn stats() -> Value { + STATE.with(|s| { + let s = s.borrow(); + json!({ + "workers": s.workers.len(), + "alive": s.workers.iter().filter(|w| w.alive).count(), + "busy": s.workers.iter().filter(|w| w.busy.is_some()).count(), + "rendered": s.rendered, + "failed": s.failed, + "lastMs": s.last_ms, + "syncs": s.syncs, + "syncBytes": s.sync_bytes, + }) + }) +} + +// ---------------------------------------------------------------- worker side + +/// Handle one frame message (a JSON [`FrameMsg`]) in a frame worker; replies are posted with +/// their pixels transferred. +#[wasm_bindgen(js_name = workerFrame)] +pub fn worker_frame(json: String) { + let msg: FrameMsg = match serde_json::from_str(&json) { + Ok(m) => m, + Err(e) => { + log::warn!("frame worker: bad message: {e}"); + return; + } + }; + let replies = SERVER.with(|c| { + let mut c = c.borrow_mut(); + let server = c.get_or_insert_with(|| { + let pool = crate::worker::pool(); + FrameServer::new(pool, Some(std::sync::Arc::new(effectcraft_expr::Expressions))) + }); + server.handle(msg) + }); + let scope: web_sys::DedicatedWorkerGlobalScope = js_sys::global().unchecked_into(); + for (r, px) in replies { + let t = serde_json::to_string(&r).unwrap_or_default(); + let res = match px { + Some(px) => { + let buf = js_sys::Uint8Array::from(&px[..]).buffer(); + scope.post_message_with_transfer( + &obj(&[("type", "frameReply".into()), ("json", t.into()), ("bytes", buf.clone().into())]), + &js_sys::Array::of1(&buf), + ) + } + None => scope.post_message(&obj(&[("type", "frameReply".into()), ("json", t.into())])), + }; + if let Err(e) = res { + log::warn!("frame worker: postMessage: {e:?}"); + } + } +} diff --git a/apps/effectcraft-web/src/lib.rs b/apps/effectcraft-web/src/lib.rs index 59de4433..96b17db2 100644 --- a/apps/effectcraft-web/src/lib.rs +++ b/apps/effectcraft-web/src/lib.rs @@ -9,8 +9,9 @@ //! | file system (`FsServices`, media reads, export writes) | a virtual file table ([`files`]) persisted to the Origin Private File System / IndexedDB ([`persist`], [`store`]) | //! | config directory (settings, shortcuts, recent projects, auto-saves) | the same store ([`store::WebConfig`]) | //! | rfd file dialogs | `` pickers; drop files on the page | +//! | Media Browser on the file system | browser storage and folders opened with the File System Access API ([`browse`]) | //! | written files (Save, Render Queue) | browser downloads (several render files as one `.zip`) | -//! | frame render threads | `Frames::pump` on the UI thread between egui frames | +//! | frame render threads | frame workers, each with a project replica fed by diffs ([`frames`]); `Frames::pump` on the UI thread without workers | //! | Render Queue / analysis threads | Web Workers running their own engine instance ([`worker`]) | //! | cpal audio output | Web Audio: an AudioWorklet fed from the preview mixdown ([`audio`]) | //! | TCP control channel / MCP | `window.effectcraft` JavaScript API ([`api`]) | @@ -26,8 +27,12 @@ mod app; #[cfg(target_arch = "wasm32")] pub mod audio; #[cfg(target_arch = "wasm32")] +pub mod browse; +#[cfg(target_arch = "wasm32")] pub mod files; #[cfg(target_arch = "wasm32")] +pub mod frames; +#[cfg(target_arch = "wasm32")] pub mod persist; #[cfg(target_arch = "wasm32")] pub mod worker; diff --git a/apps/effectcraft-web/src/worker.rs b/apps/effectcraft-web/src/worker.rs index 2d96f641..07c9aadc 100644 --- a/apps/effectcraft-web/src/worker.rs +++ b/apps/effectcraft-web/src/worker.rs @@ -5,6 +5,9 @@ //! `workerRun`), and turns the worker's messages back into [`WorkerReply`]s in the job's //! [`Inbox`]. Rendered files come back as their bytes and download like any render. //! +//! Viewer frames use separate, long-lived workers ([`crate::frames`]); Roto Brush +//! segmentations a propagation job sends back are relayed to them. +//! //! **Worker side** — `web/worker.js` instantiates the same wasm module (the page's compiled //! `WebAssembly.Module`, so nothing is compiled twice) and calls [`worker_init`], then //! [`worker_file`] per footage file and [`worker_job`] per job: a plain engine session (footage @@ -60,6 +63,9 @@ impl Offload for WorkerOffload { let Some(t) = json.as_string() else { return }; match serde_json::from_str::(&t) { Ok(r) => { + if let WorkerReply::Segs { segs } = &r { + crate::frames::relay_segs(segs); + } let done = matches!(r, WorkerReply::Done); inbox.push(r); if done { @@ -146,6 +152,11 @@ pub fn worker_init() { SESSION.with(|c| *c.borrow_mut() = Some(s)); } +/// The worker's media pool (footage files sent by the page). +pub fn pool() -> Arc { + POOL.with(|p| p.borrow().clone()).unwrap_or_else(|| Arc::new(effectcraft_media::MediaPool::new())) +} + /// A footage file for the jobs that follow. #[wasm_bindgen(js_name = workerFile)] pub fn worker_file(path: String, bytes: js_sys::Uint8Array) { diff --git a/apps/effectcraft-web/web/worker.js b/apps/effectcraft-web/web/worker.js index c29663d7..6c6853c5 100644 --- a/apps/effectcraft-web/web/worker.js +++ b/apps/effectcraft-web/web/worker.js @@ -1,7 +1,9 @@ -// An EffectCraft engine instance in a Web Worker: renders the Render Queue and runs analyses -// off the page's thread (src/worker.rs, crates/engine/src/offload.rs). The page sends the -// compiled module, then footage files and jobs; the worker replies with JSON messages and -// rendered files. +// An EffectCraft engine instance in a Web Worker (src/worker.rs, src/frames.rs, +// crates/engine/src/offload.rs and remote.rs). The page sends the compiled module, then: +// - job workers: footage files and jobs (Render Queue, analyses, Roto Brush); the worker replies +// with JSON messages and rendered files; +// - frame workers: footage files and frame messages (project syncs as diffs, render requests, +// Roto Brush segmentations); the worker replies with frames (pixels transferred). import init, * as ec from "./effectcraft_web.js"; let ready = null; @@ -16,4 +18,5 @@ self.onmessage = async (e) => { await ready; if (m.type === "file") ec.workerFile(m.path, m.bytes); else if (m.type === "job") ec.workerJob(m.json); + else if (m.type === "frame") ec.workerFrame(m.json); }; diff --git a/crates/engine/src/commands/panels_cmds.rs b/crates/engine/src/commands/panels_cmds.rs index eb056878..325e0da0 100644 --- a/crates/engine/src/commands/panels_cmds.rs +++ b/crates/engine/src/commands/panels_cmds.rs @@ -29,38 +29,61 @@ fn jobs_wait(s: &mut Session, _: &Value) -> Result { // ---------------------------------------------------------------- Media Browser -fn browser_dir(s: &Session, p: &Value) -> String { - str_p(p, "path").map(str::to_string).or_else(|| s.state.media_browser.folder.clone()).unwrap_or_else(mb::home) +fn browser(s: &Session) -> Result> { + s.media_browser().ok_or_else(|| EngineError::Other("the Media Browser is not available in this build".into())) +} + +fn browser_dir(s: &Session, b: &dyn mb::Browser, p: &Value) -> String { + str_p(p, "path").map(str::to_string).or_else(|| s.state.media_browser.folder.clone()).unwrap_or_else(|| b.home()) } fn browser_list(s: &mut Session, p: &Value) -> Result { - let dir = browser_dir(s, p); + let b = browser(s)?; + let dir = browser_dir(s, b.as_ref(), p); let only = b_p(p, "importableOnly").unwrap_or(s.state.media_browser.importable_only); - let entries = mb::list(&dir, only).map_err(EngineError::Other)?; - Ok(json!({"path": dir, "parent": mb::parent(&dir), "entries": entries, "favorites": s.state.media_browser.favorites})) + let entries = b.list(&dir, only).map_err(EngineError::Other)?; + let places: Vec = b.places().into_iter().map(|(name, path)| json!({"name": name, "path": path})).collect(); + let actions: Vec = b.actions().into_iter().map(|(id, label)| json!({"id": id, "label": label})).collect(); + Ok(json!({ + "path": dir, + "parent": b.parent(&dir), + "entries": entries, + "favorites": s.state.media_browser.favorites, + "places": places, + "actions": actions, + })) } -fn browser_enabled(_: &Session) -> std::result::Result<(), String> { - if mb::available() { Ok(()) } else { Err("the Media Browser needs the desktop app".into()) } +fn browser_enabled(s: &Session) -> std::result::Result<(), String> { + if s.media_browser().is_some() { Ok(()) } else { Err("the Media Browser is not available in this build".into()) } } fn browser_go(s: &mut Session, p: &Value) -> Result { + let b = browser(s)?; let dir = match str_p(p, "path") { - Some("..") => s.state.media_browser.folder.clone().and_then(|d| mb::parent(&d)).unwrap_or_else(mb::home), + Some("..") => s.state.media_browser.folder.clone().and_then(|d| b.parent(&d)).unwrap_or_else(|| b.home()), Some(d) => d.to_string(), - None => mb::home(), + None => b.home(), }; - if let Some(b) = b_p(p, "importableOnly") { - s.state.media_browser.importable_only = b; + if let Some(v) = b_p(p, "importableOnly") { + s.state.media_browser.importable_only = v; } // Validate before switching. - mb::list(&dir, false).map_err(EngineError::Other)?; + b.list(&dir, false).map_err(EngineError::Other)?; s.state.media_browser.folder = Some(dir.clone()); browser_list(s, &json!({"path": dir})) } +/// A browser-specific toolbar action (web: Open Folder…, Add Files…). +fn browser_action(s: &mut Session, p: &Value) -> Result { + let b = browser(s)?; + let id = str_p(p, "action").ok_or_else(|| bad("mediaBrowser.action", "missing `action`"))?; + b.action(id).map_err(|e| bad("mediaBrowser.action", e)) +} + fn browser_favorite(s: &mut Session, p: &Value, add: bool) -> Result { - let dir = browser_dir(s, p); + let b = browser(s)?; + let dir = browser_dir(s, b.as_ref(), p); let favs = &mut s.state.media_browser.favorites; favs.retain(|f| f != &dir); if add { @@ -71,6 +94,16 @@ fn browser_favorite(s: &mut Session, p: &Value, add: bool) -> Result { fn browser_import(s: &mut Session, p: &Value) -> Result { let paths = p.get("paths").or(p.get("path")).cloned().ok_or_else(|| bad("mediaBrowser.import", "missing `paths`"))?; + // Files the browser still has to read (the web app's opened folders): imported when they + // arrive. + let list: Vec = match &paths { + Value::String(one) => vec![one.clone()], + Value::Array(a) => a.iter().filter_map(|v| v.as_str().map(str::to_string)).collect(), + _ => vec![], + }; + if !browser(s)?.fetch(&list, p) { + return Ok(json!({"pending": true, "paths": list})); + } let r = s.execute("file.import", json!({"paths": paths}))?; // Add to the active comp (drag into the timeline). if b_p(p, "addToComp").unwrap_or(false) @@ -201,6 +234,15 @@ pub fn specs() -> Vec { cmd!("mediaBrowser.addFavorite", "Add to Favorites", [], None, "{path?}", browser_enabled, |s, p| browser_favorite(s, p, true)), cmd!("mediaBrowser.removeFavorite", "Remove from Favorites", [], None, "{path?}", browser_enabled, |s, p| browser_favorite(s, p, false)), cmd!("mediaBrowser.import", "Import", [], None, "{paths, addToComp?}", browser_enabled, browser_import), + cmd!( + "mediaBrowser.action", + "Media Browser Action", + [], + None, + "{action}: an action from mediaBrowser.list's `actions` (web: openFolder, addFiles, uploadFolder)", + browser_enabled, + browser_action + ), query!("mediaBrowser.fileInfo", "File Info", "{path}", file_info), query!("item.metadata", "Metadata", "{item?: id|name}", metadata), cmd!("project.setProjectComment", "Project Comment", [], None, "{comment}", always, project_comment), diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs index 64733d44..e88579c4 100644 --- a/crates/engine/src/lib.rs +++ b/crates/engine/src/lib.rs @@ -26,6 +26,7 @@ pub mod offload; pub mod prefs; pub mod preview; pub mod psd_import; +pub mod remote; pub mod render_queue; pub mod roto; pub mod scriptui; @@ -349,6 +350,11 @@ pub struct Session { pub offload: Option>, /// Analyses running in the [`Session::offload`] worker. pub offloaded: Vec, + /// What the Media Browser browses (`None`: the local file system where there is one). + pub browser: Option>, + /// Offloaded jobs that ended, newest last (kind, error): how a caller waiting for one + /// (`wait: true` over the control channel in the browser) learns its outcome. + pub offload_log: Vec<(offload::JobKind, Option)>, /// Generic background tasks (Content-Aware Fill, Scene Edit Detection), see [`jobs`]. pub tasks: Vec, /// Finished tasks, newest last (Progress panel, `jobs.list`). @@ -421,6 +427,8 @@ impl Default for Session { applied_nested_switches: None, offload: None, offloaded: vec![], + offload_log: vec![], + browser: None, tasks: vec![], job_log: vec![], next_task_id: 1, diff --git a/crates/engine/src/media_browser.rs b/crates/engine/src/media_browser.rs index afc7dc34..9b0a01da 100644 --- a/crates/engine/src/media_browser.rs +++ b/crates/engine/src/media_browser.rs @@ -1,10 +1,16 @@ -//! Media Browser (Window ▸ Media Browser): browse the local file system for footage, keep -//! favourite folders, read file metadata, and import (`file.import`). Desktop only: the web build -//! has no file system to browse ([`available`] is false there). +//! Media Browser (Window ▸ Media Browser): browse for footage, keep favourite folders, read file +//! metadata, and import (`file.import`). +//! +//! What it browses is a [`Browser`]: the local file system on the desktop ([`FsBrowser`]); in the +//! web app, a virtual tree ([`VirtualTree`]) of the browser's storage and of folders the user +//! opened through the File System Access API, whose files are read only when imported +//! ([`Browser::fetch`]). //! //! Also the metadata the Metadata panel shows for project items and files (codec, size, rate, //! duration, colour profile, dates). +use std::collections::BTreeMap; + use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; @@ -25,11 +31,153 @@ pub struct BrowserState { pub importable_only: bool, } -/// Whether this build can browse files. +/// Whether this build can browse the local file system ([`FsBrowser`]). pub fn available() -> bool { !cfg!(target_arch = "wasm32") } +/// Where the Media Browser browses. +pub trait Browser: Send + Sync { + /// The folder shown first. + fn home(&self) -> String; + /// Places listed above the favourites: (name, folder). The default is Home. + fn places(&self) -> Vec<(String, String)> { + vec![("Home".into(), self.home())] + } + /// A folder's entries: folders first, then files, by name. + fn list(&self, dir: &str, importable_only: bool) -> Result, String>; + /// The folder above `dir`. + fn parent(&self, dir: &str) -> Option { + parent(dir) + } + /// Make `paths` readable by the importer. `false`: they are being fetched, and the host runs + /// `mediaBrowser.import` with `params` again once they are (the web app reads opened folders' + /// files on demand). + fn fetch(&self, paths: &[String], params: &Value) -> bool { + let _ = (paths, params); + true + } + /// Extra toolbar actions: (id, label), run by `mediaBrowser.action`. + fn actions(&self) -> Vec<(String, String)> { + vec![] + } + fn action(&self, id: &str) -> Result { + Err(format!("unknown Media Browser action `{id}`")) + } +} + +/// The local file system (desktop). +pub struct FsBrowser; + +impl Browser for FsBrowser { + fn home(&self) -> String { + home() + } + fn list(&self, dir: &str, importable_only: bool) -> Result, String> { + list(dir, importable_only) + } +} + +impl crate::Session { + /// What the Media Browser browses: [`crate::Session::browser`], else the local file system + /// where this build has one. + pub fn media_browser(&self) -> Option> { + self.browser.clone().or_else(|| available().then(|| std::sync::Arc::new(FsBrowser) as std::sync::Arc)) + } +} + +/// One entry of a [`VirtualTree`]. +#[derive(Clone, Debug, PartialEq)] +struct VNode { + /// What importing it reads (`None`: a folder). + import: Option, + size: u64, + modified: Option, +} + +/// A browsable tree of virtual paths (`/Folder/sub/file.mov`), each file standing for an +/// importable path (the web app's file table, or a file of an opened folder). +#[derive(Clone, Debug, Default)] +pub struct VirtualTree { + nodes: BTreeMap, +} + +fn vparent(path: &str) -> Option<&str> { + if path == "/" || path.is_empty() { + return None; + } + let p = path.trim_end_matches('/'); + match p.rfind('/') { + Some(0) => Some("/"), + Some(i) => Some(&p[..i]), + None => None, + } +} + +impl VirtualTree { + /// Add a folder (and the folders above it). + pub fn add_dir(&mut self, path: &str) { + let mut cur = Some(path.trim_end_matches('/')); + while let Some(p) = cur.filter(|p| !p.is_empty() && *p != "/") { + self.nodes.entry(p.to_string()).or_insert(VNode { import: None, size: 0, modified: None }); + cur = vparent(p); + } + } + + /// Add a file shown at `path` that imports `import`. + pub fn add_file(&mut self, path: &str, import: &str, size: u64, modified: Option) { + if let Some(p) = vparent(path) { + self.add_dir(p); + } + self.nodes.insert(path.to_string(), VNode { import: Some(import.to_string()), size, modified }); + } + + /// Remove `path` and everything under it. + pub fn remove(&mut self, path: &str) { + let pre = format!("{}/", path.trim_end_matches('/')); + self.nodes.retain(|k, _| k != path && !k.starts_with(&pre)); + } + + pub fn is_dir(&self, path: &str) -> bool { + path == "/" || self.nodes.get(path.trim_end_matches('/')).is_some_and(|n| n.import.is_none()) + } + + /// What a file shown at `path` imports. + pub fn import_path(&self, path: &str) -> Option<&str> { + self.nodes.get(path).and_then(|n| n.import.as_deref()) + } + + pub fn list(&self, dir: &str, importable_only: bool) -> Result, String> { + let dir = if dir.len() > 1 { dir.trim_end_matches('/') } else { dir }; + if !self.is_dir(dir) { + return Err(format!("{dir}: no such folder")); + } + let mut out: Vec = self + .nodes + .iter() + .filter(|(k, _)| vparent(k) == Some(dir)) + .filter_map(|(k, n)| { + let name = k.rsplit('/').next().unwrap_or(k).to_string(); + let kind = n.import.as_deref().and_then(kind_of); + if importable_only && n.import.is_some() && kind.is_none() { + return None; + } + Some(Entry { + name, + // Folders browse by their virtual path; files import their own path. + path: n.import.clone().unwrap_or_else(|| k.clone()), + is_dir: n.import.is_none(), + size: n.size, + modified: n.modified, + kind, + }) + }) + .collect(); + out.sort_by(|a, b| b.is_dir.cmp(&a.is_dir).then_with(|| a.name.to_lowercase().cmp(&b.name.to_lowercase()))); + Ok(out) + } +} + /// One directory entry. #[derive(Clone, Debug, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] @@ -261,6 +409,32 @@ mod tests { std::fs::remove_dir_all(&dir).unwrap(); } + #[test] + fn virtual_tree_lists_like_a_folder() { + let mut t = VirtualTree::default(); + t.add_file("/Browser Storage/clip.mov", "/clip.mov", 10, Some(5)); + t.add_file("/Folders/Shoot/B/b.png", "/Folders/Shoot/B/b.png", 3, None); + t.add_file("/Folders/Shoot/a.wav", "/Folders/Shoot/a.wav", 4, None); + t.add_file("/Folders/Shoot/notes.txt", "/Folders/Shoot/notes.txt", 1, None); + t.add_dir("/Folders/Empty"); + let names = |t: &VirtualTree, d: &str, only: bool| t.list(d, only).unwrap().into_iter().map(|e| e.name).collect::>(); + assert_eq!(names(&t, "/", false), ["Browser Storage", "Folders"]); + assert_eq!(names(&t, "/Folders", false), ["Empty", "Shoot"]); + assert_eq!(names(&t, "/Folders/Shoot", false), ["B", "a.wav", "notes.txt"]); + assert_eq!(names(&t, "/Folders/Shoot/", true), ["B", "a.wav"]); + let e = &t.list("/Browser Storage", false).unwrap()[0]; + assert_eq!((e.path.as_str(), e.size, e.modified, e.kind, e.is_dir), ("/clip.mov", 10, Some(5), Some("video"), false)); + let d = &t.list("/Folders", false).unwrap()[1]; + assert_eq!((d.path.as_str(), d.is_dir), ("/Folders/Shoot", true)); + assert_eq!(t.import_path("/Browser Storage/clip.mov"), Some("/clip.mov")); + assert!(t.list("/Folders/Shoot/a.wav", false).is_err(), "a file is not a folder"); + assert!(t.list("/Nope", false).is_err()); + t.remove("/Folders/Shoot"); + assert_eq!(names(&t, "/Folders", false), ["Empty"]); + assert_eq!(vparent("/a"), Some("/")); + assert_eq!(vparent("/"), None); + } + #[test] fn iso_dates() { assert_eq!(iso_date(0), "1970-01-01T00:00:00Z"); diff --git a/crates/engine/src/offload.rs b/crates/engine/src/offload.rs index 3e92fbd4..712549bd 100644 --- a/crates/engine/src/offload.rs +++ b/crates/engine/src/offload.rs @@ -16,10 +16,11 @@ //! | Track Motion / Stabilize Motion | the tracker analysis | the tracker group | //! | Mask tracking | the mask track | the mask group | //! | Roto Brush Freeze | the freeze | the effect group | +//! | Roto Brush propagation | the propagation | segmentations ([`WorkerReply::Segs`], streamed while it runs) go into the page's segmentation cache | //! | 3D Camera Tracker analysis | [`Session::start_camera`] | the effect's property group | //! -//! Roto Brush propagation fills the session's segmentation cache rather than the project, so it -//! keeps running on the UI thread. +//! Viewer frames go to a separate, long-lived frame worker fed with project diffs +//! ([`crate::remote`]). //! //! Cancelling terminates the worker: what an analysis tracked so far is dropped and the Render //! Queue item being rendered becomes "User Stopped". @@ -76,6 +77,13 @@ pub enum WorkerJob { layer: LayerId, effect: Uid, }, + /// Roto Brush propagation: the segmentations come back as [`WorkerReply::Segs`]. + RotoPropagate { + comp: ItemId, + layer: LayerId, + effect: Uid, + direction: crate::roto::Direction, + }, Track { comp: ItemId, layer: LayerId, @@ -103,6 +111,7 @@ impl WorkerJob { WorkerJob::Warp { .. } => JobKind::Warp, WorkerJob::Camera { .. } => JobKind::Camera, WorkerJob::RotoFreeze { .. } => JobKind::RotoFreeze, + WorkerJob::RotoPropagate { .. } => JobKind::RotoPropagate, WorkerJob::Track { .. } => JobKind::Track, WorkerJob::MaskTrack { .. } => JobKind::MaskTrack, } @@ -112,9 +121,10 @@ impl WorkerJob { pub fn target(&self) -> Option<(ItemId, LayerId, Uid)> { match *self { WorkerJob::Render { .. } => None, - WorkerJob::Warp { comp, layer, effect } | WorkerJob::Camera { comp, layer, effect } | WorkerJob::RotoFreeze { comp, layer, effect } => { - Some((comp, layer, effect)) - } + WorkerJob::Warp { comp, layer, effect } + | WorkerJob::Camera { comp, layer, effect } + | WorkerJob::RotoFreeze { comp, layer, effect } + | WorkerJob::RotoPropagate { comp, layer, effect, .. } => Some((comp, layer, effect)), WorkerJob::Track { comp, layer, tracker, .. } => Some((comp, layer, tracker)), WorkerJob::MaskTrack { comp, layer, mask, .. } => Some((comp, layer, mask)), } @@ -128,6 +138,7 @@ pub enum JobKind { Warp, Camera, RotoFreeze, + RotoPropagate, Track, MaskTrack, } @@ -140,6 +151,7 @@ impl JobKind { JobKind::Warp => "Warp Stabilizer Analysis", JobKind::Camera => "3D Camera Tracker Analysis", JobKind::RotoFreeze => "Freeze", + JobKind::RotoPropagate => "Roto Brush Propagation", JobKind::Track => "Analyze Track", JobKind::MaskTrack => "Track Mask", } @@ -170,6 +182,10 @@ pub enum WorkerReply { group: Box, message: String, }, + /// Roto Brush segmentations computed so far (propagation), not sent before. + Segs { + segs: Vec, + }, Failed { error: String, }, @@ -177,6 +193,41 @@ pub enum WorkerReply { Done, } +/// One full-resolution Roto Brush segmentation: its chain key and the base64 of +/// [`effectcraft_track::roto::FrameSeg::to_bytes`]. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct SegData { + pub key: u64, + pub data: String, +} + +impl SegData { + pub fn new(key: u64, seg: &effectcraft_track::roto::FrameSeg) -> SegData { + SegData { key, data: effectcraft_track::roto::rle::base64_encode(&seg.to_bytes()) } + } + + /// Put the segmentation into this instance's cache (what the Roto Brush effect renders from). + pub fn store(&self) -> bool { + let Some(seg) = effectcraft_track::roto::rle::base64_decode(&self.data).and_then(|b| effectcraft_track::roto::FrameSeg::from_bytes(&b)) else { + return false; + }; + effectcraft_effects::roto::store(self.key, 1.0, Arc::new(seg)); + true + } +} + +/// The full-resolution segmentations of `keys` that are cached here and not in `sent` yet. +pub fn segs_to_send(keys: &[u64], sent: &mut std::collections::HashSet) -> Vec { + let new: Vec<(u64, Arc)> = + keys.iter().filter(|k| !sent.contains(k)).filter_map(|k| effectcraft_effects::roto::cached(*k, 1.0).map(|s| (*k, s))).collect(); + new.into_iter() + .map(|(k, s)| { + sent.insert(k); + SegData::new(k, &s) + }) + .collect() +} + /// Replies waiting for the UI thread. #[derive(Default)] pub struct Inbox(Mutex>); @@ -292,6 +343,7 @@ fn run_job(s: &mut Session, job: WorkerJob, post: Post) -> Result<(), String> { WorkerJob::RotoFreeze { comp, layer, effect } => analysis(s, kind, target, post, |s| { s.start_roto(comp, layer, effect, crate::roto::RotoTask::Freeze, crate::roto::Direction::Both, true).map(|_| ()) }), + WorkerJob::RotoPropagate { comp, layer, effect, direction } => propagate(s, comp, layer, effect, direction, post), WorkerJob::Track { comp, layer, tracker, direction, settings, points, times } => analysis(s, kind, target, post, |s| { let work = crate::tracking::Work { project: s.project.clone(), @@ -323,6 +375,38 @@ fn run_job(s: &mut Session, job: WorkerJob, post: Post) -> Result<(), String> { } } +/// Roto Brush propagation in a worker: the segmentations stream back (at most every 250 ms) +/// as they are computed, so stopping the job keeps what was propagated. +fn propagate(s: &mut Session, comp: ItemId, layer: LayerId, effect: Uid, dir: crate::roto::Direction, post: Post) -> Result<(), String> { + let keys: Vec = s.roto_chain(comp, layer, effect).map(|c| c.keys.values().copied().collect()).ok_or("no Roto Brush strokes")?; + let sent = Rc::new(RefCell::new(std::collections::HashSet::new())); + let (p, k2, s2) = (post.clone(), keys.clone(), sent.clone()); + let mut last = None::; + let hook = Box::new(move |done: u64, total: u64| { + if last.is_none_or(|t: web_time::Instant| t.elapsed().as_millis() >= 250) { + last = Some(web_time::Instant::now()); + let segs = segs_to_send(&k2, &mut s2.borrow_mut()); + if !segs.is_empty() { + p(WorkerReply::Segs { segs }); + } + p(WorkerReply::Progress { done, total }); + } + }); + s.events.clear(); + with_hook(hook, || s.start_roto(comp, layer, effect, crate::roto::RotoTask::Propagate, dir, true))?; + let segs = segs_to_send(&keys, &mut sent.borrow_mut()); + if !segs.is_empty() { + post(WorkerReply::Segs { segs }); + } + if let Some(e) = s.events.iter().find_map(|e| match e { + Event::Toast { message, error: true } => Some(message.clone()), + _ => None, + }) { + return Err(e); + } + Ok(()) +} + fn analysis( s: &mut Session, kind: JobKind, @@ -398,6 +482,8 @@ pub struct OffloadedJob { pub uid: Uid, pub inbox: Arc, pub progress: OffloadProgress, + /// Segmentations received (Roto Brush propagation). + pub segs: u64, started: web_time::Instant, } @@ -432,7 +518,17 @@ impl Session { let inbox = Arc::new(Inbox::default()); let req = WorkerRequest { id, project: self.project.to_json(), files: footage_files(&self.project), job }; off.start(req, inbox.clone())?; - self.offloaded.push(OffloadedJob { id, kind, comp, layer, uid, inbox, progress: OffloadProgress::default(), started: web_time::Instant::now() }); + self.offloaded.push(OffloadedJob { + id, + kind, + comp, + layer, + uid, + inbox, + progress: OffloadProgress::default(), + segs: 0, + started: web_time::Instant::now(), + }); self.events.push(Event::ProjectChanged { revision: self.revision }); Ok(()) } @@ -469,6 +565,12 @@ impl Session { j.progress.total = total; } WorkerReply::Group { comp, layer, group, message } => results.push((j.kind, comp, layer, group, message)), + WorkerReply::Segs { segs } => { + for sd in &segs { + sd.store(); + } + j.segs += segs.len() as u64; + } WorkerReply::Failed { error } => j.progress.error = Some(error), WorkerReply::Done => j.progress.finished = true, WorkerReply::Render { .. } | WorkerReply::Item { .. } => {} @@ -495,17 +597,25 @@ impl Session { let mut done = vec![]; self.offloaded.retain(|j| { if j.progress.finished { - done.push((j.kind, j.progress.clone())); + done.push((j.kind, j.progress.clone(), j.segs)); } !j.progress.finished }); - for (kind, p) in done { + for (kind, p, segs) in done { changed = true; + self.offload_log.push((kind, p.error.clone().or_else(|| p.cancelled.then(|| "stopped".to_string())))); + if self.offload_log.len() > 32 { + self.offload_log.remove(0); + } if let Some(e) = p.error { self.events.push(Event::Toast { message: format!("{}: {e}", kind.label()), error: true }); } else if p.cancelled { - self.events.push(Event::Toast { message: format!("{} stopped", kind.label()), error: false }); + let kept = if segs > 0 { format!(" ({segs} frame(s) kept)") } else { String::new() }; + self.events.push(Event::Toast { message: format!("{} stopped{kept}", kind.label()), error: false }); + } else if kind == JobKind::RotoPropagate { + self.events.push(Event::Toast { message: format!("Roto Brush: propagated {} frame(s) in {:.1} s", segs, p.elapsed), error: false }); } + // Rendered frames depend on the result (an analysis' group, the segmentation cache). self.bump(); } changed @@ -567,6 +677,7 @@ mod tests { WorkerJob::Warp { comp, layer: LayerId(3), effect: 9 }, WorkerJob::RotoFreeze { comp, layer: LayerId(3), effect: 9 }, WorkerJob::Camera { comp, layer: LayerId(3), effect: 9 }, + WorkerJob::RotoPropagate { comp, layer: LayerId(3), effect: 9, direction: crate::roto::Direction::Forward }, WorkerJob::Track { comp, layer: LayerId(1), @@ -606,6 +717,7 @@ mod tests { WorkerReply::Item { update: ItemUpdate::Finished { id: 3, status: RenderStatus::Failed("x".into()), seconds: 1.5, output: Some("/a.gif".into()) } }, WorkerReply::Progress { done: 1, total: 2 }, WorkerReply::Group { comp, layer: LayerId(1), group: Box::new(g), message: "ok".into() }, + WorkerReply::Segs { segs: vec![SegData::new(5, &effectcraft_track::roto::FrameSeg::empty(4, 3))] }, WorkerReply::Failed { error: "boom".into() }, WorkerReply::Done, ] { diff --git a/crates/engine/src/remote.rs b/crates/engine/src/remote.rs new file mode 100644 index 00000000..ec0640ff --- /dev/null +++ b/crates/engine/src/remote.rs @@ -0,0 +1,384 @@ +//! Viewer frames rendered by another engine instance (the browser's frame worker). +//! +//! The page keeps no frame threads (a wasm build without shared memory has one thread per +//! instance), so the web app renders viewer frames in a long-lived Web Worker that holds a +//! replica of the project. The page sends the replica **project diffs** rather than the whole +//! project per edit, then render requests; the worker answers each with the frame's pixels. +//! +//! - [`diff`] / [`patch`]: a structural JSON diff of the serialized project (object members +//! added, changed or removed; arrays element-wise when their length is unchanged, else +//! replaced). Lossless (unlike JSON Merge Patch, `null` values survive). +//! - [`Mirror`] (page side): what the worker holds; turns the next project revision into a +//! [`FrameMsg::Sync`] (a full project the first time, a diff afterwards). +//! - [`FrameServer`] (worker side): applies syncs, relayed Roto Brush segmentations and render +//! requests, and renders frames as premultiplied RGBA8 ([`rgba8_premultiplied`]). +//! +//! The transport (structured-clone messages, transferable pixel buffers) is the web crate's; +//! everything here is portable and tested natively. + +use std::sync::Arc; + +use effectcraft_project::{ItemId, Project}; +use effectcraft_render::{ExprHost, FootageSource, Image, LayerCache, RenderOpts, Renderer}; +use effectcraft_time::Tick; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::offload::SegData; + +/// One step of a path into a JSON document. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(untagged)] +pub enum Step { + Key(String), + Index(usize), +} + +/// One change: set the value at `path` (creating or replacing the member/element), or remove +/// the object member at `path` (`value` absent). +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct Op { + pub path: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub value: Option, +} + +/// The changes that turn `old` into `new` (empty when equal). +pub fn diff(old: &Value, new: &Value) -> Vec { + let mut ops = vec![]; + diff_into(old, new, &mut vec![], &mut ops); + ops +} + +fn diff_into(old: &Value, new: &Value, path: &mut Vec, ops: &mut Vec) { + if old == new { + return; + } + match (old, new) { + (Value::Object(a), Value::Object(b)) => { + for (k, va) in a { + path.push(Step::Key(k.clone())); + match b.get(k) { + Some(vb) => diff_into(va, vb, path, ops), + None => ops.push(Op { path: path.clone(), value: None }), + } + path.pop(); + } + for (k, vb) in b { + if !a.contains_key(k) { + path.push(Step::Key(k.clone())); + ops.push(Op { path: path.clone(), value: Some(vb.clone()) }); + path.pop(); + } + } + } + (Value::Array(a), Value::Array(b)) if a.len() == b.len() => { + for (i, (va, vb)) in a.iter().zip(b).enumerate() { + path.push(Step::Index(i)); + diff_into(va, vb, path, ops); + path.pop(); + } + } + _ => ops.push(Op { path: path.clone(), value: Some(new.clone()) }), + } +} + +/// Apply [`diff`]'s changes to `doc`. +pub fn patch(doc: &mut Value, ops: &[Op]) -> Result<(), String> { + for op in ops { + let Some((last, parents)) = op.path.split_last() else { + *doc = op.value.clone().ok_or("cannot remove the document")?; + continue; + }; + let mut cur = &mut *doc; + for st in parents { + cur = match (st, cur) { + (Step::Key(k), Value::Object(m)) => m.get_mut(k).ok_or_else(|| format!("no member `{k}`"))?, + (Step::Index(i), Value::Array(a)) => a.get_mut(*i).ok_or_else(|| format!("no element {i}"))?, + _ => return Err("path does not match the document".into()), + }; + } + match (last, cur, &op.value) { + (Step::Key(k), Value::Object(m), Some(v)) => { + m.insert(k.clone(), v.clone()); + } + (Step::Key(k), Value::Object(m), None) => { + m.remove(k); + } + (Step::Index(i), Value::Array(a), Some(v)) => *a.get_mut(*i).ok_or_else(|| format!("no element {i}"))? = v.clone(), + _ => return Err("path does not match the document".into()), + } + } + Ok(()) +} + +/// Page → frame worker. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "camelCase")] +pub enum FrameMsg { + /// The whole project (JSON) at `revision`. + Project { revision: u64, project: String }, + /// Changes from the worker's revision `from` to `revision`. + Patch { from: u64, revision: u64, ops: Vec }, + /// Render a frame of the current project. + Render { id: u64, revision: u64, comp: ItemId, time: Tick, opts: Box }, + /// Roto Brush segmentations computed elsewhere (a propagation worker). + Segs { segs: Vec }, + /// Drop cached layer buffers (Edit ▸ Purge). + Purge, +} + +/// Frame worker → page. A [`FrameReply::Frame`]'s pixels travel next to the message (a +/// transferable buffer in the browser). +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "camelCase")] +pub enum FrameReply { + /// Premultiplied RGBA8 pixels of request `id`, rendered in `ms` milliseconds. + Frame { id: u64, width: u32, height: u32, ms: f64 }, + /// Request `id` was not rendered (`stale`: the worker holds another revision). + Failed { id: u64, error: String, stale: bool }, + /// A sync could not be applied: the page must send the whole project again. + Resync { error: String }, +} + +/// The page's record of a worker's replica. +#[derive(Default)] +pub struct Mirror { + sent: Option<(u64, Value)>, +} + +impl Mirror { + /// The revision the worker holds. + pub fn revision(&self) -> Option { + self.sent.as_ref().map(|s| s.0) + } + + /// Forget the replica (the next sync sends the whole project). + pub fn reset(&mut self) { + self.sent = None; + } + + /// The message that brings the worker to `revision` of `project` (`None` when it has it). + pub fn sync(&mut self, revision: u64, project: &Project) -> Option { + if self.revision() == Some(revision) { + return None; + } + let v = serde_json::to_value(project).unwrap_or(Value::Null); + let msg = match self.sent.take() { + Some((from, old)) => FrameMsg::Patch { from, revision, ops: diff(&old, &v) }, + None => FrameMsg::Project { revision, project: v.to_string() }, + }; + self.sent = Some((revision, v)); + Some(msg) + } +} + +/// Premultiplied f32 → premultiplied RGBA8 (colour clamped to alpha), as the viewer shows it. +pub fn rgba8_premultiplied(img: &Image) -> Vec { + let mut out = Vec::with_capacity(img.data.len() * 4); + for p in &img.data { + let a = p[3].clamp(0.0, 1.0); + let c = |v: f32| (v.clamp(0.0, a) * 255.0 + 0.5) as u8; + out.extend_from_slice(&[c(p[0]), c(p[1]), c(p[2]), (a * 255.0 + 0.5) as u8]); + } + out +} + +/// The worker's side: a project replica and what rendering needs. +pub struct FrameServer { + pub footage: Arc, + pub expr: Option>, + pub cache: Arc, + doc: Value, + project: Option>, + revision: Option, +} + +impl FrameServer { + pub fn new(footage: Arc, expr: Option>) -> FrameServer { + FrameServer { footage, expr, cache: Arc::new(LayerCache::default()), doc: Value::Null, project: None, revision: None } + } + + pub fn revision(&self) -> Option { + self.revision + } + + pub fn project(&self) -> Option<&Arc> { + self.project.as_ref() + } + + fn load(&mut self, revision: u64) -> Result<(), String> { + let p = Project::from_json(&self.doc.to_string()).map_err(|e| e.to_string())?; + self.project = Some(Arc::new(p)); + self.revision = Some(revision); + Ok(()) + } + + /// Handle one message; returns the replies with their pixels. + pub fn handle(&mut self, msg: FrameMsg) -> Vec<(FrameReply, Option>)> { + match msg { + FrameMsg::Project { revision, project } => { + let r = serde_json::from_str(&project).map_err(|e| e.to_string()).and_then(|v| { + self.doc = v; + self.load(revision) + }); + match r { + Ok(()) => vec![], + Err(e) => self.lost(e), + } + } + FrameMsg::Patch { from, revision, ops } => { + if self.revision != Some(from) { + return self.lost(format!("the replica is at {:?}, the patch starts at {from}", self.revision)); + } + match patch(&mut self.doc, &ops).and_then(|_| self.load(revision)) { + Ok(()) => vec![], + Err(e) => self.lost(e), + } + } + FrameMsg::Segs { segs } => { + for s in &segs { + s.store(); + } + vec![] + } + FrameMsg::Purge => { + self.cache.clear(); + vec![] + } + FrameMsg::Render { id, revision, comp, time, opts } => { + let fail = |error: String, stale: bool| vec![(FrameReply::Failed { id, error, stale }, None)]; + if self.revision != Some(revision) { + return fail(format!("the replica is at {:?}, not {revision}", self.revision), true); + } + let Some(p) = self.project.clone() else { return fail("no project".into(), true) }; + if p.comp(comp).is_none() { + return fail(format!("no composition {}", comp.0), false); + } + let t0 = web_time::Instant::now(); + let mut r = Renderer::new(&p, self.footage.as_ref(), *opts); + r.expr = self.expr.as_deref(); + r.cache = Some(&self.cache); + let img = r.comp_frame(comp, time); + let ms = t0.elapsed().as_secs_f64() * 1000.0; + vec![(FrameReply::Frame { id, width: img.width, height: img.height, ms }, Some(rgba8_premultiplied(&img)))] + } + } + } + + fn lost(&mut self, error: String) -> Vec<(FrameReply, Option>)> { + self.doc = Value::Null; + self.project = None; + self.revision = None; + vec![(FrameReply::Resync { error }, None)] + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::Session; + use serde_json::json; + + fn round Deserialize<'de>>(v: &T) -> T { + serde_json::from_str(&serde_json::to_string(v).unwrap()).unwrap() + } + + #[test] + fn diff_and_patch_round_trip() { + let a = json!({"a": 1, "b": [1, 2, {"c": null}], "d": {"e": "x", "f": [1]}, "g": null}); + let b = json!({"a": 2, "b": [1, 3, {"c": 4}], "d": {"f": [1, 2]}, "g": null, "h": {"i": null}}); + let ops = diff(&a, &b); + assert!(!ops.is_empty()); + let mut x = a.clone(); + patch(&mut x, &round(&ops)).unwrap(); + assert_eq!(x, b); + // Removal and the reverse direction. + let mut y = b.clone(); + patch(&mut y, &diff(&b, &a)).unwrap(); + assert_eq!(y, a); + assert!(diff(&a, &a).is_empty()); + // A whole-document change. + let mut z = json!(1); + patch(&mut z, &diff(&json!(1), &json!([1]))).unwrap(); + assert_eq!(z, json!([1])); + // A patch for another document fails cleanly. + assert!(patch(&mut json!(3), &diff(&a, &b)).is_err()); + } + + fn server(s: &Session) -> FrameServer { + FrameServer::new(s.footage.clone(), s.expr.clone()) + } + + /// Pixels the page itself would render. + fn local(s: &Session, comp: ItemId, t: Tick, opts: RenderOpts) -> Vec { + let mut r = Renderer::new(&s.project, s.footage.as_ref(), opts); + r.expr = s.expr.as_deref(); + rgba8_premultiplied(&r.comp_frame(comp, t)) + } + + #[test] + fn replica_follows_edits_through_diffs_and_renders_the_same_frames() { + let mut s = Session::default(); + s.execute("file.openDemoProject", json!({})).unwrap(); + let comp = s.active_comp_id().unwrap(); + let opts = RenderOpts { scale: 0.25, guides: true, ..Default::default() }; + let mut m = Mirror::default(); + let mut w = server(&s); + let send = |m: FrameMsg, w: &mut FrameServer| w.handle(round(&m)); + // First sync: the whole project. + let first = m.sync(s.revision, &s.project).unwrap(); + assert!(matches!(first, FrameMsg::Project { .. })); + assert!(send(first, &mut w).is_empty()); + assert!(m.sync(s.revision, &s.project).is_none(), "nothing new"); + let t = Tick::from_seconds_f64(1.0); + let render = |w: &mut FrameServer, rev: u64, id: u64| w.handle(round(&FrameMsg::Render { id, revision: rev, comp, time: t, opts: Box::new(opts) })); + let out = render(&mut w, s.revision, 1); + let (FrameReply::Frame { id, width, height, .. }, Some(px)) = &out[0] else { panic!("{:?}", out[0].0) }; + assert_eq!((*id, *width as usize * *height as usize * 4), (1, px.len())); + assert_eq!(*px, local(&s, comp, t, opts)); + // An edit travels as a small diff; the replica renders the edited frame. + s.execute("layer.newSolid", json!({"color": "#ff8800", "name": "Orange"})).unwrap(); + let msg = m.sync(s.revision, &s.project).unwrap(); + let FrameMsg::Patch { ops, .. } = &msg else { panic!("expected a patch") }; + assert!(!ops.is_empty() && serde_json::to_string(ops).unwrap().len() < s.project.to_json().len() / 2); + assert!(send(msg, &mut w).is_empty()); + assert_eq!(**w.project().unwrap(), *s.project); + let out = render(&mut w, s.revision, 2); + assert_eq!(out[0].1.as_deref(), Some(&local(&s, comp, t, opts)[..])); + // A request for another revision is refused as stale. + let out = render(&mut w, s.revision + 7, 3); + assert!(matches!(out[0].0, FrameReply::Failed { id: 3, stale: true, .. })); + // A patch from the wrong base asks for a resync; the mirror then sends the project. + let bad = FrameMsg::Patch { from: 999, revision: 1000, ops: vec![] }; + assert!(matches!(send(bad, &mut w)[0].0, FrameReply::Resync { .. })); + m.reset(); + assert!(matches!(m.sync(s.revision, &s.project), Some(FrameMsg::Project { .. }))); + } + + #[test] + fn frame_messages_round_trip() { + for m in [ + FrameMsg::Purge, + FrameMsg::Segs { segs: vec![SegData { key: 7, data: "AAAA".into() }] }, + FrameMsg::Render { + id: 3, + revision: 4, + comp: ItemId(2), + time: Tick::from_seconds_f64(0.5), + opts: Box::new(RenderOpts { scale: 0.5, draft: true, roi: Some([1.0, 2.0, 3.0, 4.0]), ..Default::default() }), + }, + ] { + let a = serde_json::to_string(&m).unwrap(); + assert_eq!(serde_json::to_string(&round(&m)).unwrap(), a); + assert!(a.starts_with("{\"type\":"), "{a}"); + } + for r in [ + FrameReply::Frame { id: 1, width: 2, height: 3, ms: 4.5 }, + FrameReply::Failed { id: 1, error: "x".into(), stale: true }, + FrameReply::Resync { error: "y".into() }, + ] { + assert_eq!(round(&r), r); + } + } +} diff --git a/crates/engine/src/roto.rs b/crates/engine/src/roto.rs index b6086d4e..2c1828f8 100644 --- a/crates/engine/src/roto.rs +++ b/crates/engine/src/roto.rs @@ -3,7 +3,8 @@ //! //! Segmentations are derived data in `effectcraft_effects::roto`'s content-keyed cache (see that //! module). `roto.propagate` fills it at full resolution in the background (or blocking with -//! `wait`), frame by frame outward from the base frame, rendering the effect's input +//! `wait`; in the browser the background is a Web Worker whose segmentations stream back, +//! [`crate::offload`]), frame by frame outward from the base frame, rendering the effect's input //! (`Renderer::layer_input`, source → masks → the effects above it). `roto.freeze` runs the same //! pass and stores the final mattes in the effect (one undo step). [`sync`], run after every //! edit, keeps each instance's hidden Input Key equal to a signature of what its frames are made @@ -403,16 +404,21 @@ pub(crate) fn job_frames(d: &effectcraft_track::roto::RotoData, dir: Direction) } impl Session { + /// The Roto Brush job running in a worker (Freeze or propagation), with its task. + fn offloaded_roto(&self) -> Option<(&crate::offload::OffloadedJob, RotoTask)> { + self.offloaded(JobKind::RotoFreeze).map(|j| (j, RotoTask::Freeze)).or_else(|| self.offloaded(JobKind::RotoPropagate).map(|j| (j, RotoTask::Propagate))) + } + pub fn is_roto_running(&self) -> bool { - self.roto_job.as_ref().is_some_and(|j| !j.is_finished()) || self.offloaded(JobKind::RotoFreeze).is_some() + self.roto_job.as_ref().is_some_and(|j| !j.is_finished()) || self.offloaded_roto().is_some() } pub fn roto_progress(&self) -> Option { self.roto_job.as_ref().map(|j| j.progress()).or_else(|| { - self.offloaded(JobKind::RotoFreeze).map(|j| { + self.offloaded_roto().map(|(j, task)| { let p = &j.progress; RotoProgress { - task: Some(RotoTask::Freeze), + task: Some(task), done: p.done, total: p.total, elapsed: p.elapsed, @@ -425,14 +431,22 @@ impl Session { /// (comp, layer, effect uid, task) of the running job. pub fn roto_target(&self) -> Option<(ItemId, LayerId, Uid, RotoTask)> { - self.roto_job - .as_ref() - .map(|j| (j.comp, j.layer, j.effect, j.task)) - .or_else(|| self.offloaded(JobKind::RotoFreeze).map(|j| (j.comp, j.layer, j.uid, RotoTask::Freeze))) + self.roto_job.as_ref().map(|j| (j.comp, j.layer, j.effect, j.task)).or_else(|| self.offloaded_roto().map(|(j, task)| (j.comp, j.layer, j.uid, task))) + } + + /// The segmentation chain a propagation of `effect` computes (its frames' cache keys). + pub fn roto_chain(&self, comp: ItemId, layer: LayerId, effect: Uid) -> Option { + let c = self.project.comp(comp)?; + let l = c.layer(layer)?; + let g = l.effects()?.groups().find(|g| g.uid == effect && is_roto(g))?; + let d = fx::data(¶ms_static(g)); + let t = comp_time_of(c, l, d.base?); + let params = params_at(&self.project, comp, l, g, t, self.expr.as_deref()); + Some(fx::Chain::new(¶ms, layer_size(&self.project, c, l), c.frame_rate.as_f64(), 1.0)) } pub fn stop_roto(&mut self) -> bool { - if self.cancel_offloaded(JobKind::RotoFreeze) { + if self.cancel_offloaded(JobKind::RotoFreeze) || self.cancel_offloaded(JobKind::RotoPropagate) { return true; } match &self.roto_job { @@ -491,9 +505,12 @@ impl Session { task, }; let strokes = str_of(g, fx::STROKES); - if !wait && task == RotoTask::Freeze && self.offloads() { + if !wait && self.offloads() { drop(work); - self.offload_analysis(crate::offload::WorkerJob::RotoFreeze { comp, layer, effect })?; + self.offload_analysis(match task { + RotoTask::Freeze => crate::offload::WorkerJob::RotoFreeze { comp, layer, effect }, + RotoTask::Propagate => crate::offload::WorkerJob::RotoPropagate { comp, layer, effect, direction: dir }, + })?; return Ok(n); } let wait = wait || cfg!(target_arch = "wasm32"); @@ -567,7 +584,7 @@ impl Session { /// instances in the background. pub fn poll_roto(&mut self, auto: bool) -> bool { let changed = self.poll_roto_job(); - if auto && self.roto_job.is_none() && self.state.roto.auto_propagate { + if auto && self.roto_job.is_none() && self.offloaded_roto().is_none() && self.state.roto.auto_propagate { while let Some((c, l, e)) = self.roto_pending.first().copied() { self.roto_pending.remove(0); if self.start_roto(c, l, e, RotoTask::Propagate, Direction::Both, false).is_ok() { diff --git a/crates/engine/src/tests_panels.rs b/crates/engine/src/tests_panels.rs index d989b28e..8f5cb7f1 100644 --- a/crates/engine/src/tests_panels.rs +++ b/crates/engine/src/tests_panels.rs @@ -516,3 +516,64 @@ fn panels_commands_are_wired() { assert!(s.is_enabled(id), "{id}"); } } + +/// A browser like the web app's: a virtual tree whose files are read on demand. +#[derive(Default)] +struct VirtualBrowser { + tree: Mutex, + fetched: Mutex>, + actions: Mutex>, +} + +impl crate::media_browser::Browser for VirtualBrowser { + fn home(&self) -> String { + "/Browser Storage".into() + } + fn places(&self) -> Vec<(String, String)> { + vec![("Browser Storage".into(), self.home()), ("Shoot".into(), "/Folders/Shoot".into())] + } + fn list(&self, dir: &str, only: bool) -> Result, String> { + self.tree.lock().unwrap().list(dir, only) + } + fn fetch(&self, paths: &[String], _: &Value) -> bool { + // Opened folders' files arrive later. + let pending: Vec = paths.iter().filter(|p| p.starts_with("/Folders/")).cloned().collect(); + self.fetched.lock().unwrap().extend(pending.iter().cloned()); + pending.is_empty() + } + fn actions(&self) -> Vec<(String, String)> { + vec![("openFolder".into(), "Open Folder…".into())] + } + fn action(&self, id: &str) -> Result { + self.actions.lock().unwrap().push(id.into()); + Ok(json!({"pending": true})) + } +} + +#[test] +fn media_browser_over_a_virtual_tree() { + let b = Arc::new(VirtualBrowser::default()); + { + let mut t = b.tree.lock().unwrap(); + t.add_dir("/Browser Storage"); + t.add_file("/Folders/Shoot/a.csv", "/Folders/Shoot/a.csv", 12, Some(1)); + } + let mut s = Session { browser: Some(b.clone()), ..Default::default() }; + let r = s.execute("mediaBrowser.list", json!({})).unwrap(); + assert_eq!(r["path"], "/Browser Storage"); + assert_eq!(r["places"][1]["path"], "/Folders/Shoot"); + assert_eq!(r["actions"][0]["id"], "openFolder"); + let r = s.execute("mediaBrowser.go", json!({"path": "/Folders/Shoot"})).unwrap(); + assert_eq!(r["entries"][0]["name"], "a.csv"); + assert_eq!(r["parent"], "/Folders"); + assert!(s.execute("mediaBrowser.go", json!({"path": "/Nope"})).is_err()); + // Importing a file that is not read yet: pending (the host imports it once it arrives). + let r = s.execute("mediaBrowser.import", json!({"paths": ["/Folders/Shoot/a.csv"]})).unwrap(); + assert_eq!(r["pending"], true); + assert_eq!(*b.fetched.lock().unwrap(), ["/Folders/Shoot/a.csv"]); + assert!(s.project.items.values().all(|i| !matches!(i.kind, ItemKind::Footage(_)))); + let r = s.execute("mediaBrowser.action", json!({"action": "openFolder"})).unwrap(); + assert_eq!(r["pending"], true); + assert_eq!(*b.actions.lock().unwrap(), ["openFolder"]); + assert!(s.execute("mediaBrowser.action", json!({})).is_err()); +} diff --git a/crates/engine/src/tests_roto.rs b/crates/engine/src/tests_roto.rs index 79403626..41d8e86e 100644 --- a/crates/engine/src/tests_roto.rs +++ b/crates/engine/src/tests_roto.rs @@ -286,3 +286,52 @@ fn propagation_is_deterministic_and_runs_in_background() { let r = s.execute("roto.stroke", json!({"layer": clip.0, "points": [[90, 110]]})).unwrap(); assert_eq!(r["radius"], 15.0); } + +/// Runs offloaded jobs at once in a second session sharing the footage, through JSON (what the +/// browser's Web Worker does, minus the thread). +struct InlineWorker(std::sync::Arc); + +impl crate::offload::Offload for InlineWorker { + fn start(&self, req: crate::offload::WorkerRequest, inbox: std::sync::Arc) -> Result<(), String> { + let req: crate::offload::WorkerRequest = serde_json::from_str(&serde_json::to_string(&req).unwrap()).unwrap(); + let mut w = Session { footage: self.0.clone(), ..Default::default() }; + let post: crate::offload::Post = std::rc::Rc::new(move |r| inbox.push(serde_json::from_str(&serde_json::to_string(&r).unwrap()).unwrap())); + crate::offload::run_request(&mut w, req, &post); + Ok(()) + } + fn cancel(&self, _: u64) {} +} + +#[test] +fn offloaded_propagation_streams_segmentations_back() { + let (mut s, clip, uid) = painted(center, "disk-offload.mov"); + s.execute("roto.span", json!({"layer": clip.0, "end": 5})).unwrap(); + s.offload = Some(std::sync::Arc::new(InlineWorker(s.footage.clone()))); + let r = s.execute("roto.propagate", json!({"layer": clip.0})).unwrap(); + assert_eq!(r["frames"], 6, "{r}"); + assert!(s.roto_job.is_none(), "offloaded, not on a thread"); + assert!(s.is_roto_running()); + assert_eq!(s.roto_progress().unwrap().task, Some(crate::roto::RotoTask::Propagate)); + assert_eq!(s.roto_target().map(|t| (t.1, t.2)), Some((clip, uid))); + assert!(s.jobs().iter().any(|j| j.id == "roto")); + // The worker has its own segmentation cache: here nothing is computed until its replies + // are applied. + forget(&s, clip, uid); + assert_eq!(s.execute("roto.status", json!({"layer": clip.0})).unwrap()["computed"], json!([])); + let rev = s.revision; + assert!(s.poll_offload()); + assert!(!s.is_roto_running()); + assert!(s.revision > rev, "frames are redrawn"); + let st = s.execute("roto.status", json!({"layer": clip.0})).unwrap(); + assert_eq!(st["computed"], json!([0, 1, 2, 3, 4, 5]), "{st}"); + assert!(s.events.iter().any(|e| matches!(e, crate::Event::Toast { message, error: false } if message.contains("propagated 6 frame")))); + // The matte matches one computed here. + let a = s.execute("roto.status", json!({"layer": clip.0, "frame": 5, "matte": true})).unwrap()["matte"].clone(); + forget(&s, clip, uid); + let b = s.execute("roto.status", json!({"layer": clip.0, "frame": 5, "matte": true})).unwrap()["matte"].clone(); + assert!(a.is_string()); + assert_eq!(a, b); + // `wait` still runs here. + s.execute("roto.propagate", json!({"layer": clip.0, "wait": true})).unwrap(); + assert!(s.offloaded.is_empty()); +} diff --git a/crates/render/src/lib.rs b/crates/render/src/lib.rs index 6204b763..25ebe466 100644 --- a/crates/render/src/lib.rs +++ b/crates/render/src/lib.rs @@ -320,7 +320,8 @@ impl FxTarget for CpuFx<'_> { } } -#[derive(Clone, Copy, Debug)] +#[derive(Clone, Copy, Debug, serde::Serialize, serde::Deserialize)] +#[serde(default, rename_all = "camelCase")] pub struct RenderOpts { /// Output scale relative to comp pixels (1 = Full, 0.5 = Half…). pub scale: f64, diff --git a/crates/track/src/roto/mod.rs b/crates/track/src/roto/mod.rs index 393f743e..7103624b 100644 --- a/crates/track/src/roto/mod.rs +++ b/crates/track/src/roto/mod.rs @@ -230,6 +230,61 @@ impl FrameSeg { } (n > 0.0).then(|| [sx / n, sy / n]) } + + /// A compact binary form (to ship a segmentation between engine instances, e.g. from a Web + /// Worker to the page): `ECSG`, width, height (u32 LE), the colour models as JSON (length + /// prefixed), then the matte and the Refine Edge band run-length encoded (u16 LE run, byte). + pub fn to_bytes(&self) -> Vec { + let models = serde_json::to_vec(&(&self.fg, &self.bg)).unwrap_or_default(); + let mut out = Vec::with_capacity(16 + models.len() + 64); + out.extend_from_slice(b"ECSG"); + out.extend_from_slice(&(self.w as u32).to_le_bytes()); + out.extend_from_slice(&(self.h as u32).to_le_bytes()); + out.extend_from_slice(&(models.len() as u32).to_le_bytes()); + out.extend_from_slice(&models); + for plane in [&self.matte, &self.refine] { + let mut i = 0; + while i < plane.len() { + let v = plane[i]; + let mut n = 1; + while i + n < plane.len() && plane[i + n] == v && n < u16::MAX as usize { + n += 1; + } + out.extend_from_slice(&(n as u16).to_le_bytes()); + out.push(v); + i += n; + } + } + out + } + + /// Inverse of [`FrameSeg::to_bytes`] (`None` for malformed input). + pub fn from_bytes(b: &[u8]) -> Option { + if b.len() < 16 || &b[..4] != b"ECSG" { + return None; + } + let u = |i: usize| u32::from_le_bytes([b[i], b[i + 1], b[i + 2], b[i + 3]]) as usize; + let (w, h, ml) = (u(4), u(8), u(12)); + let n = w.checked_mul(h)?; + let models = b.get(16..16 + ml)?; + let (fg, bg): (Gmm, Gmm) = serde_json::from_slice(models).ok()?; + let mut rest = &b[16 + ml..]; + let mut plane = || -> Option> { + let mut v = Vec::with_capacity(n); + while v.len() < n { + if rest.len() < 3 { + return None; + } + let run = u16::from_le_bytes([rest[0], rest[1]]) as usize; + v.resize(v.len() + run, rest[2]); + rest = &rest[3..]; + } + (v.len() == n).then_some(v) + }; + let matte = plane()?; + let refine = plane()?; + Some(FrameSeg { w, h, matte, refine, fg, bg }) + } } /// Straight RGB of an image (transparent pixels are black). diff --git a/crates/track/src/roto/tests.rs b/crates/track/src/roto/tests.rs index bec3f378..15bd1f68 100644 --- a/crates/track/src/roto/tests.rs +++ b/crates/track/src/roto/tests.rs @@ -89,6 +89,12 @@ fn strokes_segment_the_base_frame() { // Deterministic. let again = segment(&img, &refs, None, &SegOpts::default(), 0.0); assert_eq!(seg, again); + // The binary form round-trips exactly and is compact (run-length encoded planes). + let bytes = seg.to_bytes(); + assert_eq!(FrameSeg::from_bytes(&bytes).as_ref(), Some(&seg)); + assert!(bytes.len() < seg.w * seg.h / 4, "{} bytes", bytes.len()); + assert!(FrameSeg::from_bytes(&bytes[..bytes.len() - 1]).is_none()); + assert!(FrameSeg::from_bytes(b"nope").is_none()); } #[test] diff --git a/crates/ui-egui/src/control.rs b/crates/ui-egui/src/control.rs index ff3db561..3e6bdd10 100644 --- a/crates/ui-egui/src/control.rs +++ b/crates/ui-egui/src/control.rs @@ -55,6 +55,87 @@ pub enum Outcome { path: Option, crop: Option<[f32; 4]>, }, + /// Reply once background jobs end (a `wait: true` command whose job runs in a worker). + AwaitJobs(JobWaiter), +} + +/// A `wait: true` engine command whose job was sent to the session's offload (the browser's +/// Web Workers) instead of blocking the UI thread: the reply goes out when the job ends. +pub struct JobWaiter { + pub command: String, + pub params: Value, + /// The command's own result (with `wait: false`). + pub result: Value, + /// Ids of the jobs it started (`jobs.list` ids: `render`, `track`, `roto`…). + pub jobs: Vec, + /// Length of [`effectcraft_engine::Session::offload_log`] when it started. + pub log_len: usize, +} + +/// Commands that block with `wait` (and its default). +fn waits(command: &str, params: &Value) -> bool { + let default = command == "renderQueue.render"; + params.get("wait").and_then(Value::as_bool).unwrap_or(default) +} + +/// Run an engine command. A waiting command (`wait: true`) whose job the session offloads +/// runs with `wait: false` instead, and the reply waits for the job ([`Outcome::AwaitJobs`]), +/// so the browser's UI keeps drawing while it runs. +fn execute_engine(app: &mut EffectcraftApp, id: &str, params: Value) -> Result { + if !app.session.offloads() || !waits(id, ¶ms) { + return app.session.execute_checked(id, params).map(ok).map_err(|e| e.to_string()); + } + let before: Vec = app.session.jobs().into_iter().map(|j| j.id).collect(); + let log_len = app.session.offload_log.len(); + let mut p = params.clone(); + p["wait"] = json!(false); + let result = app.session.execute_checked(id, p).map_err(|e| e.to_string())?; + let jobs: Vec = app.session.jobs().into_iter().map(|j| j.id).filter(|j| !before.contains(j)).collect(); + if jobs.is_empty() { + return Ok(ok(result)); + } + Ok(Outcome::AwaitJobs(JobWaiter { command: id.to_string(), params, result, jobs, log_len })) +} + +impl JobWaiter { + /// Whether its jobs are still running. + pub fn pending(&self, app: &EffectcraftApp) -> bool { + app.session.jobs().iter().any(|j| self.jobs.contains(&j.id)) + } + + /// The reply: what the blocking command would have returned (`running: false`, the final + /// progress, render items and analysis status), or the job's error. + pub fn finish(self, app: &mut EffectcraftApp) -> Value { + let s = &mut app.session; + s.poll_render(); + let errors: Vec = s.offload_log.iter().skip(self.log_len).filter_map(|(_, e)| e.clone()).collect(); + if let Some(e) = errors.first() { + return json!({"ok": false, "error": e}); + } + let mut r = self.result; + if r.get("running").is_some() { + r["running"] = json!(false); + } + if self.command == "renderQueue.render" { + let ids: Vec = r["items"].as_array().map(|a| a.iter().map(|i| i["id"].clone()).collect()).unwrap_or_default(); + if let Ok(list) = s.execute("renderQueue.list", json!({})) { + r["items"] = + json!(list["items"].as_array().map(|a| a.iter().filter(|i| ids.contains(&i["id"])).cloned().collect::>()).unwrap_or_default()); + } + r["rendering"] = json!(s.is_rendering()); + } + let prefix = self.command.split('.').next().unwrap_or(""); + if matches!(prefix, "warp" | "camera" | "roto") + && r.get("layer").is_some() + && let Ok(st) = s.execute(&format!("{prefix}.status"), json!({"layer": r["layer"], "effect": r["effect"]})) + { + r["status"] = st; + } + if r.get("progress").is_some() { + r["progress"] = Value::Null; + } + json!({"ok": true, "result": r}) + } } fn ok(v: Value) -> Outcome { @@ -94,12 +175,12 @@ pub fn handle(app: &mut EffectcraftApp, ctx: &egui::Context, req: &ControlReques // `comp.new {}` creates a default comp); UI-only ids (`tool.*`, `view.*`, `window.*`, // `playback.*` …) and `ui.menu.invoke` go through the menu dispatcher like a click. let r = if req.method == "engine.execute" && effectcraft_engine::find_command(id).is_some() { - app.session.execute_checked(id, params).map_err(|e| e.to_string()) + execute_engine(app, id, params) } else { - crate::menus::invoke(app, ctx, id, params) + crate::menus::invoke(app, ctx, id, params).map(ok) }; match r { - Ok(v) => ok(v), + Ok(o) => o, Err(e) => { app.ui.status = e.clone(); err(e) diff --git a/crates/ui-egui/src/frames.rs b/crates/ui-egui/src/frames.rs index fd628351..880d1928 100644 --- a/crates/ui-egui/src/frames.rs +++ b/crates/ui-egui/src/frames.rs @@ -4,8 +4,10 @@ //! never waits for the compositor. Results land in a memory-budgeted cache keyed by (project //! revision, comp, frame, scale); the timeline draws the cached range as the green cache bar. //! -//! wasm32 has no threads: jobs wait in the same priority queue and [`Frames::pump`] renders them -//! on the UI thread between egui frames, within a time budget. +//! wasm32 has no threads: jobs wait in the same priority queue and [`Frames::pump`] hands them +//! to a [`RemoteFrames`] (the browser's frame worker, a second engine instance fed with project +//! diffs, `effectcraft_engine::remote`) or, without one, renders them on the UI thread between +//! egui frames, within a time budget. //! //! With the disk cache on (Settings ▸ Media & Disk Cache), every CPU-rendered frame is also //! written to disk under a content key (the comp's content hash, frame, scale, view and render @@ -163,6 +165,38 @@ fn frame_disk_key(keys: &ContentKeys, project: &Project, key: &FrameKey, opts_ha disk_cache::frame_key(content, key.frame, key.scale, key.view ^ opts_hash) } +/// A frame for a [`RemoteFrames`] renderer. +pub struct RemoteJob { + pub project: Arc, + pub revision: u64, + pub comp: ItemId, + pub t: Tick, + pub opts: RenderOpts, +} + +/// Premultiplied RGBA8 pixels from a [`RemoteFrames`] renderer. +pub struct RemoteFrame { + pub width: u32, + pub height: u32, + pub rgba: Vec, + /// Render time (ms). + pub ms: f64, +} + +/// Called once with a remote frame (or why there is none). +pub type RemoteDone = Box) + Send>; + +/// Renders viewer frames elsewhere (the browser's frame worker), so the UI thread never does. +pub trait RemoteFrames: Send + Sync { + /// Frames it can render at the same time (0 when it is not usable: frames then render on + /// the UI thread again). + fn slots(&self) -> usize; + /// Start rendering `job`; `done` runs when the frame arrives. + fn start(&self, job: RemoteJob, done: RemoteDone); + /// Drop cached intermediate results (Edit ▸ Purge). + fn purge(&self) {} +} + pub struct Frames { cache: Arc>, /// Keys queued or rendering. @@ -175,6 +209,9 @@ pub struct Frames { /// Render time of the last viewer (urgent) frame in ms, for the Info panel / perf readout. pub last_ms: Arc>, content_keys: ContentKeys, + /// Where frames render instead of the UI thread (wasm32), and how many it is rendering. + remote: Option>, + remote_busy: Arc>, } impl Default for Frames { @@ -190,6 +227,8 @@ impl Default for Frames { ctx: None, last_ms: Arc::new(Mutex::new(0.0)), content_keys: Arc::default(), + remote: None, + remote_busy: Arc::default(), } } } @@ -270,6 +309,9 @@ impl Frames { } pub fn clear(&self) { + if let Some(r) = &self.remote { + r.purge(); + } if let Ok(mut c) = self.cache.lock() { c.map.clear(); c.order.clear(); @@ -324,7 +366,7 @@ impl Frames { q.jobs.push(Job { key, src: src.clone(), comp, t, opts, urgent, seq }); drop(q); #[cfg(not(target_arch = "wasm32"))] - { + if self.remote.is_none() { let w = self.worker(); // One spawn per job; each spawn renders whichever queued job matters most right now. self.pool.spawn(move || { @@ -349,13 +391,102 @@ impl Frames { } } - /// Render queued frames on this thread, most important first, until the queue is empty or - /// `budget` has passed (at least one frame). Returns whether frames are still queued. Only - /// wasm32 needs this (native frames render on the pool); elsewhere it does nothing. + /// Render frames with `remote` from now on (`None`: on this thread again). + pub fn set_remote(&mut self, remote: Option>) { + self.remote = remote; + } + + /// Whether frames go to a usable [`RemoteFrames`]. + pub fn remote_active(&self) -> bool { + self.remote.as_ref().is_some_and(|r| r.slots() > 0) + } + + /// Frames the remote renderer renders at once (0 without one). + pub fn remote_slots(&self) -> usize { + self.remote.as_ref().map(|r| r.slots()).unwrap_or(0) + } + + /// Frames being rendered remotely. + pub fn remote_busy(&self) -> usize { + self.remote_busy.lock().map(|b| *b).unwrap_or(0) + } + + /// Hand queued frames to the remote renderer while it has free slots, most important first + /// (the viewer's frame, then prefetch in request order; a viewer frame requested while every + /// slot is busy goes to the first one that frees up). Frames that the GPU compositor can keep + /// on the GPU render here instead (no readback). Returns how many were started. + pub fn dispatch_remote(&self) -> usize { + let Some(remote) = self.remote.clone() else { return 0 }; + let slots = remote.slots(); + let w = self.worker(); + let mut started = 0; + loop { + let busy = self.remote_busy(); + if busy >= slots { + break; + } + let job = { + let Ok(mut q) = self.queue.lock() else { break }; + let Some(i) = Queue::best(&q.jobs) else { break }; + let job = q.jobs.swap_remove(i); + q.running.insert(job.key); + if job.urgent { + q.urgent_running += 1; + } + job + }; + if job.src.gpu.is_some() && job.src.gpu_display { + let t0 = web_time::Instant::now(); + if let Some(img) = w.render_gpu(&job) { + w.finish(&job, img, t0); + continue; + } + } + if let Ok(mut b) = self.remote_busy.lock() { + *b += 1; + } + started += 1; + let (key, urgent) = (job.key, job.urgent); + let fw = self.worker(); + let busy = self.remote_busy.clone(); + let t0 = web_time::Instant::now(); + let rj = RemoteJob { project: job.src.project.clone(), revision: key.revision, comp: job.comp, t: job.t, opts: job.opts }; + remote.start( + rj, + Box::new(move |r: Result| { + if let Ok(mut b) = busy.lock() { + *b = b.saturating_sub(1); + } + match r { + Ok(f) => { + let img = egui::ColorImage::from_rgba_premultiplied([f.width as usize, f.height as usize], &f.rgba); + fw.finish_key(key, urgent, FrameImage::Cpu(Arc::new(img)), t0); + } + Err(e) => { + // Not rendered (a stale revision, a lost worker): the viewer asks again. + log::debug!("remote frame: {e}"); + fw.release(key, urgent); + } + } + }), + ); + } + started + } + + /// Render queued frames, most important first. With a usable [`RemoteFrames`] they are only + /// handed over ([`Frames::dispatch_remote`]); otherwise they render on this thread until the + /// queue is empty or `budget` has passed (at least one frame). Returns whether frames are + /// still queued or rendering remotely. Only wasm32 needs this (native frames render on the + /// pool); elsewhere it does nothing. pub fn pump(&self, budget: std::time::Duration) -> bool { if cfg!(not(target_arch = "wasm32")) { return false; } + if self.remote_active() { + self.dispatch_remote(); + return self.queue.lock().map(|q| !q.jobs.is_empty()).unwrap_or(false) || self.remote_busy() > 0; + } let t0 = web_time::Instant::now(); let w = self.worker(); while w.run_next() { @@ -380,12 +511,10 @@ struct Worker { impl Worker { /// Render the queued job that matters most right now. `false` when the queue was empty. fn run_next(&self) -> bool { - let Worker { queue, cache, inflight, ctx, last, content_keys } = self; + let Worker { queue, content_keys, .. } = self; let job = { let Ok(mut q) = queue.lock() else { return false }; - // Urgent first (newest), then prefetch in request order. - let best = q.jobs.iter().enumerate().max_by_key(|(_, j)| (j.urgent, if j.urgent { j.seq as i64 } else { -(j.seq as i64) })).map(|(i, _)| i); - let Some(i) = best else { return false }; + let Some(i) = Queue::best(&q.jobs) else { return false }; let job = q.jobs.swap_remove(i); q.running.insert(job.key); if job.urgent { @@ -407,27 +536,50 @@ impl Worker { ci } }; - if job.urgent - && let Ok(mut l) = last.lock() - { + self.finish(&job, ci, t0); + true + } + + /// Store a finished frame and release its job. + fn finish(&self, job: &Job, img: FrameImage, t0: web_time::Instant) { + self.finish_key(job.key, job.urgent, img, t0); + } + + fn finish_key(&self, key: FrameKey, urgent: bool, img: FrameImage, t0: web_time::Instant) { + if urgent && let Ok(mut l) = self.last.lock() { *l = t0.elapsed().as_secs_f64() * 1000.0; } - if let Ok(mut c) = cache.lock() { - c.insert(job.key, ci); + if let Ok(mut c) = self.cache.lock() { + c.insert(key, img); } - if let Ok(mut inf) = inflight.lock() { - inf.remove(&job.key); + self.release(key, urgent); + } + + /// The job of `key` is no longer queued or running (rendered or dropped). + fn release(&self, key: FrameKey, urgent: bool) { + if let Ok(mut inf) = self.inflight.lock() { + inf.remove(&key); } - if let Ok(mut q) = queue.lock() { - q.running.remove(&job.key); - if job.urgent { - q.urgent_running -= 1; + if let Ok(mut q) = self.queue.lock() { + q.running.remove(&key); + if urgent { + q.urgent_running = q.urgent_running.saturating_sub(1); } } - if let Some(ctx) = ctx { + if let Some(ctx) = &self.ctx { ctx.request_repaint(); } - true + } + + /// The frame on the GPU, kept there for the viewer, if the compositor can do it alone. + fn render_gpu(&self, job: &Job) -> Option { + let g = job.src.gpu.as_ref()?; + let mut r = Renderer::new(&job.src.project, job.src.footage.as_ref(), job.opts); + r.expr = job.src.expr.as_deref(); + r.cache = Some(&job.src.layer_cache); + r.accel = Some(g as &dyn effectcraft_engine::render::Accelerator); + r.active_accel()?; + g.render_display(&r, job.comp, job.t).map(|f| FrameImage::Gpu(Arc::new(f))) } fn render_job(&self, job: &Job) -> FrameImage { @@ -465,3 +617,10 @@ struct Queue { urgent_running: usize, seq: u64, } + +impl Queue { + /// The job that matters most: urgent first (newest), then prefetch in request order. + fn best(jobs: &[Job]) -> Option { + jobs.iter().enumerate().max_by_key(|(_, j)| (j.urgent, if j.urgent { j.seq as i64 } else { -(j.seq as i64) })).map(|(i, _)| i) + } +} diff --git a/crates/ui-egui/src/lib.rs b/crates/ui-egui/src/lib.rs index 92eed7ae..d47ca3aa 100644 --- a/crates/ui-egui/src/lib.rs +++ b/crates/ui-egui/src/lib.rs @@ -143,6 +143,8 @@ pub struct EffectcraftApp { pub(crate) deferred: Vec<(ControlRequest, f64)>, pub(crate) synthetic: Vec, pub(crate) input_waiters: Vec>, + /// Control requests waiting for an offloaded job (`wait: true` in the browser). + pub(crate) job_waiters: Vec<(control::JobWaiter, Sender)>, queued_screenshots: Vec<(u64, f64, u32)>, pending_screenshots: Vec<(u64, Option, Option<[f32; 4]>, Sender, f64)>, next_token: u64, @@ -210,6 +212,7 @@ impl EffectcraftApp { deferred: vec![], synthetic: vec![], input_waiters: vec![], + job_waiters: vec![], queued_screenshots: vec![], pending_screenshots: vec![], next_token: 1, @@ -847,8 +850,9 @@ impl EffectcraftApp { fn frames_parallelism(&self) -> usize { if cfg!(target_arch = "wasm32") { - // frames render one at a time on the UI thread (`Frames::pump`) - return 1; + // frame workers render one frame each; without them frames render one at a time + // on the UI thread (`Frames::pump`) + return self.frames.remote_slots().max(1); } std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4) } @@ -881,6 +885,10 @@ impl EffectcraftApp { } } control::Outcome::AfterInput => self.input_waiters.push(reply), + control::Outcome::AwaitJobs(w) => { + self.job_waiters.push((w, reply)); + ctx.request_repaint(); + } control::Outcome::Screenshot { path, crop } => { // A covered or minimized window presents no frames (macOS skips its redraws), so // a screenshot would never arrive and nothing would tick its timeout. @@ -897,6 +905,28 @@ impl EffectcraftApp { } } self.control_rx = Some(rx); + self.finish_job_waiters(ctx); + } + + /// Reply to the control requests whose offloaded jobs ended. + fn finish_job_waiters(&mut self, ctx: &egui::Context) { + if self.job_waiters.is_empty() { + return; + } + self.session.poll_render(); + self.session.poll_offload(); + let waiters = std::mem::take(&mut self.job_waiters); + for (w, reply) in waiters { + if w.pending(self) { + self.job_waiters.push((w, reply)); + } else { + let v = w.finish(self); + let _ = reply.send(v); + } + } + if !self.job_waiters.is_empty() { + ctx.request_repaint_after(std::time::Duration::from_millis(50)); + } } fn issue_screenshots(&mut self, ctx: &egui::Context) { diff --git a/crates/ui-egui/src/panels/media_panels.rs b/crates/ui-egui/src/panels/media_panels.rs index 4f6fd0ae..8cc14dcc 100644 --- a/crates/ui-egui/src/panels/media_panels.rs +++ b/crates/ui-egui/src/panels/media_panels.rs @@ -65,13 +65,13 @@ pub fn media_browser(app: &mut EffectcraftApp, ui: &mut egui::Ui, rect: Rect) { let ctx = ui.ctx().clone(); let p = ui.painter().with_clip_rect(rect); p.rect_filled(rect, 0.0, t.panel_bg); - if !mb::available() { - p.text(rect.center(), Align2::CENTER_CENTER, "The Media Browser needs the desktop app (use File ▸ Import here)", Tokens::ui(12.0), t.text_faint); + let Some(b) = app.session.media_browser() else { + p.text(rect.center(), Align2::CENTER_CENTER, "The Media Browser is not available here (use File ▸ Import)", Tokens::ui(12.0), t.text_faint); app.auto.add("mediaBrowser.unavailable", rect, "Media Browser unavailable"); return; - } + }; let st = app.session.state.media_browser.clone(); - let dir = st.folder.clone().unwrap_or_else(mb::home); + let dir = st.folder.clone().unwrap_or_else(|| b.home()); // Listing cached per folder for a second. let lid = egui::Id::new(("mb-list", dir.clone(), st.importable_only)); let now = ctx.input(|i| i.time); @@ -79,7 +79,7 @@ pub fn media_browser(app: &mut EffectcraftApp, ui: &mut egui::Ui, rect: Rect) { let entries = match listing { Some((at, l)) if now - at < 1.0 => l, _ => { - let l = mb::list(&dir, st.importable_only); + let l = b.list(&dir, st.importable_only); ctx.data_mut(|d| d.insert_temp(lid, (now, l.clone()))); l } @@ -104,6 +104,20 @@ pub fn media_browser(app: &mut EffectcraftApp, ui: &mut egui::Ui, rect: Rect) { if only != st.importable_only { kit::exec(app, "mediaBrowser.go", json!({"path": dir, "importableOnly": only})); } + // The browser's own actions (web: Open Folder…, Add Files…), right-aligned. + let mut ax = rect.max.x - 8.0; + for (id, label) in b.actions().iter().rev() { + let w = 14.0 + label.chars().count() as f32 * 6.4; + ax -= w; + if ax < x + 96.0 { + break; + } + if kit::button(app, ui, Rect::from_min_size(pos2(ax, y), vec2(w, 20.0)), &format!("mediaBrowser.action.{id}"), label, false) { + kit::exec(app, "mediaBrowser.action", json!({"action": id})); + ctx.data_mut(|d| d.remove::<(f64, Result, String>)>(lid)); + } + ax -= 6.0; + } // The path on its own row. let _ = x; let py = y + 24.0; @@ -114,7 +128,7 @@ pub fn media_browser(app: &mut EffectcraftApp, ui: &mut egui::Ui, rect: Rect) { p.line_segment([side.right_top(), side.right_bottom()], Stroke::new(1.0, t.separator)); p.text(pos2(side.min.x + 10.0, side.min.y + 10.0), Align2::LEFT_CENTER, "Favorites", Tokens::semibold(11.5), t.text_dim); let mut fy = side.min.y + 22.0; - let mut places: Vec<(String, String)> = vec![("Home".into(), mb::home())]; + let mut places: Vec<(String, String)> = b.places(); places.extend( st.favorites.iter().map(|f| (std::path::Path::new(f).file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_else(|| f.clone()), f.clone())), ); diff --git a/crates/ui-egui/tests/ui_web_offload.rs b/crates/ui-egui/tests/ui_web_offload.rs new file mode 100644 index 00000000..54ae1a37 --- /dev/null +++ b/crates/ui-egui/tests/ui_web_offload.rs @@ -0,0 +1,204 @@ +//! Headless checks for M13.10 (web app depth), run natively with stand-ins for the browser's +//! workers: +//! +//! - viewer frames go to a [`RemoteFrames`] renderer (the browser's frame worker, here an +//! in-process [`FrameServer`] fed through JSON with project diffs) instead of rendering on the +//! UI thread, the viewer's frame first, and come back identical; +//! - a control-channel `wait: true` command whose job is offloaded replies only when the job +//! ends, while the UI keeps running. + +use std::sync::mpsc::TryRecvError; +use std::sync::{Arc, Mutex}; + +use effectcraft_engine::Session; +use effectcraft_engine::offload::{Inbox, Offload, Post, WorkerRequest}; +use effectcraft_engine::remote::{FrameMsg, FrameReply, FrameServer, Mirror}; +use effectcraft_engine::render::RenderOpts; +use effectcraft_ui_egui::frames::{FrameImage, FrameKey, Frames, RemoteDone, RemoteFrame, RemoteFrames, RemoteJob, RenderSource}; +use effectcraft_ui_egui::{ControlRequest, EffectcraftApp}; +use egui_kittest::Harness; +use serde_json::json; + +/// A frame worker in the same process: messages go through JSON like `postMessage`; rendered +/// frames are held until [`FakeWorker::deliver`] (a busy worker) or dropped by +/// [`FakeWorker::lose`]. +struct FakeWorker { + slots: usize, + state: Mutex<(Mirror, FrameServer, Vec<(RemoteDone, RemoteFrame)>, u64)>, + /// Patch / full-project syncs sent. + syncs: Mutex>, +} + +impl FakeWorker { + fn new(s: &Session, slots: usize) -> FakeWorker { + FakeWorker { slots, state: Mutex::new((Mirror::default(), FrameServer::new(s.footage.clone(), s.expr.clone()), vec![], 0)), syncs: Mutex::default() } + } + fn pending(&self) -> usize { + self.state.lock().unwrap().2.len() + } + fn take(&self) -> Vec<(RemoteDone, RemoteFrame)> { + std::mem::take(&mut self.state.lock().unwrap().2) + } + /// Post every rendered frame back. + fn deliver(&self) { + for (done, f) in self.take() { + done(Ok(f)); + } + } + /// The worker died: its frames never arrive. + fn lose(&self) { + for (done, _) in self.take() { + done(Err("worker lost".into())); + } + } +} + +fn rt serde::Deserialize<'de>>(v: &T) -> T { + serde_json::from_str(&serde_json::to_string(v).unwrap()).unwrap() +} + +impl RemoteFrames for FakeWorker { + fn slots(&self) -> usize { + self.slots + } + fn start(&self, job: RemoteJob, done: RemoteDone) { + let mut st = self.state.lock().unwrap(); + let (mirror, server, pending, next) = &mut *st; + if let Some(m) = mirror.sync(job.revision, &job.project) { + self.syncs.lock().unwrap().push(if matches!(m, FrameMsg::Patch { .. }) { "patch" } else { "project" }); + assert!(server.handle(rt(&m)).is_empty()); + } + *next += 1; + let out = server.handle(rt(&FrameMsg::Render { id: *next, revision: job.revision, comp: job.comp, time: job.t, opts: Box::new(job.opts) })); + match out.into_iter().next() { + Some((FrameReply::Frame { width, height, ms, .. }, Some(rgba))) => pending.push((done, RemoteFrame { width, height, rgba, ms })), + other => done(Err(format!("{:?}", other.map(|o| o.0)))), + } + } +} + +fn source(s: &Session) -> RenderSource { + RenderSource { + project: s.project.clone(), + footage: s.footage.clone(), + expr: s.expr.clone(), + layer_cache: s.layer_cache.clone(), + gpu: None, + gpu_display: false, + disk: None, + } +} + +#[test] +fn viewer_frames_render_remotely_from_project_diffs() { + let mut s = Session::default(); + s.execute("file.openDemoProject", json!({})).unwrap(); + let comp = s.active_comp_id().unwrap(); + let c = s.project.comp(comp).unwrap().clone(); + let opts = RenderOpts { scale: 0.25, guides: true, ..Default::default() }; + let key = |rev: u64, f: i64| FrameKey { revision: rev, comp: comp.0, frame: f, scale: 250, view: 0 }; + let worker = Arc::new(FakeWorker::new(&s, 2)); + let mut frames = Frames::default(); + frames.set_remote(Some(worker.clone())); + assert!(frames.remote_active()); + // Three prefetch frames, then the viewer's frame. + for f in 1..=3 { + frames.request(&source(&s), key(s.revision, f), comp, c.frame_rate.tick_of(f), opts); + } + frames.request_urgent(&source(&s), key(s.revision, 10), comp, c.frame_rate.tick_of(10), opts); + assert_eq!(frames.inflight(), 4); + // Nothing renders until dispatched (no frame threads with a remote renderer); the viewer's + // frame goes first, then prefetch in request order. + assert_eq!(frames.dispatch_remote(), 2); + assert_eq!(frames.remote_busy(), 2); + assert_eq!(worker.pending(), 2); + assert_eq!(frames.dispatch_remote(), 0, "both slots are busy"); + assert_eq!(*worker.syncs.lock().unwrap(), vec!["project"], "the first sync is the whole project"); + // Frames arrive: the cache holds exactly what the page would have rendered. + worker.deliver(); + assert_eq!(frames.remote_busy(), 0); + let img = match frames.get(&key(s.revision, 10)).expect("viewer frame cached") { + FrameImage::Cpu(c) => c, + FrameImage::Gpu(_) => unreachable!(), + }; + let local = effectcraft_ui_egui::frames::to_color_image(&s.render(comp, c.frame_rate.tick_of(10), opts)); + assert_eq!(img.size, local.size); + assert_eq!(img.pixels, local.pixels); + assert!(frames.is_cached(&key(s.revision, 1))); + assert!(!frames.is_cached(&key(s.revision, 2))); + // The rest; an edit later travels as a patch. + frames.dispatch_remote(); + worker.deliver(); + frames.dispatch_remote(); + worker.deliver(); + assert_eq!(frames.inflight(), 0); + s.execute("layer.newSolid", json!({"color": "#ff8800"})).unwrap(); + frames.request_urgent(&source(&s), key(s.revision, 10), comp, c.frame_rate.tick_of(10), opts); + frames.dispatch_remote(); + worker.deliver(); + assert_eq!(worker.syncs.lock().unwrap().last(), Some(&"patch")); + let FrameImage::Cpu(img) = frames.get(&key(s.revision, 10)).unwrap() else { unreachable!() }; + let local = effectcraft_ui_egui::frames::to_color_image(&s.render(comp, c.frame_rate.tick_of(10), opts)); + assert_eq!(img.pixels, local.pixels); + // A lost frame is released (requested again later), not cached. + frames.request(&source(&s), key(s.revision, 20), comp, c.frame_rate.tick_of(20), opts); + frames.dispatch_remote(); + worker.lose(); + assert_eq!(frames.inflight(), 0); + assert!(!frames.is_cached(&key(s.revision, 20))); + // Without slots the remote renderer is unusable (frames render here again). + frames.set_remote(Some(Arc::new(FakeWorker::new(&s, 0)))); + assert!(!frames.remote_active()); +} + +/// Holds offloaded jobs until the test runs them. +#[derive(Default)] +struct Manual(Mutex)>>); + +impl Offload for Manual { + fn start(&self, req: WorkerRequest, inbox: Arc) -> Result<(), String> { + self.0.lock().unwrap().push((req, inbox)); + Ok(()) + } + fn cancel(&self, _: u64) {} +} + +fn exporter() -> Arc { + let sink: Arc) + Send + Sync> = Arc::new(|_: &str, _: Vec| {}); + Arc::new(effectcraft_host::FileExporter { sink: Some(sink) }) +} + +#[test] +fn waiting_commands_reply_when_their_offloaded_job_ends() { + let off = Arc::new(Manual::default()); + let mut s = Session { offload: Some(off.clone()), exporter: Some(exporter()), ..Default::default() }; + s.execute("comp.new", json!({"name": "W", "width": 64, "height": 48, "frameRate": 10, "duration": 1})).unwrap(); + s.execute("layer.newSolid", json!({"color": "#406080"})).unwrap(); + s.execute("renderQueue.add", json!({"format": "gif", "output": "/w.gif"})).unwrap(); + let (tx, rx) = std::sync::mpsc::channel(); + let mut h = Harness::builder().with_size(egui::vec2(1280.0, 800.0)).build_eframe(|_| EffectcraftApp::new(s).with_control(rx)); + h.run_steps(2); + // `renderQueue.render` waits by default: the job goes to the worker, the reply waits. + let (req, reply) = ControlRequest::new("engine.execute", json!({"command": "renderQueue.render", "params": {}})); + tx.send(req).unwrap(); + h.run_steps(4); + assert_eq!(reply.try_recv().unwrap_err(), TryRecvError::Empty, "no reply while the job runs"); + assert!(h.state().session.is_rendering()); + let (job, inbox) = off.0.lock().unwrap().pop().expect("the render went to the worker"); + // The worker runs it. + let mut w = Session { exporter: Some(exporter()), ..Default::default() }; + let post: Post = std::rc::Rc::new(move |r| inbox.push(r)); + effectcraft_engine::offload::run_request(&mut w, job, &post); + h.run_steps(4); + let v = reply.try_recv().expect("replied once the job ended"); + assert_eq!(v["ok"], true, "{v}"); + assert_eq!(v["result"]["rendering"], false, "{v}"); + assert_eq!(v["result"]["items"][0]["status"], "Done", "{v}"); + // `wait: false` replies at once. + h.state_mut().session.execute("renderQueue.add", json!({"format": "gif", "output": "/w2.gif"})).unwrap(); + let (req, reply) = ControlRequest::new("engine.execute", json!({"command": "renderQueue.render", "params": {"wait": false}})); + tx.send(req).unwrap(); + h.run_steps(2); + let v = reply.try_recv().expect("immediate reply"); + assert_eq!(v["result"]["rendering"], true, "{v}"); +}