mirror of
https://github.com/storytold/effectcraft.git
synced 2026-10-30 13:08:07 -06:00
Merge pull request #2369 from ZJHeepfixer/apple-silicon-idle-wakeups
EffectCraft — no polling while idle: a deadline worker that sleeps, and a footage watcher that wakes the UI only on change
This commit is contained in:
@@ -9,31 +9,35 @@
|
||||
//! render reads the new pixels.
|
||||
//!
|
||||
//! The automatic part compares each footage file's size and modification time with what they
|
||||
//! were when the file was last seen. The desktop app takes those stamps every two seconds (and
|
||||
//! when its window comes back to the front) on a worker thread, so a file on a slow network share
|
||||
//! never holds up the interface ([`start_scan`], [`poll_scan`]); `footage.reloadChanged` takes them
|
||||
//! on the spot (CLI, agents, tests). Like the footage check after File ▸ Open, an automatic reload
|
||||
//! records the state of the disk: it is not an undo step and doesn't modify the project.
|
||||
//! were when the file was last seen. The desktop app keeps a [`Watcher`]: a worker thread that
|
||||
//! stamps the files every two seconds while the window is in front (every ten seconds while it
|
||||
//! is not, and at once when it comes back to the front) and wakes the interface only when a stamp
|
||||
//! changed, so a file on a slow network share never holds up the interface and an idle app never
|
||||
//! redraws for footage that didn't change; `footage.reloadChanged` takes the stamps on the spot
|
||||
//! (CLI, agents, tests). Like the footage check after File ▸ Open, an automatic reload records
|
||||
//! the state of the disk: it is not an undo step and doesn't modify the project.
|
||||
|
||||
use std::collections::BTreeSet;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender};
|
||||
use std::sync::{Arc, Mutex, PoisonError};
|
||||
use std::time::Duration;
|
||||
|
||||
use effectcraft_project::{Footage, FootageKind, ItemId, ItemKind, SourceLayer};
|
||||
use serde_json::{Value, json};
|
||||
|
||||
use crate::Session;
|
||||
use crate::{Services, Session};
|
||||
|
||||
/// A file's size and modification time (see [`crate::Services::stamp`]).
|
||||
pub type Stamp = (u64, u128);
|
||||
|
||||
/// Stamps taken on a worker thread, waiting for [`poll_scan`].
|
||||
pub type Scan = Arc<Mutex<Option<Vec<(String, Option<Stamp>)>>>>;
|
||||
/// The stamps of a set of files, in path order.
|
||||
pub type Stamps = Vec<(String, Option<Stamp>)>;
|
||||
|
||||
/// The files footage items read: each item's file and every frame of an image sequence. Image
|
||||
/// sequences count only with Automatically Reload Footage set to All Footage (After Effects'
|
||||
/// default, Non-Sequence Footage, leaves them alone: thousands of frames on a share add up).
|
||||
/// Items with a stored path that is not followed ([`crate::doc_path`]) are not watched.
|
||||
fn watched(s: &Session) -> BTreeSet<String> {
|
||||
pub fn watched(s: &Session) -> BTreeSet<String> {
|
||||
let sequences = s.prefs.import.auto_reload_footage == "all";
|
||||
let mut paths = BTreeSet::new();
|
||||
for it in s.project.items.values() {
|
||||
@@ -267,31 +271,253 @@ pub(crate) fn command(s: &mut Session, _: &Value) -> crate::Result<Value> {
|
||||
apply_stamps(s, now)
|
||||
}
|
||||
|
||||
/// Start stamping the watched files on a worker thread, unless a scan is still running (or the
|
||||
/// build has no threads: the web app, whose footage lives in browser storage).
|
||||
pub fn start_scan(s: &mut Session) {
|
||||
if s.footage_scan.is_some() || cfg!(target_arch = "wasm32") {
|
||||
return;
|
||||
/// How often the watched files are stamped while the window is in front.
|
||||
pub const FOCUSED_INTERVAL: Duration = Duration::from_secs(2);
|
||||
/// How often they are stamped while another app has the front: a file saved there is noticed
|
||||
/// as soon as the window comes back, so this only covers a second screen.
|
||||
pub const UNFOCUSED_INTERVAL: Duration = Duration::from_secs(10);
|
||||
|
||||
enum Watch {
|
||||
Paths(BTreeSet<String>),
|
||||
Focused(bool),
|
||||
}
|
||||
|
||||
/// The desktop app's footage watcher: a worker thread that stamps the watched files and hands
|
||||
/// the stamps over ([`Watcher::take`]) only when one of them changed since it last looked,
|
||||
/// waking the interface through the callback given to [`Watcher::start`]. The interface tells it
|
||||
/// what to watch and whether the window is in front; nothing watched means the thread sleeps.
|
||||
/// Dropping the watcher ends the thread.
|
||||
pub struct Watcher {
|
||||
tx: Sender<Watch>,
|
||||
result: Arc<Mutex<Option<Stamps>>>,
|
||||
paths: BTreeSet<String>,
|
||||
focused: Option<bool>,
|
||||
}
|
||||
|
||||
impl Watcher {
|
||||
/// Start the worker, stamping through `services`; `wake` runs on the worker whenever new
|
||||
/// stamps are ready to [`Watcher::take`]. `None` where there are no threads (the web app,
|
||||
/// whose footage lives in browser storage) or one couldn't be started.
|
||||
pub fn start(services: Arc<dyn Services>, wake: impl Fn() + Send + Sync + 'static) -> Option<Watcher> {
|
||||
Self::start_with(services, wake, FOCUSED_INTERVAL, UNFOCUSED_INTERVAL)
|
||||
}
|
||||
let paths = watched(s);
|
||||
if paths.is_empty() {
|
||||
return;
|
||||
|
||||
/// [`Watcher::start`] with its own stamping intervals (tests).
|
||||
pub fn start_with(services: Arc<dyn Services>, wake: impl Fn() + Send + Sync + 'static, focused: Duration, unfocused: Duration) -> Option<Watcher> {
|
||||
if cfg!(target_arch = "wasm32") {
|
||||
return None;
|
||||
}
|
||||
let (tx, rx) = std::sync::mpsc::channel();
|
||||
let result: Arc<Mutex<Option<Stamps>>> = Arc::default();
|
||||
let out = Arc::clone(&result);
|
||||
std::thread::Builder::new().name("footage-watch".into()).spawn(move || watch(&rx, &*services, &out, &wake, focused, unfocused)).ok()?;
|
||||
Some(Watcher { tx, result, paths: BTreeSet::new(), focused: None })
|
||||
}
|
||||
let services = s.services.clone();
|
||||
let slot: Scan = Arc::default();
|
||||
let out = slot.clone();
|
||||
let spawned = std::thread::Builder::new().name("footage-watch".into()).spawn(move || {
|
||||
let stamps = paths.into_iter().map(|p| (p.clone(), services.stamp(&p))).collect();
|
||||
*out.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(stamps);
|
||||
});
|
||||
if spawned.is_ok() {
|
||||
s.footage_scan = Some(slot);
|
||||
|
||||
/// The files to watch ([`watched`]); a changed set is stamped at once.
|
||||
pub fn set_watched(&mut self, paths: BTreeSet<String>) {
|
||||
if paths != self.paths {
|
||||
self.paths = paths.clone();
|
||||
let _ = self.tx.send(Watch::Paths(paths));
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether the window is in front; coming back to the front stamps the files at once.
|
||||
pub fn set_focused(&mut self, focused: bool) {
|
||||
if self.focused != Some(focused) {
|
||||
self.focused = Some(focused);
|
||||
let _ = self.tx.send(Watch::Focused(focused));
|
||||
}
|
||||
}
|
||||
|
||||
/// Stamps the worker found changed (or the first ones for a new set of files), once; for
|
||||
/// [`apply_stamps`]. `None` while nothing changed.
|
||||
pub fn take(&self) -> Option<Stamps> {
|
||||
self.result.lock().unwrap_or_else(PoisonError::into_inner).take()
|
||||
}
|
||||
}
|
||||
|
||||
/// Apply a finished scan (`None` while there is none, or it is still running).
|
||||
pub fn poll_scan(s: &mut Session) -> Option<crate::Result<Value>> {
|
||||
let stamps = s.footage_scan.as_ref()?.lock().unwrap_or_else(std::sync::PoisonError::into_inner).take()?;
|
||||
s.footage_scan = None;
|
||||
Some(apply_stamps(s, stamps))
|
||||
/// The watcher's worker: wait for a message or the interval, stamp, report only changes.
|
||||
fn watch(
|
||||
rx: &Receiver<Watch>,
|
||||
services: &dyn Services,
|
||||
out: &Mutex<Option<Stamps>>,
|
||||
wake: &(dyn Fn() + Send + Sync),
|
||||
focused_every: Duration,
|
||||
unfocused_every: Duration,
|
||||
) {
|
||||
let mut paths = BTreeSet::new();
|
||||
let mut focused = true;
|
||||
let mut last: Option<Stamps> = None;
|
||||
loop {
|
||||
let msg = if paths.is_empty() {
|
||||
rx.recv().map_err(|_| RecvTimeoutError::Disconnected)
|
||||
} else {
|
||||
rx.recv_timeout(if focused { focused_every } else { unfocused_every })
|
||||
};
|
||||
match msg {
|
||||
Ok(Watch::Paths(p)) => {
|
||||
paths = p;
|
||||
// A new set is reported whole, so the session forgets files no longer watched.
|
||||
last = None;
|
||||
}
|
||||
Ok(Watch::Focused(f)) => {
|
||||
let regained = f && !focused;
|
||||
focused = f;
|
||||
if !regained {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
Err(RecvTimeoutError::Timeout) => {}
|
||||
Err(RecvTimeoutError::Disconnected) => return,
|
||||
}
|
||||
if paths.is_empty() {
|
||||
continue;
|
||||
}
|
||||
let stamps: Stamps = paths.iter().map(|p| (p.clone(), services.stamp(p))).collect();
|
||||
if last.as_ref() != Some(&stamps) {
|
||||
last = Some(stamps.clone());
|
||||
*out.lock().unwrap_or_else(PoisonError::into_inner) = Some(stamps);
|
||||
wake();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(all(test, not(target_arch = "wasm32")))]
|
||||
mod watcher_tests {
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
use super::*;
|
||||
|
||||
/// Files whose stamps the test sets; counts how often they were stamped.
|
||||
#[derive(Default)]
|
||||
struct Disk {
|
||||
stamps: Mutex<HashMap<String, Stamp>>,
|
||||
looks: AtomicUsize,
|
||||
}
|
||||
impl Disk {
|
||||
fn set(&self, path: &str, stamp: Stamp) {
|
||||
self.stamps.lock().unwrap_or_else(PoisonError::into_inner).insert(path.into(), stamp);
|
||||
}
|
||||
}
|
||||
impl Services for Disk {
|
||||
fn read_file(&self, path: &str) -> std::io::Result<Vec<u8>> {
|
||||
Err(std::io::Error::new(std::io::ErrorKind::NotFound, path.to_string()))
|
||||
}
|
||||
fn write_file(&self, _: &str, _: &[u8]) -> std::io::Result<()> {
|
||||
Err(std::io::Error::other("read-only"))
|
||||
}
|
||||
fn stamp(&self, path: &str) -> Option<Stamp> {
|
||||
self.looks.fetch_add(1, Ordering::SeqCst);
|
||||
self.stamps.lock().unwrap_or_else(PoisonError::into_inner).get(path).copied()
|
||||
}
|
||||
}
|
||||
|
||||
const TICK: Duration = Duration::from_millis(20);
|
||||
const LONG: Duration = Duration::from_secs(60);
|
||||
|
||||
fn wait_for(w: &Watcher, wakes: &AtomicUsize, n: usize) -> Stamps {
|
||||
let t0 = std::time::Instant::now();
|
||||
loop {
|
||||
if wakes.load(Ordering::SeqCst) >= n
|
||||
&& let Some(s) = w.take()
|
||||
{
|
||||
return s;
|
||||
}
|
||||
assert!(t0.elapsed() < Duration::from_secs(10), "no report {n} within 10 s");
|
||||
std::thread::sleep(Duration::from_millis(5));
|
||||
}
|
||||
}
|
||||
|
||||
/// Wait (at most 10 s) until `ready`: a slow machine's worker gets there late, never early.
|
||||
fn wait_until(what: &str, mut ready: impl FnMut() -> bool) {
|
||||
let t0 = std::time::Instant::now();
|
||||
while !ready() {
|
||||
assert!(t0.elapsed() < Duration::from_secs(10), "{what} within 10 s");
|
||||
std::thread::sleep(Duration::from_millis(5));
|
||||
}
|
||||
}
|
||||
|
||||
fn start(disk: &Arc<Disk>, focused: Duration, unfocused: Duration) -> Option<(Watcher, Arc<AtomicUsize>)> {
|
||||
let wakes = Arc::new(AtomicUsize::new(0));
|
||||
let counter = Arc::clone(&wakes);
|
||||
let w = Watcher::start_with(
|
||||
Arc::clone(disk) as Arc<dyn Services>,
|
||||
move || {
|
||||
counter.fetch_add(1, Ordering::SeqCst);
|
||||
},
|
||||
focused,
|
||||
unfocused,
|
||||
)?;
|
||||
Some((w, wakes))
|
||||
}
|
||||
|
||||
/// The interface is woken for the first stamps of a set and for every change, and for
|
||||
/// nothing else: an unchanged file polled many times is no wakeup at all.
|
||||
#[test]
|
||||
fn the_watcher_wakes_the_interface_only_when_a_stamp_changes() {
|
||||
let disk = Arc::new(Disk::default());
|
||||
disk.set("/a.psd", (10, 1));
|
||||
let (mut w, wakes) = start(&disk, TICK, LONG).unwrap();
|
||||
w.set_focused(true);
|
||||
w.set_watched(["/a.psd".to_string()].into());
|
||||
assert_eq!(wait_for(&w, &wakes, 1), vec![("/a.psd".to_string(), Some((10, 1)))]);
|
||||
let looked = disk.looks.load(Ordering::SeqCst);
|
||||
wait_until("it keeps stamping while the window is in front", || disk.looks.load(Ordering::SeqCst) > looked + 3);
|
||||
assert_eq!(wakes.load(Ordering::SeqCst), 1, "an unchanged file is no wakeup");
|
||||
assert!(w.take().is_none());
|
||||
disk.set("/a.psd", (12, 2));
|
||||
assert_eq!(wait_for(&w, &wakes, 2), vec![("/a.psd".to_string(), Some((12, 2)))]);
|
||||
assert_eq!(wakes.load(Ordering::SeqCst), 2);
|
||||
}
|
||||
|
||||
/// Behind another app the files are stamped only at the slow interval; the window coming
|
||||
/// back to the front stamps them at once.
|
||||
#[test]
|
||||
fn behind_another_app_the_watcher_rests_until_the_window_is_back() {
|
||||
let disk = Arc::new(Disk::default());
|
||||
disk.set("/b.psd", (1, 1));
|
||||
let (mut w, wakes) = start(&disk, TICK, LONG).unwrap();
|
||||
w.set_focused(false);
|
||||
w.set_watched(["/b.psd".to_string()].into());
|
||||
wait_for(&w, &wakes, 1);
|
||||
let looked = disk.looks.load(Ordering::SeqCst);
|
||||
disk.set("/b.psd", (2, 2));
|
||||
std::thread::sleep(TICK * 15);
|
||||
assert_eq!(disk.looks.load(Ordering::SeqCst), looked, "no stamping at the front interval while behind");
|
||||
assert_eq!(wakes.load(Ordering::SeqCst), 1);
|
||||
w.set_focused(true);
|
||||
assert_eq!(wait_for(&w, &wakes, 2), vec![("/b.psd".to_string(), Some((2, 2)))]);
|
||||
}
|
||||
|
||||
/// Nothing to watch: the worker sleeps, stamping nothing; a new set is stamped at once and
|
||||
/// reported whole, and dropping the watcher ends the worker.
|
||||
#[test]
|
||||
fn nothing_watched_is_nothing_stamped_and_a_new_set_is_reported_whole() {
|
||||
let disk = Arc::new(Disk::default());
|
||||
disk.set("/c.psd", (1, 1));
|
||||
disk.set("/d.psd", (2, 2));
|
||||
let (mut w, wakes) = start(&disk, TICK, TICK).unwrap();
|
||||
w.set_focused(true);
|
||||
std::thread::sleep(TICK * 10);
|
||||
assert_eq!(disk.looks.load(Ordering::SeqCst), 0);
|
||||
w.set_watched(["/c.psd".to_string(), "/d.psd".to_string()].into());
|
||||
assert_eq!(wait_for(&w, &wakes, 1), vec![("/c.psd".to_string(), Some((1, 1))), ("/d.psd".to_string(), Some((2, 2)))]);
|
||||
w.set_watched(["/d.psd".to_string()].into());
|
||||
assert_eq!(wait_for(&w, &wakes, 2), vec![("/d.psd".to_string(), Some((2, 2)))], "the set is reported whole, so the session forgets /c.psd");
|
||||
w.set_watched(BTreeSet::new());
|
||||
// The worker may already be past its wait with the old set: that one round of stamps
|
||||
// (one file) can still come, and nothing after it.
|
||||
let looked = disk.looks.load(Ordering::SeqCst);
|
||||
std::thread::sleep(TICK * 10);
|
||||
assert!(disk.looks.load(Ordering::SeqCst) <= looked + 1, "nothing watched, nothing stamped");
|
||||
let result = Arc::downgrade(&w.result);
|
||||
drop(w);
|
||||
let t0 = std::time::Instant::now();
|
||||
while result.upgrade().is_some() {
|
||||
assert!(t0.elapsed() < Duration::from_secs(5), "the worker is still running");
|
||||
std::thread::sleep(Duration::from_millis(5));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -241,8 +241,6 @@ pub struct Session {
|
||||
/// Each footage file's size and modification time when last seen (Settings ▸ Import ▸
|
||||
/// Reload Footage Changed on Disk, [`footage_reload`]).
|
||||
pub footage_stamps: std::collections::HashMap<String, footage_reload::Stamp>,
|
||||
/// The stamps being taken on a worker thread ([`footage_reload::start_scan`]).
|
||||
pub footage_scan: Option<footage_reload::Scan>,
|
||||
/// The disk cache's footage salt and the footage it was taken for ([`Session::refresh_disk_salt`]).
|
||||
pub disk_salt: Option<(u64, u64)>,
|
||||
/// How many script-issued commands are running ([`Session::execute_for_script`]): while one
|
||||
@@ -351,7 +349,6 @@ impl Default for Session {
|
||||
stored_paths: Default::default(),
|
||||
footage_stamps: Default::default(),
|
||||
disk_salt: None,
|
||||
footage_scan: None,
|
||||
script_calls: ScriptCalls::default(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -153,7 +153,8 @@ page!(Import {
|
||||
sequence_fps: f64 = 30.0,
|
||||
report_missing_frames: bool = true,
|
||||
/// Automatically Reload Footage whose file another app changed (size or modification time),
|
||||
/// checked every two seconds and when the window comes back to the front: `off`,
|
||||
/// checked every two seconds while the window is in front, every ten seconds while it is
|
||||
/// not, and when it comes back to the front: `off`,
|
||||
/// `nonSequence` (the default, as in After Effects: image sequences are left alone) or `all`.
|
||||
auto_reload_footage: String = "nonSequence".into(),
|
||||
/// Interpret unlabeled alpha as: `ask`, `guess`, `ignore`, `straight`, `premultiplied`.
|
||||
|
||||
+170
-38
@@ -2,12 +2,15 @@
|
||||
// Browser callbacks stay on their worker thread. The workspace's wgpu feature still
|
||||
// requires Send callbacks on non-atomic wasm, and ParticleSim requires Send + Sync.
|
||||
use std::collections::BTreeMap;
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
use std::sync::Condvar;
|
||||
use std::sync::{Arc, Mutex, Weak};
|
||||
use std::time::Duration;
|
||||
|
||||
pub(crate) const TIMEOUT: Duration = Duration::from_secs(10);
|
||||
pub(crate) type Error = String;
|
||||
type Done = Box<dyn FnOnce(Result<Vec<u8>, Error>) + Send>;
|
||||
const TIMED_OUT: &str = "GPU readback timed out; context retired";
|
||||
|
||||
struct Entry {
|
||||
deadline: web_time::Instant,
|
||||
@@ -20,9 +23,70 @@ struct State {
|
||||
retired: Option<Error>,
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
timer_armed: bool,
|
||||
/// The deadline worker is to exit: the context was dropped.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
closed: bool,
|
||||
/// How many times the deadline worker woke up (tests: it must not wake while idle).
|
||||
#[cfg(all(test, not(target_arch = "wasm32")))]
|
||||
worker_wakes: u64,
|
||||
}
|
||||
/// The state and, natively, the condition the deadline worker sleeps on. The worker owns an
|
||||
/// `Arc` of this, not of the [`Readbacks`], so it can block on the condition without keeping
|
||||
/// the context alive; dropping the context closes it.
|
||||
#[derive(Default)]
|
||||
struct Inner {
|
||||
state: Mutex<State>,
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
wake: Condvar,
|
||||
}
|
||||
#[derive(Default)]
|
||||
pub(crate) struct Readbacks(Mutex<State>);
|
||||
pub(crate) struct Readbacks(Arc<Inner>);
|
||||
|
||||
/// Retire the context when a deadline has passed: the entries to fail, or `None` when none
|
||||
/// is due (or it is retired already). Decides and drains under the lock the caller holds.
|
||||
fn expire_locked(s: &mut State, now: web_time::Instant) -> Option<Vec<Entry>> {
|
||||
// Decide and drain atomically: a cancelled old deadline must not retire
|
||||
// a newer request inserted between the deadline check and retirement.
|
||||
if s.retired.is_some() || !s.pending.values().any(|e| now >= e.deadline) {
|
||||
return None;
|
||||
}
|
||||
s.retired = Some(TIMED_OUT.to_string());
|
||||
Some(std::mem::take(&mut s.pending).into_values().collect())
|
||||
}
|
||||
|
||||
/// The deadline worker. It sleeps while nothing is pending, and otherwise exactly until the
|
||||
/// earliest deadline or until what is pending changes, so an idle context costs no wakeups
|
||||
/// (a 100 ms polling loop here was ten wakeups a second for the life of the app). When a
|
||||
/// deadline has passed it retires the context and fails every pending readback.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn run_deadlines(inner: &Inner) {
|
||||
let mut s = inner.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
loop {
|
||||
if s.closed {
|
||||
return;
|
||||
}
|
||||
let now = web_time::Instant::now();
|
||||
match s.pending.values().map(|e| e.deadline).min() {
|
||||
None => s = inner.wake.wait(s).unwrap_or_else(std::sync::PoisonError::into_inner),
|
||||
Some(deadline) if deadline > now => {
|
||||
s = inner.wake.wait_timeout(s, deadline - now).unwrap_or_else(std::sync::PoisonError::into_inner).0;
|
||||
}
|
||||
Some(_) => {
|
||||
let expired = expire_locked(&mut s, now);
|
||||
drop(s);
|
||||
for entry in expired.into_iter().flatten() {
|
||||
notify(entry.done, Err(TIMED_OUT.into()));
|
||||
}
|
||||
s = inner.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
#[cfg(test)]
|
||||
{
|
||||
s.worker_wakes += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
pub(crate) struct Ticket {
|
||||
owner: Weak<Readbacks>,
|
||||
id: u64,
|
||||
@@ -50,23 +114,24 @@ impl Readbacks {
|
||||
let me = Arc::new(Self::default());
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
{
|
||||
let weak = Arc::downgrade(&me);
|
||||
let inner = Arc::clone(&me.0);
|
||||
std::thread::Builder::new()
|
||||
.name("gpu-readback-deadlines".into())
|
||||
.spawn(move || {
|
||||
loop {
|
||||
std::thread::sleep(Duration::from_millis(100));
|
||||
let Some(me) = weak.upgrade() else { break };
|
||||
me.expire(web_time::Instant::now());
|
||||
}
|
||||
})
|
||||
.spawn(move || run_deadlines(&inner))
|
||||
.map_err(|e| format!("starting GPU readback deadline worker: {e}"))?;
|
||||
}
|
||||
Ok(me)
|
||||
}
|
||||
|
||||
/// Have the deadline worker look at what is pending again (natively; the browser's timer
|
||||
/// re-arms itself).
|
||||
fn poke(&self) {
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
self.0.wake.notify_one();
|
||||
}
|
||||
|
||||
pub(crate) fn check(&self) -> Result<(), Error> {
|
||||
let s = self.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let s = self.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
match &s.retired {
|
||||
Some(e) => Err(e.clone()),
|
||||
None => Ok(()),
|
||||
@@ -76,7 +141,7 @@ impl Readbacks {
|
||||
pub(crate) fn start(self: &Arc<Self>, done: impl FnOnce(Result<Vec<u8>, Error>) + wgpu::WasmNotSend + 'static) -> Result<Ticket, Error> {
|
||||
let mut done: Option<Done> = Some(Box::new(done));
|
||||
let result = {
|
||||
let mut s = self.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let mut s = self.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
if let Some(e) = &s.retired {
|
||||
Err(e.clone())
|
||||
} else if s.pending.len() >= 256 {
|
||||
@@ -96,8 +161,10 @@ impl Readbacks {
|
||||
{
|
||||
notify(done, Err(e.clone()));
|
||||
}
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
if result.is_ok() {
|
||||
// The worker may be asleep with nothing pending: this is its new earliest deadline.
|
||||
self.poke();
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
self.arm_timer();
|
||||
}
|
||||
result
|
||||
@@ -106,7 +173,7 @@ impl Readbacks {
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
fn arm_timer(self: &Arc<Self>) {
|
||||
let delay = {
|
||||
let mut s = self.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let mut s = self.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
if s.timer_armed || s.retired.is_some() {
|
||||
return;
|
||||
}
|
||||
@@ -123,13 +190,14 @@ impl Readbacks {
|
||||
|
||||
pub(crate) fn retire(&self, error: Error) {
|
||||
let pending = {
|
||||
let mut s = self.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let mut s = self.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
if s.retired.is_some() {
|
||||
return;
|
||||
}
|
||||
s.retired = Some(error.clone());
|
||||
std::mem::take(&mut s.pending)
|
||||
};
|
||||
self.poke();
|
||||
for (_, entry) in pending {
|
||||
notify(entry.done, Err(error.clone()));
|
||||
}
|
||||
@@ -137,34 +205,49 @@ impl Readbacks {
|
||||
|
||||
pub(crate) fn cancel_pending(&self, error: Error) {
|
||||
let pending = {
|
||||
let mut s = self.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let mut s = self.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
std::mem::take(&mut s.pending)
|
||||
};
|
||||
self.poke();
|
||||
for (_, entry) in pending {
|
||||
notify(entry.done, Err(error.clone()));
|
||||
}
|
||||
}
|
||||
|
||||
/// Retire the context when a deadline has passed (the browser's timer and tests; natively
|
||||
/// the deadline worker does this itself).
|
||||
#[cfg(any(test, target_arch = "wasm32"))]
|
||||
fn expire(&self, now: web_time::Instant) {
|
||||
let error = "GPU readback timed out; context retired".to_string();
|
||||
let pending = {
|
||||
let mut s = self.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
// Decide and drain atomically: a cancelled old deadline must not retire
|
||||
// a newer request inserted between the deadline check and retirement.
|
||||
if s.retired.is_some() || !s.pending.values().any(|e| now >= e.deadline) {
|
||||
return;
|
||||
}
|
||||
s.retired = Some(error.clone());
|
||||
std::mem::take(&mut s.pending)
|
||||
let expired = {
|
||||
let mut s = self.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
expire_locked(&mut s, now)
|
||||
};
|
||||
for (_, entry) in pending {
|
||||
notify(entry.done, Err(error.clone()));
|
||||
for entry in expired.into_iter().flatten() {
|
||||
notify(entry.done, Err(TIMED_OUT.into()));
|
||||
}
|
||||
}
|
||||
|
||||
/// Move every pending deadline to `deadline` and wake the worker, as a test's stand-in for
|
||||
/// waiting out the production timeout.
|
||||
#[cfg(all(test, not(target_arch = "wasm32")))]
|
||||
fn move_deadlines_for_test(&self, deadline: web_time::Instant) {
|
||||
{
|
||||
let mut s = self.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
for entry in s.pending.values_mut() {
|
||||
entry.deadline = deadline;
|
||||
}
|
||||
}
|
||||
self.poke();
|
||||
}
|
||||
}
|
||||
impl Drop for Readbacks {
|
||||
fn drop(&mut self) {
|
||||
self.retire("GPU readback context dropped".into());
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
{
|
||||
self.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner).closed = true;
|
||||
self.0.wake.notify_all();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -194,12 +277,14 @@ impl Layout {
|
||||
|
||||
impl Ticket {
|
||||
pub(crate) fn active(&self) -> bool {
|
||||
self.owner.upgrade().is_some_and(|me| me.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner).pending.contains_key(&self.id))
|
||||
self.owner.upgrade().is_some_and(|me| me.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner).pending.contains_key(&self.id))
|
||||
}
|
||||
pub(crate) fn complete(self, result: Result<Vec<u8>, Error>) {
|
||||
let Some(me) = self.owner.upgrade() else { return };
|
||||
let entry = me.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner).pending.remove(&self.id);
|
||||
let entry = me.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner).pending.remove(&self.id);
|
||||
if let Some(entry) = entry {
|
||||
// Its deadline is gone: the worker sleeps until the next one, or until something new.
|
||||
me.poke();
|
||||
notify(entry.done, result);
|
||||
}
|
||||
}
|
||||
@@ -215,7 +300,7 @@ fn schedule(me: Weak<Readbacks>, delay: Duration) -> Result<(), Error> {
|
||||
.map_err(|_| "GPU deadline timer unavailable".to_string())?;
|
||||
let callback = Closure::once(move || {
|
||||
if let Some(me) = me.upgrade() {
|
||||
me.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner).timer_armed = false;
|
||||
me.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner).timer_armed = false;
|
||||
me.expire(web_time::Instant::now());
|
||||
// If the earliest request completed, arm for the next live deadline.
|
||||
me.arm_timer();
|
||||
@@ -377,7 +462,7 @@ mod tests {
|
||||
assert_eq!(deferred.in_flight(), 0);
|
||||
assert_eq!(wakes.0.load(Ordering::SeqCst), 1);
|
||||
assert!(matches!(settled.as_mut().poll(&mut Context::from_waker(&waker)), Poll::Ready(())));
|
||||
assert!(me.0.lock().unwrap().pending.is_empty());
|
||||
assert!(me.0.state.lock().unwrap().pending.is_empty());
|
||||
rx.recv_timeout(Duration::from_secs(2)).unwrap();
|
||||
rx.recv_timeout(Duration::from_secs(2)).unwrap();
|
||||
|
||||
@@ -390,17 +475,64 @@ mod tests {
|
||||
tx2.send(()).unwrap();
|
||||
})
|
||||
.unwrap();
|
||||
{
|
||||
let mut state = me.0.lock().unwrap();
|
||||
for entry in state.pending.values_mut() {
|
||||
entry.deadline = web_time::Instant::now();
|
||||
}
|
||||
}
|
||||
me.move_deadlines_for_test(web_time::Instant::now());
|
||||
rx.recv_timeout(Duration::from_secs(2)).unwrap();
|
||||
assert!(me.0.lock().unwrap().pending.is_empty());
|
||||
assert!(me.0.state.lock().unwrap().pending.is_empty());
|
||||
assert!(me.check().is_err());
|
||||
}
|
||||
|
||||
/// With nothing pending the deadline worker sleeps: no periodic wakeups (it used to poll
|
||||
/// every 100 ms for the life of the app, ten wakeups a second while idle).
|
||||
#[test]
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn an_idle_deadline_worker_does_not_wake_up() {
|
||||
let me = Readbacks::new().unwrap();
|
||||
std::thread::sleep(Duration::from_millis(350));
|
||||
let wakes = me.0.state.lock().unwrap().worker_wakes;
|
||||
assert!(wakes <= 1, "the worker woke {wakes} times while nothing was pending");
|
||||
// A readback that completes in time: the worker is told, and goes back to sleep.
|
||||
let ticket = me.start(|r| assert!(r.is_ok())).unwrap();
|
||||
ticket.complete(Ok(vec![1]));
|
||||
std::thread::sleep(Duration::from_millis(350));
|
||||
let wakes = me.0.state.lock().unwrap().worker_wakes;
|
||||
assert!(wakes <= 3, "the worker woke {wakes} times for one completed readback");
|
||||
assert!(me.check().is_ok());
|
||||
}
|
||||
|
||||
/// A deadline that passes wakes the worker exactly then, without the old polling loop.
|
||||
#[test]
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn a_pending_deadline_wakes_the_worker_when_it_is_due() {
|
||||
let me = Readbacks::new().unwrap();
|
||||
let (tx, rx) = std::sync::mpsc::channel();
|
||||
let _ticket = me
|
||||
.start(move |r| {
|
||||
assert!(r.is_err());
|
||||
tx.send(web_time::Instant::now()).unwrap();
|
||||
})
|
||||
.unwrap();
|
||||
let due = web_time::Instant::now() + Duration::from_millis(200);
|
||||
me.move_deadlines_for_test(due);
|
||||
let fired = rx.recv_timeout(Duration::from_secs(5)).expect("the deadline fires");
|
||||
assert!(fired >= due, "fired {:?} early", due - fired);
|
||||
assert!(fired < due + Duration::from_secs(2), "fired {:?} late", fired - due);
|
||||
assert!(me.check().is_err());
|
||||
}
|
||||
|
||||
/// Dropping the last handle lets the worker exit: it holds the state, not the context.
|
||||
#[test]
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn dropping_the_context_stops_the_deadline_worker() {
|
||||
let me = Readbacks::new().unwrap();
|
||||
let inner = Arc::downgrade(&me.0);
|
||||
drop(me);
|
||||
let t0 = web_time::Instant::now();
|
||||
while inner.upgrade().is_some() {
|
||||
assert!(t0.elapsed() < Duration::from_secs(5), "the worker is still holding the state");
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn missing_callback_times_out_once_and_rejects_late_success() {
|
||||
let me = Arc::new(Readbacks::default());
|
||||
|
||||
@@ -146,27 +146,58 @@ fn a_file_still_being_written_is_tried_again() {
|
||||
assert!(near(pixel(&s, cid, 32, 24), BLUE));
|
||||
}
|
||||
|
||||
/// The desktop app's watcher ([`footage_reload::Watcher`]) notices a footage file another app
|
||||
/// saved, hands its stamps over once, and reloads it through [`footage_reload::apply_stamps`];
|
||||
/// while nothing changes it hands over nothing (so the interface is never woken for it).
|
||||
#[test]
|
||||
fn background_scan_reloads_changed_footage() {
|
||||
let path = tmp("scan.psd");
|
||||
fn the_watcher_notices_a_footage_file_changed_on_disk() {
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::time::{Duration, Instant};
|
||||
let path = tmp("watched.psd");
|
||||
std::fs::write(&path, flat(RED, false)).unwrap();
|
||||
let (mut s, _, cid) = flat_in_comp(&path);
|
||||
let scan = |s: &mut Session| {
|
||||
footage_reload::start_scan(s);
|
||||
let t0 = std::time::Instant::now();
|
||||
let wakes = Arc::new(AtomicUsize::new(0));
|
||||
let counter = Arc::clone(&wakes);
|
||||
let fast = Duration::from_millis(20);
|
||||
let mut w = footage_reload::Watcher::start_with(
|
||||
s.services.clone(),
|
||||
move || {
|
||||
counter.fetch_add(1, Ordering::SeqCst);
|
||||
},
|
||||
fast,
|
||||
Duration::from_secs(60),
|
||||
)
|
||||
.unwrap();
|
||||
w.set_focused(true);
|
||||
w.set_watched(footage_reload::watched(&s));
|
||||
// Report `n`: the worker stores the stamps and then wakes the interface, so wait for the
|
||||
// wake too, not only for the stamps.
|
||||
let next = |w: &footage_reload::Watcher, n: usize| {
|
||||
let t0 = Instant::now();
|
||||
loop {
|
||||
if let Some(r) = footage_reload::poll_scan(s) {
|
||||
return r.unwrap();
|
||||
if wakes.load(Ordering::SeqCst) >= n
|
||||
&& let Some(stamps) = w.take()
|
||||
{
|
||||
return stamps;
|
||||
}
|
||||
assert!(t0.elapsed().as_secs() < 10, "scan never finished");
|
||||
std::thread::yield_now();
|
||||
assert!(t0.elapsed() < Duration::from_secs(10), "the watcher never reported");
|
||||
std::thread::sleep(Duration::from_millis(5));
|
||||
}
|
||||
};
|
||||
assert_eq!(scan(&mut s), json!({"files": 0, "reloaded": 0}));
|
||||
// First look: remembered, nothing to reload.
|
||||
assert_eq!(footage_reload::apply_stamps(&mut s, next(&w, 1)).unwrap(), json!({"files": 0, "reloaded": 0}));
|
||||
assert_eq!(wakes.load(Ordering::SeqCst), 1);
|
||||
std::thread::sleep(fast * 15);
|
||||
assert!(w.take().is_none(), "an unchanged file is never reported");
|
||||
assert_eq!(wakes.load(Ordering::SeqCst), 1, "and never wakes the interface");
|
||||
// Saved in another app: reported once, reloaded, shown.
|
||||
save(&path, &flat(GREEN, true));
|
||||
assert_eq!(scan(&mut s), json!({"files": 1, "reloaded": 1}));
|
||||
assert_eq!(footage_reload::apply_stamps(&mut s, next(&w, 2)).unwrap(), json!({"files": 1, "reloaded": 1}));
|
||||
assert!(near(pixel(&s, cid, 32, 24), GREEN));
|
||||
assert!(s.footage_scan.is_none());
|
||||
assert_eq!(wakes.load(Ordering::SeqCst), 2);
|
||||
std::thread::sleep(fast * 10);
|
||||
assert!(w.take().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -76,6 +76,8 @@ impl EffectcraftApp {
|
||||
gpu_checked: false,
|
||||
gpu_failures: None,
|
||||
gpu_failure_applied: None,
|
||||
footage_watcher: None,
|
||||
footage_watch_key: None,
|
||||
device_lost_close: None,
|
||||
dialog_answers: std::sync::mpsc::channel(),
|
||||
file_dialogs: FileDialogsUp::default(),
|
||||
|
||||
@@ -532,33 +532,40 @@ impl EffectcraftApp {
|
||||
}
|
||||
ctx.request_repaint_after(std::time::Duration::from_secs(10));
|
||||
}
|
||||
self.tick_footage_reload(ctx, now);
|
||||
self.tick_footage_reload(ctx);
|
||||
}
|
||||
|
||||
/// Settings ▸ Import ▸ Reload Footage Changed on Disk: every two seconds, and as soon as the
|
||||
/// window comes back to the front (after saving in another app), stamp the footage files on a
|
||||
/// worker thread and reload the ones that changed.
|
||||
fn tick_footage_reload(&mut self, ctx: &egui::Context, now: f64) {
|
||||
/// Settings ▸ Import ▸ Reload Footage Changed on Disk: a worker thread stamps the footage
|
||||
/// files (every two seconds while the window is in front, and as soon as it comes back to
|
||||
/// the front after saving in another app) and wakes this frame only when a stamp changed;
|
||||
/// the changed files are reloaded here. Nothing here asks for a repaint: an idle app with
|
||||
/// unchanged footage stays asleep.
|
||||
fn tick_footage_reload(&mut self, ctx: &egui::Context) {
|
||||
use effectcraft_engine::footage_reload;
|
||||
if let Some(Err(e)) = footage_reload::poll_scan(&mut self.session) {
|
||||
if let Some(stamps) = self.footage_watcher.as_ref().and_then(footage_reload::Watcher::take)
|
||||
&& let Err(e) = footage_reload::apply_stamps(&mut self.session, stamps)
|
||||
{
|
||||
self.ui.status = e.to_string();
|
||||
}
|
||||
let (last_key, focus_key) = (egui::Id::unique("footageReload.lastScan"), egui::Id::unique("footageReload.focused"));
|
||||
let focused = self.focused;
|
||||
let was_focused: bool = ctx.data(|d| d.get_temp(focus_key)).unwrap_or(focused);
|
||||
ctx.data_mut(|d| d.insert_temp(focus_key, focused));
|
||||
if self.session.prefs.import.auto_reload_footage == "off" || self.session.project.items.is_empty() {
|
||||
let off = self.session.prefs.import.auto_reload_footage == "off" || self.session.project.items.is_empty();
|
||||
if off {
|
||||
if let Some(w) = &mut self.footage_watcher {
|
||||
w.set_watched(Default::default());
|
||||
}
|
||||
self.footage_watch_key = None;
|
||||
return;
|
||||
}
|
||||
let last: f64 = ctx.data(|d| d.get_temp(last_key)).unwrap_or(0.0);
|
||||
if now - last >= 2.0 || (focused && !was_focused) {
|
||||
ctx.data_mut(|d| d.insert_temp(last_key, now));
|
||||
footage_reload::start_scan(&mut self.session);
|
||||
if self.footage_watcher.is_none() {
|
||||
let wake = ctx.clone();
|
||||
self.footage_watcher = footage_reload::Watcher::start(self.session.services.clone(), move || wake.request_repaint());
|
||||
}
|
||||
if self.session.footage_scan.is_some() {
|
||||
ctx.request_repaint_after(std::time::Duration::from_millis(100));
|
||||
} else {
|
||||
ctx.request_repaint_after(std::time::Duration::from_secs(2));
|
||||
let Some(watcher) = &mut self.footage_watcher else { return };
|
||||
watcher.set_focused(self.focused);
|
||||
// Which files are watched follows the project and the preference, not the frame.
|
||||
let key = (self.session.revision, self.session.prefs.import.auto_reload_footage.clone());
|
||||
if self.footage_watch_key.as_ref() != Some(&key) {
|
||||
self.footage_watch_key = Some(key);
|
||||
watcher.set_watched(footage_reload::watched(&self.session));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -312,6 +312,11 @@ pub struct EffectcraftApp {
|
||||
gpu_checked: bool,
|
||||
gpu_failures: Option<gpu_failure::GpuFailureBridge>,
|
||||
gpu_failure_applied: Option<bool>,
|
||||
/// Settings ▸ Import ▸ Reload Footage Changed on Disk: the worker that stamps the footage
|
||||
/// files, started with the first project that has items ([`Self::tick_footage_reload`]).
|
||||
footage_watcher: Option<effectcraft_engine::footage_reload::Watcher>,
|
||||
/// The project revision and preference the watched set was last computed for.
|
||||
footage_watch_key: Option<(u64, String)>,
|
||||
/// The graphics device was lost and the window is closing (see `close_after_device_loss`);
|
||||
/// `true` keeps crash recovery for the next launch.
|
||||
device_lost_close: Option<bool>,
|
||||
|
||||
+5
-3
@@ -120,9 +120,11 @@ Commands (CLI, MCP, control channel):
|
||||
- `import.reportMissingFrames`: gaps in an image sequence's numbering are reported on import
|
||||
(they show as colour-bar placeholders, as in After Effects; see [footage.md](footage.md))
|
||||
- `import.autoReloadFootage`: Automatically Reload Footage, `off`, `nonSequence` (the default, as
|
||||
in After Effects) or `all` (image sequences too). Every two seconds,
|
||||
and as soon as the window comes back to the front, the desktop app compares each footage file's
|
||||
size and modification time with what they were, and reloads what another app saved: a
|
||||
in After Effects) or `all` (image sequences too). Every two seconds while its window is in
|
||||
front (every ten seconds while another app has the front), and as soon as the window comes back
|
||||
to the front, the desktop app compares each footage file's size and modification time with what
|
||||
they were, on a worker thread that only wakes the interface when something changed, and reloads
|
||||
what another app saved: a
|
||||
Photoshop file edited in PhotoCraft or Photoshop shows its new pixels without File ▸ Reload
|
||||
Footage. It is not an undo step and doesn't modify the project; a layer item stays on its layer
|
||||
(found again by name), and a file that can't be read yet (still being written) is tried again once it changes again.
|
||||
|
||||
Reference in New Issue
Block a user