nng协议nni_taskq_sys_init(void) 对nni_taskq

nng协议nni_taskq_sys_init(void) 对nni_taskq_systq 初始化

函数调用关系：nni_init(void) --> nni_plat_init(nni_init_helper) 今天分析的函数位于 nni_init_helper()函数中的 nni_taskq_sys_init() 。 这个函数主要用于针对全局变量 nni_taskq_systq 进行申请空间并初始化。

1. 确定线程数量 num_thr

这个通过宏定义或者函数获得。没定义则取CPU数量 * 2；最大值为16. 如果小于2，则指定num_thr 的值为2. 最少2个，最多16个。

2.执行初始化 nni_taskq_init(&nni_taskq_systq, num_thr)

nni_taskq 的定义：

struct nni_taskq_thr{nni_taskq *tqt_tq;nni_thr    tqt_thread;
}  struct    nni_taskq {nni_list       tq_tasks;nni_mtx        tq_mtx;nni_cv         tq_sched_cv;nni_cv         tq_wait_cv;nni_taskq_thr  *tq_threads;int            tq_nthreads;bool           tq_run;
}

基本步骤：

1. 为tq 申请空间，为tq->tq_threads 申请内存空间

2. 对线程数量赋值，tq->tq_nthreads = nthr .

3. 链表初始化 nni_taskq_systq->tq-tasks 互斥锁初始化，条件变量初始化

//请详细分析并给出使用范例
nni_mtx_init(&tq->mtx);
nni_cv_init(&tq->tq_sched_cv, &tq->tq_mtx);
nni_cv_init(&tq->tq_wait_cv, &tq->tq_mtx);

4. 循环对全局变量nni_taskq_systq 的成员变量所有的 tq_threads[i] 进行初始化。

5. th_threads[i] 的初始化流程：

tq->th_threads[i].tqt_tq = tq;

执行nni_thr_init 对 tqt_thread[i].tqt_thread 进行初始化。这个函数和其他的初始化类似，首先对第一层变量进行赋值，然后调用nni_plat_thr_init，初始化后的值为：

//tq的类型是  nni_taskq  *tq;
nni_thr_init(&tq->tq_threads[i].tqt_thread, nni_taskq_thread, &tq->tq_threads[i]);tq->tq_threads[i].tqt_tq = tq;&tq->tq_threads[i].tqt_thread 的类型是nni_thr
//nni_thr_init
tq->tq_threads[i].tqt_thread->done  = 0;
tq->tq_threads[i].tqt_thread->start = 0;
tq->tq_threads[i].tqt_thread->stop  = 0;
tq->tq_threads[i].tqt_thread->fn    = nni_taskq_thread;
tq->tq_threads[i].tqt_thread->arg   = &tq->tq_threads[i];//tq->tq_threads[i].tqt_thread->thr的类型是 nni_plat_thr//nni_plat_thr_init(&thr->thr, nni_thr_wrap, thr)
tq->tq_threads[i].tqt_thread->thr->func = nni_thr_wrap;
tq->tq_threads[i].tqt_thread->thr->arg  = tq->tq_threads[i].tqt_thread;

在nni_plat_thr_init(nni_plat_thr *thr, void (*fn)(void *), void *arg)中创建线程，执行线程函数：nni_plat_thr_main(void *arg)；

在线程函数中调用函数指针：thr->func(thr->arg); 也就是nni_plat_init 初始化的thr->func 亦即：nni_thr_wrap，参数是nni_plat_init 中设置的参数，这里是 &tq->tq_threads[i]；由于初始化时，start==0， stop==0 所以函数将等待在条件变量上。

static void
nni_thr_wrap(void *arg)
{nni_thr *thr = arg;int      start;nni_plat_mtx_lock(&thr->mtx);while (((start = thr->start) == 0) && (thr->stop == 0)) { //初始化条件，函数将等待在条件变量nni_plat_cv_wait(&thr->cv);}nni_plat_mtx_unlock(&thr->mtx);if ((start) && (thr->fn != NULL)) {thr->fn(thr->arg);}nni_plat_mtx_lock(&thr->mtx);thr->done = 1;nni_plat_cv_wake(&thr->cv);nni_plat_mtx_unlock(&thr->mtx);
}

在 nni_taskq_init 的最后，这里将执行nni_thr_run 函数，设置条件 start == 1 并唤醒等待在条件变量上的线程。

for (int i = 0; i < tq->tq_nthreads; i++) {nni_thr_run(&tq->tq_threads[i].tqt_thread);
}

函数继续执行，调用 nni_taskq_thread函数，参数是 &tq->tq_threads[i] ；

执行简图请参考resolv_thr 的初始化

static void
nni_taskq_thread(void *self)
{nni_taskq_thr *thr = self;nni_taskq     *tq  = thr->tqt_tq;nni_task      *task;nni_thr_set_name(NULL, "nng:task");nni_mtx_lock(&tq->tq_mtx);for (;;) {if ((task = nni_list_first(&tq->tq_tasks)) != NULL) {nni_list_remove(&tq->tq_tasks, task);nni_mtx_unlock(&tq->tq_mtx);task->task_cb(task->task_arg);nni_mtx_lock(&task->task_mtx);task->task_busy--;if (task->task_busy == 0) {nni_cv_wake(&task->task_cv);}nni_mtx_unlock(&task->task_mtx);nni_mtx_lock(&tq->tq_mtx);continue;}if (!tq->tq_run) {break;}nni_cv_wait(&tq->tq_sched_cv);}nni_mtx_unlock(&tq->tq_mtx);
}