bthread源码剖析（一）: 基本概念与TaskControl初始化

bthread是协程吗？

如果你使用过brpc，那么对bthread应该并不陌生。毫不夸张地说，brpc的精华全在bthread上了。bthread可以理解为“协程”，尽管官方文档的FAQ中，并不称之为协程。见：

https://github.com/apache/incubator-brpc/blob/master/docs/cn/bthread.md

若说到pthread大家都不陌生，是POSIX标准中定义的一套线程模型。应用于Unix Like系统，在Linux上pthread API的具体实现是NPTL库实现的。在Linux系统上，其实没有真正的线程，其采用的是LWP（轻量级进程）实现的线程。而bthread是brpc实现的一套“协程”，当然这并不是传统意义上的协程。就像1个进程可以开辟N个线程一样。传统意义上的协程是一个线程中开辟多个协程，也就是通常意义的N:1协程。比如微信开源的libco就是N:1的，libco属于非对称协程，区分caller和callee，而bthread是M:N的“协程”，每个bthread之间的平等的，所谓的M:N是指协程可以在线程间迁移。熟悉Go语言的朋友，应该对goroutine并不陌生，它也是M:N的。关于Go这部分描述，可以阅读：

[Goroutine并发调度模型深入之实现一个协程池]

当然准确的说法goroutine也并不等同于协程。不过由于通常也称goroutine为协程，从此种理解上来讲，bthread也可算是协程，只是不是传统意义上的协程！当然，咬文嚼字，没必要。

要实现M:N其中关键就是：工作窃取（Work Stealing）算法。不过在真正展开介绍工作窃取之前，我们先透视一下bthread的组成部分。

bthread的三个T

讲到bthread，首先要讲的三大件：TaskControl、TaskGroup、TaskMeta。以下简称TC、TG、TM。

三个都Task前缀，即任务。一个bthread表征的就是一个任务。而TM基本就是表示一个bthread（一个任务）上下文的完整结构体了。

TC进程内全局唯一，是bthread调控的中枢。TG和bthread用到的线程数(pthread）相当，每个线程都有一个TG。brpc中通常将运行着TG的pthread称之为 worker。因为并不是brpc server中的所有线程都是worker，因为brpc也会在没有TG的的pthread上运行代码。

虽然前面我们说bthread并不严格从属于一个pthread，但是bthread在运行的时候是必然要在一个pthread中的worker中被调用执行的。

TaskControl 先看TC是如何被创建的，TC对象的直接创建（new和初始化）是在get_or_new_task_control()中，这个函数顾名思义，就是获取TC，有则返之，无则造之。所以TC是进程内全局唯一的，也就是只有一个实例。而TG和TM是一个比一个多。

下面展示一下get_or_new_task_control的被调用链（表示函数的被调用关系），也就能更直观的发现TC是如何被创建的。

• get_or_new_task_control

• bthread_start_background

• bthread_start_urgent

• start_from_non_worker

• bthread_timer_add

我们通常用的bthread_start_background()或者定期器bthread_timer_add()都会调用到get_or_new_task_control。

看一下我们熟悉的bthread_start_background()的定义：

int bthread_start_background(bthread_t* __restrict tid,
                             const bthread_attr_t* __restrict attr,
                             void * (*fn)(void*),
                             void* __restrict arg) {
    bthread::TaskGroup* g = bthread::tls_task_group;
    if (g) {
        // start from worker
        return g->start_background<false>(tid, attr, fn, arg);
    }
    return bthread::start_from_non_worker(tid, attr, fn, arg);
}

如果能获取到thread local的TG，那么直接用这个TG来运行任务。如果获取不到TG，则说明当前还没有bthread的上下文（TC、TG都没有），所以调用start_from_non_worker()函数，其中再调用get_or_new_task_control()函数，从而创建出一个TC来。

get_or_new_task_control()的基本逻辑，直接看代码，主要地方我补充了注释：

inline TaskControl* get_or_new_task_control() {
    // 1. 全局变量TC（g_task_control）初始化原子变量
    butil::atomic<TaskControl*>* p = (butil::atomic<TaskControl*>*)&g_task_control;
    // 2.1 通过原子变量进行load，取出TC指针，如果不为空，直接返回
    TaskControl* c = p->load(butil::memory_order_consume);
    if (c != NULL) {
        return c;
    }
    // 2.2. 竞争加自旋锁，重复上一操作
    BAIDU_SCOPED_LOCK(g_task_control_mutex);
    c = p->load(butil::memory_order_consume);
    if (c != NULL) {
        return c;
    }

    // 2. 走到这，说明TC确实为NULL，开始new一个
    c = new (std::nothrow) TaskControl;
    if (NULL == c) {
        return NULL;
    }
    // 3. 用并发度concurrency来初始化全局TC
    int concurrency = FLAGS_bthread_min_concurrency > 0 ?
        FLAGS_bthread_min_concurrency :
        FLAGS_bthread_concurrency;
    if (c->init(concurrency) != 0) {
        LOG(ERROR) << "Fail to init g_task_control";
        delete c;
        return NULL;
    }

    // 4. 将全局TC存入原子变量中
    p->store(c, butil::memory_order_release);
    return c;
}

串完上述逻辑，我们来关注一下TC的初始化操作：TaskControl::init()

TaskControl::init() 源码十分简单：

   for (int i = 0; i < _concurrency; ++i) {
        const int rc = pthread_create(&_workers[i], NULL, worker_thread, this);
        if (rc) {
            LOG(ERROR) << "Fail to create _workers[" << i << "], " << berror(rc);
            return -1;
        }
    }

所谓的TC的初始化，就是调用了pthread_create()创建了N个新线程。N就是前面提到的并发度：concurrency。每个pthread线程的回调函数为worker_thread，通过这个函数也便引出了本文真正的主角TaskGroup了。

TaskControl::worker_thread() 毋庸置疑，这是一个static函数（回调函数一般为static）。在这个函数中会创建了一个TaskGroup。去掉一些日志逻辑，我们来看下源码，请大家关注我加的注释部分：

void* TaskControl::worker_thread(void* arg) {
    // 1. TG创建前的处理，里面也是回调g_worker_start_fun函数来执行操作，
    // 可以通过 bthread_set_worker_startfn() 来设置这个回调函数，
    // 实际但是很少用到这个功能
    run_worker_startfn();    

    // 2. 获取TC的指针
    TaskControl* c = static_cast<TaskControl*>(arg);
    // 2.1 创建一个TG
    TaskGroup* g = c->create_group();
    TaskStatistics stat;
    if (NULL == g) {
        LOG(ERROR) << "Fail to create TaskGroup in pthread=" << pthread_self();
        return NULL;
    }

    // 3.1 把thread local的tls_task_group 用刚才创建的TG来初始化
    tls_task_group = g;
    // 3.2 worker计数加1（_nworkers是bvar::Adder<int64_t>类型)
    c->_nworkers << 1;

    // 4. TG运行主任务（死循环）
    g->run_main_task();

    // 5. TG结束时返回状态信息，后面其实有输出stat的日志，这里去掉了
    stat = g->main_stat();
    // ...

    // 6. 各种清理操作
    tls_task_group = NULL;
    g->destroy_self();
    c->_nworkers << -1;
    return NULL;
}

至此我们的观察视角也就从TC平稳的过渡到TG了。其中TG最主要的函数就是run_main_task()。而你们心心念念的work stealing也不远了。

TaskGroup::run_main_task()内容较多，我们留到下一篇文章。现在我们先把目光聚焦在TC中，worker_thread()的主要步骤，我都用注释标明。其中值得一提的是2.1 调用create_group()去创建TG的地方，其源码如下（请大家继续关注我的注释）：

TaskGroup* TaskControl::create_group() {
    // 1. new出来一个TG
    TaskGroup* g = new (std::nothrow) TaskGroup(this);
    if (NULL == g) {
        LOG(FATAL) << "Fail to new TaskGroup";
        return NULL;
    }
    // 2. 进行TG的初始化
    if (g->init(FLAGS_task_group_runqueue_capacity) != 0) {
        LOG(ERROR) << "Fail to init TaskGroup";
        delete g;
        return NULL;
    }
    // 3. 将初始化成功的TG，添加到TC记录TG指针数组中
    if (_add_group(g) != 0) {
        delete g;
        return NULL;
    }
    return g;
}

第2部分的TG初始化我们也留待以后再讲，第三部分_add_group()，定义如下：

int TaskControl::_add_group(TaskGroup* g) {
    if (__builtin_expect(NULL == g, 0)) {
        return -1;
    }
    std::unique_lock<butil::Mutex> mu(_modify_group_mutex);
    if (_stop) {
        return -1;
    }
    size_t ngroup = _ngroup.load(butil::memory_order_relaxed);
    if (ngroup < (size_t)BTHREAD_MAX_CONCURRENCY) {
        _groups[ngroup] = g;
        _ngroup.store(ngroup + 1, butil::memory_order_release);
    }
    mu.unlock();
    // See the comments in _destroy_group
    // TODO: Not needed anymore since non-worker pthread cannot have TaskGroup
    signal_task(65536);
    return 0;
}

加brpc的互斥锁（butil::Mutex)，然后获取当前TG的个数：_ngroup，然后存储到_groups数组中。接着回写_ngroup加1，并回写其值。

关注一下这个函数开始时候的判断逻辑，我们谈一个“课外知识”：

   if (__builtin_expect(NULL == g, 0)) {
        return -1;
    }

__builtin_expect() 是gcc提供的一个用于优化分支预测的函数。大多人可能不了解分支预测。对于写惯了if else 的程序员来说，可能不了解if 和 else 中条件的顺序与性能之间的关系，比如：

if (NULL == g) {
    fun1();
} else {
    fun2();
}

和

if (NULL != g) {
    fun2();
} else {
    fun1();
}

这两种写法有区别么？答案是有。具体哪些写法的性能相对高一些，和g 与 NULL的概率有关。如果g绝大概率为NULL，那么第一种性能高，否则是第二种。

CPU中有流水线，具体而言是CPU有指令预取功能，CPU预先取出下一条指令，可以减少CPU等待取指令的耗时，从而可以提供CPU的效率。但若存在跳转指令（jmp，比如if else的时候）， CPU在执行当前指令时，从内存中预取出了当前指令的下一条指令。执行完当前指令后，CPU发现不是要执行预取的指令，则需要重新取指令。当然效率会低。

但是若要程序员如此费心的编排if else，那可就太累了。因此__builtin_expect()应运而生，其作用是"允许程序员将最有可能执行的分支告诉编译器"。它的用法为：

__builtin_expect(EXP, N)

意思是：EXP==N的概率很大。从而让编译器自己生产出高效的代码。所以

    if (__builtin_expect(NULL == g, 0)) {
        return -1;
    }

就是告诉编译器：NULL != g 的概率很大！

好了，一则小知识分享给大家。关于bthread的解读，请关注后续文章！

本文由哈喽比特于2年以前收录，如有侵权请联系我们。
文章来源：https://mp.weixin.qq.com/s/PqLRz01JudI88M7zwau7sQ