1. Context Switching

bthread 中使用 libcontext 实现协程间的切换，原理类似 汇编魔法实现 C++ 协程 中的方法。看一个单元测试中的例子（ 在线执行 ）：

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <iostream>
typedef void *bthread_fcontext_t;
extern "C" bthread_fcontext_t bthread_make_fcontext(void *sp, size_t size,
                                                    void (*fn)(intptr_t));
__asm(
    ".text\n"
    ".globl bthread_make_fcontext\n"
    ".type bthread_make_fcontext,@function\n"
    ".align 16\n"
    "bthread_make_fcontext:\n"
    "    movq  %rdi, %rax\n"
    "    andq  $-16, %rax\n"
    "    leaq  -0x48(%rax), %rax\n"
    "    movq  %rdx, 0x38(%rax)\n"
    "    stmxcsr  (%rax)\n"
    "    fnstcw   0x4(%rax)\n"
    "    leaq  finish(%rip), %rcx\n"
    "    movq  %rcx, 0x40(%rax)\n"
    "    ret \n"
    "finish:\n"
    "    xorq  %rdi, %rdi\n"
    "    call  _exit@PLT\n"
    "    hlt\n"
    ".size bthread_make_fcontext,.-bthread_make_fcontext\n"
    ".section .note.GNU-stack,\"\",%progbits\n"
    ".previous\n");
extern "C" intptr_t bthread_jump_fcontext(bthread_fcontext_t *ofc,
                                          bthread_fcontext_t nfc, intptr_t vp);
__asm(
    ".text\n"
    ".globl bthread_jump_fcontext\n"
    ".type bthread_jump_fcontext,@function\n"
    ".align 16\n"
    "bthread_jump_fcontext:\n"
    "    pushq  %rbp  \n"
    "    pushq  %rbx  \n"
    "    pushq  %r15  \n"
    "    pushq  %r14  \n"
    "    pushq  %r13  \n"
    "    pushq  %r12  \n"
    "    leaq  -0x8(%rsp), %rsp\n"
    "    movq  %rsp, (%rdi)\n"
    "    movq  %rsi, %rsp\n"
    "    leaq  0x8(%rsp), %rsp\n"
    "    popq  %r12  \n"
    "    popq  %r13  \n"
    "    popq  %r14  \n"
    "    popq  %r15  \n"
    "    popq  %rbx  \n"
    "    popq  %rbp  \n"
    "    popq  %r8\n"
    "    movq  %rdx, %rax\n"
    "    movq  %rdx, %rdi\n"
    "    jmp  *%r8\n"
    ".size bthread_jump_fcontext,.-bthread_jump_fcontext\n"
    ".section .note.GNU-stack,\"\",%progbits\n"
    ".previous\n");
bthread_fcontext_t fcm;
bthread_fcontext_t fc;
typedef std::pair<int, int> pair_t;
static void f(intptr_t param) {
  pair_t *p = (pair_t *)param;
  printf("In Routine: fcm %p fc %p\n", fcm, fc);
  p = (pair_t *)bthread_jump_fcontext(&fc, fcm,
                                      (intptr_t)(p->first + p->second));
  printf("In Routine Again: fcm %p fc %p\n", fcm, fc);
  bthread_jump_fcontext(&fc, fcm, (intptr_t)(p->first + p->second));
int main() {
  fcm = NULL;
  std::size_t size(8192);
  void *sp = malloc(size);
  pair_t p(std::make_pair(2, 7));
  fc = bthread_make_fcontext((char *)sp + size, size, f);
  printf("Start Routine: fcm %p fc %p\n", fcm, fc);
  int res = (int)bthread_jump_fcontext(&fcm, fc, (intptr_t)&p);
  printf("Back to Main: %d + %d = %d\n", p.first, p.second, res);
  p = std::make_pair(5, 6);
  printf("Resume Routine: fcm %p fc %p\n", fcm, fc);
  res = (int)bthread_jump_fcontext(&fcm, fc, (intptr_t)&p);
  printf("Back to Main Again: %d + %d = %d\n", p.first, p.second, res);
  return 0;
bthread 中使用 TaskMeta 结构封装协程的元信息：
// task_meta.h
struct TaskMeta {
  // [Not Reset]
  butil::atomic<ButexWaiter*> current_waiter;
  uint64_t current_sleep;
  // A builtin flag to mark if the thread is stopping.
  bool stop;
  // The thread is interrupted and should wake up from some blocking ops.
  bool interrupted;
  // Scheduling of the thread can be delayed.
  bool about_to_quit;
  // [Not Reset] guarantee visibility of version_butex.
  pthread_spinlock_t version_lock;
  // [Not Reset] only modified by one bthread at any time, no need to be atomic
  uint32_t* version_butex;
  // The identifier. It does not have to be here, however many code is
  // simplified if they can get tid from TaskMeta.
  bthread_t tid;
  // User function and argument
  void* (*fn)(void*);
  void* arg;
  // Stack of this task.
  ContextualStack* stack;
  // Attributes creating this task
  bthread_attr_t attr;
  // Statistics
  int64_t cpuwide_start_ns;
  TaskStatistics stat;
  // bthread local storage, sync with tls_bls (defined in task_group.cpp)
  // when the bthread is created or destroyed.
  // DO NOT use this field directly, use tls_bls instead.
  LocalStorage local_storage;
 public:
  // Only initialize [Not Reset] fields, other fields will be reset in
  // bthread_start* functions
  TaskMeta()
    : current_waiter(NULL)
      , current_sleep(0)
      , stack(NULL) {
      pthread_spin_init(&version_lock, 0);
      version_butex = butex_create_checked<uint32_t>();
      *version_butex = 1;
  ~TaskMeta() {
    butex_destroy(version_butex);
    version_butex = NULL;
    pthread_spin_destroy(&version_lock);
  void set_stack(ContextualStack* s) {
    stack = s;
  ContextualStack* release_stack() {
    ContextualStack* tmp = stack;
    stack = NULL;
    return tmp;
  StackType stack_type() const {
    return static_cast<StackType>(attr.stack_type);
这里先只关注协程栈 stack。协程首次调度前会分配对应的调用栈：




    

// task_group_inl.h
ContextualStack* stk = get_stack(next_meta->stack_type(), task_runner);
// stack_inl.h
inline ContextualStack* get_stack(StackType type, void (*entry)(intptr_t)) {
  switch (type) {
    case STACK_TYPE_PTHREAD:
      return NULL;
    case STACK_TYPE_SMALL:
      return StackFactory<SmallStackClass>::get_stack(entry);
    case STACK_TYPE_NORMAL:  // 默认使用 Normal
      return StackFactory<NormalStackClass>::get_stack(entry);
    case STACK_TYPE_LARGE:
      return StackFactory<LargeStackClass>::get_stack(entry);
    case STACK_TYPE_MAIN:
      return StackFactory<MainStackClass>::get_stack(entry);
  return NULL;
template <typename StackClass> struct StackFactory {
  struct Wrapper : public ContextualStack {
    explicit Wrapper(void (*entry)(intptr_t)) {
      // 分配栈内存，默认带内存保护，注意这里的 StackClass::stack_size_flag
      if (allocate_stack_storage(&storage, *StackClass::stack_size_flag,
                                 FLAGS_guard_page_size) != 0) {
        storage.zeroize();
        context = NULL;
        return;
      // 初始化 context
      context = bthread_make_fcontext(storage.bottom, storage.stacksize, entry);
      stacktype = (StackType)StackClass::stacktype;
    ~Wrapper() {
      if (context) {
        context = NULL;
        // 析构时释放
        deallocate_stack_storage(&storage);
        storage.zeroize();
  static ContextualStack* get_stack(void (*entry)(intptr_t)) {
    return butil::get_object<Wrapper>(entry);
  static void return_stack(ContextualStack* sc) {
    // 对象池回收时不会执行析构，申请的栈空间可以复用
    butil::return_object(static_cast<Wrapper*>(sc));
// stack.h
struct StackStorage {
  int stacksize;
  int guardsize;
  void* bottom;
  unsigned valgrind_stack_id;
  // Clears all members.
  void zeroize() {
    stacksize = 0;
    guardsize = 0;
    bottom = NULL;
    valgrind_stack_id = 0;
// Allocate a piece of stack.
int allocate_stack_storage(StackStorage* s, int stacksize, int guardsize);
// Deallocate a piece of stack. Parameters MUST be returned or set by the
// corresponding allocate_stack_storage() otherwise behavior is undefined.
void deallocate_stack_storage(StackStorage* s);
enum StackType {
  STACK_TYPE_MAIN = 0,
  STACK_TYPE_PTHREAD = BTHREAD_STACKTYPE_PTHREAD,
  STACK_TYPE_SMALL = BTHREAD_STACKTYPE_SMALL,
  STACK_TYPE_NORMAL = BTHREAD_STACKTYPE_NORMAL,
  STACK_TYPE_LARGE = BTHREAD_STACKTYPE_LARGE
struct ContextualStack {
  bthread_fcontext_t context;
  StackType stacktype;
  StackStorage storage;
// stack.cpp
int allocate_stack_storage(StackStorage* s, int stacksize_in, int guardsize_in) {
  const static int PAGESIZE = getpagesize();
  const int PAGESIZE_M1 = PAGESIZE - 1;
  const int MIN_STACKSIZE = PAGESIZE * 2;
  const int MIN_GUARDSIZE = PAGESIZE;
  // Align stacksize
  const int stacksize =
    (std::max(stacksize_in, MIN_STACKSIZE) + PAGESIZE_M1) &
    ~PAGESIZE_M1;  // 栈对齐，PAGESIZE 一般是 4K
  if (guardsize_in <= 0) {
    ...
  } else {
    // Align guardsize
    const int guardsize =
      (std::max(guardsize_in, MIN_GUARDSIZE) + PAGESIZE_M1) &
      ~PAGESIZE_M1;
    const int memsize = stacksize + guardsize;
    void* const mem = mmap(NULL, memsize, (PROT_READ | PROT_WRITE),
                           (MAP_PRIVATE | MAP_ANONYMOUS), -1, 0);
    if (MAP_FAILED == mem) {
      PLOG_EVERY_SECOND(ERROR) 
        << "Fail to mmap size=" << memsize << " stack_count="
        << s_stack_count.load(butil::memory_order_relaxed)
        << ", possibly limited by /proc/sys/vm/max_map_count";
      // may fail due to limit of max_map_count (65536 in default)
      return -1;
    void* aligned_mem = (void*)(((intptr_t)mem + PAGESIZE_M1) & ~PAGESIZE_M1);
    if (aligned_mem != mem) {
      LOG_ONCE(ERROR) << "addr=" << mem << " returned by mmap is not "
        "aligned by pagesize=" << PAGESIZE;
    const int offset = (char*)aligned_mem - (char*)mem;
    // 使用 mprotect，当栈溢出时抛出异常
    if (guardsize <= offset ||
        mprotect(aligned_mem, guardsize - offset, PROT_NONE) != 0) {
      munmap(mem, memsize);
      PLOG_EVERY_SECOND(ERROR) 
        << "Fail to mprotect " << (void*)aligned_mem << " length="
        << guardsize - offset; 
      return -1;
    s_stack_count.fetch_add(1, butil::memory_order_relaxed);
    s->bottom = (char*)mem + memsize;
    s->stacksize = stacksize;
    s->guardsize = guardsize;
    if (RunningOnValgrind()) {
      s->valgrind_stack_id = VALGRIND_STACK_REGISTER(
        s->bottom, (char*)s->bottom - stacksize);
    } else {
      s->valgrind_stack_id = 0;
    return 0;
int* SmallStackClass::stack_size_flag = &FLAGS_stack_size_small;
int* NormalStackClass::stack_size_flag = &FLAGS_stack_size_normal;
int* LargeStackClass::stack_size_flag = &FLAGS_stack_size_large;
2. Work Stealing
核心思路是当本线程内没有待执行的任务时，从其他线程的任务队列中窃取任务执行。首先来看 work stealing 时使用的无锁队列 src/bthread/work_stealing_queue.h：




    

template <typename T>
class WorkStealingQueue {
 public:
  WorkStealingQueue() : _bottom(1), _capacity(0), _buffer(NULL), _top(1) {}
  int init(size_t capacity) {
    if (_capacity != 0) {
      LOG(ERROR) << "Already initialized";
      return -1;
    if (capacity == 0) {
      LOG(ERROR) << "Invalid capacity=" << capacity;
      return -1;
    if (capacity & (capacity - 1)) {
      LOG(ERROR) << "Invalid capacity=" << capacity
                 << " which must be power of 2";
      return -1;
    _buffer = new (std::nothrow) T[capacity];
    if (NULL == _buffer) {
      return -1;
    _capacity = capacity;
    return 0;
  // 从底部追加，非线程安全，与 steal 线程安全
  bool push(const T& x) {
    const size_t b = _bottom.load(butil::memory_order_relaxed);
    const size_t t = _top.load(butil::memory_order_acquire);
    if (b >= t + _capacity) {  // Full queue.
      return false;
    _buffer[b & (_capacity - 1)] = x;
    _bottom.store(b + 1, butil::memory_order_release);
    return true;
  // 从底部弹出，非线程安全，与 steal 线程安全
  bool pop(T* val) {
    const size_t b = _bottom.load(butil::memory_order_relaxed);
    size_t t = _top.load(butil::memory_order_relaxed);
    if (t >= b) {
      // fast check since we call pop() in each sched.
      // Stale _top which is smaller should not enter this branch.
      return false;
    const size_t newb = b - 1;
    _bottom.store(newb, butil::memory_order_relaxed);
    butil::atomic_thread_fence(butil::memory_order_seq_cst);
    t = _top.load(butil::memory_order_relaxed);
    if (t > newb) {
      _bottom.store(b, butil::memory_order_relaxed);
      return false;
    *val = _buffer[newb & (_capacity - 1)];
    if (t != newb) {
      return true;
    // Single last element, compete with steal()
    // 对于最后一个元素，使用 CAS 保证和 steal 并发时的线程安全
    const bool popped = _top.compare_exchange_strong(
        t, t + 1, butil::memory_order_seq_cst, butil::memory_order_relaxed);
    _bottom.store(b, butil::memory_order_relaxed);
    return popped;
  // 从顶部窃取，线程安全
  bool steal(T* val) {
    size_t t = _top.load(butil::memory_order_acquire);
    size_t b = _bottom.load(butil::memory_order_acquire);
    if (t >= b) {
      // Permit false negative for performance considerations.
      return false;
    do {
      butil::atomic_thread_fence(butil::memory_order_seq_cst);
      b = _bottom.load(butil::memory_order_acquire);
      if (t >= b) {
        return false;
      *val = _buffer[t & (_capacity - 1)];
      // CAS 保证线程安全
    } while (!_top.compare_exchange_strong(
        t, t + 1, butil::memory_order_seq_cst, butil::memory_order_relaxed));
    return true;
 private:
  // Copying a concurrent structure makes no sense.
  DISALLOW_COPY_AND_ASSIGN(WorkStealingQueue);
  butil::atomic<size_t> _bottom;
  size_t _capacity;
  T* _buffer;
  butil::atomic<size_t> BAIDU_CACHELINE_ALIGNMENT _top;  // 分开到两个 CacheLine
push 和 pop 仅在底部操作，非线程安全。steal 仅在顶部窃取，通过 CAS 保证线程安全。




    

接着来看 bthread 启动的流程：
// test/bthread_unittest.cpp
TEST_F(BthreadTest, sanity) {
  LOG(INFO) << "main thread " << pthread_self();
  bthread_t th1;
  ASSERT_EQ(0, bthread_start_urgent(&th1, NULL, misc, (void*)1));
  LOG(INFO) << "back to main thread " << th1 << " " << pthread_self();
  ASSERT_EQ(0, bthread_join(th1, NULL));
// bthread.cpp
int bthread_start_urgent(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 bthread::TaskGroup::start_foreground(&g, tid, attr, fn, arg);
  // 首次执行，需要初始化
  return bthread::start_from_non_worker(tid, attr, fn, arg);
BUTIL_FORCE_INLINE int
  start_from_non_worker(bthread_t* __restrict tid,
                        const bthread_attr_t* __restrict attr,
                        void * (*fn)(void*),
                        void* __restrict arg) {
  // 获取 TaskControl 全局单例
  TaskControl* c = get_or_new_task_control();
  if (NULL == c) {
    return ENOMEM;
  if (attr != NULL && (attr->flags & BTHREAD_NOSIGNAL)) {
    // Remember the TaskGroup to insert NOSIGNAL tasks for 2 reasons:
    // 1. NOSIGNAL is often for creating many bthreads in batch,
    //    inserting into the same TaskGroup maximizes the batch.
    // 2. bthread_flush() needs to know which TaskGroup to flush.
    TaskGroup* g = tls_task_group_nosignal;
    if (NULL == g) {
      g = c->choose_one_group();
      tls_task_group_nosignal = g;
    return g->start_background<true>(tid, attr, fn, arg);
  // 加入队列
  return c->choose_one_group()->start_background<true>(
    tid, attr, fn, arg);
inline TaskControl* get_or_new_task_control() {
  butil::atomic<TaskControl*>* p = (butil::atomic<TaskControl*>*)&g_task_control;
  TaskControl* c = p->load(butil::memory_order_consume);
  if (c != NULL) {
    return c;
  BAIDU_SCOPED_LOCK(g_task_control_mutex);  // 全局锁
  c = p->load(butil::memory_order_consume);
  if (c != NULL) {
    return c;
  c = new (std::nothrow) TaskControl;
  if (NULL == c) {
    return NULL;
  int concurrency = FLAGS_bthread_min_concurrency > 0 ?
    FLAGS_bthread_min_concurrency :
  FLAGS_bthread_concurrency;
  // 初始化，concurrency 为工作线程数
  if (c->init(concurrency) != 0) {
    LOG(ERROR) << "Fail to init g_task_control";
    delete c;
    return NULL;
  p->store(c, butil::memory_order_release);
  return c;
// task_control.cpp
int TaskControl::init(int concurrency) {
  if (_concurrency != 0) {
    LOG(ERROR) << "Already initialized";
    return -1;
  if (concurrency <= 0) {
    LOG(ERROR) << "Invalid concurrency=" << concurrency;
    return -1;
  _concurrency = concurrency;
  // Make sure TimerThread is ready.
  if (get_or_create_global_timer_thread() == NULL) {
    LOG(ERROR) << "Fail to get global_timer_thread";
    return -1;
  _workers.resize(_concurrency);   
  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;
  _worker_usage_second.expose("bthread_worker_usage");
  _switch_per_second.expose("bthread_switch_second");
  _signal_per_second.expose("bthread_signal_second");
  _status.expose("bthread_group_status");
  // Wait for at least one group is added so that choose_one_group()
  // never returns NULL.
  // TODO: Handle the case that worker quits before add_group
  while (_ngroup == 0) {
    usleep(100);  // TODO: Elaborate
  return 0;
bthread 后台会开启多个 worker_thread 线程执行 bthread 任务：




    

// task_control.cpp
void* TaskControl::worker_thread(void* arg) {
  run_worker_startfn();
  TaskControl* c = static_cast<TaskControl*>(arg);
  TaskGroup* g = c->create_group();  // 每个线程有一个对应的 TaskGroup
  TaskStatistics stat;
  if (NULL == g) {
    LOG(ERROR) << "Fail to create TaskGroup in pthread=" << pthread_self();
    return NULL;
  BT_VLOG << "Created worker=" << pthread_self()
    << " bthread=" << g->main_tid();
  tls_task_group = g;  // 使用 TLS 存储线程对应的 TaskGroup
  c->_nworkers << 1;
  g->run_main_task();  // 任务主循环
  stat = g->main_stat();
  BT_VLOG << "Destroying worker=" << pthread_self() << " bthread="
    << g->main_tid() << " idle=" << stat.cputime_ns / 1000000.0
    << "ms uptime=" << g->current_uptime_ns() / 1000000.0 << "ms";
  tls_task_group = NULL;
  g->destroy_self();
  c->_nworkers << -1;
  return NULL;
// task_group.cpp
void TaskGroup::run_main_task() {
  bvar::PassiveStatus<double> cumulated_cputime(
    get_cumulated_cputime_from_this, this);
  std::unique_ptr<bvar::PerSecond<bvar::PassiveStatus<double> > > usage_bvar;
  TaskGroup* dummy = this;
  bthread_t tid;
  while (wait_task(&tid)) {  // 获取任务
    TaskGroup::sched_to(&dummy, tid);  // 调度执行
    DCHECK_EQ(this, dummy);
    DCHECK_EQ(_cur_meta->stack, _main_stack);
    if (_cur_meta->tid != _main_tid) {
      TaskGroup::task_runner(1/*skip remained*/);
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);
bool steal_task(bthread_t* tid) {
  // 本地队列中有任务，优先本地
  if (_remote_rq.pop(tid)) {
    return true;
  // 否则通过 TaskControl 窃取全局的任务
  return _control->steal_task(tid, &_steal_seed, _steal_offset);
// task_control.cpp
bool TaskControl::steal_task(bthread_t* tid, size_t* seed, size_t offset) {
  // 1: Acquiring fence is paired with releasing fence in _add_group to
  // avoid accessing uninitialized slot of _groups.
  const size_t ngroup = _ngroup.load(butil::memory_order_acquire/*1*/);
  if (0 == ngroup) {
    return false;
  // NOTE: Don't return inside `for' iteration since we need to update |seed|
  bool stolen = false;
  size_t s = *seed;
  for (size_t i = 0; i < ngroup; ++i, s += offset) {
    TaskGroup* g = _groups[s % ngroup];
    // g is possibly NULL because of concurrent _destroy_group
    if (g) {
      if (g->_rq.steal(tid)) {  // 无锁窃取
        stolen = true;
        break;
      if (g->_remote_rq.pop(tid)) {  // 有锁窃取
        stolen = true;
        break;
  *seed = s;
  return stolen;
// task_group_inl.h
inline void TaskGroup::sched_to(TaskGroup** pg, bthread_t next_tid) {
  TaskMeta* next_meta = address_meta(next_tid);
  if (next_meta->stack == NULL) {
    ContextualStack* stk = get_stack(next_meta->stack_type(), task_runner);
    if (stk) {
      next_meta->set_stack(stk);
    } else {
      // stack_type is BTHREAD_STACKTYPE_PTHREAD or out of memory,
      // In latter case, attr is forced to be BTHREAD_STACKTYPE_PTHREAD.
      // This basically means that if we can't allocate stack, run
      // the task in pthread directly.
      next_meta->attr.stack_type = BTHREAD_STACKTYPE_PTHREAD;
      next_meta->set_stack((*pg)->_main_stack);
  // Update now_ns only when wait_task did yield.
  sched_to(pg, next_meta);  // 执行
// task_group.cpp
void TaskGroup::sched_to(TaskGroup** pg, TaskMeta* next_meta) {
  TaskGroup* g = *pg;
  // Save errno so that errno is bthread-specific.
  const int saved_errno = errno;
  void* saved_unique_user_ptr = tls_unique_user_ptr;
  TaskMeta* const cur_meta = g->_cur_meta;
  const int64_t now = butil::cpuwide_time_ns();
  const int64_t elp_ns = now - g->_last_run_ns;
  g->_last_run_ns = now;
  cur_meta->stat.cputime_ns += elp_ns;
  if (cur_meta->tid != g->main_tid()) {
    g->_cumulated_cputime_ns += elp_ns;
  ++cur_meta->stat.nswitch;
  ++ g->_nswitch;
  // Switch to the task
  if (__builtin_expect(next_meta != cur_meta, 1)) {
    g->_cur_meta = next_meta;
    // Switch tls_bls
    cur_meta->local_storage = tls_bls;
    tls_bls = next_meta->local_storage;
    // Logging must be done after switching the local storage, since the logging lib
    // use bthread local storage internally, or will cause memory leak.
    if ((cur_meta->attr.flags & BTHREAD_LOG_CONTEXT_SWITCH) ||
        (next_meta->attr.flags & BTHREAD_LOG_CONTEXT_SWITCH)) {
      LOG(INFO) << "Switch bthread: " << cur_meta->tid << " -> "
        << next_meta->tid;
    if (cur_meta->stack != NULL) {
      if (next_meta->stack != cur_meta->stack) {
        jump_stack(cur_meta->stack, next_meta->stack);  // 协程切换
        // probably went to another group, need to assign g again.
        g = tls_task_group;
    // else because of ending_sched(including pthread_task->pthread_task)
  } else {
    LOG(FATAL) << "bthread=" << g->current_tid() << " sched_to itself!";
  while (g->_last_context_remained) {
    RemainedFn fn = g->_last_context_remained;
    g->_last_context_remained = NULL;
    fn(g->_last_context_remained_arg);
    g = tls_task_group;
  // Restore errno
  errno = saved_errno;
  tls_unique_user_ptr = saved_unique_user_ptr;
  *pg = g;
// stack_inl.h
inline void jump_stack(ContextualStack* from, ContextualStack* to) {
  bthread_jump_fcontext(&from->context, to->context, 0/*not skip remained*/);
从外部线程通过 TaskControl 新增 bthread 的流程：




    

// task_group.cpp
template <bool REMOTE>
int TaskGroup::start_background(bthread_t* __restrict th,
                                const bthread_attr_t* __restrict attr,
                                void * (*fn)(void*),
                                void* __restrict arg) {
    if (__builtin_expect(!fn, 0)) {
        return EINVAL;
    const int64_t start_ns = butil::cpuwide_time_ns();
    const bthread_attr_t using_attr = (attr ? *attr : BTHREAD_ATTR_NORMAL);
    butil::ResourceId<TaskMeta> slot;
    TaskMeta* m = butil::get_resource(&slot);
    if (__builtin_expect(!m, 0)) {
        return ENOMEM;
    CHECK(m->current_waiter.load(butil::memory_order_relaxed) == NULL);
    m->stop = false;
    m->interrupted = false;
    m->about_to_quit = false;
    m->fn = fn;
    m->arg = arg;
    CHECK(m->stack == NULL);
    m->attr = using_attr;
    m->local_storage = LOCAL_STORAGE_INIT;
    m->cpuwide_start_ns = start_ns;
    m->stat = EMPTY_STAT;
    m->tid = make_tid(*m->version_butex, slot);
    *th = m->tid;
    if (using_attr.flags & BTHREAD_LOG_START_AND_FINISH) {
        LOG(INFO) << "Started bthread " << m->tid;
    _control->_nbthreads << 1;
    if (REMOTE) {
        // 外部线程
        ready_to_run_remote(m->tid, (using_attr.flags & BTHREAD_NOSIGNAL));
    } else {
        ready_to_run(m->tid, (using_attr.flags & BTHREAD_NOSIGNAL));
    return 0;
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);  // 唤醒工作线程执行任务
3. Futex & Butex
Linux 中提供了快速的用户态锁 futex，bRPC 中进行了简单的封装：




    

// sys_futex.h
inline int futex_wait_private(void *addr1, int expected,
                              const timespec *timeout) {
  // 当 *addr1 == expected 时，线程挂起等待
  return syscall(SYS_futex, addr1, (FUTEX_WAIT | FUTEX_PRIVATE_FLAG), expected,
                 timeout, NULL, 0);
inline int futex_wake_private(void *addr1, int nwake) {
  // 唤醒 nwake 个等待的线程
  return syscall(SYS_futex, addr1, (FUTEX_WAKE | FUTEX_PRIVATE_FLAG), nwake,
                 NULL, NULL, 0);
对于 MacOS，bthread 也提供了用户态的实现，可以自行查看 src/bthread/sys_futex.cpp 中对应的实现。使用 futex 可以构造更快速的互斥锁：
// mutex.h
class FastPthreadMutex {
public:
  FastPthreadMutex() : _futex(0) {}
  ~FastPthreadMutex() {}
  void lock();
  void unlock();
  bool try_lock();
private:
  DISALLOW_COPY_AND_ASSIGN(FastPthreadMutex);
  int lock_contended();
  unsigned _futex;
// mutex.cpp
// Implement bthread_mutex_t related functions
struct MutexInternal {
  butil::static_atomic<unsigned char> locked;
  butil::static_atomic<unsigned char> contended;
  unsigned short padding;
const MutexInternal MUTEX_CONTENDED_RAW = {{1}, {1}, 0};
const MutexInternal MUTEX_LOCKED_RAW = {{1}, {0}, 0};
// Define as macros rather than constants which can't be put in read-only
// section and affected by initialization-order fiasco.
#define BTHREAD_MUTEX_CONTENDED                                                \
  (*(const unsigned *)&bthread::MUTEX_CONTENDED_RAW)
#define BTHREAD_MUTEX_LOCKED (*(const unsigned *)&bthread::MUTEX_LOCKED_RAW)
int FastPthreadMutex::lock_contended() {
  butil::atomic<unsigned> *whole = (butil::atomic<unsigned> *)&_futex;
  // 将状态原子修改为加锁 + 等待
  // 如果这期间锁被释放了，那么该函数直接退出
  while (whole->exchange(BTHREAD_MUTEX_CONTENDED) & BTHREAD_MUTEX_LOCKED) {
    // 否则调用 futex 尝试挂起当前线程，并且要求状态为加锁 + 等待（因为中间锁可能会被释放）
    if (futex_wait_private(whole, BTHREAD_MUTEX_CONTENDED, NULL) < 0 &&
        errno != EWOULDBLOCK) {
      return errno;
    // futex_wait_private 非 EWOULDBLOCK 失败的情况下，继续循环重试
  return 0;
void FastPthreadMutex::lock() {
  bthread::MutexInternal *split = (bthread::MutexInternal *)&_futex;
  // 加锁，如果 locked 原先的值为 0，lock 直接返回成功
  if (split->locked.exchange(1, butil::memory_order_acquire)) {
    // 如果已经是加锁状态，那么进入锁竞争的处理
    (void)lock_contended();
bool FastPthreadMutex::try_lock() {
  bthread::MutexInternal *split = (bthread::MutexInternal *)&_futex;
  return !split->locked.exchange(1, butil::memory_order_acquire);
void FastPthreadMutex::unlock() {
  butil::atomic<unsigned> *whole = (butil::atomic<unsigned> *)&_futex;
  const unsigned prev = whole->exchange(0, butil::memory_order_release);
  // CAUTION: the mutex may be destroyed, check comments before butex_create
  if (prev != BTHREAD_MUTEX_LOCKED) {
    // 如果有挂起等待的线程，执行唤醒
    futex_wake_private(whole, 1);
写了一个简单的测试样例，这里的 FastPthreadMutex 可以比 std::mutex 快一倍。
对于 bthread，如果直接使用互斥锁，会导致整个工作线程阻塞，进而影响同线程下的其他 bthread。这里更好的解决方案是仅挂起 bthread，主动挂起出让 CPU 让其他 bthread 继续执行。bthread 中使用 butex 提供类 futex 的同步原语，下面来看具体实现：
// butex.h
// Create a butex which is a futex-like 32-bit primitive for synchronizing
// bthreads/pthreads.
// Returns a pointer to 32-bit data, NULL on failure.
// NOTE: all butexes are private(not inter-process).
void *butex_create();
// Wake up at most 1 thread waiting on |butex|.
// Returns # of threads woken up.
int butex_wake(void *butex);
// Atomically wait on |butex| if *butex equals |expected_value|, until the
// butex is woken up by butex_wake*, or CLOCK_REALTIME reached |abstime| if
// abstime is not NULL.
// About |abstime|:
//   Different from FUTEX_WAIT, butex_wait uses absolute time.
// Returns 0 on success, -1 otherwise and errno is set.
int butex_wait(void *butex, int expected_value, const timespec *abstime);
// butex.cpp
struct ButexWaiter : public butil::LinkNode<ButexWaiter> {
  // tids of pthreads are 0
  bthread_t tid;
  // Erasing node from middle of LinkedList is thread-unsafe, we need
  // to hold its container's lock.
  butil::atomic<Butex *> container;
struct ButexBthreadWaiter : public ButexWaiter {
  TaskMeta *task_meta;
  TimerThread::TaskId sleep_id;
  WaiterState waiter_state;
  int expected_value;
  Butex *initial_butex;
  TaskControl *control;
// 双向链表
typedef butil::LinkedList<ButexWaiter> ButexWaiterList;
struct BAIDU_CACHELINE_ALIGNMENT Butex {
  Butex() {}
  ~Butex() {}
  butil::atomic<int> value;
  ButexWaiterList waiters;
  internal::FastPthreadMutex waiter_lock;
// 注意 value 的地址与 butex 对象一致，方便后面转换
BAIDU_CASSERT(offsetof(Butex, value) == 0, offsetof_value_must_0);
void *butex_create() {
  Butex *b = butil::get_object<Butex>();
  if (b) {
    return &b->value;
  return NULL;
int butex_wait(void *arg, int expected_value, const timespec *abstime) {
  Butex *b = container_of(static_cast<butil::atomic<int> *>(arg), Butex, value);
  if (b->value.load(butil::memory_order_relaxed) != expected_value) {
    errno = EWOULDBLOCK;
    // Sometimes we may take actions immediately after unmatched butex,
    // this fence makes sure that we see changes before changing butex.
    butil::atomic_thread_fence(butil::memory_order_acquire);
    return -1;
  TaskGroup *g = tls_task_group;
  if (NULL == g || g->is_current_pthread_task()) {
    return butex_wait_from_pthread(g, b, expected_value, abstime);
  // 在栈上创建 waiter 对象
  ButexBthreadWaiter bbw;
  // tid is 0 iff the thread is non-bthread
  bbw.tid = g->current_tid();
  bbw.container.store(NULL, butil::memory_order_relaxed);
  bbw.task_meta = g->current_task();
  bbw.sleep_id = 0;
  bbw.waiter_state = WAITER_STATE_READY;
  bbw.expected_value = expected_value;
  bbw.initial_butex = b;
  bbw.control = g->control();
  if (abstime != NULL) {
    ...  // 先忽略对时间的处理
  // release fence matches with acquire fence in interrupt_and_consume_waiters
  // in task_group.cpp to guarantee visibility of `interrupted'.
  bbw.task_meta->current_waiter.store(&bbw, butil::memory_order_release);
  g->set_remained(wait_for_butex, &bbw);  // 在执行下个 bthread 前执行 wait_for_butex
  TaskGroup::sched(&g);  // 主动 Yield
  // erase_from_butex_and_wakeup (called by TimerThread) is possibly still
  // running and using bbw. The chance is small, just spin until it's done.
  BT_LOOP_WHEN(unsleep_if_necessary(&bbw, get_global_timer_thread()) < 0,
               30 /*nops before sched_yield*/);
  // If current_waiter is NULL, TaskGroup::interrupt() is running and using bbw.
  // Spin until current_waiter != NULL.
  BT_LOOP_WHEN(bbw.task_meta->current_waiter.exchange(
                   NULL, butil::memory_order_acquire) == NULL,
               30 /*nops before sched_yield*/);
  bool is_interrupted = false;
  if (bbw.task_meta->interrupted) {
    // Race with set and may consume multiple interruptions, which are OK.
    bbw.task_meta->interrupted = false;
    is_interrupted = true;
  // If timed out as well as value unmatched, return ETIMEDOUT.
  if (WAITER_STATE_TIMEDOUT == bbw.waiter_state) {
    errno = ETIMEDOUT;
    return -1;
  } else if (WAITER_STATE_UNMATCHEDVALUE == bbw.waiter_state) {
    errno = EWOULDBLOCK;
    return -1;
  } else if (is_interrupted) {
    errno = EINTR;
    return -1;
  return 0;
static void wait_for_butex(void *arg) {
  ButexBthreadWaiter *const bw = static_cast<ButexBthreadWaiter *>(arg);
  Butex *const b = bw->initial_butex;
  // 1: waiter with timeout should have waiter_state == WAITER_STATE_READY
  //    before they're queued, otherwise the waiter is already timedout
  //    and removed by TimerThread, in which case we should stop queueing.
  // Visibility of waiter_state:
  //    [bthread]                         [TimerThread]
  //    waiter_state = TIMED
  //    tt_lock { add task }
  //                                      tt_lock { get task }
  //                                      waiter_lock { waiter_state=TIMEDOUT }
  //    waiter_lock { use waiter_state }
  // tt_lock represents TimerThread::_mutex. Visibility of waiter_state is
  // sequenced by two locks, both threads are guaranteed to see the correct
  // value.
    BAIDU_SCOPED_LOCK(b->waiter_lock);
    // 加锁后，校验 value 和 expected_value 是否相等
    if (b->value.load(butil::memory_order_relaxed) != bw->expected_value) {
      bw->waiter_state = WAITER_STATE_UNMATCHEDVALUE;
    } else if (bw->waiter_state == WAITER_STATE_READY /*1*/ &&
               !bw->task_meta->interrupted) {
      b->waiters.Append(bw);  // 加入 waiters 队列
      bw->container.store(b, butil::memory_order_relaxed);
      return;
  // b->container is NULL which makes erase_from_butex_and_wakeup() and
  // TaskGroup::interrupt() no-op, there's no race between following code and
  // the two functions. The on-stack ButexBthreadWaiter is safe to use and
  // bw->waiter_state will not change again.
  // 如果没有成功 wait，则清理定时器，并切回原协程
  unsleep_if_necessary(bw, get_global_timer_thread());
  tls_task_group->ready_to_run(bw->tid);
  // FIXME: jump back to original thread is buggy.
  // // Value unmatched or waiter is already woken up by TimerThread, jump
  // // back to original bthread.
  // TaskGroup* g = tls_task_group;
  // ReadyToRunArgs args = { g->current_tid(), false };
  // g->set_remained(TaskGroup::ready_to_run_in_worker, &args);
  // // 2: Don't run remained because we're already in a remained function
  // //    otherwise stack may overflow.
  // TaskGroup::sched_to(&g, bw->tid, false/*2*/);
int butex_wake(void *arg) {
  Butex *b = container_of(static_cast<butil::atomic<int> *>(arg), Butex, value);
  ButexWaiter *front = NULL;
    BAIDU_SCOPED_LOCK(b->waiter_lock);
    if (b->waiters.empty()) {
      return 0;
    front = b->waiters.head()->value();
    front->RemoveFromList();  // 从队列中取出等待的 waiter
    front->container.store(NULL, butil::memory_order_relaxed);
  if (front->tid == 0) {
    wakeup_pthread(static_cast<ButexPthreadWaiter *>(front));
    return 1;
  ButexBthreadWaiter *bbw = static_cast<ButexBthreadWaiter *>(front);
  unsleep_if_necessary(bbw, get_global_timer_thread());
  TaskGroup *g = tls_task_group;
  if (g) {
    TaskGroup::exchange(&g, bbw->tid);  // 如果在 bthread 环境中，直接唤醒
  } else {
    // 否认触发重新调度
    bbw->control->choose_one_group()->ready_to_run_remote(bbw->tid);
  return 1;
PS：一些题外话
没错，Tokio 源码分析被“鸽”置了。去年下半年组里项目非常忙，导致周末也不想学习。好在项目赶在年前上线了，年初的答辩也挺顺利，今年就有更多时间自我提升了。开 bRPC 的新坑是因为该项目里有不少先进经验可以应用到自己的工作上，Tokio 会排在 bRPC 之后补上。
          Except where otherwise noted, content on this site is licensed under a CC BY-SA 4.0 license.