gst/gstbin.c: don't

Original commit message from CVS:
* gst/gstbin.c: (gst_bin_iterate_func), (gst_bin_iterate):
don't
* gst/schedulers/entryscheduler.c: (safe_cothread_switch),
(safe_cothread_destroy),
(gst_entry_scheduler_remove_all_cothreads),
(gst_entry_scheduler_reset), (_remove_cothread),
(gst_entry_scheduler_state_transition):
hold off cothread destruction if we're not in main cothread
* configure.ac:
* testsuite/Makefile.am:
add new test dir
* testsuite/schedulers/.cvsignore:
* testsuite/schedulers/Makefile.am:
add tests
* testsuite/schedulers/relink.c: (cb_handoff), (main):
check relinking and adding/removing elements from a running pipeline
* testsuite/schedulers/unlink.c: (cb_handoff), (main):
check unlinking in a running pipeline
* testsuite/schedulers/unref.c: (cb_handoff), (main):
check unreffing a running pipeline
* testsuite/schedulers/useless_iteration.c: (main):
check iterating a pipeline that contains running threads works

18 files changed, 678 insertions, 53 deletions

diff --git a/ChangeLog b/ChangeLog
index 831770942..e2d0bcc8a 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,28 @@
+2004-05-18  Benjamin Otte  <in7y118@public.uni-hamburg.de>
+
+	* gst/gstbin.c: (gst_bin_iterate_func), (gst_bin_iterate):
+	  don't
+	* gst/schedulers/entryscheduler.c: (safe_cothread_switch),
+	(safe_cothread_destroy),
+	(gst_entry_scheduler_remove_all_cothreads),
+	(gst_entry_scheduler_reset), (_remove_cothread),
+	(gst_entry_scheduler_state_transition):
+	  hold off cothread destruction if we're not in main cothread
+	* configure.ac:
+	* testsuite/Makefile.am:
+	  add new test dir
+	* testsuite/schedulers/.cvsignore:
+	* testsuite/schedulers/Makefile.am:
+	  add tests
+	* testsuite/schedulers/relink.c: (cb_handoff), (main):
+	  check relinking and adding/removing elements from a running pipeline
+	* testsuite/schedulers/unlink.c: (cb_handoff), (main):
+	  check unlinking in a running pipeline
+	* testsuite/schedulers/unref.c: (cb_handoff), (main):
+	  check unreffing a running pipeline
+	* testsuite/schedulers/useless_iteration.c: (main):
+	  check iterating a pipeline that contains running threads works
+
 2004-05-18  David Schleef  <ds@schleef.org>
 
 	* docs/gst/Makefile.am: Add all-local target for when HAVE_GTK_DOC
diff --git a/configure.ac b/configure.ac
index 96794e655..e163b5ddd 100644
--- a/configure.ac
+++ b/configure.ac
@@ -670,6 +670,7 @@ testsuite/negotiation/Makefile
 testsuite/parse/Makefile
 testsuite/plugin/Makefile
 testsuite/refcounting/Makefile
+testsuite/schedulers/Makefile
 testsuite/tags/Makefile
 testsuite/threads/Makefile
 examples/Makefile
diff --git a/gst/gstbin.c b/gst/gstbin.c
index ae9488a41..339c55b77 100644
--- a/gst/gstbin.c
+++ b/gst/gstbin.c
@@ -1112,17 +1112,35 @@ gst_bin_restore_thyself (GstObject * object, xmlNodePtr self)
 static gboolean
 gst_bin_iterate_func (GstBin * bin)
 {
+  GstScheduler *sched = GST_ELEMENT_SCHED (bin);
+
   /* only iterate if this is the manager bin */
-  if (GST_ELEMENT_SCHED (bin) &&
-      GST_ELEMENT_SCHED (bin)->parent == GST_ELEMENT (bin)) {
+  if (sched && sched->parent == GST_ELEMENT (bin)) {
     GstSchedulerState state;
 
-    state = gst_scheduler_iterate (GST_ELEMENT_SCHED (bin));
+    state = gst_scheduler_iterate (sched);
 
     if (state == GST_SCHEDULER_STATE_RUNNING) {
       return TRUE;
     } else if (state == GST_SCHEDULER_STATE_ERROR) {
       gst_element_set_state (GST_ELEMENT (bin), GST_STATE_PAUSED);
+    } else if (state == GST_SCHEDULER_STATE_STOPPED) {
+      /* check if we have children scheds that are still running */
+      /* FIXME: remove in 0.9? autouseless because iterations gone? */
+      GList *walk;
+
+      for (walk = sched->schedulers; walk; walk = g_list_next (walk)) {
+        GstScheduler *test = walk->data;
+
+        g_return_val_if_fail (test->parent, FALSE);
+        if (GST_STATE (test->parent) == GST_STATE_PLAYING) {
+          GST_CAT_DEBUG_OBJECT (GST_CAT_SCHEDULING, bin,
+              "current bin is not iterating, but children are, "
+              "so returning TRUE anyway...");
+          g_usleep (1);
+          return TRUE;
+        }
+      }
     }
   } else {
     g_warning ("bin \"%s\" is not the managing bin, can't be iterated on!\n",
@@ -1150,25 +1168,13 @@ gst_bin_iterate (GstBin * bin)
   g_return_val_if_fail (GST_IS_BIN (bin), FALSE);
 
   GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, bin, "starting iteration");
-
   gst_object_ref (GST_OBJECT (bin));
 
   running = FALSE;
   g_signal_emit (G_OBJECT (bin), gst_bin_signals[ITERATE], 0, &running);
 
-  GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, bin, "finished iteration");
-
-  if (!running) {
-    if (GST_STATE (bin) == GST_STATE_PLAYING &&
-        GST_STATE_PENDING (bin) == GST_STATE_VOID_PENDING) {
-      GST_CAT_DEBUG (GST_CAT_SCHEDULING,
-          "[%s]: polling for child shutdown after useless iteration",
-          GST_ELEMENT_NAME (bin));
-      g_usleep (1);
-      running = TRUE;
-    }
-  }
   gst_object_unref (GST_OBJECT (bin));
+  GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, bin, "finished iteration");
 
   return running;
 }
diff --git a/gst/schedulers/entryscheduler.c b/gst/schedulers/entryscheduler.c
index 10dfcbee6..fd86bfbfd 100644
--- a/gst/schedulers/entryscheduler.c
+++ b/gst/schedulers/entryscheduler.c
@@ -130,6 +130,8 @@ struct _GstEntryScheduler
   GList *schedule_possible;     /* possible entry points */
   GList *waiting;               /* elements waiting for the clock */
   gboolean error;               /* if an element threw an error */
+
+  GSList *reaping;              /* cothreads we need to destroy but can't */
 };
 
 struct _GstEntrySchedulerClass
@@ -526,10 +528,37 @@ can_schedule (Entry * entry)
   }
 }
 
-#define safe_cothread_switch(sched,cothread) G_STMT_START{ \
-  if (do_cothread_get_current (sched->context) != cothread) \
-    do_cothread_switch (cothread); \
-}G_STMT_END
+static void
+safe_cothread_switch (GstEntryScheduler * scheduler, cothread * thread)
+{
+  GList *list;
+  cothread *cur = do_cothread_get_current (scheduler->context);
+
+  if (cur == thread) {
+    GST_LOG_OBJECT (scheduler, "switch to same cothread, ignoring");
+  }
+
+  for (list = scheduler->schedule_possible; list; list = g_list_next (list)) {
+    if (ENTRY_IS_COTHREAD (list->data)) {
+      CothreadPrivate *priv = (CothreadPrivate *) list->data;
+
+      if (priv->thread == thread)
+        gst_object_ref (GST_OBJECT (priv->element));
+      if (priv->thread == cur)
+        gst_object_unref (GST_OBJECT (priv->element));
+    }
+  }
+  do_cothread_switch (thread);
+  if (cur == do_cothread_get_main (scheduler->context)) {
+    GSList *walk;
+
+    for (walk = scheduler->reaping; walk; walk = g_slist_next (walk)) {
+      do_cothread_destroy (walk->data);
+    }
+    g_slist_free (scheduler->reaping);
+    scheduler->reaping = NULL;
+  }
+}
 
 /* the meat - no guarantee as to which cothread this function is called */
 static void
@@ -739,26 +768,46 @@ gst_entry_scheduler_setup (GstScheduler * sched)
 }
 
 static void
-gst_entry_scheduler_reset (GstScheduler * sched)
+safe_cothread_destroy (CothreadPrivate * thread)
 {
-#if 0
-  /* FIXME: do we need to destroy cothreads ourselves? */
-  GList *elements = GST_ENTRY_SCHEDULER (sched)->elements;
+  GstEntryScheduler *scheduler = thread->sched;
 
-  while (elements) {
-    GstElement *element = GST_ELEMENT (elements->data);
+  if (do_cothread_get_current (scheduler->context) ==
+      do_cothread_get_main (scheduler->context)) {
+    do_cothread_destroy (thread->thread);
+  } else {
+    GST_WARNING_OBJECT (scheduler, "delaying destruction of cothread %p",
+        thread->thread);
+    scheduler->reaping = g_slist_prepend (scheduler->reaping, thread->thread);
+  }
+  thread->thread = NULL;
+}
 
-    if (GST_ELEMENT_THREADSTATE (element)) {
-      do_cothread_destroy (GST_ELEMENT_THREADSTATE (element));
-      GST_ELEMENT_THREADSTATE (element) = NULL;
+static void
+gst_entry_scheduler_remove_all_cothreads (GstEntryScheduler * scheduler)
+{
+  GList *list;
+
+  for (list = scheduler->schedule_possible; list; list = g_list_next (list)) {
+    if (ENTRY_IS_COTHREAD (list->data)) {
+      CothreadPrivate *priv = (CothreadPrivate *) list->data;
+
+      if (priv->thread)
+        safe_cothread_destroy (priv);
     }
-    elements = g_list_next (elements);
   }
-#endif
+}
+
+static void
+gst_entry_scheduler_reset (GstScheduler * sched)
+{
+  GstEntryScheduler *scheduler = GST_ENTRY_SCHEDULER (sched);
 
-  if (GST_ENTRY_SCHEDULER (sched)->context) {
-    do_cothread_context_destroy (GST_ENTRY_SCHEDULER (sched)->context);
-    GST_ENTRY_SCHEDULER (sched)->context = NULL;
+  if (scheduler->context) {
+    g_return_if_fail (scheduler->reaping == NULL);
+    gst_entry_scheduler_remove_all_cothreads (scheduler);
+    do_cothread_context_destroy (scheduler->context);
+    scheduler->context = NULL;
   }
 }
 
@@ -813,7 +862,7 @@ _remove_cothread (CothreadPrivate * priv)
   sched->schedule_possible = g_list_remove (sched->schedule_possible, priv);
 
   if (priv->thread)
-    do_cothread_destroy (priv->thread);
+    safe_cothread_destroy (priv);
   g_free (priv);
 }
 
@@ -856,23 +905,11 @@ gst_entry_scheduler_state_transition (GstScheduler * scheduler,
       break;
     case GST_STATE_PAUSED_TO_READY:
       if (element == scheduler->parent) {
-        GList *list;
-
-        for (list = sched->schedule_possible; list; list = g_list_next (list)) {
-          if (ENTRY_IS_COTHREAD (list->data)) {
-            CothreadPrivate *priv = (CothreadPrivate *) list->data;
-
-            if (priv->thread) {
-              do_cothread_destroy (priv->thread);
-              priv->thread = NULL;
-            }
-          }
-        }
+        gst_entry_scheduler_remove_all_cothreads (sched);
       }
       if (element->sched_private != NULL
           && ELEMENT_PRIVATE (element)->thread != NULL) {
-        do_cothread_destroy (ELEMENT_PRIVATE (element)->thread);
-        ELEMENT_PRIVATE (element)->thread = NULL;
+        safe_cothread_destroy (ELEMENT_PRIVATE (element));
       }
       break;
     case GST_STATE_READY_TO_NULL:
diff --git a/tests/old/testsuite/Makefile.am b/tests/old/testsuite/Makefile.am
index 74b849319..6d6b0ef70 100644
--- a/tests/old/testsuite/Makefile.am
+++ b/tests/old/testsuite/Makefile.am
@@ -16,11 +16,11 @@ endif
 SUBDIRS = bins bytestream cleanup dynparams ghostpads \
 	caps plugin elements clock refcounting tags threads \
 	indexers debug $(GST_PARSE_DIRS) $(GST_DEBUG_DIRS) \
-	dlopen negotiation
+	dlopen negotiation schedulers
 
 DIST_SUBDIRS = bins bytestream caps cleanup clock dynparams elements indexers \
 		plugin refcounting tags threads parse debug ghostpads \
-		dlopen negotiation
+		dlopen negotiation schedulers
 
 tests_pass = test_gst_init
 tests_fail = 
diff --git a/tests/old/testsuite/schedulers/.gitignore b/tests/old/testsuite/schedulers/.gitignore
new file mode 100644
index 000000000..c66a29a03
--- /dev/null
+++ b/tests/old/testsuite/schedulers/.gitignore
@@ -0,0 +1,7 @@
+relink_sink
+relink_src
+unlink_sink
+unlink_src
+unref_sink
+unref_src
+useless_iteration
diff --git a/tests/old/testsuite/schedulers/Makefile.am b/tests/old/testsuite/schedulers/Makefile.am
new file mode 100644
index 000000000..eaba16e2e
--- /dev/null
+++ b/tests/old/testsuite/schedulers/Makefile.am
@@ -0,0 +1,23 @@
+include ../Rules
+
+tests_pass = 									\
+	unlink_src unlink_sink							\
+	relink_src relink_sink							\
+	unref_src unref_sink							\
+	useless_iteration
+
+tests_fail = 
+tests_ignore =
+
+unlink_src_SOURCES = unlink.c
+unlink_src_CFLAGS = $(AM_CFLAGS) -DELEMENT=src
+unlink_sink_SOURCES = unlink.c
+unlink_sink_CFLAGS = $(AM_CFLAGS) -DELEMENT=sink
+relink_src_SOURCES = relink.c
+relink_src_CFLAGS = $(AM_CFLAGS) -DELEMENT=src -DOTHER_ELEMENT=sink
+relink_sink_SOURCES = relink.c
+relink_sink_CFLAGS = $(AM_CFLAGS) -DELEMENT=sink -DOTHER_ELEMENT=src
+unref_src_SOURCES = unref.c
+unref_src_CFLAGS = $(AM_CFLAGS) -DELEMENT=src
+unref_sink_SOURCES = unref.c
+unref_sink_CFLAGS = $(AM_CFLAGS) -DELEMENT=sink
diff --git a/tests/old/testsuite/schedulers/relink.c b/tests/old/testsuite/schedulers/relink.c
new file mode 100644
index 000000000..a76daf0df
--- /dev/null
+++ b/tests/old/testsuite/schedulers/relink.c
@@ -0,0 +1,73 @@
+/* GStreamer
+ * Copyright (C) 2004 Benjamin Otte <in7y118@public.uni-hamburg.de>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU 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
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free
+ * Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#include <gst/gst.h>
+
+GstElement *pipeline, *src, *sink;
+
+static void
+cb_handoff (GstElement * element, GstBuffer * buffer, GstPad * pad,
+    gpointer unused)
+{
+  if (GST_PAD_PEER (pad)) {
+    g_print ("relinking...\n");
+    gst_pad_unlink (pad, GST_PAD_PEER (pad));
+    gst_bin_remove (GST_BIN (pipeline), OTHER_ELEMENT);
+    OTHER_ELEMENT =
+        gst_element_factory_make ("fake" G_STRINGIFY (OTHER_ELEMENT), NULL);
+    g_assert (OTHER_ELEMENT);
+    gst_bin_add (GST_BIN (pipeline), OTHER_ELEMENT);
+    gst_element_sync_state_with_parent (OTHER_ELEMENT);
+    gst_element_link (ELEMENT, OTHER_ELEMENT);
+  }
+}
+
+gint
+main (gint argc, gchar ** argv)
+{
+  guint i = 0;
+
+  gst_init (&argc, &argv);
+
+  g_print ("setting up...\n");
+  /* setup pipeline */
+  pipeline = gst_element_factory_make ("pipeline", NULL);
+  g_assert (pipeline);
+  src = gst_element_factory_make ("fakesrc", NULL);
+  g_assert (src);
+  sink = gst_element_factory_make ("fakesink", NULL);
+  g_assert (sink);
+  gst_bin_add_many (GST_BIN (pipeline), src, sink, NULL);
+  gst_element_link (src, sink);
+  /* setup special stuff */
+  g_object_set (ELEMENT, "signal-handoffs", TRUE, NULL);
+  g_signal_connect (ELEMENT, "handoff", (GCallback) cb_handoff, NULL);
+
+  /* run pipeline */
+  g_print ("running...\n");
+  if (gst_element_set_state (pipeline, GST_STATE_PLAYING) != GST_STATE_SUCCESS)
+    g_assert_not_reached ();
+  while (i++ < 10 && gst_bin_iterate (GST_BIN (pipeline)));
+
+  g_print ("cleaning up...\n");
+  gst_object_unref (GST_OBJECT (pipeline));
+  pipeline = NULL;
+
+  g_print ("done.\n");
+  return 0;
+}
diff --git a/tests/old/testsuite/schedulers/unlink.c b/tests/old/testsuite/schedulers/unlink.c
new file mode 100644
index 000000000..11ae3e5c1
--- /dev/null
+++ b/tests/old/testsuite/schedulers/unlink.c
@@ -0,0 +1,64 @@
+/* GStreamer
+ * Copyright (C) 2004 Benjamin Otte <in7y118@public.uni-hamburg.de>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU 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
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free
+ * Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#include <gst/gst.h>
+
+static void
+cb_handoff (GstElement * element, GstBuffer * buffer, GstPad * pad,
+    gpointer unused)
+{
+  if (GST_PAD_PEER (pad)) {
+    g_print ("unlinking...\n");
+    gst_pad_unlink (pad, GST_PAD_PEER (pad));
+  }
+}
+
+gint
+main (gint argc, gchar ** argv)
+{
+  GstElement *pipeline, *src, *sink;
+
+  gst_init (&argc, &argv);
+
+  g_print ("setting up...\n");
+  /* setup pipeline */
+  pipeline = gst_element_factory_make ("pipeline", NULL);
+  g_assert (pipeline);
+  src = gst_element_factory_make ("fakesrc", NULL);
+  g_assert (src);
+  sink = gst_element_factory_make ("fakesink", NULL);
+  g_assert (sink);
+  gst_bin_add_many (GST_BIN (pipeline), src, sink, NULL);
+  gst_element_link (src, sink);
+  /* setup special stuff */
+  g_object_set (ELEMENT, "signal-handoffs", TRUE, NULL);
+  g_signal_connect (ELEMENT, "handoff", (GCallback) cb_handoff, NULL);
+
+  /* run pipeline */
+  g_print ("running...\n");
+  if (gst_element_set_state (pipeline, GST_STATE_PLAYING) != GST_STATE_SUCCESS)
+    g_assert_not_reached ();
+  while (gst_bin_iterate (GST_BIN (pipeline)));
+
+  g_print ("cleaning up...\n");
+  gst_object_unref (GST_OBJECT (pipeline));
+  pipeline = NULL;
+
+  g_print ("done.\n");
+  return 0;
+}
diff --git a/tests/old/testsuite/schedulers/unref.c b/tests/old/testsuite/schedulers/unref.c
new file mode 100644
index 000000000..543620a7a
--- /dev/null
+++ b/tests/old/testsuite/schedulers/unref.c
@@ -0,0 +1,61 @@
+/* GStreamer
+ * Copyright (C) 2004 Benjamin Otte <in7y118@public.uni-hamburg.de>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU 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
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free
+ * Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#include <gst/gst.h>
+
+GstElement *pipeline, *src, *sink;
+
+static void
+cb_handoff (GstElement * element, GstBuffer * buffer, GstPad * pad,
+    gpointer unused)
+{
+  if (pipeline) {
+    g_print ("unreffing...\n");
+    gst_object_unref (GST_OBJECT (pipeline));
+    pipeline = NULL;
+  }
+}
+
+gint
+main (gint argc, gchar ** argv)
+{
+  gst_init (&argc, &argv);
+
+  g_print ("setting up...\n");
+  /* setup pipeline */
+  pipeline = gst_element_factory_make ("pipeline", NULL);
+  g_assert (pipeline);
+  src = gst_element_factory_make ("fakesrc", NULL);
+  g_assert (src);
+  sink = gst_element_factory_make ("fakesink", NULL);
+  g_assert (sink);
+  gst_bin_add_many (GST_BIN (pipeline), src, sink, NULL);
+  gst_element_link (src, sink);
+  /* setup special stuff */
+  g_object_set (ELEMENT, "signal-handoffs", TRUE, NULL);
+  g_signal_connect (ELEMENT, "handoff", (GCallback) cb_handoff, NULL);
+
+  /* run pipeline */
+  g_print ("running...\n");
+  if (gst_element_set_state (pipeline, GST_STATE_PLAYING) != GST_STATE_SUCCESS)
+    g_assert_not_reached ();
+  while (pipeline && gst_bin_iterate (GST_BIN (pipeline)));
+
+  g_print ("done.\n");
+  return 0;
+}
diff --git a/tests/old/testsuite/schedulers/useless_iteration.c b/tests/old/testsuite/schedulers/useless_iteration.c
new file mode 100644
index 000000000..d064c648c
--- /dev/null
+++ b/tests/old/testsuite/schedulers/useless_iteration.c
@@ -0,0 +1,50 @@
+/* GStreamer
+ * Copyright (C) 2004 Benjamin Otte <in7y118@public.uni-hamburg.de>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU 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
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free
+ * Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#include <gst/gst.h>
+
+gint
+main (gint argc, gchar ** argv)
+{
+  GstElement *pipeline;
+  GError *error = NULL;
+  guint i = 0;
+
+  gst_init (&argc, &argv);
+
+  g_print ("setting up...\n");
+  /* setup pipeline */
+  pipeline = gst_parse_launch ("pipeline.( { fakesrc ! fakesink } )", &error);
+  g_assert (error == NULL);
+  g_assert (pipeline);
+
+  /* run pipeline */
+  g_print ("running...\n");
+  if (gst_element_set_state (pipeline, GST_STATE_PLAYING) != GST_STATE_SUCCESS)
+    g_assert_not_reached ();
+  while (i < 100 && gst_bin_iterate (GST_BIN (pipeline)))
+    i++;
+
+  g_print ("cleaning up... (%d iterations)\n", i);
+  g_assert (i == 100);
+  gst_object_unref (GST_OBJECT (pipeline));
+  pipeline = NULL;
+
+  g_print ("done.\n");
+  return 0;
+}
diff --git a/testsuite/Makefile.am b/testsuite/Makefile.am
index 74b849319..6d6b0ef70 100644
--- a/testsuite/Makefile.am
+++ b/testsuite/Makefile.am
@@ -16,11 +16,11 @@ endif
 SUBDIRS = bins bytestream cleanup dynparams ghostpads \
 	caps plugin elements clock refcounting tags threads \
 	indexers debug $(GST_PARSE_DIRS) $(GST_DEBUG_DIRS) \
-	dlopen negotiation
+	dlopen negotiation schedulers
 
 DIST_SUBDIRS = bins bytestream caps cleanup clock dynparams elements indexers \
 		plugin refcounting tags threads parse debug ghostpads \
-		dlopen negotiation
+		dlopen negotiation schedulers
 
 tests_pass = test_gst_init
 tests_fail = 
diff --git a/testsuite/schedulers/.gitignore b/testsuite/schedulers/.gitignore
new file mode 100644
index 000000000..c66a29a03
--- /dev/null
+++ b/testsuite/schedulers/.gitignore
@@ -0,0 +1,7 @@
+relink_sink
+relink_src
+unlink_sink
+unlink_src
+unref_sink
+unref_src
+useless_iteration
diff --git a/testsuite/schedulers/Makefile.am b/testsuite/schedulers/Makefile.am
new file mode 100644
index 000000000..eaba16e2e
--- /dev/null
+++ b/testsuite/schedulers/Makefile.am
@@ -0,0 +1,23 @@
+include ../Rules
+
+tests_pass = 									\
+	unlink_src unlink_sink							\
+	relink_src relink_sink							\
+	unref_src unref_sink							\
+	useless_iteration
+
+tests_fail = 
+tests_ignore =
+
+unlink_src_SOURCES = unlink.c
+unlink_src_CFLAGS = $(AM_CFLAGS) -DELEMENT=src
+unlink_sink_SOURCES = unlink.c
+unlink_sink_CFLAGS = $(AM_CFLAGS) -DELEMENT=sink
+relink_src_SOURCES = relink.c
+relink_src_CFLAGS = $(AM_CFLAGS) -DELEMENT=src -DOTHER_ELEMENT=sink
+relink_sink_SOURCES = relink.c
+relink_sink_CFLAGS = $(AM_CFLAGS) -DELEMENT=sink -DOTHER_ELEMENT=src
+unref_src_SOURCES = unref.c
+unref_src_CFLAGS = $(AM_CFLAGS) -DELEMENT=src
+unref_sink_SOURCES = unref.c
+unref_sink_CFLAGS = $(AM_CFLAGS) -DELEMENT=sink
diff --git a/testsuite/schedulers/relink.c b/testsuite/schedulers/relink.c
new file mode 100644
index 000000000..a76daf0df
--- /dev/null
+++ b/testsuite/schedulers/relink.c
@@ -0,0 +1,73 @@
+/* GStreamer
+ * Copyright (C) 2004 Benjamin Otte <in7y118@public.uni-hamburg.de>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU 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
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free
+ * Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#include <gst/gst.h>
+
+GstElement *pipeline, *src, *sink;
+
+static void
+cb_handoff (GstElement * element, GstBuffer * buffer, GstPad * pad,
+    gpointer unused)
+{
+  if (GST_PAD_PEER (pad)) {
+    g_print ("relinking...\n");
+    gst_pad_unlink (pad, GST_PAD_PEER (pad));
+    gst_bin_remove (GST_BIN (pipeline), OTHER_ELEMENT);
+    OTHER_ELEMENT =
+        gst_element_factory_make ("fake" G_STRINGIFY (OTHER_ELEMENT), NULL);
+    g_assert (OTHER_ELEMENT);
+    gst_bin_add (GST_BIN (pipeline), OTHER_ELEMENT);
+    gst_element_sync_state_with_parent (OTHER_ELEMENT);
+    gst_element_link (ELEMENT, OTHER_ELEMENT);
+  }
+}
+
+gint
+main (gint argc, gchar ** argv)
+{
+  guint i = 0;
+
+  gst_init (&argc, &argv);
+
+  g_print ("setting up...\n");
+  /* setup pipeline */
+  pipeline = gst_element_factory_make ("pipeline", NULL);
+  g_assert (pipeline);
+  src = gst_element_factory_make ("fakesrc", NULL);
+  g_assert (src);
+  sink = gst_element_factory_make ("fakesink", NULL);
+  g_assert (sink);
+  gst_bin_add_many (GST_BIN (pipeline), src, sink, NULL);
+  gst_element_link (src, sink);
+  /* setup special stuff */
+  g_object_set (ELEMENT, "signal-handoffs", TRUE, NULL);
+  g_signal_connect (ELEMENT, "handoff", (GCallback) cb_handoff, NULL);
+
+  /* run pipeline */
+  g_print ("running...\n");
+  if (gst_element_set_state (pipeline, GST_STATE_PLAYING) != GST_STATE_SUCCESS)
+    g_assert_not_reached ();
+  while (i++ < 10 && gst_bin_iterate (GST_BIN (pipeline)));
+
+  g_print ("cleaning up...\n");
+  gst_object_unref (GST_OBJECT (pipeline));
+  pipeline = NULL;
+
+  g_print ("done.\n");
+  return 0;
+}
diff --git a/testsuite/schedulers/unlink.c b/testsuite/schedulers/unlink.c
new file mode 100644
index 000000000..11ae3e5c1
--- /dev/null
+++ b/testsuite/schedulers/unlink.c
@@ -0,0 +1,64 @@
+/* GStreamer
+ * Copyright (C) 2004 Benjamin Otte <in7y118@public.uni-hamburg.de>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU 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
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free
+ * Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#include <gst/gst.h>
+
+static void
+cb_handoff (GstElement * element, GstBuffer * buffer, GstPad * pad,
+    gpointer unused)
+{
+  if (GST_PAD_PEER (pad)) {
+    g_print ("unlinking...\n");
+    gst_pad_unlink (pad, GST_PAD_PEER (pad));
+  }
+}
+
+gint
+main (gint argc, gchar ** argv)
+{
+  GstElement *pipeline, *src, *sink;
+
+  gst_init (&argc, &argv);
+
+  g_print ("setting up...\n");
+  /* setup pipeline */
+  pipeline = gst_element_factory_make ("pipeline", NULL);
+  g_assert (pipeline);
+  src = gst_element_factory_make ("fakesrc", NULL);
+  g_assert (src);
+  sink = gst_element_factory_make ("fakesink", NULL);
+  g_assert (sink);
+  gst_bin_add_many (GST_BIN (pipeline), src, sink, NULL);
+  gst_element_link (src, sink);
+  /* setup special stuff */
+  g_object_set (ELEMENT, "signal-handoffs", TRUE, NULL);
+  g_signal_connect (ELEMENT, "handoff", (GCallback) cb_handoff, NULL);
+
+  /* run pipeline */
+  g_print ("running...\n");
+  if (gst_element_set_state (pipeline, GST_STATE_PLAYING) != GST_STATE_SUCCESS)
+    g_assert_not_reached ();
+  while (gst_bin_iterate (GST_BIN (pipeline)));
+
+  g_print ("cleaning up...\n");
+  gst_object_unref (GST_OBJECT (pipeline));
+  pipeline = NULL;
+
+  g_print ("done.\n");
+  return 0;
+}
diff --git a/testsuite/schedulers/unref.c b/testsuite/schedulers/unref.c
new file mode 100644
index 000000000..543620a7a
--- /dev/null
+++ b/testsuite/schedulers/unref.c
@@ -0,0 +1,61 @@
+/* GStreamer
+ * Copyright (C) 2004 Benjamin Otte <in7y118@public.uni-hamburg.de>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU 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
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free
+ * Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#include <gst/gst.h>
+
+GstElement *pipeline, *src, *sink;
+
+static void
+cb_handoff (GstElement * element, GstBuffer * buffer, GstPad * pad,
+    gpointer unused)
+{
+  if (pipeline) {
+    g_print ("unreffing...\n");
+    gst_object_unref (GST_OBJECT (pipeline));
+    pipeline = NULL;
+  }
+}
+
+gint
+main (gint argc, gchar ** argv)
+{
+  gst_init (&argc, &argv);
+
+  g_print ("setting up...\n");
+  /* setup pipeline */
+  pipeline = gst_element_factory_make ("pipeline", NULL);
+  g_assert (pipeline);
+  src = gst_element_factory_make ("fakesrc", NULL);
+  g_assert (src);
+  sink = gst_element_factory_make ("fakesink", NULL);
+  g_assert (sink);
+  gst_bin_add_many (GST_BIN (pipeline), src, sink, NULL);
+  gst_element_link (src, sink);
+  /* setup special stuff */
+  g_object_set (ELEMENT, "signal-handoffs", TRUE, NULL);
+  g_signal_connect (ELEMENT, "handoff", (GCallback) cb_handoff, NULL);
+
+  /* run pipeline */
+  g_print ("running...\n");
+  if (gst_element_set_state (pipeline, GST_STATE_PLAYING) != GST_STATE_SUCCESS)
+    g_assert_not_reached ();
+  while (pipeline && gst_bin_iterate (GST_BIN (pipeline)));
+
+  g_print ("done.\n");
+  return 0;
+}
diff --git a/testsuite/schedulers/useless_iteration.c b/testsuite/schedulers/useless_iteration.c
new file mode 100644
index 000000000..d064c648c
--- /dev/null
+++ b/testsuite/schedulers/useless_iteration.c
@@ -0,0 +1,50 @@
+/* GStreamer
+ * Copyright (C) 2004 Benjamin Otte <in7y118@public.uni-hamburg.de>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU 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
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free
+ * Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#include <gst/gst.h>
+
+gint
+main (gint argc, gchar ** argv)
+{
+  GstElement *pipeline;
+  GError *error = NULL;
+  guint i = 0;
+
+  gst_init (&argc, &argv);
+
+  g_print ("setting up...\n");
+  /* setup pipeline */
+  pipeline = gst_parse_launch ("pipeline.( { fakesrc ! fakesink } )", &error);
+  g_assert (error == NULL);
+  g_assert (pipeline);
+
+  /* run pipeline */
+  g_print ("running...\n");
+  if (gst_element_set_state (pipeline, GST_STATE_PLAYING) != GST_STATE_SUCCESS)
+    g_assert_not_reached ();
+  while (i < 100 && gst_bin_iterate (GST_BIN (pipeline)))
+    i++;
+
+  g_print ("cleaning up... (%d iterations)\n", i);
+  g_assert (i == 100);
+  gst_object_unref (GST_OBJECT (pipeline));
+  pipeline = NULL;
+
+  g_print ("done.\n");
+  return 0;
+}