pad: Fix for multiple blocking probes interaction.

Change the way the marshalled flag in the internal ProbeMarshall state
is handled when iterating over pad probes so that it only counts
probes that still exist and would be called when retrying.

This improves the way that removing a blocking probe works when
there are multiple blocking probes for different conditions (data vs
events for example).

As a side-effect, probes aren't put into the the called_probes array
unless they actually match the current probe type and would be called,
potentially reducing the number of hooks that get stored in the
called_probes array, and the cost of the looping check on retries.

Fixes https://gitlab.freedesktop.org/gstreamer/gstreamer/-/issues/658

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

diff --git a/gst/gstpad.c b/gst/gstpad.c
index 6ae914b..93a7a65 100644 (file)
--- a/gst/gstpad.c
+++ b/gst/gstpad.c
@@ -3533,15 +3533,13 @@ gst_pad_query_default (GstPad * pad, GstObject * parent, GstQuery * query)
 
 #define N_STACK_ALLOCATE_PROBES (16)
 
-static void
-probe_hook_marshal (GHook * hook, ProbeMarshall * data)
+/* A helper that checks if a probe was already
+ * in the called_probes list, and adds it if
+ * not. Used to avoid calling probes a 2nd time when
+ * looping again after probe removal */
+static gboolean
+check_probe_already_called (GHook * hook, ProbeMarshall * data)
 {
-  GstPad *pad = data->pad;
-  GstPadProbeInfo *info = data->info;
-  GstPadProbeType type, flags;
-  GstPadProbeCallback callback;
-  GstPadProbeReturn ret;
-  gpointer original_data;
   guint i;
 
   /* if we have called this callback, do nothing. But only check
@@ -3549,9 +3547,7 @@ probe_hook_marshal (GHook * hook, ProbeMarshall * data)
   if (data->retry) {
     for (i = 0; i < data->n_called_probes; i++) {
       if (data->called_probes[i] == hook->hook_id) {
-        GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad,
-            "hook %lu already called", hook->hook_id);
-        return;
+        return TRUE;
       }
     }
   }
@@ -3573,6 +3569,20 @@ probe_hook_marshal (GHook * hook, ProbeMarshall * data)
   }
   data->called_probes[data->n_called_probes++] = hook->hook_id;
 
+  /* This probe was not alraedy called */
+  return FALSE;
+}
+
+static void
+probe_hook_marshal (GHook * hook, ProbeMarshall * data)
+{
+  GstPad *pad = data->pad;
+  GstPadProbeInfo *info = data->info;
+  GstPadProbeType type, flags;
+  GstPadProbeCallback callback;
+  GstPadProbeReturn ret;
+  gpointer original_data;
+
   flags = hook->flags >> G_HOOK_FLAG_USER_SHIFT;
   type = info->type;
   original_data = info->data;
@@ -3618,14 +3628,26 @@ probe_hook_marshal (GHook * hook, ProbeMarshall * data)
       (flags & GST_PAD_PROBE_TYPE_EVENT_FLUSH & type) == 0)
     goto no_match;
 
+  if (check_probe_already_called (hook, data)) {
+    /* Reset marshalled = TRUE here, because the probe
+     * was already called and set it the first time around,
+     * and we may want to keep blocking on it.
+     *
+     * https://gitlab.freedesktop.org/gstreamer/gstreamer/-/issues/658
+     */
+    data->marshalled = TRUE;
+    goto already_called;
+  }
+
   GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad,
       "hook %lu with flags 0x%08x matches", hook->hook_id, flags);
 
-  data->marshalled = TRUE;
-
   callback = (GstPadProbeCallback) hook->func;
-  if (callback == NULL)
+  if (callback == NULL) {
+    /* No callback is equivalent to just returning GST_PAD_PROBE_OK */
+    data->marshalled = TRUE;
     return;
+  }
 
   info->id = hook->hook_id;
 
@@ -3638,6 +3660,18 @@ probe_hook_marshal (GHook * hook, ProbeMarshall * data)
 
   GST_OBJECT_LOCK (pad);
 
+  /* If the probe callback asked for the
+   * probe to be removed, don't set the marshalled flag
+   * otherwise, you can get a case where you return
+   * GST_PAD_PROBE_REMOVE from a buffer probe and
+   * then the pad blocks anyway if there's any other
+   * blocking probes installed.
+   *
+   * https://gitlab.freedesktop.org/gstreamer/gstreamer/-/issues/658
+   */
+  if (ret != GST_PAD_PROBE_REMOVE)
+    data->marshalled = TRUE;
+
   if ((flags & GST_PAD_PROBE_TYPE_IDLE))
     pad->priv->idle_running--;
 
@@ -3686,6 +3720,12 @@ no_match:
         hook->hook_id, flags, info->type);
     return;
   }
+already_called:
+  {
+    GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad,
+        "hook %lu already called", hook->hook_id);
+    return;
+  }
 }
 
 /* a probe that does not take or return any data */
@@ -3774,7 +3814,6 @@ do_probe_callbacks (GstPad * pad, GstPadProbeInfo * info,
   data.info = info;
   data.pass = FALSE;
   data.handled = FALSE;
-  data.marshalled = FALSE;
   data.dropped = FALSE;
 
   /* We stack-allocate for N_STACK_ALLOCATE_PROBES hooks as a first step. If more are needed,
@@ -3797,6 +3836,10 @@ again:
   GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "do probes");
   cookie = pad->priv->probe_list_cookie;
 
+  /* Clear the marshalled flag before doing callbacks. Only if
+   * there are matching callbacks still will it get set */
+  data.marshalled = FALSE;
+
   g_hook_list_marshal (&pad->probes, TRUE,
       (GHookMarshaller) probe_hook_marshal, &data);