gstpad: Avoid race in (un)setting EOS flag on sinkpads
[platform/upstream/gstreamer.git] / subprojects / gstreamer / gst / gsttaskpool.c
1 /* GStreamer
2  * Copyright (C) 2009 Wim Taymans <wim.taymans@gmail.com>
3  *
4  * gsttaskpool.c: Pool for streaming threads
5  *
6  * This library is free software; you can redistribute it and/or
7  * modify it under the terms of the GNU Library General Public
8  * License as published by the Free Software Foundation; either
9  * version 2 of the License, or (at your option) any later version.
10  *
11  * This library is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14  * Library General Public License for more details.
15  *
16  * You should have received a copy of the GNU Library General Public
17  * License along with this library; if not, write to the
18  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
19  * Boston, MA 02110-1301, USA.
20  */
21
22 /**
23  * SECTION:gsttaskpool
24  * @title: GstTaskPool
25  * @short_description: Pool of GStreamer streaming threads
26  * @see_also: #GstTask, #GstPad
27  *
28  * This object provides an abstraction for creating threads. The default
29  * implementation uses a regular GThreadPool to start tasks.
30  *
31  * Subclasses can be made to create custom threads.
32  */
33
34 #include "gst_private.h"
35
36 #include "gstinfo.h"
37 #include "gsttaskpool.h"
38 #include "gsterror.h"
39
40 GST_DEBUG_CATEGORY_STATIC (taskpool_debug);
41 #define GST_CAT_DEFAULT (taskpool_debug)
42
43 #ifndef GST_DISABLE_GST_DEBUG
44 static void gst_task_pool_finalize (GObject * object);
45 #endif
46
47 #define _do_init \
48 { \
49   GST_DEBUG_CATEGORY_INIT (taskpool_debug, "taskpool", 0, "Thread pool"); \
50 }
51
52 G_DEFINE_TYPE_WITH_CODE (GstTaskPool, gst_task_pool, GST_TYPE_OBJECT, _do_init);
53
54 typedef struct
55 {
56   GstTaskPoolFunction func;
57   gpointer user_data;
58 } TaskData;
59
60 static void
61 default_func (TaskData * tdata, GstTaskPool * pool)
62 {
63   GstTaskPoolFunction func;
64   gpointer user_data;
65
66   func = tdata->func;
67   user_data = tdata->user_data;
68   g_slice_free (TaskData, tdata);
69
70   func (user_data);
71 }
72
73 static void
74 default_prepare (GstTaskPool * pool, GError ** error)
75 {
76   GST_OBJECT_LOCK (pool);
77   pool->pool = g_thread_pool_new ((GFunc) default_func, pool, -1, FALSE, error);
78   GST_OBJECT_UNLOCK (pool);
79 }
80
81 static void
82 default_cleanup (GstTaskPool * pool)
83 {
84   GThreadPool *pool_;
85
86   GST_OBJECT_LOCK (pool);
87   pool_ = pool->pool;
88   pool->pool = NULL;
89   GST_OBJECT_UNLOCK (pool);
90
91   if (pool_) {
92     /* Shut down all the threads, we still process the ones scheduled
93      * because the unref happens in the thread function.
94      * Also wait for currently running ones to finish. */
95     g_thread_pool_free (pool_, FALSE, TRUE);
96   }
97 }
98
99 static gpointer
100 default_push (GstTaskPool * pool, GstTaskPoolFunction func,
101     gpointer user_data, GError ** error)
102 {
103   TaskData *tdata;
104
105   tdata = g_slice_new (TaskData);
106   tdata->func = func;
107   tdata->user_data = user_data;
108
109   GST_OBJECT_LOCK (pool);
110   if (pool->pool)
111     g_thread_pool_push (pool->pool, tdata, error);
112   else {
113     g_slice_free (TaskData, tdata);
114     g_set_error_literal (error, GST_CORE_ERROR, GST_CORE_ERROR_FAILED,
115         "No thread pool");
116
117   }
118   GST_OBJECT_UNLOCK (pool);
119
120   return NULL;
121 }
122
123 static void
124 default_join (GstTaskPool * pool, gpointer id)
125 {
126   /* we do nothing here, we can't join from the pools */
127 }
128
129 static void
130 default_dispose_handle (GstTaskPool * pool, gpointer id)
131 {
132   /* we do nothing here, the default handle is NULL */
133 }
134
135 static void
136 gst_task_pool_class_init (GstTaskPoolClass * klass)
137 {
138   GObjectClass *gobject_class;
139   GstTaskPoolClass *gsttaskpool_class;
140
141   gobject_class = (GObjectClass *) klass;
142   gsttaskpool_class = (GstTaskPoolClass *) klass;
143
144 #ifndef GST_DISABLE_GST_DEBUG
145   gobject_class->finalize = gst_task_pool_finalize;
146 #endif
147
148   gsttaskpool_class->prepare = default_prepare;
149   gsttaskpool_class->cleanup = default_cleanup;
150   gsttaskpool_class->push = default_push;
151   gsttaskpool_class->join = default_join;
152   gsttaskpool_class->dispose_handle = default_dispose_handle;
153 }
154
155 static void
156 gst_task_pool_init (GstTaskPool * pool)
157 {
158 }
159
160 #ifndef GST_DISABLE_GST_DEBUG
161 static void
162 gst_task_pool_finalize (GObject * object)
163 {
164   GST_DEBUG ("taskpool %p finalize", object);
165
166   G_OBJECT_CLASS (gst_task_pool_parent_class)->finalize (object);
167 }
168 #endif
169 /**
170  * gst_task_pool_new:
171  *
172  * Create a new default task pool. The default task pool will use a regular
173  * GThreadPool for threads.
174  *
175  * Returns: (transfer full): a new #GstTaskPool. gst_object_unref() after usage.
176  */
177 GstTaskPool *
178 gst_task_pool_new (void)
179 {
180   GstTaskPool *pool;
181
182   pool = g_object_new (GST_TYPE_TASK_POOL, NULL);
183
184   /* clear floating flag */
185   gst_object_ref_sink (pool);
186
187   return pool;
188 }
189
190 /**
191  * gst_task_pool_prepare:
192  * @pool: a #GstTaskPool
193  * @error: an error return location
194  *
195  * Prepare the taskpool for accepting gst_task_pool_push() operations.
196  *
197  * MT safe.
198  */
199 void
200 gst_task_pool_prepare (GstTaskPool * pool, GError ** error)
201 {
202   GstTaskPoolClass *klass;
203
204   g_return_if_fail (GST_IS_TASK_POOL (pool));
205
206   klass = GST_TASK_POOL_GET_CLASS (pool);
207
208   if (klass->prepare)
209     klass->prepare (pool, error);
210 }
211
212 /**
213  * gst_task_pool_cleanup:
214  * @pool: a #GstTaskPool
215  *
216  * Wait for all tasks to be stopped. This is mainly used internally
217  * to ensure proper cleanup of internal data structures in test suites.
218  *
219  * MT safe.
220  */
221 void
222 gst_task_pool_cleanup (GstTaskPool * pool)
223 {
224   GstTaskPoolClass *klass;
225
226   g_return_if_fail (GST_IS_TASK_POOL (pool));
227
228   klass = GST_TASK_POOL_GET_CLASS (pool);
229
230   if (klass->cleanup)
231     klass->cleanup (pool);
232 }
233
234 /**
235  * gst_task_pool_push:
236  * @pool: a #GstTaskPool
237  * @func: (scope async): the function to call
238  * @user_data: (closure): data to pass to @func
239  * @error: return location for an error
240  *
241  * Start the execution of a new thread from @pool.
242  *
243  * Returns: (transfer full) (nullable): a pointer that should be used
244  * for the gst_task_pool_join function. This pointer can be %NULL, you
245  * must check @error to detect errors. If the pointer is not %NULL and
246  * gst_task_pool_join() is not used, call gst_task_pool_dispose_handle()
247  * instead.
248  */
249 gpointer
250 gst_task_pool_push (GstTaskPool * pool, GstTaskPoolFunction func,
251     gpointer user_data, GError ** error)
252 {
253   GstTaskPoolClass *klass;
254
255   g_return_val_if_fail (GST_IS_TASK_POOL (pool), NULL);
256
257   klass = GST_TASK_POOL_GET_CLASS (pool);
258
259   if (klass->push == NULL)
260     goto not_supported;
261
262   return klass->push (pool, func, user_data, error);
263
264   /* ERRORS */
265 not_supported:
266   {
267     g_warning ("pushing tasks on pool %p is not supported", pool);
268     return NULL;
269   }
270 }
271
272 /**
273  * gst_task_pool_join:
274  * @pool: a #GstTaskPool
275  * @id: (transfer full) (nullable): the id
276  *
277  * Join a task and/or return it to the pool. @id is the id obtained from
278  * gst_task_pool_push(). The default implementation does nothing, as the
279  * default #GstTaskPoolClass::push implementation always returns %NULL.
280  *
281  * This method should only be called with the same @pool instance that provided
282  * @id.
283  */
284 void
285 gst_task_pool_join (GstTaskPool * pool, gpointer id)
286 {
287   GstTaskPoolClass *klass;
288
289   g_return_if_fail (GST_IS_TASK_POOL (pool));
290
291   klass = GST_TASK_POOL_GET_CLASS (pool);
292
293   if (klass->join)
294     klass->join (pool, id);
295 }
296
297 /**
298  * gst_task_pool_dispose_handle:
299  * @pool: a #GstTaskPool
300  * @id: (transfer full) (nullable): the id
301  *
302  * Dispose of the handle returned by gst_task_pool_push(). This does
303  * not need to be called with the default implementation as the default
304  * #GstTaskPoolClass::push implementation always returns %NULL. This does not need to be
305  * called either when calling gst_task_pool_join(), but should be called
306  * when joining is not necessary, but gst_task_pool_push() returned a
307  * non-%NULL value.
308  *
309  * This method should only be called with the same @pool instance that provided
310  * @id.
311  *
312  * Since: 1.20
313  */
314 void
315 gst_task_pool_dispose_handle (GstTaskPool * pool, gpointer id)
316 {
317   GstTaskPoolClass *klass;
318
319   g_return_if_fail (GST_IS_TASK_POOL (pool));
320
321   klass = GST_TASK_POOL_GET_CLASS (pool);
322
323   if (klass->dispose_handle)
324     klass->dispose_handle (pool, id);
325 }
326
327 typedef struct
328 {
329   gboolean done;
330   guint64 id;
331   GstTaskPoolFunction func;
332   gpointer user_data;
333   GMutex done_lock;
334   GCond done_cond;
335   gint refcount;
336 } SharedTaskData;
337
338 static SharedTaskData *
339 shared_task_data_ref (SharedTaskData * tdata)
340 {
341   g_atomic_int_add (&tdata->refcount, 1);
342
343   return tdata;
344 }
345
346 static void
347 shared_task_data_unref (SharedTaskData * tdata)
348 {
349   if (g_atomic_int_dec_and_test (&tdata->refcount)) {
350     g_mutex_clear (&tdata->done_lock);
351     g_cond_clear (&tdata->done_cond);
352     g_slice_free (SharedTaskData, tdata);
353   }
354 }
355
356 struct _GstSharedTaskPoolPrivate
357 {
358   guint max_threads;
359 };
360
361 #define GST_SHARED_TASK_POOL_CAST(pool)       ((GstSharedTaskPool*)(pool))
362
363 G_DEFINE_TYPE_WITH_PRIVATE (GstSharedTaskPool, gst_shared_task_pool,
364     GST_TYPE_TASK_POOL);
365
366 static void
367 shared_func (SharedTaskData * tdata, GstTaskPool * pool)
368 {
369   tdata->func (tdata->user_data);
370
371   g_mutex_lock (&tdata->done_lock);
372   tdata->done = TRUE;
373   g_cond_signal (&tdata->done_cond);
374   g_mutex_unlock (&tdata->done_lock);
375
376   shared_task_data_unref (tdata);
377 }
378
379 static gpointer
380 shared_push (GstTaskPool * pool, GstTaskPoolFunction func,
381     gpointer user_data, GError ** error)
382 {
383   SharedTaskData *ret = NULL;
384
385   GST_OBJECT_LOCK (pool);
386
387   if (!pool->pool) {
388     GST_OBJECT_UNLOCK (pool);
389     goto done;
390   }
391
392   ret = g_slice_new (SharedTaskData);
393
394   ret->done = FALSE;
395   ret->func = func;
396   ret->user_data = user_data;
397   g_atomic_int_set (&ret->refcount, 1);
398   g_cond_init (&ret->done_cond);
399   g_mutex_init (&ret->done_lock);
400
401   g_thread_pool_push (pool->pool, shared_task_data_ref (ret), error);
402
403   GST_OBJECT_UNLOCK (pool);
404
405 done:
406   return ret;
407 }
408
409 static void
410 shared_join (GstTaskPool * pool, gpointer id)
411 {
412   SharedTaskData *tdata;
413
414   if (!id)
415     return;
416
417   tdata = (SharedTaskData *) id;
418
419   g_mutex_lock (&tdata->done_lock);
420   while (!tdata->done) {
421     g_cond_wait (&tdata->done_cond, &tdata->done_lock);
422   }
423   g_mutex_unlock (&tdata->done_lock);
424
425   shared_task_data_unref (tdata);
426 }
427
428 static void
429 shared_dispose_handle (GstTaskPool * pool, gpointer id)
430 {
431   SharedTaskData *tdata;
432
433   if (!id)
434     return;
435
436   tdata = (SharedTaskData *) id;
437
438
439   shared_task_data_unref (tdata);
440 }
441
442 static void
443 shared_prepare (GstTaskPool * pool, GError ** error)
444 {
445   GstSharedTaskPool *shared_pool = GST_SHARED_TASK_POOL_CAST (pool);
446
447   GST_OBJECT_LOCK (pool);
448   pool->pool =
449       g_thread_pool_new ((GFunc) shared_func, pool,
450       shared_pool->priv->max_threads, FALSE, error);
451   GST_OBJECT_UNLOCK (pool);
452 }
453
454 static void
455 gst_shared_task_pool_class_init (GstSharedTaskPoolClass * klass)
456 {
457   GstTaskPoolClass *taskpoolclass = GST_TASK_POOL_CLASS (klass);
458
459   taskpoolclass->prepare = shared_prepare;
460   taskpoolclass->push = shared_push;
461   taskpoolclass->join = shared_join;
462   taskpoolclass->dispose_handle = shared_dispose_handle;
463 }
464
465 static void
466 gst_shared_task_pool_init (GstSharedTaskPool * pool)
467 {
468   GstSharedTaskPoolPrivate *priv;
469
470   priv = pool->priv = gst_shared_task_pool_get_instance_private (pool);
471   priv->max_threads = 1;
472 }
473
474 /**
475  * gst_shared_task_pool_set_max_threads:
476  * @pool: a #GstSharedTaskPool
477  * @max_threads: Maximum number of threads to spawn.
478  *
479  * Update the maximal number of threads the @pool may spawn. When
480  * the maximal number of threads is reduced, existing threads are not
481  * immediately shut down, see g_thread_pool_set_max_threads().
482  *
483  * Setting @max_threads to 0 effectively freezes the pool.
484  *
485  * Since: 1.20
486  */
487 void
488 gst_shared_task_pool_set_max_threads (GstSharedTaskPool * pool,
489     guint max_threads)
490 {
491   GstTaskPool *taskpool;
492
493   g_return_if_fail (GST_IS_SHARED_TASK_POOL (pool));
494
495   taskpool = GST_TASK_POOL (pool);
496
497   GST_OBJECT_LOCK (pool);
498   if (taskpool->pool)
499     g_thread_pool_set_max_threads (taskpool->pool, max_threads, NULL);
500   pool->priv->max_threads = max_threads;
501   GST_OBJECT_UNLOCK (pool);
502 }
503
504 /**
505  * gst_shared_task_pool_get_max_threads:
506  * @pool: a #GstSharedTaskPool
507  *
508  * Returns: the maximum number of threads @pool is configured to spawn
509  * Since: 1.20
510  */
511 guint
512 gst_shared_task_pool_get_max_threads (GstSharedTaskPool * pool)
513 {
514   guint ret;
515
516   g_return_val_if_fail (GST_IS_SHARED_TASK_POOL (pool), 0);
517
518   GST_OBJECT_LOCK (pool);
519   ret = pool->priv->max_threads;
520   GST_OBJECT_UNLOCK (pool);
521
522   return ret;
523 }
524
525 /**
526  * gst_shared_task_pool_new:
527  *
528  * Create a new shared task pool. The shared task pool will queue tasks on
529  * a maximum number of threads, 1 by default.
530  *
531  * Do not use a #GstSharedTaskPool to manage potentially inter-dependent tasks such
532  * as pad tasks, as having one task waiting on another to return before returning
533  * would cause obvious deadlocks if they happen to share the same thread.
534  *
535  * Returns: (transfer full): a new #GstSharedTaskPool. gst_object_unref() after usage.
536  * Since: 1.20
537  */
538 GstTaskPool *
539 gst_shared_task_pool_new (void)
540 {
541   GstTaskPool *pool;
542
543   pool = g_object_new (GST_TYPE_SHARED_TASK_POOL, NULL);
544
545   /* clear floating flag */
546   gst_object_ref_sink (pool);
547
548   return pool;
549 }