1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
use std::{
sync::{
Mutex,
atomic::{AtomicU64, Ordering},
},
time::{Duration, Instant},
};
/// Shared wall-clock reference for pacing decoded frames to their
/// presentation time. Whichever branch (video, audio, ...) processes a
/// frame first sets the anchor; every other branch reads the same one, so
/// they agree on t=0 instead of each drifting from its own first frame.
///
/// Owned by [`crate::pipeline::Pipeline`] (one per pipeline, shared with
/// every [`crate::elements::Pacer`] via the `wire` closure) so
/// `Pipeline::pause`/`resume` can keep it in sync with the rest of the
/// pipeline — see those for why a `Pacer`, mid-playback, needs this to be
/// pause-aware and not just a fixed anchor.
pub struct Clock {
state: Mutex<State>,
/// Incremented before a control request starts cascading through the
/// pipeline. A `Pacer` compares this with the last generation it
/// acknowledged in `control()` so a long presentation-time wait can
/// return promptly and let the owning worker process that request.
interrupt_epoch: AtomicU64,
}
#[derive(Clone, Copy)]
enum State {
/// Never started — `start()` anchors to *now* on first call.
Unset,
Running {
start: Instant,
},
Paused {
start: Instant,
paused_at: Instant,
},
}
impl Default for Clock {
fn default() -> Self {
Self::new()
}
}
impl Clock {
pub fn new() -> Self {
Self {
state: Mutex::new(State::Unset),
interrupt_epoch: AtomicU64::new(0),
}
}
/// Signals paced waits to return without changing the clock's playback
/// anchor. The actual pause/seek/stop state change still happens through
/// the ordinary synchronous control cascade.
pub(crate) fn interrupt(&self) {
self.interrupt_epoch.fetch_add(1, Ordering::Release);
}
pub(crate) fn interrupt_epoch(&self) -> u64 {
self.interrupt_epoch.load(Ordering::Acquire)
}
/// The instant playback started, set on first call — shifted forward
/// on every [`Clock::resume`] by however long the clock spent paused,
/// so `now - start()` stays continuous across a pause/resume cycle
/// instead of jumping by the pause's real duration. Callers that pace
/// against this (see `Pacer::wait_for`) need to
/// call it fresh each time, not cache the first result — the whole
/// point is that it can move.
pub fn start(&self) -> Instant {
let mut state = self.state.lock().unwrap();
match *state {
State::Unset => {
let now = Instant::now();
*state = State::Running { start: now };
now
}
State::Running { start } => start,
State::Paused { start, .. } => start,
}
}
/// Pause-aware time elapsed since this clock was first anchored.
pub(crate) fn elapsed(&self) -> Duration {
let state = self.state.lock().unwrap();
match *state {
State::Unset => Duration::ZERO,
State::Running { start } => Instant::now().saturating_duration_since(start),
State::Paused { start, paused_at } => paused_at.saturating_duration_since(start),
}
}
/// Freezes the clock in place. No-op if unset (nothing running yet)
/// or already paused.
pub fn pause(&self) {
let mut state = self.state.lock().unwrap();
if let State::Running { start } = *state {
*state = State::Paused {
start,
paused_at: Instant::now(),
};
}
}
/// Undoes [`Clock::pause`] by shifting `start` forward by however long
/// this pause lasted. No-op if not currently paused.
pub fn resume(&self) {
let mut state = self.state.lock().unwrap();
if let State::Paused { start, paused_at } = *state {
let shift = Instant::now().saturating_duration_since(paused_at);
*state = State::Running {
start: start + shift,
};
}
}
/// Back to the same "never started" state as a freshly constructed
/// `Clock` — the next [`Clock::start`] call re-anchors t=0 to
/// *that* moment, same lazy-first-caller-wins semantics as initial
/// startup (see the type docs). Unconditional, regardless of current
/// state.
///
/// Called on [`crate::control::ControlMsg::Seek`]
/// (see [`crate::pipeline::Pipeline::seek`]): the old anchor measured
/// real time elapsed *for the pre-seek position* — after a jump, a
/// `Pacer`'s `elapsed_secs` (relative to its own now-reset
/// `first_pts`) starts over from ~0 too, so pairing it with the
/// stale anchor would compute a `due` far in the past and skip
/// sleeping entirely, dumping every post-seek frame with no pacing.
/// This is the wall-clock half of that same fix — `Pacer::first_pts`
/// resetting is the pts half; both are needed together.
pub fn reset(&self) {
let mut state = self.state.lock().unwrap();
*state = State::Unset;
}
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use super::*;
#[test]
fn pause_shifts_start_forward_by_the_pause_duration() {
let clock = Clock::new();
let first = clock.start();
clock.pause();
std::thread::sleep(Duration::from_millis(30));
clock.resume();
let after = clock.start();
assert!(
after >= first + Duration::from_millis(20),
"expected start() to shift forward by roughly the pause duration"
);
}
/// Regression test for the bug found manually testing `seek_render`:
/// without `reset()`, a `Pacer` re-anchoring only its `first_pts` (not
/// the shared `Clock`) after a seek computed `due` times far in the
/// past — `start()` kept returning the *original* anchor no matter
/// how long ago that was — so every post-seek frame skipped its sleep
/// entirely. `reset()` must make the next `start()` anchor to a fresh
/// "now", not the original one.
#[test]
fn reset_makes_the_next_start_anchor_to_a_fresh_now() {
let clock = Clock::new();
let original = clock.start();
std::thread::sleep(Duration::from_millis(30));
clock.reset();
let after_reset = clock.start();
assert!(
after_reset >= original + Duration::from_millis(20),
"expected start() after reset() to anchor to a fresh instant, \
not keep returning the original one"
);
}
}