streamsynchronizer: Send GAP event to finish preroll when change state from PLAYING to PAUSED

Wait in the event function when EOS is received until all pads are EOS
and then forward the EOS event from each pads own event function.

Also send a new GAP event for EOS pads from the event function whenever
going from PLAYING->PAUSED by shortly waking up the GCond. This is needed
to allow sinks to pre-roll again, as they did not receive EOS yet because
we blocked that, but also will never get data again.

https://bugzilla.gnome.org/show_bug.cgi?id=736655

1 files changed, 117 insertions, 22 deletions

diff --git a/gst/playback/gststreamsynchronizer.c b/gst/playback/gststreamsynchronizer.c
index 9e06ee766..dfc955faa 100644
--- a/gst/playback/gststreamsynchronizer.c
+++ b/gst/playback/gststreamsynchronizer.c
@@ -67,6 +67,8 @@ typedef struct
   gboolean wait;                /* TRUE if waiting/blocking */
   gboolean new_stream;
   gboolean is_eos;              /* TRUE if EOS was received */
+  gboolean eos_sent;            /* when EOS was sent downstream */
+  gboolean flushing;            /* set after flush-start and before flush-stop */
   gboolean seen_data;
 
   GCond stream_finish_cond;
@@ -202,6 +204,52 @@ out:
   return ret;
 }
 
+/* must be called with the STREAM_SYNCHRONIZER_LOCK */
+static gboolean
+gst_stream_synchronizer_wait (GstStreamSynchronizer * self, GstPad * pad)
+{
+  gboolean ret = FALSE;
+  GstStream *stream;
+
+  while (!self->eos && !self->flushing) {
+    stream = gst_pad_get_element_private (pad);
+    if (!stream) {
+      GST_WARNING_OBJECT (pad, "EOS for unknown stream");
+      return ret;
+    }
+    if (stream->flushing) {
+      GST_DEBUG_OBJECT (pad, "Flushing");
+      break;
+    }
+
+    if (self->send_gap_event) {
+      GstEvent *event;
+
+      if (!GST_CLOCK_TIME_IS_VALID (stream->segment.position)) {
+        GST_WARNING_OBJECT (pad, "Have no position and can't send GAP event");
+        self->send_gap_event = FALSE;
+        continue;
+      }
+
+      event = gst_event_new_gap (stream->segment.position, GST_CLOCK_TIME_NONE);
+      GST_DEBUG_OBJECT (pad, "Send GAP event");
+      /* drop lock when sending GAP event, which may block in e.g. preroll */
+      GST_STREAM_SYNCHRONIZER_UNLOCK (self);
+      ret = gst_pad_push_event (pad, event);
+      GST_STREAM_SYNCHRONIZER_LOCK (self);
+      if (!ret) {
+        GST_ERROR_OBJECT (pad, "Failed to send GAP event");
+        return ret;
+      }
+      self->send_gap_event = FALSE;
+    }
+
+    g_cond_wait (&stream->stream_finish_cond, &self->lock);
+  }
+
+  return TRUE;
+}
+
 /* sinkpad functions */
 static gboolean
 gst_stream_synchronizer_sink_event (GstPad * pad, GstObject * parent,
@@ -241,6 +289,8 @@ gst_stream_synchronizer_sink_event (GstPad * pad, GstObject * parent,
       if ((have_group_id && stream->group_id != group_id) || (!have_group_id
               && stream->stream_start_seqnum != seqnum)) {
         stream->is_eos = FALSE;
+        stream->eos_sent = FALSE;
+        stream->flushing = FALSE;
         stream->stream_start_seqnum = seqnum;
         stream->group_id = group_id;
 
@@ -395,8 +445,10 @@ gst_stream_synchronizer_sink_event (GstPad * pad, GstObject * parent,
 
       GST_STREAM_SYNCHRONIZER_LOCK (self);
       stream = gst_pad_get_element_private (pad);
+      self->eos = FALSE;
       if (stream) {
         GST_DEBUG_OBJECT (pad, "Flushing streams");
+        stream->flushing = TRUE;
         g_cond_broadcast (&stream->stream_finish_cond);
       }
       GST_STREAM_SYNCHRONIZER_UNLOCK (self);
@@ -413,9 +465,10 @@ gst_stream_synchronizer_sink_event (GstPad * pad, GstObject * parent,
         gst_segment_init (&stream->segment, GST_FORMAT_UNDEFINED);
 
         stream->is_eos = FALSE;
+        stream->eos_sent = FALSE;
+        stream->flushing = FALSE;
         stream->wait = FALSE;
         stream->new_stream = FALSE;
-        stream->seen_data = FALSE;
         g_cond_broadcast (&stream->stream_finish_cond);
       }
       GST_STREAM_SYNCHRONIZER_UNLOCK (self);
@@ -451,6 +504,8 @@ gst_stream_synchronizer_sink_event (GstPad * pad, GstObject * parent,
       else
         timestamp = stream->segment.stop;
 
+      stream->segment.position = timestamp;
+
       for (l = self->streams; l; l = l->next) {
         GstStream *ostream = l->data;
 
@@ -461,6 +516,7 @@ gst_stream_synchronizer_sink_event (GstPad * pad, GstObject * parent,
 
       if (all_eos) {
         GST_DEBUG_OBJECT (self, "All streams are EOS -- forwarding");
+        self->eos = TRUE;
         for (l = self->streams; l; l = l->next) {
           GstStream *ostream = l->data;
           /* local snapshot of current pads */
@@ -468,41 +524,52 @@ gst_stream_synchronizer_sink_event (GstPad * pad, GstObject * parent,
           pads = g_slist_prepend (pads, ostream->srcpad);
         }
       }
-      GST_STREAM_SYNCHRONIZER_UNLOCK (self);
-      /* drop lock when sending eos, which may block in e.g. preroll */
       if (pads) {
         GstPad *pad;
         GSList *epad;
+        GstStream *ostream;
 
         ret = TRUE;
         epad = pads;
         while (epad) {
           pad = epad->data;
-          GST_DEBUG_OBJECT (pad, "Pushing EOS");
-          ret = ret && gst_pad_push_event (pad, gst_event_new_eos ());
+          ostream = gst_pad_get_element_private (pad);
+          if (ostream) {
+            g_cond_broadcast (&ostream->stream_finish_cond);
+          }
+
           gst_object_unref (pad);
           epad = g_slist_next (epad);
         }
         g_slist_free (pads);
       } else {
-        /* if EOS, but no data has passed, then send something to replace EOS
-         * for preroll purposes */
-        if (!seen_data) {
-          GstEvent *gap_event;
-
-          gap_event = gst_event_new_gap (timestamp, GST_CLOCK_TIME_NONE);
-          ret = gst_pad_push_event (srcpad, gap_event);
-        } else {
-          GstEvent *gap_event;
-
-          /* FIXME: Also send a GAP event to let audio sinks start their
-           * clock in case they did not have enough data yet */
-          gap_event = gst_event_new_gap (timestamp, GST_CLOCK_TIME_NONE);
-          ret = gst_pad_push_event (srcpad, gap_event);
+        if (seen_data) {
+          self->send_gap_event = TRUE;
+          ret = gst_stream_synchronizer_wait (self, srcpad);
+        }
+      }
+
+      /* send eos if haven't seen data. seen_data will be ture if data buffer
+       * of the track have received in anytime. sink is ready if seen_data is
+       * true, so can send GAP event. Will send EOS if sink isn't ready. The
+       * scenario for the case is one track haven't any media data and then
+       * send EOS. Or no any valid media data in one track, so decoder can't
+       * get valid CAPS for the track. sink can't ready without received CAPS.*/
+      if (!seen_data || self->eos) {
+        GST_DEBUG_OBJECT (pad, "send EOS event");
+        /* drop lock when sending eos, which may block in e.g. preroll */
+        GST_STREAM_SYNCHRONIZER_UNLOCK (self);
+        ret = gst_pad_push_event (srcpad, gst_event_new_eos ());
+        GST_STREAM_SYNCHRONIZER_LOCK (self);
+        stream = gst_pad_get_element_private (pad);
+        if (stream) {
+          stream->eos_sent = TRUE;
         }
       }
+
       gst_object_unref (srcpad);
       gst_event_unref (event);
+      GST_STREAM_SYNCHRONIZER_UNLOCK (self);
       goto done;
     }
     default:
@@ -599,7 +666,8 @@ gst_stream_synchronizer_sink_chain (GstPad * pad, GstObject * parent,
       GstStream *ostream = l->data;
       gint64 position;
 
-      if (!ostream->is_eos || ostream->segment.format != GST_FORMAT_TIME)
+      if (!ostream->is_eos || ostream->eos_sent ||
+          ostream->segment.format != GST_FORMAT_TIME)
         continue;
 
       if (ostream->segment.position != -1)
@@ -651,6 +719,7 @@ gst_stream_synchronizer_request_new_pad (GstElement * element,
   stream->stream_start_seqnum = G_MAXUINT32;
   stream->segment_seqnum = G_MAXUINT32;
   stream->group_id = G_MAXUINT;
+  stream->seen_data = FALSE;
 
   tmp = g_strdup_printf ("sink_%u", self->current_stream_number);
   stream->sinkpad = gst_pad_new_from_static_template (&sinktemplate, tmp);
@@ -795,6 +864,7 @@ gst_stream_synchronizer_change_state (GstElement * element,
     case GST_STATE_CHANGE_NULL_TO_READY:
       GST_DEBUG_OBJECT (self, "State change NULL->READY");
       self->shutdown = FALSE;
+      self->send_gap_event = FALSE;
       break;
     case GST_STATE_CHANGE_READY_TO_PAUSED:
       GST_DEBUG_OBJECT (self, "State change READY->PAUSED");
@@ -802,18 +872,21 @@ gst_stream_synchronizer_change_state (GstElement * element,
       self->have_group_id = TRUE;
       self->group_id = G_MAXUINT;
       self->shutdown = FALSE;
+      self->flushing = FALSE;
+      self->eos = FALSE;
       break;
     case GST_STATE_CHANGE_PAUSED_TO_READY:{
       GList *l;
 
-      GST_DEBUG_OBJECT (self, "State change READY->NULL");
+      GST_DEBUG_OBJECT (self, "State change PAUSED->READY");
 
       GST_STREAM_SYNCHRONIZER_LOCK (self);
+      self->flushing = TRUE;
+      self->shutdown = TRUE;
       for (l = self->streams; l; l = l->next) {
         GstStream *ostream = l->data;
         g_cond_broadcast (&ostream->stream_finish_cond);
       }
-      self->shutdown = TRUE;
       GST_STREAM_SYNCHRONIZER_UNLOCK (self);
     }
     default:
@@ -826,6 +899,25 @@ gst_stream_synchronizer_change_state (GstElement * element,
     return ret;
 
   switch (transition) {
+    case GST_STATE_CHANGE_PLAYING_TO_PAUSED:{
+      GList *l;
+
+      GST_DEBUG_OBJECT (self, "State change PLAYING->PAUSED");
+
+      GST_STREAM_SYNCHRONIZER_LOCK (self);
+      for (l = self->streams; l; l = l->next) {
+        GstStream *stream = l->data;
+        /* send GAP event to sink to finished pre-roll. The reason is function
+         * chain () will be blocked on pad_push (), so can't trigger the track
+         * which reach EOS to send GAP event. */
+        if (stream->is_eos && !stream->eos_sent) {
+          self->send_gap_event = TRUE;
+          g_cond_broadcast (&stream->stream_finish_cond);
+        }
+      }
+      GST_STREAM_SYNCHRONIZER_UNLOCK (self);
+      break;
+    }
     case GST_STATE_CHANGE_PAUSED_TO_READY:{
       GList *l;
 
@@ -833,6 +925,7 @@ gst_stream_synchronizer_change_state (GstElement * element,
       self->group_start_time = 0;
 
       GST_STREAM_SYNCHRONIZER_LOCK (self);
+      self->send_gap_event = FALSE;
       for (l = self->streams; l; l = l->next) {
         GstStream *stream = l->data;
 
@@ -840,6 +933,8 @@ gst_stream_synchronizer_change_state (GstElement * element,
         stream->wait = FALSE;
         stream->new_stream = FALSE;
         stream->is_eos = FALSE;
+        stream->eos_sent = FALSE;
+        stream->flushing = FALSE;
       }
       GST_STREAM_SYNCHRONIZER_UNLOCK (self);
       break;