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 }