ges-source: make sure internal decodebin doesn't get seeked too early

When uridecodebin exposes pads for its streams, we immediately ghost
the relevant (selected) one and let composition send a seek as soon as a
buffer is probed.

This means that sometimes uridecodebin is still linking elements
internally (for non-selected streams) and sees flush events travel down
the elements it is still busy trying to link / forward sticky events to.

This causes all sorts of nasty issues, which can be avoided by simply
blocking all data flow from the source until no-more-pads has been
emitted by uridecodebin (or whatever sub_element is wrapped).

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/3971>

1 files changed, 66 insertions, 2 deletions

diff --git a/subprojects/gst-editing-services/ges/ges-source.c b/subprojects/gst-editing-services/ges/ges-source.c
index 43b5f42707..862b23f778 100644
--- a/subprojects/gst-editing-services/ges/ges-source.c
+++ b/subprojects/gst-editing-services/ges/ges-source.c
@@ -40,6 +40,9 @@ struct _GESSourcePrivate
   GstElement *last_converter;
   GstPad *ghostpad;
 
+  GList *sub_element_probes;
+  GMutex sub_element_lock;
+
   gboolean is_rendering_smartly;
 };
 
@@ -77,6 +80,41 @@ link_elements (GstElement * bin, GPtrArray * elements)
   return prev;
 }
 
+typedef struct
+{
+  GstPad *pad;
+  gulong probe_id;
+} ProbeData;
+
+static GstPadProbeReturn
+pad_probe_cb (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
+{
+  return GST_PAD_PROBE_OK;
+}
+
+static void
+_release_probe_data (ProbeData * pdata)
+{
+  gst_pad_remove_probe (pdata->pad, pdata->probe_id);
+  gst_object_unref (pdata->pad);
+  g_free (pdata);
+}
+
+static void
+_no_more_pads_cb (GstElement * element, GESSource * self)
+{
+  GESSourcePrivate *priv = self->priv;
+
+  GST_DEBUG_OBJECT (self,
+      "Unblocking after no more pads from sub_element %" GST_PTR_FORMAT,
+      element);
+  g_mutex_lock (&priv->sub_element_lock);
+  g_list_free_full (priv->sub_element_probes,
+      (GDestroyNotify) _release_probe_data);
+  priv->sub_element_probes = NULL;
+  g_mutex_unlock (&priv->sub_element_lock);
+}
+
 static void
 _set_ghost_pad_target (GESSource * self, GstPad * srcpad, GstElement * element)
 {
@@ -133,6 +171,26 @@ _set_ghost_pad_target (GESSource * self, GstPad * srcpad, GstElement * element)
   }
 }
 
+static void
+_pad_added_cb (GstElement * element, GstPad * srcpad, GESSource * self)
+{
+  GESSourcePrivate *priv = self->priv;
+  ProbeData *pdata = g_new0 (ProbeData, 1);
+
+  GST_LOG_OBJECT (self, "blocking sub_element srcpad %" GST_PTR_FORMAT, srcpad);
+
+  pdata->probe_id = gst_pad_add_probe (srcpad,
+      GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM,
+      (GstPadProbeCallback) pad_probe_cb, NULL, NULL);
+  pdata->pad = gst_object_ref (srcpad);
+
+  g_mutex_lock (&priv->sub_element_lock);
+  priv->sub_element_probes = g_list_append (priv->sub_element_probes, pdata);
+  g_mutex_unlock (&priv->sub_element_lock);
+
+  _set_ghost_pad_target (self, srcpad, element);
+}
+
 /* @elements: (transfer-full) */
 GstElement *
 ges_source_create_topbin (GESSource * source, const gchar * bin_name,
@@ -173,8 +231,10 @@ ges_source_create_topbin (GESSource * source, const gchar * bin_name,
     gst_object_unref (sub_srcpad);
   } else {
     GST_INFO_OBJECT (source, "Waiting for pad added");
-    g_signal_connect_swapped (sub_element, "pad-added",
-        G_CALLBACK (_set_ghost_pad_target), source);
+    g_signal_connect (sub_element, "pad-added",
+        G_CALLBACK (_pad_added_cb), source);
+    g_signal_connect (sub_element, "no-more-pads",
+        G_CALLBACK (_no_more_pads_cb), source);
   }
   g_ptr_array_free (elements, TRUE);
 
@@ -215,6 +275,9 @@ ges_source_dispose (GObject * object)
   gst_clear_object (&priv->last_converter);
   gst_clear_object (&priv->topbin);
   gst_clear_object (&priv->ghostpad);
+  g_list_free_full (priv->sub_element_probes,
+      (GDestroyNotify) _release_probe_data);
+  g_mutex_clear (&priv->sub_element_lock);
 
   G_OBJECT_CLASS (ges_source_parent_class)->dispose (object);
 }
@@ -236,4 +299,5 @@ static void
 ges_source_init (GESSource * self)
 {
   self->priv = ges_source_get_instance_private (self);
+  g_mutex_init (&self->priv->sub_element_lock);
 }