blob: 20b8aa41b8a5ff4c4ea94b9d142d74ba5a25e1c9 [file] [log] [blame]
Nicolas Dechesnebeb43202011-09-30 00:54:14 +02001/* 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öge01f23672013-07-14 10:55:08 +020018 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
19 * Boston, MA 02110-1301, USA.
Nicolas Dechesnebeb43202011-09-30 00:54:14 +020020 */
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 Dechesnebeb43202011-09-30 00:54:14 +020031 */
32
33#include "gst_private.h"
34
35#include "gstinfo.h"
36#include "gsttaskpool.h"
37
38GST_DEBUG_CATEGORY_STATIC (taskpool_debug);
39#define GST_CAT_DEFAULT (taskpool_debug)
40
41#ifndef GST_DISABLE_GST_DEBUG
42static 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
50G_DEFINE_TYPE_WITH_CODE (GstTaskPool, gst_task_pool, GST_TYPE_OBJECT, _do_init);
51
52typedef struct
53{
54 GstTaskPoolFunction func;
55 gpointer user_data;
56} TaskData;
57
58static void
59default_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
71static void
72default_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
79static void
80default_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
93static gpointer
94default_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
114static void
115default_join (GstTaskPool * pool, gpointer id)
116{
117 /* we do nothing here, we can't join from the pools */
118}
119
120static void
121gst_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
139static void
140gst_task_pool_init (GstTaskPool * pool)
141{
Sebastian Dröge5254e832014-06-22 17:16:06 +0200142 /* clear floating flag */
143 gst_object_ref_sink (pool);
Nicolas Dechesnebeb43202011-09-30 00:54:14 +0200144}
145
146#ifndef GST_DISABLE_GST_DEBUG
147static void
148gst_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 Dechesnebeb43202011-09-30 00:54:14 +0200162 */
163GstTaskPool *
164gst_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 Dechesnebeb43202011-09-30 00:54:14 +0200181 */
182void
183gst_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 Dechesnebeb43202011-09-30 00:54:14 +0200203 */
204void
205gst_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öged1f921c2012-08-08 18:10:27 +0200220 * @func: (scope async): the function to call
Nicolas Dechesnebeb43202011-09-30 00:54:14 +0200221 * @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ögee2d05632014-06-28 11:12:35 +0200226 * 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 Dechesnebeb43202011-09-30 00:54:14 +0200229 */
230gpointer
231gst_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 */
246not_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 Dechesnebeb43202011-09-30 00:54:14 +0200260 */
261void
262gst_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}