1134 lines
47 KiB
Rust
1134 lines
47 KiB
Rust
//! Background jobs where the platform has no threads (the browser).
|
|
//!
|
|
//! On the desktop, renders and analyses run on threads that share the session's project and
|
|
//! footage. A wasm build without atomics has one thread per instance, so the web app runs a
|
|
//! second engine instance in a Web Worker instead and talks to it with messages: an
|
|
//! [`Offload`] (set as [`Session::offload`]) ships a [`WorkerRequest`] — the serialized project,
|
|
//! the footage files it reads and the job — and feeds the worker's [`WorkerReply`]s back into an
|
|
//! [`Inbox`] that the session drains every frame ([`Session::poll_render`],
|
|
//! [`Session::poll_offload`]). The worker side is [`run_request_async`]: a plain session fed the
|
|
//! project runs the job and reports through a callback. With a WebGPU device of the worker's
|
|
//! own (deferred readbacks, `effectcraft_render::passes`) the job's frames render on the GPU in
|
|
//! passes, the job awaiting the device between them; [`run_request`] runs it blocking.
|
|
//!
|
|
//! | Job | Worker runs | Reply applied on the UI thread |
|
|
//! |---|---|---|
|
|
//! | Render Queue | the export of every queued item | item status, progress; files arrive through the host |
|
|
//! | Warp Stabilizer analysis | [`Session::start_warp`], then the stabilization plan | the effect's property group (one undo step); the plan's summary ([`WorkerReply::WarpPlan`]) |
|
|
//! | Warp Stabilizer stabilization (settings changed after the analysis) | the plan ([`effectcraft_effects::warp_stab::summary_at`]) | its summary, for `warp.status` |
|
|
//! | 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 |
|
|
//! | Content-Aware Fill | the fill ([`crate::commands::content_fill::FillPlan`]) | its PNG files (to the host's storage), then the fill layer (one undo step) |
|
|
//!
|
|
//! 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".
|
|
|
|
use std::cell::{Cell, RefCell};
|
|
use std::future::Future;
|
|
use std::pin::Pin;
|
|
use std::rc::Rc;
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use effectcraft_keyframe::ShapePath;
|
|
use effectcraft_project::render_queue::{RenderQueueItem, RenderStatus};
|
|
use effectcraft_project::tracking::TrackerSettings;
|
|
use effectcraft_project::{ItemId, ItemKind, LayerId, Project, PropGroup, Uid};
|
|
use effectcraft_render::{Accelerator, Backend, RenderOpts};
|
|
use effectcraft_time::Tick;
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
use crate::mask_track::MaskMethod;
|
|
use crate::render_queue::{ItemUpdate, JobShared, JobState};
|
|
use crate::tracking::PointValues;
|
|
use crate::{Event, Session};
|
|
|
|
/// One job for a worker.
|
|
#[derive(Clone, Debug, Serialize, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct WorkerRequest {
|
|
pub id: u64,
|
|
/// The project ([`Project::to_json`]).
|
|
pub project: String,
|
|
/// Footage files the job reads; the host ships their bytes before the request.
|
|
pub files: Vec<String>,
|
|
pub job: WorkerJob,
|
|
}
|
|
|
|
#[derive(Clone, Debug, Serialize, Deserialize)]
|
|
#[serde(tag = "kind", rename_all = "camelCase")]
|
|
pub enum WorkerJob {
|
|
/// Render Queue: per item, one (item with that output module, resolved output path) per
|
|
/// output module.
|
|
Render {
|
|
items: Vec<Vec<(RenderQueueItem, String)>>,
|
|
},
|
|
Warp {
|
|
comp: ItemId,
|
|
layer: LayerId,
|
|
effect: Uid,
|
|
},
|
|
/// The stabilization plan of an analysed Warp Stabilizer at comp time `time` (its settings
|
|
/// there): seconds of solving for a long Subspace Warp clip, so not on the page. The
|
|
/// summary comes back as [`WorkerReply::WarpPlan`].
|
|
WarpPlan {
|
|
comp: ItemId,
|
|
layer: LayerId,
|
|
effect: Uid,
|
|
time: Tick,
|
|
},
|
|
Camera {
|
|
comp: ItemId,
|
|
layer: LayerId,
|
|
effect: Uid,
|
|
},
|
|
RotoFreeze {
|
|
comp: ItemId,
|
|
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,
|
|
tracker: Uid,
|
|
direction: crate::tracking::Direction,
|
|
settings: TrackerSettings,
|
|
points: Vec<PointValues>,
|
|
times: Vec<Tick>,
|
|
},
|
|
MaskTrack {
|
|
comp: ItemId,
|
|
layer: LayerId,
|
|
mask: Uid,
|
|
direction: crate::tracking::Direction,
|
|
path: ShapePath,
|
|
method: MaskMethod,
|
|
times: Vec<Tick>,
|
|
},
|
|
/// Content-Aware Fill: the files go to the host's storage, the plan and their paths come
|
|
/// back as [`WorkerReply::Fill`].
|
|
ContentFill {
|
|
plan: Box<crate::commands::content_fill::FillPlan>,
|
|
},
|
|
}
|
|
|
|
impl WorkerJob {
|
|
pub fn kind(&self) -> JobKind {
|
|
match self {
|
|
WorkerJob::Render { .. } => JobKind::Render,
|
|
WorkerJob::Warp { .. } => JobKind::Warp,
|
|
WorkerJob::WarpPlan { .. } => JobKind::WarpPlan,
|
|
WorkerJob::Camera { .. } => JobKind::Camera,
|
|
WorkerJob::RotoFreeze { .. } => JobKind::RotoFreeze,
|
|
WorkerJob::RotoPropagate { .. } => JobKind::RotoPropagate,
|
|
WorkerJob::Track { .. } => JobKind::Track,
|
|
WorkerJob::MaskTrack { .. } => JobKind::MaskTrack,
|
|
WorkerJob::ContentFill { .. } => JobKind::ContentFill,
|
|
}
|
|
}
|
|
|
|
/// (comp, layer, property group uid) an analysis writes to.
|
|
pub fn target(&self) -> Option<(ItemId, LayerId, Uid)> {
|
|
match *self {
|
|
WorkerJob::Render { .. } => None,
|
|
WorkerJob::Warp { comp, layer, effect }
|
|
| WorkerJob::WarpPlan { 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)),
|
|
WorkerJob::ContentFill { ref plan } => Some((plan.comp, plan.layer, 0)),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub enum JobKind {
|
|
Render,
|
|
Warp,
|
|
WarpPlan,
|
|
Camera,
|
|
RotoFreeze,
|
|
RotoPropagate,
|
|
Track,
|
|
MaskTrack,
|
|
ContentFill,
|
|
}
|
|
|
|
impl JobKind {
|
|
/// The undo step an analysis result becomes.
|
|
pub fn label(self) -> &'static str {
|
|
match self {
|
|
JobKind::Render => "Render",
|
|
JobKind::Warp => "Warp Stabilizer Analysis",
|
|
JobKind::WarpPlan => "Warp Stabilizer: Stabilizing",
|
|
JobKind::Camera => "3D Camera Tracker Analysis",
|
|
JobKind::RotoFreeze => "Freeze",
|
|
JobKind::RotoPropagate => "Roto Brush Propagation",
|
|
JobKind::Track => "Analyze Track",
|
|
JobKind::MaskTrack => "Track Mask",
|
|
JobKind::ContentFill => "Content-Aware Fill",
|
|
}
|
|
}
|
|
}
|
|
|
|
/// What a worker reports.
|
|
#[derive(Clone, Debug, Serialize, Deserialize)]
|
|
#[serde(tag = "type", rename_all = "camelCase")]
|
|
pub enum WorkerReply {
|
|
/// Render Queue progress.
|
|
Render {
|
|
state: JobState,
|
|
},
|
|
/// A Render Queue item started or finished.
|
|
Item {
|
|
update: ItemUpdate,
|
|
},
|
|
/// Analysis progress (frames).
|
|
Progress {
|
|
done: u64,
|
|
total: u64,
|
|
},
|
|
/// An analysis finished: its property group after the job, and the job's message.
|
|
Group {
|
|
comp: ItemId,
|
|
layer: LayerId,
|
|
group: Box<PropGroup>,
|
|
message: String,
|
|
},
|
|
/// A Warp Stabilizer plan's summary ([`effectcraft_effects::warp_stab::summary_key`]): the
|
|
/// page keeps it for `warp.status` instead of solving the plan itself.
|
|
WarpPlan {
|
|
key: [u64; 2],
|
|
summary: Box<effectcraft_effects::warp_stab::PlanSummary>,
|
|
},
|
|
/// Roto Brush segmentations computed so far (propagation), not sent before.
|
|
Segs {
|
|
segs: Vec<SegData>,
|
|
},
|
|
/// A Content-Aware Fill finished: its plan and the PNG files written (already in the
|
|
/// host's storage).
|
|
Fill {
|
|
plan: Box<crate::commands::content_fill::FillPlan>,
|
|
files: Vec<String>,
|
|
},
|
|
Failed {
|
|
error: String,
|
|
},
|
|
/// The job ended (always the last reply).
|
|
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>>);
|
|
|
|
impl Inbox {
|
|
pub fn push(&self, r: WorkerReply) {
|
|
self.0.lock().unwrap_or_else(|e| e.into_inner()).push(r);
|
|
}
|
|
pub fn take(&self) -> Vec<WorkerReply> {
|
|
std::mem::take(&mut *self.0.lock().unwrap_or_else(|e| e.into_inner()))
|
|
}
|
|
}
|
|
|
|
/// Runs jobs off the UI thread (the web app: a pool of Web Workers).
|
|
pub trait Offload: Send + Sync {
|
|
/// Start `req`; replies go to `inbox` (ending with [`WorkerReply::Done`]).
|
|
fn start(&self, req: WorkerRequest, inbox: Arc<Inbox>) -> Result<(), String>;
|
|
/// Stop job `id` now (the worker is terminated; no more replies arrive).
|
|
fn cancel(&self, id: u64);
|
|
}
|
|
|
|
static NEXT_ID: AtomicU64 = AtomicU64::new(1);
|
|
|
|
pub fn next_id() -> u64 {
|
|
NEXT_ID.fetch_add(1, Ordering::Relaxed)
|
|
}
|
|
|
|
/// Footage files a project reads (paths, sequence frames).
|
|
pub fn footage_files(p: &Project) -> Vec<String> {
|
|
let mut v: Vec<String> = p
|
|
.items
|
|
.values()
|
|
.filter_map(|i| match &i.kind {
|
|
ItemKind::Footage(f) if !f.path.is_empty() => Some(std::iter::once(f.path.clone()).chain(f.sequence.iter().cloned()).collect::<Vec<_>>()),
|
|
_ => None,
|
|
})
|
|
.flatten()
|
|
.collect();
|
|
v.sort();
|
|
v.dedup();
|
|
v
|
|
}
|
|
|
|
// ---------------------------------------------------------------- progress hook (worker side)
|
|
|
|
thread_local! {
|
|
static HOOK: RefCell<Option<Box<dyn FnMut(u64, u64)>>> = const { RefCell::new(None) };
|
|
}
|
|
|
|
/// Analyses call this as frames complete; a worker forwards it (no-op elsewhere).
|
|
pub fn report(done: u64, total: u64) {
|
|
HOOK.with(|h| {
|
|
if let Ok(mut h) = h.try_borrow_mut()
|
|
&& let Some(f) = h.as_mut()
|
|
{
|
|
f(done, total);
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Run `f` (a future: an analysis awaited to its end) with `hook` receiving its progress.
|
|
async fn with_hook<R>(hook: Box<dyn FnMut(u64, u64)>, f: impl Future<Output = R>) -> R {
|
|
HOOK.with(|h| *h.borrow_mut() = Some(hook));
|
|
let r = f.await;
|
|
HOOK.with(|h| *h.borrow_mut() = None);
|
|
r
|
|
}
|
|
|
|
// ---------------------------------------------------------------- GPU in job workers
|
|
|
|
/// A boxed analysis future (see [`run_or_queue`]).
|
|
type Queued = Pin<Box<dyn Future<Output = ()>>>;
|
|
|
|
thread_local! {
|
|
/// The job worker's accelerator while it runs a request: a WebGPU device with deferred
|
|
/// readbacks (see [`job_accel`]).
|
|
static JOB_ACCEL: RefCell<Option<Arc<dyn Accelerator>>> = const { RefCell::new(None) };
|
|
/// Analyses started while [`run_request_async`] runs a request on this thread: awaited by it
|
|
/// (`None`: not inside one, analyses run to completion when started).
|
|
static QUEUED: RefCell<Option<Vec<Queued>>> = const { RefCell::new(None) };
|
|
}
|
|
|
|
/// The accelerator analyses render their input frames with: the job worker's GPU while it runs
|
|
/// a request ([`run_request_async`] with a session whose accelerator defers its readbacks).
|
|
/// Elsewhere (the desktop's analysis threads) analyses render on the CPU, as before.
|
|
pub(crate) fn job_accel() -> Option<Arc<dyn Accelerator>> {
|
|
JOB_ACCEL.with(|a| a.borrow().clone())
|
|
}
|
|
|
|
/// Render options of an analysis's input frames: 100 %; with an accelerator, the project's
|
|
/// renderer (Mercury GPU Acceleration: its GPU effects on the device; Software Only: the CPU).
|
|
pub(crate) fn analysis_opts(accel: Option<&dyn Accelerator>) -> RenderOpts {
|
|
RenderOpts { scale: 1.0, backend: if accel.is_some() { Backend::Auto } else { Backend::Cpu }, ..Default::default() }
|
|
}
|
|
|
|
/// A render in passes ([`effectcraft_render::passes::in_passes`]): its value.
|
|
pub(crate) async fn in_passes<'a, T>(accel: Option<&'a dyn Accelerator>, render: impl FnMut(Option<&'a dyn Accelerator>) -> T) -> T {
|
|
effectcraft_render::passes::in_passes(accel, render).await.value
|
|
}
|
|
|
|
/// `work()` while the next frame renders (`fetch`): side by side on the thread pool, or — with
|
|
/// `deferred` readbacks, where a frame renders in passes awaiting the GPU — one after the other.
|
|
pub(crate) async fn join_fetch<R: Send, F: Send, Fut: Future<Output = F>>(
|
|
deferred: bool,
|
|
work: impl FnOnce() -> R + Send,
|
|
fetch: impl FnOnce() -> Fut + Send,
|
|
) -> (R, F) {
|
|
if deferred {
|
|
let r = work();
|
|
return (r, fetch().await);
|
|
}
|
|
rayon::join(work, || effectcraft_render::passes::block_on(fetch()))
|
|
}
|
|
|
|
/// Run an analysis started with `wait`: inside [`run_request_async`] it is queued and awaited
|
|
/// there (its frames may render in passes); anywhere else it runs to completion now.
|
|
pub(crate) fn run_or_queue(f: impl Future<Output = ()> + 'static) {
|
|
let f: Queued = Box::pin(f);
|
|
let now = QUEUED.with(|q| match q.borrow_mut().as_mut() {
|
|
Some(v) => {
|
|
v.push(f);
|
|
None
|
|
}
|
|
None => Some(f),
|
|
});
|
|
if let Some(f) = now {
|
|
effectcraft_render::passes::block_on(f);
|
|
}
|
|
}
|
|
|
|
/// Await the analyses [`run_or_queue`] queued, then let the session apply their results.
|
|
async fn drain(s: &mut Session) {
|
|
loop {
|
|
let next = QUEUED.with(|q| q.borrow_mut().as_mut().and_then(|v| (!v.is_empty()).then(|| v.remove(0))));
|
|
match next {
|
|
Some(f) => f.await,
|
|
None => break,
|
|
}
|
|
}
|
|
s.poll_track();
|
|
s.poll_mask_track();
|
|
s.poll_warp(false);
|
|
s.poll_camera(false);
|
|
s.poll_roto(false);
|
|
}
|
|
|
|
// ---------------------------------------------------------------- worker side
|
|
|
|
/// Where a worker sends its replies.
|
|
pub type Post = Rc<dyn Fn(WorkerReply)>;
|
|
|
|
/// Run `req` on `s` (a session with footage, expressions and an exporter, but no project yet),
|
|
/// blocking, reporting through `post`. Always ends with [`WorkerReply::Done`].
|
|
pub fn run_request(s: &mut Session, req: WorkerRequest, post: &Post) {
|
|
effectcraft_render::passes::block_on(run_request_async(s, req, post));
|
|
}
|
|
|
|
/// [`run_request`] as a future: with an accelerator whose readbacks are deferred
|
|
/// ([`Session::accel`]: a browser job worker's own WebGPU device) the Render Queue's frames
|
|
/// and the analyses' input frames render on it in passes, awaiting the device between them
|
|
/// (the browser delivers readbacks only while the worker is back in its event loop).
|
|
pub async fn run_request_async(s: &mut Session, req: WorkerRequest, post: &Post) {
|
|
let accel = s.accel.clone().filter(|a| a.miss_gate().is_some());
|
|
if let Some(a) = &accel {
|
|
// Buffers computed from placeholders of a pass that missed are not kept.
|
|
s.layer_cache.set_gate(a.miss_gate());
|
|
}
|
|
JOB_ACCEL.with(|a| *a.borrow_mut() = accel);
|
|
QUEUED.with(|q| *q.borrow_mut() = Some(vec![]));
|
|
match Project::from_json(&req.project) {
|
|
Ok(p) => {
|
|
s.replace_project(p, None);
|
|
if let Err(e) = run_job(s, req.job, post.clone()).await {
|
|
post(WorkerReply::Failed { error: e });
|
|
}
|
|
}
|
|
Err(e) => post(WorkerReply::Failed { error: format!("project: {e}") }),
|
|
}
|
|
QUEUED.with(|q| *q.borrow_mut() = None);
|
|
JOB_ACCEL.with(|a| *a.borrow_mut() = None);
|
|
post(WorkerReply::Done);
|
|
}
|
|
|
|
async fn run_job(s: &mut Session, job: WorkerJob, post: Post) -> Result<(), String> {
|
|
let kind = job.kind();
|
|
let target = job.target();
|
|
match job {
|
|
WorkerJob::Render { items } => {
|
|
let exporter = s.exporter.clone().ok_or("export is not available in this build")?;
|
|
let shared = JobShared::default();
|
|
// Item updates and progress go out as they happen (progress at most every 100 ms).
|
|
let last = Cell::new(None::<web_time::Instant>);
|
|
let notify = |sh: &JobShared, force: bool| {
|
|
let ups = std::mem::take(&mut *crate::render_queue::lock(&sh.updates));
|
|
for u in ups {
|
|
post(WorkerReply::Item { update: u });
|
|
}
|
|
if force || last.get().is_none_or(|t| t.elapsed().as_millis() >= 100) {
|
|
last.set(Some(web_time::Instant::now()));
|
|
post(WorkerReply::Render { state: sh.snapshot() });
|
|
}
|
|
};
|
|
crate::render_queue::run_items(s, exporter, items, &shared, &mut |sh| notify(sh, false)).await;
|
|
notify(&shared, true);
|
|
Ok(())
|
|
}
|
|
WorkerJob::Warp { comp, layer, effect } => analysis(s, kind, target, post, |s| s.start_warp(comp, layer, effect, true).map(|_| ())).await,
|
|
WorkerJob::WarpPlan { comp, layer, effect, time } => {
|
|
let r = warp_plan(s, comp, layer, effect, time).ok_or("the Warp Stabilizer is not analysed")?;
|
|
post(r);
|
|
Ok(())
|
|
}
|
|
WorkerJob::Camera { comp, layer, effect } => analysis(s, kind, target, post, |s| s.start_camera(comp, layer, effect, true).map(|_| ())).await,
|
|
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(|_| ())
|
|
})
|
|
.await
|
|
}
|
|
WorkerJob::RotoPropagate { comp, layer, effect, direction } => propagate(s, comp, layer, effect, direction, post).await,
|
|
WorkerJob::ContentFill { plan } => fill(s, *plan, post).await,
|
|
WorkerJob::Track { comp, layer, tracker, direction, settings, points, times } => {
|
|
analysis(s, kind, target, post, |s| {
|
|
let work = crate::tracking::Work {
|
|
project: s.project.clone(),
|
|
footage: s.footage.clone(),
|
|
expr: s.expr.clone(),
|
|
cache: s.layer_cache.clone(),
|
|
comp,
|
|
layer,
|
|
settings,
|
|
points,
|
|
times,
|
|
};
|
|
s.start_track(work, tracker, direction, true)
|
|
})
|
|
.await
|
|
}
|
|
WorkerJob::MaskTrack { comp, layer, mask, direction, path, method, times } => {
|
|
analysis(s, kind, target, post, |s| {
|
|
let work = crate::mask_track::MaskWork {
|
|
project: s.project.clone(),
|
|
footage: s.footage.clone(),
|
|
expr: s.expr.clone(),
|
|
cache: s.layer_cache.clone(),
|
|
comp,
|
|
layer,
|
|
path,
|
|
method,
|
|
times,
|
|
face_model: if method.is_face() { s.models.face() } else { None },
|
|
};
|
|
s.start_mask_track(work, mask, direction, true)
|
|
})
|
|
.await
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Content-Aware Fill in a worker: the layer's frames render on the worker's GPU (in passes),
|
|
/// the files are written through the session's services (the browser: back to the page's
|
|
/// storage) and the plan comes back for the page to add the fill layer.
|
|
async fn fill(s: &mut Session, plan: crate::commands::content_fill::FillPlan, post: Post) -> Result<(), String> {
|
|
let p = post.clone();
|
|
let mut last = None::<web_time::Instant>;
|
|
let mut progress = move |done: u64, total: u64| {
|
|
if last.is_none_or(|t: web_time::Instant| t.elapsed().as_millis() >= 100) {
|
|
last = Some(web_time::Instant::now());
|
|
p(WorkerReply::Progress { done, total });
|
|
}
|
|
true
|
|
};
|
|
let accel = job_accel();
|
|
let files = crate::commands::content_fill::run_plan(
|
|
&plan,
|
|
&s.project,
|
|
s.footage.as_ref(),
|
|
s.expr.as_deref(),
|
|
&s.layer_cache,
|
|
s.services.as_ref(),
|
|
accel.as_deref(),
|
|
&mut progress,
|
|
&mut |_| {},
|
|
)
|
|
.await?;
|
|
post(WorkerReply::Fill { plan: Box::new(plan), files });
|
|
Ok(())
|
|
}
|
|
|
|
/// 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.
|
|
async 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, async {
|
|
let r = s.start_roto(comp, layer, effect, crate::roto::RotoTask::Propagate, dir, true);
|
|
drain(s).await;
|
|
r
|
|
})
|
|
.await?;
|
|
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(())
|
|
}
|
|
|
|
async fn analysis(
|
|
s: &mut Session,
|
|
kind: JobKind,
|
|
target: Option<(ItemId, LayerId, Uid)>,
|
|
post: Post,
|
|
run: impl FnOnce(&mut Session) -> Result<(), String>,
|
|
) -> Result<(), String> {
|
|
let (comp, layer, uid) = target.ok_or("no target")?;
|
|
let rev = s.revision;
|
|
// Progress at most every 100 ms.
|
|
let p = post.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() >= 100) {
|
|
last = Some(web_time::Instant::now());
|
|
p(WorkerReply::Progress { done, total });
|
|
}
|
|
});
|
|
with_hook(hook, async {
|
|
let r = run(s);
|
|
// The analysis was queued: await it, then the session writes its result.
|
|
drain(s).await;
|
|
r
|
|
})
|
|
.await?;
|
|
let message = s
|
|
.events
|
|
.iter()
|
|
.rev()
|
|
.find_map(|e| match e {
|
|
Event::Toast { message, .. } => Some(message.clone()),
|
|
_ => None,
|
|
})
|
|
.unwrap_or_default();
|
|
let failed = s.events.iter().any(|e| matches!(e, Event::Toast { error: true, .. }));
|
|
if failed {
|
|
return Err(message);
|
|
}
|
|
if s.revision == rev && kind != JobKind::Track && kind != JobKind::MaskTrack {
|
|
// Nothing written (e.g. the analysis was empty).
|
|
return Err(if message.is_empty() { "nothing was analysed".into() } else { message });
|
|
}
|
|
let group = s.project.comp(comp).and_then(|c| c.layer(layer)).and_then(|l| l.props.find_group(uid)).cloned().ok_or("the analysed property is gone")?;
|
|
// The stabilization plan too (before the group, so the page has it once it shows the
|
|
// analysis): solving it would otherwise stall the page at its first `warp.status`.
|
|
if kind == JobKind::Warp
|
|
&& let Some(r) = warp_plan(s, comp, layer, uid, s.time_of(comp))
|
|
{
|
|
post(r);
|
|
}
|
|
post(WorkerReply::Group { comp, layer, group: Box::new(group), message });
|
|
Ok(())
|
|
}
|
|
|
|
/// The Warp Stabilizer plan summary of (comp, layer, effect) with its settings at comp time `t`
|
|
/// (computing the plan), as a reply.
|
|
fn warp_plan(s: &Session, comp: ItemId, layer: LayerId, effect: Uid, t: Tick) -> Option<WorkerReply> {
|
|
let c = s.project.comp(comp)?;
|
|
let l = c.layer(layer)?;
|
|
let g = l.props.find_group(effect)?;
|
|
let ctx = effectcraft_render::EvalCtx::new(&s.project, comp, c, t);
|
|
let params = effectcraft_effects::flatten_params(g, &mut |pr| ctx.value(l, pr));
|
|
let (key, summary) = effectcraft_effects::warp_stab::summary(¶ms)?;
|
|
Some(WorkerReply::WarpPlan { key, summary: Box::new((*summary).clone()) })
|
|
}
|
|
|
|
// ---------------------------------------------------------------- UI side
|
|
|
|
/// Analysis progress of an offloaded job.
|
|
#[derive(Clone, Debug, Default, Serialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct OffloadProgress {
|
|
pub done: u64,
|
|
pub total: u64,
|
|
pub elapsed: f64,
|
|
pub finished: bool,
|
|
pub cancelled: bool,
|
|
pub error: Option<String>,
|
|
}
|
|
|
|
/// Frames per second of `done` frames in `elapsed` seconds.
|
|
pub(crate) fn rate(done: u64, elapsed: f64) -> f64 {
|
|
if elapsed > 0.0 { done as f64 / elapsed } else { 0.0 }
|
|
}
|
|
|
|
impl OffloadProgress {
|
|
pub(crate) fn track(&self) -> crate::tracking::TrackProgress {
|
|
crate::tracking::TrackProgress { done: self.done, total: self.total, elapsed: self.elapsed, fps: rate(self.done, self.elapsed), ..Default::default() }
|
|
}
|
|
}
|
|
|
|
/// An analysis running in a worker.
|
|
pub struct OffloadedJob {
|
|
pub id: u64,
|
|
pub kind: JobKind,
|
|
pub comp: ItemId,
|
|
pub layer: LayerId,
|
|
pub uid: Uid,
|
|
pub inbox: Arc<Inbox>,
|
|
pub progress: OffloadProgress,
|
|
/// Segmentations received (Roto Brush propagation).
|
|
pub segs: u64,
|
|
started: web_time::Instant,
|
|
}
|
|
|
|
/// A Render Queue job running in a worker (see [`crate::render_queue::RenderJob`]).
|
|
pub struct RemoteRender {
|
|
pub id: u64,
|
|
pub inbox: Arc<Inbox>,
|
|
/// The queue items in the job.
|
|
pub ids: Vec<u64>,
|
|
}
|
|
|
|
impl Session {
|
|
/// Whether jobs started without `wait` go to the [`Offload`].
|
|
pub fn offloads(&self) -> bool {
|
|
self.offload.is_some()
|
|
}
|
|
|
|
/// The offloaded analysis of `kind`, if one runs.
|
|
pub fn offloaded(&self, kind: JobKind) -> Option<&OffloadedJob> {
|
|
self.offloaded.iter().find(|j| j.kind == kind && !j.progress.finished)
|
|
}
|
|
|
|
/// Send an analysis to the worker.
|
|
pub(crate) fn offload_analysis(&mut self, job: WorkerJob) -> Result<(), String> {
|
|
let off = self.offload.clone().ok_or("no background worker")?;
|
|
let kind = job.kind();
|
|
let (comp, layer, uid) = job.target().ok_or("not an analysis")?;
|
|
if self.offloaded(kind).is_some() {
|
|
return Err(format!("a {} job is already running", kind.label()));
|
|
}
|
|
let id = next_id();
|
|
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(),
|
|
segs: 0,
|
|
started: web_time::Instant::now(),
|
|
});
|
|
self.events.push(Event::ProjectChanged { revision: self.revision });
|
|
Ok(())
|
|
}
|
|
|
|
/// Cancel the offloaded analysis of `kind` (nothing is written).
|
|
pub fn cancel_offloaded(&mut self, kind: JobKind) -> bool {
|
|
let off = self.offload.clone();
|
|
let Some(j) = self.offloaded.iter_mut().find(|j| j.kind == kind && !j.progress.finished) else { return false };
|
|
if let Some(off) = off {
|
|
off.cancel(j.id);
|
|
}
|
|
j.progress.cancelled = true;
|
|
j.progress.finished = true;
|
|
true
|
|
}
|
|
|
|
/// Apply worker replies to offloaded analyses (results become one undo step each) and drop
|
|
/// finished jobs. Frontends call this every frame while [`Session::offloaded`] is not empty.
|
|
pub fn poll_offload(&mut self) -> bool {
|
|
if self.offloaded.is_empty() {
|
|
return false;
|
|
}
|
|
let mut changed = false;
|
|
let mut results = vec![];
|
|
let mut fills = vec![];
|
|
for j in &mut self.offloaded {
|
|
if j.progress.cancelled {
|
|
continue;
|
|
}
|
|
for r in j.inbox.take() {
|
|
changed = true;
|
|
match r {
|
|
WorkerReply::Progress { done, total } => {
|
|
j.progress.done = done;
|
|
j.progress.total = total;
|
|
}
|
|
WorkerReply::Group { comp, layer, group, message } => results.push((j.kind, comp, layer, group, message)),
|
|
WorkerReply::Fill { plan, files } => fills.push((plan, files)),
|
|
WorkerReply::WarpPlan { key, summary } => effectcraft_effects::warp_stab::store_summary(key, *summary),
|
|
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 { .. } => {}
|
|
}
|
|
}
|
|
j.progress.elapsed = j.started.elapsed().as_secs_f64();
|
|
}
|
|
for (plan, files) in fills {
|
|
let n = files.len();
|
|
match crate::commands::content_fill::apply_plan(self, &plan, files) {
|
|
Ok(_) => self.events.push(Event::Toast { message: format!("Content-Aware Fill: {n} frame(s) filled"), error: false }),
|
|
Err(e) => self.events.push(Event::Toast { message: format!("Content-Aware Fill: {e}"), error: true }),
|
|
}
|
|
}
|
|
for (kind, comp, layer, group, message) in results {
|
|
let uid = group.uid;
|
|
let r = self.edit(kind.label(), None, |p, _| {
|
|
let g = p
|
|
.comp_mut(comp)
|
|
.and_then(|c| c.layer_mut(layer))
|
|
.and_then(|l| l.props.find_group_mut(uid))
|
|
.ok_or_else(|| crate::EngineError::Other("the analysed property was removed during the analysis".into()))?;
|
|
*g = *group;
|
|
Ok(())
|
|
});
|
|
match r {
|
|
Ok(()) => self.events.push(Event::Toast { message, error: false }),
|
|
Err(e) => self.events.push(Event::Toast { message: format!("{}: {e}", kind.label()), error: true }),
|
|
}
|
|
}
|
|
let mut done = vec![];
|
|
self.offloaded.retain(|j| {
|
|
if j.progress.finished {
|
|
done.push((j.kind, j.progress.clone(), j.segs));
|
|
}
|
|
!j.progress.finished
|
|
});
|
|
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 {
|
|
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);
|
|
// not on a plan summary (`warp.status` only).
|
|
if kind != JobKind::WarpPlan {
|
|
self.bump();
|
|
}
|
|
}
|
|
changed
|
|
}
|
|
|
|
/// Feed a remote render's replies into its [`JobShared`]; a cancelled one is wrapped up here
|
|
/// (the item being rendered and the ones after it become "User Stopped").
|
|
pub(crate) fn drain_remote_render(&mut self) {
|
|
let Some(job) = &self.render_job else { return };
|
|
let Some(remote) = &job.remote else { return };
|
|
let shared = job.shared.clone();
|
|
for r in remote.inbox.take() {
|
|
match r {
|
|
WorkerReply::Render { state } => {
|
|
let fin = crate::render_queue::lock(&shared.state).finished;
|
|
*crate::render_queue::lock(&shared.state) = JobState { finished: fin, ..state };
|
|
}
|
|
WorkerReply::Item { update } => crate::render_queue::lock(&shared.updates).push(update),
|
|
WorkerReply::Failed { error } => {
|
|
log::warn!("render worker: {error}");
|
|
let cur = crate::render_queue::lock(&shared.state).current;
|
|
if let Some(id) = cur {
|
|
crate::render_queue::lock(&shared.updates).push(ItemUpdate::Finished {
|
|
id,
|
|
status: RenderStatus::Failed(error),
|
|
seconds: 0.0,
|
|
output: None,
|
|
});
|
|
}
|
|
}
|
|
WorkerReply::Done => crate::render_queue::lock(&shared.state).finished = true,
|
|
_ => {}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
fn round<T: Serialize + for<'de> Deserialize<'de>>(v: &T) -> String {
|
|
let a = serde_json::to_string(v).unwrap();
|
|
let back: T = serde_json::from_str(&a).unwrap();
|
|
let b = serde_json::to_string(&back).unwrap();
|
|
assert_eq!(a, b);
|
|
a
|
|
}
|
|
|
|
#[test]
|
|
fn protocol_round_trips() {
|
|
let mut s = Session::default();
|
|
s.execute("file.openDemoProject", serde_json::json!({})).unwrap();
|
|
let comp = s.active_comp_id().unwrap();
|
|
s.execute("renderQueue.add", serde_json::json!({})).unwrap();
|
|
let item = s.project.render_queue[0].clone();
|
|
let reqs = [
|
|
WorkerJob::Render { items: vec![vec![(item.clone(), "/out.gif".into())]] },
|
|
WorkerJob::Warp { comp, layer: LayerId(3), effect: 9 },
|
|
WorkerJob::WarpPlan { comp, layer: LayerId(3), effect: 9, time: Tick::from_seconds_f64(0.5) },
|
|
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),
|
|
tracker: 4,
|
|
direction: crate::tracking::Direction::Backward,
|
|
settings: TrackerSettings::default(),
|
|
points: vec![PointValues {
|
|
uid: 5,
|
|
spec: effectcraft_track::PointSpec { center: [1.0, 2.0], feature_size: [3.0, 4.0], search_offset: [0.5, 0.0], search_size: [9.0, 9.0] },
|
|
attach_offset: [0.25, -1.0],
|
|
}],
|
|
times: vec![Tick::ZERO, Tick::from_seconds_f64(0.5)],
|
|
},
|
|
WorkerJob::MaskTrack {
|
|
comp,
|
|
layer: LayerId(2),
|
|
mask: 7,
|
|
direction: crate::tracking::Direction::Forward,
|
|
path: ShapePath::ellipse([10.0, 20.0], 5.0, 6.0),
|
|
method: MaskMethod::Perspective,
|
|
times: vec![Tick::from_seconds_f64(1.0)],
|
|
},
|
|
WorkerJob::ContentFill {
|
|
plan: Box::new(crate::commands::content_fill::FillPlan {
|
|
comp,
|
|
layer: LayerId(2),
|
|
name: "Clip".into(),
|
|
width: 4,
|
|
height: 3,
|
|
times: vec![Tick::ZERO],
|
|
method: "object".into(),
|
|
lighting: 0.5,
|
|
expand: 1,
|
|
reference: Some((LayerId(1), Tick::ZERO)),
|
|
dir: "/Fill/Clip_Fill_1".into(),
|
|
}),
|
|
},
|
|
];
|
|
for job in reqs {
|
|
let kind = job.kind();
|
|
let req = WorkerRequest { id: 42, project: s.project.to_json(), files: footage_files(&s.project), job };
|
|
let t = round(&req);
|
|
assert!(t.contains("\"kind\":"), "{kind:?}");
|
|
let back: WorkerRequest = serde_json::from_str(&t).unwrap();
|
|
assert_eq!(back.job.kind(), kind);
|
|
assert_eq!(Project::from_json(&back.project).unwrap(), *s.project);
|
|
}
|
|
let g = s.active_comp().unwrap().layers[0].props.clone();
|
|
for r in [
|
|
WorkerReply::Render { state: JobState { current: Some(3), done: 4, total: 10, items_total: 1, ..Default::default() } },
|
|
WorkerReply::Item { update: ItemUpdate::Started { id: 3, unix: 1_700_000_000 } },
|
|
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::WarpPlan {
|
|
key: [1, u64::MAX],
|
|
summary: Box::new(effectcraft_effects::warp_stab::PlanSummary {
|
|
warps: vec![[[1.0, 0.0, 2.5], [0.0, 1.0, -1.0], [0.0, 0.0, 1.0]]],
|
|
auto_scale: 1.08,
|
|
crop: Some([1.0, 2.0, 300.0, 200.0]),
|
|
valid_fraction: 0.9,
|
|
}),
|
|
},
|
|
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::Fill {
|
|
plan: Box::new(crate::commands::content_fill::FillPlan {
|
|
comp,
|
|
layer: LayerId(2),
|
|
name: "Clip".into(),
|
|
width: 4,
|
|
height: 3,
|
|
times: vec![Tick::ZERO],
|
|
method: "object".into(),
|
|
lighting: 0.5,
|
|
expand: 1,
|
|
reference: Some((LayerId(1), Tick::ZERO)),
|
|
dir: "/Fill/Clip_Fill_1".into(),
|
|
}),
|
|
files: vec!["/Fill/Clip_Fill_1/fill_00000.png".into()],
|
|
},
|
|
WorkerReply::Failed { error: "boom".into() },
|
|
WorkerReply::Done,
|
|
] {
|
|
let t = round(&r);
|
|
assert!(t.starts_with("{\"type\":"), "{t}");
|
|
}
|
|
}
|
|
|
|
/// An [`Offload`] that runs the job right away in a second session, through JSON (what a
|
|
/// Web Worker does, minus the threads).
|
|
struct Inline;
|
|
impl Offload for Inline {
|
|
fn start(&self, req: WorkerRequest, inbox: Arc<Inbox>) -> Result<(), String> {
|
|
let req: WorkerRequest = serde_json::from_str(&serde_json::to_string(&req).unwrap()).unwrap();
|
|
let mut w = Session { exporter: Some(Arc::new(crate::rq_tests::MockExporter { log: Default::default() })), ..Default::default() };
|
|
let post: Post = Rc::new(move |r| inbox.push(serde_json::from_str(&serde_json::to_string(&r).unwrap()).unwrap()));
|
|
run_request(&mut w, req, &post);
|
|
Ok(())
|
|
}
|
|
fn cancel(&self, _: u64) {}
|
|
}
|
|
|
|
#[test]
|
|
fn render_runs_through_the_offload() {
|
|
let mut s = Session {
|
|
exporter: Some(Arc::new(crate::rq_tests::MockExporter { log: Default::default() })),
|
|
offload: Some(Arc::new(Inline)),
|
|
..Default::default()
|
|
};
|
|
s.execute("file.openDemoProject", serde_json::json!({})).unwrap();
|
|
s.execute("renderQueue.add", serde_json::json!({"output": "/x.gif"})).unwrap();
|
|
let ids = s.start_render(false).unwrap();
|
|
// (the inline offload has replied already; replies are applied when polled)
|
|
s.poll_render();
|
|
assert!(s.render_job.is_none());
|
|
let it = s.project.render_queue.iter().find(|i| i.id == ids[0]).unwrap();
|
|
assert_eq!(it.status, RenderStatus::Done, "{:?}", it.status);
|
|
assert!(it.last_output.as_deref().is_some_and(|o| o.ends_with("x.gif")));
|
|
// `wait` renders inline.
|
|
s.execute("renderQueue.add", serde_json::json!({"output": "/y.gif"})).unwrap();
|
|
s.start_render(true).unwrap();
|
|
assert!(s.render_job.is_none() || s.render_job.as_ref().unwrap().remote.is_none());
|
|
}
|
|
|
|
/// Runs jobs in a session sharing `footage` (the synthetic clip of the tracking tests).
|
|
struct InlineWith(Arc<dyn effectcraft_render::FootageSource>);
|
|
impl Offload for InlineWith {
|
|
fn start(&self, req: WorkerRequest, inbox: Arc<Inbox>) -> Result<(), String> {
|
|
let req: WorkerRequest = serde_json::from_str(&serde_json::to_string(&req).unwrap()).unwrap();
|
|
let mut w = Session { footage: self.0.clone(), ..Default::default() };
|
|
let post: Post = Rc::new(move |r| inbox.push(serde_json::from_str(&serde_json::to_string(&r).unwrap()).unwrap()));
|
|
run_request(&mut w, req, &post);
|
|
Ok(())
|
|
}
|
|
fn cancel(&self, _: u64) {}
|
|
}
|
|
|
|
/// Never replies (a job still running); records cancellations.
|
|
#[derive(Default)]
|
|
struct Hanging(Mutex<Vec<u64>>);
|
|
impl Offload for Hanging {
|
|
fn start(&self, _: WorkerRequest, _: Arc<Inbox>) -> Result<(), String> {
|
|
Ok(())
|
|
}
|
|
fn cancel(&self, id: u64) {
|
|
self.0.lock().unwrap().push(id);
|
|
}
|
|
}
|
|
|
|
fn pose(f: u32) -> ([f64; 2], f64, f64) {
|
|
([150.0 + 2.0 * f as f64, 110.0 + 1.0 * f as f64], 0.0, 1.0)
|
|
}
|
|
|
|
#[test]
|
|
fn track_analysis_runs_through_the_offload_as_one_undo_step() {
|
|
let (mut s, clip, _) = crate::tests_track::setup(crate::tests_track::patch_frames(pose));
|
|
s.offload = Some(Arc::new(InlineWith(s.footage.clone())));
|
|
s.execute("track.motion", serde_json::json!({})).unwrap();
|
|
s.execute("track.setPoint", serde_json::json!({"point": 1, "center": pose(0).0, "featureSize": [36, 36], "searchSize": [80, 80]})).unwrap();
|
|
let undo = s.history.undo.len();
|
|
s.execute("track.analyze", serde_json::json!({"direction": "forward"})).unwrap();
|
|
assert!(s.is_tracking() && s.track_job.is_none(), "offloaded, not on a thread");
|
|
assert!(s.track_progress().is_some());
|
|
assert!(s.poll_offload());
|
|
assert!(!s.is_tracking() && s.offloaded.is_empty());
|
|
let l = s.active_comp().unwrap().layer(clip).unwrap();
|
|
let tp = l.trackers().next().unwrap().0.track_points().next().unwrap().clone();
|
|
let fc = tp.get("featureCenter").unwrap();
|
|
assert_eq!(fc.keys.len(), 25);
|
|
let v = fc.value_at(crate::tests_track::frame_time(12)).as_vec2();
|
|
assert!((v[0] - pose(12).0[0]).abs() < 0.3 && (v[1] - pose(12).0[1]).abs() < 0.3, "{v:?}");
|
|
assert_eq!(s.history.undo.len(), undo + 1);
|
|
assert_eq!(s.history.undo.last().unwrap().0, "Analyze Track");
|
|
assert!(s.events.iter().any(|e| matches!(e, Event::Toast { message, error: false } if message.starts_with("Tracked"))));
|
|
assert!(s.undo());
|
|
let l = s.active_comp().unwrap().layer(clip).unwrap();
|
|
assert!(l.trackers().next().unwrap().0.track_points().next().unwrap().get("featureCenter").unwrap().keys.is_empty());
|
|
// `wait` runs inline.
|
|
s.redo();
|
|
s.execute("track.analyze", serde_json::json!({"direction": "forward", "wait": true})).ok();
|
|
assert!(s.offloaded.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn cancelling_offloaded_jobs() {
|
|
let (mut s, _, _) = crate::tests_track::setup(crate::tests_track::patch_frames(pose));
|
|
let off = Arc::new(Hanging::default());
|
|
s.offload = Some(off.clone());
|
|
s.exporter = Some(Arc::new(crate::rq_tests::MockExporter { log: Default::default() }));
|
|
// Analysis: stopping drops the job, nothing is written.
|
|
s.execute("track.motion", serde_json::json!({})).unwrap();
|
|
let rev = s.revision;
|
|
s.execute("track.analyze", serde_json::json!({})).unwrap();
|
|
assert!(s.is_tracking());
|
|
assert!(s.execute("track.analyze", serde_json::json!({})).is_err(), "one at a time");
|
|
assert!(s.stop_track());
|
|
s.poll_offload();
|
|
assert!(!s.is_tracking());
|
|
assert_eq!(off.0.lock().unwrap().len(), 1);
|
|
assert!(s.events.iter().any(|e| matches!(e, Event::Toast { message, .. } if message.contains("stopped"))));
|
|
assert!(s.revision > rev);
|
|
// Render: the queued items become User Stopped.
|
|
s.execute("renderQueue.add", serde_json::json!({"output": "/a.gif"})).unwrap();
|
|
let ids = s.start_render(false).unwrap();
|
|
assert!(s.is_rendering());
|
|
assert!(s.stop_render());
|
|
s.poll_render();
|
|
assert!(!s.is_rendering());
|
|
assert_eq!(off.0.lock().unwrap().len(), 2);
|
|
let it = s.project.render_queue.iter().find(|i| i.id == ids[0]).unwrap();
|
|
assert_eq!(it.status, RenderStatus::UserStopped);
|
|
}
|
|
|
|
#[test]
|
|
fn progress_hook_is_scoped() {
|
|
report(1, 2); // no hook: nothing happens
|
|
let seen = Rc::new(RefCell::new(vec![]));
|
|
let s2 = seen.clone();
|
|
effectcraft_render::passes::block_on(with_hook(Box::new(move |d, t| s2.borrow_mut().push((d, t))), async {
|
|
report(1, 3);
|
|
report(2, 3);
|
|
}));
|
|
report(3, 3);
|
|
assert_eq!(*seen.borrow(), vec![(1, 3), (2, 3)]);
|
|
}
|
|
}
|