Nicolas Dechesne | beb4320 | 2011-09-30 00:54:14 +0200 | [diff] [blame] | 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 |
Sebastian Dröge | 01f2367 | 2013-07-14 10:55:08 +0200 | [diff] [blame] | 18 | * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, |
| 19 | * Boston, MA 02110-1301, USA. |
Nicolas Dechesne | beb4320 | 2011-09-30 00:54:14 +0200 | [diff] [blame] | 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. |
Nicolas Dechesne | beb4320 | 2011-09-30 00:54:14 +0200 | [diff] [blame] | 31 | */ |
| 32 | |
| 33 | #include "gst_private.h" |
| 34 | |
| 35 | #include "gstinfo.h" |
| 36 | #include "gsttaskpool.h" |
| 37 | |
| 38 | GST_DEBUG_CATEGORY_STATIC (taskpool_debug); |
| 39 | #define GST_CAT_DEFAULT (taskpool_debug) |
| 40 | |
| 41 | #ifndef GST_DISABLE_GST_DEBUG |
| 42 | static void gst_task_pool_finalize (GObject * object); |
| 43 | #endif |
| 44 | |
| 45 | #define _do_init \ |
| 46 | { \ |
| 47 | GST_DEBUG_CATEGORY_INIT (taskpool_debug, "taskpool", 0, "Thread pool"); \ |
| 48 | } |
| 49 | |
| 50 | G_DEFINE_TYPE_WITH_CODE (GstTaskPool, gst_task_pool, GST_TYPE_OBJECT, _do_init); |
| 51 | |
| 52 | typedef struct |
| 53 | { |
| 54 | GstTaskPoolFunction func; |
| 55 | gpointer user_data; |
| 56 | } TaskData; |
| 57 | |
| 58 | static void |
| 59 | default_func (TaskData * tdata, GstTaskPool * pool) |
| 60 | { |
| 61 | GstTaskPoolFunction func; |
| 62 | gpointer user_data; |
| 63 | |
| 64 | func = tdata->func; |
| 65 | user_data = tdata->user_data; |
| 66 | g_slice_free (TaskData, tdata); |
| 67 | |
| 68 | func (user_data); |
| 69 | } |
| 70 | |
| 71 | static void |
| 72 | default_prepare (GstTaskPool * pool, GError ** error) |
| 73 | { |
| 74 | GST_OBJECT_LOCK (pool); |
| 75 | pool->pool = g_thread_pool_new ((GFunc) default_func, pool, -1, FALSE, NULL); |
| 76 | GST_OBJECT_UNLOCK (pool); |
| 77 | } |
| 78 | |
| 79 | static void |
| 80 | default_cleanup (GstTaskPool * pool) |
| 81 | { |
| 82 | GST_OBJECT_LOCK (pool); |
| 83 | if (pool->pool) { |
| 84 | /* Shut down all the threads, we still process the ones scheduled |
| 85 | * because the unref happens in the thread function. |
| 86 | * Also wait for currently running ones to finish. */ |
| 87 | g_thread_pool_free (pool->pool, FALSE, TRUE); |
| 88 | pool->pool = NULL; |
| 89 | } |
| 90 | GST_OBJECT_UNLOCK (pool); |
| 91 | } |
| 92 | |
| 93 | static gpointer |
| 94 | default_push (GstTaskPool * pool, GstTaskPoolFunction func, |
| 95 | gpointer user_data, GError ** error) |
| 96 | { |
| 97 | TaskData *tdata; |
| 98 | |
| 99 | tdata = g_slice_new (TaskData); |
| 100 | tdata->func = func; |
| 101 | tdata->user_data = user_data; |
| 102 | |
| 103 | GST_OBJECT_LOCK (pool); |
| 104 | if (pool->pool) |
| 105 | g_thread_pool_push (pool->pool, tdata, error); |
| 106 | else { |
| 107 | g_slice_free (TaskData, tdata); |
| 108 | } |
| 109 | GST_OBJECT_UNLOCK (pool); |
| 110 | |
| 111 | return NULL; |
| 112 | } |
| 113 | |
| 114 | static void |
| 115 | default_join (GstTaskPool * pool, gpointer id) |
| 116 | { |
| 117 | /* we do nothing here, we can't join from the pools */ |
| 118 | } |
| 119 | |
| 120 | static void |
| 121 | gst_task_pool_class_init (GstTaskPoolClass * klass) |
| 122 | { |
| 123 | GObjectClass *gobject_class; |
| 124 | GstTaskPoolClass *gsttaskpool_class; |
| 125 | |
| 126 | gobject_class = (GObjectClass *) klass; |
| 127 | gsttaskpool_class = (GstTaskPoolClass *) klass; |
| 128 | |
| 129 | #ifndef GST_DISABLE_GST_DEBUG |
| 130 | gobject_class->finalize = gst_task_pool_finalize; |
| 131 | #endif |
| 132 | |
| 133 | gsttaskpool_class->prepare = default_prepare; |
| 134 | gsttaskpool_class->cleanup = default_cleanup; |
| 135 | gsttaskpool_class->push = default_push; |
| 136 | gsttaskpool_class->join = default_join; |
| 137 | } |
| 138 | |
| 139 | static void |
| 140 | gst_task_pool_init (GstTaskPool * pool) |
| 141 | { |
Sebastian Dröge | 5254e83 | 2014-06-22 17:16:06 +0200 | [diff] [blame] | 142 | /* clear floating flag */ |
| 143 | gst_object_ref_sink (pool); |
Nicolas Dechesne | beb4320 | 2011-09-30 00:54:14 +0200 | [diff] [blame] | 144 | } |
| 145 | |
| 146 | #ifndef GST_DISABLE_GST_DEBUG |
| 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 | #endif |
| 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: (transfer full): a new #GstTaskPool. gst_object_unref() after usage. |
Nicolas Dechesne | beb4320 | 2011-09-30 00:54:14 +0200 | [diff] [blame] | 162 | */ |
| 163 | GstTaskPool * |
| 164 | gst_task_pool_new (void) |
| 165 | { |
| 166 | GstTaskPool *pool; |
| 167 | |
| 168 | pool = g_object_newv (GST_TYPE_TASK_POOL, 0, NULL); |
| 169 | |
| 170 | return pool; |
| 171 | } |
| 172 | |
| 173 | /** |
| 174 | * gst_task_pool_prepare: |
| 175 | * @pool: a #GstTaskPool |
| 176 | * @error: an error return location |
| 177 | * |
| 178 | * Prepare the taskpool for accepting gst_task_pool_push() operations. |
| 179 | * |
| 180 | * MT safe. |
Nicolas Dechesne | beb4320 | 2011-09-30 00:54:14 +0200 | [diff] [blame] | 181 | */ |
| 182 | void |
| 183 | gst_task_pool_prepare (GstTaskPool * pool, GError ** error) |
| 184 | { |
| 185 | GstTaskPoolClass *klass; |
| 186 | |
| 187 | g_return_if_fail (GST_IS_TASK_POOL (pool)); |
| 188 | |
| 189 | klass = GST_TASK_POOL_GET_CLASS (pool); |
| 190 | |
| 191 | if (klass->prepare) |
| 192 | klass->prepare (pool, error); |
| 193 | } |
| 194 | |
| 195 | /** |
| 196 | * gst_task_pool_cleanup: |
| 197 | * @pool: a #GstTaskPool |
| 198 | * |
| 199 | * Wait for all tasks to be stopped. This is mainly used internally |
| 200 | * to ensure proper cleanup of internal data structures in test suites. |
| 201 | * |
| 202 | * MT safe. |
Nicolas Dechesne | beb4320 | 2011-09-30 00:54:14 +0200 | [diff] [blame] | 203 | */ |
| 204 | void |
| 205 | gst_task_pool_cleanup (GstTaskPool * pool) |
| 206 | { |
| 207 | GstTaskPoolClass *klass; |
| 208 | |
| 209 | g_return_if_fail (GST_IS_TASK_POOL (pool)); |
| 210 | |
| 211 | klass = GST_TASK_POOL_GET_CLASS (pool); |
| 212 | |
| 213 | if (klass->cleanup) |
| 214 | klass->cleanup (pool); |
| 215 | } |
| 216 | |
| 217 | /** |
| 218 | * gst_task_pool_push: |
| 219 | * @pool: a #GstTaskPool |
Sebastian Dröge | d1f921c | 2012-08-08 18:10:27 +0200 | [diff] [blame] | 220 | * @func: (scope async): the function to call |
Nicolas Dechesne | beb4320 | 2011-09-30 00:54:14 +0200 | [diff] [blame] | 221 | * @user_data: (closure): data to pass to @func |
| 222 | * @error: return location for an error |
| 223 | * |
| 224 | * Start the execution of a new thread from @pool. |
| 225 | * |
Sebastian Dröge | e2d0563 | 2014-06-28 11:12:35 +0200 | [diff] [blame] | 226 | * Returns: (transfer none) (nullable): a pointer that should be used |
| 227 | * for the gst_task_pool_join function. This pointer can be %NULL, you |
| 228 | * must check @error to detect errors. |
Nicolas Dechesne | beb4320 | 2011-09-30 00:54:14 +0200 | [diff] [blame] | 229 | */ |
| 230 | gpointer |
| 231 | gst_task_pool_push (GstTaskPool * pool, GstTaskPoolFunction func, |
| 232 | gpointer user_data, GError ** error) |
| 233 | { |
| 234 | GstTaskPoolClass *klass; |
| 235 | |
| 236 | g_return_val_if_fail (GST_IS_TASK_POOL (pool), NULL); |
| 237 | |
| 238 | klass = GST_TASK_POOL_GET_CLASS (pool); |
| 239 | |
| 240 | if (klass->push == NULL) |
| 241 | goto not_supported; |
| 242 | |
| 243 | return klass->push (pool, func, user_data, error); |
| 244 | |
| 245 | /* ERRORS */ |
| 246 | not_supported: |
| 247 | { |
| 248 | g_warning ("pushing tasks on pool %p is not supported", pool); |
| 249 | return NULL; |
| 250 | } |
| 251 | } |
| 252 | |
| 253 | /** |
| 254 | * gst_task_pool_join: |
| 255 | * @pool: a #GstTaskPool |
| 256 | * @id: the id |
| 257 | * |
| 258 | * Join a task and/or return it to the pool. @id is the id obtained from |
| 259 | * gst_task_pool_push(). |
Nicolas Dechesne | beb4320 | 2011-09-30 00:54:14 +0200 | [diff] [blame] | 260 | */ |
| 261 | void |
| 262 | gst_task_pool_join (GstTaskPool * pool, gpointer id) |
| 263 | { |
| 264 | GstTaskPoolClass *klass; |
| 265 | |
| 266 | g_return_if_fail (GST_IS_TASK_POOL (pool)); |
| 267 | |
| 268 | klass = GST_TASK_POOL_GET_CLASS (pool); |
| 269 | |
| 270 | if (klass->join) |
| 271 | klass->join (pool, id); |
| 272 | } |