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
{
gboolean pushed;
GstFlowReturn result;
+ gboolean removed;
} PushData;
static GstPad *gst_tee_request_new_pad (GstElement * element,
g_free (tee->last_message);
+ g_mutex_free (tee->dyn_lock);
+
G_OBJECT_CLASS (parent_class)->finalize (object);
}
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;
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);
gst_tee_release_pad (GstElement * element, GstPad * pad)
{
GstTee *tee;
+ PushData *data;
tee = GST_TEE (element);
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
/* 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)
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);
* need to unref the buffer */
if (res == GST_FLOW_OK)
gst_buffer_unref (*buf);
+ *buf = NULL;
goto retry;
}
if (res == GST_FLOW_OK) {
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",
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);
* 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
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)
{
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;
}
projects / platform / upstream / gstreamer.git / commitdiff