Merge branch 'master' into 0.11
[platform/upstream/gstreamer.git] / 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., 59 Temple Place - Suite 330,
20  * Boston, MA 02111-1307, 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 #GStaticRecMutex 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_create(). 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 it 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 perposes, 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 2010-03-15 (0.10.29)
69  */
70
71 #include "gst_private.h"
72
73 #include "gstinfo.h"
74 #include "gsttask.h"
75
76 #include <stdio.h>
77
78 #ifdef HAVE_SYS_PRCTL_H
79 #include <sys/prctl.h>
80 #endif
81
82 GST_DEBUG_CATEGORY_STATIC (task_debug);
83 #define GST_CAT_DEFAULT (task_debug)
84
85 #define SET_TASK_STATE(t,s) (g_atomic_int_set (&GST_TASK_STATE(t), (s)))
86 #define GET_TASK_STATE(t)   (g_atomic_int_get (&GST_TASK_STATE(t)))
87
88 #define GST_TASK_GET_PRIVATE(obj)  \
89    (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_TASK, GstTaskPrivate))
90
91 struct _GstTaskPrivate
92 {
93   /* callbacks for managing the thread of this task */
94   GstTaskThreadCallbacks thr_callbacks;
95   gpointer thr_user_data;
96   GDestroyNotify thr_notify;
97
98   gboolean prio_set;
99   GThreadPriority priority;
100
101   /* configured pool */
102   GstTaskPool *pool;
103
104   /* remember the pool and id that is currently running. */
105   gpointer id;
106   GstTaskPool *pool_id;
107 };
108
109 #ifdef _MSC_VER
110 #include <windows.h>
111
112 struct _THREADNAME_INFO
113 {
114   DWORD dwType;                 // must be 0x1000
115   LPCSTR szName;                // pointer to name (in user addr space)
116   DWORD dwThreadID;             // thread ID (-1=caller thread)
117   DWORD dwFlags;                // reserved for future use, must be zero
118 };
119 typedef struct _THREADNAME_INFO THREADNAME_INFO;
120
121 void
122 SetThreadName (DWORD dwThreadID, LPCSTR szThreadName)
123 {
124   THREADNAME_INFO info;
125   info.dwType = 0x1000;
126   info.szName = szThreadName;
127   info.dwThreadID = dwThreadID;
128   info.dwFlags = 0;
129
130   __try {
131     RaiseException (0x406D1388, 0, sizeof (info) / sizeof (DWORD),
132         (DWORD *) & info);
133   }
134   __except (EXCEPTION_CONTINUE_EXECUTION) {
135   }
136 }
137 #endif
138
139 static void gst_task_finalize (GObject * object);
140
141 static void gst_task_func (GstTask * task);
142
143 static GStaticMutex pool_lock = G_STATIC_MUTEX_INIT;
144
145 #define _do_init \
146 { \
147   GST_DEBUG_CATEGORY_INIT (task_debug, "task", 0, "Processing tasks"); \
148 }
149
150 G_DEFINE_TYPE_WITH_CODE (GstTask, gst_task, GST_TYPE_OBJECT, _do_init);
151
152 static void
153 init_klass_pool (GstTaskClass * klass)
154 {
155   g_static_mutex_lock (&pool_lock);
156   if (klass->pool) {
157     gst_task_pool_cleanup (klass->pool);
158     gst_object_unref (klass->pool);
159   }
160   klass->pool = gst_task_pool_new ();
161   gst_task_pool_prepare (klass->pool, NULL);
162   g_static_mutex_unlock (&pool_lock);
163 }
164
165 static void
166 gst_task_class_init (GstTaskClass * klass)
167 {
168   GObjectClass *gobject_class;
169
170   gobject_class = (GObjectClass *) klass;
171
172   g_type_class_add_private (klass, sizeof (GstTaskPrivate));
173
174   gobject_class->finalize = gst_task_finalize;
175
176   init_klass_pool (klass);
177 }
178
179 static void
180 gst_task_init (GstTask * task)
181 {
182   GstTaskClass *klass;
183
184   klass = GST_TASK_GET_CLASS (task);
185
186   task->priv = GST_TASK_GET_PRIVATE (task);
187   task->running = FALSE;
188   task->thread = NULL;
189   task->lock = NULL;
190   task->cond = g_cond_new ();
191   SET_TASK_STATE (task, GST_TASK_STOPPED);
192   task->priv->prio_set = FALSE;
193
194   /* use the default klass pool for this task, users can
195    * override this later */
196   g_static_mutex_lock (&pool_lock);
197   task->priv->pool = gst_object_ref (klass->pool);
198   g_static_mutex_unlock (&pool_lock);
199 }
200
201 static void
202 gst_task_finalize (GObject * object)
203 {
204   GstTask *task = GST_TASK (object);
205   GstTaskPrivate *priv = task->priv;
206
207   GST_DEBUG ("task %p finalize", task);
208
209   if (priv->thr_notify)
210     priv->thr_notify (priv->thr_user_data);
211   priv->thr_notify = NULL;
212   priv->thr_user_data = NULL;
213
214   gst_object_unref (priv->pool);
215
216   /* task thread cannot be running here since it holds a ref
217    * to the task so that the finalize could not have happened */
218   g_cond_free (task->cond);
219   task->cond = NULL;
220
221   G_OBJECT_CLASS (gst_task_parent_class)->finalize (object);
222 }
223
224 /* should be called with the object LOCK */
225 static void
226 gst_task_configure_name (GstTask * task)
227 {
228 #ifdef HAVE_SYS_PRCTL_H
229   const gchar *name;
230   gchar thread_name[17] = { 0, };
231
232   GST_OBJECT_LOCK (task);
233   name = GST_OBJECT_NAME (task);
234
235   /* set the thread name to something easily identifiable */
236   if (!snprintf (thread_name, 17, "%s", GST_STR_NULL (name))) {
237     GST_DEBUG_OBJECT (task, "Could not create thread name for '%s'", name);
238   } else {
239     GST_DEBUG_OBJECT (task, "Setting thread name to '%s'", thread_name);
240     if (prctl (PR_SET_NAME, (unsigned long int) thread_name, 0, 0, 0))
241       GST_DEBUG_OBJECT (task, "Failed to set thread name");
242   }
243   GST_OBJECT_UNLOCK (task);
244 #endif
245 #ifdef _MSC_VER
246   const gchar *name;
247   name = GST_OBJECT_NAME (task);
248
249   /* set the thread name to something easily identifiable */
250   GST_DEBUG_OBJECT (task, "Setting thread name to '%s'", name);
251   SetThreadName (-1, name);
252 #endif
253 }
254
255 static void
256 gst_task_func (GstTask * task)
257 {
258   GStaticRecMutex *lock;
259   GThread *tself;
260   GstTaskPrivate *priv;
261
262   priv = task->priv;
263
264   tself = g_thread_self ();
265
266   GST_DEBUG ("Entering task %p, thread %p", task, tself);
267
268   /* we have to grab the lock to get the mutex. We also
269    * mark our state running so that nobody can mess with
270    * the mutex. */
271   GST_OBJECT_LOCK (task);
272   if (GET_TASK_STATE (task) == GST_TASK_STOPPED)
273     goto exit;
274   lock = GST_TASK_GET_LOCK (task);
275   if (G_UNLIKELY (lock == NULL))
276     goto no_lock;
277   task->thread = tself;
278   /* only update the priority when it was changed */
279   if (priv->prio_set)
280     g_thread_set_priority (tself, priv->priority);
281   GST_OBJECT_UNLOCK (task);
282
283   /* fire the enter_thread callback when we need to */
284   if (priv->thr_callbacks.enter_thread)
285     priv->thr_callbacks.enter_thread (task, tself, priv->thr_user_data);
286
287   /* locking order is TASK_LOCK, LOCK */
288   g_static_rec_mutex_lock (lock);
289   /* configure the thread name now */
290   gst_task_configure_name (task);
291
292   while (G_LIKELY (GET_TASK_STATE (task) != GST_TASK_STOPPED)) {
293     if (G_UNLIKELY (GET_TASK_STATE (task) == GST_TASK_PAUSED)) {
294       GST_OBJECT_LOCK (task);
295       while (G_UNLIKELY (GST_TASK_STATE (task) == GST_TASK_PAUSED)) {
296         gint t;
297
298         t = g_static_rec_mutex_unlock_full (lock);
299         if (t <= 0) {
300           g_warning ("wrong STREAM_LOCK count %d", t);
301         }
302         GST_TASK_SIGNAL (task);
303         GST_TASK_WAIT (task);
304         GST_OBJECT_UNLOCK (task);
305         /* locking order.. */
306         if (t > 0)
307           g_static_rec_mutex_lock_full (lock, t);
308
309         GST_OBJECT_LOCK (task);
310         if (G_UNLIKELY (GET_TASK_STATE (task) == GST_TASK_STOPPED)) {
311           GST_OBJECT_UNLOCK (task);
312           goto done;
313         }
314       }
315       GST_OBJECT_UNLOCK (task);
316     }
317
318     task->func (task->data);
319   }
320 done:
321   g_static_rec_mutex_unlock (lock);
322
323   GST_OBJECT_LOCK (task);
324   task->thread = NULL;
325
326 exit:
327   if (priv->thr_callbacks.leave_thread) {
328     /* fire the leave_thread callback when we need to. We need to do this before
329      * we signal the task and with the task lock released. */
330     GST_OBJECT_UNLOCK (task);
331     priv->thr_callbacks.leave_thread (task, tself, priv->thr_user_data);
332     GST_OBJECT_LOCK (task);
333   } else {
334     /* restore normal priority when releasing back into the pool, we will not
335      * touch the priority when a custom callback has been installed. */
336     g_thread_set_priority (tself, G_THREAD_PRIORITY_NORMAL);
337   }
338   /* now we allow messing with the lock again by setting the running flag to
339    * FALSE. Together with the SIGNAL this is the sign for the _join() to
340    * complete.
341    * Note that we still have not dropped the final ref on the task. We could
342    * check here if there is a pending join() going on and drop the last ref
343    * before releasing the lock as we can be sure that a ref is held by the
344    * caller of the join(). */
345   task->running = FALSE;
346   GST_TASK_SIGNAL (task);
347   GST_OBJECT_UNLOCK (task);
348
349   GST_DEBUG ("Exit task %p, thread %p", task, g_thread_self ());
350
351   gst_object_unref (task);
352   return;
353
354 no_lock:
355   {
356     g_warning ("starting task without a lock");
357     goto exit;
358   }
359 }
360
361 /**
362  * gst_task_cleanup_all:
363  *
364  * Wait for all tasks to be stopped. This is mainly used internally
365  * to ensure proper cleanup of internal data structures in test suites.
366  *
367  * MT safe.
368  */
369 void
370 gst_task_cleanup_all (void)
371 {
372   GstTaskClass *klass;
373
374   if ((klass = g_type_class_peek (GST_TYPE_TASK))) {
375     init_klass_pool (klass);
376   }
377 }
378
379 /**
380  * gst_task_create:
381  * @func: The #GstTaskFunction to use
382  * @data: (closure): User data to pass to @func
383  *
384  * Create a new Task that will repeatedly call the provided @func
385  * with @data as a parameter. Typically the task will run in
386  * a new thread.
387  *
388  * The function cannot be changed after the task has been created. You
389  * must create a new #GstTask to change the function.
390  *
391  * This function will not yet create and start a thread. Use gst_task_start() or
392  * gst_task_pause() to create and start the GThread.
393  *
394  * Before the task can be used, a #GStaticRecMutex must be configured using the
395  * gst_task_set_lock() function. This lock will always be acquired while
396  * @func is called.
397  *
398  * Returns: (transfer full): A new #GstTask.
399  *
400  * MT safe.
401  */
402 GstTask *
403 gst_task_create (GstTaskFunction func, gpointer data)
404 {
405   GstTask *task;
406
407   task = g_object_newv (GST_TYPE_TASK, 0, NULL);
408   task->func = func;
409   task->data = data;
410
411   GST_DEBUG ("Created task %p", task);
412
413   return task;
414 }
415
416 /**
417  * gst_task_set_lock:
418  * @task: The #GstTask to use
419  * @mutex: The #GMutex to use
420  *
421  * Set the mutex used by the task. The mutex will be acquired before
422  * calling the #GstTaskFunction.
423  *
424  * This function has to be called before calling gst_task_pause() or
425  * gst_task_start().
426  *
427  * MT safe.
428  */
429 void
430 gst_task_set_lock (GstTask * task, GStaticRecMutex * mutex)
431 {
432   GST_OBJECT_LOCK (task);
433   if (G_UNLIKELY (task->running))
434     goto is_running;
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_set_priority:
450  * @task: a #GstTask
451  * @priority: a new priority for @task
452  *
453  * Changes the priority of @task to @priority.
454  *
455  * Note: try not to depend on task priorities.
456  *
457  * MT safe.
458  *
459  * Since: 0.10.24
460  */
461 void
462 gst_task_set_priority (GstTask * task, GThreadPriority priority)
463 {
464   GstTaskPrivate *priv;
465   GThread *thread;
466
467   g_return_if_fail (GST_IS_TASK (task));
468
469   priv = task->priv;
470
471   GST_OBJECT_LOCK (task);
472   priv->prio_set = TRUE;
473   priv->priority = priority;
474   thread = task->thread;
475   if (thread != NULL) {
476     /* if this task already has a thread, we can configure the priority right
477      * away, else we do that when we assign a thread to the task. */
478     g_thread_set_priority (thread, priority);
479   }
480   GST_OBJECT_UNLOCK (task);
481 }
482
483 /**
484  * gst_task_get_pool:
485  * @task: a #GstTask
486  *
487  * Get the #GstTaskPool that this task will use for its streaming
488  * threads.
489  *
490  * MT safe.
491  *
492  * Returns: (transfer full): the #GstTaskPool used by @task. gst_object_unref()
493  * after usage.
494  *
495  * Since: 0.10.24
496  */
497 GstTaskPool *
498 gst_task_get_pool (GstTask * task)
499 {
500   GstTaskPool *result;
501   GstTaskPrivate *priv;
502
503   g_return_val_if_fail (GST_IS_TASK (task), NULL);
504
505   priv = task->priv;
506
507   GST_OBJECT_LOCK (task);
508   result = gst_object_ref (priv->pool);
509   GST_OBJECT_UNLOCK (task);
510
511   return result;
512 }
513
514 /**
515  * gst_task_set_pool:
516  * @task: a #GstTask
517  * @pool: (transfer none): a #GstTaskPool
518  *
519  * Set @pool as the new GstTaskPool for @task. Any new streaming threads that
520  * will be created by @task will now use @pool.
521  *
522  * MT safe.
523  *
524  * Since: 0.10.24
525  */
526 void
527 gst_task_set_pool (GstTask * task, GstTaskPool * pool)
528 {
529   GstTaskPool *old;
530   GstTaskPrivate *priv;
531
532   g_return_if_fail (GST_IS_TASK (task));
533   g_return_if_fail (GST_IS_TASK_POOL (pool));
534
535   priv = task->priv;
536
537   GST_OBJECT_LOCK (task);
538   if (priv->pool != pool) {
539     old = priv->pool;
540     priv->pool = gst_object_ref (pool);
541   } else
542     old = NULL;
543   GST_OBJECT_UNLOCK (task);
544
545   if (old)
546     gst_object_unref (old);
547 }
548
549
550 /**
551  * gst_task_set_thread_callbacks:
552  * @task: The #GstTask to use
553  * @callbacks: (in): a #GstTaskThreadCallbacks pointer
554  * @user_data: (closure): user data passed to the callbacks
555  * @notify: called when @user_data is no longer referenced
556  *
557  * Set callbacks which will be executed when a new thread is needed, the thread
558  * function is entered and left and when the thread is joined.
559  *
560  * By default a thread for @task will be created from a default thread pool.
561  *
562  * Objects can use custom GThreads or can perform additional configuration of
563  * the threads (such as changing the thread priority) by installing callbacks.
564  *
565  * MT safe.
566  *
567  * Since: 0.10.24
568  */
569 void
570 gst_task_set_thread_callbacks (GstTask * task,
571     GstTaskThreadCallbacks * callbacks, gpointer user_data,
572     GDestroyNotify notify)
573 {
574   GDestroyNotify old_notify;
575
576   g_return_if_fail (task != NULL);
577   g_return_if_fail (GST_IS_TASK (task));
578   g_return_if_fail (callbacks != NULL);
579
580   GST_OBJECT_LOCK (task);
581   old_notify = task->priv->thr_notify;
582
583   if (old_notify) {
584     gpointer old_data;
585
586     old_data = task->priv->thr_user_data;
587
588     task->priv->thr_user_data = NULL;
589     task->priv->thr_notify = NULL;
590     GST_OBJECT_UNLOCK (task);
591
592     old_notify (old_data);
593
594     GST_OBJECT_LOCK (task);
595   }
596   task->priv->thr_callbacks = *callbacks;
597   task->priv->thr_user_data = user_data;
598   task->priv->thr_notify = notify;
599   GST_OBJECT_UNLOCK (task);
600 }
601
602 /**
603  * gst_task_get_state:
604  * @task: The #GstTask to query
605  *
606  * Get the current state of the task.
607  *
608  * Returns: The #GstTaskState of the task
609  *
610  * MT safe.
611  */
612 GstTaskState
613 gst_task_get_state (GstTask * task)
614 {
615   GstTaskState result;
616
617   g_return_val_if_fail (GST_IS_TASK (task), GST_TASK_STOPPED);
618
619   result = GET_TASK_STATE (task);
620
621   return result;
622 }
623
624 /* make sure the task is running and start a thread if it's not.
625  * This function must be called with the task LOCK. */
626 static gboolean
627 start_task (GstTask * task)
628 {
629   gboolean res = TRUE;
630   GError *error = NULL;
631   GstTaskPrivate *priv;
632
633   priv = task->priv;
634
635   /* new task, We ref before so that it remains alive while
636    * the thread is running. */
637   gst_object_ref (task);
638   /* mark task as running so that a join will wait until we schedule
639    * and exit the task function. */
640   task->running = TRUE;
641
642   /* push on the thread pool, we remember the original pool because the user
643    * could change it later on and then we join to the wrong pool. */
644   priv->pool_id = gst_object_ref (priv->pool);
645   priv->id =
646       gst_task_pool_push (priv->pool_id, (GstTaskPoolFunction) gst_task_func,
647       task, &error);
648
649   if (error != NULL) {
650     g_warning ("failed to create thread: %s", error->message);
651     g_error_free (error);
652     res = FALSE;
653   }
654   return res;
655 }
656
657
658 /**
659  * gst_task_set_state:
660  * @task: a #GstTask
661  * @state: the new task state
662  *
663  * Sets the state of @task to @state.
664  *
665  * The @task must have a lock associated with it using
666  * gst_task_set_lock() when going to GST_TASK_STARTED or GST_TASK_PAUSED or
667  * this function will return %FALSE.
668  *
669  * MT safe.
670  *
671  * Returns: %TRUE if the state could be changed.
672  *
673  * Since: 0.10.24
674  */
675 gboolean
676 gst_task_set_state (GstTask * task, GstTaskState state)
677 {
678   GstTaskState old;
679   gboolean res = TRUE;
680
681   g_return_val_if_fail (GST_IS_TASK (task), FALSE);
682
683   GST_DEBUG_OBJECT (task, "Changing task %p to state %d", task, state);
684
685   GST_OBJECT_LOCK (task);
686   if (state != GST_TASK_STOPPED)
687     if (G_UNLIKELY (GST_TASK_GET_LOCK (task) == NULL))
688       goto no_lock;
689
690   /* if the state changed, do our thing */
691   old = GET_TASK_STATE (task);
692   if (old != state) {
693     SET_TASK_STATE (task, state);
694     switch (old) {
695       case GST_TASK_STOPPED:
696         /* If the task already has a thread scheduled we don't have to do
697          * anything. */
698         if (G_UNLIKELY (!task->running))
699           res = start_task (task);
700         break;
701       case GST_TASK_PAUSED:
702         /* when we are paused, signal to go to the new state */
703         GST_TASK_SIGNAL (task);
704         break;
705       case GST_TASK_STARTED:
706         /* if we were started, we'll go to the new state after the next
707          * iteration. */
708         break;
709     }
710   }
711   GST_OBJECT_UNLOCK (task);
712
713   return res;
714
715   /* ERRORS */
716 no_lock:
717   {
718     GST_WARNING_OBJECT (task, "state %d set on task without a lock", state);
719     GST_OBJECT_UNLOCK (task);
720     g_warning ("task without a lock can't be set to state %d", state);
721     return FALSE;
722   }
723 }
724
725 /**
726  * gst_task_start:
727  * @task: The #GstTask to start
728  *
729  * Starts @task. The @task must have a lock associated with it using
730  * gst_task_set_lock() or this function will return %FALSE.
731  *
732  * Returns: %TRUE if the task could be started.
733  *
734  * MT safe.
735  */
736 gboolean
737 gst_task_start (GstTask * task)
738 {
739   return gst_task_set_state (task, GST_TASK_STARTED);
740 }
741
742 /**
743  * gst_task_stop:
744  * @task: The #GstTask to stop
745  *
746  * Stops @task. This method merely schedules the task to stop and
747  * will not wait for the task to have completely stopped. Use
748  * gst_task_join() to stop and wait for completion.
749  *
750  * Returns: %TRUE if the task could be stopped.
751  *
752  * MT safe.
753  */
754 gboolean
755 gst_task_stop (GstTask * task)
756 {
757   return gst_task_set_state (task, GST_TASK_STOPPED);
758 }
759
760 /**
761  * gst_task_pause:
762  * @task: The #GstTask to pause
763  *
764  * Pauses @task. This method can also be called on a task in the
765  * stopped state, in which case a thread will be started and will remain
766  * in the paused state. This function does not wait for the task to complete
767  * the paused state.
768  *
769  * Returns: %TRUE if the task could be paused.
770  *
771  * MT safe.
772  */
773 gboolean
774 gst_task_pause (GstTask * task)
775 {
776   return gst_task_set_state (task, GST_TASK_PAUSED);
777 }
778
779 /**
780  * gst_task_join:
781  * @task: The #GstTask to join
782  *
783  * Joins @task. After this call, it is safe to unref the task
784  * and clean up the lock set with gst_task_set_lock().
785  *
786  * The task will automatically be stopped with this call.
787  *
788  * This function cannot be called from within a task function as this
789  * would cause a deadlock. The function will detect this and print a
790  * g_warning.
791  *
792  * Returns: %TRUE if the task could be joined.
793  *
794  * MT safe.
795  */
796 gboolean
797 gst_task_join (GstTask * task)
798 {
799   GThread *tself;
800   GstTaskPrivate *priv;
801   gpointer id;
802   GstTaskPool *pool = NULL;
803
804   priv = task->priv;
805
806   g_return_val_if_fail (GST_IS_TASK (task), FALSE);
807
808   tself = g_thread_self ();
809
810   GST_DEBUG_OBJECT (task, "Joining task %p, thread %p", task, tself);
811
812   /* we don't use a real thread join here because we are using
813    * thread pools */
814   GST_OBJECT_LOCK (task);
815   if (G_UNLIKELY (tself == task->thread))
816     goto joining_self;
817   SET_TASK_STATE (task, GST_TASK_STOPPED);
818   /* signal the state change for when it was blocked in PAUSED. */
819   GST_TASK_SIGNAL (task);
820   /* we set the running flag when pushing the task on the thread pool.
821    * This means that the task function might not be called when we try
822    * to join it here. */
823   while (G_LIKELY (task->running))
824     GST_TASK_WAIT (task);
825   /* clean the thread */
826   task->thread = NULL;
827   /* get the id and pool to join */
828   pool = priv->pool_id;
829   id = priv->id;
830   priv->pool_id = NULL;
831   priv->id = NULL;
832   GST_OBJECT_UNLOCK (task);
833
834   if (pool) {
835     if (id)
836       gst_task_pool_join (pool, id);
837     gst_object_unref (pool);
838   }
839
840   GST_DEBUG_OBJECT (task, "Joined task %p", task);
841
842   return TRUE;
843
844   /* ERRORS */
845 joining_self:
846   {
847     GST_WARNING_OBJECT (task, "trying to join task from its thread");
848     GST_OBJECT_UNLOCK (task);
849     g_warning ("\nTrying to join task %p from its thread would deadlock.\n"
850         "You cannot change the state of an element from its streaming\n"
851         "thread. Use g_idle_add() or post a GstMessage on the bus to\n"
852         "schedule the state change from the main thread.\n", task);
853     return FALSE;
854   }
855 }