bthread源码剖析（四）: 通过ParkingLot实现Worker间任务状态同步-阿里云开发者社区

通过之前的文章我们知道TaskGroup（以下简称TG）是在死循环等待任务，然后切换栈去执行任务。在当前TG没有任务的时候会进行“工作窃取”窃取其他TG的任务。在没有任务的时候TG会“休眠”，当任务出现的时候被唤醒然后消费。

这个思路和线程中的条件变量类似。条件变量是线程间同步的一种方式。而bthread实现worker间的状态同步是通过“ParkingLot”。并且实现了也有与条件变量类似的wait(阻塞并等待）和signal(通知并唤醒）的操作。

ParkingLot与TaskControl

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);

TaskControl::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()函数。

ParkingLot 的基础定义

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又是何方神圣呢？

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);

参数解析：

uaddr指针指向一个整型，存储一个整数。

op表示要执行的操作类型，比如唤醒(FUTEX_WAKE)、等待(FUTEX_WAIT)

val表示一个值，注意：对于不同的op类型，val语义不同。

对于等待操作：如果uaddr存储的整型与val相同则继续休眠等待。等待时间就是timeout参数。

对于唤醒操作：val表示，最多唤醒val 个阻塞等待uaddr上的“消费者”（之前对同一个uaddr调用过FUTEX_WAIT，姑且称之为消费者，其实在brpc语境中，就是阻塞的worker）。

timeout表示超时时间，仅对op类型为等待时有用。就是休眠等待的最长时间。在

uaddr2和val3可以忽略。

返回值解析：

对于等待操作：成功返回0，失败返回-1

对于唤醒操作：成功返回唤醒的之前阻塞在futex上的“消费者”个数。失败返回-1。

所以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()了。

继续 TaskControl::signal_task(int num_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()

ParkingLot::stop()

TaskControl::stop_and_join()

正常我们都不会调用，bthread_stop_world()，所以在_last_pl_state.stopped()在服务正常运转的情况下都不会为false。

bthread源码剖析（四）: 通过ParkingLot实现Worker间任务状态同步

ParkingLot与TaskControl

从生产者的角度出发

TaskControl::signal_task(int num_task)

ParkingLot 的基础定义

futex_wake_private()

继续 TaskControl::signal_task(int num_task)

热门文章

最新文章

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

bthread源码剖析（四）: 通过ParkingLot实现Worker间任务状态同步

ParkingLot与TaskControl

从生产者的角度出发

TaskControl::signal_task(int num_task)

ParkingLot 的基础定义

futex_wake_private()

继续 TaskControl::signal_task(int num_task)

热门文章

最新文章

相关电子书