ez_ffmpeg/core/scheduler/progress/
handle.rs

1//! The cloneable, read-only progress handle.
2
3use super::snapshot::{OutputProgress, Progress};
4use super::state::ProgressState;
5use super::tracker::ProgressTracker;
6use crate::core::scheduler::ffmpeg_scheduler::{STATUS_ABORT, STATUS_END, STATUS_PAUSE};
7use std::sync::atomic::{AtomicUsize, Ordering};
8use std::sync::Arc;
9
10/// A cheap, cloneable, thread-safe, **read-only** view of a scheduled FFmpeg
11/// job's progress, obtained from `FfmpegScheduler::progress_handle` in the
12/// `Running` or `Paused` state.
13///
14/// The handle holds only reference-counted `std` synchronization state — no
15/// FFmpeg pointers, no callbacks — so it is `Send + Sync`, can be cloned
16/// freely across threads, and **remains safe to use after the scheduler has
17/// been consumed** (by `wait()`/`stop()`/`abort()`) **or dropped**: from
18/// then on [`snapshot`](Self::snapshot) keeps returning the job's final
19/// frozen values with [`ProgressState::Ended`].
20///
21/// Observing progress is pull-based: call `snapshot()` whenever a fresh
22/// value is wanted. Each call is a handful of atomic loads; no locks are
23/// taken and no FFmpeg call is made.
24#[derive(Clone)]
25pub struct ProgressHandle {
26    /// The scheduler's status word (`STATUS_*` in `ffmpeg_scheduler`).
27    status: Arc<AtomicUsize>,
28    /// The scheduler's seqlock-style pause parity: odd means a
29    /// pause()/resume() transition is in flight or the job is paused.
30    pause_epoch: Arc<AtomicUsize>,
31    tracker: Arc<ProgressTracker>,
32}
33
34impl ProgressHandle {
35    pub(crate) fn new(
36        status: Arc<AtomicUsize>,
37        pause_epoch: Arc<AtomicUsize>,
38        tracker: Arc<ProgressTracker>,
39    ) -> Self {
40        Self {
41            status,
42            pause_epoch,
43            tracker,
44        }
45    }
46
47    /// Takes a point-in-time snapshot of the job's progress.
48    pub fn snapshot(&self) -> Progress {
49        // Sample the state FIRST: reading it after the values would let a
50        // completion sealed in between produce a snapshot claiming `Ended`
51        // while its values predate the freeze. Sampled this way, a snapshot
52        // that says `Ended` was taken entirely after the latch sealed, so
53        // its values are the frozen finals.
54        let state = self.state();
55        let elapsed = self.tracker.elapsed();
56        let outputs = self
57            .tracker
58            .outputs()
59            .iter()
60            .enumerate()
61            .map(|(index, telemetry)| OutputProgress::collect(index, telemetry, elapsed))
62            .collect();
63        Progress::new(state, elapsed, outputs)
64    }
65
66    /// Whether every tracked worker of the job has torn down — the same
67    /// edge `wait()`/`stop()` unblock on, and the point at which
68    /// [`snapshot`](Self::snapshot) values (elapsed included) freeze.
69    ///
70    /// Note this is stricter than `FfmpegScheduler::is_ended`, which reports
71    /// the terminal status *signal* (a stopping job answers `true` there
72    /// while workers may still be flushing).
73    pub fn is_ended(&self) -> bool {
74        self.tracker.is_completed()
75    }
76
77    /// Derives the job-level state. Precedence: the sealed completion latch
78    /// wins (workers all gone — `Ended`), then a published terminal signal
79    /// (`Finishing`: teardown in flight), then pause, then the
80    /// producers-drained edge separating `Running` from `Finishing`.
81    fn state(&self) -> ProgressState {
82        if self.tracker.is_completed() {
83            return ProgressState::Ended;
84        }
85        match self.status.load(Ordering::Acquire) {
86            STATUS_PAUSE => ProgressState::Paused,
87            STATUS_END | STATUS_ABORT => ProgressState::Finishing,
88            _ => {
89                // STATUS_RUN (STATUS_INIT is unreachable through a handle:
90                // they are only obtainable after start() published RUN).
91                // An odd pause epoch marks the pause()/resume() transition
92                // windows, where "paused" is the conservative answer (see
93                // the epoch invariants on FfmpegScheduler::pause).
94                if self.pause_epoch.load(Ordering::Acquire) % 2 == 1 {
95                    ProgressState::Paused
96                } else if self.tracker.inputs_drained() {
97                    ProgressState::Finishing
98                } else {
99                    ProgressState::Running
100                }
101            }
102        }
103    }
104}
105
106impl std::fmt::Debug for ProgressHandle {
107    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
108        f.debug_struct("ProgressHandle")
109            .field("state", &self.state())
110            .field("is_ended", &self.is_ended())
111            .finish_non_exhaustive()
112    }
113}
ez_ffmpeg/core/scheduler/progress/handle.rs

ez_ffmpeg/core/scheduler/progress/
handle.rs
