mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-12-18 14:26:43 +00:00
Code Issues 1 Projects Releases Wiki Activity

queue: fixed racy unit tests

Browse source 
Fixes #600004
This commit is contained in:
Jonas Holmberg 2010-08-18 15:12:45 +02:00 committed by Wim Taymans
parent 6ee61e4cb3
commit 63c0e75e15
1 changed files with 234 additions and 138 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

372
tests/check/elements/queue.c
View file

@ -24,13 +24,24 @@
#include <gst/check/gstcheck.h>
static gint overrun_count = 0;
static gint underrun_count = 0;
#define UNDERRUN_LOCK() (g_mutex_lock (underrun_mutex))
#define UNDERRUN_UNLOCK() (g_mutex_unlock (underrun_mutex))
#define UNDERRUN_SIGNAL() (g_cond_signal (underrun_cond))
#define UNDERRUN_WAIT() (g_cond_wait (underrun_cond, underrun_mutex))
static GstElement *queue;
/* For ease of programming we use globals to keep refs for our floating
* src and sink pads we create; otherwise we always have to do get_pad,
* get_peer, and then remove references in every test function */
static GstPad *mysrcpad, *mysinkpad;
static GstPad *mysrcpad;
static GstPad *mysinkpad;
static gint overrun_count;
static GMutex *underrun_mutex;
static GCond *underrun_cond;
static gint underrun_count;
static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
GST_PAD_SINK,
@ -52,35 +63,68 @@ static void
queue_underrun (GstElement * queue, gpointer user_data)
{
GST_DEBUG ("queue underrun");
g_mutex_lock (check_mutex);
UNDERRUN_LOCK ();
underrun_count++;
g_cond_signal (check_cond);
g_mutex_unlock (check_mutex);
}
static GstElement *
setup_queue (void)
{
GstElement *queue;
GST_DEBUG ("setup_queue");
overrun_count = 0;
underrun_count = 0;
queue = gst_check_setup_element ("queue");
g_signal_connect (queue, "overrun", G_CALLBACK (queue_overrun), NULL);
g_signal_connect (queue, "underrun", G_CALLBACK (queue_underrun), NULL);
return queue;
UNDERRUN_SIGNAL ();
UNDERRUN_UNLOCK ();
}
static void
cleanup_queue (GstElement * queue)
setup (void)
{
GST_DEBUG ("setup_queue");
queue = gst_check_setup_element ("queue");
g_signal_connect (queue, "underrun", G_CALLBACK (queue_underrun), NULL);
mysrcpad = gst_check_setup_src_pad (queue, &srctemplate, NULL);
gst_pad_set_active (mysrcpad, TRUE);
overrun_count = 0;
underrun_mutex = g_mutex_new ();
underrun_cond = g_cond_new ();
underrun_count = 0;
}
static void
cleanup (void)
{
GST_DEBUG ("cleanup_queue");
gst_check_drop_buffers ();
g_cond_free (underrun_cond);
underrun_cond = NULL;
g_mutex_free (underrun_mutex);
underrun_mutex = NULL;
gst_pad_set_active (mysrcpad, FALSE);
gst_check_teardown_src_pad (queue);
gst_check_teardown_element (queue);
queue = NULL;
}
/* setup the sinkpad on a playing queue element. gst_check_setup_sink_pad()
* does not work in this case since it does not activate the pad before linking
* it. */
static GstPad *
setup_sink_pad (GstElement * element, GstStaticPadTemplate * tmpl)
{
GstPad *srcpad;
GstPad *sinkpad;
sinkpad = gst_pad_new_from_static_template (tmpl, "sink");
fail_if (sinkpad == NULL);
srcpad = gst_element_get_static_pad (element, "src");
fail_if (srcpad == NULL);
gst_pad_set_chain_function (sinkpad, gst_check_chain_func);
gst_pad_set_active (sinkpad, TRUE);
fail_unless (gst_pad_link (srcpad, sinkpad) == GST_PAD_LINK_OK);
gst_object_unref (srcpad);
return sinkpad;
}
/* set queue size to 2 buffers
@ -89,38 +133,44 @@ cleanup_queue (GstElement * queue)
*/
GST_START_TEST (test_non_leaky_underrun)
{
GstElement *queue;
GstBuffer *buffer = NULL;
queue = setup_queue ();
g_signal_connect (queue, "overrun", G_CALLBACK (queue_overrun), NULL);
g_object_set (G_OBJECT (queue), "max-size-buffers", 2, NULL);
mysinkpad = gst_check_setup_sink_pad (queue, &sinktemplate, NULL);
gst_pad_set_active (mysinkpad, TRUE);
g_object_set (G_OBJECT (queue), "max-size-buffers", 2, NULL);
GST_DEBUG ("starting");
g_mutex_lock (check_mutex);
UNDERRUN_LOCK ();
fail_unless (gst_element_set_state (queue,
GST_STATE_PLAYING) == GST_STATE_CHANGE_SUCCESS,
"could not set to playing");
g_cond_wait (check_cond, check_mutex);
g_mutex_unlock (check_mutex);
UNDERRUN_WAIT ();
UNDERRUN_UNLOCK ();
fail_unless (overrun_count == 0);
fail_unless (underrun_count == 1);
fail_unless (buffer == NULL);
GST_DEBUG ("stopping");
fail_unless (gst_element_set_state (queue,
GST_STATE_NULL) == GST_STATE_CHANGE_SUCCESS, "could not set to null");
/* cleanup */
gst_pad_set_active (mysinkpad, FALSE);
gst_check_teardown_sink_pad (queue);
cleanup_queue (queue);
}
GST_END_TEST;
static void
queue_overrun_link_and_activate (GstElement * queue, gpointer user_data)
{
GST_DEBUG ("queue overrun");
overrun_count++;
/* link the src pad of the queue to make it dequeue buffers */
mysinkpad = setup_sink_pad (queue, &sinktemplate);
}
/* set queue size to 2 buffers
* push 2 buffers
* check over/underuns
@ -129,13 +179,13 @@ GST_END_TEST;
*/
GST_START_TEST (test_non_leaky_overrun)
{
GstElement *queue;
GstBuffer *buffer1, *buffer2, *buffer3;
GstBuffer *buffer1;
GstBuffer *buffer2;
GstBuffer *buffer3;
GstBuffer *buffer;
queue = setup_queue ();
mysrcpad = gst_check_setup_src_pad (queue, &srctemplate, NULL);
mysinkpad = gst_check_setup_sink_pad (queue, &sinktemplate, NULL);
gst_pad_set_active (mysrcpad, TRUE);
g_signal_connect (queue, "overrun",
G_CALLBACK (queue_overrun_link_and_activate), NULL);
g_object_set (G_OBJECT (queue), "max-size-buffers", 2, NULL);
GST_DEBUG ("starting");
@ -143,10 +193,11 @@ GST_START_TEST (test_non_leaky_overrun)
fail_unless (gst_element_set_state (queue,
GST_STATE_PLAYING) == GST_STATE_CHANGE_SUCCESS,
"could not set to playing");
fail_unless (overrun_count == 0);
fail_unless (underrun_count == 0);
buffer1 = gst_buffer_new_and_alloc (4);
ASSERT_BUFFER_REFCOUNT (buffer1, "buffer", 1);
/* pushing gives away my reference ... */
/* pushing gives away my reference */
gst_pad_push (mysrcpad, buffer1);
GST_DEBUG ("added 1st");
@ -154,8 +205,6 @@ GST_START_TEST (test_non_leaky_overrun)
fail_unless (underrun_count == 0);
buffer2 = gst_buffer_new_and_alloc (4);
ASSERT_BUFFER_REFCOUNT (buffer2, "buffer", 1);
/* pushing gives away my reference ... */
gst_pad_push (mysrcpad, buffer2);
GST_DEBUG ("added 2nd");
@ -163,20 +212,49 @@ GST_START_TEST (test_non_leaky_overrun)
fail_unless (underrun_count == 0);
buffer3 = gst_buffer_new_and_alloc (4);
ASSERT_BUFFER_REFCOUNT (buffer3, "buffer", 1);
/* pushing gives away my reference ... */
/* lock the check_mutex to block the first buffer pushed to mysinkpad */
g_mutex_lock (check_mutex);
/* the next call to gst_pad_push will emit the overrun signal. The signal
* handler queue_overrun() (above) increases overrun_count, links and
* activates mysinkpad. The queue task then dequeues a buffer and
* gst_pad_push() will return. */
gst_pad_push (mysrcpad, buffer3);
GST_DEBUG ("stopping");
GST_DEBUG ("added 3rd");
fail_unless (overrun_count == 1);
fail_unless (underrun_count == 0);
/* now let the queue push all buffers */
while (g_list_length (buffers) < 3) {
g_cond_wait (check_cond, check_mutex);
}
g_mutex_unlock (check_mutex);
fail_unless (overrun_count == 1);
/* make sure we get the underrun signal before we check underrun_count */
UNDERRUN_LOCK ();
while (underrun_count < 1) {
UNDERRUN_WAIT ();
}
UNDERRUN_UNLOCK ();
fail_unless (underrun_count == 1);
buffer = g_list_nth (buffers, 0)->data;
fail_unless (buffer == buffer1);
buffer = g_list_nth (buffers, 1)->data;
fail_unless (buffer == buffer2);
buffer = g_list_nth (buffers, 2)->data;
fail_unless (buffer == buffer3);
GST_DEBUG ("stopping");
fail_unless (gst_element_set_state (queue,
GST_STATE_NULL) == GST_STATE_CHANGE_SUCCESS, "could not set to null");
/* cleanup */
gst_pad_set_active (mysrcpad, FALSE);
gst_check_teardown_src_pad (queue);
gst_pad_set_active (mysinkpad, FALSE);
gst_check_teardown_sink_pad (queue);
cleanup_queue (queue);
}
GST_END_TEST;
@ -190,25 +268,24 @@ GST_END_TEST;
*/
GST_START_TEST (test_leaky_upstream)
{
GstElement *queue;
GstBuffer *buffer1, *buffer2, *buffer3;
GstBuffer *buffer1;
GstBuffer *buffer2;
GstBuffer *buffer3;
GstBuffer *buffer;
queue = setup_queue ();
mysrcpad = gst_check_setup_src_pad (queue, &srctemplate, NULL);
mysinkpad = gst_check_setup_sink_pad (queue, &sinktemplate, NULL);
g_signal_connect (queue, "overrun", G_CALLBACK (queue_overrun), NULL);
g_object_set (G_OBJECT (queue), "max-size-buffers", 2, "leaky", 1, NULL);
gst_pad_set_active (mysrcpad, TRUE);
GST_DEBUG ("starting");
fail_unless (gst_element_set_state (queue,
GST_STATE_PLAYING) == GST_STATE_CHANGE_SUCCESS,
"could not set to playing");
fail_unless (overrun_count == 0);
fail_unless (underrun_count == 0);
buffer1 = gst_buffer_new_and_alloc (4);
ASSERT_BUFFER_REFCOUNT (buffer1, "buffer", 1);
/* pushing gives away my reference ... */
/* pushing gives away my reference */
gst_pad_push (mysrcpad, buffer1);
GST_DEBUG ("added 1st");
@ -216,8 +293,6 @@ GST_START_TEST (test_leaky_upstream)
fail_unless (underrun_count == 0);
buffer2 = gst_buffer_new_and_alloc (4);
ASSERT_BUFFER_REFCOUNT (buffer2, "buffer", 1);
/* pushing gives away my reference ... */
gst_pad_push (mysrcpad, buffer2);
GST_DEBUG ("added 2nd");
@ -225,68 +300,82 @@ GST_START_TEST (test_leaky_upstream)
fail_unless (underrun_count == 0);
buffer3 = gst_buffer_new_and_alloc (4);
ASSERT_BUFFER_REFCOUNT (buffer3, "buffer", 1);
/* pushing gives away my reference ... */
gst_pad_push (mysrcpad, gst_buffer_ref (buffer3));
g_mutex_lock (check_mutex);
/* start the src-task briefly leak buffer3 */
gst_pad_set_active (mysinkpad, TRUE);
g_cond_wait (check_cond, check_mutex);
g_mutex_unlock (check_mutex);
gst_pad_set_active (mysinkpad, FALSE);
GST_DEBUG ("stopping");
fail_unless (g_list_length (buffers) > 0);
buffer = g_list_first (buffers)->data;
fail_unless (buffer == buffer1);
ASSERT_BUFFER_REFCOUNT (buffer3, "buffer", 1);
/* buffer3 will be leaked, keep a ref so refcount can be checked below */
gst_buffer_ref (buffer3);
gst_pad_push (mysrcpad, buffer3);
GST_DEBUG ("added 3rd");
/* it still triggers overrun when leaking */
fail_unless (overrun_count == 1);
fail_unless (underrun_count == 0);
/* wait for underrun and check that we got buffer1 and buffer2 only */
UNDERRUN_LOCK ();
mysinkpad = setup_sink_pad (queue, &sinktemplate);
UNDERRUN_WAIT ();
UNDERRUN_UNLOCK ();
fail_unless (overrun_count == 1);
fail_unless (underrun_count == 1);
fail_unless (g_list_length (buffers) == 2);
buffer = g_list_nth (buffers, 0)->data;
fail_unless (buffer == buffer1);
buffer = g_list_nth (buffers, 1)->data;
fail_unless (buffer == buffer2);
ASSERT_BUFFER_REFCOUNT (buffer3, "buffer", 1);
gst_buffer_unref (buffer3);
GST_DEBUG ("stopping");
fail_unless (gst_element_set_state (queue,
GST_STATE_NULL) == GST_STATE_CHANGE_SUCCESS, "could not set to null");
/* cleanup */
gst_pad_set_active (mysrcpad, FALSE);
gst_buffer_unref (buffer3);
gst_check_teardown_src_pad (queue);
gst_pad_set_active (mysinkpad, FALSE);
gst_check_teardown_sink_pad (queue);
cleanup_queue (queue);
}
GST_END_TEST;
/* set queue size to 2 buffers
* push 2 buffers
* check over/underuns
* push 1 more buffer
* check over/underuns again
* check which buffer was leaked
*/
GST_START_TEST (test_leaky_downstream)
{
GstElement *queue;
GstBuffer *buffer1, *buffer2, *buffer3;
GstBuffer *buffer1;
GstBuffer *buffer2;
GstBuffer *buffer3;
GstBuffer *buffer;
queue = setup_queue ();
mysrcpad = gst_check_setup_src_pad (queue, &srctemplate, NULL);
mysinkpad = gst_check_setup_sink_pad (queue, &sinktemplate, NULL);
g_object_set (G_OBJECT (queue), "leaky", 2, "max-size-buffers", 2, NULL);
gst_pad_set_active (mysrcpad, TRUE);
g_signal_connect (queue, "overrun", G_CALLBACK (queue_overrun), NULL);
g_object_set (G_OBJECT (queue), "max-size-buffers", 2, "leaky", 2, NULL);
GST_DEBUG ("starting");
fail_unless (gst_element_set_state (queue,
GST_STATE_PLAYING) == GST_STATE_CHANGE_SUCCESS,
"could not set to playing");
fail_unless (overrun_count == 0);
fail_unless (underrun_count == 0);
buffer1 = gst_buffer_new_and_alloc (4);
ASSERT_BUFFER_REFCOUNT (buffer1, "buffer", 1);
/* pushing gives away my reference ... */
gst_pad_push (mysrcpad, gst_buffer_ref (buffer1));
/* buffer1 will be leaked, keep a ref so refcount can be checked below */
gst_buffer_ref (buffer1);
/* pushing gives away one reference */
gst_pad_push (mysrcpad, buffer1);
GST_DEBUG ("added 1st");
fail_unless (overrun_count == 0);
fail_unless (underrun_count == 0);
buffer2 = gst_buffer_new_and_alloc (4);
ASSERT_BUFFER_REFCOUNT (buffer2, "buffer", 1);
/* pushing gives away my reference ... */
gst_pad_push (mysrcpad, buffer2);
GST_DEBUG ("added 2nd");
@ -294,55 +383,57 @@ GST_START_TEST (test_leaky_downstream)
fail_unless (underrun_count == 0);
buffer3 = gst_buffer_new_and_alloc (4);
ASSERT_BUFFER_REFCOUNT (buffer3, "buffer", 1);
/* pushing gives away my reference ... */
gst_pad_push (mysrcpad, buffer3);
g_mutex_lock (check_mutex);
/* start the src-task briefly and leak buffer1 */
gst_pad_set_active (mysinkpad, TRUE);
g_cond_wait (check_cond, check_mutex);
g_mutex_unlock (check_mutex);
gst_pad_set_active (mysinkpad, FALSE);
GST_DEBUG ("stopping");
fail_unless (g_list_length (buffers) > 0);
buffer = g_list_first (buffers)->data;
fail_unless (buffer == buffer2);
ASSERT_BUFFER_REFCOUNT (buffer1, "buffer", 1);
GST_DEBUG ("added 3rd");
/* it still triggers overrun when leaking */
fail_unless (overrun_count == 1);
fail_unless (underrun_count == 0);
/* wait for underrun and check that we got buffer1 and buffer2 only */
UNDERRUN_LOCK ();
mysinkpad = setup_sink_pad (queue, &sinktemplate);
UNDERRUN_WAIT ();
UNDERRUN_UNLOCK ();
fail_unless (overrun_count == 1);
fail_unless (underrun_count == 1);
fail_unless (g_list_length (buffers) == 2);
ASSERT_BUFFER_REFCOUNT (buffer1, "buffer", 1);
gst_buffer_unref (buffer1);
buffer = g_list_nth (buffers, 0)->data;
fail_unless (buffer == buffer2);
buffer = g_list_nth (buffers, 1)->data;
fail_unless (buffer == buffer3);
GST_DEBUG ("stopping");
fail_unless (gst_element_set_state (queue,
GST_STATE_NULL) == GST_STATE_CHANGE_SUCCESS, "could not set to null");
/* cleanup */
gst_pad_set_active (mysrcpad, FALSE);
gst_buffer_unref (buffer1);
gst_check_teardown_src_pad (queue);
gst_pad_set_active (mysinkpad, FALSE);
gst_check_teardown_sink_pad (queue);
cleanup_queue (queue);
}
GST_END_TEST;
/* set queue size to 5 buffers
* pull 1 buffer
* check over/underuns
/* set queue size to 6 buffers and 7 seconds
* push 7 buffers with and without duration
* check current-level-time
*/
GST_START_TEST (test_time_level)
{
GstElement *queue;
GstBuffer *buffer = NULL;
GstClockTime time;
queue = setup_queue ();
mysrcpad = gst_check_setup_src_pad (queue, &srctemplate, NULL);
mysinkpad = gst_check_setup_sink_pad (queue, &sinktemplate, NULL);
g_signal_connect (queue, "overrun",
G_CALLBACK (queue_overrun_link_and_activate), NULL);
g_object_set (G_OBJECT (queue), "max-size-buffers", 6, NULL);
g_object_set (G_OBJECT (queue), "max-size-time", 10 * GST_SECOND, NULL);
gst_pad_set_active (mysrcpad, TRUE);
gst_pad_set_active (mysinkpad, TRUE);
g_object_set (G_OBJECT (queue), "max-size-time", 7 * GST_SECOND, NULL);
GST_DEBUG ("starting");
@ -353,8 +444,7 @@ GST_START_TEST (test_time_level)
/* push buffer without duration */
buffer = gst_buffer_new_and_alloc (4);
GST_BUFFER_TIMESTAMP (buffer) = GST_SECOND;
ASSERT_BUFFER_REFCOUNT (buffer, "buffer", 1);
/* pushing gives away my reference ... */
/* pushing gives away my reference */
gst_pad_push (mysrcpad, buffer);
/* level should be 1 seconds because buffer has no duration and starts at 1
@ -365,8 +455,6 @@ GST_START_TEST (test_time_level)
/* second push should set the level to 2 second */
buffer = gst_buffer_new_and_alloc (4);
GST_BUFFER_TIMESTAMP (buffer) = 2 * GST_SECOND;
ASSERT_BUFFER_REFCOUNT (buffer, "buffer", 1);
/* pushing gives away my reference ... */
gst_pad_push (mysrcpad, buffer);
g_object_get (G_OBJECT (queue), "current-level-time", &time, NULL);
@ -379,7 +467,6 @@ GST_START_TEST (test_time_level)
GST_BUFFER_TIMESTAMP (buffer) = 3 * GST_SECOND;
GST_BUFFER_DURATION (buffer) = 1 * GST_SECOND;
ASSERT_BUFFER_REFCOUNT (buffer, "buffer", 1);
/* pushing gives away my reference ... */
gst_pad_push (mysrcpad, buffer);
g_object_get (G_OBJECT (queue), "current-level-time", &time, NULL);
@ -391,7 +478,6 @@ GST_START_TEST (test_time_level)
GST_BUFFER_TIMESTAMP (buffer) = 5 * GST_SECOND;
GST_BUFFER_DURATION (buffer) = 1 * GST_SECOND;
ASSERT_BUFFER_REFCOUNT (buffer, "buffer", 1);
/* pushing gives away my reference ... */
gst_pad_push (mysrcpad, buffer);
g_object_get (G_OBJECT (queue), "current-level-time", &time, NULL);
@ -402,8 +488,6 @@ GST_START_TEST (test_time_level)
buffer = gst_buffer_new_and_alloc (4);
GST_BUFFER_TIMESTAMP (buffer) = 5 * GST_SECOND;
GST_BUFFER_DURATION (buffer) = 1 * GST_SECOND;
ASSERT_BUFFER_REFCOUNT (buffer, "buffer", 1);
/* pushing gives away my reference ... */
gst_pad_push (mysrcpad, buffer);
g_object_get (G_OBJECT (queue), "current-level-time", &time, NULL);
@ -413,19 +497,30 @@ GST_START_TEST (test_time_level)
* previous buffer actually had a duration of 2 SECONDS */
buffer = gst_buffer_new_and_alloc (4);
GST_BUFFER_TIMESTAMP (buffer) = 7 * GST_SECOND;
ASSERT_BUFFER_REFCOUNT (buffer, "buffer", 1);
/* pushing gives away my reference ... */
gst_pad_push (mysrcpad, buffer);
g_object_get (G_OBJECT (queue), "current-level-time", &time, NULL);
fail_if (time != 7 * GST_SECOND);
/* eighth push should cause overrun */
fail_unless (overrun_count == 0);
buffer = gst_buffer_new_and_alloc (4);
GST_BUFFER_TIMESTAMP (buffer) = 8 * GST_SECOND;
/* the next call to gst_pad_push will emit the overrun signal. The signal
* handler queue_overrun() (above) increases overrun_count, links and
* activates mysinkpad. The queue task then dequeues a buffer and
* gst_pad_push() will return. */
gst_pad_push (mysrcpad, buffer);
fail_unless (overrun_count == 1);
GST_DEBUG ("stopping");
fail_unless (gst_element_set_state (queue,
GST_STATE_NULL) == GST_STATE_CHANGE_SUCCESS, "could not set to null");
/* cleanup */
gst_pad_set_active (mysinkpad, FALSE);
gst_check_teardown_sink_pad (queue);
cleanup_queue (queue);
}
GST_END_TEST;
@ -437,6 +532,7 @@ queue_suite (void)
TCase *tc_chain = tcase_create ("general");
suite_add_tcase (s, tc_chain);
tcase_add_checked_fixture (tc_chain, setup, cleanup);
tcase_add_test (tc_chain, test_non_leaky_underrun);
tcase_add_test (tc_chain, test_non_leaky_overrun);
tcase_add_test (tc_chain, test_leaky_upstream);