/* GStreamer

 * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>

 *                    2005 Wim Taymans <wim@fluendo.com>

 *

 * gsttask.c: Streaming tasks

 *

 * 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.

 */

/**

 * SECTION:gsttask

 * @short_description: Abstraction of GStreamer streaming threads.

 * @see_also: #GstElement, #GstPad

 *

 * #GstTask is used by #GstElement and #GstPad to provide the data passing

 * threads in a #GstPipeline.

 *

 * A #GstPad will typically start a #GstTask to push or pull data to/from the

 * peer pads. Most source elements start a #GstTask to push data. In some cases

 * a demuxer element can start a #GstTask to pull data from a peer element. This

 * is typically done when the demuxer can perform random access on the upstream

 * peer element for improved performance.

 *

 * Although convenience functions exist on #GstPad to start/pause/stop tasks, it 

 * might sometimes be needed to create a #GstTask manually if it is not related to

 * a #GstPad.

 *

 * Before the #GstTask can be run, it needs a #GStaticRecMutex that can be set with

 * gst_task_set_lock().

 *

 * The task can be started, paused and stopped with gst_task_start(), gst_task_pause()

 * and gst_task_stop() respectively.

 *

 * A #GstTask will repeadedly call the #GstTaskFunction with the user data

 * that was provided when creating the task with gst_task_create(). Before calling

 * the function it will acquire the provided lock.

 *

 * Stopping a task with gst_task_stop() will not immediatly make sure the task is

 * not running anymore. Use gst_task_join() to make sure the task is completely

 * stopped and the thread is stopped.

 *

 * After creating a #GstTask, use gst_object_unref() to free its resources. This can

 * only be done it the task is not running anymore.

 *

 * Last reviewed on 2006-02-13 (0.10.4)

 */

#include "gst_private.h"

#include "gstinfo.h"

#include "gsttask.h"

#ifdef __SYMBIAN32__

#include <glib_global.h>

#endif

GST_DEBUG_CATEGORY_STATIC (task_debug);

#define GST_CAT_DEFAULT (task_debug)

static void gst_task_class_init (GstTaskClass * klass);

static void gst_task_init (GstTask * task);

static void gst_task_finalize (GObject * object);

static void gst_task_func (GstTask * task, GstTaskClass * tclass);

static GstObjectClass *parent_class = NULL;

static GStaticMutex pool_lock = G_STATIC_MUTEX_INIT;

#ifdef __SYMBIAN32__

EXPORT_C

#endif

GType

gst_task_get_type (void)

{

  static GType _gst_task_type = 0;

  if (G_UNLIKELY (_gst_task_type == 0)) {

    static const GTypeInfo task_info = {

      sizeof (GstTaskClass),

      NULL,

      NULL,

      (GClassInitFunc) gst_task_class_init,

      NULL,

      NULL,

      sizeof (GstTask),

      0,

      (GInstanceInitFunc) gst_task_init,

      NULL

    };

    _gst_task_type =

        g_type_register_static (GST_TYPE_OBJECT, "GstTask", &task_info, 0);

    GST_DEBUG_CATEGORY_INIT (task_debug, "task", 0, "Processing tasks");

  }

  return _gst_task_type;

}

static void

gst_task_class_init (GstTaskClass * klass)

{

  GObjectClass *gobject_class;

  gobject_class = (GObjectClass *) klass;

  parent_class = g_type_class_peek_parent (klass);

  gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_task_finalize);

  klass->pool = g_thread_pool_new (

      (GFunc) gst_task_func, klass, -1, FALSE, NULL);

}

static void

gst_task_init (GstTask * task)

{

  task->running = FALSE;

  task->abidata.ABI.thread = NULL;

  task->lock = NULL;

  task->cond = g_cond_new ();

  task->state = GST_TASK_STOPPED;

}

static void

gst_task_finalize (GObject * object)

{

  GstTask *task = GST_TASK (object);

  GST_DEBUG ("task %p finalize", task);

  /* task thread cannot be running here since it holds a ref

   * to the task so that the finalize could not have happened */

  g_cond_free (task->cond);

  task->cond = NULL;

  G_OBJECT_CLASS (parent_class)->finalize (object);

}

static void

gst_task_func (GstTask * task, GstTaskClass * tclass)

{

  GStaticRecMutex *lock;

  GThread *tself;

  tself = g_thread_self ();

  GST_DEBUG ("Entering task %p, thread %p", task, tself);

  /* we have to grab the lock to get the mutex. We also

   * mark our state running so that nobody can mess with

   * the mutex. */

  GST_OBJECT_LOCK (task);

  if (task->state == GST_TASK_STOPPED)

    goto exit;

  lock = GST_TASK_GET_LOCK (task);

  if (G_UNLIKELY (lock == NULL))

    goto no_lock;

  task->abidata.ABI.thread = tself;

  GST_OBJECT_UNLOCK (task);

  /* locking order is TASK_LOCK, LOCK */

  g_static_rec_mutex_lock (lock);

  GST_OBJECT_LOCK (task);

  while (G_LIKELY (task->state != GST_TASK_STOPPED)) {

    while (G_UNLIKELY (task->state == GST_TASK_PAUSED)) {

      gint t;

      t = g_static_rec_mutex_unlock_full (lock);

      if (t <= 0) {

        g_warning ("wrong STREAM_LOCK count %d", t);

      }

      GST_TASK_SIGNAL (task);

      GST_TASK_WAIT (task);

      GST_OBJECT_UNLOCK (task);

      /* locking order.. */

      if (t > 0)

        g_static_rec_mutex_lock_full (lock, t);

      GST_OBJECT_LOCK (task);

      if (G_UNLIKELY (task->state == GST_TASK_STOPPED))

        goto done;

    }

    GST_OBJECT_UNLOCK (task);

    task->func (task->data);

    GST_OBJECT_LOCK (task);

  }

done:

  GST_OBJECT_UNLOCK (task);

  g_static_rec_mutex_unlock (lock);

  GST_OBJECT_LOCK (task);

  task->abidata.ABI.thread = NULL;

exit:

  /* now we allow messing with the lock again by setting the running flag to

   * FALSE. Together with the SIGNAL this is the sign for the _join() to 

   * complete. 

   * Note that we still have not dropped the final ref on the task. We could

   * check here if there is a pending join() going on and drop the last ref

   * before releasing the lock as we can be sure that a ref is held by the

   * caller of the join(). */

  task->running = FALSE;

  GST_TASK_SIGNAL (task);

  GST_OBJECT_UNLOCK (task);

  GST_DEBUG ("Exit task %p, thread %p", task, g_thread_self ());

  gst_object_unref (task);

  return;

no_lock:

  {

    g_warning ("starting task without a lock");

    goto exit;

  }

}

/**

 * gst_task_cleanup_all:

 *

 * Wait for all tasks to be stopped. This is mainly used internally

 * to ensure proper cleanup of internal datastructures in testsuites.

 *

 * MT safe.

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif

void

gst_task_cleanup_all (void)

{

  GstTaskClass *klass;

  if ((klass = g_type_class_peek (GST_TYPE_TASK))) {

    g_static_mutex_lock (&pool_lock);

    if (klass->pool) {

      /* Shut down all the threads, we still process the ones scheduled

       * because the unref happens in the thread function.

       * Also wait for currently running ones to finish. */

      g_thread_pool_free (klass->pool, 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);

    }

    g_static_mutex_unlock (&pool_lock);

  }

}

/**

 * gst_task_create:

 * @func: The #GstTaskFunction to use

 * @data: User data to pass to @func

 *

 * Create a new Task that will repeadedly call the provided @func

 * with @data as a parameter. Typically the task will run in

 * a new thread.

 *

 * The function cannot be changed after the task has been created. You

 * must create a new GstTask to change the function.

 *

 * Returns: A new #GstTask.

 *

 * MT safe.

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif

GstTask *

gst_task_create (GstTaskFunction func, gpointer data)

{

  GstTask *task;

  task = g_object_new (GST_TYPE_TASK, NULL);

  task->func = func;

  task->data = data;

  GST_DEBUG ("Created task %p", task);

  return task;

}

/**

 * gst_task_set_lock:

 * @task: The #GstTask to use

 * @mutex: The GMutex to use

 *

 * Set the mutex used by the task. The mutex will be acquired before

 * calling the #GstTaskFunction.

 *

 * This function has to be called before calling gst_task_pause() or

 * gst_task_start().

 *

 * MT safe.

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif

void

gst_task_set_lock (GstTask * task, GStaticRecMutex * mutex)

{

  GST_OBJECT_LOCK (task);

  if (G_UNLIKELY (task->running))

    goto is_running;

  GST_TASK_GET_LOCK (task) = mutex;

  GST_OBJECT_UNLOCK (task);

  return;

  /* ERRORS */

is_running:

  {

    GST_OBJECT_UNLOCK (task);

    g_warning ("cannot call set_lock on a running task");

  }

}

/**

 * gst_task_get_state:

 * @task: The #GstTask to query

 *

 * Get the current state of the task.

 *

 * Returns: The #GstTaskState of the task

 *

 * MT safe.

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif

GstTaskState

gst_task_get_state (GstTask * task)

{

  GstTaskState result;

  g_return_val_if_fail (GST_IS_TASK (task), GST_TASK_STOPPED);

  GST_OBJECT_LOCK (task);

  result = task->state;

  GST_OBJECT_UNLOCK (task);

  return result;

}

/**

 * gst_task_start:

 * @task: The #GstTask to start

 *

 * Starts @task. The @task must have a lock associated with it using

 * gst_task_set_lock() or thsi function will return FALSE.

 *

 * Returns: TRUE if the task could be started.

 *

 * MT safe.

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif

gboolean

gst_task_start (GstTask * task)

{

  GstTaskState old;

  g_return_val_if_fail (GST_IS_TASK (task), FALSE);

  GST_DEBUG_OBJECT (task, "Starting task %p", task);

  GST_OBJECT_LOCK (task);

  if (G_UNLIKELY (GST_TASK_GET_LOCK (task) == NULL))

    goto no_lock;

  old = task->state;

  task->state = GST_TASK_STARTED;

  switch (old) {

    case GST_TASK_STOPPED:

    {

      GstTaskClass *tclass;

      /* If the task already has a thread scheduled we don't have to do

       * anything. */

      if (task->running)

        break;

      /* new task, push on threadpool. We ref before so

       * that it remains alive while on the threadpool. */

      gst_object_ref (task);

      /* mark task as running so that a join will wait until we schedule

       * and exit the task function. */

      task->running = TRUE;

      tclass = GST_TASK_GET_CLASS (task);

      g_static_mutex_lock (&pool_lock);

      g_thread_pool_push (tclass->pool, task, NULL);

      g_static_mutex_unlock (&pool_lock);

      break;

    }

    case GST_TASK_PAUSED:

      /* PAUSE to PLAY, signal */

      GST_TASK_SIGNAL (task);

      break;

    case GST_TASK_STARTED:

      /* was OK */

      break;

  }

  GST_OBJECT_UNLOCK (task);

  return TRUE;

  /* ERRORS */

no_lock:

  {

    GST_WARNING_OBJECT (task, "starting task without a lock");

    GST_OBJECT_UNLOCK (task);

    g_warning ("starting task without a lock");

    return FALSE;

  }

}

/**

 * gst_task_stop:

 * @task: The #GstTask to stop

 *

 * Stops @task. This method merely schedules the task to stop and

 * will not wait for the task to have completely stopped. Use

 * gst_task_join() to stop and wait for completion.

 *

 * Returns: TRUE if the task could be stopped.

 *

 * MT safe.

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif

gboolean

gst_task_stop (GstTask * task)

{

  GstTaskClass *tclass;

  GstTaskState old;

  g_return_val_if_fail (GST_IS_TASK (task), FALSE);

  tclass = GST_TASK_GET_CLASS (task);

  GST_DEBUG_OBJECT (task, "Stopping task %p", task);

  GST_OBJECT_LOCK (task);

  old = task->state;

  task->state = GST_TASK_STOPPED;

  switch (old) {

    case GST_TASK_STOPPED:

      break;

    case GST_TASK_PAUSED:

      GST_TASK_SIGNAL (task);

      break;

    case GST_TASK_STARTED:

      break;

  }

  GST_OBJECT_UNLOCK (task);

  return TRUE;

}

/**

 * gst_task_pause:

 * @task: The #GstTask to pause

 *

 * Pauses @task. This method can also be called on a task in the

 * stopped state, in which case a thread will be started and will remain

 * in the paused state. This function does not wait for the task to complete

 * the paused state.

 *

 * Returns: TRUE if the task could be paused.

 *

 * MT safe.

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif 

gboolean

gst_task_pause (GstTask * task)

{

  GstTaskState old;

  g_return_val_if_fail (GST_IS_TASK (task), FALSE);

  GST_DEBUG_OBJECT (task, "Pausing task %p", task);

  GST_OBJECT_LOCK (task);

  if (G_UNLIKELY (GST_TASK_GET_LOCK (task) == NULL))

    goto no_lock;

  old = task->state;

  task->state = GST_TASK_PAUSED;

  switch (old) {

    case GST_TASK_STOPPED:

    {

      GstTaskClass *tclass;

      if (task->running)

        break;

      gst_object_ref (task);

      task->running = TRUE;