diff --git a/gst/gstpad.c b/gst/gstpad.c index b762465b8e..a7e1c37da9 100644 --- a/gst/gstpad.c +++ b/gst/gstpad.c @@ -140,6 +140,11 @@ struct _GstPadPrivate gint using; guint probe_list_cookie; guint probe_cookie; + + /* counter of how many idle probes are running directly from the add_probe + * call. Used to block any data flowing in the pad while the idle callback + * Doesn't finish its work */ + gint idle_running; }; typedef struct @@ -149,6 +154,8 @@ typedef struct } GstProbe; #define PROBE_COOKIE(h) (((GstProbe *)(h))->cookie) +#define GST_PAD_IS_RUNNING_IDLE_PROBE(p) \ + (((GstPad *)(p))->priv->idle_running > 0) typedef struct { @@ -1387,6 +1394,7 @@ gst_pad_add_probe (GstPad * pad, GstPadProbeType mask, /* Keep another ref, the callback could destroy the pad */ gst_object_ref (pad); + pad->priv->idle_running++; /* the pad is idle now, we can signal the idle callback now */ GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, @@ -1416,6 +1424,10 @@ gst_pad_add_probe (GstPad * pad, GstPadProbeType mask, GST_DEBUG_OBJECT (pad, "probe returned %d", ret); break; } + pad->priv->idle_running--; + if (pad->priv->idle_running == 0) { + GST_PAD_BLOCK_BROADCAST (pad); + } GST_OBJECT_UNLOCK (pad); gst_object_unref (pad); @@ -3401,6 +3413,38 @@ no_match: #define PROBE_PULL(pad,mask,data,offs,size,label) \ PROBE_FULL(pad, mask, data, offs, size, label); +static GstFlowReturn +do_pad_idle_probe_wait (GstPad * pad) +{ + while (GST_PAD_IS_RUNNING_IDLE_PROBE (pad)) { + GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, + "waiting idle probe to be removed"); + GST_OBJECT_FLAG_SET (pad, GST_PAD_FLAG_BLOCKING); + GST_PAD_BLOCK_WAIT (pad); + GST_OBJECT_FLAG_UNSET (pad, GST_PAD_FLAG_BLOCKING); + GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "We got unblocked"); + + if (G_UNLIKELY (GST_PAD_IS_FLUSHING (pad))) + return GST_FLOW_FLUSHING; + } + return GST_FLOW_OK; +} + +#define PROBE_TYPE_IS_SERIALIZED(i) \ + ( \ + ( \ + (((i)->type & (GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM | \ + GST_PAD_PROBE_TYPE_EVENT_FLUSH)) && \ + GST_EVENT_IS_SERIALIZED ((i)->data)) \ + ) || ( \ + (((i)->type & GST_PAD_PROBE_TYPE_QUERY_DOWNSTREAM) && \ + GST_QUERY_IS_SERIALIZED ((i)->data)) \ + ) || ( \ + ((i)->type & (GST_PAD_PROBE_TYPE_BUFFER | \ + GST_PAD_PROBE_TYPE_BUFFER_LIST)) \ + ) \ + ) + static GstFlowReturn do_probe_callbacks (GstPad * pad, GstPadProbeInfo * info, GstFlowReturn defaultval) @@ -3419,6 +3463,11 @@ do_probe_callbacks (GstPad * pad, GstPadProbeInfo * info, is_block = (info->type & GST_PAD_PROBE_TYPE_BLOCK) == GST_PAD_PROBE_TYPE_BLOCK; + if (is_block && PROBE_TYPE_IS_SERIALIZED (info)) { + if (do_pad_idle_probe_wait (pad) == GST_FLOW_FLUSHING) + goto flushing; + } + again: GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "do probes cookie %u", data.cookie);