decodebin2: only allow 'lower' multiqueues to emit buffering messages

When we have a scenario of demuxers linked to demuxers, decodebin2
will create multiqueue at different levels of the pipeline. The problem
is that only the lowest multiqueue's should do the buffering messaging,
as they will handle with the raw streams data.

When all multiqueues are doing buffering, the upper ones can handle
large buffers that easily fill them, moving from 0% to 100% from
buffer to buffer, causing too much buffering messages to be posted.
This hangs the pipeline unnecessarily and might lead to deadlocks.

1 files changed, 69 insertions, 0 deletions

diff --git a/gst/playback/gstdecodebin2.c b/gst/playback/gstdecodebin2.c
index 4577155b7..b871cb87c 100644
--- a/gst/playback/gstdecodebin2.c
+++ b/gst/playback/gstdecodebin2.c
@@ -455,12 +455,15 @@ static gboolean gst_decode_chain_is_complete (GstDecodeChain * chain);
 static gboolean gst_decode_chain_expose (GstDecodeChain * chain,
     GList ** endpads, gboolean * missing_plugin);
 static gboolean gst_decode_chain_is_drained (GstDecodeChain * chain);
+static gboolean gst_decode_chain_reset_buffering (GstDecodeChain * chain);
 static gboolean gst_decode_group_is_complete (GstDecodeGroup * group);
 static GstPad *gst_decode_group_control_demuxer_pad (GstDecodeGroup * group,
     GstPad * pad);
 static gboolean gst_decode_group_is_drained (GstDecodeGroup * group);
+static gboolean gst_decode_group_reset_buffering (GstDecodeGroup * group);
 
 static gboolean gst_decode_bin_expose (GstDecodeBin * dbin);
+static void gst_decode_bin_reset_buffering (GstDecodeBin * dbin);
 
 #define CHAIN_MUTEX_LOCK(chain) G_STMT_START {				\
     GST_LOG_OBJECT (chain->dbin,					\
@@ -2707,10 +2710,42 @@ are_final_caps (GstDecodeBin * dbin, GstCaps * caps)
   return res;
 }
 
+/* gst_decode_bin_reset_buffering:
+ *
+ * Enables buffering on the last multiqueue of each group only,
+ * disabling the rest
+ *
+ */
+static void
+gst_decode_bin_reset_buffering (GstDecodeBin * dbin)
+{
+  if (!dbin->use_buffering)
+    return;
+
+  GST_DEBUG_OBJECT (dbin, "Reseting multiqueues buffering");
+  CHAIN_MUTEX_LOCK (dbin->decode_chain);
+  gst_decode_chain_reset_buffering (dbin->decode_chain);
+  CHAIN_MUTEX_UNLOCK (dbin->decode_chain);
+}
+
 /****
  * GstDecodeChain functions
  ****/
 
+static gboolean
+gst_decode_chain_reset_buffering (GstDecodeChain * chain)
+{
+  GstDecodeGroup *group;
+
+  group = chain->active_group;
+  GST_LOG_OBJECT (chain->dbin, "Resetting chain %p buffering, active group: %p",
+      chain, group);
+  if (group) {
+    return gst_decode_group_reset_buffering (group);
+  }
+  return FALSE;
+}
+
 /* gst_decode_chain_get_current_group:
  *
  * Returns the current group of this chain, to which
@@ -3528,6 +3563,37 @@ out:
   return drained;
 }
 
+static gboolean
+gst_decode_group_reset_buffering (GstDecodeGroup * group)
+{
+  GList *l;
+  gboolean ret = TRUE;
+
+  GST_DEBUG_OBJECT (group->dbin, "Group reset buffering %p %s", group,
+      GST_ELEMENT_NAME (group->multiqueue));
+  for (l = group->children; l; l = l->next) {
+    GstDecodeChain *chain = l->data;
+
+    CHAIN_MUTEX_LOCK (chain);
+    if (!gst_decode_chain_reset_buffering (chain)) {
+      ret = FALSE;
+    }
+    CHAIN_MUTEX_UNLOCK (chain);
+  }
+
+  if (ret) {
+    /* all chains are buffering already, no need to do it here */
+    g_object_set (group->multiqueue, "use-buffering", FALSE, NULL);
+  } else {
+    g_object_set (group->multiqueue, "use-buffering", TRUE, NULL);
+  }
+
+  GST_DEBUG_OBJECT (group->dbin, "Setting %s buffering to %d",
+      GST_ELEMENT_NAME (group->multiqueue), !ret);
+  return TRUE;
+}
+
+
 /* sort_end_pads:
  * GCompareFunc to use with lists of GstPad.
  * Sorts pads by mime type.
@@ -3831,6 +3897,9 @@ gst_decode_bin_expose (GstDecodeBin * dbin)
     return TRUE;
   }
 
+  /* going to expose something, reset buffering */
+  gst_decode_bin_reset_buffering (dbin);
+
   /* Set all already exposed pads to blocked */
   for (tmp = endpads; tmp; tmp = tmp->next) {
     GstDecodePad *dpad = tmp->data;