mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2025-02-17 11:45:25 +00:00
Code Issues 1 Projects Releases Wiki Activity

plugins/elements/gsttee.*: Protect pad_alloc with a new lock so that we can be sure that nothing is performing a pad_...

Browse source 
Original commit message from CVS:
Patch by: Ole André Vadla Ravnås  <ole.andre.ravnas at tandberg com>
* 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.
This commit is contained in:
Ole André Vadla Ravnås 2008-08-15 17:01:07 +00:00 committed by Wim Taymans
parent 05bfed2da8
commit fe969e6391
4 changed files with 240 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
ChangeLog
View file

@ -1,3 +1,19 @@
2008-08-15 Wim Taymans <wim.taymans@collabora.co.uk>
Patch by: Ole André Vadla Ravnås <ole.andre.ravnas at tandberg com>
* 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 <thijsvermeir@gmail.com>
* gst/gstpad.h:

41
plugins/elements/gsttee.c
View file

@ -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);

3
plugins/elements/gsttee.h
View file

@ -65,6 +65,9 @@ struct _GstTee {
GstElement element;
/*< private >*/
/* lock protecting dynamic pads */
GMutex *dyn_lock;
GstPad *sinkpad;
GstPad *allocpad;
gint pad_counter;

183
tests/check/elements/tee.c
View file

@ -3,6 +3,8 @@
* unit test for tee
*
* Copyright (C) <2007> Wim Taymans <wim dot taymans at gmail dot com>
* Copyright (C) <2008> Ole André Vadla Ravnås <ole.andre.ravnas@tandberg.com>
* Copyright (C) <2008> Christian Berentsen <christian.berentsen@tandberg.com>
*
* 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;
}