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., 59 Temple Place - Suite 330,

    19  * Boston, MA 02111-1307, USA.

    20  */

    21 

    22 /**

    23  * SECTION:gsttaskpool

    24  * @short_description: Pool of GStreamer streaming threads

    25  * @see_also: #GstTask, #GstPad

    26  *

    27  * This object provides an abstraction for creating threads. The default

    28  * implementation uses a regular GThreadPool to start tasks.

    29  *

    30  * Subclasses can be made to create custom threads.

    31  *

    32  * Last reviewed on 2009-04-23 (0.10.24)

    33  */

    34 

    35 #include "gst_private.h"

    36 

    37 #include "gstinfo.h"

    38 #include "gsttaskpool.h"

    39 #ifdef __SYMBIAN32__

    40 #include <glib_global.h>

    41 #endif

    42 

    43 GST_DEBUG_CATEGORY_STATIC (taskpool_debug);

    44 #define GST_CAT_DEFAULT (taskpool_debug)

    45 

    46 static void gst_task_pool_class_init (GstTaskPoolClass * klass);

    47 static void gst_task_pool_init (GstTaskPool * pool);

    48 static void gst_task_pool_finalize (GObject * object);

    49 

    50 #define _do_init \

    51 { \

    52   GST_DEBUG_CATEGORY_INIT (taskpool_debug, "taskpool", 0, "Thread pool"); \

    53 }

    54 

    55 G_DEFINE_TYPE_WITH_CODE (GstTaskPool, gst_task_pool, GST_TYPE_OBJECT, _do_init);

    56 

    57 typedef struct

    58 {

    59   GstTaskPoolFunction func;

    60   gpointer user_data;

    61 } TaskData;

    62 

    63 static void

    64 default_func (TaskData * tdata, GstTaskPool * pool)

    65 {

    66   GstTaskPoolFunction func;

    67   gpointer user_data;

    68 

    69   func = tdata->func;

    70   user_data = tdata->user_data;

    71   g_slice_free (TaskData, tdata);

    72 

    73   func (user_data);

    74 }

    75 

    76 static void

    77 default_prepare (GstTaskPool * pool, GError ** error)

    78 {

    79   GST_OBJECT_LOCK (pool);

    80   pool->pool = g_thread_pool_new ((GFunc) default_func, pool, -1, FALSE, NULL);

    81   GST_OBJECT_UNLOCK (pool);

    82 }

    83 

    84 static void

    85 default_cleanup (GstTaskPool * pool)

    86 {

    87   GST_OBJECT_LOCK (pool);

    88   if (pool->pool) {

    89     /* Shut down all the threads, we still process the ones scheduled

    90      * because the unref happens in the thread function.

    91      * Also wait for currently running ones to finish. */

    92     g_thread_pool_free (pool->pool, FALSE, TRUE);

    93     pool->pool = NULL;

    94   }

    95   GST_OBJECT_UNLOCK (pool);

    96 }

    97 

    98 static gpointer

    99 default_push (GstTaskPool * pool, GstTaskPoolFunction func,

   100     gpointer user_data, GError ** error)

   101 {

   102   TaskData *tdata;

   103 

   104   tdata = g_slice_new (TaskData);

   105   tdata->func = func;

   106   tdata->user_data = user_data;

   107 

   108   GST_OBJECT_LOCK (pool);

   109   if (pool->pool)

   110     g_thread_pool_push (pool->pool, tdata, error);

   111   else {

   112     g_slice_free (TaskData, tdata);

   113   }

   114   GST_OBJECT_UNLOCK (pool);

   115 

   116   return NULL;

   117 }

   118 

   119 static void

   120 default_join (GstTaskPool * pool, gpointer id)

   121 {

   122   /* we do nothing here, we can't join from the pools */

   123 }

   124 

   125 static void

   126 gst_task_pool_class_init (GstTaskPoolClass * klass)

   127 {

   128   GObjectClass *gobject_class;

   129   GstTaskPoolClass *gsttaskpool_class;

   130 

   131   gobject_class = (GObjectClass *) klass;

   132   gsttaskpool_class = (GstTaskPoolClass *) klass;

   133 

   134   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_task_pool_finalize);

   135 

   136   gsttaskpool_class->prepare = default_prepare;

   137   gsttaskpool_class->cleanup = default_cleanup;

   138   gsttaskpool_class->push = default_push;

   139   gsttaskpool_class->join = default_join;

   140 }

   141 

   142 static void

   143 gst_task_pool_init (GstTaskPool * pool)

   144 {

   145 }

   146 

   147 static void

   148 gst_task_pool_finalize (GObject * object)

   149 {

   150   GST_DEBUG ("taskpool %p finalize", object);

   151 

   152   G_OBJECT_CLASS (gst_task_pool_parent_class)->finalize (object);

   153 }

   154 

   155 /**

   156  * gst_task_pool_new:

   157  *

   158  * Create a new default task pool. The default task pool will use a regular

   159  * GThreadPool for threads.

   160  *

   161  * Returns: a new #GstTaskPool. gst_object_unref() after usage.

   162  */

   163 #ifdef __SYMBIAN32__

   164 EXPORT_C

   165 #endif

   166 

   167 GstTaskPool *

   168 gst_task_pool_new (void)

   169 {

   170   GstTaskPool *pool;

   171 

   172   pool = g_object_new (GST_TYPE_TASK_POOL, NULL);

   173 

   174   return pool;

   175 }

   176 

   177 /**

   178  * gst_task_pool_prepare:

   179  * @pool: a #GstTaskPool

   180  * @error: an error return location

   181  *

   182  * Prepare the taskpool for accepting gst_task_pool_push() operations.

   183  *

   184  * MT safe.

   185  */

   186 #ifdef __SYMBIAN32__

   187 EXPORT_C

   188 #endif

   189 

   190 void

   191 gst_task_pool_prepare (GstTaskPool * pool, GError ** error)

   192 {

   193   GstTaskPoolClass *klass;

   194 

   195   g_return_if_fail (GST_IS_TASK_POOL (pool));

   196 

   197   klass = GST_TASK_POOL_GET_CLASS (pool);

   198 

   199   if (klass->prepare)

   200     klass->prepare (pool, error);

   201 }

   202 

   203 /**

   204  * gst_task_pool_cleanup:

   205  * @pool: a #GstTaskPool

   206  *

   207  * Wait for all tasks to be stopped. This is mainly used internally

   208  * to ensure proper cleanup of internal data structures in test suites.

   209  *

   210  * MT safe.

   211  */

   212 #ifdef __SYMBIAN32__

   213 EXPORT_C

   214 #endif

   215 

   216 void

   217 gst_task_pool_cleanup (GstTaskPool * pool)

   218 {

   219   GstTaskPoolClass *klass;

   220 

   221   g_return_if_fail (GST_IS_TASK_POOL (pool));

   222 

   223   klass = GST_TASK_POOL_GET_CLASS (pool);

   224 

   225   if (klass->cleanup)

   226     klass->cleanup (pool);

   227 }

   228 

   229 /**

   230  * gst_task_pool_push:

   231  * @pool: a #GstTaskPool

   232  * @func: the function to call

   233  * @user_data: data to pass to @func

   234  * @error: return location for an error

   235  *

   236  * Start the execution of a new thread from @pool.

   237  *

   238  * Returns: a pointer that should be used for the gst_task_pool_join

   239  * function. This pointer can be NULL, you must check @error to detect

   240  * errors.

   241  */

   242 #ifdef __SYMBIAN32__

   243 EXPORT_C

   244 #endif

   245 

   246 gpointer

   247 gst_task_pool_push (GstTaskPool * pool, GstTaskPoolFunction func,

   248     gpointer user_data, GError ** error)

   249 {

   250   GstTaskPoolClass *klass;

   251 

   252   g_return_val_if_fail (GST_IS_TASK_POOL (pool), NULL);

   253 

   254   klass = GST_TASK_POOL_GET_CLASS (pool);

   255 

   256   if (klass->push == NULL)

   257     goto not_supported;

   258 

   259   return klass->push (pool, func, user_data, error);

   260 

   261   /* ERRORS */

   262 not_supported:

   263   {

   264     g_warning ("pushing tasks on pool %p is not supported", pool);

   265     return NULL;

   266   }

   267 }

   268 

   269 /**

   270  * gst_task_pool_join:

   271  * @pool: a #GstTaskPool

   272  * @id: the id

   273  *

   274  * Join a task and/or return it to the pool. @id is the id obtained from 

   275  * gst_task_pool_push().

   276  */

   277 #ifdef __SYMBIAN32__

   278 EXPORT_C

   279 #endif

   280 

   281 void

   282 gst_task_pool_join (GstTaskPool * pool, gpointer id)

   283 {

   284   GstTaskPoolClass *klass;

   285 

   286   g_return_if_fail (GST_IS_TASK_POOL (pool));

   287 

   288   klass = GST_TASK_POOL_GET_CLASS (pool);

   289 

   290   if (klass->join)

   291     klass->join (pool, id);

   292 }