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 <noreply@anthropic.com>
This commit is contained in:
Brandon Thomas
parent
0f591e0856
commit
5bb6768766
|
|
@ -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();
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<dyn effectcraft_engine::offload::Offload>),
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -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 `<input type=file>`, so every browser) are copied
|
||||
//! there and stay across visits: the recent list.
|
||||
//! - **Folders** (`/Folders/<name>/…`): 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 <name>**).
|
||||
|
||||
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<String>,
|
||||
disconnected: Vec<String>,
|
||||
}
|
||||
|
||||
static STATE: LazyLock<Mutex<State>> = 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<String> {
|
||||
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<Vec<Entry>, 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<String> {
|
||||
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<String> = 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::<Vec<_>>().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<Value, String> {
|
||||
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::<js_sys::Uint8Array>().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}`"))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<String>,
|
||||
busy: Option<(u64, RemoteDone)>,
|
||||
alive: bool,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct State {
|
||||
workers: Vec<Fw>,
|
||||
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<dyn FnMut(u32, JsValue, JsValue, JsValue)>;
|
||||
|
||||
thread_local! {
|
||||
static STATE: RefCell<State> = RefCell::new(State::default());
|
||||
static HANDLER: RefCell<Option<OnMessage>> = const { RefCell::new(None) };
|
||||
/// The worker's frame server (worker side).
|
||||
static SERVER: RefCell<Option<FrameServer>> = 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::<js_sys::Uint8Array>().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:?}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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 | `<input type=file>` 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;
|
||||
|
|
|
|||
|
|
@ -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::<WorkerReply>(&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<effectcraft_media::MediaPool> {
|
||||
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) {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
};
|
||||
|
|
|
|||
|
|
@ -29,38 +29,61 @@ fn jobs_wait(s: &mut Session, _: &Value) -> Result<Value> {
|
|||
|
||||
// ---------------------------------------------------------------- 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<std::sync::Arc<dyn mb::Browser>> {
|
||||
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<Value> {
|
||||
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<Value> = b.places().into_iter().map(|(name, path)| json!({"name": name, "path": path})).collect();
|
||||
let actions: Vec<Value> = 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<Value> {
|
||||
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<Value> {
|
||||
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<Value> {
|
||||
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<Value> {
|
|||
|
||||
fn browser_import(s: &mut Session, p: &Value) -> Result<Value> {
|
||||
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<String> = 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<CommandSpec> {
|
|||
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),
|
||||
|
|
|
|||
|
|
@ -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<Arc<dyn offload::Offload>>,
|
||||
/// Analyses running in the [`Session::offload`] worker.
|
||||
pub offloaded: Vec<offload::OffloadedJob>,
|
||||
/// What the Media Browser browses (`None`: the local file system where there is one).
|
||||
pub browser: Option<Arc<dyn media_browser::Browser>>,
|
||||
/// 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<String>)>,
|
||||
/// Generic background tasks (Content-Aware Fill, Scene Edit Detection), see [`jobs`].
|
||||
pub tasks: Vec<jobs::Task>,
|
||||
/// 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,
|
||||
|
|
|
|||
|
|
@ -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<Vec<Entry>, String>;
|
||||
/// The folder above `dir`.
|
||||
fn parent(&self, dir: &str) -> Option<String> {
|
||||
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<Value, String> {
|
||||
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<Vec<Entry>, 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<std::sync::Arc<dyn Browser>> {
|
||||
self.browser.clone().or_else(|| available().then(|| std::sync::Arc::new(FsBrowser) as std::sync::Arc<dyn Browser>))
|
||||
}
|
||||
}
|
||||
|
||||
/// One entry of a [`VirtualTree`].
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
struct VNode {
|
||||
/// What importing it reads (`None`: a folder).
|
||||
import: Option<String>,
|
||||
size: u64,
|
||||
modified: Option<u64>,
|
||||
}
|
||||
|
||||
/// 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<String, VNode>,
|
||||
}
|
||||
|
||||
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<u64>) {
|
||||
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<Vec<Entry>, 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<Entry> = 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::<Vec<_>>();
|
||||
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");
|
||||
|
|
|
|||
|
|
@ -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<PropGroup>,
|
||||
message: String,
|
||||
},
|
||||
/// Roto Brush segmentations computed so far (propagation), not sent before.
|
||||
Segs {
|
||||
segs: Vec<SegData>,
|
||||
},
|
||||
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<u64>) -> Vec<SegData> {
|
||||
let new: Vec<(u64, Arc<effectcraft_track::roto::FrameSeg>)> =
|
||||
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<Vec<WorkerReply>>);
|
||||
|
|
@ -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<u64> = 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::<web_time::Instant>;
|
||||
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<Inbox>,
|
||||
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,
|
||||
] {
|
||||
|
|
|
|||
|
|
@ -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<Step>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub value: Option<Value>,
|
||||
}
|
||||
|
||||
/// The changes that turn `old` into `new` (empty when equal).
|
||||
pub fn diff(old: &Value, new: &Value) -> Vec<Op> {
|
||||
let mut ops = vec![];
|
||||
diff_into(old, new, &mut vec![], &mut ops);
|
||||
ops
|
||||
}
|
||||
|
||||
fn diff_into(old: &Value, new: &Value, path: &mut Vec<Step>, ops: &mut Vec<Op>) {
|
||||
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<Op> },
|
||||
/// Render a frame of the current project.
|
||||
Render { id: u64, revision: u64, comp: ItemId, time: Tick, opts: Box<RenderOpts> },
|
||||
/// Roto Brush segmentations computed elsewhere (a propagation worker).
|
||||
Segs { segs: Vec<SegData> },
|
||||
/// 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<u64> {
|
||||
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<FrameMsg> {
|
||||
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<u8> {
|
||||
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<dyn FootageSource>,
|
||||
pub expr: Option<Arc<dyn ExprHost>>,
|
||||
pub cache: Arc<LayerCache>,
|
||||
doc: Value,
|
||||
project: Option<Arc<Project>>,
|
||||
revision: Option<u64>,
|
||||
}
|
||||
|
||||
impl FrameServer {
|
||||
pub fn new(footage: Arc<dyn FootageSource>, expr: Option<Arc<dyn ExprHost>>) -> FrameServer {
|
||||
FrameServer { footage, expr, cache: Arc::new(LayerCache::default()), doc: Value::Null, project: None, revision: None }
|
||||
}
|
||||
|
||||
pub fn revision(&self) -> Option<u64> {
|
||||
self.revision
|
||||
}
|
||||
|
||||
pub fn project(&self) -> Option<&Arc<Project>> {
|
||||
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<Vec<u8>>)> {
|
||||
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<Vec<u8>>)> {
|
||||
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<T: Serialize + for<'de> 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<u8> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<RotoProgress> {
|
||||
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<fx::Chain> {
|
||||
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() {
|
||||
|
|
|
|||
|
|
@ -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<crate::media_browser::VirtualTree>,
|
||||
fetched: Mutex<Vec<String>>,
|
||||
actions: Mutex<Vec<String>>,
|
||||
}
|
||||
|
||||
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<Vec<crate::media_browser::Entry>, String> {
|
||||
self.tree.lock().unwrap().list(dir, only)
|
||||
}
|
||||
fn fetch(&self, paths: &[String], _: &Value) -> bool {
|
||||
// Opened folders' files arrive later.
|
||||
let pending: Vec<String> = 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<Value, String> {
|
||||
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());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<dyn effectcraft_render::FootageSource>);
|
||||
|
||||
impl crate::offload::Offload for InlineWorker {
|
||||
fn start(&self, req: crate::offload::WorkerRequest, inbox: std::sync::Arc<crate::offload::Inbox>) -> 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());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<u8> {
|
||||
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<FrameSeg> {
|
||||
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<Vec<u8>> {
|
||||
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).
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
|
|
@ -55,6 +55,87 @@ pub enum Outcome {
|
|||
path: Option<String>,
|
||||
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<String>,
|
||||
/// 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<Outcome, String> {
|
||||
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<String> = 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<String> = 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<String> = 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<Value> = 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::<Vec<_>>()).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)
|
||||
|
|
|
|||
|
|
@ -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<Project>,
|
||||
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<u8>,
|
||||
/// Render time (ms).
|
||||
pub ms: f64,
|
||||
}
|
||||
|
||||
/// Called once with a remote frame (or why there is none).
|
||||
pub type RemoteDone = Box<dyn FnOnce(Result<RemoteFrame, String>) + 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<Mutex<Cache>>,
|
||||
/// 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<Mutex<f64>>,
|
||||
content_keys: ContentKeys,
|
||||
/// Where frames render instead of the UI thread (wasm32), and how many it is rendering.
|
||||
remote: Option<Arc<dyn RemoteFrames>>,
|
||||
remote_busy: Arc<Mutex<usize>>,
|
||||
}
|
||||
|
||||
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<Arc<dyn RemoteFrames>>) {
|
||||
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<RemoteFrame, String>| {
|
||||
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<FrameImage> {
|
||||
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<usize> {
|
||||
jobs.iter().enumerate().max_by_key(|(_, j)| (j.urgent, if j.urgent { j.seq as i64 } else { -(j.seq as i64) })).map(|(i, _)| i)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -143,6 +143,8 @@ pub struct EffectcraftApp {
|
|||
pub(crate) deferred: Vec<(ControlRequest, f64)>,
|
||||
pub(crate) synthetic: Vec<egui::Event>,
|
||||
pub(crate) input_waiters: Vec<Sender<serde_json::Value>>,
|
||||
/// Control requests waiting for an offloaded job (`wait: true` in the browser).
|
||||
pub(crate) job_waiters: Vec<(control::JobWaiter, Sender<serde_json::Value>)>,
|
||||
queued_screenshots: Vec<(u64, f64, u32)>,
|
||||
pending_screenshots: Vec<(u64, Option<String>, Option<[f32; 4]>, Sender<serde_json::Value>, 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) {
|
||||
|
|
|
|||
|
|
@ -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<Vec<mb::Entry>, 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())),
|
||||
);
|
||||
|
|
|
|||
|
|
@ -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<Vec<&'static str>>,
|
||||
}
|
||||
|
||||
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<T: serde::Serialize + for<'de> 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<Vec<(WorkerRequest, Arc<Inbox>)>>);
|
||||
|
||||
impl Offload for Manual {
|
||||
fn start(&self, req: WorkerRequest, inbox: Arc<Inbox>) -> Result<(), String> {
|
||||
self.0.lock().unwrap().push((req, inbox));
|
||||
Ok(())
|
||||
}
|
||||
fn cancel(&self, _: u64) {}
|
||||
}
|
||||
|
||||
fn exporter() -> Arc<effectcraft_host::FileExporter> {
|
||||
let sink: Arc<dyn Fn(&str, Vec<u8>) + Send + Sync> = Arc::new(|_: &str, _: Vec<u8>| {});
|
||||
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}");
|
||||
}
|
||||
Loading…
Reference in New Issue