From: Sebastian Dröge Date: Wed, 12 Feb 2020 11:11:43 +0000 (+0200) Subject: appsrc/appsink: Make setting/replacing callbacks thread-safe X-Git-Url: http://review.tizen.org/git/?a=commitdiff_plain;h=9a55945c0bbb43ee8cd336e90c31edeb35614720;p=platform%2Fupstream%2Fgst-plugins-base.git appsrc/appsink: Make setting/replacing callbacks thread-safe Previously we would simply use them without any locking at all, while using the object lock for setting them. Nothing prevented new callbacks to be set in the meantime, potentially calling a callback with already freed user_data. To prevent this move the callbacks into a reference counted struct and use the appsrc/appsink mutex to protect access to it, which is used in all functions calling the callbacks already anyway. Fixes https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/issues/729 --- diff --git a/gst-libs/gst/app/gstappsink.c b/gst-libs/gst/app/gstappsink.c index 695d871..964b2c1 100644 --- a/gst-libs/gst/app/gstappsink.c +++ b/gst-libs/gst/app/gstappsink.c @@ -80,6 +80,34 @@ typedef enum APP_WAITING = 1 << 1, /* application thread is waiting for streaming thread */ } GstAppSinkWaitStatus; +typedef struct +{ + GstAppSinkCallbacks callbacks; + gpointer user_data; + GDestroyNotify destroy_notify; + gint ref_count; +} Callbacks; + +static Callbacks * +callbacks_ref (Callbacks * callbacks) +{ + g_atomic_int_inc (&callbacks->ref_count); + + return callbacks; +} + +static void +callbacks_unref (Callbacks * callbacks) +{ + if (!g_atomic_int_dec_and_test (&callbacks->ref_count)) + return; + + if (callbacks->destroy_notify) + callbacks->destroy_notify (callbacks->user_data); + + g_free (callbacks); +} + struct _GstAppSinkPrivate { GstCaps *caps; @@ -104,9 +132,7 @@ struct _GstAppSinkPrivate gboolean is_eos; gboolean buffer_lists_supported; - GstAppSinkCallbacks callbacks; - gpointer user_data; - GDestroyNotify notify; + Callbacks *callbacks; GstSample *sample; }; @@ -476,21 +502,18 @@ gst_app_sink_dispose (GObject * obj) GstAppSink *appsink = GST_APP_SINK_CAST (obj); GstAppSinkPrivate *priv = appsink->priv; GstMiniObject *queue_obj; + Callbacks *callbacks = NULL; GST_OBJECT_LOCK (appsink); if (priv->caps) { gst_caps_unref (priv->caps); priv->caps = NULL; } - if (priv->notify) { - priv->notify (priv->user_data); - } - priv->user_data = NULL; - priv->notify = NULL; - GST_OBJECT_UNLOCK (appsink); g_mutex_lock (&priv->mutex); + if (priv->callbacks) + callbacks = g_steal_pointer (&priv->callbacks); while ((queue_obj = gst_queue_array_pop_head (priv->queue))) gst_mini_object_unref (queue_obj); gst_buffer_replace (&priv->preroll_buffer, NULL); @@ -502,6 +525,8 @@ gst_app_sink_dispose (GObject * obj) } g_mutex_unlock (&priv->mutex); + g_clear_pointer (&callbacks, callbacks_unref); + G_OBJECT_CLASS (parent_class)->dispose (obj); } @@ -715,6 +740,7 @@ gst_app_sink_event (GstBaseSink * sink, GstEvent * event) break; case GST_EVENT_EOS:{ gboolean emit = TRUE; + Callbacks *callbacks = NULL; g_mutex_lock (&priv->mutex); GST_DEBUG_OBJECT (appsink, "receiving EOS"); @@ -748,14 +774,19 @@ gst_app_sink_event (GstBaseSink * sink, GstEvent * event) } if (priv->flushing) emit = FALSE; + + if (emit && priv->callbacks) + callbacks = callbacks_ref (priv->callbacks); g_mutex_unlock (&priv->mutex); if (emit) { /* emit EOS now */ - if (priv->callbacks.eos) - priv->callbacks.eos (appsink, priv->user_data); + if (callbacks && callbacks->callbacks.eos) + callbacks->callbacks.eos (appsink, callbacks->user_data); else g_signal_emit (appsink, gst_app_sink_signals[SIGNAL_EOS], 0); + + g_clear_pointer (&callbacks, callbacks_unref); } break; @@ -784,6 +815,7 @@ gst_app_sink_preroll (GstBaseSink * psink, GstBuffer * buffer) GstAppSink *appsink = GST_APP_SINK_CAST (psink); GstAppSinkPrivate *priv = appsink->priv; gboolean emit; + Callbacks *callbacks = NULL; g_mutex_lock (&priv->mutex); if (priv->flushing) @@ -796,10 +828,12 @@ gst_app_sink_preroll (GstBaseSink * psink, GstBuffer * buffer) g_cond_signal (&priv->cond); emit = priv->emit_signals; + if (priv->callbacks) + callbacks = callbacks_ref (priv->callbacks); g_mutex_unlock (&priv->mutex); - if (priv->callbacks.new_preroll) { - res = priv->callbacks.new_preroll (appsink, priv->user_data); + if (callbacks && callbacks->callbacks.new_preroll) { + res = callbacks->callbacks.new_preroll (appsink, callbacks->user_data); } else { res = GST_FLOW_OK; if (emit) @@ -807,6 +841,8 @@ gst_app_sink_preroll (GstBaseSink * psink, GstBuffer * buffer) &res); } + g_clear_pointer (&callbacks, callbacks_unref); + return res; flushing: @@ -870,6 +906,7 @@ gst_app_sink_render_common (GstBaseSink * psink, GstMiniObject * data, GstAppSink *appsink = GST_APP_SINK_CAST (psink); GstAppSinkPrivate *priv = appsink->priv; gboolean emit; + Callbacks *callbacks = NULL; restart: g_mutex_lock (&priv->mutex); @@ -929,15 +966,19 @@ restart: g_cond_signal (&priv->cond); emit = priv->emit_signals; + if (priv->callbacks) + callbacks = callbacks_ref (priv->callbacks); g_mutex_unlock (&priv->mutex); - if (priv->callbacks.new_sample) { - ret = priv->callbacks.new_sample (appsink, priv->user_data); + if (callbacks && callbacks->callbacks.new_sample) { + ret = callbacks->callbacks.new_sample (appsink, callbacks->user_data); } else { ret = GST_FLOW_OK; if (emit) g_signal_emit (appsink, gst_app_sink_signals[SIGNAL_NEW_SAMPLE], 0, &ret); } + g_clear_pointer (&callbacks, callbacks_unref); + return ret; flushing: @@ -1721,12 +1762,15 @@ not_started: * * If callbacks are installed, no signals will be emitted for performance * reasons. + * + * Before 1.16.3 it was not possible to change the callbacks in a thread-safe + * way. */ void gst_app_sink_set_callbacks (GstAppSink * appsink, GstAppSinkCallbacks * callbacks, gpointer user_data, GDestroyNotify notify) { - GDestroyNotify old_notify; + Callbacks *old_callbacks, *new_callbacks = NULL; GstAppSinkPrivate *priv; g_return_if_fail (GST_IS_APP_SINK (appsink)); @@ -1734,26 +1778,20 @@ gst_app_sink_set_callbacks (GstAppSink * appsink, priv = appsink->priv; - GST_OBJECT_LOCK (appsink); - old_notify = priv->notify; - - if (old_notify) { - gpointer old_data; - - old_data = priv->user_data; - - priv->user_data = NULL; - priv->notify = NULL; - GST_OBJECT_UNLOCK (appsink); + if (callbacks) { + new_callbacks = g_new0 (Callbacks, 1); + new_callbacks->callbacks = *callbacks; + new_callbacks->user_data = user_data; + new_callbacks->destroy_notify = notify; + new_callbacks->ref_count = 1; + } - old_notify (old_data); + g_mutex_lock (&priv->mutex); + old_callbacks = g_steal_pointer (&priv->callbacks); + priv->callbacks = g_steal_pointer (&new_callbacks); + g_mutex_unlock (&priv->mutex); - GST_OBJECT_LOCK (appsink); - } - priv->callbacks = *callbacks; - priv->user_data = user_data; - priv->notify = notify; - GST_OBJECT_UNLOCK (appsink); + g_clear_pointer (&old_callbacks, callbacks_unref); } /*** GSTURIHANDLER INTERFACE *************************************************/ diff --git a/gst-libs/gst/app/gstappsrc.c b/gst-libs/gst/app/gstappsrc.c index 63395db..ed92e5a 100644 --- a/gst-libs/gst/app/gstappsrc.c +++ b/gst-libs/gst/app/gstappsrc.c @@ -108,6 +108,35 @@ typedef enum APP_WAITING = 1 << 1, /* application thread is waiting for streaming thread */ } GstAppSrcWaitStatus; +typedef struct +{ + GstAppSrcCallbacks callbacks; + gpointer user_data; + GDestroyNotify destroy_notify; + gint ref_count; +} Callbacks; + +static Callbacks * +callbacks_ref (Callbacks * callbacks) +{ + g_atomic_int_inc (&callbacks->ref_count); + + return callbacks; +} + +static void +callbacks_unref (Callbacks * callbacks) +{ + if (!g_atomic_int_dec_and_test (&callbacks->ref_count)) + return; + + if (callbacks->destroy_notify) + callbacks->destroy_notify (callbacks->user_data); + + g_free (callbacks); +} + + struct _GstAppSrcPrivate { GCond cond; @@ -138,9 +167,7 @@ struct _GstAppSrcPrivate gboolean emit_signals; guint min_percent; - GstAppSrcCallbacks callbacks; - gpointer user_data; - GDestroyNotify notify; + Callbacks *callbacks; }; GST_DEBUG_CATEGORY_STATIC (app_src_debug); @@ -621,6 +648,7 @@ gst_app_src_dispose (GObject * obj) { GstAppSrc *appsrc = GST_APP_SRC_CAST (obj); GstAppSrcPrivate *priv = appsrc->priv; + Callbacks *callbacks = NULL; GST_OBJECT_LOCK (appsrc); if (priv->current_caps) { @@ -631,18 +659,16 @@ gst_app_src_dispose (GObject * obj) gst_caps_unref (priv->last_caps); priv->last_caps = NULL; } - if (priv->notify) { - priv->notify (priv->user_data); - } - priv->user_data = NULL; - priv->notify = NULL; - GST_OBJECT_UNLOCK (appsrc); g_mutex_lock (&priv->mutex); + if (priv->callbacks) + callbacks = g_steal_pointer (&priv->callbacks); gst_app_src_flush_queued (appsrc, FALSE); g_mutex_unlock (&priv->mutex); + g_clear_pointer (&callbacks, callbacks_unref); + G_OBJECT_CLASS (parent_class)->dispose (obj); } @@ -997,6 +1023,8 @@ gst_app_src_do_seek (GstBaseSrc * src, GstSegment * segment) GstAppSrcPrivate *priv = appsrc->priv; gint64 desired_position; gboolean res = FALSE; + gboolean emit; + Callbacks *callbacks = NULL; desired_position = segment->position; @@ -1007,20 +1035,23 @@ gst_app_src_do_seek (GstBaseSrc * src, GstSegment * segment) GST_DEBUG_OBJECT (appsrc, "seeking to %" G_GINT64_FORMAT ", format %s", desired_position, gst_format_get_name (segment->format)); - if (priv->callbacks.seek_data) - res = priv->callbacks.seek_data (appsrc, desired_position, priv->user_data); - else { - gboolean emit; - - g_mutex_lock (&priv->mutex); - emit = priv->emit_signals; - g_mutex_unlock (&priv->mutex); + g_mutex_lock (&priv->mutex); + emit = priv->emit_signals; + if (priv->callbacks) + callbacks = callbacks_ref (priv->callbacks); + g_mutex_unlock (&priv->mutex); - if (emit) - g_signal_emit (appsrc, gst_app_src_signals[SIGNAL_SEEK_DATA], 0, - desired_position, &res); + if (callbacks && callbacks->callbacks.seek_data) { + res = + callbacks->callbacks.seek_data (appsrc, desired_position, + callbacks->user_data); + } else if (emit) { + g_signal_emit (appsrc, gst_app_src_signals[SIGNAL_SEEK_DATA], 0, + desired_position, &res); } + g_clear_pointer (&callbacks, callbacks_unref); + if (res) { GST_DEBUG_OBJECT (appsrc, "flushing queue"); g_mutex_lock (&priv->mutex); @@ -1041,20 +1072,25 @@ gst_app_src_emit_seek (GstAppSrc * appsrc, guint64 offset) gboolean res = FALSE; gboolean emit; GstAppSrcPrivate *priv = appsrc->priv; + Callbacks *callbacks = NULL; emit = priv->emit_signals; + if (priv->callbacks) + callbacks = callbacks_ref (priv->callbacks); g_mutex_unlock (&priv->mutex); GST_DEBUG_OBJECT (appsrc, "we are at %" G_GINT64_FORMAT ", seek to %" G_GINT64_FORMAT, priv->offset, offset); - if (priv->callbacks.seek_data) - res = priv->callbacks.seek_data (appsrc, offset, priv->user_data); + if (callbacks && callbacks->callbacks.seek_data) + res = callbacks->callbacks.seek_data (appsrc, offset, callbacks->user_data); else if (emit) g_signal_emit (appsrc, gst_app_src_signals[SIGNAL_SEEK_DATA], 0, offset, &res); + g_clear_pointer (&callbacks, callbacks_unref); + g_mutex_lock (&priv->mutex); return res; @@ -1067,17 +1103,22 @@ gst_app_src_emit_need_data (GstAppSrc * appsrc, guint size) { gboolean emit; GstAppSrcPrivate *priv = appsrc->priv; + Callbacks *callbacks = NULL; emit = priv->emit_signals; + if (priv->callbacks) + callbacks = callbacks_ref (priv->callbacks); g_mutex_unlock (&priv->mutex); /* we have no data, we need some. We fire the signal with the size hint. */ - if (priv->callbacks.need_data) - priv->callbacks.need_data (appsrc, size, priv->user_data); + if (callbacks && callbacks->callbacks.need_data) + callbacks->callbacks.need_data (appsrc, size, callbacks->user_data); else if (emit) g_signal_emit (appsrc, gst_app_src_signals[SIGNAL_NEED_DATA], 0, size, NULL); + g_clear_pointer (&callbacks, callbacks_unref); + g_mutex_lock (&priv->mutex); /* we can be flushing now because we released the lock */ } @@ -1817,18 +1858,23 @@ gst_app_src_push_internal (GstAppSrc * appsrc, GstBuffer * buffer, priv->queued_bytes, priv->max_bytes); if (first) { + Callbacks *callbacks = NULL; gboolean emit; emit = priv->emit_signals; + if (priv->callbacks) + callbacks = callbacks_ref (priv->callbacks); /* only signal on the first push */ g_mutex_unlock (&priv->mutex); - if (priv->callbacks.enough_data) - priv->callbacks.enough_data (appsrc, priv->user_data); + if (callbacks && callbacks->callbacks.enough_data) + callbacks->callbacks.enough_data (appsrc, callbacks->user_data); else if (emit) g_signal_emit (appsrc, gst_app_src_signals[SIGNAL_ENOUGH_DATA], 0, NULL); + g_clear_pointer (&callbacks, callbacks_unref); + g_mutex_lock (&priv->mutex); /* continue to check for flushing/eos after releasing the lock */ first = FALSE; @@ -2081,12 +2127,15 @@ flushing: * * If callbacks are installed, no signals will be emitted for performance * reasons. + * + * Before 1.16.3 it was not possible to change the callbacks in a thread-safe + * way. */ void gst_app_src_set_callbacks (GstAppSrc * appsrc, GstAppSrcCallbacks * callbacks, gpointer user_data, GDestroyNotify notify) { - GDestroyNotify old_notify; + Callbacks *old_callbacks, *new_callbacks = NULL; GstAppSrcPrivate *priv; g_return_if_fail (GST_IS_APP_SRC (appsrc)); @@ -2094,26 +2143,20 @@ gst_app_src_set_callbacks (GstAppSrc * appsrc, priv = appsrc->priv; - GST_OBJECT_LOCK (appsrc); - old_notify = priv->notify; - - if (old_notify) { - gpointer old_data; - - old_data = priv->user_data; - - priv->user_data = NULL; - priv->notify = NULL; - GST_OBJECT_UNLOCK (appsrc); + if (callbacks) { + new_callbacks = g_new0 (Callbacks, 1); + new_callbacks->callbacks = *callbacks; + new_callbacks->user_data = user_data; + new_callbacks->destroy_notify = notify; + new_callbacks->ref_count = 1; + } - old_notify (old_data); + g_mutex_lock (&priv->mutex); + old_callbacks = g_steal_pointer (&priv->callbacks); + priv->callbacks = g_steal_pointer (&new_callbacks); + g_mutex_unlock (&priv->mutex); - GST_OBJECT_LOCK (appsrc); - } - priv->callbacks = *callbacks; - priv->user_data = user_data; - priv->notify = notify; - GST_OBJECT_UNLOCK (appsrc); + g_clear_pointer (&old_callbacks, callbacks_unref); } /*** GSTURIHANDLER INTERFACE *************************************************/