media_pp\core/
playback_clock.rs

1use std::{
2    sync::{Arc, Mutex},
3    time::Duration,
4};
5
6use thiserror::Error as ThisError;
7
8use crate::clock::Clock;
9
10/// Which source currently defines the pipeline's media position.
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum PlaybackMaster {
13    /// No timestamped stream has established a position yet.
14    Unavailable,
15    /// Media position advances from the pipeline's pause-aware wall clock.
16    Wall,
17    /// An audio renderer owns the clock but has not started the endpoint yet.
18    AudioPriming,
19    /// An audio endpoint's played-sample position is the master clock.
20    Audio,
21}
22
23#[derive(Debug, Clone, Copy, PartialEq, Eq, ThisError)]
24pub enum PlaybackClockError {
25    #[error("this pipeline already has an audio playback-clock master")]
26    AudioMasterAlreadyRegistered,
27
28    #[error("the audio playback-clock registration is stale")]
29    StaleAudioMaster,
30}
31
32/// Pipeline-wide media clock shared by audio output and video scheduling.
33///
34/// [`Clock`] remains the pipeline's monotonic control/pause clock. This
35/// type adds the media-timeline position and can hand that position from a
36/// wall-clock fallback to one audio renderer without letting the position
37/// jump backwards. It deliberately contains no WASAPI types: an audio
38/// backend publishes device-position snapshots through its private
39/// registration, while video scheduling only reads the resulting position.
40pub struct PlaybackClock {
41    wall_clock: Arc<Clock>,
42    state: Mutex<State>,
43}
44
45#[derive(Clone, Copy)]
46enum State {
47    Unavailable {
48        next_registration: u64,
49    },
50    Wall {
51        anchor_ns: i64,
52        anchor_elapsed: Duration,
53        next_registration: u64,
54    },
55    AudioPriming {
56        registration: u64,
57        held_ns: Option<i64>,
58        next_registration: u64,
59    },
60    // Only an audio renderer moves the clock into these two, and the only one
61    // in this crate is behind `wasapi-renderer`. They are dead in a build
62    // without it, but they are the timeline contract `PlaybackClock` exists to
63    // provide — gating them on a backend feature would invert that. See
64    // `AudioMasterRegistration`.
65    #[allow(dead_code)]
66    Audio {
67        registration: u64,
68        position_ns: i64,
69        sampled_elapsed: Duration,
70        submitted_until_ns: i64,
71        running: bool,
72        next_registration: u64,
73    },
74    #[allow(dead_code)]
75    AudioFallback {
76        registration: u64,
77        anchor_ns: i64,
78        anchor_elapsed: Duration,
79        next_registration: u64,
80    },
81}
82
83impl PlaybackClock {
84    pub(crate) fn new(wall_clock: Arc<Clock>) -> Self {
85        Self {
86            wall_clock,
87            state: Mutex::new(State::Unavailable {
88                next_registration: 1,
89            }),
90        }
91    }
92
93    pub fn master(&self) -> PlaybackMaster {
94        match *self.state.lock().unwrap() {
95            State::Unavailable { .. } => PlaybackMaster::Unavailable,
96            State::Wall { .. } | State::AudioFallback { .. } => PlaybackMaster::Wall,
97            State::AudioPriming { .. } => PlaybackMaster::AudioPriming,
98            State::Audio { .. } => PlaybackMaster::Audio,
99        }
100    }
101
102    #[cfg(test)]
103    pub(crate) fn position_ns(&self) -> Option<i64> {
104        let state = self.state.lock().unwrap();
105        position_at(*state, self.wall_clock.elapsed())
106    }
107
108    pub(crate) fn interrupt_epoch(&self) -> u64 {
109        self.wall_clock.interrupt_epoch()
110    }
111
112    /// Establishes a wall-clock media origin if no stream owns one yet.
113    /// Returns the current position after doing so.
114    #[cfg(test)]
115    pub(crate) fn ensure_wall_origin(&self, media_ns: i64) -> Option<i64> {
116        let mut state = self.state.lock().unwrap();
117        if let State::Unavailable { next_registration } = *state {
118            self.wall_clock.start();
119            let elapsed = self.wall_clock.elapsed();
120            *state = State::Wall {
121                anchor_ns: media_ns,
122                anchor_elapsed: elapsed,
123                next_registration,
124            };
125        }
126        position_at(*state, self.wall_clock.elapsed())
127    }
128
129    pub(crate) fn video_snapshot(&self, media_ns: i64) -> (PlaybackMaster, Option<i64>) {
130        let mut state = self.state.lock().unwrap();
131        if let State::Unavailable { next_registration } = *state {
132            self.wall_clock.start();
133            let elapsed = self.wall_clock.elapsed();
134            *state = State::Wall {
135                anchor_ns: media_ns,
136                anchor_elapsed: elapsed,
137                next_registration,
138            };
139        }
140        let elapsed = self.wall_clock.elapsed();
141        let master = match *state {
142            State::Unavailable { .. } => PlaybackMaster::Unavailable,
143            State::Wall { .. } | State::AudioFallback { .. } => PlaybackMaster::Wall,
144            State::AudioPriming { .. } => PlaybackMaster::AudioPriming,
145            State::Audio { .. } => PlaybackMaster::Audio,
146        };
147        (master, position_at(*state, elapsed))
148    }
149
150    /// Claims the timeline for one audio renderer. Unused in a build without
151    /// an audio renderer (see `AudioMasterRegistration`), hence the `allow`.
152    #[allow(dead_code)]
153    pub(crate) fn register_audio_master(
154        self: &Arc<Self>,
155    ) -> Result<AudioMasterRegistration, PlaybackClockError> {
156        let mut state = self.state.lock().unwrap();
157        let elapsed = self.wall_clock.elapsed();
158        let (held_ns, registration, next_registration) = match *state {
159            State::Unavailable { next_registration } => {
160                (None, next_registration, next_registration.wrapping_add(1))
161            }
162            State::Wall {
163                next_registration, ..
164            } => (
165                position_at(*state, elapsed),
166                next_registration,
167                next_registration.wrapping_add(1),
168            ),
169            State::AudioPriming { .. } | State::Audio { .. } | State::AudioFallback { .. } => {
170                return Err(PlaybackClockError::AudioMasterAlreadyRegistered);
171            }
172        };
173        *state = State::AudioPriming {
174            registration,
175            held_ns,
176            next_registration,
177        };
178        Ok(AudioMasterRegistration {
179            clock: self.clone(),
180            registration,
181        })
182    }
183
184    /// Resets media state for a seek while retaining the current audio
185    /// renderer's ownership. The next timestamp/device sample establishes
186    /// the post-seek position.
187    pub(crate) fn reset_for_seek(&self) {
188        let mut state = self.state.lock().unwrap();
189        *state = match *state {
190            State::Unavailable { next_registration }
191            | State::Wall {
192                next_registration, ..
193            } => State::Unavailable { next_registration },
194            State::AudioPriming {
195                registration,
196                next_registration,
197                ..
198            }
199            | State::Audio {
200                registration,
201                next_registration,
202                ..
203            }
204            | State::AudioFallback {
205                registration,
206                next_registration,
207                ..
208            } => State::AudioPriming {
209                registration,
210                held_ns: None,
211                next_registration,
212            },
213        };
214    }
215
216    #[allow(dead_code)]
217    fn release_audio_master(&self, registration: u64) {
218        let mut state = self.state.lock().unwrap();
219        let elapsed = self.wall_clock.elapsed();
220        let (matches, next_registration) = match *state {
221            State::AudioPriming {
222                registration: current,
223                next_registration,
224                ..
225            }
226            | State::Audio {
227                registration: current,
228                next_registration,
229                ..
230            }
231            | State::AudioFallback {
232                registration: current,
233                next_registration,
234                ..
235            } => (current == registration, next_registration),
236            State::Unavailable { .. } | State::Wall { .. } => return,
237        };
238        if !matches {
239            return;
240        }
241        *state = match position_at(*state, elapsed) {
242            Some(anchor_ns) => State::Wall {
243                anchor_ns,
244                anchor_elapsed: elapsed,
245                next_registration,
246            },
247            None => State::Unavailable { next_registration },
248        };
249    }
250}
251
252/// Exclusive, generation-checked writer owned by one audio renderer.
253/// Dropping it hands the last known position back to the wall clock.
254///
255/// The only audio renderer in this crate is `WasapiRenderer`, behind the
256/// `wasapi-renderer` feature, so a build without it constructs this nowhere and
257/// every method below is dead. That is why the `allow`s here are deliberate
258/// rather than a `cfg(feature = "wasapi-renderer")` gate: `PlaybackClock` is
259/// the backend-independent timeline every renderer binds to, and teaching it
260/// about one backend's Cargo feature would invert that relationship. The
261/// crate's own tests exercise this path, so it is covered even when no shipped
262/// element uses it.
263#[allow(dead_code)]
264pub(crate) struct AudioMasterRegistration {
265    clock: Arc<PlaybackClock>,
266    registration: u64,
267}
268
269#[allow(dead_code)]
270impl AudioMasterRegistration {
271    pub(crate) fn priming_target_ns(&self) -> Result<Option<i64>, PlaybackClockError> {
272        match *self.clock.state.lock().unwrap() {
273            State::AudioPriming {
274                registration,
275                held_ns,
276                ..
277            } if registration == self.registration => Ok(held_ns),
278            State::Audio { registration, .. } if registration == self.registration => Ok(None),
279            State::AudioFallback { registration, .. } if registration == self.registration => {
280                Ok(None)
281            }
282            _ => Err(PlaybackClockError::StaleAudioMaster),
283        }
284    }
285
286    pub(crate) fn publish(
287        &self,
288        position_ns: i64,
289        submitted_until_ns: i64,
290        running: bool,
291    ) -> Result<(), PlaybackClockError> {
292        let mut state = self.clock.state.lock().unwrap();
293        self.clock.wall_clock.start();
294        let elapsed = self.clock.wall_clock.elapsed();
295        let (held_ns, next_registration) = match *state {
296            State::AudioPriming {
297                registration,
298                held_ns,
299                next_registration,
300            } if registration == self.registration => (held_ns, next_registration),
301            State::Audio {
302                registration,
303                next_registration,
304                ..
305            } if registration == self.registration => (None, next_registration),
306            State::AudioFallback {
307                registration,
308                next_registration,
309                ..
310            } if registration == self.registration => (None, next_registration),
311            _ => return Err(PlaybackClockError::StaleAudioMaster),
312        };
313
314        // A master handoff must never make video scheduling move backwards.
315        let position_ns = held_ns.map_or(position_ns, |held| position_ns.max(held));
316        let submitted_until_ns = submitted_until_ns.max(position_ns);
317        *state = State::Audio {
318            registration: self.registration,
319            position_ns,
320            sampled_elapsed: elapsed,
321            submitted_until_ns,
322            running,
323            next_registration,
324        };
325        Ok(())
326    }
327
328    /// Audio ended before another stream: continue from its final played
329    /// position using the wall clock while retaining this registration so
330    /// a second renderer cannot race the still-attached one.
331    pub(crate) fn finish(&self, position_ns: i64) -> Result<(), PlaybackClockError> {
332        let mut state = self.clock.state.lock().unwrap();
333        let elapsed = self.clock.wall_clock.elapsed();
334        let next_registration = match *state {
335            State::AudioPriming {
336                registration,
337                next_registration,
338                ..
339            }
340            | State::Audio {
341                registration,
342                next_registration,
343                ..
344            } if registration == self.registration => next_registration,
345            _ => return Err(PlaybackClockError::StaleAudioMaster),
346        };
347        *state = State::AudioFallback {
348            registration: self.registration,
349            anchor_ns: position_ns,
350            anchor_elapsed: elapsed,
351            next_registration,
352        };
353        Ok(())
354    }
355
356    pub(crate) fn reset_for_seek(&self) -> Result<(), PlaybackClockError> {
357        let mut state = self.clock.state.lock().unwrap();
358        let next_registration = match *state {
359            State::AudioPriming {
360                registration,
361                next_registration,
362                ..
363            }
364            | State::Audio {
365                registration,
366                next_registration,
367                ..
368            }
369            | State::AudioFallback {
370                registration,
371                next_registration,
372                ..
373            } if registration == self.registration => next_registration,
374            _ => return Err(PlaybackClockError::StaleAudioMaster),
375        };
376        *state = State::AudioPriming {
377            registration: self.registration,
378            held_ns: None,
379            next_registration,
380        };
381        Ok(())
382    }
383}
384
385impl Drop for AudioMasterRegistration {
386    fn drop(&mut self) {
387        self.clock.release_audio_master(self.registration);
388    }
389}
390
391fn position_at(state: State, elapsed: Duration) -> Option<i64> {
392    match state {
393        State::Unavailable { .. } => None,
394        State::Wall {
395            anchor_ns,
396            anchor_elapsed,
397            ..
398        } => Some(add_duration(
399            anchor_ns,
400            elapsed.saturating_sub(anchor_elapsed),
401        )),
402        State::AudioPriming { held_ns, .. } => held_ns,
403        State::Audio {
404            position_ns,
405            sampled_elapsed,
406            submitted_until_ns,
407            running,
408            ..
409        } => {
410            let projected = if running {
411                add_duration(position_ns, elapsed.saturating_sub(sampled_elapsed))
412            } else {
413                position_ns
414            };
415            Some(projected.min(submitted_until_ns))
416        }
417        State::AudioFallback {
418            anchor_ns,
419            anchor_elapsed,
420            ..
421        } => Some(add_duration(
422            anchor_ns,
423            elapsed.saturating_sub(anchor_elapsed),
424        )),
425    }
426}
427
428fn add_duration(value_ns: i64, duration: Duration) -> i64 {
429    let delta = duration.as_nanos().min(i64::MAX as u128) as i64;
430    value_ns.saturating_add(delta)
431}
432
433#[cfg(test)]
434mod tests {
435    use std::{thread, time::Duration};
436
437    use super::*;
438
439    #[test]
440    fn wall_origin_advances_and_freezes_with_pipeline_clock() {
441        let wall = Arc::new(Clock::new());
442        let playback = PlaybackClock::new(wall.clone());
443        assert!(playback.ensure_wall_origin(1_000).unwrap() >= 1_000);
444        thread::sleep(Duration::from_millis(20));
445        assert!(playback.position_ns().unwrap() >= 10_000_000);
446
447        wall.pause();
448        let paused = playback.position_ns().unwrap();
449        thread::sleep(Duration::from_millis(20));
450        assert_eq!(playback.position_ns(), Some(paused));
451    }
452
453    #[test]
454    fn audio_handoff_never_moves_backwards_and_release_continues_on_wall() {
455        let wall = Arc::new(Clock::new());
456        let playback = Arc::new(PlaybackClock::new(wall));
457        playback.ensure_wall_origin(50_000_000);
458        let audio = playback.register_audio_master().unwrap();
459        let held = audio.priming_target_ns().unwrap().unwrap();
460
461        audio
462            .publish(held - 10_000_000, held + 100_000_000, true)
463            .unwrap();
464        assert!(playback.position_ns().unwrap() >= held);
465        drop(audio);
466        let released = playback.position_ns().unwrap();
467        thread::sleep(Duration::from_millis(10));
468        assert!(playback.position_ns().unwrap() >= released);
469        assert_eq!(playback.master(), PlaybackMaster::Wall);
470    }
471
472    #[test]
473    fn only_one_audio_master_can_publish_and_seek_retains_its_generation() {
474        let wall = Arc::new(Clock::new());
475        let playback = Arc::new(PlaybackClock::new(wall));
476        let audio = playback.register_audio_master().unwrap();
477        assert!(matches!(
478            playback.register_audio_master(),
479            Err(PlaybackClockError::AudioMasterAlreadyRegistered)
480        ));
481
482        playback.reset_for_seek();
483        audio.publish(2_000, 3_000, true).unwrap();
484        assert_eq!(playback.master(), PlaybackMaster::Audio);
485    }
486
487    #[test]
488    fn audio_projection_is_capped_at_submitted_media() {
489        let wall = Arc::new(Clock::new());
490        let playback = Arc::new(PlaybackClock::new(wall));
491        let audio = playback.register_audio_master().unwrap();
492        audio.publish(10, 1_000_000, true).unwrap();
493        thread::sleep(Duration::from_millis(5));
494        assert_eq!(playback.position_ns(), Some(1_000_000));
495    }
496
497    #[test]
498    fn finished_audio_continues_on_wall_and_can_reset_for_seek() {
499        let wall = Arc::new(Clock::new());
500        let playback = Arc::new(PlaybackClock::new(wall));
501        let audio = playback.register_audio_master().unwrap();
502        audio.publish(1_000, 2_000, false).unwrap();
503        audio.finish(2_000).unwrap();
504        assert_eq!(playback.master(), PlaybackMaster::Wall);
505        thread::sleep(Duration::from_millis(5));
506        assert!(playback.position_ns().unwrap() > 2_000);
507
508        audio.reset_for_seek().unwrap();
509        assert_eq!(playback.master(), PlaybackMaster::AudioPriming);
510        assert_eq!(playback.position_ns(), None);
511    }
512}
media_pp\core/playback_clock.rs

media_pp\core/
playback_clock.rs
