libs/gst/base/gstbasesink.c: Don't try to report a 0 position when we don't know, return -1 and FALSE instead. This m...

Original commit message from CVS:
* libs/gst/base/gstbasesink.c: (gst_base_sink_get_sync_times),
(gst_base_sink_do_sync), (gst_base_sink_preroll_object),
(gst_base_sink_event), (gst_base_sink_get_position_last),
(gst_base_sink_get_position_paused), (gst_base_sink_get_position),
(gst_base_sink_change_state):
Don't try to report a 0 position when we don't know, return -1 and FALSE
instead. This mostly happens when we are prerolling.
Make sure we can report the right position before we post the ASYNC_DONE
message so that a message handler can query position without races.
* tests/check/generic/sinks.c: (send_eos), (GST_START_TEST),
(async_done_handoff), (async_done_func), (send_buffer),
(async_done_eos_func), (gst_sinks_suite):
Add two tests for the above.

ChangeLog

@@ -1,3 +1,20 @@
+2007-11-06  Wim Taymans  <wim.taymans@gmail.com>
+
+	* libs/gst/base/gstbasesink.c: (gst_base_sink_get_sync_times),
+	(gst_base_sink_do_sync), (gst_base_sink_preroll_object),
+	(gst_base_sink_event), (gst_base_sink_get_position_last),
+	(gst_base_sink_get_position_paused), (gst_base_sink_get_position),
+	(gst_base_sink_change_state):
+	Don't try to report a 0 position when we don't know, return -1 and FALSE
+	instead. This mostly happens when we are prerolling.
+	Make sure we can report the right position before we post the ASYNC_DONE
+	message so that a message handler can query position without races.
+
+	* tests/check/generic/sinks.c: (send_eos), (GST_START_TEST),
+	(async_done_handoff), (async_done_func), (send_buffer),
+	(async_done_eos_func), (gst_sinks_suite):
+	Add two tests for the above.
+
 2007-11-06  Wim Taymans  <wim.taymans@gmail.com>
 
 	* MAINTAINERS:

libs/gst/base/gstbasesink.c

@@ -1274,7 +1274,8 @@ async_failed:
 static gboolean
 gst_base_sink_get_sync_times (GstBaseSink * basesink, GstMiniObject * obj,
     GstClockTime * rsstart, GstClockTime * rsstop,
-    GstClockTime * rrstart, GstClockTime * rrstop, gboolean * do_sync)
+    GstClockTime * rrstart, GstClockTime * rrstop, gboolean * do_sync,
+    GstSegment * segment)
 {
   GstBaseSinkClass *bclass;
   GstBuffer *buffer;
@@ -1283,7 +1284,9 @@ gst_base_sink_get_sync_times (GstBaseSink * basesink, GstMiniObject * obj,
   gint64 rstart, rstop;         /* clipped timestamps converted to running time */
   GstClockTime sstart, sstop;   /* clipped timestamps converted to stream time */
   GstFormat format;
-  GstSegment *segment;
+  GstBaseSinkPrivate *priv;
+
+  priv = basesink->priv;
 
   /* start with nothing */
   start = stop = sstart = sstop = rstart = rstop = -1;
@@ -1294,8 +1297,8 @@ gst_base_sink_get_sync_times (GstBaseSink * basesink, GstMiniObject * obj,
     switch (GST_EVENT_TYPE (event)) {
         /* EOS event needs syncing */
       case GST_EVENT_EOS:
-        sstart = sstop = basesink->priv->current_sstop;
+        sstart = sstop = priv->current_sstop;
-        rstart = rstop = basesink->priv->eos_rtime;
+        rstart = rstop = priv->eos_rtime;
         *do_sync = rstart != -1;
         GST_DEBUG_OBJECT (basesink, "sync times for EOS %" GST_TIME_FORMAT,
             GST_TIME_ARGS (rstart));
@@ -1332,7 +1335,6 @@ gst_base_sink_get_sync_times (GstBaseSink * basesink, GstMiniObject * obj,
       GST_TIME_ARGS (stop), *do_sync);
 
   /* collect segment and format for code clarity */
-  segment = &basesink->segment;
   format = segment->format;
 
   /* no timestamp clipping if we did not * get a TIME segment format */
@@ -1605,17 +1607,20 @@ gst_base_sink_do_sync (GstBaseSink * basesink, GstPad * pad,
   GstClockTimeDiff jitter;
   gboolean syncable;
   GstClockReturn status = GST_CLOCK_OK;
-  GstClockTime sstart, sstop, rstart, rstop;
+  GstClockTime rstart, rstop, sstart, sstop;
   gboolean do_sync;
+  GstBaseSinkPrivate *priv;
+
+  priv = basesink->priv;
 
   sstart = sstop = rstart = rstop = -1;
   do_sync = TRUE;
 
-  basesink->priv->current_rstart = -1;
+  priv->current_rstart = -1;
 
-  /* update timing information for this object */
+  /* get timing information for this object against the render segment */
   syncable = gst_base_sink_get_sync_times (basesink, obj,
-      &sstart, &sstop, &rstart, &rstop, &do_sync);
+      &sstart, &sstop, &rstart, &rstop, &do_sync, &basesink->segment);
 
   /* a syncable object needs to participate in preroll and
    * clocking. All buffers and EOS are syncable. */
@@ -1623,17 +1628,10 @@ gst_base_sink_do_sync (GstBaseSink * basesink, GstPad * pad,
     goto not_syncable;
 
   /* store timing info for current object */
-  basesink->priv->current_rstart = rstart;
+  priv->current_rstart = rstart;
-  basesink->priv->current_rstop = (rstop != -1 ? rstop : rstart);
+  priv->current_rstop = (rstop != -1 ? rstop : rstart);
   /* save sync time for eos when the previous object needed sync */
-  basesink->priv->eos_rtime = (do_sync ? basesink->priv->current_rstop : -1);
+  priv->eos_rtime = (do_sync ? priv->current_rstop : -1);
-
-  /* lock because we read this when answering the POSITION 
-   * query. */
-  GST_OBJECT_LOCK (basesink);
-  basesink->priv->current_sstart = sstart;
-  basesink->priv->current_sstop = (sstop != -1 ? sstop : sstart);
-  GST_OBJECT_UNLOCK (basesink);
 
 again:
   /* first do preroll, this makes sure we commit our state
@@ -1658,6 +1656,13 @@ again:
     }
   }
 
+  /* After rendering we store the position of the last buffer so that we can use
+   * it to report the position. We need to take the lock here. */
+  GST_OBJECT_LOCK (basesink);
+  priv->current_sstart = sstart;
+  priv->current_sstop = (sstop != -1 ? sstop : sstart);
+  GST_OBJECT_UNLOCK (basesink);
+
   if (!do_sync)
     goto done;
 
@@ -1687,7 +1692,7 @@ again:
   }
 
   /* successful syncing done, record observation */
-  basesink->priv->current_jitter = jitter;
+  priv->current_jitter = jitter;
 
   /* check if the object should be dropped */
   *late = gst_base_sink_is_too_late (basesink, obj, rstart, rstop,
@@ -2111,11 +2116,13 @@ gst_base_sink_preroll_object (GstBaseSink * basesink, GstPad * pad,
   if (G_LIKELY (GST_IS_BUFFER (obj))) {
     GstBaseSinkClass *bclass;
     GstBuffer *buf;
+    GstClockTime timestamp;
 
     buf = GST_BUFFER_CAST (obj);
+    timestamp = GST_BUFFER_TIMESTAMP (buf);
 
     GST_DEBUG_OBJECT (basesink, "preroll buffer %" GST_TIME_FORMAT,
-        GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)));
+        GST_TIME_ARGS (timestamp));
 
     gst_base_sink_set_last_buffer (basesink, buf);
 
@@ -2394,6 +2401,7 @@ gst_base_sink_event (GstPad * pad, GstEvent * event)
       gst_segment_init (basesink->abidata.ABI.clip_segment,
           GST_FORMAT_UNDEFINED);
       basesink->have_newsegment = FALSE;
+
       /* for position reporting */
       GST_OBJECT_LOCK (basesink);
       basesink->priv->current_sstart = -1;
@@ -2965,16 +2973,19 @@ gst_base_sink_get_position_last (GstBaseSink * basesink, gint64 * cur)
 {
   /* return last observed stream time */
   *cur = basesink->priv->current_sstop;
+  GST_DEBUG_OBJECT (basesink, "POSITION: %" GST_TIME_FORMAT,
+      GST_TIME_ARGS (*cur));
   return TRUE;
 }
 
-/* get the position when we are PAUSED */
+/* get the position when we are PAUSED, this is the stream time of the buffer
-/* FIXME, not entirely correct if we have preroll_queue_len > 1 and
+ * that prerolled. If no buffer is prerolled (we are still flushing), this
- * there are multiple segments in the queue since we calculate on the
+ * value will be -1. */
- * total segments, not the first one. */
 static gboolean
 gst_base_sink_get_position_paused (GstBaseSink * basesink, gint64 * cur)
 {
+  gboolean res;
+
   *cur = basesink->priv->current_sstart;
 
   if (*cur != -1)
@@ -2982,7 +2993,11 @@ gst_base_sink_get_position_paused (GstBaseSink * basesink, gint64 * cur)
   else
     *cur = basesink->abidata.ABI.clip_segment->time;
 
-  return TRUE;
+  res = (*cur != -1);
+  GST_DEBUG_OBJECT (basesink, "POSITION: %" GST_TIME_FORMAT,
+      GST_TIME_ARGS (*cur));
+
+  return res;
 }
 
 static gboolean
@@ -3094,12 +3109,14 @@ done:
   /* special cases */
 in_eos:
   {
+    GST_DEBUG_OBJECT (basesink, "position in EOS");
     res = gst_base_sink_get_position_last (basesink, cur);
     GST_OBJECT_UNLOCK (basesink);
     goto done;
   }
 in_pause:
   {
+    GST_DEBUG_OBJECT (basesink, "position in PAUSED");
     res = gst_base_sink_get_position_paused (basesink, cur);
     GST_OBJECT_UNLOCK (basesink);
     goto done;
@@ -3107,8 +3124,9 @@ in_pause:
 wrong_state:
   {
     /* in NULL or READY we always return 0 */
-    res = TRUE;
+    GST_DEBUG_OBJECT (basesink, "position in wrong state, return -1");
-    *cur = 0;
+    res = FALSE;
+    *cur = -1;
     GST_OBJECT_UNLOCK (basesink);
     goto done;
   }
@@ -3118,6 +3136,8 @@ no_sync:
      * that upstream can answer */
     if ((*cur = basesink->priv->current_sstart) != -1)
       res = TRUE;
+    GST_DEBUG_OBJECT (basesink, "no sync, res %d, POSITION %" GST_TIME_FORMAT,
+        res, GST_TIME_ARGS (*cur));
     GST_OBJECT_UNLOCK (basesink);
     return res;
   }
@@ -3218,8 +3238,8 @@ gst_base_sink_change_state (GstElement * element, GstStateChange transition)
       basesink->have_preroll = FALSE;
       basesink->need_preroll = TRUE;
       basesink->playing_async = TRUE;
-      priv->current_sstart = 0;
+      priv->current_sstart = -1;
-      priv->current_sstop = 0;
+      priv->current_sstop = -1;
       priv->eos_rtime = -1;
       priv->latency = 0;
       basesink->eos = FALSE;
@@ -3362,8 +3382,8 @@ gst_base_sink_change_state (GstElement * element, GstStateChange transition)
       } else {
         GST_DEBUG_OBJECT (basesink, "PAUSED to READY, don't need_preroll");
       }
-      priv->current_sstart = 0;
+      priv->current_sstart = -1;
-      priv->current_sstop = 0;
+      priv->current_sstop = -1;
       priv->have_latency = FALSE;
       gst_base_sink_set_last_buffer (basesink, NULL);
       GST_PAD_PREROLL_UNLOCK (basesink->sinkpad);

tests/check/generic/sinks.c

@@ -975,6 +975,269 @@ GST_START_TEST (test_fake_eos)
 
 GST_END_TEST;
 
+/* this variable is updated in the same thread, first it is set by the
+ * handoff-preroll signal, then it is checked when the ASYNC_DONE is posted on
+ * the bus */
+static gboolean have_preroll = FALSE;
+
+static void
+async_done_handoff (GstElement * element, GstBuffer * buf, GstPad * pad,
+    GstElement * sink)
+{
+  GST_DEBUG ("we have the preroll buffer");
+  have_preroll = TRUE;
+}
+
+/* when we get the ASYNC_DONE, query the position */
+static GstBusSyncReply
+async_done_func (GstBus * bus, GstMessage * msg, GstElement * sink)
+{
+  if (GST_MESSAGE_TYPE (msg) == GST_MESSAGE_ASYNC_DONE) {
+    GstFormat format;
+    gint64 position;
+
+    GST_DEBUG ("we have ASYNC_DONE now");
+    fail_unless (have_preroll == TRUE, "no preroll buffer received");
+
+    /* get the position now */
+    format = GST_FORMAT_TIME;
+    gst_element_query_position (sink, &format, &position);
+
+    GST_DEBUG ("we have position %" GST_TIME_FORMAT, GST_TIME_ARGS (position));
+
+    fail_unless (position == 10 * GST_SECOND, "position is wrong");
+  }
+
+  /* we can drop the message, nothing is listening for it. */
+  return GST_BUS_DROP;
+}
+
+static void
+send_buffer (GstPad * sinkpad)
+{
+  GstBuffer *buffer;
+  GstStateChangeReturn ret;
+
+  /* push a second buffer */
+  GST_DEBUG ("pushing last buffer");
+  buffer = gst_buffer_new_and_alloc (10);
+  GST_BUFFER_TIMESTAMP (buffer) = 200 * GST_SECOND;
+  GST_BUFFER_DURATION (buffer) = 100 * GST_SECOND;
+
+  /* this function will initially block */
+  ret = gst_pad_chain (sinkpad, buffer);
+  fail_unless (ret == GST_FLOW_OK, "no OK flow return");
+}
+
+/* when we get the ASYNC_DONE message from a sink, we want the sink to be able
+ * to report the duration and position. The sink should also have called the
+ * render method. */
+GST_START_TEST (test_async_done)
+{
+  GstElement *sink;
+  GstBuffer *buffer;
+  GstEvent *event;
+  GstStateChangeReturn ret;
+  GstPad *sinkpad;
+  GstFlowReturn res;
+  GstBus *bus;
+  GThread *thread;
+  GstFormat format;
+  gint64 position;
+  gboolean qret;
+
+  sink = gst_element_factory_make ("fakesink", "sink");
+  g_object_set (G_OBJECT (sink), "sync", TRUE, NULL);
+  g_object_set (G_OBJECT (sink), "preroll-queue-len", 2, NULL);
+  g_object_set (G_OBJECT (sink), "signal-handoffs", TRUE, NULL);
+
+  g_signal_connect (sink, "preroll-handoff", (GCallback) async_done_handoff,
+      sink);
+
+  sinkpad = gst_element_get_pad (sink, "sink");
+
+  ret = gst_element_set_state (sink, GST_STATE_PAUSED);
+  fail_unless (ret == GST_STATE_CHANGE_ASYNC, "no ASYNC state return");
+
+  /* set bus on element synchronously listen for ASYNC_DONE */
+  bus = gst_bus_new ();
+  gst_element_set_bus (sink, bus);
+  gst_bus_set_sync_handler (bus, (GstBusSyncHandler) async_done_func, sink);
+
+  /* make newsegment, this sets the position to 10sec when the buffer prerolls */
+  GST_DEBUG ("sending segment");
+  event =
+      gst_event_new_new_segment (FALSE, 1.0, GST_FORMAT_TIME, 0, -1,
+      10 * GST_SECOND);
+  res = gst_pad_send_event (sinkpad, event);
+
+  /* We have not yet received any buffers so we are still in the READY state,
+   * the position is therefore still not queryable. */
+  format = GST_FORMAT_TIME;
+  position = -1;
+  qret = gst_element_query_position (sink, &format, &position);
+  fail_unless (qret == FALSE, "position wrong");
+  fail_unless (position == -1, "position is wrong");
+
+  /* Since we are paused and the preroll queue has a length of 2, this function
+   * will return immediatly, the preroll handoff will be called and the stream
+   * position should now be 10 seconds. */
+  GST_DEBUG ("pushing first buffer");
+  buffer = gst_buffer_new_and_alloc (10);
+  GST_BUFFER_TIMESTAMP (buffer) = 1 * GST_SECOND;
+  GST_BUFFER_DURATION (buffer) = 100 * GST_SECOND;
+  res = gst_pad_chain (sinkpad, buffer);
+  fail_unless (res == GST_FLOW_OK, "no OK flow return");
+
+  /* scond buffer, will not block either but position should still be 10
+   * seconds */
+  GST_DEBUG ("pushing second buffer");
+  buffer = gst_buffer_new_and_alloc (10);
+  GST_BUFFER_TIMESTAMP (buffer) = 100 * GST_SECOND;
+  GST_BUFFER_DURATION (buffer) = 100 * GST_SECOND;
+  res = gst_pad_chain (sinkpad, buffer);
+  fail_unless (res == GST_FLOW_OK, "no OK flow return");
+
+  /* check if position is still 10 seconds */
+  format = GST_FORMAT_TIME;
+  gst_element_query_position (sink, &format, &position);
+  GST_DEBUG ("first buffer position %" GST_TIME_FORMAT,
+      GST_TIME_ARGS (position));
+  fail_unless (position == 10 * GST_SECOND, "position is wrong");
+
+  /* last buffer, blocks because preroll queue is filled. Start the push in a
+   * new thread so that we can check the position */
+  GST_DEBUG ("starting thread");
+  thread = g_thread_create ((GThreadFunc) send_buffer, sinkpad, TRUE, NULL);
+  fail_if (thread == NULL, "no thread");
+
+  GST_DEBUG ("waiting 1 second");
+  g_usleep (G_USEC_PER_SEC);
+  GST_DEBUG ("waiting done");
+
+  /* check if position is still 10 seconds. This is racy because  the above
+   * thread might not yet have started the push, because of the above sleep,
+   * this is very unlikely, though. */
+  format = GST_FORMAT_TIME;
+  gst_element_query_position (sink, &format, &position);
+  GST_DEBUG ("second buffer position %" GST_TIME_FORMAT,
+      GST_TIME_ARGS (position));
+  fail_unless (position == 10 * GST_SECOND, "position is wrong");
+
+  /* now we go to playing. This should unlock and stop the above thread. */
+  GST_DEBUG ("going to PLAYING");
+  ret = gst_element_set_state (sink, GST_STATE_PLAYING);
+
+  /* join the thread. At this point we know the sink processed the last buffer
+   * and the position should now be 210 seconds; the time of the last buffer we
+   * pushed */
+  GST_DEBUG ("joining thread");
+  g_thread_join (thread);
+
+  format = GST_FORMAT_TIME;
+  gst_element_query_position (sink, &format, &position);
+  GST_DEBUG ("last buffer position %" GST_TIME_FORMAT,
+      GST_TIME_ARGS (position));
+  fail_unless (position == 210 * GST_SECOND, "position is wrong");
+
+  gst_object_unref (sinkpad);
+
+  gst_element_set_state (sink, GST_STATE_NULL);
+  gst_object_unref (sink);
+  gst_object_unref (bus);
+}
+
+GST_END_TEST;
+
+/* when we get the ASYNC_DONE, query the position */
+static GstBusSyncReply
+async_done_eos_func (GstBus * bus, GstMessage * msg, GstElement * sink)
+{
+  if (GST_MESSAGE_TYPE (msg) == GST_MESSAGE_ASYNC_DONE) {
+    GstFormat format;
+    gint64 position;
+
+    GST_DEBUG ("we have ASYNC_DONE now");
+
+    /* get the position now */
+    format = GST_FORMAT_TIME;
+    gst_element_query_position (sink, &format, &position);
+
+    GST_DEBUG ("we have position %" GST_TIME_FORMAT, GST_TIME_ARGS (position));
+
+    fail_unless (position == 10 * GST_SECOND, "position is wrong");
+  }
+  /* we can drop the message, nothing is listening for it. */
+  return GST_BUS_DROP;
+}
+
+/* when we get the ASYNC_DONE message from a sink, we want the sink to be able
+ * to report the duration and position. The sink should also have called the
+ * render method. */
+GST_START_TEST (test_async_done_eos)
+{
+  GstElement *sink;
+  GstEvent *event;
+  GstStateChangeReturn ret;
+  GstPad *sinkpad;
+  gboolean res;
+  GstBus *bus;
+  GstFormat format;
+  gint64 position;
+  gboolean qret;
+
+  sink = gst_element_factory_make ("fakesink", "sink");
+  g_object_set (G_OBJECT (sink), "sync", TRUE, NULL);
+  g_object_set (G_OBJECT (sink), "preroll-queue-len", 1, NULL);
+
+  sinkpad = gst_element_get_pad (sink, "sink");
+
+  ret = gst_element_set_state (sink, GST_STATE_PAUSED);
+  fail_unless (ret == GST_STATE_CHANGE_ASYNC, "no ASYNC state return");
+
+  /* set bus on element synchronously listen for ASYNC_DONE */
+  bus = gst_bus_new ();
+  gst_element_set_bus (sink, bus);
+  gst_bus_set_sync_handler (bus, (GstBusSyncHandler) async_done_eos_func, sink);
+
+  /* make newsegment, this sets the position to 10sec when the buffer prerolls */
+  GST_DEBUG ("sending segment");
+  event =
+      gst_event_new_new_segment (FALSE, 1.0, GST_FORMAT_TIME, 0, -1,
+      10 * GST_SECOND);
+  res = gst_pad_send_event (sinkpad, event);
+
+  /* We have not yet received any buffers so we are still in the READY state,
+   * the position is therefore still not queryable. */
+  format = GST_FORMAT_TIME;
+  position = -1;
+  qret = gst_element_query_position (sink, &format, &position);
+  fail_unless (qret == FALSE, "position wrong");
+  fail_unless (position == -1, "position is wrong");
+
+  /* Since we are paused and the preroll queue has a length of 1, this function
+   * will return immediatly. The EOS will complete the preroll and the  
+   * position should now be 10 seconds. */
+  GST_DEBUG ("pushing EOS");
+  event = gst_event_new_eos ();
+  res = gst_pad_send_event (sinkpad, event);
+  fail_unless (res == TRUE, "no TRUE return");
+
+  /* check if position is still 10 seconds */
+  format = GST_FORMAT_TIME;
+  gst_element_query_position (sink, &format, &position);
+  GST_DEBUG ("EOS position %" GST_TIME_FORMAT, GST_TIME_ARGS (position));
+  fail_unless (position == 10 * GST_SECOND, "position is wrong");
+
+  gst_object_unref (sinkpad);
+
+  gst_element_set_state (sink, GST_STATE_NULL);
+  gst_object_unref (sink);
+  gst_object_unref (bus);
+}
+
+GST_END_TEST;
+
 /* test: try changing state of sinks */
 Suite *
 gst_sinks_suite (void)
@@ -1002,6 +1265,8 @@ gst_sinks_suite (void)
   tcase_add_test (tc_chain, test_add_live2);
   tcase_add_test (tc_chain, test_bin_live);
   tcase_add_test (tc_chain, test_fake_eos);
+  tcase_add_test (tc_chain, test_async_done);
+  tcase_add_test (tc_chain, test_async_done_eos);
 
   return s;
 }