d7ff3449c0e72f3588e9637f100fa99511a82a5b
[platform/upstream/gstreamer.git] / 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  * @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
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 {
142   /* clear floating flag */
143   gst_object_ref_sink (pool);
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.
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.
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.
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
220  * @func: (scope async): the function to call
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  *
226  * Returns: (transfer none): a pointer that should be used for the
227  * gst_task_pool_join function. This pointer can be %NULL, you must
228  * check @error to detect errors.
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().
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 }