queue2: Post buffering messages earlier in ringbuffer mode

In ringbuffer mode we need to make sure we post buffering messages *before*
blocking to wait for data to be drained.

Without this, we would end up in situations like this:
* pipeline is pre-rolling
* Downstream demuxer/decoder has pushed data to all sinks, and demuxer thread
  is blocking downstream (i.e. not pulling from upstream/queue2).
* Therefore pipeline has pre-rolled ...
* ... but queue2 hasn't filled up yet, therefore the application waits for
  the buffering 100% messages before setting the pipeline to PLAYING
* But queue2 can't post that message, since the 100% message will be posted
  *after* there is room available for that last buffer.

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

1 files changed, 27 insertions, 5 deletions

diff --git a/plugins/elements/gstqueue2.c b/plugins/elements/gstqueue2.c
index 5e55ba294..8e5a5dc6c 100644
--- a/plugins/elements/gstqueue2.c
+++ b/plugins/elements/gstqueue2.c
@@ -275,6 +275,7 @@ static gboolean gst_queue2_is_filled (GstQueue2 * queue);
 
 static void update_cur_level (GstQueue2 * queue, GstQueue2Range * range);
 static void update_in_rates (GstQueue2 * queue, gboolean force);
+static GstMessage *gst_queue2_get_buffering_message (GstQueue2 * queue);
 static void gst_queue2_post_buffering (GstQueue2 * queue);
 
 typedef enum
@@ -1002,13 +1003,12 @@ get_buffering_stats (GstQueue2 * queue, gint percent, GstBufferingMode * mode,
   }
 }
 
-static void
-gst_queue2_post_buffering (GstQueue2 * queue)
+/* Called with the lock taken */
+static GstMessage *
+gst_queue2_get_buffering_message (GstQueue2 * queue)
 {
   GstMessage *msg = NULL;
 
-  g_mutex_lock (&queue->buffering_post_lock);
-  GST_QUEUE2_MUTEX_LOCK (queue);
   if (queue->percent_changed) {
     gint percent = queue->buffering_percent;
 
@@ -1020,6 +1020,18 @@ gst_queue2_post_buffering (GstQueue2 * queue)
     gst_message_set_buffering_stats (msg, queue->mode, queue->avg_in,
         queue->avg_out, queue->buffering_left);
   }
+
+  return msg;
+}
+
+static void
+gst_queue2_post_buffering (GstQueue2 * queue)
+{
+  GstMessage *msg = NULL;
+
+  g_mutex_lock (&queue->buffering_post_lock);
+  GST_QUEUE2_MUTEX_LOCK (queue);
+  msg = gst_queue2_get_buffering_message (queue);
   GST_QUEUE2_MUTEX_UNLOCK (queue);
 
   if (msg != NULL)
@@ -2028,8 +2040,18 @@ gst_queue2_create_write (GstQueue2 * queue, GstBuffer * buffer)
     update_cur_level (queue, queue->current);
 
     /* update the buffering status */
-    if (queue->use_buffering)
+    if (queue->use_buffering) {
+      GstMessage *msg;
       update_buffering (queue);
+      msg = gst_queue2_get_buffering_message (queue);
+      if (msg) {
+        GST_QUEUE2_MUTEX_UNLOCK (queue);
+        g_mutex_lock (&queue->buffering_post_lock);
+        gst_element_post_message (GST_ELEMENT_CAST (queue), msg);
+        g_mutex_unlock (&queue->buffering_post_lock);
+        GST_QUEUE2_MUTEX_LOCK (queue);
+      }
+    }
 
     GST_INFO_OBJECT (queue, "cur_level.bytes %u (max %" G_GUINT64_FORMAT ")",
         queue->cur_level.bytes, QUEUE_MAX_BYTES (queue));