media_pp\elements\filter/
video_synchronizer.rs

1use std::{collections::VecDeque, sync::Arc, thread, time::Duration};
2
3use crate::pp_log::{PpLog, pp_debug, pp_info};
4use ffmpeg_next as ffmpeg;
5use thiserror::Error as ThisError;
6
7use crate::{
8    buffer::MediaBuffer,
9    control::ControlMsg,
10    element::{Element, ElementType, Sink, Source, element_pp_log},
11    pad::SrcPad,
12    playback_clock::{PlaybackClock, PlaybackMaster},
13    time::{InvalidTimeBase, MediaTimestamp, TimeBase},
14};
15
16const INTERRUPT_POLL_INTERVAL: Duration = Duration::from_millis(10);
17const FALLBACK_FRAME_DURATION: Duration = Duration::from_millis(40);
18
19fn nanoseconds() -> TimeBase {
20    TimeBase::new_unchecked(ffmpeg::Rational::new(1, 1_000_000_000))
21}
22
23#[derive(Debug, ThisError)]
24pub enum VideoSynchronizerError {
25    #[error(
26        "invalid time base {numerator}/{denominator}: both numerator and denominator must be positive"
27    )]
28    InvalidTimeBase { numerator: i32, denominator: i32 },
29
30    #[error("VideoSynchronizer only schedules decoded Video frames, got a {0}")]
31    UnsupportedBuffer(&'static str),
32
33    #[error("VideoSynchronizer cannot schedule a video frame without a PTS")]
34    MissingPts,
35}
36
37enum Decision {
38    Render,
39    Drop,
40    Wait(Duration),
41    Hold,
42}
43
44/// Schedules decoded video against the pipeline's current playback master.
45///
46/// In wall-master mode this replaces [`crate::elements::Pacer`]: the first
47/// video PTS establishes the media origin and early frames wait. Once an
48/// audio renderer registers and starts, the same instance automatically
49/// compares video PTS with the played-audio position, waiting for early
50/// frames and dropping frames more than one frame-duration late. During
51/// audio priming it holds the in-flight frame so the wall-to-audio handoff
52/// cannot make the picture run ahead.
53///
54/// Do not put a `Pacer` in the same video branch; that would pace twice.
55/// Put a [`crate::queue::Queue`] upstream so waits do not block demux/decode.
56pub struct VideoSynchronizer {
57    pp_log: PpLog,
58    name: Arc<str>,
59    time_base: TimeBase,
60    playback_clock: Arc<PlaybackClock>,
61    interrupt_epoch: u64,
62    last_pts: Option<i64>,
63    frame_duration: Duration,
64    pending: VecDeque<MediaBuffer>,
65    pad: SrcPad,
66}
67
68impl VideoSynchronizer {
69    pub fn new(
70        name: impl Into<String>,
71        time_base: ffmpeg::Rational,
72        playback_clock: Arc<PlaybackClock>,
73    ) -> Result<Self, VideoSynchronizerError> {
74        let name: Arc<str> = name.into().into();
75        let pp_log = element_pp_log(ElementType::VideoSynchronizer, &name, None);
76        let time_base = TimeBase::try_new(time_base).map_err(
77            |InvalidTimeBase {
78                 numerator,
79                 denominator,
80             }| VideoSynchronizerError::InvalidTimeBase {
81                numerator,
82                denominator,
83            },
84        )?;
85        let interrupt_epoch = playback_clock.interrupt_epoch();
86        pp_info!(pp_log: &pp_log, "created: time_base={time_base:?}");
87        Ok(Self {
88            name: name.clone(),
89            pp_log,
90            time_base,
91            playback_clock,
92            interrupt_epoch,
93            last_pts: None,
94            frame_duration: FALLBACK_FRAME_DURATION,
95            pending: VecDeque::new(),
96            pad: SrcPad::new(format!("{name}_src")),
97        })
98    }
99
100    fn timestamp_ns(&self, pts: i64) -> i64 {
101        MediaTimestamp::new_unchecked(pts, self.time_base).rescale(nanoseconds())
102    }
103
104    fn observe_frame_duration(&mut self, pts: i64) {
105        if let Some(delta) = self.last_pts.and_then(|last| pts.checked_sub(last))
106            && delta > 0
107        {
108            let duration = Duration::from_nanos(self.timestamp_ns(delta).max(0) as u64);
109            if !duration.is_zero() {
110                self.frame_duration = duration;
111            }
112        }
113        self.last_pts = Some(pts);
114    }
115
116    #[cfg(test)]
117    fn decision(&mut self, pts: i64) -> Decision {
118        self.observe_frame_duration(pts);
119        self.decision_without_observing(pts)
120    }
121
122    fn wait_for(&mut self, pts: i64) -> WaitOutcome {
123        self.observe_frame_duration(pts);
124        loop {
125            if self.playback_clock.interrupt_epoch() != self.interrupt_epoch {
126                return WaitOutcome::Interrupted;
127            }
128            match self.decision_without_observing(pts) {
129                Decision::Render => return WaitOutcome::Render,
130                Decision::Drop => return WaitOutcome::Drop,
131                Decision::Wait(wait) => thread::sleep(wait.min(INTERRUPT_POLL_INTERVAL)),
132                Decision::Hold => thread::sleep(INTERRUPT_POLL_INTERVAL),
133            }
134        }
135    }
136
137    fn decision_without_observing(&self, pts: i64) -> Decision {
138        let frame_ns = self.timestamp_ns(pts);
139        let (master, position) = self.playback_clock.video_snapshot(frame_ns);
140        match master {
141            PlaybackMaster::Unavailable => Decision::Render,
142            PlaybackMaster::AudioPriming => Decision::Hold,
143            PlaybackMaster::Wall => match position {
144                Some(position_ns) if frame_ns > position_ns => {
145                    Decision::Wait(ns_duration(frame_ns.saturating_sub(position_ns)))
146                }
147                _ => Decision::Render,
148            },
149            PlaybackMaster::Audio => {
150                let Some(position_ns) = position else {
151                    return Decision::Hold;
152                };
153                if frame_ns > position_ns {
154                    Decision::Wait(ns_duration(frame_ns.saturating_sub(position_ns)))
155                } else if position_ns.saturating_sub(frame_ns) > duration_ns(self.frame_duration) {
156                    Decision::Drop
157                } else {
158                    Decision::Render
159                }
160            }
161        }
162    }
163}
164
165enum WaitOutcome {
166    Render,
167    Drop,
168    Interrupted,
169}
170
171impl Element for VideoSynchronizer {
172    fn name(&self) -> Arc<str> {
173        self.name.clone()
174    }
175
176    fn element_type(&self) -> ElementType {
177        ElementType::VideoSynchronizer
178    }
179
180    fn pp_log(&self) -> &PpLog {
181        &self.pp_log
182    }
183
184    fn pp_log_mut(&mut self) -> &mut PpLog {
185        &mut self.pp_log
186    }
187}
188
189impl Source for VideoSynchronizer {
190    fn src_pads(&mut self) -> &mut [SrcPad] {
191        std::slice::from_mut(&mut self.pad)
192    }
193}
194
195impl Sink for VideoSynchronizer {
196    fn consume(&mut self, buf: MediaBuffer) -> crate::error::Result<()> {
197        match &buf {
198            MediaBuffer::Video(_) | MediaBuffer::Eos => {}
199            other => return Err(VideoSynchronizerError::UnsupportedBuffer(other.kind()).into()),
200        }
201
202        self.pending.push_back(buf);
203        while let Some(buf) = self.pending.pop_front() {
204            let outcome = match &buf {
205                MediaBuffer::Video(frame) => {
206                    let pts = frame.pts().ok_or(VideoSynchronizerError::MissingPts)?;
207                    self.wait_for(pts)
208                }
209                MediaBuffer::Eos => WaitOutcome::Render,
210                _ => unreachable!("buffer kind validated before queueing"),
211            };
212            match outcome {
213                WaitOutcome::Render => self.pad.push(buf)?,
214                WaitOutcome::Drop => pp_debug!(self, "dropping late video frame"),
215                WaitOutcome::Interrupted => {
216                    self.pending.push_front(buf);
217                    return Ok(());
218                }
219            }
220        }
221        Ok(())
222    }
223
224    fn control(&mut self, msg: ControlMsg) -> crate::error::Result<()> {
225        self.interrupt_epoch = self.playback_clock.interrupt_epoch();
226        match msg {
227            ControlMsg::Seek(_) | ControlMsg::Stop => {
228                self.pending.clear();
229                self.last_pts = None;
230                self.frame_duration = FALLBACK_FRAME_DURATION;
231            }
232            ControlMsg::Pause | ControlMsg::Resume => {}
233        }
234        self.pad.control(msg)
235    }
236}
237
238fn ns_duration(ns: i64) -> Duration {
239    Duration::from_nanos(ns.max(0) as u64)
240}
241
242fn duration_ns(duration: Duration) -> i64 {
243    duration.as_nanos().min(i64::MAX as u128) as i64
244}
245
246#[cfg(test)]
247mod tests {
248    use super::*;
249    use crate::{clock::Clock, playback_clock::PlaybackClock, pool::UnboundObjectPool};
250
251    fn synchronizer(clock: Arc<PlaybackClock>) -> VideoSynchronizer {
252        VideoSynchronizer::new("sync", ffmpeg::Rational::new(1, 1_000), clock).unwrap()
253    }
254
255    #[test]
256    fn first_video_timestamp_establishes_wall_origin() {
257        let playback = Arc::new(PlaybackClock::new(Arc::new(Clock::new())));
258        let mut sync = synchronizer(playback.clone());
259        assert!(matches!(sync.decision(5_000), Decision::Render));
260        assert_eq!(playback.master(), PlaybackMaster::Wall);
261        assert!(playback.position_ns().unwrap() >= 5_000_000_000);
262    }
263
264    #[test]
265    fn audio_priming_holds_video_and_audio_master_drops_late_frames() {
266        let playback = Arc::new(PlaybackClock::new(Arc::new(Clock::new())));
267        let mut sync = synchronizer(playback.clone());
268        let audio = playback.register_audio_master().unwrap();
269        assert!(matches!(sync.decision(1_000), Decision::Hold));
270
271        audio.publish(2_000_000_000, 3_000_000_000, false).unwrap();
272        assert!(matches!(sync.decision(1_000), Decision::Drop));
273        assert!(matches!(sync.decision(2_010), Decision::Wait(_)));
274    }
275
276    #[test]
277    fn invalid_time_base_and_non_video_input_are_typed_errors() {
278        let playback = Arc::new(PlaybackClock::new(Arc::new(Clock::new())));
279        assert!(matches!(
280            VideoSynchronizer::new("sync", ffmpeg::Rational::new(0, 1), playback.clone()),
281            Err(VideoSynchronizerError::InvalidTimeBase { .. })
282        ));
283        let mut sync = synchronizer(playback);
284        assert!(matches!(
285            sync.consume(MediaBuffer::Packet(Arc::new(ffmpeg::Packet::empty()))),
286            Err(crate::error::Error::VideoSynchronizerError(_))
287        ));
288
289        let pool = UnboundObjectPool::new(0, ffmpeg::frame::Video::empty, |_| {});
290        assert!(matches!(
291            sync.consume(MediaBuffer::Video(Arc::new(pool.get()))),
292            Err(crate::error::Error::VideoSynchronizerError(
293                VideoSynchronizerError::MissingPts
294            ))
295        ));
296    }
297}
media_pp\elements\filter/video_synchronizer.rs

media_pp\elements\filter/
video_synchronizer.rs
