通过之前的文章我们知道TaskGroup（以下简称TG）是在死循环等待任务，然后切换栈去执行任务。在当前TG没有任务的时候会进行“工作窃取”窃取其他TG的任务。在没有任务的时候TG会“休眠”，当任务出现的时候被唤醒然后消费。
这个思路和线程中的条件变量类似。条件变量是线程间同步的一种方式。而bthread实现worker间的状态同步是通过“ParkingLot”。并且实现了也有与条件变量类似的wait(阻塞并等待）和signal(通知并唤醒）的操作。
ParkingLot（以下简称PL）直译是停车场，你可以理解成停放worker的停车场。我们暂时先不展开PL的定义。而是看一下ParkingLot与TaskControl（以下简称TC）与TaskGroup的关系。
TC中有ParkingLot类型的成员，是一个数组：
static const int PARKING_LOT_NUM = 4;
ParkingLot _pl[PARKING_LOT_NUM];也就是说一个TC有4个PL对象。因为全局只有一个TC，所以也就是全局只有4个PL。
TG中也有PL相关的成员(BTHREAD_DONT_SAVE_PARKING_STATE是开启的）：
ParkingLot* _pl;
#ifndef BTHREAD_DONT_SAVE_PARKING_STATE
ParkingLot::State _last_pl_state;
#endif_pl和_last_pl_state。_pl只是一个指针，其实他也源自TC中的pl。看TG的构造函数。
TaskGroup::TaskGroup(TaskControl* c)
... // 初始化列表，给成员赋值默认值，这里忽略
{
_steal_seed = butil::fast_rand();
_steal_offset = OFFSET_TABLE[_steal_seed % ARRAY_SIZE(OFFSET_TABLE)];
_pl = &c->_pl[butil::fmix64(pthread_numeric_id()) % TaskControl::PARKING_LOT_NUM];
}
butil::fmix64()是一个hash函数，用的murmurhash的算法，将输入的整型映射成另外一个整型。这里用pthread线程的id作为参赛，进行hash，然后把结果再对PARKING_LOT_NUM取模。相当于是从TC的4个PL中选择了一个PL，赋值给了TG！
换言之，TC下面的所有TG（worker）被分成了4组，每组共享一个PL。通过PL在调控TG之间bthread任务的生产与消费。之所以用4个PL，而不是一个PL，大概率也是为了减少race condition（竞争状态）减少性能开销。
我们常用的bthread_start_background()会调用TG的start_background()。
在TaskGroup::start_background()中的定义中有：
if (REMOTE) {
ready_to_run_remote(m->tid, (using_attr.flags & BTHREAD_NOSIGNAL));
} else {
ready_to_run(m->tid, (using_attr.flags & BTHREAD_NOSIGNAL));
}
ready_to_run_remote()和ready_to_run()的第二个参数nosignal，需要创建bthread任务的时候，给bthread设置属性：BTHREAD_NOSIGNAL。比如：
// 样例
bthread_t th;
bthread_attr_t tmp = BTHREAD_ATTR_NORMAL | BTHREAD_NOSIGNAL;
bthread_start_background(&th, &tmp, ProcessInputMessage, call_back_func);
不过通常我们调用bthread_start_background()的时候，第二个参数是设置为NULL的。所以可以暂时忽略nosignal相关逻辑。默认都是走signal的。注意这里的说的signal不是Unix C环境编程里面的信号。而是brpc自己给bthread实现的一套调控TG（worker）等待与唤醒的信号。
回看ready_to_run_remote()和ready_to_run()。ready_to_run()就是把任务入队到TG的 rq，ready_to_run_remote()是在当前线程不是brpc的worker()的时候（在worker外创建的 bthread任务），把任务通过TC入队到某个TG的 remote_rq。
ready_to_run()源码定义如下：
void TaskGroup::ready_to_run(bthread_t tid, bool nosignal) {
push_rq(tid);
if (nosignal) {
++_num_nosignal;
} else {
const int additional_signal = _num_nosignal;
_num_nosignal = 0;
_nsignaled += 1 + additional_signal;
_control->signal_task(1 + additional_signal);
}
}
ready_to_run()比较简洁，我们继续看下ready_to_run_remote()的定义：
void TaskGroup::ready_to_run_remote(bthread_t tid, bool nosignal) {
_remote_rq._mutex.lock();
while (!_remote_rq.push_locked(tid)) {
flush_nosignal_tasks_remote_locked(_remote_rq._mutex);
LOG_EVERY_SECOND(ERROR) << "_remote_rq is full, capacity="
<< _remote_rq.capacity();
::usleep(1000);
_remote_rq._mutex.lock();
}
if (nosignal) {
++_remote_num_nosignal;
_remote_rq._mutex.unlock();
} else {
const int additional_signal = _remote_num_nosignal;
_remote_num_nosignal = 0;
_remote_nsignaled += 1 + additional_signal;
_remote_rq._mutex.unlock();
_control->signal_task(1 + additional_signal);
}
}
先给当前TG的 remote_rq 加互斥锁。然后对 remote_rq 进行入队操作，这里是一个while循环，只有入队失败就执行flush_nosignal_tasks_remote_locked()然后休眠1ms，然后重新尝试入队。
这里入队失败的唯一原因就是remote_rq 的容量满了。flush_nosignal_tasks_remote_locked()的操作也无非就是发出一个信号，让remote_rq中的任务（TM/bthread)尽快被消费掉。给新的任务入队留出空间。另外flush_nosignal_tasks_remote_locked()内会做解锁操作，所以休眠1ms之后需要重新加锁。
回看ready_to_run_remote()，在while结束之后。表示新任务已经入队。前面已讲，nosignal多为false，所以忽略if(nosignal)的部分，关注else的部分。用当前remote_rq中还没有通知的任务个数+1，去做通知操作。也就是调用TaskControl的signal_task()。其实就是通知其他人来消费。
// Tell other groups that `n' tasks was just added to caller's runqueue
void signal_task(int num_task);看代码：
if (num_task <= 0) {
return;
}
// TODO(gejun): Current algorithm does not guarantee enough threads will
// be created to match caller's requests. But in another side, there's also
// many useless signalings according to current impl. Capping the concurrency
// is a good balance between performance and timeliness of scheduling.
if (num_task > 2) {
num_task = 2;
}
num_task 小于等于0 则返回，如果大于2，则重置为2。也就是说下面逻辑中num_task的有效值只有1和2。在上方“戈君”（BRPC作者）的注释中提到，把num_task不超过2，是在性能和调度时间直接的一种平衡。
这句话如何理解呢？其实是这样，如果TC的signal_task()通知的任务个数多，那么队列被消费的也就越快。消费的快本来是好事，但是也有个问题就是我们现在之所以走到signal_task()是因为我们在“生产”bthread任务，也就是说在执行bthread_start_background()（或其他函数）创建新任务。这个函数调用是阻塞的，如果signal_task()通知的任务个数太多，则会导致bthread_start_background()阻塞的时间拉长。所以这里说是找到一种平衡。
int start_index = butil::fmix64(pthread_numeric_id()) % PARKING_LOT_NUM;
num_task -= _pl[start_index].signal(1);
start_index计算方式和刚才给TG分配PL的相同，主要就是找到了当前TG（worker）所归属的PL。然后调用这个PL的成员函数signal(1)进行通知。好了，先暂停“生产者”函数调用视角。看下PL的定义，以及其signal()函数。
class BAIDU_CACHELINE_ALIGNMENT ParkingLot {
public:
class State {
public:
State(): val(0) {}
bool stopped() const { return val & 1; }
private:
friend class ParkingLot;
State(int val) : val(val) {}
int val;
};
ParkingLot() : _pending_signal(0) {}
... 成员函数：signal(int)、get_state()、wait()、stop()
private:
// higher 31 bits for signalling, LSB for stopping.
butil::atomic<int> _pending_signal;
};
有一个内部类State，其构造函数可以接收一个int。PL是它的友元，另外PL有一个私有成员_pending_signal，是一个原子类型。初始为0。
接着我们看下PL的成员函数signal(int)，也就是前面调用的那个。
// Wake up at most `num_task' workers.
// Returns #workers woken up.
int signal(int num_task) {
_pending_signal.fetch_add((num_task << 1), butil::memory_order_release);
return futex_wake_private(&_pending_signal, num_task);
}注释有言：唤醒最多num_task个worker，返回唤醒的worker。
代码实现中，寥寥两行。先给_pending_signal 加上num_task <<1（即num_task*2）。这里之所以累加的数字，要经过左移操作，其目的只是为了让其成为偶数。为什么这里需要一个偶数呢？在文章尾部会有讲解，大家稍安勿躁。
接着调用futex_wake_private(&_pending_signal, num_task)。那么问题又来了，futex_wake_private又是何方神圣呢？
在src/bthread/sys_futex.h中有定义。另外该文件中还有阈值配套的函数futex_wait_private()
inline int futex_wake_private(void* addr1, int nwake) {
return syscall(SYS_futex, addr1, (FUTEX_WAKE | FUTEX_PRIVATE_FLAG),
nwake, NULL, NULL, 0);
}
其实就是对于系统调用SYS_futex的封装。这里之所以通过syscall()传参，而不是直接调用的方式，来调用它。是因为SYS_futex没有被glibc export成库函数。我们通常使用的fork()、open()、write()等函数虽然也被称为系统调用，但其实是glibc把系统调用给export出来的封装函数。
继续介绍一下SYS_futex调用。就是通常说的futex，它是一种用户态和内核态混合的同步机制，可以简单理解为是一种效率较高的同步机制。pthread的很多API大多基于futex实现，细节不再展开。futex系统调用的API声明如下：
int futex(int *uaddr, int op, int val, const struct timespec *timeout,
int *uaddr2, int val3);参数解析：
返回值解析：
所以futex_wake_private()里面的syscall()等价于：
futex(&_pending_signal, (FUTEX_WAKE|FUTEX_PRIVATE_FLAG), num_task, NULL, NULL, 0);
FUTEX_WAKE是唤醒操作，FUTEX_PRIVATE_FLAG是一个标记，表示不和其他进程共享，可以减少开销。由于是唤醒操作，在brpc语境下，其返回值就是阻塞的worker个数。它的返回值会一路透传给futex_wake_private()以及PL的signal()函数。
彼时我们的观察视角也可以开始回溯，回到TC的signal_task()了。
int start_index = butil::fmix64(pthread_numeric_id()) % PARKING_LOT_NUM;
num_task -= _pl[start_index].signal(1);
_pl[start_index].signal(1)的返回值就是返回的worker个数了。然后将num_task减去唤醒的个数就是需要唤醒，但未唤醒的任务个数。接着看：
if (num_task > 0) {
for (int i = 1; i < PARKING_LOT_NUM && num_task > 0; ++i) {
if (++start_index >= PARKING_LOT_NUM) {
start_index = 0;
}
num_task -= _pl[start_index].signal(1);
}
}如果num_task不为0，则继续遍历TC的下一个PL，开始执行signal()操作去唤醒阻塞的worker。
接着：
if (num_task > 0 &&
FLAGS_bthread_min_concurrency > 0 && // test min_concurrency for performance
_concurrency.load(butil::memory_order_relaxed) < FLAGS_bthread_concurrency) {
// TODO: Reduce this lock
BAIDU_SCOPED_LOCK(g_task_control_mutex);
if (_concurrency.load(butil::memory_order_acquire) < FLAGS_bthread_concurrency) {
add_workers(1);
}
}如果任务还有剩余（表示消费者不够用），并且全局TC的并发度（_concurrency）小于gflag中配置的bthread_min_concurrency，那么就调用add_workers()去增加worker的数量。所以FLAGS_bthread_concurrency是worker(或者说是TG、pthread）个数的硬门槛。
好了，至此从“生产”bthread任务的角度，已经串完了整个流程。再从消费者的角度看一下ParkingLot。
其实上一篇文章已经对“消费”bthread任务的流程，讲的比较多了，其中涉及到了工作窃取（work stealing）以及汇编语言完成的栈空间切换。但是其中涉及到pl的部分没有重点介绍，我们来回顾一下TG的wait_task()函数。该函数是用来等待任务出现的。
bool TaskGroup::wait_task(bthread_t* tid) {
do {
if (_last_pl_state.stopped()) {
return false;
}
_pl->wait(_last_pl_state);
if (steal_task(tid)) {
return true;
}
} while (true);
}
_last_pl_state是ParkingLot::State，是TG的一个成员。回看其定义：
class State {
public:
State(): val(0) {}
bool stopped() const { return val & 1; }
private:
friend class ParkingLot;
State(int val) : val(val) {}
int val;
};
TG初始化的时候_last_pl_state是无参数构造的，所以其val是0。
看下它的stopped()，其实就是判断val是否是奇数！由于我们生产任务时，调用pl的signal()总是累加一个偶数(num_task <<1)：
_pending_signal.fetch_add((num_task << 1), butil::memory_order_release);
所以TaskGroup::wait_task()中第一个if。if(_last_pl_state.stopped()) 在正常情况下都是不成立的！不会触发return。而是继续向下走到了：
//TaskGroup::wait_task中
...
_pl->wait(_last_pl_state);
去等待任务出现。这个wait()在ParkingLot类中定义如下：
// Wait for tasks.
// If the `expected_state' does not match, wait() may finish directly.
void wait(const State& expected_state) {
futex_wait_private(&_pending_signal, expected_state.val, NULL);
}
和生产流程中我们看到的wake()类似，这里的其对等操作wait()，封装的是futex_wait_private()。闲言少叙，其最终等价于：
futex(&_pending_signal, (FUTEX_WAIT|FUTEX_PRIVATE_FLAG), expected_state.val, NULL, NULL, 0);
关于futex的等待操作，在介绍唤醒操作时也已经提及。这里结合参数可以这样理解，它阻塞在&_pending_signal这里，因为expected_state实际传入的是_last_pl_state，所以该wait操作其预期值也便是_last_pl_state.val。如果&_pending_signal存储的值和_last_pl_state.val相同则阻塞（也就是说还没有任务出现），否则解除阻塞。走到：
//TaskGroup::wait_task中
...
if (steal_task(tid)) {
return true;
}
去调用TG的steal_task()找任务。定义如下
(忽略宏BTHREAD_DONT_SAVE_PARKING_STATE）
bool steal_task(bthread_t* tid) {
if (_remote_rq.pop(tid)) {
return true;
}
_last_pl_state = _pl->get_state();
return _control->steal_task(tid, &_steal_seed, _steal_offset);
}
在当前TG的_remote_rq无任务的时候，_last_pl_state会从pl同步一次状态。
PL中的get_state()定义如下：
// Get a state for later wait().
State get_state() {
return _pending_signal.load(butil::memory_order_acquire);
}所以_last_pl_state同步的就是_pending_signal的最新值。其实从last_pl_state的名字早就可以看出，它存储的是上一次pl的状态了！
值得一提的是：&_pending_signal中存储的值其实并不表示任务的个数，尽管来任务来临时，它会做一次加法，但加的并不是任务数，并且在任务被消费后不会做减法。这里面值是没有具体意义的，其变化仅仅是一种状态“同步”的媒介！就像小说和电影中的工具人！。
好了，前面说了_last_pl_state正常情况下，判断stopped()都是不成立的，那么什么时候会成立呢？还是在ParkingLot中，它有一个stop()成员函数：
// Wakeup suspended wait() and make them unwaitable ever.
void stop() {
_pending_signal.fetch_or(1);
futex_wake_private(&_pending_signal, 10000);
}
其中会做fetch_or(1)操作，经此一役，_last_pl_state必然为奇数。而调用pl的stop()函数的地方只有一处，那就是TC中的stop_and_join()，而stop_and_join()又只在bthread_stop_world()这个函数调用的中被调用。调用链如下：
正常我们都不会调用，bthread_stop_world()，所以在_last_pl_state.stopped()在服务正常运转的情况下都不会为false。
Copyright© 2013-2020
All Rights Reserved 京ICP备2023019179号-8