Gstreamer中的task

在gstreamer中，可以在需要的时候方便的创建一个task作为独立线程，task可以是普通的task，也可以是pad task，这个需要根据具体需求来定，下面先看下v4l2和avdemux里面task的代码。

task的创建

gst-plugins-good/sys/v4l2/gstv4l2videodec.c

通过gst_pad_start_task开始一个pad task，task函数是gst_v4l2_video_dec_loop。作用是启动v4l2的解码task，并且当task退出时，将禁用输入处理以在drain时unlock输入，防止阻塞：

if (!gst_pad_start_task (decoder->srcpad, (GstTaskFunction) gst_v4l2_video_dec_loop, self, NULL))
    goto start_task_failed;

gst-libav/ext/libav/gstavdemux.c

创建一个demux->task，task函数是gst_ffmpegdemux_loop，gst_task_init会通过gst_task_new间接调用，在init函数中调用ensure_klass_pool确保_global_task_pool的线程池可用，start_task的时候将task通过gst_task_pool_push放到task的线程池里面，下面gst_pad_start_task的调用栈就可以看到。



demux->task = gst_task_new ((GstTaskFunction) gst_ffmpegdemux_loop, demux, NULL);
g_rec_mutex_init (&demux->task_lock);


gst_task_set_lock (demux->task, &demux->task_lock);

gst_ffmpegdemux_loop和gst_v4l2_video_dec_loop两个函数是task的loop函数，在gstreamer中，task不像pthread的中的task，需要有一个循环来保证常驻task一直存在。而是通过前面的线程池来运行。

task的运行

看gst_pad_start_task的代码和栈就可以清楚的看到task的new，init，start都是在gst_pad_start_task里面实现的，这些在如果不是通过gst_pad_start_task启动的task，都需要手动调用。

1   gst_task_init                      gsttask.c     249  0x7ffff7b49fa3 
2   g_type_create_instance                                0x7ffff75559c5 
4   g_object_new_with_properties                          0x7ffff7537ee5 
5   g_object_new                                          0x7ffff7538961 
6   gst_task_new                       gsttask.c     482  0x7ffff7b4a5ad 
7   gst_pad_start_task                 gstpad.c      6287 0x7ffff7b1d7fe

调用start_task的栈：

1   gst_task_pool_push                 gsttaskpool.c 252  0x7ffff7b4b820 
2   start_task                         gsttask.c     712  0x7ffff7b49f19 
3   gst_task_set_state_unlocked        gsttask.c     744  0x7ffff7b4ad3e 
4   gst_task_set_state                 gsttask.c     792  0x7ffff7b4ad3e                                                          
7   gst_pad_start_task                 gstpad.c      6287 0x7ffff7b1d7fe

gst_task_pool_push是在start_task函数中调用的，将task放到线程池里面，指定taskpool函数为gst_task_func:

  
  priv->pool_id = gst_object_ref (priv->pool);
  priv->id =
      gst_task_pool_push (priv->pool_id, (GstTaskPoolFunction) gst_task_func,
      task, &error);

gst_task_pool_push中继续调用GstTaskPoolClass的push函数default_push，在default_push函数中会调用g_thread_pool_push函数放到g_thread_pool里面，后面task的运行就是GLib的g_thread_pool相关的实现了。

gpointer
gst_task_pool_push (GstTaskPool * pool, GstTaskPoolFunction func,
                        gpointer user_data, GError ** error)
{
    GstTaskPoolClass *klass;
    g_return_val_if_fail (GST_IS_TASK_POOL (pool), NULL);

    klass = GST_TASK_POOL_GET_CLASS (pool);
    if (klass->push == NULL)
        goto not_supported;

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

注意：task不允许没有lock而使用，否则会报错：

task without a lock can't be set to state 0

通过gst_task_set_lock函数原型可以看到，这里task需要的lock是GRecMutex类型。

下面代码是curlhttpsrc的代码，可以有个初步的认识：

subprojects/gst-plugins-bad/ext/curl/gstcurlhttpsrc.c

g_rec_mutex_init (&klass->multi_task_context.task_rec_mutex);

klass->multi_task_context.state = GSTCURL_MULTI_LOOP_STATE_RUNNING;


klass->multi_task_context.task = gst_task_new (
    (GstTaskFunction) gst_curl_http_src_curl_multi_loop,
    (gpointer) & klass->multi_task_context, NULL);


gst_task_set_lock (klass->multi_task_context.task,
                   &klass->multi_task_context.task_rec_mutex);


if (gst_task_start (klass->multi_task_context.task) == FALSE) {
    GSTCURL_ERROR_PRINT ("Couldn't start curl_multi task! Aborting.");
    abort ();
}

GMutex

GMutex结构是一个不透明的数据结构，代表一个互斥量（mutex），它可以用来保护数据免受共享访问。

GRecMutex

GRecMutex结构是一个不透明的数据结构，代表一个递归互斥。它类似于GMutex，不同的是，它可以在同一线程中多次锁定GRecMutex而不造成死锁。在这样做的时候，必须注意在递归互斥被锁定的时候解锁它。

Gstreamer中的task

Java相关栏目本月热门文章