From fe969e63917a4b868371f311349edfb31093d8a8 Mon Sep 17 00:00:00 2001 From: =?utf8?q?Ole=20Andr=C3=A9=20Vadla=20Ravn=C3=A5s?= Date: Fri, 15 Aug 2008 17:01:07 +0000 Subject: [PATCH] plugins/elements/gsttee.*: Protect pad_alloc with a new lock so that we can be sure that nothing is performing a pad_... MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Original commit message from CVS: Patch by: Ole André Vadla Ravnås * plugins/elements/gsttee.c: (gst_tee_finalize), (gst_tee_init), (gst_tee_request_new_pad), (gst_tee_release_pad), (gst_tee_find_buffer_alloc), (gst_tee_buffer_alloc): * plugins/elements/gsttee.h: Protect pad_alloc with a new lock so that we can be sure that nothing is performing a pad_alloc when removing the pad. Fixes #547835. * tests/check/elements/tee.c: (buffer_alloc_harness_setup), (buffer_alloc_harness_teardown), (app_thread_func), (final_sinkpad_bufferalloc), (GST_START_TEST), (tee_suite): Added testcase for shutdown race. --- ChangeLog | 16 ++++ plugins/elements/gsttee.c | 41 +++++++++- plugins/elements/gsttee.h | 3 + tests/check/elements/tee.c | 183 +++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 240 insertions(+), 3 deletions(-) diff --git a/ChangeLog b/ChangeLog index 0c4c3fa..4448380 100644 --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,19 @@ +2008-08-15 Wim Taymans + + Patch by: Ole André Vadla Ravnås + + * plugins/elements/gsttee.c: (gst_tee_finalize), (gst_tee_init), + (gst_tee_request_new_pad), (gst_tee_release_pad), + (gst_tee_find_buffer_alloc), (gst_tee_buffer_alloc): + * plugins/elements/gsttee.h: + Protect pad_alloc with a new lock so that we can be sure that nothing is + performing a pad_alloc when removing the pad. Fixes #547835. + + * tests/check/elements/tee.c: (buffer_alloc_harness_setup), + (buffer_alloc_harness_teardown), (app_thread_func), + (final_sinkpad_bufferalloc), (GST_START_TEST), (tee_suite): + Added testcase for shutdown race. + 2008-08-14 Thijs Vermeir * gst/gstpad.h: diff --git a/plugins/elements/gsttee.c b/plugins/elements/gsttee.c index a15d3e7..a689fc6 100644 --- a/plugins/elements/gsttee.c +++ b/plugins/elements/gsttee.c @@ -62,6 +62,10 @@ gst_tee_pull_mode_get_type (void) return type; } +/* lock to protect request pads from being removed while downstream */ +#define GST_TEE_DYN_LOCK(tee) g_mutex_lock ((tee)->dyn_lock) +#define GST_TEE_DYN_UNLOCK(tee) g_mutex_unlock ((tee)->dyn_lock) + #define DEFAULT_PROP_NUM_SRC_PADS 0 #define DEFAULT_PROP_HAS_SINK_LOOP FALSE #define DEFAULT_PROP_HAS_CHAIN TRUE @@ -97,6 +101,7 @@ typedef struct { gboolean pushed; GstFlowReturn result; + gboolean removed; } PushData; static GstPad *gst_tee_request_new_pad (GstElement * element, @@ -146,6 +151,8 @@ gst_tee_finalize (GObject * object) g_free (tee->last_message); + g_mutex_free (tee->dyn_lock); + G_OBJECT_CLASS (parent_class)->finalize (object); } @@ -197,6 +204,8 @@ gst_tee_class_init (GstTeeClass * klass) static void gst_tee_init (GstTee * tee, GstTeeClass * g_class) { + tee->dyn_lock = g_mutex_new (); + tee->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink"); tee->sink_mode = GST_ACTIVATE_NONE; @@ -242,6 +251,7 @@ gst_tee_request_new_pad (GstElement * element, GstPadTemplate * templ, data = g_new0 (PushData, 1); data->pushed = FALSE; data->result = GST_FLOW_NOT_LINKED; + data->removed = FALSE; g_object_set_qdata_full (G_OBJECT (srcpad), push_data, data, g_free); GST_OBJECT_UNLOCK (tee); @@ -292,6 +302,7 @@ static void gst_tee_release_pad (GstElement * element, GstPad * pad) { GstTee *tee; + PushData *data; tee = GST_TEE (element); @@ -302,9 +313,16 @@ gst_tee_release_pad (GstElement * element, GstPad * pad) tee->allocpad = NULL; GST_OBJECT_UNLOCK (tee); + /* wait for pending pad_alloc to finish */ + GST_TEE_DYN_LOCK (tee); + /* mark the pad as removed so that future pad_alloc fails with NOT_LINKED. */ + data = g_object_get_qdata (G_OBJECT (pad), push_data); + data->removed = TRUE; + gst_pad_set_active (pad, FALSE); gst_element_remove_pad (GST_ELEMENT_CAST (tee), pad); + GST_TEE_DYN_UNLOCK (tee); } static void @@ -373,7 +391,7 @@ gst_tee_get_property (GObject * object, guint prop_id, GValue * value, /* we have no previous source pad we can use to proxy the pad alloc. Loop over * the source pads, try to alloc a buffer on each one of them. Keep a reference * to the first pad that succeeds, we will be using it to alloc more buffers - * later. */ + * later. must be called with the OBJECT_LOCK on tee. */ static GstFlowReturn gst_tee_find_buffer_alloc (GstTee * tee, guint64 offset, guint size, GstCaps * caps, GstBuffer ** buf) @@ -390,13 +408,20 @@ retry: while (pads) { GstPad *pad; + PushData *data; pad = GST_PAD_CAST (pads->data); gst_object_ref (pad); GST_DEBUG_OBJECT (tee, "try alloc on pad %s:%s", GST_DEBUG_PAD_NAME (pad)); GST_OBJECT_UNLOCK (tee); - res = gst_pad_alloc_buffer (pad, offset, size, caps, buf); + GST_TEE_DYN_LOCK (tee); + data = g_object_get_qdata (G_OBJECT (pad), push_data); + if (!data->removed) + res = gst_pad_alloc_buffer (pad, offset, size, caps, buf); + else + res = GST_FLOW_NOT_LINKED; + GST_TEE_DYN_UNLOCK (tee); GST_DEBUG_OBJECT (tee, "got return value %d", res); @@ -409,6 +434,7 @@ retry: * need to unref the buffer */ if (res == GST_FLOW_OK) gst_buffer_unref (*buf); + *buf = NULL; goto retry; } if (res == GST_FLOW_OK) { @@ -439,6 +465,8 @@ gst_tee_buffer_alloc (GstPad * pad, guint64 offset, guint size, GST_OBJECT_LOCK (tee); if ((allocpad = tee->allocpad)) { + PushData *data; + /* if we had a previous pad we used for allocating a buffer, continue using * it. */ GST_DEBUG_OBJECT (tee, "using pad %s:%s for alloc", @@ -446,7 +474,14 @@ gst_tee_buffer_alloc (GstPad * pad, guint64 offset, guint size, gst_object_ref (allocpad); GST_OBJECT_UNLOCK (tee); - res = gst_pad_alloc_buffer (allocpad, offset, size, caps, buf); + GST_TEE_DYN_LOCK (tee); + data = g_object_get_qdata (G_OBJECT (allocpad), push_data); + if (!data->removed) + res = gst_pad_alloc_buffer (allocpad, offset, size, caps, buf); + else + res = GST_FLOW_NOT_LINKED; + GST_TEE_DYN_UNLOCK (tee); + gst_object_unref (allocpad); GST_OBJECT_LOCK (tee); diff --git a/plugins/elements/gsttee.h b/plugins/elements/gsttee.h index 67d4c61..d767125 100644 --- a/plugins/elements/gsttee.h +++ b/plugins/elements/gsttee.h @@ -65,6 +65,9 @@ struct _GstTee { GstElement element; /*< private >*/ + /* lock protecting dynamic pads */ + GMutex *dyn_lock; + GstPad *sinkpad; GstPad *allocpad; gint pad_counter; diff --git a/tests/check/elements/tee.c b/tests/check/elements/tee.c index e48bc40..bc017dd 100644 --- a/tests/check/elements/tee.c +++ b/tests/check/elements/tee.c @@ -3,6 +3,8 @@ * unit test for tee * * Copyright (C) <2007> Wim Taymans + * Copyright (C) <2008> Ole André Vadla Ravnås + * Copyright (C) <2008> Christian Berentsen * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Library General Public @@ -164,6 +166,185 @@ GST_START_TEST (test_stress) GST_END_TEST; +static GstFlowReturn +final_sinkpad_bufferalloc (GstPad * pad, guint64 offset, guint size, + GstCaps * caps, GstBuffer ** buf); + +typedef struct +{ + GstElement *tee; + GstCaps *caps; + GstPad *start_srcpad; + GstPad *tee_sinkpad; + GstPad *tee_srcpad; + GstPad *final_sinkpad; + GThread *app_thread; + gint countdown; + gboolean app_thread_prepped; + gboolean bufferalloc_blocked; +} BufferAllocHarness; + +void +buffer_alloc_harness_setup (BufferAllocHarness * h, gint countdown) +{ + h->tee = gst_check_setup_element ("tee"); + fail_if (h->tee == NULL); + + h->countdown = countdown; + + fail_unless_equals_int (gst_element_set_state (h->tee, GST_STATE_PLAYING), + TRUE); + + h->caps = gst_caps_new_simple ("video/x-raw-yuv", NULL); + + h->start_srcpad = gst_pad_new ("src", GST_PAD_SRC); + fail_if (h->start_srcpad == NULL); + fail_unless (gst_pad_set_caps (h->start_srcpad, h->caps) == TRUE); + fail_unless (gst_pad_set_active (h->start_srcpad, TRUE) == TRUE); + + h->tee_sinkpad = gst_element_get_static_pad (h->tee, "sink"); + fail_if (h->tee_sinkpad == NULL); + + h->tee_srcpad = gst_element_get_request_pad (h->tee, "src%d"); + fail_if (h->tee_srcpad == NULL); + + h->final_sinkpad = gst_pad_new ("sink", GST_PAD_SINK); + fail_if (h->final_sinkpad == NULL); + gst_pad_set_bufferalloc_function (h->final_sinkpad, + final_sinkpad_bufferalloc); + fail_unless (gst_pad_set_caps (h->final_sinkpad, h->caps) == TRUE); + fail_unless (gst_pad_set_active (h->final_sinkpad, TRUE) == TRUE); + g_object_set_qdata (G_OBJECT (h->final_sinkpad), + g_quark_from_static_string ("buffer-alloc-harness"), h); + + fail_unless_equals_int (gst_pad_link (h->start_srcpad, h->tee_sinkpad), + GST_PAD_LINK_OK); + fail_unless_equals_int (gst_pad_link (h->tee_srcpad, h->final_sinkpad), + GST_PAD_LINK_OK); +} + +void +buffer_alloc_harness_teardown (BufferAllocHarness * h) +{ + g_thread_join (h->app_thread); + + gst_pad_set_active (h->final_sinkpad, FALSE); + gst_object_unref (h->final_sinkpad); + gst_object_unref (h->tee_srcpad); + gst_object_unref (h->tee_sinkpad); + gst_pad_set_active (h->start_srcpad, FALSE); + gst_object_unref (h->start_srcpad); + gst_caps_unref (h->caps); + gst_check_teardown_element (h->tee); +} + +static gpointer +app_thread_func (gpointer data) +{ + BufferAllocHarness *h = data; + + /* Signal that we are about to call release_request_pad(). */ + g_mutex_lock (check_mutex); + h->app_thread_prepped = TRUE; + g_cond_signal (check_cond); + g_mutex_unlock (check_mutex); + + /* Simulate that the app releases the pad while the streaming thread is in + * buffer_alloc below. */ + gst_element_release_request_pad (h->tee, h->tee_srcpad); + + /* Signal the bufferalloc function below if it's still waiting. */ + g_mutex_lock (check_mutex); + h->bufferalloc_blocked = FALSE; + g_cond_signal (check_cond); + g_mutex_unlock (check_mutex); + + return NULL; +} + +static GstFlowReturn +final_sinkpad_bufferalloc (GstPad * pad, guint64 offset, guint size, + GstCaps * caps, GstBuffer ** buf) +{ + BufferAllocHarness *h; + GTimeVal deadline; + + h = g_object_get_qdata (G_OBJECT (pad), + g_quark_from_static_string ("buffer-alloc-harness")); + g_assert (h != NULL); + + if (--(h->countdown) == 0) { + /* Time to make the app release the pad. */ + h->app_thread_prepped = FALSE; + h->bufferalloc_blocked = TRUE; + + h->app_thread = g_thread_create (app_thread_func, h, TRUE, NULL); + fail_if (h->app_thread == NULL); + + /* Wait for the app thread to get ready to call release_request_pad(). */ + g_mutex_lock (check_mutex); + while (!h->app_thread_prepped) + g_cond_wait (check_cond, check_mutex); + g_mutex_unlock (check_mutex); + + /* Now wait for it to do that within a second, to avoid deadlocking + * in the event of future changes to the locking semantics. */ + g_mutex_lock (check_mutex); + g_get_current_time (&deadline); + deadline.tv_sec += 1; + while (h->bufferalloc_blocked) { + if (!g_cond_timed_wait (check_cond, check_mutex, &deadline)) + break; + } + g_mutex_unlock (check_mutex); + } + + *buf = gst_buffer_new_and_alloc (size); + gst_buffer_set_caps (*buf, caps); + + return GST_FLOW_OK; +} + +/* Simulate an app releasing the pad while the first alloc_buffer() is in + * progress. */ +GST_START_TEST (test_release_while_buffer_alloc) +{ + BufferAllocHarness h; + GstBuffer *buf; + + buffer_alloc_harness_setup (&h, 1); + + fail_unless_equals_int (gst_pad_alloc_buffer (h.start_srcpad, 0, 1, h.caps, + &buf), GST_FLOW_OK); + gst_buffer_unref (buf); + + buffer_alloc_harness_teardown (&h); +} + +GST_END_TEST; + +/* Simulate an app releasing the pad while the second alloc_buffer() is in + * progress. */ +GST_START_TEST (test_release_while_second_buffer_alloc) +{ + BufferAllocHarness h; + GstBuffer *buf; + + buffer_alloc_harness_setup (&h, 2); + + fail_unless_equals_int (gst_pad_alloc_buffer (h.start_srcpad, 0, 1, h.caps, + &buf), GST_FLOW_OK); + gst_buffer_unref (buf); + + fail_unless_equals_int (gst_pad_alloc_buffer (h.start_srcpad, 0, 1, h.caps, + &buf), GST_FLOW_OK); + gst_buffer_unref (buf); + + buffer_alloc_harness_teardown (&h); +} + +GST_END_TEST; + static Suite * tee_suite (void) { @@ -173,6 +354,8 @@ tee_suite (void) suite_add_tcase (s, tc_chain); tcase_add_test (tc_chain, test_num_buffers); tcase_add_test (tc_chain, test_stress); + tcase_add_test (tc_chain, test_release_while_buffer_alloc); + tcase_add_test (tc_chain, test_release_while_second_buffer_alloc); return s; } -- 2.7.4