Index: debian/changelog
===================================================================
--- debian/changelog	(revision 6538)
+++ debian/changelog	(working copy)
@@ -1,3 +1,9 @@
+gstreamer0.10 (0.10.13-0osso10.1) unstable; urgency=low
+
+  * Add realtime/time critical priority support
+
+ -- Jussi Laako <jussi.laako@nokia.com>  Mon,  1 Nov 2007 14:56:45 +0200
+
 gstreamer0.10 (0.10.13-0osso10) unstable; urgency=low
 
   * Queue eos handling fix from:
Index: libs/gst/base/gstbasesrc.c
===================================================================
--- libs/gst/base/gstbasesrc.c	(revision 6538)
+++ libs/gst/base/gstbasesrc.c	(working copy)
@@ -1135,6 +1135,8 @@
   src->data.ABI.running = TRUE;
   /* and restart the task in case it got paused explicitely or by
    * the FLUSH_START event we pushed out. */
+  if (src->timing_critical)
+	  GST_PAD_TIMING_CRITICAL(src->srcpad) = TRUE;
   gst_pad_start_task (src->srcpad, (GstTaskFunction) gst_base_src_loop,
       src->srcpad);
 
@@ -2320,3 +2322,8 @@
     return result;
   }
 }
+
+void gst_base_src_set_critical (GstBaseSrc *src, gboolean critical)
+{
+	src->timing_critical = critical;
+}
Index: libs/gst/base/gstbasesrc.h
===================================================================
--- libs/gst/base/gstbasesrc.h	(revision 6538)
+++ libs/gst/base/gstbasesrc.h	(working copy)
@@ -73,6 +73,7 @@
 
   /*< protected >*/
   GstPad	*srcpad;
+  gboolean	 timing_critical;
 
   /* available to subclass implementations */
   /* MT-protected (with LIVE_LOCK) */
@@ -240,6 +241,8 @@
 gboolean        gst_base_src_query_latency (GstBaseSrc *src, gboolean * live,
                                             GstClockTime * min_latency, 
 					    GstClockTime * max_latency);
+void		gst_base_src_set_critical  (GstBaseSrc *src,
+                                            gboolean critical);
 
 G_END_DECLS
 
Index: gst/gstpad.c
===================================================================
--- gst/gstpad.c	(revision 6538)
+++ gst/gstpad.c	(working copy)
@@ -4306,6 +4306,8 @@
   task = GST_PAD_TASK (pad);
   if (task == NULL) {
     task = gst_task_create (func, data);
+    if (GST_PAD_TIMING_CRITICAL(pad))
+      task->rt = TRUE;
     gst_task_set_lock (task, GST_PAD_GET_STREAM_LOCK (pad));
     GST_PAD_TASK (pad) = task;
     GST_DEBUG_OBJECT (pad, "created task");
Index: gst/gstbin.c
===================================================================
--- gst/gstbin.c	(revision 6538)
+++ gst/gstbin.c	(working copy)
@@ -367,7 +367,6 @@
   GObjectClass *gobject_class;
   GstObjectClass *gstobject_class;
   GstElementClass *gstelement_class;
-  GError *err;
 
   gobject_class = (GObjectClass *) klass;
   gstobject_class = (GstObjectClass *) klass;
@@ -440,13 +439,8 @@
   klass->remove_element = GST_DEBUG_FUNCPTR (gst_bin_remove_func);
   klass->handle_message = GST_DEBUG_FUNCPTR (gst_bin_handle_message_func);
 
-  GST_DEBUG ("creating bin thread pool");
-  err = NULL;
-  klass->pool =
-      g_thread_pool_new ((GFunc) gst_bin_continue_func, NULL, -1, FALSE, &err);
-  if (err != NULL) {
-    g_critical ("could alloc threadpool %s", err->message);
-  }
+  /* delayed pool creation */
+  klass->pool = NULL;
 }
 
 static void
@@ -2417,10 +2411,32 @@
 bin_push_state_continue (GstBin * bin)
 {
   GstBinClass *klass;
+  GError *err;
+  GThreadScheduler sched;
+  gint prio;
 
   /* mark the bin dirty */
   GST_OBJECT_LOCK (bin);
   klass = GST_BIN_GET_CLASS (bin);
+
+  if (klass->pool == NULL) {
+    GST_DEBUG ("creating bin thread pool");
+    if (bin->rt) {
+      sched = G_THREAD_SCHEDULER_FIFO;
+      prio = g_thread_scheduler_get_priority_min(sched);
+    } else {
+      sched = G_THREAD_SCHEDULER_DEFAULT;
+      prio = 0;
+    }
+    err = NULL;
+    klass->pool =
+        g_thread_pool_new_with_priority ((GFunc) gst_bin_continue_func,
+            NULL, -1, FALSE, sched, prio, &err);
+    if (err != NULL) {
+      g_critical ("could not alloc threadpool %s", err->message);
+    }
+  }
+
   GST_DEBUG_OBJECT (bin, "pushing continue on thread pool");
   gst_object_ref (bin);
   g_thread_pool_push (klass->pool, bin, NULL);
Index: gst/gstpad.h
===================================================================
--- gst/gstpad.h	(revision 6538)
+++ gst/gstpad.h	(working copy)
@@ -602,6 +602,9 @@
   gint				 do_buffer_signals;
   gint				 do_event_signals;
 
+  /* timing critical pad, like audio pad */
+  gboolean			 timing_critical;
+
   /*< private >*/
   gpointer _gst_reserved[GST_PADDING];
 };
@@ -655,6 +658,7 @@
 
 #define GST_PAD_DO_BUFFER_SIGNALS(pad) 	(GST_PAD_CAST(pad)->do_buffer_signals)
 #define GST_PAD_DO_EVENT_SIGNALS(pad) 	(GST_PAD_CAST(pad)->do_event_signals)
+#define GST_PAD_TIMING_CRITICAL(pad)	(GST_PAD_CAST(pad)->timing_critical)
 
 #define GST_PAD_IS_LINKED(pad)		(GST_PAD_PEER(pad) != NULL)
 #define GST_PAD_IS_BLOCKED(pad)		(GST_OBJECT_FLAG_IS_SET (pad, GST_PAD_BLOCKED))
@@ -844,7 +848,7 @@
 gboolean		gst_pad_send_event			(GstPad *pad, GstEvent *event);
 
 /* pad tasks */
-gboolean		gst_pad_start_task			(GstPad *pad, GstTaskFunction func,
+gboolean		gst_pad_start_task			(GstPad *pad,GstTaskFunction func,
 								 gpointer data);
 gboolean		gst_pad_pause_task			(GstPad *pad);
 gboolean		gst_pad_stop_task			(GstPad *pad);
Index: gst/gstbin.h
===================================================================
--- gst/gstbin.h	(revision 6538)
+++ gst/gstbin.h	(working copy)
@@ -114,6 +114,8 @@
   GstClock	*provided_clock;
   GstElement    *clock_provider;
 
+  gboolean       rt;
+
   GstBinPrivate *priv;
 
   /*< private >*/
Index: gst/gsttask.c
===================================================================
--- gst/gsttask.c	(revision 6538)
+++ gst/gsttask.c	(working copy)
@@ -116,6 +116,11 @@
 
   klass->pool = g_thread_pool_new (
       (GFunc) gst_task_func, klass, -1, FALSE, NULL);
+  klass->rtpool = g_thread_pool_new_with_priority (
+      (GFunc) gst_task_func, klass, -1, FALSE,
+      G_THREAD_SCHEDULER_FIFO,
+      g_thread_scheduler_get_priority_min(G_THREAD_SCHEDULER_FIFO),
+      NULL);
 }
 
 static void
@@ -244,10 +249,16 @@
        * because the unref happens in the thread function.
        * Also wait for currently running ones to finish. */
       g_thread_pool_free (klass->pool, FALSE, TRUE);
+      g_thread_pool_free (klass->rtpool, FALSE, TRUE);
       /* create new pool, so we can still do something after this
        * call. */
       klass->pool = g_thread_pool_new (
           (GFunc) gst_task_func, klass, -1, FALSE, NULL);
+      klass->rtpool = g_thread_pool_new_with_priority (
+          (GFunc) gst_task_func, klass, -1, FALSE,
+          G_THREAD_SCHEDULER_FIFO,
+          g_thread_scheduler_get_priority_min(G_THREAD_SCHEDULER_FIFO),
+          NULL);
     }
     g_static_mutex_unlock (&pool_lock);
   }
@@ -386,7 +397,10 @@
       tclass = GST_TASK_GET_CLASS (task);
 
       g_static_mutex_lock (&pool_lock);
-      g_thread_pool_push (tclass->pool, task, NULL);
+      if (task->rt) { g_message("*** pushing task to rtpool\n");
+        g_thread_pool_push (tclass->rtpool, task, NULL); }
+      else { g_message("*** pushing task to pool\n");
+        g_thread_pool_push (tclass->pool, task, NULL); }
       g_static_mutex_unlock (&pool_lock);
       break;
     }
@@ -495,7 +509,10 @@
       tclass = GST_TASK_GET_CLASS (task);
 
       g_static_mutex_lock (&pool_lock);
-      g_thread_pool_push (tclass->pool, task, NULL);
+      if (task->rt)
+        g_thread_pool_push(tclass->rtpool, task, NULL);
+      else
+        g_thread_pool_push (tclass->pool, task, NULL);
       g_static_mutex_unlock (&pool_lock);
       break;
     }
Index: gst/gsttask.h
===================================================================
--- gst/gsttask.h	(revision 6538)
+++ gst/gsttask.h	(working copy)
@@ -132,6 +132,8 @@
 
   gboolean	   running;
 
+  gboolean         rt;
+
   /*< private >*/
   union {
     struct {
@@ -149,6 +151,7 @@
 
   /*< private >*/
   GThreadPool *pool;
+  GThreadPool *rtpool;
 
   /*< private >*/
   gpointer _gst_reserved[GST_PADDING];
