alphacombine: De-couple flush-start/stop events handling

There is no guarantee that any FLUSH_STOP event is preceded by a FLUSH_START. The element now stops flushing once it has received a FLUSH_STOP on all its sink pads.

Fixes #4174 (closed)

subprojects/gst-plugins-bad/gst/codecalpha/gstalphacombine.c

@@ -113,6 +113,11 @@ struct _GstAlphaCombine
      done flushing. Protected by buffer_lock */
   guint flush_stops;
 
+  /* sink pad being blocked while waiting for the other pad to receive
+   * FLUSH_STOP. Protected by buffer_lock */
+  GstPad *blocked_pad;
+  gulong pad_block_id;
+
   GstVideoInfo sink_vinfo;
   GstVideoInfo alpha_vinfo;
   GstVideoFormat src_format;
@@ -160,16 +165,42 @@ gst_alpha_combine_unlock (GstAlphaCombine * self)
   g_mutex_unlock (&self->buffer_lock);
 }
 
+static GstPadProbeReturn
+pad_blocked (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
+{
+  GstAlphaCombine *self = user_data;
+
+  GST_DEBUG_OBJECT (self, "pad %s:%s blocked", GST_DEBUG_PAD_NAME (pad));
+  return GST_PAD_PROBE_OK;
+}
+
 static void
-gst_alpha_combine_unlock_stop (GstAlphaCombine * self)
+gst_alpha_combine_unlock_stop (GstAlphaCombine * self, GstPad * pad)
 {
   g_mutex_lock (&self->buffer_lock);
   self->flush_stops++;
 
   /* Keep flushing until we have received a FLUSH_STOP on every sink pad. */
-  if (self->flush_stops < 2) {
-    g_mutex_unlock (&self->buffer_lock);
-    return;
+  if (self->flush_stops >= 1 && pad) {
+    if (self->flush_stops == 2) {
+      GST_DEBUG_OBJECT (self,
+          "Both sink pads received FLUSH_STOP, unblocking them");
+      g_assert (self->pad_block_id);
+      g_assert (self->blocked_pad);
+      gst_pad_remove_probe (self->blocked_pad, self->pad_block_id);
+      self->pad_block_id = 0;
+      self->blocked_pad = NULL;
+    } else {
+      GST_DEBUG_OBJECT (pad, "FLUSH_STOP received, blocking");
+      g_assert (!self->pad_block_id);
+      self->pad_block_id =
+          gst_pad_add_probe (pad,
+          GST_PAD_PROBE_TYPE_BLOCK | GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM,
+          pad_blocked, self, NULL);
+      self->blocked_pad = pad;
+      g_mutex_unlock (&self->buffer_lock);
+      return;
+    }
   }
 
   self->flushing = FALSE;
@@ -501,7 +532,7 @@ gst_alpha_combine_sink_event (GstPad * pad, GstObject * object,
       gst_alpha_combine_unlock (self);
       break;
     case GST_EVENT_FLUSH_STOP:
-      gst_alpha_combine_unlock_stop (self);
+      gst_alpha_combine_unlock_stop (self, pad);
       break;
     case GST_EVENT_CAPS:
     {
@@ -532,7 +563,7 @@ gst_alpha_combine_alpha_event (GstPad * pad, GstObject * object,
       gst_alpha_combine_unlock (self);
       break;
     case GST_EVENT_FLUSH_STOP:
-      gst_alpha_combine_unlock_stop (self);
+      gst_alpha_combine_unlock_stop (self, pad);
       gst_alpha_combine_reset (self);
       break;
     case GST_EVENT_CAPS:
@@ -596,7 +627,7 @@ gst_alpha_combine_change_state (GstElement * element, GstStateChange transition)
 
   switch (transition) {
     case GST_STATE_CHANGE_READY_TO_PAUSED:
-      gst_alpha_combine_unlock_stop (self);
+      gst_alpha_combine_unlock_stop (self, NULL);
       break;
     case GST_STATE_CHANGE_PAUSED_TO_READY:
       gst_alpha_combine_unlock (self);
@@ -632,6 +663,12 @@ gst_alpha_combine_dispose (GObject * object)
   g_clear_object (&self->alpha_pad);
   g_clear_object (&self->src_pad);
 
+  if (G_UNLIKELY (self->blocked_pad)) {
+    g_assert (self->pad_block_id);
+    gst_pad_remove_probe (self->blocked_pad, self->pad_block_id);
+  }
+  g_clear_object (&self->blocked_pad);
+
   G_OBJECT_CLASS (parent_class)->dispose (object);
 }
 
@@ -680,6 +717,8 @@ gst_alpha_combine_init (GstAlphaCombine * self)
   self->src_pad = gst_element_get_static_pad (GST_ELEMENT (self), "src");
   self->flushing = FALSE;
   self->flush_stops = 0;
+  self->pad_block_id = 0;
+  self->blocked_pad = NULL;
 
   g_mutex_init (&self->buffer_lock);
   g_cond_init (&self->buffer_cond);