gst/gsttask.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *                    2005 Wim Taymans <wim@fluendo.com>
   4  *
   5  * gsttask.c: Streaming tasks
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  20  * Boston, MA 02110-1301, USA.
  21  */
  22
  23 /**
  24  * SECTION:gsttask
  25  * @short_description: Abstraction of GStreamer streaming threads.
  26  * @see_also: #GstElement, #GstPad
  27  *
  28  * #GstTask is used by #GstElement and #GstPad to provide the data passing
  29  * threads in a #GstPipeline.
  30  *
  31  * A #GstPad will typically start a #GstTask to push or pull data to/from the
  32  * peer pads. Most source elements start a #GstTask to push data. In some cases
  33  * a demuxer element can start a #GstTask to pull data from a peer element. This
  34  * is typically done when the demuxer can perform random access on the upstream
  35  * peer element for improved performance.
  36  *
  37  * Although convenience functions exist on #GstPad to start/pause/stop tasks, it
  38  * might sometimes be needed to create a #GstTask manually if it is not related to
  39  * a #GstPad.
  40  *
  41  * Before the #GstTask can be run, it needs a #GRecMutex that can be set with
  42  * gst_task_set_lock().
  43  *
  44  * The task can be started, paused and stopped with gst_task_start(), gst_task_pause()
  45  * and gst_task_stop() respectively or with the gst_task_set_state() function.
  46  *
  47  * A #GstTask will repeatedly call the #GstTaskFunction with the user data
  48  * that was provided when creating the task with gst_task_new(). While calling
  49  * the function it will acquire the provided lock. The provided lock is released
  50  * when the task pauses or stops.
  51  *
  52  * Stopping a task with gst_task_stop() will not immediately make sure the task is
  53  * not running anymore. Use gst_task_join() to make sure the task is completely
  54  * stopped and the thread is stopped.
  55  *
  56  * After creating a #GstTask, use gst_object_unref() to free its resources. This can
  57  * only be done when the task is not running anymore.
  58  *
  59  * Task functions can send a #GstMessage to send out-of-band data to the
  60  * application. The application can receive messages from the #GstBus in its
  61  * mainloop.
  62  *
  63  * For debugging purposes, the task will configure its object name as the thread
  64  * name on Linux. Please note that the object name should be configured before the
  65  * task is started; changing the object name after the task has been started, has
  66  * no effect on the thread name.
  67  *
  68  * Last reviewed on 2012-03-29 (0.11.3)
  69  */
  70
  71 #include "gst_private.h"
  72
  73 #include "gstinfo.h"
  74 #include "gsttask.h"
  75 #include "glib-compat-private.h"
  76
  77 #include <stdio.h>
  78
  79 #ifdef HAVE_SYS_PRCTL_H
  80 #include <sys/prctl.h>
  81 #endif
  82
  83 GST_DEBUG_CATEGORY_STATIC (task_debug);
  84 #define GST_CAT_DEFAULT (task_debug)
  85
  86 #define SET_TASK_STATE(t,s) (g_atomic_int_set (&GST_TASK_STATE(t), (s)))
  87 #define GET_TASK_STATE(t)   ((GstTaskState) g_atomic_int_get (&GST_TASK_STATE(t)))
  88
  89 #define GST_TASK_GET_PRIVATE(obj)  \
  90    (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_TASK, GstTaskPrivate))
  91
  92 struct _GstTaskPrivate
  93 {
  94   /* callbacks for managing the thread of this task */
  95   GstTaskThreadFunc enter_func;
  96   gpointer enter_user_data;
  97   GDestroyNotify enter_notify;
  98
  99   GstTaskThreadFunc leave_func;
 100   gpointer leave_user_data;
 101   GDestroyNotify leave_notify;
 102
 103   /* configured pool */
 104   GstTaskPool *pool;
 105
 106   /* remember the pool and id that is currently running. */
 107   gpointer id;
 108   GstTaskPool *pool_id;
 109 };
 110
 111 #ifdef _MSC_VER
 112 #include <windows.h>
 113
 114 struct _THREADNAME_INFO
 115 {
 116   DWORD dwType;                 // must be 0x1000
 117   LPCSTR szName;                // pointer to name (in user addr space)
 118   DWORD dwThreadID;             // thread ID (-1=caller thread)
 119   DWORD dwFlags;                // reserved for future use, must be zero
 120 };
 121 typedef struct _THREADNAME_INFO THREADNAME_INFO;
 122
 123 void
 124 SetThreadName (DWORD dwThreadID, LPCSTR szThreadName)
 125 {
 126   THREADNAME_INFO info;
 127   info.dwType = 0x1000;
 128   info.szName = szThreadName;
 129   info.dwThreadID = dwThreadID;
 130   info.dwFlags = 0;
 131
 132   __try {
 133     RaiseException (0x406D1388, 0, sizeof (info) / sizeof (DWORD),
 134         (DWORD *) & info);
 135   }
 136   __except (EXCEPTION_CONTINUE_EXECUTION) {
 137   }
 138 }
 139 #endif
 140
 141 static void gst_task_finalize (GObject * object);
 142
 143 static void gst_task_func (GstTask * task);
 144
 145 static GMutex pool_lock;
 146
 147 #define _do_init \
 148 { \
 149   GST_DEBUG_CATEGORY_INIT (task_debug, "task", 0, "Processing tasks"); \
 150 }
 151
 152 G_DEFINE_TYPE_WITH_CODE (GstTask, gst_task, GST_TYPE_OBJECT, _do_init);
 153
 154 static void
 155 init_klass_pool (GstTaskClass * klass)
 156 {
 157   g_mutex_lock (&pool_lock);
 158   if (klass->pool) {
 159     gst_task_pool_cleanup (klass->pool);
 160     gst_object_unref (klass->pool);
 161   }
 162   klass->pool = gst_task_pool_new ();
 163   gst_task_pool_prepare (klass->pool, NULL);
 164   g_mutex_unlock (&pool_lock);
 165 }
 166
 167 static void
 168 gst_task_class_init (GstTaskClass * klass)
 169 {
 170   GObjectClass *gobject_class;
 171
 172   gobject_class = (GObjectClass *) klass;
 173
 174   g_type_class_add_private (klass, sizeof (GstTaskPrivate));
 175
 176   gobject_class->finalize = gst_task_finalize;
 177
 178   init_klass_pool (klass);
 179 }
 180
 181 static void
 182 gst_task_init (GstTask * task)
 183 {
 184   GstTaskClass *klass;
 185
 186   klass = GST_TASK_GET_CLASS (task);
 187
 188   task->priv = GST_TASK_GET_PRIVATE (task);
 189   task->running = FALSE;
 190   task->thread = NULL;
 191   task->lock = NULL;
 192   g_cond_init (&task->cond);
 193   SET_TASK_STATE (task, GST_TASK_STOPPED);
 194
 195   /* use the default klass pool for this task, users can
 196    * override this later */
 197   g_mutex_lock (&pool_lock);
 198   task->priv->pool = gst_object_ref (klass->pool);
 199   g_mutex_unlock (&pool_lock);
 200
 201   /* clear floating flag */
 202   gst_object_ref_sink (task);
 203 }
 204
 205 static void
 206 gst_task_finalize (GObject * object)
 207 {
 208   GstTask *task = GST_TASK (object);
 209   GstTaskPrivate *priv = task->priv;
 210
 211   GST_DEBUG ("task %p finalize", task);
 212
 213   if (priv->enter_notify)
 214     priv->enter_notify (priv->enter_user_data);
 215
 216   if (priv->leave_notify)
 217     priv->leave_notify (priv->leave_user_data);
 218
 219   if (task->notify)
 220     task->notify (task->user_data);
 221
 222   gst_object_unref (priv->pool);
 223
 224   /* task thread cannot be running here since it holds a ref
 225    * to the task so that the finalize could not have happened */
 226   g_cond_clear (&task->cond);
 227
 228   G_OBJECT_CLASS (gst_task_parent_class)->finalize (object);
 229 }
 230
 231 /* should be called with the object LOCK */
 232 static void
 233 gst_task_configure_name (GstTask * task)
 234 {
 235 #if defined(HAVE_SYS_PRCTL_H) && defined(PR_SET_NAME)
 236   const gchar *name;
 237   gchar thread_name[17] = { 0, };
 238
 239   GST_OBJECT_LOCK (task);
 240   name = GST_OBJECT_NAME (task);
 241
 242   /* set the thread name to something easily identifiable */
 243   if (!snprintf (thread_name, 17, "%s", GST_STR_NULL (name))) {
 244     GST_DEBUG_OBJECT (task, "Could not create thread name for '%s'", name);
 245   } else {
 246     GST_DEBUG_OBJECT (task, "Setting thread name to '%s'", thread_name);
 247     if (prctl (PR_SET_NAME, (unsigned long int) thread_name, 0, 0, 0))
 248       GST_DEBUG_OBJECT (task, "Failed to set thread name");
 249   }
 250   GST_OBJECT_UNLOCK (task);
 251 #endif
 252 #ifdef _MSC_VER
 253   const gchar *name;
 254   name = GST_OBJECT_NAME (task);
 255
 256   /* set the thread name to something easily identifiable */
 257   GST_DEBUG_OBJECT (task, "Setting thread name to '%s'", name);
 258   SetThreadName (-1, name);
 259 #endif
 260 }
 261
 262 static void
 263 gst_task_func (GstTask * task)
 264 {
 265   GRecMutex *lock;
 266   GThread *tself;
 267   GstTaskPrivate *priv;
 268
 269   priv = task->priv;
 270
 271   tself = g_thread_self ();
 272
 273   GST_DEBUG ("Entering task %p, thread %p", task, tself);
 274
 275   /* we have to grab the lock to get the mutex. We also
 276    * mark our state running so that nobody can mess with
 277    * the mutex. */
 278   GST_OBJECT_LOCK (task);
 279   if (GET_TASK_STATE (task) == GST_TASK_STOPPED)
 280     goto exit;
 281   lock = GST_TASK_GET_LOCK (task);
 282   if (G_UNLIKELY (lock == NULL))
 283     goto no_lock;
 284   task->thread = tself;
 285   GST_OBJECT_UNLOCK (task);
 286
 287   /* fire the enter_func callback when we need to */
 288   if (priv->enter_func)
 289     priv->enter_func (task, tself, priv->enter_user_data);
 290
 291   /* locking order is TASK_LOCK, LOCK */
 292   g_rec_mutex_lock (lock);
 293   /* configure the thread name now */
 294   gst_task_configure_name (task);
 295
 296   while (G_LIKELY (GET_TASK_STATE (task) != GST_TASK_STOPPED)) {
 297     if (G_UNLIKELY (GET_TASK_STATE (task) == GST_TASK_PAUSED)) {
 298       GST_OBJECT_LOCK (task);
 299       while (G_UNLIKELY (GST_TASK_STATE (task) == GST_TASK_PAUSED)) {
 300         g_rec_mutex_unlock (lock);
 301
 302         GST_TASK_SIGNAL (task);
 303         GST_INFO_OBJECT (task, "Task going to paused");
 304         GST_TASK_WAIT (task);
 305         GST_INFO_OBJECT (task, "Task resume from paused");
 306         GST_OBJECT_UNLOCK (task);
 307         /* locking order.. */
 308         g_rec_mutex_lock (lock);
 309
 310         GST_OBJECT_LOCK (task);
 311         if (G_UNLIKELY (GET_TASK_STATE (task) == GST_TASK_STOPPED)) {
 312           GST_OBJECT_UNLOCK (task);
 313           goto done;
 314         }
 315       }
 316       GST_OBJECT_UNLOCK (task);
 317     }
 318
 319     task->func (task->user_data);
 320   }
 321 done:
 322   g_rec_mutex_unlock (lock);
 323
 324   GST_OBJECT_LOCK (task);
 325   task->thread = NULL;
 326
 327 exit:
 328   if (priv->leave_func) {
 329     /* fire the leave_func callback when we need to. We need to do this before
 330      * we signal the task and with the task lock released. */
 331     GST_OBJECT_UNLOCK (task);
 332     priv->leave_func (task, tself, priv->leave_user_data);
 333     GST_OBJECT_LOCK (task);
 334   }
 335   /* now we allow messing with the lock again by setting the running flag to
 336    * FALSE. Together with the SIGNAL this is the sign for the _join() to
 337    * complete.
 338    * Note that we still have not dropped the final ref on the task. We could
 339    * check here if there is a pending join() going on and drop the last ref
 340    * before releasing the lock as we can be sure that a ref is held by the
 341    * caller of the join(). */
 342   task->running = FALSE;
 343   GST_TASK_SIGNAL (task);
 344   GST_OBJECT_UNLOCK (task);
 345
 346   GST_DEBUG ("Exit task %p, thread %p", task, g_thread_self ());
 347
 348   gst_object_unref (task);
 349   return;
 350
 351 no_lock:
 352   {
 353     g_warning ("starting task without a lock");
 354     goto exit;
 355   }
 356 }
 357
 358 /**
 359  * gst_task_cleanup_all:
 360  *
 361  * Wait for all tasks to be stopped. This is mainly used internally
 362  * to ensure proper cleanup of internal data structures in test suites.
 363  *
 364  * MT safe.
 365  */
 366 void
 367 gst_task_cleanup_all (void)
 368 {
 369   GstTaskClass *klass;
 370
 371   if ((klass = g_type_class_peek (GST_TYPE_TASK))) {
 372     init_klass_pool (klass);
 373   }
 374 }
 375
 376 /**
 377  * gst_task_new:
 378  * @func: The #GstTaskFunction to use
 379  * @user_data: User data to pass to @func
 380  * @notify: the function to call when @user_data is no longer needed.
 381  *
 382  * Create a new Task that will repeatedly call the provided @func
 383  * with @user_data as a parameter. Typically the task will run in
 384  * a new thread.
 385  *
 386  * The function cannot be changed after the task has been created. You
 387  * must create a new #GstTask to change the function.
 388  *
 389  * This function will not yet create and start a thread. Use gst_task_start() or
 390  * gst_task_pause() to create and start the GThread.
 391  *
 392  * Before the task can be used, a #GRecMutex must be configured using the
 393  * gst_task_set_lock() function. This lock will always be acquired while
 394  * @func is called.
 395  *
 396  * Returns: (transfer full): A new #GstTask.
 397  *
 398  * MT safe.
 399  */
 400 GstTask *
 401 gst_task_new (GstTaskFunction func, gpointer user_data, GDestroyNotify notify)
 402 {
 403   GstTask *task;
 404
 405   task = g_object_newv (GST_TYPE_TASK, 0, NULL);
 406   task->func = func;
 407   task->user_data = user_data;
 408   task->notify = notify;
 409
 410   GST_DEBUG ("Created task %p", task);
 411
 412   return task;
 413 }
 414
 415 /**
 416  * gst_task_set_lock:
 417  * @task: The #GstTask to use
 418  * @mutex: The #GRecMutex to use
 419  *
 420  * Set the mutex used by the task. The mutex will be acquired before
 421  * calling the #GstTaskFunction.
 422  *
 423  * This function has to be called before calling gst_task_pause() or
 424  * gst_task_start().
 425  *
 426  * MT safe.
 427  */
 428 void
 429 gst_task_set_lock (GstTask * task, GRecMutex * mutex)
 430 {
 431   GST_OBJECT_LOCK (task);
 432   if (G_UNLIKELY (task->running))
 433     goto is_running;
 434   GST_INFO ("setting stream lock %p on task %p", mutex, task);
 435   GST_TASK_GET_LOCK (task) = mutex;
 436   GST_OBJECT_UNLOCK (task);
 437
 438   return;
 439
 440   /* ERRORS */
 441 is_running:
 442   {
 443     GST_OBJECT_UNLOCK (task);
 444     g_warning ("cannot call set_lock on a running task");
 445   }
 446 }
 447
 448 /**
 449  * gst_task_get_pool:
 450  * @task: a #GstTask
 451  *
 452  * Get the #GstTaskPool that this task will use for its streaming
 453  * threads.
 454  *
 455  * MT safe.
 456  *
 457  * Returns: (transfer full): the #GstTaskPool used by @task. gst_object_unref()
 458  * after usage.
 459  */
 460 GstTaskPool *
 461 gst_task_get_pool (GstTask * task)
 462 {
 463   GstTaskPool *result;
 464   GstTaskPrivate *priv;
 465
 466   g_return_val_if_fail (GST_IS_TASK (task), NULL);
 467
 468   priv = task->priv;
 469
 470   GST_OBJECT_LOCK (task);
 471   result = gst_object_ref (priv->pool);
 472   GST_OBJECT_UNLOCK (task);
 473
 474   return result;
 475 }
 476
 477 /**
 478  * gst_task_set_pool:
 479  * @task: a #GstTask
 480  * @pool: (transfer none): a #GstTaskPool
 481  *
 482  * Set @pool as the new GstTaskPool for @task. Any new streaming threads that
 483  * will be created by @task will now use @pool.
 484  *
 485  * MT safe.
 486  */
 487 void
 488 gst_task_set_pool (GstTask * task, GstTaskPool * pool)
 489 {
 490   GstTaskPool *old;
 491   GstTaskPrivate *priv;
 492
 493   g_return_if_fail (GST_IS_TASK (task));
 494   g_return_if_fail (GST_IS_TASK_POOL (pool));
 495
 496   priv = task->priv;
 497
 498   GST_OBJECT_LOCK (task);
 499   if (priv->pool != pool) {
 500     old = priv->pool;
 501     priv->pool = gst_object_ref (pool);
 502   } else
 503     old = NULL;
 504   GST_OBJECT_UNLOCK (task);
 505
 506   if (old)
 507     gst_object_unref (old);
 508 }
 509
 510 /**
 511  * gst_task_set_enter_callback:
 512  * @task: The #GstTask to use
 513  * @enter_func: (in): a #GstTaskThreadFunc
 514  * @user_data: user data passed to @enter_func
 515  * @notify: called when @user_data is no longer referenced
 516  *
 517  * Call @enter_func when the task function of @task is entered. @user_data will
 518  * be passed to @enter_func and @notify will be called when @user_data is no
 519  * longer referenced.
 520  */
 521 void
 522 gst_task_set_enter_callback (GstTask * task, GstTaskThreadFunc enter_func,
 523     gpointer user_data, GDestroyNotify notify)
 524 {
 525   GDestroyNotify old_notify;
 526
 527   g_return_if_fail (task != NULL);
 528   g_return_if_fail (GST_IS_TASK (task));
 529
 530   GST_OBJECT_LOCK (task);
 531   if ((old_notify = task->priv->enter_notify)) {
 532     gpointer old_data = task->priv->enter_user_data;
 533
 534     task->priv->enter_user_data = NULL;
 535     task->priv->enter_notify = NULL;
 536     GST_OBJECT_UNLOCK (task);
 537
 538     old_notify (old_data);
 539
 540     GST_OBJECT_LOCK (task);
 541   }
 542   task->priv->enter_func = enter_func;
 543   task->priv->enter_user_data = user_data;
 544   task->priv->enter_notify = notify;
 545   GST_OBJECT_UNLOCK (task);
 546 }
 547
 548 /**
 549  * gst_task_set_leave_callback:
 550  * @task: The #GstTask to use
 551  * @leave_func: (in): a #GstTaskThreadFunc
 552  * @user_data: user data passed to @leave_func
 553  * @notify: called when @user_data is no longer referenced
 554  *
 555  * Call @leave_func when the task function of @task is left. @user_data will
 556  * be passed to @leave_func and @notify will be called when @user_data is no
 557  * longer referenced.
 558  */
 559 void
 560 gst_task_set_leave_callback (GstTask * task, GstTaskThreadFunc leave_func,
 561     gpointer user_data, GDestroyNotify notify)
 562 {
 563   GDestroyNotify old_notify;
 564
 565   g_return_if_fail (task != NULL);
 566   g_return_if_fail (GST_IS_TASK (task));
 567
 568   GST_OBJECT_LOCK (task);
 569   if ((old_notify = task->priv->leave_notify)) {
 570     gpointer old_data = task->priv->leave_user_data;
 571
 572     task->priv->leave_user_data = NULL;
 573     task->priv->leave_notify = NULL;
 574     GST_OBJECT_UNLOCK (task);
 575
 576     old_notify (old_data);
 577
 578     GST_OBJECT_LOCK (task);
 579   }
 580   task->priv->leave_func = leave_func;
 581   task->priv->leave_user_data = user_data;
 582   task->priv->leave_notify = notify;
 583   GST_OBJECT_UNLOCK (task);
 584 }
 585
 586 /**
 587  * gst_task_get_state:
 588  * @task: The #GstTask to query
 589  *
 590  * Get the current state of the task.
 591  *
 592  * Returns: The #GstTaskState of the task
 593  *
 594  * MT safe.
 595  */
 596 GstTaskState
 597 gst_task_get_state (GstTask * task)
 598 {
 599   GstTaskState result;
 600
 601   g_return_val_if_fail (GST_IS_TASK (task), GST_TASK_STOPPED);
 602
 603   result = GET_TASK_STATE (task);
 604
 605   return result;
 606 }
 607
 608 /* make sure the task is running and start a thread if it's not.
 609  * This function must be called with the task LOCK. */
 610 static gboolean
 611 start_task (GstTask * task)
 612 {
 613   gboolean res = TRUE;
 614   GError *error = NULL;
 615   GstTaskPrivate *priv;
 616
 617   priv = task->priv;
 618
 619   /* new task, We ref before so that it remains alive while
 620    * the thread is running. */
 621   gst_object_ref (task);
 622   /* mark task as running so that a join will wait until we schedule
 623    * and exit the task function. */
 624   task->running = TRUE;
 625
 626   /* push on the thread pool, we remember the original pool because the user
 627    * could change it later on and then we join to the wrong pool. */
 628   priv->pool_id = gst_object_ref (priv->pool);
 629   priv->id =
 630       gst_task_pool_push (priv->pool_id, (GstTaskPoolFunction) gst_task_func,
 631       task, &error);
 632
 633   if (error != NULL) {
 634     g_warning ("failed to create thread: %s", error->message);
 635     g_error_free (error);
 636     res = FALSE;
 637   }
 638   return res;
 639 }
 640
 641
 642 /**
 643  * gst_task_set_state:
 644  * @task: a #GstTask
 645  * @state: the new task state
 646  *
 647  * Sets the state of @task to @state.
 648  *
 649  * The @task must have a lock associated with it using
 650  * gst_task_set_lock() when going to GST_TASK_STARTED or GST_TASK_PAUSED or
 651  * this function will return %FALSE.
 652  *
 653  * MT safe.
 654  *
 655  * Returns: %TRUE if the state could be changed.
 656  */
 657 gboolean
 658 gst_task_set_state (GstTask * task, GstTaskState state)
 659 {
 660   GstTaskState old;
 661   gboolean res = TRUE;
 662
 663   g_return_val_if_fail (GST_IS_TASK (task), FALSE);
 664
 665   GST_DEBUG_OBJECT (task, "Changing task %p to state %d", task, state);
 666
 667   GST_OBJECT_LOCK (task);
 668   if (state != GST_TASK_STOPPED)
 669     if (G_UNLIKELY (GST_TASK_GET_LOCK (task) == NULL))
 670       goto no_lock;
 671
 672   /* if the state changed, do our thing */
 673   old = GET_TASK_STATE (task);
 674   if (old != state) {
 675     SET_TASK_STATE (task, state);
 676     switch (old) {
 677       case GST_TASK_STOPPED:
 678         /* If the task already has a thread scheduled we don't have to do
 679          * anything. */
 680         if (G_UNLIKELY (!task->running))
 681           res = start_task (task);
 682         break;
 683       case GST_TASK_PAUSED:
 684         /* when we are paused, signal to go to the new state */
 685         GST_TASK_SIGNAL (task);
 686         break;
 687       case GST_TASK_STARTED:
 688         /* if we were started, we'll go to the new state after the next
 689          * iteration. */
 690         break;
 691     }
 692   }
 693   GST_OBJECT_UNLOCK (task);
 694
 695   return res;
 696
 697   /* ERRORS */
 698 no_lock:
 699   {
 700     GST_WARNING_OBJECT (task, "state %d set on task without a lock", state);
 701     GST_OBJECT_UNLOCK (task);
 702     g_warning ("task without a lock can't be set to state %d", state);
 703     return FALSE;
 704   }
 705 }
 706
 707 /**
 708  * gst_task_start:
 709  * @task: The #GstTask to start
 710  *
 711  * Starts @task. The @task must have a lock associated with it using
 712  * gst_task_set_lock() or this function will return %FALSE.
 713  *
 714  * Returns: %TRUE if the task could be started.
 715  *
 716  * MT safe.
 717  */
 718 gboolean
 719 gst_task_start (GstTask * task)
 720 {
 721   return gst_task_set_state (task, GST_TASK_STARTED);
 722 }
 723
 724 /**
 725  * gst_task_stop:
 726  * @task: The #GstTask to stop
 727  *
 728  * Stops @task. This method merely schedules the task to stop and
 729  * will not wait for the task to have completely stopped. Use
 730  * gst_task_join() to stop and wait for completion.
 731  *
 732  * Returns: %TRUE if the task could be stopped.
 733  *
 734  * MT safe.
 735  */
 736 gboolean
 737 gst_task_stop (GstTask * task)
 738 {
 739   return gst_task_set_state (task, GST_TASK_STOPPED);
 740 }
 741
 742 /**
 743  * gst_task_pause:
 744  * @task: The #GstTask to pause
 745  *
 746  * Pauses @task. This method can also be called on a task in the
 747  * stopped state, in which case a thread will be started and will remain
 748  * in the paused state. This function does not wait for the task to complete
 749  * the paused state.
 750  *
 751  * Returns: %TRUE if the task could be paused.
 752  *
 753  * MT safe.
 754  */
 755 gboolean
 756 gst_task_pause (GstTask * task)
 757 {
 758   return gst_task_set_state (task, GST_TASK_PAUSED);
 759 }
 760
 761 /**
 762  * gst_task_join:
 763  * @task: The #GstTask to join
 764  *
 765  * Joins @task. After this call, it is safe to unref the task
 766  * and clean up the lock set with gst_task_set_lock().
 767  *
 768  * The task will automatically be stopped with this call.
 769  *
 770  * This function cannot be called from within a task function as this
 771  * would cause a deadlock. The function will detect this and print a
 772  * g_warning.
 773  *
 774  * Returns: %TRUE if the task could be joined.
 775  *
 776  * MT safe.
 777  */
 778 gboolean
 779 gst_task_join (GstTask * task)
 780 {
 781   GThread *tself;
 782   GstTaskPrivate *priv;
 783   gpointer id;
 784   GstTaskPool *pool = NULL;
 785
 786   priv = task->priv;
 787
 788   g_return_val_if_fail (GST_IS_TASK (task), FALSE);
 789
 790   tself = g_thread_self ();
 791
 792   GST_DEBUG_OBJECT (task, "Joining task %p, thread %p", task, tself);
 793
 794   /* we don't use a real thread join here because we are using
 795    * thread pools */
 796   GST_OBJECT_LOCK (task);
 797   if (G_UNLIKELY (tself == task->thread))
 798     goto joining_self;
 799   SET_TASK_STATE (task, GST_TASK_STOPPED);
 800   /* signal the state change for when it was blocked in PAUSED. */
 801   GST_TASK_SIGNAL (task);
 802   /* we set the running flag when pushing the task on the thread pool.
 803    * This means that the task function might not be called when we try
 804    * to join it here. */
 805   while (G_LIKELY (task->running))
 806     GST_TASK_WAIT (task);
 807   /* clean the thread */
 808   task->thread = NULL;
 809   /* get the id and pool to join */
 810   pool = priv->pool_id;
 811   id = priv->id;
 812   priv->pool_id = NULL;
 813   priv->id = NULL;
 814   GST_OBJECT_UNLOCK (task);
 815
 816   if (pool) {
 817     if (id)
 818       gst_task_pool_join (pool, id);
 819     gst_object_unref (pool);
 820   }
 821
 822   GST_DEBUG_OBJECT (task, "Joined task %p", task);
 823
 824   return TRUE;
 825
 826   /* ERRORS */
 827 joining_self:
 828   {
 829     GST_WARNING_OBJECT (task, "trying to join task from its thread");
 830     GST_OBJECT_UNLOCK (task);
 831     g_warning ("\nTrying to join task %p from its thread would deadlock.\n"
 832         "You cannot change the state of an element from its streaming\n"
 833         "thread. Use g_idle_add() or post a GstMessage on the bus to\n"
 834         "schedule the state change from the main thread.\n", task);
 835     return FALSE;
 836   }
 837 }