From 1119ed66202fcef3cf11d02ff6ce97fcb99c5aa4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Sebastian=20Dr=C3=B6ge?= <sebastian@centricular.com>
Date: Tue, 20 Jun 2023 10:36:51 +0300
Subject: [PATCH] livesync: Wait for the end timestamp of the previous buffer
 before looking at queue

Previously livesync was waiting for the start timestamp of the current
buffer after looking at the queue and right before pushing it
downstream. This meant that it generally looked too early at the queue
and especially that upstream had to provide the next buffer already at
the start timestamp of the previous one.

Instead, now wait before looking at the queue and wait for the end
timestamp of the previous buffer. Once the previous buffer has expired,
a new buffer really needs to be available or otherwise a filler buffer
has to be pushed downstream.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1250>
---
 utils/livesync/src/livesync/imp.rs | 70 +++++++++++++++---------------
 1 file changed, 36 insertions(+), 34 deletions(-)

diff --git a/utils/livesync/src/livesync/imp.rs b/utils/livesync/src/livesync/imp.rs
index 4d4e6a09..929e86b8 100644
--- a/utils/livesync/src/livesync/imp.rs
+++ b/utils/livesync/src/livesync/imp.rs
@@ -1028,6 +1028,42 @@ impl LiveSync {
         }
         state.srcresult?;
 
+        if let Some(out_timestamp) = state.out_timestamp {
+            let sync_ts = out_timestamp.end;
+
+            let element = self.obj();
+
+            let base_time = element.base_time().ok_or_else(|| {
+                gst::error!(CAT, imp: self, "Missing base time");
+                gst::FlowError::Flushing
+            })?;
+
+            let clock = element.clock().ok_or_else(|| {
+                gst::error!(CAT, imp: self, "Missing clock");
+                gst::FlowError::Flushing
+            })?;
+
+            let clock_id = clock.new_single_shot_id(base_time + sync_ts);
+            state.clock_id = Some(clock_id.clone());
+
+            gst::trace!(
+                CAT,
+                imp: self,
+                "Waiting for clock to reach {}",
+                clock_id.time(),
+            );
+
+            let (res, _) = MutexGuard::unlocked(&mut state, || clock_id.wait());
+            gst::trace!(CAT, imp: self, "Clock returned {res:?}",);
+
+            if res == Err(gst::ClockError::Unscheduled) {
+                return Err(gst::FlowError::Flushing);
+            }
+
+            state.srcresult?;
+            state.clock_id = None;
+        }
+
         let in_item = state.queue.pop_front();
         gst::trace!(CAT, imp: self, "Unqueueing {:?}", in_item);
 
@@ -1184,40 +1220,6 @@ impl LiveSync {
             state.out_segment = Some(segment);
         }
 
-        {
-            let element = self.obj();
-
-            let base_time = element.base_time().ok_or_else(|| {
-                gst::error!(CAT, imp: self, "Missing base time");
-                gst::FlowError::Flushing
-            })?;
-
-            let clock = element.clock().ok_or_else(|| {
-                gst::error!(CAT, imp: self, "Missing clock");
-                gst::FlowError::Flushing
-            })?;
-
-            let clock_id = clock.new_single_shot_id(base_time + sync_ts);
-            state.clock_id = Some(clock_id.clone());
-
-            gst::trace!(
-                CAT,
-                imp: self,
-                "Waiting for clock to reach {}",
-                clock_id.time(),
-            );
-
-            let (res, _) = MutexGuard::unlocked(&mut state, || clock_id.wait());
-            gst::trace!(CAT, imp: self, "Clock returned {res:?}",);
-
-            if res == Err(gst::ClockError::Unscheduled) {
-                return Err(gst::FlowError::Flushing);
-            }
-
-            state.srcresult?;
-            state.clock_id = None;
-        }
-
         state.num_out += 1;
         if duplicate {
             state.num_duplicate += 1;