gst/gstbus.c: Add debugging messages.

Original commit message from CVS:
* gst/gstbus.c: (gst_bus_post), (gst_bus_have_pending),
(gst_bus_set_flushing), (gst_bus_pop), (gst_bus_peek),
(gst_bus_source_dispatch):
Add debugging messages.
Make internal methods static.
Handle the case where the bus is flushed in the handler.

* gst/gstelement.c: (gst_element_get_bus):
Fix refcount in _get_bus();

* gst/gstpipeline.c: (gst_pipeline_change_state),
(gst_pipeline_get_clock_func):
Clock refcounting fixes.
Handle the case where preroll timed out more gracefully.

* gst/gstsystemclock.c: (gst_system_clock_dispose):
Clean up the internal thread in dispose. This is needed
for subclasses that actually get disposed.

* gst/schedulers/threadscheduler.c:
(gst_thread_scheduler_class_init), (gst_thread_scheduler_func),
(gst_thread_scheduler_dispose):
Free thread pool in dispose.

ChangeLog

@@ -1,3 +1,29 @@
+2005-06-28  Wim Taymans  <wim@fluendo.com>
+
+	* gst/gstbus.c: (gst_bus_post), (gst_bus_have_pending),
+	(gst_bus_set_flushing), (gst_bus_pop), (gst_bus_peek),
+	(gst_bus_source_dispatch):
+	Add debugging messages.
+	Make internal methods static.
+	Handle the case where the bus is flushed in the handler.
+	
+	* gst/gstelement.c: (gst_element_get_bus):
+	Fix refcount in _get_bus();
+
+	* gst/gstpipeline.c: (gst_pipeline_change_state),
+	(gst_pipeline_get_clock_func):
+	Clock refcounting fixes.
+	Handle the case where preroll timed out more gracefully.
+	
+	* gst/gstsystemclock.c: (gst_system_clock_dispose):
+	Clean up the internal thread in dispose. This is needed
+	for subclasses that actually get disposed.
+	
+	* gst/schedulers/threadscheduler.c:
+	(gst_thread_scheduler_class_init), (gst_thread_scheduler_func),
+	(gst_thread_scheduler_dispose):
+	Free thread pool in dispose.
+
 2005-06-28  Andy Wingo  <wingo@pobox.com>
 
 	* tests/network-clock-utils.scm (debug, print-event): New utils.

gst/gstbus.c

@@ -185,6 +185,7 @@ gst_bus_post (GstBus * bus, GstMessage * message)
 
   GST_LOCK (bus);
   if (GST_FLAG_IS_SET (bus, GST_BUS_FLUSHING)) {
+    GST_DEBUG_OBJECT (bus, "bus is flushing");
     gst_message_unref (message);
     GST_UNLOCK (bus);
     return FALSE;
@@ -275,6 +276,8 @@ gst_bus_have_pending (GstBus * bus)
   length = g_queue_get_length (bus->queue);
   g_mutex_unlock (bus->queue_lock);
 
+  GST_DEBUG ("have %d pending", length);
+
   return (length > 0);
 }
 
@@ -299,9 +302,12 @@ gst_bus_set_flushing (GstBus * bus, gboolean flushing)
   if (flushing) {
     GST_FLAG_SET (bus, GST_BUS_FLUSHING);
 
+    GST_DEBUG ("set bus flushing");
+
     while ((message = gst_bus_pop (bus)))
       gst_message_unref (message);
   } else {
+    GST_DEBUG ("unset bus flushing");
     GST_FLAG_UNSET (bus, GST_BUS_FLUSHING);
   }
 
@@ -329,6 +335,8 @@ gst_bus_pop (GstBus * bus)
   message = g_queue_pop_head (bus->queue);
   g_mutex_unlock (bus->queue_lock);
 
+  GST_DEBUG ("pop on bus, got message %p", message);
+
   return message;
 }
 
@@ -354,6 +362,8 @@ gst_bus_peek (GstBus * bus)
   message = g_queue_peek_head (bus->queue);
   g_mutex_unlock (bus->queue_lock);
 
+  GST_DEBUG ("peek on bus, got message %p", message);
+
   return message;
 }
 
@@ -386,20 +396,20 @@ typedef struct
   GstBus *bus;
 } GstBusSource;
 
-gboolean
+static gboolean
 gst_bus_source_prepare (GSource * source, gint * timeout)
 {
   *timeout = -1;
   return gst_bus_have_pending (((GstBusSource *) source)->bus);
 }
 
-gboolean
+static gboolean
 gst_bus_source_check (GSource * source)
 {
   return gst_bus_have_pending (((GstBusSource *) source)->bus);
 }
 
-gboolean
+static gboolean
 gst_bus_source_dispatch (GSource * source, GSourceFunc callback,
     gpointer user_data)
 {
@@ -412,6 +422,8 @@ gst_bus_source_dispatch (GSource * source, GSourceFunc callback,
 
   message = gst_bus_peek (bsource->bus);
 
+  GST_DEBUG ("have message %p", message);
+
   g_return_val_if_fail (message != NULL, TRUE);
 
   if (!handler) {
@@ -420,15 +432,28 @@ gst_bus_source_dispatch (GSource * source, GSourceFunc callback,
     return FALSE;
   }
 
+  GST_DEBUG ("calling dispatch with %p", message);
+
   needs_pop = handler (bsource->bus, message, user_data);
 
-  if (needs_pop)
-    gst_message_unref (gst_bus_pop (bsource->bus));
+  GST_DEBUG ("handler returns %d", needs_pop);
+  if (needs_pop) {
+    message = gst_bus_pop (bsource->bus);
+    if (message) {
+      gst_message_unref (message);
+    } else {
+      /* after executing the handler, the app could have disposed
+       * the pipeline and set the bus to flushing. It is possible
+       * then that there are no more messages on the bus. this is
+       * not a problem. */
+      GST_DEBUG ("handler requested pop but no message on the bus");
+    }
+  }
 
   return TRUE;
 }
 
-void
+static void
 gst_bus_source_finalize (GSource * source)
 {
   GstBusSource *bsource = (GstBusSource *) source;

gst/gstelement.c

@@ -2307,7 +2307,7 @@ gst_element_set_bus (GstElement * element, GstBus * bus)
  *
  * Returns the bus of the element.
  *
- * Returns: the element's #GstBus.
+ * Returns: the element's #GstBus. unref after usage.
  *
  * MT safe.
  */
@@ -2320,6 +2320,7 @@ gst_element_get_bus (GstElement * element)
 
   GST_LOCK (element);
   result = GST_ELEMENT_BUS (element);
+  gst_object_ref (result);
   GST_UNLOCK (element);
 
   return result;

gst/gstpipeline.c

@@ -395,6 +395,7 @@ gst_pipeline_change_state (GstElement * element)
 
       clock = gst_element_get_clock (element);
       gst_element_set_clock (element, clock);
+      gst_object_unref (clock);
       pipeline->eosed = NULL;
       break;
     }
@@ -452,8 +453,7 @@ gst_pipeline_change_state (GstElement * element)
    * intermediate state.
    * FIXME this can block forever, better do this in a worker
    * thread or use a timeout? */
-  if (result == GST_STATE_ASYNC &&
-      (GST_STATE_FINAL (pipeline) != GST_STATE_PENDING (pipeline))) {
+  if (result == GST_STATE_ASYNC) {
     GTimeVal *timeval, timeout;
 
     GST_STATE_UNLOCK (pipeline);
@@ -468,6 +468,11 @@ gst_pipeline_change_state (GstElement * element)
     GST_UNLOCK (pipeline);
 
     result = gst_element_get_state (element, NULL, NULL, timeval);
+    if (result == GST_STATE_ASYNC) {
+      GST_WARNING ("timeout in PREROLL, forcing next state change");
+      g_warning ("timeout in PREROLL, forcing next state change");
+      result = GST_STATE_SUCCESS;
+    }
     GST_STATE_LOCK (pipeline);
   }
 
@@ -528,10 +533,7 @@ gst_pipeline_get_clock_func (GstElement * element)
     /* no clock, use a system clock */
     if (!clock) {
       clock = gst_system_clock_obtain ();
-      /* we unref since this function is not supposed to increase refcount
-       * of clock object returned; this is ok since the systemclock always
-       * has a refcount of at least one in the current code. */
-      gst_object_unref (clock);
+
       GST_CAT_DEBUG (GST_CAT_CLOCK, "pipeline obtained system clock: %p (%s)",
           clock, clock ? GST_STR_NULL (GST_OBJECT_NAME (clock)) : "-");
     } else {

gst/gstsystemclock.c

@@ -137,6 +137,29 @@ gst_system_clock_dispose (GObject * object)
 
     /* no parent dispose here, this is bad enough already */
   } else {
+    GstSystemClock *sysclock = GST_SYSTEM_CLOCK (clock);
+    GList *entries;
+
+    /* else we have to stop the thread */
+    GST_LOCK (clock);
+    sysclock->stopping = TRUE;
+    /* unschedule all entries */
+    for (entries = clock->entries; entries; entries = g_list_next (entries)) {
+      GstClockEntry *entry = (GstClockEntry *) entries->data;
+
+      GST_CAT_DEBUG (GST_CAT_CLOCK, "unscheduling entry %p", entry);
+      entry->status = GST_CLOCK_UNSCHEDULED;
+    }
+    g_list_free (clock->entries);
+    clock->entries = NULL;
+    GST_CLOCK_SIGNAL (clock);
+    GST_UNLOCK (clock);
+
+    if (sysclock->thread)
+      g_thread_join (sysclock->thread);
+    sysclock->thread = NULL;
+    GST_CAT_DEBUG (GST_CAT_CLOCK, "joined thread");
+
     G_OBJECT_CLASS (parent_class)->dispose (object);
   }
 }

gst/schedulers/threadscheduler.c

@@ -226,6 +226,7 @@ gst_thread_scheduler_task_pause (GstTask * task)
 
 static void gst_thread_scheduler_class_init (gpointer g_class, gpointer data);
 static void gst_thread_scheduler_init (GstThreadScheduler * object);
+static void gst_thread_scheduler_dispose (GObject * object);
 
 GType
 gst_thread_scheduler_get_type (void)
@@ -257,11 +258,18 @@ static void gst_thread_scheduler_reset (GstScheduler * sched);
 static GstTask *gst_thread_scheduler_create_task (GstScheduler * sched,
     GstTaskFunction func, gpointer data);
 
+static GObjectClass *parent_class = NULL;
+
 static void
 gst_thread_scheduler_class_init (gpointer klass, gpointer class_data)
 {
+  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
   GstSchedulerClass *scheduler = GST_SCHEDULER_CLASS (klass);
 
+  parent_class = g_type_class_ref (GST_TYPE_SCHEDULER);
+
+  gobject_class->dispose = gst_thread_scheduler_dispose;
+
   scheduler->setup = gst_thread_scheduler_setup;
   scheduler->reset = gst_thread_scheduler_reset;
   scheduler->create_task = gst_thread_scheduler_create_task;
@@ -312,6 +320,17 @@ gst_thread_scheduler_init (GstThreadScheduler * scheduler)
   scheduler->pool = g_thread_pool_new (
       (GFunc) gst_thread_scheduler_func, scheduler, -1, FALSE, NULL);
 }
+static void
+gst_thread_scheduler_dispose (GObject * object)
+{
+  GstThreadScheduler *scheduler = GST_THREAD_SCHEDULER (object);
+
+  if (scheduler->pool) {
+    g_thread_pool_free (scheduler->pool, FALSE, TRUE);
+    scheduler->pool = NULL;
+  }
+  G_OBJECT_CLASS (parent_class)->dispose (object);
+}
 
 static GstTask *
 gst_thread_scheduler_create_task (GstScheduler * sched, GstTaskFunction func,