libs/gst/base/gstcollectpads.c: Unlock mutex when removing an unknown pad.

Original commit message from CVS:
Patch by: Alessandro Decina <alessandro at nnva dot org>
* libs/gst/base/gstcollectpads.c: (gst_collect_pads_remove_pad):
Unlock mutex when removing an unknown pad.
Fixes #343334.
* tests/check/Makefile.am:
* tests/check/libs/collectpads.c: (collected_cb), (push_buffer),
(push_event), (setup), (teardown), (GST_START_TEST),
(gst_collect_pads_suite), (main):
Added collecpads check, disabled for now as check crashes for
some reason.

ChangeLog

@@ -1,3 +1,18 @@
+2006-05-30  Wim Taymans  <wim@fluendo.com>
+
+	Patch by: Alessandro Decina <alessandro at nnva dot org>
+
+	* libs/gst/base/gstcollectpads.c: (gst_collect_pads_remove_pad):
+	Unlock mutex when removing an unknown pad.
+	Fixes #343334.
+
+	* tests/check/Makefile.am:
+	* tests/check/libs/collectpads.c: (collected_cb), (push_buffer),
+	(push_event), (setup), (teardown), (GST_START_TEST),
+	(gst_collect_pads_suite), (main):
+	Added collecpads check, disabled for now as check crashes for
+	some reason.
+
 2006-05-29  Wim Taymans  <wim@fluendo.com>
 
 	* libs/gst/base/gstcollectpads.c: (gst_collect_pads_finalize):

libs/gst/base/gstcollectpads.c

@@ -309,6 +309,7 @@ gst_collect_pads_remove_pad (GstCollectPads * pads, GstPad * pad)
 unknown_pad:
   {
     GST_WARNING ("cannot remove unknown pad %s:%s", GST_DEBUG_PAD_NAME (pad));
+    GST_COLLECT_PADS_PAD_UNLOCK (pads);
     return FALSE;
   }
 }

tests/check/Makefile.am

@@ -109,6 +109,9 @@ libs_basesrc_LDADD = \
 libs_adapter_LDADD = \
 	$(top_builddir)/libs/gst/base/libgstbase-@GST_MAJORMINOR@.la \
 	$(LDADD)
+libs_collectpads_LDADD = \
+	$(top_builddir)/libs/gst/base/libgstbase-@GST_MAJORMINOR@.la \
+	$(LDADD)
 libs_controller_LDADD = \
 	$(top_builddir)/libs/gst/controller/libgstcontroller-@GST_MAJORMINOR@.la \
 	$(LDADD)
@@ -124,7 +127,8 @@ libs_typefindhelper_LDADD = \
 
 # valgrind testing
 # these just need valgrind fixing, period
-VALGRIND_TO_FIX =
+VALGRIND_TO_FIX =		\
+	libs/collectpads
 
 VALGRIND_IGNORE = \
 	pipelines/stress

tests/check/libs/collectpads.c

@@ -0,0 +1,276 @@
+/*
+ * collectpads.c - GstCollectPads testsuite
+ * Copyright (C) 2006 Alessandro Decina <alessandro@nnva.org>
+ *
+ * Authors:
+ *   Alessandro Decina <alessandro@nnva.org>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+#include <gst/check/gstcheck.h>
+#include <gst/base/gstcollectpads.h>
+
+typedef struct
+{
+  char foo;
+} BadCollectData;
+
+typedef struct
+{
+  GstCollectData data;
+  GstPad *pad;
+  GstBuffer *buffer;
+  GstEvent *event;
+} TestData;
+
+static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
+    GST_PAD_SRC,
+    GST_PAD_ALWAYS,
+    GST_STATIC_CAPS_ANY);
+
+static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
+    GST_PAD_SINK,
+    GST_PAD_ALWAYS,
+    GST_STATIC_CAPS_ANY);
+
+static GstCollectPads *collect;
+static gboolean collected;
+static GstPad *srcpad1, *srcpad2;
+static GstPad *sinkpad1, *sinkpad2;
+static TestData *data1, *data2;
+
+GMutex *lock;
+GCond *cond;
+
+static GstFlowReturn
+collected_cb (GstCollectPads * pads, gpointer user_data)
+{
+  g_mutex_lock (lock);
+  collected = TRUE;
+  g_cond_signal (cond);
+  g_mutex_unlock (lock);
+
+  return GST_FLOW_OK;
+}
+
+static gpointer
+push_buffer (gpointer user_data)
+{
+  TestData *test_data = (TestData *) user_data;
+
+  fail_unless (gst_pad_push (test_data->pad, test_data->buffer)
+      == GST_FLOW_OK);
+
+  return NULL;
+}
+
+static gpointer
+push_event (gpointer user_data)
+{
+  TestData *test_data = (TestData *) user_data;
+
+  fail_unless (gst_pad_push_event (test_data->pad, test_data->event) == TRUE);
+
+  return NULL;
+}
+
+static void
+setup ()
+{
+  collect = gst_collect_pads_new ();
+  gst_collect_pads_set_function (collect, collected_cb, NULL);
+
+  srcpad1 = gst_pad_new_from_static_template (&srctemplate, "src1");
+  srcpad2 = gst_pad_new_from_static_template (&srctemplate, "src2");
+  sinkpad1 = gst_pad_new_from_static_template (&sinktemplate, "sink1");
+  sinkpad2 = gst_pad_new_from_static_template (&sinktemplate, "sink2");
+  fail_unless (gst_pad_link (srcpad1, sinkpad1) == GST_PAD_LINK_OK);
+  fail_unless (gst_pad_link (srcpad2, sinkpad2) == GST_PAD_LINK_OK);
+
+  cond = g_cond_new ();
+  lock = g_mutex_new ();
+  data1 = NULL;
+  data2 = NULL;
+  collected = FALSE;
+}
+
+static void
+teardown ()
+{
+  gst_object_unref (sinkpad1);
+  gst_object_unref (sinkpad2);
+  gst_object_unref (collect);
+  g_cond_free (cond);
+  g_mutex_free (lock);
+}
+
+GST_START_TEST (test_pad_add_remove)
+{
+  ASSERT_CRITICAL (gst_collect_pads_add_pad (collect, sinkpad1,
+          sizeof (BadCollectData)));
+
+  data1 = (TestData *) gst_collect_pads_add_pad (collect,
+      sinkpad1, sizeof (TestData));
+  fail_unless (data1 != NULL);
+
+  fail_unless (gst_collect_pads_remove_pad (collect, sinkpad2) == FALSE);
+  fail_unless (gst_collect_pads_remove_pad (collect, sinkpad1) == TRUE);
+}
+
+GST_END_TEST;
+
+GST_START_TEST (test_collect)
+{
+  GstBuffer *buf1, *buf2, *tmp;
+  GThread *thread1, *thread2;
+
+  data1 = (TestData *) gst_collect_pads_add_pad (collect,
+      sinkpad1, sizeof (TestData));
+  fail_unless (data1 != NULL);
+
+  data2 = (TestData *) gst_collect_pads_add_pad (collect,
+      sinkpad2, sizeof (TestData));
+  fail_unless (data2 != NULL);
+
+  buf1 = gst_buffer_new ();
+  buf2 = gst_buffer_new ();
+
+  /* start collect pads */
+  gst_collect_pads_start (collect);
+
+  /* push buffers on the pads */
+  data1->pad = srcpad1;
+  data1->buffer = buf1;
+  thread1 = g_thread_create (push_buffer, data1, TRUE, NULL);
+  /* here thread1 is blocked and srcpad1 has a queued buffer */
+  fail_unless (collected == FALSE);
+
+  data2->pad = srcpad2;
+  data2->buffer = buf2;
+  thread2 = g_thread_create (push_buffer, data2, TRUE, NULL);
+
+  /* now both pads have a buffer */
+  g_mutex_lock (lock);
+  while (collected == FALSE)
+    g_cond_wait (cond, lock);
+  fail_unless (collected == TRUE);
+  g_mutex_unlock (lock);
+
+  tmp = gst_collect_pads_pop (collect, (GstCollectData *) data1);
+  fail_unless (tmp == buf1);
+  tmp = gst_collect_pads_pop (collect, (GstCollectData *) data2);
+  fail_unless (tmp == buf2);
+
+  /* these will return immediately as at this point the threads have been
+   * unlocked and are finished */
+  g_thread_join (thread1);
+  g_thread_join (thread2);
+
+  gst_collect_pads_stop (collect);
+
+  gst_buffer_unref (buf1);
+  gst_buffer_unref (buf2);
+}
+
+GST_END_TEST;
+
+GST_START_TEST (test_collect_eos)
+{
+  GstBuffer *buf1, *tmp;
+  GThread *thread1, *thread2;
+
+  data1 = (TestData *) gst_collect_pads_add_pad (collect,
+      sinkpad1, sizeof (TestData));
+  fail_unless (data1 != NULL);
+
+  data2 = (TestData *) gst_collect_pads_add_pad (collect,
+      sinkpad2, sizeof (TestData));
+  fail_unless (data2 != NULL);
+
+  buf1 = gst_buffer_new ();
+
+  /* start collect pads */
+  gst_collect_pads_start (collect);
+
+  /* push a buffer on srcpad1 and EOS on srcpad2 */
+  data1->pad = srcpad1;
+  data1->buffer = buf1;
+  thread1 = g_thread_create (push_buffer, data1, TRUE, NULL);
+  /* here thread1 is blocked and srcpad1 has a queued buffer */
+  fail_unless (collected == FALSE);
+
+  data2->pad = srcpad2;
+  data2->event = gst_event_new_eos ();
+  thread2 = g_thread_create (push_event, data2, TRUE, NULL);
+  /* now sinkpad1 has a buffer and sinkpad2 has EOS */
+  g_mutex_lock (lock);
+  while (collected == FALSE)
+    g_cond_wait (cond, lock);
+  fail_unless (collected == TRUE);
+  g_mutex_unlock (lock);
+
+  tmp = gst_collect_pads_pop (collect, (GstCollectData *) data1);
+  fail_unless (tmp == buf1);
+  /* sinkpad2 has EOS so a NULL buffer is returned */
+  tmp = gst_collect_pads_pop (collect, (GstCollectData *) data2);
+  fail_unless (tmp == NULL);
+
+  /* these will return immediately as when the data is popped the threads are
+   * unlocked and will terminate */
+  g_thread_join (thread1);
+  g_thread_join (thread2);
+
+  gst_collect_pads_stop (collect);
+
+  gst_buffer_unref (buf1);
+}
+
+GST_END_TEST;
+
+static Suite *
+gst_collect_pads_suite ()
+{
+  Suite *suite;
+  TCase *general;
+
+  suite = suite_create ("GstCollectPads");
+  general = tcase_create ("general");
+  suite_add_tcase (suite, general);
+  tcase_add_checked_fixture (general, setup, teardown);
+  tcase_add_test (general, test_pad_add_remove);
+  tcase_add_test (general, test_collect);
+  tcase_add_test (general, test_collect_eos);
+
+  return suite;
+}
+
+int
+main (int argc, char **argv)
+{
+  int nf;
+
+  Suite *suite = gst_collect_pads_suite ();
+  SRunner *runner = srunner_create (suite);
+
+  gst_check_init (&argc, &argv);
+
+  srunner_run_all (runner, CK_NORMAL);
+  nf = srunner_ntests_failed (runner);
+  srunner_free (runner);
+
+  return nf;
+}