Merge branch 'sklug/upstream/fix-stream-synchronization-v1' into 'master'

RFC: Improve/fix timestamping issues with gstpipewiresrc

See merge request pipewire/pipewire!2786

spa/plugins/libcamera/libcamera-source.cpp

@@ -1060,20 +1060,12 @@ void handle_completed_request(struct impl *impl, libcamera::Request *request)
 	const FrameMetadata &fmd = buffer->metadata();
 
 	if (impl->clock) {
-		double target = (double)port->info.rate.num / port->info.rate.denom;
+		/*
-		double corr;
+		 * The clock has SPA_IO_CLOCK_FLAG_NO_RATE set, so there is no
-
+		 * need to update any rate specific fields.
-		if (impl->dll.bw == 0.0) {
+		 * As libcamera uses CLOCK_MONOTONIC internally, there is no
-			spa_dll_set_bw(&impl->dll, SPA_DLL_BW_MAX, port->info.rate.denom, port->info.rate.denom);
+		 * need to adjust the timestamps.
-			impl->clock->next_nsec = fmd.timestamp;
+		 */
-			corr = 1.0;
-		} else {
-			double diff = ((double)impl->clock->next_nsec - (double)fmd.timestamp) / SPA_NSEC_PER_SEC;
-			double error = port->info.rate.denom * (diff - target);
-			corr = spa_dll_update(&impl->dll, SPA_CLAMPD(error, -128., 128.));
-		}
-		/* FIXME, we should follow the driver clock and target_ values.
-		 * for now we ignore and use our own. */
 		impl->clock->target_rate = port->rate;
 		impl->clock->target_duration = 1;
 
@@ -1082,8 +1074,8 @@ void handle_completed_request(struct impl *impl, libcamera::Request *request)
 		impl->clock->position = fmd.sequence;
 		impl->clock->duration = 1;
 		impl->clock->delay = 0;
-		impl->clock->rate_diff = corr;
+		impl->clock->rate_diff = 1.0;
-		impl->clock->next_nsec += (uint64_t) (target * SPA_NSEC_PER_SEC * corr);
+		impl->clock->next_nsec = fmd.timestamp+1;
 	}
 
 	if (b->h) {

src/gst/gstpipewireclock.c

@@ -22,6 +22,7 @@ gst_pipewire_clock_new (GstPipeWireStream *stream, GstClockTime last_time)
   g_weak_ref_set (&clock->stream, stream);
   clock->last_time = last_time;
   clock->time_offset = last_time;
+  clock->start_time_valid = false;
 
   return GST_CLOCK_CAST (clock);
 }
@@ -38,19 +39,35 @@ gst_pipewire_clock_get_internal_time (GstClock * clock)
     return pclock->last_time;
 
   now = pw_stream_get_nsec(s->pwstream);
+
 #if 1
   struct pw_time t;
   if (s->pwstream == NULL ||
       pw_stream_get_time_n (s->pwstream, &t, sizeof(t)) < 0 ||
-      t.rate.denom == 0)
+      t.rate.denom == 0) {
+        pclock->start_time_valid = false;
+        if (now < pclock->last_time)
           return pclock->last_time;
+        return now;
+  }
 
-  result = gst_util_uint64_scale (t.ticks, GST_SECOND * t.rate.num, t.rate.denom);
+  uint64_t elapsed = gst_util_uint64_scale (t.ticks, GST_SECOND * t.rate.num, t.rate.denom);
+
+  if (!pclock->start_time_valid) {
+    pclock->start_time = t.now - elapsed;
+    pclock->start_time_valid = true;
+  }
+
+  result = pclock->start_time + elapsed;
   result += now - t.now;
 
   result += pclock->time_offset;
   pclock->last_time = result;
 
+  if ( ABS(GST_CLOCK_DIFF(now, result)) > GST_SECOND * 5 ) {
+    GST_ERROR ("clock: %p Large timedrift detected. Something is wrong.", pclock);
+  }
+
   GST_DEBUG ("%"PRId64", %d/%d %"PRId64" %"PRId64" %"PRId64,
                 t.ticks, t.rate.num, t.rate.denom, t.now, result, now);
 #else

src/gst/gstpipewireclock.h

@@ -22,6 +22,8 @@ struct _GstPipeWireClock {
   GWeakRef stream;
 
   GstClockTime last_time;
+  GstClockTime start_time;
+  bool start_time_valid;
   GstClockTimeDiff time_offset;
 };
 

src/gst/gstpipewiresrc.c

@@ -116,8 +116,6 @@ static gboolean gst_pipewire_src_start (GstBaseSrc * basesrc);
 static gboolean gst_pipewire_src_stop (GstBaseSrc * basesrc);
 static gboolean gst_pipewire_src_event (GstBaseSrc * src, GstEvent * event);
 static gboolean gst_pipewire_src_query (GstBaseSrc * src, GstQuery * query);
-static void gst_pipewire_src_get_times (GstBaseSrc * basesrc, GstBuffer * buffer,
-    GstClockTime * start, GstClockTime * end);
 
 static void
 gst_pipewire_src_set_property (GObject * object, guint prop_id,
@@ -501,7 +499,6 @@ gst_pipewire_src_class_init (GstPipeWireSrcClass * klass)
   gstbasesrc_class->stop = gst_pipewire_src_stop;
   gstbasesrc_class->event = gst_pipewire_src_event;
   gstbasesrc_class->query = gst_pipewire_src_query;
-  gstbasesrc_class->get_times = gst_pipewire_src_get_times;
   gstpushsrc_class->create = gst_pipewire_src_create;
 
   GST_DEBUG_CATEGORY_INIT (pipewire_src_debug, "pipewiresrc", 0,
@@ -531,7 +528,6 @@ gst_pipewire_src_init (GstPipeWireSrc * src)
   src->autoconnect = DEFAULT_AUTOCONNECT;
   src->min_latency = 0;
   src->max_latency = GST_CLOCK_TIME_NONE;
-  src->last_buffer_clock_time = GST_CLOCK_TIME_NONE;
   src->n_buffers = 0;
   src->flushing_on_remove_buffer = FALSE;
   src->on_disconnect = DEFAULT_ON_DISCONNECT;
@@ -1523,40 +1519,13 @@ gst_pipewire_src_query (GstBaseSrc * src, GstQuery * query)
   return res;
 }
 
-static void
-gst_pipewire_src_get_times (GstBaseSrc * basesrc, GstBuffer * buffer,
-    GstClockTime * start, GstClockTime * end)
-{
-  GstPipeWireSrc *pwsrc = GST_PIPEWIRE_SRC (basesrc);
-
-  /* for live sources, sync on the timestamp of the buffer */
-  if (gst_base_src_is_live (basesrc)) {
-    GstClockTime timestamp = GST_BUFFER_PTS (buffer);
-
-    if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
-      /* get duration to calculate end time */
-      GstClockTime duration = GST_BUFFER_DURATION (buffer);
-
-      if (GST_CLOCK_TIME_IS_VALID (duration)) {
-        *end = timestamp + duration;
-      }
-      *start = timestamp;
-    }
-  } else {
-    *start = GST_CLOCK_TIME_NONE;
-    *end = GST_CLOCK_TIME_NONE;
-  }
-
-  GST_LOG_OBJECT (pwsrc, "start %" GST_TIME_FORMAT " (%" G_GUINT64_FORMAT
-      "), end %" GST_TIME_FORMAT " (%" G_GUINT64_FORMAT ")",
-      GST_TIME_ARGS (*start), *start, GST_TIME_ARGS (*end), *end);
-}
-
 static GstFlowReturn
 gst_pipewire_src_create (GstPushSrc * psrc, GstBuffer ** buffer)
 {
   GstPipeWireSrc *pwsrc;
+  GstClockTime pts, dts, base_time;
   const char *error = NULL;
+  GstClock *clock;
   GstBuffer *buf;
   gboolean update_time = FALSE, timeout = FALSE;
   GstCaps *caps = NULL;
@@ -1615,21 +1584,12 @@ gst_pipewire_src_create (GstPushSrc * psrc, GstBuffer ** buffer)
       GST_LOG_OBJECT (pwsrc, "popped buffer %p", buf);
       if (buf != NULL) {
         if (pwsrc->resend_last || pwsrc->keepalive_time > 0) {
-          GstClock *clock;
           GstBuffer *old;
 
           old = pwsrc->last_buffer;
           pwsrc->last_buffer = gst_buffer_copy (buf);
           gst_buffer_unref (old);
           gst_buffer_add_parent_buffer_meta (pwsrc->last_buffer, buf);
-
-          clock = gst_element_get_clock (GST_ELEMENT_CAST (pwsrc));
-          if (clock != NULL) {
-            pwsrc->last_buffer_clock_time = gst_clock_get_time (clock);
-            gst_object_unref (clock);
-          } else {
-            pwsrc->last_buffer_clock_time = GST_CLOCK_TIME_NONE;
-          }
         }
         break;
       }
@@ -1652,37 +1612,51 @@ gst_pipewire_src_create (GstPushSrc * psrc, GstBuffer ** buffer)
   pw_thread_loop_unlock (pwsrc->stream->core->loop);
 
   *buffer = buf;
+  clock = gst_element_get_clock (GST_ELEMENT_CAST (pwsrc));
 
   if (update_time) {
-    GstClock *clock;
-    GstClockTime current_clock_time;
-
-    clock = gst_element_get_clock (GST_ELEMENT_CAST (pwsrc));
     if (clock != NULL) {
-      current_clock_time = gst_clock_get_time (clock);
+      pts = dts = gst_clock_get_time (clock);
       gst_object_unref (clock);
     } else {
-      current_clock_time = GST_CLOCK_TIME_NONE;
+      pts = dts = GST_CLOCK_TIME_NONE;
     }
 
-    if (GST_CLOCK_TIME_IS_VALID (current_clock_time) &&
+    GST_LOG_OBJECT (pwsrc, "Sending keepalive buffer");
-        GST_CLOCK_TIME_IS_VALID (pwsrc->last_buffer_clock_time) &&
-        GST_CLOCK_TIME_IS_VALID (GST_BUFFER_PTS (*buffer)) &&
-        GST_CLOCK_TIME_IS_VALID (GST_BUFFER_DTS (*buffer))) {
-      GstClockTime diff;
-
-      diff = current_clock_time - pwsrc->last_buffer_clock_time;
-
-      GST_BUFFER_PTS (*buffer) += diff;
-      GST_BUFFER_DTS (*buffer) += diff;
   } else {
-      GST_BUFFER_PTS (*buffer) = GST_BUFFER_DTS (*buffer) = current_clock_time;
+    pts = GST_BUFFER_PTS (*buffer);
+    dts = GST_BUFFER_DTS (*buffer);
   }
 
-    GST_LOG_OBJECT (pwsrc, "Sending keepalive buffer pts/dts: %" GST_TIME_FORMAT
+  /*
-      " (%" G_GUINT64_FORMAT ")", GST_TIME_ARGS (current_clock_time),
+   * We need to map the pipwire time to gstreamer time. If the gstreamer clock
-      current_clock_time);
+   * is provided by us, we can safely use the base_time of the element.
+   * Otherwise we can not assume that the gstreamer clock is CLOCK_MONOTONIC and
+   * must therefore fall back to our own base_time. This might introduce a bit
+   * of jitter.
+   */
+  base_time = 0;
+  if (pwsrc->is_live) {
+    if (clock == pwsrc->stream->clock) {
+      base_time = gst_element_get_base_time (GST_ELEMENT_CAST (pwsrc));
+    } else {
+      base_time = pwsrc->pw_base_time;
     }
+  }
+
+  if (GST_CLOCK_TIME_IS_VALID (pts))
+    pts = (pts >= base_time ? pts - base_time : 0);
+  if (GST_CLOCK_TIME_IS_VALID (dts))
+    dts = (dts >= base_time ? dts - base_time : 0);
+
+  GST_LOG_OBJECT (pwsrc,
+      "pts %" G_GUINT64_FORMAT ", dts %" G_GUINT64_FORMAT
+      ", base-time %" GST_TIME_FORMAT " -> %" GST_TIME_FORMAT ", %" GST_TIME_FORMAT,
+      GST_BUFFER_PTS (*buffer), GST_BUFFER_DTS (*buffer), GST_TIME_ARGS (base_time),
+      GST_TIME_ARGS (pts), GST_TIME_ARGS (dts));
+
+  GST_BUFFER_PTS (*buffer) = pts;
+  GST_BUFFER_DTS (*buffer) = dts;
 
   return GST_FLOW_OK;
 
@@ -1781,6 +1755,7 @@ gst_pipewire_src_change_state (GstElement * element, GstStateChange transition)
       GST_DEBUG_OBJECT (this, "activating stream");
 
       pw_thread_loop_lock (this->stream->core->loop);
+      this->pw_base_time = pw_stream_get_nsec (this->stream->pwstream);
       pw_stream_set_active (this->stream->pwstream, true);
       /* if state have been paused for longer time, the underlying node might
        * be moved from idle to suspended, which would mean format cleared via

src/gst/gstpipewiresrc.h

@@ -79,11 +79,11 @@ struct _GstPipeWireSrc {
 
   gboolean is_live;
   int64_t delay;
+  uint64_t pw_base_time;
   GstClockTime min_latency;
   GstClockTime max_latency;
 
   GstBuffer *last_buffer;
-  GstClockTime last_buffer_clock_time;
 
   enum spa_meta_videotransform_value transform_value;