More massive changes to the scheduling system. Moved the scheduling code to...

More massive changes to the scheduling system.  Moved the scheduling code to gstscheduler.[ch], so an child bin can r...

Original commit message from CVS:
More massive changes to the scheduling system.  Moved the scheduling code
to gstscheduler.[ch], so an child bin can replace the scheduler.
Introduced the concept of chains, which are subsets of the list of managed
elements for a given manager bin, which get scheduled as separate entities.
gst_bin_iterate_func should be pretty much fixed now, the scheduling code
gets to do all the hard work.

Cothreaded case work in the couple tests I've tried, chained is next.

gst/Makefile.am

@@ -46,7 +46,8 @@ libgst_la_SOURCES =	\
 	gstmeta.c	\
 	gsttee.c	\
 	gstxml.c	\
-	cothreads.c
+	cothreads.c	\
+	gstscheduler.c
 
 libgstincludedir = $(includedir)/gst
 libgstinclude_HEADERS =	\
@@ -78,7 +79,8 @@ libgstinclude_HEADERS =	\
 	gsttee.h	\
 	gstxml.h	\
 	gstdebug.h	\
-	cothreads.h
+	cothreads.h	\
+	gstscheduler.h
 
 noinst_HEADERS =	\
 	gstarch.h       \

gst/elements/gstidentity.c

@@ -134,6 +134,7 @@ gst_identity_loop (GstElement *element)
   
   do {
     buf = gst_pad_pull (identity->sinkpad);
+    g_print("(%s:%s)i ",GST_DEBUG_PAD_NAME(identity->sinkpad));
 
     gst_pad_push (identity->srcpad, buf);
 

gst/elements/gstqueue.c

@@ -49,6 +49,7 @@ enum {
   ARG_0,
   ARG_LEVEL,
   ARG_MAX_LEVEL,
+  ARG_BLOCK,
 };
 
 
@@ -106,6 +107,8 @@ gst_queue_class_init (GstQueueClass *klass)
                            GTK_ARG_READABLE, ARG_LEVEL);
   gtk_object_add_arg_type ("GstQueue::max_level", GTK_TYPE_INT,
                            GTK_ARG_READWRITE, ARG_MAX_LEVEL);
+  gtk_object_add_arg_type ("GstQueue::block", GTK_TYPE_BOOL,
+                           GTK_ARG_READWRITE, ARG_BLOCK);
 
   gtkobject_class->set_arg = gst_queue_set_arg;  
   gtkobject_class->get_arg = gst_queue_get_arg;
@@ -116,7 +119,8 @@ gst_queue_class_init (GstQueueClass *klass)
 static void 
 gst_queue_init (GstQueue *queue) 
 {
-  GST_FLAG_SET (queue, GST_ELEMENT_SCHEDULE_PASSIVELY);
+  // scheduling on this kind of element is, well, interesting
+  GST_FLAG_SET (queue, GST_ELEMENT_DECOUPLED);
 
   queue->sinkpad = gst_pad_new ("sink", GST_PAD_SINK);
   gst_pad_set_chain_function (queue->sinkpad, GST_DEBUG_FUNCPTR(gst_queue_chain));
@@ -129,6 +133,7 @@ gst_queue_init (GstQueue *queue)
   queue->queue = NULL;
   queue->level_buffers = 0;
   queue->max_buffers = 20;
+  queue->block = TRUE;
   queue->level_bytes = 0;
   queue->size_buffers = 0;
   queue->size_bytes = 0;
@@ -237,6 +242,12 @@ gst_queue_get (GstPad *pad)
   DEBUG("queue: %s push %d %ld %p\n", name, queue->level_buffers, pthread_self (), queue->emptycond);
   DEBUG("queue: %s have queue lock\n", name);
 
+  // we bail if there's nothing there
+  if (!queue->level_buffers && !queue->block) {
+    GST_UNLOCK(queue);
+    return NULL;
+  }
+
   while (!queue->level_buffers) {
     STATUS("queue: %s U released lock\n");
     GST_UNLOCK (queue);
@@ -312,6 +323,9 @@ gst_queue_set_arg (GtkObject *object, GtkArg *arg, guint id)
     case ARG_MAX_LEVEL:
       queue->max_buffers = GTK_VALUE_INT (*arg);
       break;
+    case ARG_BLOCK:
+      queue->block = GTK_VALUE_BOOL (*arg);
+      break;
     default:
       break;
   }
@@ -334,6 +348,9 @@ gst_queue_get_arg (GtkObject *object, GtkArg *arg, guint id)
     case ARG_MAX_LEVEL:
       GTK_VALUE_INT (*arg) = queue->max_buffers;
       break;
+    case ARG_BLOCK:
+      GTK_VALUE_BOOL (*arg) = queue->block;
+      break;
     default:
       arg->type = GTK_TYPE_INVALID;
       break;

gst/elements/gstqueue.h

@@ -59,6 +59,7 @@ struct _GstQueue {
 
   gint level_buffers;	/* number of buffers queued here */
   gint max_buffers;	/* maximum number of buffers queued here */
+  gboolean block;	/* if set to FALSE, _get returns NULL if queue empty */
   gint level_bytes;	/* number of bytes queued here */
   gint size_buffers;	/* size of queue in buffers */
   gint size_bytes;	/* size of queue in bytes */

gst/gstbin.h

@@ -54,6 +54,7 @@ typedef enum {
 
 typedef struct _GstBin GstBin;
 typedef struct _GstBinClass GstBinClass;
+typedef struct __GstBinChain _GstBinChain;
 
 struct _GstBin {
   GstElement element;
@@ -66,12 +67,14 @@ struct _GstBin {
   gboolean need_cothreads;
   GList *managed_elements;
   gint num_managed_elements;
+
+  GList *chains;
+  gint num_chains;
   GList *entries;
   gint num_entries;
 
   cothread_context *threadcontext;
   gboolean use_cothreads;
-  GList *outside_schedules;
 };
 
 struct _GstBinClass {
@@ -87,10 +90,21 @@ struct _GstBinClass {
                                  	 GtkType type);
   /* create a plan for the execution of the bin */
   void 		(*create_plan) 		(GstBin *bin);
+  void 		(*schedule) 		(GstBin *bin);
   /* run a full iteration of operation */
   void		(*iterate) 		(GstBin *bin);
 };
 
+struct __GstBinChain {
+  GList *elements;
+  gint num_elements;
+
+  GList *entries;
+
+  gboolean use_cothreads;
+
+  GstElement *entry;
+};  
 
 
 GtkType 	gst_bin_get_type		(void);
@@ -109,6 +123,7 @@ GstElement*	gst_bin_get_by_name		(GstBin *bin,
 GList*		gst_bin_get_list		(GstBin *bin);
 
 void 		gst_bin_create_plan		(GstBin *bin);
+void 		gst_bin_schedule		(GstBin *bin);
 gboolean 	gst_bin_set_state_type		(GstBin *bin,
 						 GstElementState state,
 						 GtkType type);

gst/gstdebug.h

@@ -101,6 +101,7 @@ G_GNUC_UNUSED static GModule *_debug_self_module = NULL;
   (_debug_string != NULL) ? \
     fprintf(stderr,GST_DEBUG_PREFIX("%s: "format , _debug_string , ## args )) : \
     fprintf(stderr,GST_DEBUG_PREFIX(": "format , ## args ))
+#define DEBUG_NOPREFIX(format,args...) fprintf(stderr,format , ## args )
 #define DEBUG_ENTER(format, args...) \
   fprintf(stderr,GST_DEBUG_PREFIX(format": entering\n" , ## args ))
 #define DEBUG_SET_STRING(format, args...) \
@@ -112,6 +113,7 @@ G_GNUC_UNUSED static GModule *_debug_self_module = NULL;
 #define DEBUG_LEAVE_STRING DEBUG_LEAVE("%s",_debug_string)
 #else
 #define DEBUG(format, args...)
+#define DEBUG_NOPREFIX(format, args...)
 #define DEBUG_ENTER(format, args...)
 #define DEBUG_LEAVE(format, args...)
 #define DEBUG_SET_STRING(format, args...)

gst/gstelement.h

@@ -85,8 +85,9 @@ static inline char *_gst_print_statename(int state) {
 typedef enum {
   // element is complex (for some def.) and generally require a cothread
   GST_ELEMENT_COMPLEX		= GST_OBJECT_FLAG_LAST,
-  // not to be scheduled directly, let others trigger all events
-  GST_ELEMENT_SCHEDULE_PASSIVELY,
+  // input and output pads aren't directly coupled to each other
+  // examples: queues, multi-output async readers, etc.
+  GST_ELEMENT_DECOUPLED,
   // this element should be placed in a thread if at all possible
   GST_ELEMENT_THREAD_SUGGESTED,
   // this element is incable of seeking (FIXME: does this apply to filters?)
@@ -96,6 +97,8 @@ typedef enum {
   GST_ELEMENT_NEW_LOOPFUNC,
   // the cothread holding this element needs to be stopped
   GST_ELEMENT_COTHREAD_STOPPING,
+  // the element has to be scheduled as a cothread for any sanity
+  GST_ELEMENT_USE_COTHREAD,
 
   /* use some padding for future expansion */
   GST_ELEMENT_FLAG_LAST		= GST_OBJECT_FLAG_LAST + 8,

gst/gstscheduler.h

+/* Gnome-Streamer
+ * Copyright (C) <1999> Erik Walthinsen <omega@cse.ogi.edu>
+ *
+ * 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.
+ */
+
+
+#ifndef __GST_SCHEDULER_H__
+#define __GST_SCHEDULER_H__
+
+#include <gst/gstbin.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif /* __cplusplus */
+
+
+void gst_bin_schedule_func(GstBin *bin);
+
+
+#ifdef __cplusplus   
+}
+#endif /* __cplusplus */
+
+#endif /* __GST_SCHEDULER_H__ */

gst/gstthread.c

@@ -116,6 +116,8 @@ gst_thread_class_init (GstThreadClass *klass)
 static void 
 gst_thread_init (GstThread *thread) 
 {
+  DEBUG("initializing thread '%s'\n",gst_element_get_name(GST_ELEMENT(thread)));
+
   // we're a manager by default
   GST_FLAG_SET (thread, GST_BIN_FLAG_MANAGER);
 

plugins/elements/gstidentity.c

@@ -134,6 +134,7 @@ gst_identity_loop (GstElement *element)
   
   do {
     buf = gst_pad_pull (identity->sinkpad);
+    g_print("(%s:%s)i ",GST_DEBUG_PAD_NAME(identity->sinkpad));
 
     gst_pad_push (identity->srcpad, buf);
 

plugins/elements/gstqueue.c

@@ -49,6 +49,7 @@ enum {
   ARG_0,
   ARG_LEVEL,
   ARG_MAX_LEVEL,
+  ARG_BLOCK,
 };
 
 
@@ -106,6 +107,8 @@ gst_queue_class_init (GstQueueClass *klass)
                            GTK_ARG_READABLE, ARG_LEVEL);
   gtk_object_add_arg_type ("GstQueue::max_level", GTK_TYPE_INT,
                            GTK_ARG_READWRITE, ARG_MAX_LEVEL);
+  gtk_object_add_arg_type ("GstQueue::block", GTK_TYPE_BOOL,
+                           GTK_ARG_READWRITE, ARG_BLOCK);
 
   gtkobject_class->set_arg = gst_queue_set_arg;  
   gtkobject_class->get_arg = gst_queue_get_arg;
@@ -116,7 +119,8 @@ gst_queue_class_init (GstQueueClass *klass)
 static void 
 gst_queue_init (GstQueue *queue) 
 {
-  GST_FLAG_SET (queue, GST_ELEMENT_SCHEDULE_PASSIVELY);
+  // scheduling on this kind of element is, well, interesting
+  GST_FLAG_SET (queue, GST_ELEMENT_DECOUPLED);
 
   queue->sinkpad = gst_pad_new ("sink", GST_PAD_SINK);
   gst_pad_set_chain_function (queue->sinkpad, GST_DEBUG_FUNCPTR(gst_queue_chain));
@@ -129,6 +133,7 @@ gst_queue_init (GstQueue *queue)
   queue->queue = NULL;
   queue->level_buffers = 0;
   queue->max_buffers = 20;
+  queue->block = TRUE;
   queue->level_bytes = 0;
   queue->size_buffers = 0;
   queue->size_bytes = 0;
@@ -237,6 +242,12 @@ gst_queue_get (GstPad *pad)
   DEBUG("queue: %s push %d %ld %p\n", name, queue->level_buffers, pthread_self (), queue->emptycond);
   DEBUG("queue: %s have queue lock\n", name);
 
+  // we bail if there's nothing there
+  if (!queue->level_buffers && !queue->block) {
+    GST_UNLOCK(queue);
+    return NULL;
+  }
+
   while (!queue->level_buffers) {
     STATUS("queue: %s U released lock\n");
     GST_UNLOCK (queue);
@@ -312,6 +323,9 @@ gst_queue_set_arg (GtkObject *object, GtkArg *arg, guint id)
     case ARG_MAX_LEVEL:
       queue->max_buffers = GTK_VALUE_INT (*arg);
       break;
+    case ARG_BLOCK:
+      queue->block = GTK_VALUE_BOOL (*arg);
+      break;
     default:
       break;
   }
@@ -334,6 +348,9 @@ gst_queue_get_arg (GtkObject *object, GtkArg *arg, guint id)
     case ARG_MAX_LEVEL:
       GTK_VALUE_INT (*arg) = queue->max_buffers;
       break;
+    case ARG_BLOCK:
+      GTK_VALUE_BOOL (*arg) = queue->block;
+      break;
     default:
       arg->type = GTK_TYPE_INVALID;
       break;

plugins/elements/gstqueue.h

@@ -59,6 +59,7 @@ struct _GstQueue {
 
   gint level_buffers;	/* number of buffers queued here */
   gint max_buffers;	/* maximum number of buffers queued here */
+  gboolean block;	/* if set to FALSE, _get returns NULL if queue empty */
   gint level_bytes;	/* number of bytes queued here */
   gint size_buffers;	/* size of queue in buffers */
   gint size_bytes;	/* size of queue in bytes */