/* 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()

 *

 *

 * 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 GStaticMutex pool_lock = G_STATIC_MUTEX_INIT;

  }

}

static void

gst_task_class_init (GstTaskClass * klass)

{

  GObjectClass *gobject_class;

  gobject_class = (GObjectClass *) klass;

  gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_task_finalize);

}

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;

}

static void

{

  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

 *

 * MT safe.

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif

void

gst_task_cleanup_all (void)

{

  GstTaskClass *klass;

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

  }

}

/**

 * gst_task_create:

 * @func: The #GstTaskFunction to use

 * @data: User data to pass to @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

 *

 * 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

 *

 * 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");

  }

}