clocksync: Add a new property "sync-to-first" for automatic ts-offset setup

Add a new property so that clocksync can setup "ts-offset" value
based on the first buffer and pipeline's running time when the
first arrived. Newly update "ts-offset" in this case would be
a value that allows outputting the first buffer without clock waiting.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/702>

4 files changed, 175 insertions, 8 deletions

diff --git a/docs/plugins/gst_plugins_cache.json b/docs/plugins/gst_plugins_cache.json
index 3c64ac2288..66f5bd1f6c 100644
--- a/docs/plugins/gst_plugins_cache.json
+++ b/docs/plugins/gst_plugins_cache.json
@@ -92,6 +92,18 @@
                         "type": "gboolean",
                         "writable": true
                     },
+                    "sync-to-first": {
+                        "blurb": "Automatically set ts-offset based on running time of the first buffer and pipeline's running time (i.e., ts-offset = \"pipeline running time\" - \"buffer running time\"). When enabled, clocksync element will update ts-offset on the first buffer per flush event or READY to PAUSED state change. This property can be useful in case that buffer timestamp does not necessarily have to be synchronized with pipeline's running time, but duration of the buffer through clocksync element needs to be synchronized with the amount of clock time go. Note that mixed use of ts-offset and this property would be racy if clocksync element is running already.",
+                        "conditionally-available": false,
+                        "construct": false,
+                        "construct-only": false,
+                        "controllable": false,
+                        "default": "false",
+                        "mutable": "null",
+                        "readable": true,
+                        "type": "gboolean",
+                        "writable": true
+                    },
                     "ts-offset": {
                         "blurb": "Timestamp offset in nanoseconds for synchronisation, negative for earlier sync",
                         "conditionally-available": false,
diff --git a/plugins/elements/gstclocksync.c b/plugins/elements/gstclocksync.c
index 4d5d3e2544..5b841db5f8 100644
--- a/plugins/elements/gstclocksync.c
+++ b/plugins/elements/gstclocksync.c
@@ -56,14 +56,19 @@ GST_DEBUG_CATEGORY_STATIC (gst_clock_sync_debug);
 /* ClockSync args */
 #define DEFAULT_SYNC                    TRUE
 #define DEFAULT_TS_OFFSET               0
+#define DEFAULT_SYNC_TO_FIRST           FALSE
 
 enum
 {
   PROP_0,
   PROP_SYNC,
-  PROP_TS_OFFSET
+  PROP_TS_OFFSET,
+  PROP_SYNC_TO_FIRST,
+  PROP_LAST
 };
 
+static GParamSpec *properties[PROP_LAST];
+
 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
     GST_PAD_SINK,
     GST_PAD_ALWAYS,
@@ -116,15 +121,42 @@ gst_clock_sync_class_init (GstClockSyncClass * klass)
   gobject_class->get_property = gst_clock_sync_get_property;
   gobject_class->finalize = gst_clock_sync_finalize;
 
-  g_object_class_install_property (gobject_class, PROP_SYNC,
+  properties[PROP_SYNC] =
       g_param_spec_boolean ("sync", "Synchronize",
-          "Synchronize to pipeline clock", DEFAULT_SYNC,
-          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
-  g_object_class_install_property (gobject_class, PROP_TS_OFFSET,
+      "Synchronize to pipeline clock", DEFAULT_SYNC,
+      G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
+  properties[PROP_TS_OFFSET] =
       g_param_spec_int64 ("ts-offset", "Timestamp offset for synchronisation",
-          "Timestamp offset in nanoseconds for synchronisation, negative for earlier sync",
-          G_MININT64, G_MAXINT64, DEFAULT_TS_OFFSET,
-          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+      "Timestamp offset in nanoseconds for synchronisation, negative for earlier sync",
+      G_MININT64, G_MAXINT64, DEFAULT_TS_OFFSET,
+      G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
+
+  /**
+   * GstClockSync:sync-to-first:
+   *
+   * When enabled, clocksync elemenet will adjust "ts-offset" value
+   * automatically by using given timestamp of the first buffer and running
+   * time of pipeline, so that clocksync element can output the first buffer
+   * immediately without clock waiting.
+   *
+   * Since: 1.20
+   */
+  properties[PROP_SYNC_TO_FIRST] =
+      g_param_spec_boolean ("sync-to-first",
+      "Sync to first",
+      "Automatically set ts-offset based on running time of the first "
+      "buffer and pipeline's running time "
+      "(i.e., ts-offset = \"pipeline running time\" - \"buffer running time\"). "
+      "When enabled, clocksync element will update ts-offset on the first "
+      "buffer per flush event or READY to PAUSED state change. "
+      "This property can be useful in case that buffer timestamp does not "
+      "necessarily have to be synchronized with pipeline's running time, "
+      "but duration of the buffer through clocksync element needs to be "
+      "synchronized with the amount of clock time go. "
+      "Note that mixed use of ts-offset and this property would be racy "
+      "if clocksync element is running already.",
+      DEFAULT_SYNC_TO_FIRST, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
+  g_object_class_install_properties (gobject_class, PROP_LAST, properties);
 
   gstelement_class->change_state =
       GST_DEBUG_FUNCPTR (gst_clocksync_change_state);
@@ -172,6 +204,7 @@ gst_clock_sync_init (GstClockSync * clocksync)
 
   clocksync->ts_offset = DEFAULT_TS_OFFSET;
   clocksync->sync = DEFAULT_SYNC;
+  clocksync->sync_to_first = DEFAULT_SYNC_TO_FIRST;
   g_cond_init (&clocksync->blocked_cond);
 
   GST_OBJECT_FLAG_SET (clocksync, GST_ELEMENT_FLAG_REQUIRE_CLOCK);
@@ -211,6 +244,9 @@ gst_clock_sync_set_property (GObject * object, guint prop_id,
     case PROP_TS_OFFSET:
       clocksync->ts_offset = g_value_get_int64 (value);
       break;
+    case PROP_SYNC_TO_FIRST:
+      clocksync->sync_to_first = g_value_get_boolean (value);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
@@ -230,6 +266,9 @@ gst_clock_sync_get_property (GObject * object, guint prop_id,
     case PROP_TS_OFFSET:
       g_value_set_int64 (value, clocksync->ts_offset);
       break;
+    case PROP_SYNC_TO_FIRST:
+      g_value_set_boolean (value, clocksync->sync_to_first);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
@@ -346,6 +385,8 @@ gst_clock_sync_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
       clocksync->flushing = FALSE;
       gst_segment_init (&clocksync->segment, GST_FORMAT_UNDEFINED);
       GST_OBJECT_UNLOCK (clocksync);
+      clocksync->is_first = TRUE;
+      break;
     default:
       break;
   }
@@ -355,6 +396,42 @@ gst_clock_sync_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
   return ret;
 }
 
+static void
+gst_clock_sync_update_ts_offset (GstClockSync * clocksync,
+    GstClockTime runtimestamp)
+{
+  GstClock *clock;
+  GstClockTimeDiff ts_offset = 0;
+  GstClockTime running_time;
+
+  if (!clocksync->sync_to_first || !clocksync->is_first || !clocksync->sync)
+    return;
+
+  GST_OBJECT_LOCK (clocksync);
+  clock = GST_ELEMENT_CLOCK (clocksync);
+  if (!clock) {
+    GST_DEBUG_OBJECT (clocksync, "We have no clock");
+    GST_OBJECT_UNLOCK (clocksync);
+    return;
+  }
+
+  running_time = gst_clock_get_time (clock) -
+      GST_ELEMENT_CAST (clocksync)->base_time;
+  ts_offset = GST_CLOCK_DIFF (runtimestamp, running_time);
+  GST_OBJECT_UNLOCK (clocksync);
+
+  GST_DEBUG_OBJECT (clocksync, "Running time %" GST_TIME_FORMAT
+      ", running time stamp %" GST_TIME_FORMAT ", calculated ts-offset %"
+      GST_STIME_FORMAT, GST_TIME_ARGS (running_time),
+      GST_TIME_ARGS (runtimestamp), GST_STIME_ARGS (ts_offset));
+
+  clocksync->is_first = FALSE;
+  if (ts_offset != clocksync->ts_offset) {
+    clocksync->ts_offset = ts_offset;
+    g_object_notify_by_pspec (G_OBJECT (clocksync), properties[PROP_TS_OFFSET]);
+  }
+}
+
 static GstFlowReturn
 gst_clock_sync_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
 {
@@ -388,6 +465,8 @@ gst_clock_sync_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
     else if (GST_CLOCK_TIME_IS_VALID (runpts))
       runtimestamp = runpts;
 
+    gst_clock_sync_update_ts_offset (clocksync, runtimestamp);
+
     ret = gst_clocksync_do_sync (clocksync, runtimestamp);
     if (ret != GST_FLOW_OK) {
       GST_LOG_OBJECT (clocksync,
@@ -431,6 +510,8 @@ gst_clock_sync_chain_list (GstPad * pad, GstObject * parent,
     else if (GST_CLOCK_TIME_IS_VALID (runpts))
       runtimestamp = runpts;
 
+    gst_clock_sync_update_ts_offset (clocksync, runtimestamp);
+
     ret = gst_clocksync_do_sync (clocksync, runtimestamp);
     if (ret != GST_FLOW_OK) {
       gst_buffer_list_unref (buffer_list);
@@ -512,6 +593,7 @@ gst_clocksync_change_state (GstElement * element, GstStateChange transition)
       GST_OBJECT_UNLOCK (clocksync);
       if (clocksync->sync)
         no_preroll = TRUE;
+      clocksync->is_first = TRUE;
       break;
     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
       GST_OBJECT_LOCK (clocksync);
diff --git a/plugins/elements/gstclocksync.h b/plugins/elements/gstclocksync.h
index 82d929e3e3..647d73e158 100644
--- a/plugins/elements/gstclocksync.h
+++ b/plugins/elements/gstclocksync.h
@@ -59,6 +59,8 @@ struct _GstClockSync
   GCond          blocked_cond;
   gboolean       blocked;
   GstClockTimeDiff  ts_offset;
+  gboolean sync_to_first;
+  gboolean is_first;
 
   GstClockTime   upstream_latency;
 };
diff --git a/tests/check/elements/clocksync.c b/tests/check/elements/clocksync.c
index 3069e72e63..df04b6b9f7 100644
--- a/tests/check/elements/clocksync.c
+++ b/tests/check/elements/clocksync.c
@@ -168,6 +168,76 @@ GST_START_TEST (test_stopping_element_unschedules_sync)
 
 GST_END_TEST;
 
+typedef struct
+{
+  guint notify_count;
+  GstClockTimeDiff ts_offset;
+} ClockSyncTestData;
+
+static void
+clock_sync_ts_offset_changed_cb (GstElement * clocksync, GParamSpec * pspec,
+    ClockSyncTestData * data)
+{
+  data->notify_count++;
+  g_object_get (clocksync, "ts-offset", &data->ts_offset, NULL);
+}
+
+GST_START_TEST (test_sync_to_first)
+{
+  /* the reason to use the queue in front of the clocksync element
+     is to effectively make gst_harness_push asynchronous, not locking
+     up the test, waiting for gst_clock_id_wait */
+  GstHarness *h =
+      gst_harness_new_parse ("queue ! clocksync sync-to-first=true");
+  GstBuffer *buf;
+  GstClock *clock;
+  GstClockTime timestamp = 123456789;
+  GstElement *clocksync;
+  ClockSyncTestData data;
+  data.notify_count = 0;
+  data.ts_offset = 0;
+
+  clocksync = gst_harness_find_element (h, "clocksync");
+  g_signal_connect (clocksync, "notify::ts-offset",
+      G_CALLBACK (clock_sync_ts_offset_changed_cb), &data);
+  gst_object_unref (clocksync);
+
+  /* use testclock */
+  gst_harness_use_testclock (h);
+  gst_harness_set_src_caps_str (h, "mycaps");
+
+  /* make a buffer and set the timestamp */
+  buf = gst_buffer_new ();
+  GST_BUFFER_PTS (buf) = timestamp;
+
+  /* push the buffer, and verify it does *not* make it through */
+  gst_harness_push (h, buf);
+  fail_unless_equals_int (0, gst_harness_buffers_in_queue (h));
+
+  /* verify the clocksync element has registered exactly one GstClockID */
+  fail_unless (gst_harness_wait_for_clock_id_waits (h, 1, 42));
+
+  /* crank the clock and pull the buffer */
+  gst_harness_crank_single_clock_wait (h);
+  buf = gst_harness_pull (h);
+
+  /* verify that the buffer has the right timestamp, and that the time on
+     the clock is equal to the timestamp */
+  fail_unless_equals_int64 (timestamp, GST_BUFFER_PTS (buf));
+  clock = gst_element_get_clock (h->element);
+  /* this buffer must be pushed without clock waiting */
+  fail_unless_equals_int64 (gst_clock_get_time (clock), 0);
+  fail_unless_equals_int (data.notify_count, 1);
+  fail_unless_equals_int64 (data.ts_offset, -timestamp);
+
+  /* cleanup */
+  gst_object_unref (clock);
+  gst_buffer_unref (buf);
+  gst_harness_teardown (h);
+}
+
+GST_END_TEST;
+
 static Suite *
 clocksync_suite (void)
 {
@@ -179,6 +249,7 @@ clocksync_suite (void)
   tcase_add_test (tc_chain, test_sync_on_timestamp);
   tcase_add_test (tc_chain, test_stopping_element_unschedules_sync);
   tcase_add_test (tc_chain, test_no_sync_on_timestamp);
+  tcase_add_test (tc_chain, test_sync_to_first);
 
 
   return s;