docs/design/part-element-sink.txt: Updated document.

Original commit message from CVS:
* docs/design/part-element-sink.txt:
Updated document.

* libs/gst/base/gstbasesink.c: (gst_base_sink_init),
(gst_base_sink_finalize), (gst_base_sink_preroll_queue_flush),
(gst_base_sink_configure_segment), (gst_base_sink_commit_state),
(gst_base_sink_get_sync_times), (gst_base_sink_wait_clock),
(gst_base_sink_do_sync), (gst_base_sink_render_object),
(gst_base_sink_preroll_object),
(gst_base_sink_queue_object_unlocked),
(gst_base_sink_queue_object), (gst_base_sink_event),
(gst_base_sink_chain_unlocked), (gst_base_sink_chain),
(gst_base_sink_loop), (gst_base_sink_activate_pull),
(gst_base_sink_get_position), (gst_base_sink_change_state):
* libs/gst/base/gstbasesink.h:
Totally refactored matching the design doc.
Use two segments, one to clip incomming buffers and another to
perform sync.
Handle queueing correctly, bypass the queue when playing.
Make EOS cancelable.
Handle errors correctly when operating in pull based mode.

* tests/check/elements/fakesink.c: (GST_START_TEST),
(fakesink_suite):
Added new check for sinks.

ChangeLog

@@ -1,3 +1,31 @@
+2006-02-02  Wim Taymans  <wim@fluendo.com>
+
+	* docs/design/part-element-sink.txt:
+	Updated document.
+
+	* libs/gst/base/gstbasesink.c: (gst_base_sink_init),
+	(gst_base_sink_finalize), (gst_base_sink_preroll_queue_flush),
+	(gst_base_sink_configure_segment), (gst_base_sink_commit_state),
+	(gst_base_sink_get_sync_times), (gst_base_sink_wait_clock),
+	(gst_base_sink_do_sync), (gst_base_sink_render_object),
+	(gst_base_sink_preroll_object),
+	(gst_base_sink_queue_object_unlocked),
+	(gst_base_sink_queue_object), (gst_base_sink_event),
+	(gst_base_sink_chain_unlocked), (gst_base_sink_chain),
+	(gst_base_sink_loop), (gst_base_sink_activate_pull),
+	(gst_base_sink_get_position), (gst_base_sink_change_state):
+	* libs/gst/base/gstbasesink.h:
+	Totally refactored matching the design doc.
+	Use two segments, one to clip incomming buffers and another to
+	perform sync.
+	Handle queueing correctly, bypass the queue when playing.
+	Make EOS cancelable.
+	Handle errors correctly when operating in pull based mode.
+
+	* tests/check/elements/fakesink.c: (GST_START_TEST),
+	(fakesink_suite):
+	Added new check for sinks.
+
 2006-02-02  Wim Taymans  <wim@fluendo.com>
 
 	* gst/gstsegment.c: (gst_segment_clip):

docs/design/part-element-sink.txt

@@ -31,96 +31,170 @@ Events other than EOS do not complete the preroll stage.
 sink overview
 -------------
 
-  /* commit the state. We return TRUE if we can continue
+ - TODO: PREROLL_LOCK can be removed and we can safely use the STREAM_LOCK.
-   * streaming, FALSE in the case we go to a READY or NULL state.
+
-   * if we go to PLAYING, we don't need to block on preroll.
+
-   */
+
+  # commit the state. We return TRUE if we can continue
+  # streaming, FALSE in the case we go to a READY or NULL state.
+  # if we go to PLAYING, we don't need to block on preroll.
   commit
+  {
     LOCK
-    switch (pending) {
+    switch (pending)
       case PLAYING:
-        need_preroll = FALSE;
+        need_preroll = FALSE
-	break;
+	break
       case PAUSED:
-	break;
+	break
       case READY:
       case NULL:
-        return FALSE;
+        return FALSE
       case VOID:
-        return TRUE;
+        return TRUE
-    }
-    /* update state */
-    state = pending;
-    next = VOID;
-    pending = VOID;
-    UNLOCK
-    return TRUE;
 
-  /* handle a prerollable item (EOS or buffer). It is
+    # update state
-   * always called with the PREROLL_LOCK helt.
+    state = pending
-   * need_preroll indicates that we must perform, commit and
+    next = VOID
-   * potentially block on preroll.
+    pending = VOID
-   */
+    UNLOCK
-  handle (time)
+    return TRUE
+  }
+
+  # sync an object. We have to wait for the element to reach
+  # the PLAYING state before we can wait on the clock.
+  # some items do not need synchronisation (most events) so the
+  # get_times method returns FALSE (not syncable)
+  # need_preroll indicates that we are not in the PLAYING state
+  # and therefore need to commit and potentially block on preroll
+  # if our clock_wait got interrupted we commit and block again.
+  # The reason for this is that the current item being rendered is
+  # not yet finished and we can use that item to finish preroll.
+  do_sync (obj)
+  {
+    # get timing information for this object
+    syncable = get_times (obj, &start, &stop)
+    if (!syncable)
+      return OK;
  again:
-    while (need_preroll) {
+    while (need_preroll)
-      preroll
+      if (need_commit)
+        need_commit = FALSE
         if (!commit)
           return WRONG_STATE
-      /* commit could have made us not need preroll anymore. */
+
-      if (need_preroll) {
+      if (need_preroll)
-        /* release PREROLL_LOCK and wait. prerolled can be observed
+        # release PREROLL_LOCK and wait. prerolled can be observed
-	 * and will be TRUE */
+        # and will be TRUE
-        prerolled = TRUE;
+        prerolled = TRUE
         PREROLL_WAIT (releasing PREROLL_LOCK)
-        prerolled = FALSE;
+        prerolled = FALSE
         if (flushing)
           return WRONG_STATE
-      }
+
-    }
+    if (valid (start || stop))
-    if (clock && sync) {
-      /* the only way we can regain the prerolled state is when
-       * the clock entry gets unscheduled, we then preroll (again) on the
-       * current item, else we render and preroll on the next buffer. */
       PREROLL_UNLOCK
-      ret = wait_clock;
+      end_time = stop
+      ret = wait_clock (obj,start)
       PREROLL_LOCK
-      if (flushing)                | /* sinks that sync on buffer contents do like this */
+      if (flushing)
-        return WRONG_STATE         | while (more_to_render) {
+        return WRONG_STATE 
-      if (ret == UNSCHEDULED)      |   ret = render 
+      # if the clock was unscheduled, we redo the
-        goto again;                |   if (ret == interrupted)
+      # preroll
-    }                              |     prerolled = TRUE;
+      if (ret == UNSCHEDULED)
-    render                   ----->|     PREROLL_WAIT (releasing PREROLL_LOCK)
+        goto again          
-                                   |     prerolled = FALSE;
+  }
+
+  # render a prerollable item (EOS or buffer). It is
+  # always called with the PREROLL_LOCK helt.
+  render_object (obj)
+  {
+    ret = do_sync (obj)
+    if (ret != OK)
+      return ret;
+
+    # preroll and syncing done, now we can render
+    render(obj)
+  }
+                                   | # sinks that sync on buffer contents do like this
+                                   | while (more_to_render)
+                                   |   ret = render 
+                                   |   if (ret == interrupted)
+                                   |     prerolled = TRUE
+    render (buffer)          ----->|     PREROLL_WAIT (releasing PREROLL_LOCK)
+                                   |     prerolled = FALSE
                                    |     if (flushing)
                                    |       return WRONG_STATE
-                                   | }
+                                   | 
-  /* various event functions */
+
+  # queue a prerollable item (EOS or buffer). It is
+  # always called with the PREROLL_LOCK helt.
+  # This function will commit the state when receiving the
+  # first prerollable item.
+  # items are then added to the rendering queue or rendered
+  # right away if no preroll is needed.
+  queue (obj, prerollable) 
+  {
+    if (prerollable)
+      queuelen++
+
+    if (need_preroll)
+      # first item in the queue while we need preroll
+      # will complete state change and call preroll
+      if (queuelen == 1)
+        preroll (obj)
+        if (need_commit)
+          need_commit = FALSE
+          if (!commit)
+            return WRONG_STATE
+
+      # then see if we need more preroll items before we
+      # can block
+      if (need_preroll)
+        if (queuelen <= maxqueue)
+          queue.add (obj)
+          return OK
+
+        # now clear the queue and render each item before
+	# rendering the current item.
+        while (queue.hasItem)
+          render_object (queue.remove())
+
+    render_object (obj)
+    queuelen = 0
+  }
+
+  # various event functions
   event
     EOS:
+      # events must complete preroll too
       STREAM_LOCK
       PREROLL_LOCK
       if (flushing)
         return FALSE
-      ret = handle (end_time);
+      ret = queue (event, TRUE)
       if (ret == WRONG_STATE)
         return FALSE
-      post_eos
-      eos = TRUE;
       PREROLL_UNLOCK
       STREAM_UNLOCK
-      break;
+      break
     NEWSEGMENT:
+      # the newsegment must be used to clip incomming
+      # buffers. Then then go into the queue as non-prerollable
+      # items used for syncing the buffers
       STREAM_LOCK
       PREROLL_LOCK
       if (flushing)
         return FALSE
       set_clip
-      event 
+      ret = queue (event, FALSE)
+      if (ret == WRONG_STATE)
+        return FALSE
       PREROLL_UNLOCK
       STREAM_UNLOCK
-      break;
+      break
     FLUSH_START:
+      # set flushing and unblock all that is waiting
       event                   ----> subclasses can interrupt render
       PREROLL_LOCK
       flushing = TRUE
@@ -130,67 +204,89 @@ sink overview
       STREAM_LOCK
       lost_state
       STREAM_UNLOCK
-      break;
+      break
     FLUSH_END:
+      # unset flushing and clear all data and eos
       STREAM_LOCK
       event 
       PREROLL_LOCK
+      queue.clear
+      queuelen = 0
       flushing = FALSE
-      eos = FALSE;
+      eos = FALSE
       PREROLL_UNLOCK
       STREAM_UNLOCK
-      break;
+      break
 
+  # the chain function checks the buffer falls within the
+  # configured segment and queues the buffer for preroll and
+  # rendering
   chain
     STREAM_LOCK
     PREROLL_LOCK
     if (flushing)
       return WRONG_STATE
     if (clip)
-      handle (time);
+      queue (buffer, TRUE)
     PREROLL_UNLOCK
     STREAM_UNLOCK
 
   state
     switch (transition)
       READY_PAUSED:
-        ret = ASYNC;
+        # no datapassing is going on so we always return ASYNC
+        ret = ASYNC
+	need_commit = TRUE
         eos = FALSE
         flushing = FALSE
-	need_preroll = TRUE;
+	need_preroll = TRUE
-	prerolled = FALSE;
+	prerolled = FALSE
-        break;
+        break
       PAUSED_PLAYING:
+        # we grab the preroll lock. This we can only do if the
+	# chain function is either doing some clock sync, we are
+	# waiting for preroll or the chain function is not being called.
         PREROLL_LOCK
 	if (prerolled || eos)
-          PREROLL_SIGNAL
+          ret = OK
-          ret = OK;
+	  need_commit = FALSE
-	  need_preroll = FALSE;
+	  need_preroll = FALSE
 	  if (eos)
 	    post_eos
 	  else
-	  need_preroll = TRUE;
+            PREROLL_SIGNAL
-          ret = ASYNC;
+	else
+	  need_preroll = TRUE
+	  need_commit = TRUE
+          ret = ASYNC
         PREROLL_UNLOCK
-        break;
+        break
       PLAYING_PAUSED:
                            ---> subclass can interrupt render
+        # we grab the preroll lock. This we can only do if the
+	# chain function is either doing some clock sync
+	# or the chain function is not being called. 
         PREROLL_LOCK
-	need_preroll = TRUE;
+	need_preroll = TRUE
         unlock_clock
 	if (prerolled || eos)
-          ret = OK;
+          ret = OK
 	else
-          ret = ASYNC;
+          ret = ASYNC
         PREROLL_UNLOCK
-        break;
+        break
       PAUSED_READY:
                            ---> subclass can interrupt render
+        # we grab the preroll lock. Set to flushing and unlock
+	# everything. This should exit the chain functions and stop
+	# streaming.
         PREROLL_LOCK
         flushing = TRUE
         unlock_clock
+        queue.clear
+        queuelen = 0
         PREROLL_SIGNAL
-        ret = OK;
+        ret = OK
         PREROLL_UNLOCK
-        break;
+        break
 

libs/gst/base/gstbasesink.h

@@ -88,7 +88,15 @@ struct _GstBaseSink {
   gboolean       flushing;
 
   /*< private >*/
-  gpointer       _gst_reserved[GST_PADDING_LARGE];
+  union {
+    struct {
+      /* segment used for clipping incomming buffers */
+      GstSegment  *clip_segment;
+    } ABI;
+    /* adding + 0 to mark ABI change to be undone later */
+    gpointer _gst_reserved[GST_PADDING_LARGE + 0];
+  } abidata;
+
 };
 
 struct _GstBaseSinkClass {

tests/check/elements/fakesink.c

@@ -166,7 +166,7 @@ GST_START_TEST (test_clipping)
     fail_unless (current == GST_STATE_PAUSED);
     fail_unless (pending == GST_STATE_VOID_PENDING);
 
-    /* pause should render the buffer */
+    /* playing should render the buffer */
     ret = gst_element_set_state (sink, GST_STATE_PLAYING);
     fail_unless (ret == GST_STATE_CHANGE_SUCCESS);
 
@@ -217,6 +217,81 @@ GST_START_TEST (test_clipping)
 
 GST_END_TEST;
 
+GST_START_TEST (test_preroll_sync)
+{
+  GstElement *pipeline, *sink;
+  GstPad *sinkpad;
+  GstStateChangeReturn ret;
+
+  /* create sink */
+  pipeline = gst_pipeline_new ("pipeline");
+  fail_if (pipeline == NULL);
+
+  sink = gst_element_factory_make ("fakesink", "sink");
+  fail_if (sink == NULL);
+  g_object_set (G_OBJECT (sink), "sync", TRUE, NULL);
+
+  gst_bin_add (GST_BIN (pipeline), sink);
+
+  sinkpad = gst_element_get_pad (sink, "sink");
+  fail_if (sinkpad == NULL);
+
+  /* make pipeline and element ready to accept data */
+  ret = gst_element_set_state (pipeline, GST_STATE_PAUSED);
+  fail_unless (ret == GST_STATE_CHANGE_ASYNC);
+
+  /* send segment */
+  {
+    GstEvent *segment;
+    gboolean eret;
+
+    GST_DEBUG ("sending segment");
+    segment = gst_event_new_new_segment (FALSE,
+        1.0, GST_FORMAT_TIME, 0 * GST_SECOND, 2 * GST_SECOND, 0 * GST_SECOND);
+
+    eret = gst_pad_send_event (sinkpad, segment);
+    fail_if (eret == FALSE);
+  }
+
+  /* send buffer that should block and finish preroll */
+  {
+    GstBuffer *buffer;
+    GstFlowReturn fret;
+    ChainData *data;
+    GstState current, pending;
+
+    buffer = gst_buffer_new ();
+    GST_BUFFER_TIMESTAMP (buffer) = 1 * GST_SECOND;
+    GST_BUFFER_DURATION (buffer) = 1 * GST_SECOND;
+
+    GST_DEBUG ("sending buffer to finish preroll");
+    data = chain_async (sinkpad, buffer);
+    fail_if (data == NULL);
+
+    /* state should now eventually change to PAUSED */
+    ret =
+        gst_element_get_state (pipeline, &current, &pending,
+        GST_CLOCK_TIME_NONE);
+    fail_unless (ret == GST_STATE_CHANGE_SUCCESS);
+    fail_unless (current == GST_STATE_PAUSED);
+    fail_unless (pending == GST_STATE_VOID_PENDING);
+
+    /* playing should render the buffer */
+    ret = gst_element_set_state (pipeline, GST_STATE_PLAYING);
+    fail_unless (ret == GST_STATE_CHANGE_SUCCESS);
+
+    /* and we should get a success return value */
+    fret = chain_async_return (data);
+    fail_if (fret != GST_FLOW_OK);
+  }
+  gst_element_set_state (pipeline, GST_STATE_NULL);
+  gst_element_get_state (pipeline, NULL, NULL, GST_CLOCK_TIME_NONE);
+  gst_object_unref (sinkpad);
+  gst_object_unref (pipeline);
+}
+
+GST_END_TEST;
+
 Suite *
 fakesink_suite (void)
 {
@@ -225,6 +300,7 @@ fakesink_suite (void)
 
   suite_add_tcase (s, tc_chain);
   tcase_add_test (tc_chain, test_clipping);
+  tcase_add_test (tc_chain, test_preroll_sync);
 
   return s;
 }