long_screen/
driver.rs

1//! The driver API at the crate root (feature `driver`): [`run`] and its options, observer, control and result.
2
3use crate::context::{t, Ctx};
4pub use crate::engine::catalog::Locale;
5pub use crate::output::{CanvasInfo, RunResult};
6#[cfg(feature = "decode")]
7pub use crate::source::VideoSource;
8pub use crate::source::{Frame, FrameSource, MediaInfo, RawSource, SourceError};
9pub use crate::store::Store;
10use crate::value::{Js, Obj};
11use crate::{present, render, reveal, scan, solve};
12use std::fmt;
13use std::path::PathBuf;
14use std::sync::atomic::{AtomicBool, Ordering};
15use std::sync::Arc;
16use std::time::{Instant, SystemTime, UNIX_EPOCH};
17
18/// How conflicting content (an area that changed while it was on screen) is resolved.
19#[derive(Clone, Copy, Debug, PartialEq, Eq)]
20pub enum TemporalPolicy {
21    /// Keep the first moment the area was seen whole (the page's default).
22    Stable,
23    /// Replace it with each later complete moment.
24    Latest,
25}
26
27/// Whether the print is also shown inside the recording's own frame (chrome, toolbars).
28#[derive(Clone, Copy, Debug, PartialEq, Eq)]
29pub enum Framing {
30    /// Also build a framed presentation canvas from the first frame (the page's default).
31    Context,
32    /// Only the reconstructed content.
33    Region,
34}
35
36/// The run settings the page offers (`Settings` in src/types.ts), with its defaults.
37#[derive(Clone, Debug, PartialEq)]
38pub struct Settings {
39    /// Motion is analysed on frames scaled down so the longer side is at most this many pixels (default 640).
40    pub analysis_size: u32,
41    /// Working-memory budget of the tile cache, in MB (default 128).
42    pub memory_mb: f64,
43    /// Output tile side in pixels, a multiple of 16 (default 512).
44    pub tile_size: u32,
45    pub temporal_policy: TemporalPolicy,
46    pub framing: Framing,
47}
48
49impl Default for Settings {
50    fn default() -> Settings {
51        Settings {
52            analysis_size: 640,
53            memory_mb: 128.0,
54            tile_size: 512,
55            temporal_policy: TemporalPolicy::Stable,
56            framing: Framing::Context,
57        }
58    }
59}
60
61impl Settings {
62    pub(crate) fn framing_context(&self) -> bool {
63        self.framing == Framing::Context
64    }
65
66    pub(crate) fn stable(&self) -> bool {
67        self.temporal_policy == TemporalPolicy::Stable
68    }
69
70    /// The `settings` object the project row persists, in the page's field order.
71    fn to_js(&self) -> Obj {
72        Obj::new()
73            .with("analysisSize", self.analysis_size)
74            .with("memoryMB", self.memory_mb)
75            .with("tileSize", self.tile_size)
76            .with("regions", Js::Arr(Vec::new()))
77            .with(
78                "temporalPolicy",
79                if self.stable() { "stable" } else { "latest" },
80            )
81            .with("decoder", "precise")
82            .with("compatibilityFPS", 60u32)
83            .with(
84                "framing",
85                if self.framing_context() {
86                    "context"
87                } else {
88                    "region"
89                },
90            )
91            .with("compute", "cpu")
92    }
93}
94
95/// Everything a run takes besides its source.
96#[derive(Clone, Debug)]
97pub struct Options {
98    pub settings: Settings,
99    /// The language of the diagnostics and progress messages the observer and the result show (default English).
100    /// Persisted rows are always written in zh, as the page writes them.
101    pub locale: Locale,
102    /// Threads the kernels run on; `None` uses rayon's global pool (one per core). Without the `rayon` feature the
103    /// kernels run on the calling thread, and more than one thread is an [`Error::Threads`].
104    pub threads: Option<usize>,
105    /// Keep encoded tiles in memory up to this many bytes, then move them to `spill_dir` (default: no limit).
106    pub store_budget: Option<usize>,
107    /// Where encoded tiles go past `store_budget` (default: a fresh directory under the system temp directory).
108    pub spill_dir: Option<PathBuf>,
109    /// The settings object to persist instead of the one `settings` describes (the parity harness replays the
110    /// page's own).
111    #[doc(hidden)]
112    pub settings_row: Option<Obj>,
113}
114
115impl Default for Options {
116    fn default() -> Options {
117        Options {
118            settings: Settings::default(),
119            locale: Locale::En,
120            threads: None,
121            store_budget: None,
122            spill_dir: None,
123            settings_row: None,
124        }
125    }
126}
127
128/// One progress event.
129#[derive(Clone, Debug, PartialEq)]
130pub struct Progress {
131    /// `scanning`, `solving`, `optimizing`, `rendering`, `revealing`, `framing`, `pyramid`, then `complete` or
132    /// `partial`.
133    pub phase: String,
134    /// 0…1 within the phase.
135    pub fraction: f64,
136    /// Frames done in the phase.
137    pub frames: u32,
138    /// Frames the container lists, when it lists them.
139    pub total: Option<u64>,
140    /// Seconds into the recording.
141    pub time: f64,
142    pub message: String,
143    /// The canvas the event concerns, if any.
144    pub canvas: Option<String>,
145}
146
147/// One diagnostic, in the run's display language.
148#[derive(Clone, Debug, PartialEq)]
149pub struct Diagnostic {
150    pub code: String,
151    /// `info`, `warning` or `error`.
152    pub severity: String,
153    pub message: String,
154    pub action: Option<String>,
155    pub frame: Option<u32>,
156    pub time: Option<f64>,
157    pub canvas: Option<String>,
158    /// How many times this code has fired so far in the run.
159    pub occurrences: u32,
160    /// The whole row (as the page stores it, but in the display language).
161    pub row: Obj,
162}
163
164impl Diagnostic {
165    pub(crate) fn from_row(row: Obj, occurrences: u32) -> Diagnostic {
166        let num = |k: &str| row.get(k).map(Js::num).filter(|n| n.is_finite());
167        Diagnostic {
168            code: row.str("code").unwrap_or_default().to_string(),
169            severity: row.str("severity").unwrap_or_default().to_string(),
170            message: row.str("message").unwrap_or_default().to_string(),
171            action: row.str("action").map(String::from),
172            frame: num("frame").map(|f| f as u32),
173            time: num("time"),
174            canvas: row.str("canvasId").map(String::from),
175            occurrences,
176            row,
177        }
178    }
179}
180
181/// Receives a run's events on the run's thread. Both methods default to doing nothing.
182pub trait Observer: Send {
183    fn progress(&mut self, _progress: &Progress) {}
184    fn diagnostic(&mut self, _diagnostic: &Diagnostic) {}
185}
186
187/// No events.
188impl Observer for () {}
189
190/// A closure observes progress only.
191impl<F: FnMut(&Progress) + Send> Observer for F {
192    fn progress(&mut self, progress: &Progress) {
193        self(progress)
194    }
195}
196
197/// Pause and stop for a running [`run`], from any thread. Clones share the same flags. A stop ends the current
198/// pass at its next step and keeps what is done (the result is `partial`); a pause holds the run between steps.
199#[derive(Clone, Debug, Default)]
200pub struct Control {
201    stop: Arc<AtomicBool>,
202    paused: Arc<AtomicBool>,
203}
204
205impl Control {
206    pub fn new() -> Control {
207        Control::default()
208    }
209
210    /// Asks the run to stop (and wakes it if paused).
211    pub fn stop(&self) {
212        self.stop.store(true, Ordering::SeqCst);
213    }
214
215    pub fn set_paused(&self, paused: bool) {
216        self.paused.store(paused, Ordering::SeqCst);
217    }
218
219    pub fn is_paused(&self) -> bool {
220        self.paused.load(Ordering::SeqCst)
221    }
222
223    pub(crate) fn stop_requested(&self) -> bool {
224        self.stop.load(Ordering::SeqCst)
225    }
226
227    pub(crate) fn take_stop(&self) -> bool {
228        self.stop.swap(false, Ordering::SeqCst)
229    }
230}
231
232/// Why a run could not produce anything.
233#[derive(Debug)]
234pub enum Error {
235    /// The settings cannot be used (for example a tile size that is not a multiple of 16).
236    Settings(String),
237    /// The thread pool could not be built.
238    Threads(String),
239    /// The run failed before any frame was rendered; the result's diagnostics say more.
240    Failed {
241        message: String,
242        result: Box<RunResult>,
243    },
244}
245
246impl fmt::Display for Error {
247    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
248        match self {
249            Error::Settings(m) => write!(f, "invalid settings: {m}"),
250            Error::Threads(m) => write!(f, "could not start the thread pool: {m}"),
251            Error::Failed { message, .. } => f.write_str(message),
252        }
253    }
254}
255
256impl std::error::Error for Error {}
257
258/// A run id like the page's: unique, and not a word that occurs in any message.
259fn run_id() -> String {
260    let nanos = SystemTime::now()
261        .duration_since(UNIX_EPOCH)
262        .map_or(0, |d| d.as_nanos());
263    format!("n{:x}{:x}", nanos, std::process::id())
264}
265
266/// `new Date().toISOString()`.
267fn iso_now() -> String {
268    let d = SystemTime::now()
269        .duration_since(UNIX_EPOCH)
270        .unwrap_or_default();
271    let (secs, millis) = (d.as_secs() as i64, d.subsec_millis());
272    let (days, rem) = (secs.div_euclid(86400), secs.rem_euclid(86400));
273    // Civil date from days since 1970-01-01 (Howard Hinnant's algorithm).
274    let z = days + 719468;
275    let era = z.div_euclid(146097);
276    let doe = z - era * 146097;
277    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
278    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
279    let mp = (5 * doy + 2) / 153;
280    let day = doy - (153 * mp + 2) / 5 + 1;
281    let month = if mp < 10 { mp + 3 } else { mp - 9 };
282    let year = yoe + era * 400 + (month <= 2) as i64;
283    format!(
284        "{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}.{millis:03}Z",
285        rem / 3600,
286        rem / 60 % 60,
287        rem % 60
288    )
289}
290
291/// Runs the whole reconstruction on `source`. Progress and diagnostics go to `observer` as they happen; `control`
292/// pauses or stops the run from another thread. Returns the result even when the run ended `partial` (a stop, or a
293/// failure after some frames were rendered); only a run that rendered nothing is an [`Error::Failed`].
294pub fn run(
295    source: &mut dyn FrameSource,
296    options: &Options,
297    observer: &mut dyn Observer,
298    control: &Control,
299) -> Result<RunResult, Error> {
300    let s = &options.settings;
301    if s.tile_size < 16 || !s.tile_size.is_multiple_of(16) {
302        return Err(Error::Settings(
303            "the tile size must be a positive multiple of 16".into(),
304        ));
305    }
306    #[cfg(feature = "rayon")]
307    match options.threads {
308        Some(n) => {
309            let pool = rayon::ThreadPoolBuilder::new()
310                .num_threads(n.max(1))
311                .build()
312                .map_err(|e| Error::Threads(e.to_string()))?;
313            pool.install(|| run_in_pool(source, options, observer, control))
314        }
315        None => run_in_pool(source, options, observer, control),
316    }
317    // Without rayon the kernels run on the calling thread only.
318    #[cfg(not(feature = "rayon"))]
319    match options.threads {
320        Some(n) if n > 1 => Err(Error::Threads(
321            "this build has no thread pool (the `rayon` feature is off)".into(),
322        )),
323        _ => run_in_pool(source, options, observer, control),
324    }
325}
326
327/// Threads the kernels run on in the current pool.
328fn pool_threads() -> usize {
329    #[cfg(feature = "rayon")]
330    return rayon::current_num_threads();
331    #[cfg(not(feature = "rayon"))]
332    return 1;
333}
334
335fn run_in_pool(
336    source: &mut dyn FrameSource,
337    options: &Options,
338    observer: &mut dyn Observer,
339    control: &Control,
340) -> Result<RunResult, Error> {
341    let store = match (options.store_budget, &options.spill_dir) {
342        (Some(budget), dir) => {
343            let dir = dir
344                .clone()
345                .unwrap_or_else(|| std::env::temp_dir().join(format!("long-screen-{}", run_id())));
346            Store::spilling(dir, budget).map_err(|e| Error::Settings(e.to_string()))?
347        }
348        (None, _) => Store::in_memory(),
349    };
350    let settings_js = options
351        .settings_row
352        .clone()
353        .unwrap_or_else(|| options.settings.to_js());
354    let info = source.info().clone();
355    let mut ctx = Ctx::new(
356        info,
357        observer,
358        options.settings.clone(),
359        settings_js,
360        control.clone(),
361        options.locale,
362        store,
363        pool_threads(),
364    )
365    .map_err(Error::Settings)?;
366    ctx.run_id = run_id();
367    ctx.created = iso_now();
368    let started = Instant::now();
369    ctx.persist();
370    startup_diagnostics(&mut ctx);
371    let outcome = stages(&mut ctx, source);
372    let failure = match outcome {
373        Ok(()) => None,
374        Err(message) => {
375            handle_failure(&mut ctx, &message);
376            Some(message)
377        }
378    };
379    let result = RunResult::new(ctx, started.elapsed().as_secs_f64());
380    match failure {
381        Some(message) if result.rendered_frames() == 0 => Err(Error::Failed {
382            message,
383            result: Box::new(result),
384        }),
385        _ => Ok(result),
386    }
387}
388
389/// The run's assumptions and inputs, before the first stage.
390fn startup_diagnostics(ctx: &mut Ctx) {
391    let noise = ctx.noise;
392    let (mode, codec) = (ctx.info.mode.clone(), ctx.info.codec.clone());
393    ctx.emit(&|l| {
394        let margin = if noise != 0.0 {
395            t(l, "diag.MODEL_ASSUMPTIONS.marginLossy", &[])
396        } else {
397            t(l, "diag.MODEL_ASSUMPTIONS.marginLossless", &[])
398        };
399        Obj::new()
400            .with("code", "MODEL_ASSUMPTIONS")
401            .with("severity", "info")
402            .with(
403                "message",
404                t(
405                    l,
406                    "diag.MODEL_ASSUMPTIONS.message",
407                    &[
408                        ("noise", crate::engine::catalog::Param::Num(noise)),
409                        ("margin", crate::engine::catalog::Param::Str(&margin)),
410                    ],
411                ),
412            )
413            .with("action", t(l, "diag.MODEL_ASSUMPTIONS.action", &[]))
414            .with(
415                "detail",
416                Obj::new()
417                    .with("noise", noise)
418                    .with("lossless", noise == 0.0)
419                    .with("source", mode.as_str())
420                    .with("codec", codec.as_str()),
421            )
422    });
423    if ctx.settings_js.str("decoder") == Some("compatibility") {
424        let fps = ctx.settings_js.num("compatibilityFPS");
425        ctx.emit(&|l| {
426            Obj::new()
427                .with("code", "APPROXIMATE_DECODER")
428                .with("severity", "warning")
429                .with(
430                    "message",
431                    t(
432                        l,
433                        "diag.APPROXIMATE_DECODER.message",
434                        &[("fps", crate::engine::catalog::Param::Num(fps))],
435                    ),
436                )
437                .with("action", t(l, "diag.APPROXIMATE_DECODER.action", &[]))
438        });
439    }
440    let size = ctx.settings.analysis_size;
441    if ctx.info.width > size as usize || ctx.info.height > size as usize {
442        ctx.emit(&|l| {
443            Obj::new()
444                .with("code", "ANALYSIS_PYRAMID")
445                .with("severity", "info")
446                .with(
447                    "message",
448                    t(
449                        l,
450                        "diag.ANALYSIS_PYRAMID.message",
451                        &[("size", crate::engine::catalog::Param::Int(size as i64))],
452                    ),
453                )
454        });
455    }
456    for warning in ctx.info.warnings.clone() {
457        ctx.emit(&|_| {
458            Obj::new()
459                .with("code", "MEDIA_NOTICE")
460                .with("severity", "warning")
461                .with("message", warning.as_str())
462        });
463    }
464    let (w, h) = (ctx.info.width as f64, ctx.info.height as f64);
465    if w * h * 4.0 * 4.0 > ctx.settings.memory_mb * 1024.0 * 1024.0 * 0.8 {
466        ctx.emit(&|l| {
467            Obj::new()
468                .with("code", "FRAME_MEMORY_PRESSURE")
469                .with("severity", "warning")
470                .with("message", t(l, "diag.FRAME_MEMORY_PRESSURE.message", &[]))
471                .with("action", t(l, "diag.FRAME_MEMORY_PRESSURE.action", &[]))
472        });
473    }
474}
475
476fn ms(since: Instant) -> f64 {
477    since.elapsed().as_secs_f64() * 1000.0
478}
479
480/// The stages in the page's order: scan, solve, render, reveal, framing, pyramid.
481fn stages(ctx: &mut Ctx, source: &mut dyn FrameSource) -> Result<(), String> {
482    let at = Instant::now();
483    let features = scan::scan(ctx, source)?;
484    ctx.timing("scanMS", ms(at));
485    if ctx.frames == 0 {
486        return Err(
487            "No observations could be decoded. Nothing has been marked as reconstructed.".into(),
488        );
489    }
490    let at = Instant::now();
491    solve::solve(ctx, source, features)?;
492    ctx.timing("solveMS", ms(at));
493    let at = Instant::now();
494    render::render(ctx, source)?;
495    ctx.timing("renderMS", ms(at));
496    reveal::reveal(ctx, source)?;
497    let at = Instant::now();
498    present::framing(ctx);
499    ctx.timing("framingMS", ms(at));
500    let at = Instant::now();
501    ctx.phase = "pyramid";
502    ctx.status = "pyramid";
503    ctx.persist();
504    present::pyramid(ctx);
505    if ctx.stop_requested {
506        ctx.partial = true;
507        ctx.stop_requested = false;
508    }
509    ctx.timing("pyramidMS", ms(at));
510    let stats = ctx.tiles.cache.stats();
511    let mut performance = ctx.timings.clone();
512    performance.set("exactDuplicateFrames", ctx.duplicates);
513    performance.set("skippedPaints", ctx.skipped_paints);
514    performance.set("tileEncodes", stats.encoded);
515    performance.set("tileDecodes", stats.decoded);
516    performance.set("tileEvictions", stats.evictions);
517    performance.set(
518        "compute",
519        Obj::new()
520            .with("backend", "cpu")
521            .with("threads", ctx.threads),
522    );
523    performance.set("consistencyVotedLayers", ctx.voted_layers);
524    performance.set("consistencyThinLayers", ctx.thin_layers);
525    performance.set(
526        "note",
527        "Stage timings include decoding, storage and yields; not GPU-only kernel time.",
528    );
529    ctx.put("performance", performance);
530    ctx.status = if ctx.partial { "partial" } else { "complete" };
531    ctx.persist();
532    let message = ctx.tr(
533        if ctx.partial {
534            "progress.runPartial"
535        } else {
536            "progress.runComplete"
537        },
538        &[],
539    );
540    let (status, rendered, duration) = (ctx.status, ctx.rendered_frames, ctx.info.duration);
541    ctx.event(status, 1.0, rendered, duration, message);
542    Ok(())
543}
544
545/// The page's top-level catch: keep what was committed, journal the failure.
546fn handle_failure(ctx: &mut Ctx, message: &str) {
547    ctx.status = if ctx.rendered_frames > 0 {
548        "partial"
549    } else {
550        "error"
551    };
552    ctx.error = Some(message.to_string());
553    let message = message.to_string();
554    ctx.emit(&|l| {
555        Obj::new()
556            .with("code", "PROCESSING_ERROR")
557            .with("severity", "error")
558            .with("message", message.as_str())
559            .with("action", t(l, "diag.RUN_FAILURE.action", &[]))
560    });
561    let run = ctx.key("");
562    let _ = ctx.tiles.flush(&mut ctx.store, &run, None);
563    ctx.persist();
564}
long_screen/driver.rs

long_screen/
driver.rs
