<TITLE>GstTaskPool</TITLE>
GstTaskPool
GstTaskPoolClass
+GstTaskPoolFunction
gst_task_pool_new
-gst_task_pool_set_func
gst_task_pool_prepare
gst_task_pool_push
gst_task_pool_join
static void gst_task_init (GstTask * task);
static void gst_task_finalize (GObject * object);
-static void gst_task_func (GstTask * task, GstTaskClass * tclass);
+static void gst_task_func (GstTask * task);
static GStaticMutex pool_lock = G_STATIC_MUTEX_INIT;
gst_object_unref (klass->pool);
}
klass->pool = gst_task_pool_new ();
- gst_task_pool_set_func (klass->pool, (GFunc) gst_task_func, klass);
gst_task_pool_prepare (klass->pool, NULL);
g_static_mutex_unlock (&pool_lock);
}
}
static void
-gst_task_func (GstTask * task, GstTaskClass * tclass)
+gst_task_func (GstTask * task)
{
GStaticRecMutex *lock;
GThread *tself;
/* push on the thread pool, we remember the original pool because the user
* could change it later on and then we join to the wrong pool. */
priv->pool_id = gst_object_ref (priv->pool);
- priv->id = gst_task_pool_push (priv->pool_id, task, &error);
+ priv->id =
+ gst_task_pool_push (priv->pool_id, (GstTaskPoolFunction) gst_task_func,
+ task, &error);
if (error != NULL) {
g_warning ("failed to create thread: %s", error->message);
G_DEFINE_TYPE_WITH_CODE (GstTaskPool, gst_task_pool, GST_TYPE_OBJECT, _do_init);
+typedef struct
+{
+ GstTaskPoolFunction func;
+ gpointer user_data;
+} TaskData;
+
static void
-default_prepare (GstTaskPool * pool, GFunc func, gpointer user_data,
- GError ** error)
+default_func (TaskData * tdata, GstTaskPool * pool)
+{
+ GstTaskPoolFunction func;
+ gpointer user_data;
+
+ func = tdata->func;
+ user_data = tdata->user_data;
+ g_slice_free (TaskData, tdata);
+
+ func (user_data);
+}
+
+static void
+default_prepare (GstTaskPool * pool, GError ** error)
{
GST_OBJECT_LOCK (pool);
- pool->pool = g_thread_pool_new ((GFunc) func, user_data, -1, FALSE, NULL);
+ pool->pool = g_thread_pool_new ((GFunc) default_func, pool, -1, FALSE, NULL);
GST_OBJECT_UNLOCK (pool);
}
}
static gpointer
-default_push (GstTaskPool * pool, gpointer data, GError ** error)
+default_push (GstTaskPool * pool, GstTaskPoolFunction func,
+ gpointer user_data, GError ** error)
{
+ TaskData *tdata;
+
+ tdata = g_slice_new (TaskData);
+ tdata->func = func;
+ tdata->user_data = user_data;
+
GST_OBJECT_LOCK (pool);
if (pool->pool)
- g_thread_pool_push (pool->pool, data, error);
+ g_thread_pool_push (pool->pool, tdata, error);
+ else {
+ g_slice_free (TaskData, tdata);
+ }
GST_OBJECT_UNLOCK (pool);
return NULL;
static void
default_join (GstTaskPool * pool, gpointer id)
{
- /* does nothing, we can't join for threads from the threadpool */
+ /* we do nothing here, we can't join from the pools */
}
static void
return pool;
}
-void
-gst_task_pool_set_func (GstTaskPool * pool, GFunc func, gpointer user_data)
-{
- g_return_if_fail (GST_IS_TASK_POOL (pool));
-
- pool->func = func;
- pool->user_data = user_data;
-}
-
-
/**
* gst_task_pool_prepare:
* @pool: a #GstTaskPool
klass = GST_TASK_POOL_GET_CLASS (pool);
if (klass->prepare)
- klass->prepare (pool, pool->func, pool->user_data, error);
+ klass->prepare (pool, error);
}
/**
/**
* gst_task_pool_push:
* @pool: a #GstTaskPool
- * @data: data to pass to the thread function
+ * @func: the function to call
+ * @user_data: data to pass to @func
* @error: return location for an error
*
* Start the execution of a new thread from @pool.
* errors.
*/
gpointer
-gst_task_pool_push (GstTaskPool * pool, gpointer data, GError ** error)
+gst_task_pool_push (GstTaskPool * pool, GstTaskPoolFunction func,
+ gpointer user_data, GError ** error)
{
GstTaskPoolClass *klass;
if (klass->push == NULL)
goto not_supported;
- return klass->push (pool, data, error);
+ return klass->push (pool, func, user_data, error);
/* ERRORS */
not_supported:
typedef struct _GstTaskPool GstTaskPool;
typedef struct _GstTaskPoolClass GstTaskPoolClass;
+typedef void (*GstTaskPoolFunction) (void *data);
+
/**
* GstTaskPool:
*
GstObject object;
/*< private >*/
- GFunc func;
- gpointer user_data;
-
GThreadPool *pool;
gpointer _gst_reserved[GST_PADDING];
GstObjectClass parent_class;
/*< public >*/
- void (*prepare) (GstTaskPool *pool, GFunc func,
- gpointer user_data, GError **error);
+ void (*prepare) (GstTaskPool *pool, GError **error);
void (*cleanup) (GstTaskPool *pool);
- gpointer (*push) (GstTaskPool *pool, gpointer data, GError **error);
+ gpointer (*push) (GstTaskPool *pool, GstTaskPoolFunction func,
+ gpointer user_data, GError **error);
void (*join) (GstTaskPool *pool, gpointer id);
/*< private >*/
GType gst_task_pool_get_type (void);
GstTaskPool * gst_task_pool_new (void);
-
-void gst_task_pool_set_func (GstTaskPool *pool,
- GFunc func, gpointer user_data);
-
void gst_task_pool_prepare (GstTaskPool *pool, GError **error);
-gpointer gst_task_pool_push (GstTaskPool *pool, gpointer data,
- GError **error);
+gpointer gst_task_pool_push (GstTaskPool *pool, GstTaskPoolFunction func,
+ gpointer user_data, GError **error);
void gst_task_pool_join (GstTaskPool *pool, gpointer id);
void gst_task_pool_cleanup (GstTaskPool *pool);
stream-status
+rtpool-test
*.bb
*.bbg
*.da
* Boston, MA 02111-1307, USA.
*/
+#include <pthread.h>
+
#include "testrtpool.h"
static void test_rt_pool_class_init (TestRTPoolClass * klass);
static void test_rt_pool_init (TestRTPool * pool);
static void test_rt_pool_finalize (GObject * object);
+typedef struct
+{
+ pthread_t thread;
+} TestRTId;
+
G_DEFINE_TYPE (TestRTPool, test_rt_pool, GST_TYPE_TASK_POOL);
static void
-default_prepare (GstTaskPool * pool, GFunc func, gpointer user_data,
- GError ** error)
+default_prepare (GstTaskPool * pool, GError ** error)
{
+ /* we don't do anything here. We could construct a pool of threads here that
+ * we could reuse later but we don't */
g_message ("prepare Realtime pool %p", pool);
}
}
static gpointer
-default_push (GstTaskPool * pool, gpointer data, GError ** error)
+default_push (GstTaskPool * pool, GstTaskPoolFunction func, gpointer data,
+ GError ** error)
{
- g_message ("pushing Realtime pool %p", pool);
+ TestRTId *tid;
+ gint res;
+ pthread_attr_t attr;
+ //struct sched_param param;
+
+ g_message ("pushing Realtime pool %p, %p", pool, func);
- *error = g_error_new (1, 1, "not supported");
+ tid = g_slice_new0 (TestRTId);
- return NULL;
+ pthread_attr_init (&attr);
+ /*
+ pthread_attr_setschedpolicy (&attr, SCHED_RR);
+ param.sched_priority = 50;
+ pthread_attr_setschedparam (&attr, ¶m);
+ */
+
+ res = pthread_create (&tid->thread, &attr, (void *(*)(void *)) func, data);
+
+ if (res < 0) {
+ g_set_error (error, G_THREAD_ERROR, G_THREAD_ERROR_AGAIN,
+ "Error creating thread: %s", g_strerror (res));
+ g_slice_free (TestRTId, tid);
+ tid = NULL;
+ }
+
+ return tid;
}
static void
default_join (GstTaskPool * pool, gpointer id)
{
+ TestRTId *tid = (TestRTId *) id;
+
g_message ("joining Realtime pool %p", pool);
+
+ pthread_join (tid->thread, NULL);
+
+ g_slice_free (TestRTId, tid);
}
static void
gst_task_pool_new
gst_task_pool_prepare
gst_task_pool_push
- gst_task_pool_set_func
gst_task_set_lock
gst_task_set_pool
gst_task_set_priority