decodebin2: Handle flushing with multiple decode groups

When an upstream element wants to flush downstream, we need to take
all chains/groups into consideration.

To that effect, when a FLUSH_START event is seen, after having it
sent downstream we mark all those chains/groups as "drained" (as if
they had seen a EOS event on the endpads).

When a FLUSH_STOP event is received, we check if we need to switch groups.
This is done by checking if there are next groups. If so, we will switch
over to the latest next_group. The actual switch will be done when
that group is blocked.

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

1 files changed, 116 insertions, 0 deletions

diff --git a/gst/playback/gstdecodebin2.c b/gst/playback/gstdecodebin2.c
index bb1909570..5bec1bf97 100644
--- a/gst/playback/gstdecodebin2.c
+++ b/gst/playback/gstdecodebin2.c
@@ -309,6 +309,8 @@ static GstCaps *gst_decode_bin_get_caps (GstDecodeBin * dbin);
 static void caps_notify_cb (GstPad * pad, GParamSpec * unused,
     GstDecodeChain * chain);
 
+static void flush_chain (GstDecodeChain * chain, gboolean flushing);
+static void flush_group (GstDecodeGroup * group, gboolean flushing);
 static GstPad *find_sink_pad (GstElement * element);
 static GstStateChangeReturn gst_decode_bin_change_state (GstElement * element,
     GstStateChange transition);
@@ -2015,6 +2017,54 @@ is_simple_demuxer_factory (GstElementFactory * factory)
   return FALSE;
 }
 
+static GstPadProbeReturn
+demuxer_source_pad_probe (GstPad * pad, GstPadProbeInfo * info,
+    gpointer user_data)
+{
+  GstEvent *event = GST_PAD_PROBE_INFO_EVENT (info);
+  GstDecodeGroup *group = (GstDecodeGroup *) user_data;
+  GstDecodeChain *parent_chain = group->parent;
+
+  GST_DEBUG_OBJECT (pad, "Saw event %s", GST_EVENT_TYPE_NAME (event));
+  /* Check if we are the active group, if not we need to proxy the flush
+   * events to the other groups (of which at least one is exposed, ensuring
+   * flushing properly propagates downstream of decodebin */
+  if (parent_chain->active_group == group)
+    return GST_PAD_PROBE_OK;
+
+  switch (GST_EVENT_TYPE (event)) {
+    case GST_EVENT_FLUSH_START:
+    case GST_EVENT_FLUSH_STOP:
+    {
+      GList *tmp;
+      GST_DEBUG_OBJECT (pad, "Proxying flush events to inactive groups");
+      /* Proxy to active group */
+      for (tmp = parent_chain->active_group->reqpads; tmp; tmp = tmp->next) {
+        GstPad *reqpad = (GstPad *) tmp->data;
+        gst_pad_send_event (reqpad, gst_event_ref (event));
+      }
+      /* Proxy to other non-active groups (except ourself) */
+      for (tmp = parent_chain->next_groups; tmp; tmp = tmp->next) {
+        GList *tmp2;
+        GstDecodeGroup *tmpgroup = (GstDecodeGroup *) tmp->data;
+        if (tmpgroup != group) {
+          for (tmp2 = tmpgroup->reqpads; tmp; tmp = tmp->next) {
+            GstPad *reqpad = (GstPad *) tmp2->data;
+            gst_pad_send_event (reqpad, gst_event_ref (event));
+          }
+        }
+      }
+      flush_chain (parent_chain,
+          GST_EVENT_TYPE (event) == GST_EVENT_FLUSH_START);
+    }
+      break;
+    default:
+      break;
+  }
+
+  return GST_PAD_PROBE_OK;
+}
+
 /* connect_pad:
  *
  * Try to connect the given pad to an element created from one of the factories,
@@ -2048,6 +2098,10 @@ connect_pad (GstDecodeBin * dbin, GstElement * src, GstDecodePad * dpad,
         "is a demuxer, connecting the pad through multiqueue '%s'",
         GST_OBJECT_NAME (chain->parent->multiqueue));
 
+    /* Set a flush-start/-stop probe on the downstream events */
+    gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_EVENT_FLUSH,
+        demuxer_source_pad_probe, chain->parent, NULL);
+
     decode_pad_set_target (dpad, NULL);
     if (!(mqpad = gst_decode_group_control_demuxer_pad (chain->parent, pad)))
       goto beach;
@@ -3846,6 +3900,68 @@ out:
   return complete;
 }
 
+/* Flushing group/chains */
+static void
+flush_group (GstDecodeGroup * group, gboolean flushing)
+{
+  GList *tmp;
+
+  GST_DEBUG ("group %p flushing:%d", group, flushing);
+
+  if (group->drained == flushing)
+    return;
+  for (tmp = group->children; tmp; tmp = tmp->next) {
+    GstDecodeChain *chain = (GstDecodeChain *) tmp->data;
+    flush_chain (chain, flushing);
+  }
+  GST_DEBUG ("Setting group %p to drained:%d", group, flushing);
+  group->drained = flushing;
+}
+
+static void
+flush_chain (GstDecodeChain * chain, gboolean flushing)
+{
+  GList *tmp;
+  GstDecodeBin *dbin = chain->dbin;
+
+  GST_DEBUG_OBJECT (dbin, "chain %p (pad %s:%s) flushing:%d", chain,
+      GST_DEBUG_PAD_NAME (chain->pad), flushing);
+  if (chain->drained == flushing)
+    return;
+  /* if unflushing, check if we should switch to last group */
+  if (flushing == FALSE && chain->next_groups) {
+    GstDecodeGroup *target_group =
+        (GstDecodeGroup *) g_list_last (chain->next_groups)->data;
+    gst_decode_chain_start_free_hidden_groups_thread (chain);
+    /* Hide active group (we're sure it's not that one we'll be using) */
+    GST_DEBUG_OBJECT (dbin, "Switching from active group %p to group %p",
+        chain->active_group, target_group);
+    gst_decode_group_hide (chain->active_group);
+    chain->old_groups = g_list_prepend (chain->old_groups, chain->active_group);
+    chain->active_group = target_group;
+    /* Hide all groups but the target_group */
+    for (tmp = chain->next_groups; tmp; tmp = tmp->next) {
+      GstDecodeGroup *group = (GstDecodeGroup *) tmp->data;
+      if (group != target_group) {
+        gst_decode_group_hide (group);
+        chain->old_groups = g_list_prepend (chain->old_groups, group);
+      }
+    }
+    /* Clear next groups */
+    g_list_free (chain->next_groups);
+    chain->next_groups = NULL;
+  }
+  /* Mark all groups as flushing */
+  if (chain->active_group)
+    flush_group (chain->active_group, flushing);
+  for (tmp = chain->next_groups; tmp; tmp = tmp->next) {
+    GstDecodeGroup *group = (GstDecodeGroup *) tmp->data;
+    flush_group (group, flushing);
+  }
+  GST_DEBUG ("Setting chain %p to drained:%d", chain, flushing);
+  chain->drained = flushing;
+}
+
 static gboolean
 drain_and_switch_chains (GstDecodeChain * chain, GstDecodePad * drainpad,
     gboolean * last_group, gboolean * drained, gboolean * switched);