解谜 Dart VM中的线程池：并发编程艺术的详细分析

准备

• 在这个地址下载源码
• VSCode 安装好「C/C++ for Visual Studio Code」 插件
• 了解 Dart Isolate 的基础使用

打开 vm/thread_pool.h 与 vm/thread_pool.cc 两个文件。

开始

众所周知如需要在 Dart 开发中要使用「多线程」的能力那肯定离不开 Isolate ，与传统多线程的概念不同 Dart 语言下的 Isolate 之间进行数据同步与通信时需要借助「消息机制」。Isolate 在设计理念上淡化了线程的概念却更接近进程，在使用上与进程间通信类似但又比进程间通信更简单。实际上 Isolate 在 Runtime 底层实现依然还是使用「多线程」的能力，它通过包装与抽象多线程让 Dart 代码可以直接使用 Isolate 来替代多线程，其中「线程池（以下用 ThreadPool 替代）」是这层抽象的基础。根据官方的定义：

在使用 Isolate 时能保证任意线程不会同时进入多个 Isolate 且同一个 Isolate 不会被多个线程同时运行

Dart Runtime 中的 ThreadPool 不仅用来承载线程(Worker)还承载了任务(Task)，Worker 代表「消费者」来消费当前的 Task（引用 MessageHandler），这两个类型是 ThreadPool 的内置类型本文后面会详细介绍。

---

看源码之前切换到上帝视角来看 ThreadPool 的基础作用：

• C++ Isolate 类型与 Dart 中的 Isolate 类型对应

• 每个 C++ Isolate 关联了一个 MessageHandler 对象，MessageHandler 保存了当前 Isolate 的所有 Message

• Message 包含 Dart 代码传进来的基础数据与 port_id，Message 可来自于当前 Isolate 或另一个 Isolate，Message 被消费时会将数据再回调给 Dart 代码（另一个 Dart Isolate 回调）

• Isolate 负责触发 ThreadPool 对 Task （Task 会引用 MessageHandler）的创建流程，创建好的 Task 会保存到 ThreadPool 自身的队列中等待自身 Worker 来消费

• 每个 Worker 会保证消费掉一个 Task 内的所有 Message 后再进入另一个 Task

• 多个 Isolate 共享同一个 ThreadPool

锁

本小节为 C++ 基础，C++ 大佬可略过

在源码中总是能看到类似下面的代码，函数作用域内又定义了一个内部作用域，从语法层面上看这个内部作用域似乎没有什么意义，有没有它也不影响代码的执行逻辑。那它有什么作用呢？不卖关子，这里其实是 C++ 多线程下「锁」的巧妙用法。

bool ThreadPool::RunImpl(std::unique_ptr<Task> task) {
  Worker* new_worker = nullptr;
  { // 声明内部作用域
    MonitorLocker ml(&pool_monitor_); // 定义变量
    if (shutting_down_) {
      return false;
    }
    new_worker = ScheduleTaskLocked(&ml, std::move(task));
  } //  作用域结束
  if (new_worker != nullptr) {
    new_worker->StartThread();
  }
  return true;
}

一般情况下我们在多线程下使用锁时都是直接在临界资源访问前后直接调用加解锁的代码，伪代码示例如下：

int globalCount = 1;

void threadEntry() {
    // 直接加锁
    thread_lock();
    globalCount++;
    // 直接解锁
    thread_unlock();
}

这种写法很直观，访问临界资源前加锁占用当前资源，防止其它线程再访问当前资源，访问结束后再解锁释放当前资源。注意这里的加解锁一定得是一个对称操作，有加锁的代码就一定要对称出现解锁代码否则问题很严重。但如果要保护的临界资源较长（加解锁保护的代码较长）或者相同的加解锁代码太多又或者有提前 return 边界条件，如何防止解锁代码漏写就比较麻烦了。

C++ 类定义时有构造函数与析构函数，当 new 出一个对象时构造函数会被调用，delete 释放对象时析构函数会被调用，所以当类的临时变量超过作用域它的析构函数会被调用。正是利用 C++ 的这个特点可以通过添加作用域的方式，来保护临界资源。来一个简单示例：

#include <iostream>
#include <mutex>

// 简化版的互斥锁类
class MyMutex {
public:
    MyMutex(std::mutex& mtx) : mutexRef(mtx) {
        mutexRef.lock();
    }

    ~MyMutex() {
        mutexRef.unlock();
    }

private:
    std::mutex& mutexRef;
};

int globalCount = 1;

int main() {
    // 一个互斥锁
    std::mutex _mutex;

    // 在作用域中创建 MyMutex 对象，构造函数加锁，离开作用域时析构函数解锁
    {
        MyMutex myMutex(_mutex);
        globalCount++;
    }
    // MyMutex 对象离开作用域后，互斥锁已被解锁

    return 0;
}

回到 ThreadPool 中的锁类型，ThreadPool 中有涉及到了 Runtime 中两种锁类型 Monitor 与 MonitorLocker。从 Monitor 的构造函数与析构函数可以看出，Monitor 正是利用了析构特性实现了超出作用域自动解锁的能力，是对条件锁（也称条件变量）的一层包装。同时 Monitor 也用来屏蔽不同 OS 的锁实现，下面的代码正是在 MacOS/iOS 实现（os_thread_macos.cc）。

条件变量扩展阅读：pthread 条件变量

// 构造函数
Monitor::Monitor() {
  pthread_mutexattr_t attr;
  int result = pthread_mutexattr_init(&attr);
  result = pthread_mutex_init(data_.mutex(), &attr);

  result = pthread_mutexattr_destroy(&attr);

  result = pthread_cond_init(data_.cond(), nullptr);
}
// 析构函数
Monitor::~Monitor() {
  int result = pthread_mutex_destroy(data_.mutex());
  result = pthread_cond_destroy(data_.cond());
}

而 MonitorLocker 则更为直接，从类定义来看它仅仅是对 Monitor 进行了二次包装。知识点：在 Dart Runtime 中并不直接使用 Monitor 进行锁操作，而是使用 Monitor 的封装类 MonitorLocker 进行锁操作。

class MonitorLocker {
 public:
  explicit MonitorLocker(Monitor* monitor) : monitor_(monitor) {
    monitor_->Enter();
  }

  virtual ~MonitorLocker() { monitor_->Exit(); }

  Monitor::WaitResult Wait(int64_t millis = Monitor::kNoTimeout) {
    return monitor_->Wait(millis);
  }

  void Notify() { monitor_->Notify(); }

  void NotifyAll() { monitor_->NotifyAll(); }

 private:
  // 对不同操作系统条件变量的封装
  Monitor* const monitor_;
}

上面对 ThreadPool 中涉及到的锁进行了介绍，相信看到类似的代码后不会再感到困惑。

两个类型

本文开头提到了线程池模型中的生产者（Task）与消费者（Worker），整个 ThreadPool 都是围绕这两个类型的队列进行逻辑处理，本小节将重点介绍这两个类型。

Task

Task 被定义在 ThreadPool 中，是 ThreadPool 的内置类型，同时通过搜索整个 Runtime 仓库可知 Task 也是一个基类，它派生出了不同的子类，每个子类都代表某种任务。

基类 Task 内只有一个函数 Run, 同时它的构造函数被 protected 修饰，说明它及它的子类都只能在 ThreadPool 及 ThreadPool 子类中实例化。实际上在 Task 类定义的下方就有 Task 的实例化逻辑，只不过它是用模板（C++ 中的模板相当于泛型）实现。

class ThreadPool {
 public:
  // 基类 Task 的定义，继承自 IntrusiveDListEntry 说明它的子类可以进行列队操作
  class Task : public IntrusiveDListEntry<Task> {
   protected:
    Task() {}

   public:
    virtual ~Task() {}
    // 虚函数，由子负责类实现
    virtual void Run() = 0;
  };

  // 模板(泛型)函数，负责实例化 Task 子类
  template <typename T, typename... Args>
  bool Run(Args&&... args) {
    return RunImpl(std::unique_ptr<Task>(new T(std::forward<Args>(args)...)));
  }

  private:
  using TaskList = IntrusiveDList<Task>;
  TaskList tasks_;
}

Task 子类实例化后便调用了 ThreadPool::RunImpl 方法，MessageHandler 正是由此触发了 Task 的创建流程。Task 子类众多，这里我们暂时只关注 MessageHandler 相关的部分，也就是 MessageHandlerTask，可以看看它的实现。

Worker

Worker 字面意思有「工具人」的味道，从类定义来看它持有当前线程（os_thread_ 成员变量），且有一个 Main 静态函数，说明这个 Main 是线程的入口函数。

class ThreadPool {
  public:
  
  private:
    class Worker : public IntrusiveDListEntry<Worker> {
     public:
        explicit Worker(ThreadPool* pool);

        void StartThread();

     private:
        friend class ThreadPool;

        // 线程创建后的入口函数
        static void Main(uword args);

        ThreadPool* pool_;
        ThreadJoinId join_id_;
        // 持有当前函数
        OSThread* os_thread_ = nullptr;
    };
    using WorkerList = IntrusiveDList<Worker>;
    // 不同状态的 Worker 队列
    WorkerList running_workers_;
    WorkerList idle_workers_;
    WorkerList dead_workers_;
}

通过 Worker 的实现可知， Main 函数在 ThreadPool::Worker::StartThread 方法中被传入操作系统线程开始执行。

void ThreadPool::Worker::StartThread() {
  int result = OSThread::Start("DartWorker", &Worker::Main,
                               reinterpret_cast<uword>(this));
  // 省略 result 判断
}

依然以 MacOS/iOS 系统平台代码为例，线程创建的代码如下所示（对源码略有简化）。从源码来看，线程的创建并没有任何特殊处理。

int OSThread::Start(const char* name,
                    ThreadStartFunction function,
                    uword parameter) {
  // ... 省略其它代码
  //  ThreadPool::Worker::Main 函数与参数保存在 data 对象中
  ThreadStartData* data = new ThreadStartData(name, function, parameter);
  pthread_t tid;
  // 创建线程
  result = pthread_create(&tid, &attr, ThreadStart, data);
  // ... 省略其它代码

  return 0;
}

// Worker 优先级全局常量，默认值为：kMinInt
int FLAG_worker_thread_priority = Flags::Register_int(&FLAG_worker_thread_priority, "worker_thread_priority", kMinInt, "The thread priority the VM should use for new worker threads.");

static void* ThreadStart(void* data_ptr) {
  // 如果优先级不为 kMinInt 时则设置线程优选级
  if (FLAG_worker_thread_priority != kMinInt) {
    // 这里的 FLAG_worker_thread_priority 全局变量默认值为 kMinInt
    // 所以优选级不会永远不会被设置
    const pthread_t thread = pthread_self();
    int policy = SCHED_FIFO;
    struct sched_param schedule;
    pthread_getschedparam(thread, &policy, &schedule);
    schedule.sched_priority = FLAG_worker_thread_priority;
    pthread_setschedparam(thread, policy, &schedule);
  }

  // 取出 ThreadPool::Worker 内的静态 Main 函数与参数
  OSThread::ThreadStartFunction function = data->function();
  uword parameter = data->parameter();

  // 调用 ThreadPool::Worker::Main 函数
  function(parameter);

  return nullptr;
}

所有线程创建后均使用默认优先级（pthread_create 创建的线程默认优先级为 0），说明 Dart 还没有针对 Apple M 系列的芯片做针对性的性能优化，这可能会使 Dart 在计算密集型的场景处于不利位置。因为根据少数派这篇文章所述，Apple M 系列芯片使用大小核架构（大核：性能核心简称 P 核，小核：效能核心简称 E 核），优先级低的线程只会分配到 E 核心上，只有当 E 核心分配满了才会分配 P 核心。

扩展阅读：M1 CPU 那么多的核，macOS 是怎样管理的？。虽然这篇文章所述的优先级均是 QoS （NSOperation）优先级，但根据苹果的 Prioritize Work with Quality of Service Classes 文档与 XNU 源码 可知 QoS 与 pthread 优先级存在映身关系。

如果你的 Dart 应用（包含 Flutter 桌面 App，甚至 Dart 编译前端）对性能有更高要求，理论上可以尝试更改 FLAG_worker_thread_priority 的默认值（如：63）然后重新编译 Dart SDK，让 MacOS 操作系统强制优先分配 P 核心来提升性能。（由于我这边没有 M 芯片 Mac 无法做验证，如果你做了相关验证请一定要让我知道 🥳）

核心

上面介绍完了 ThreadPool 的基础知识，如果有 C/C++ 基础知识其实就能完全看懂这部分代码了，这里只对对一些核心细节进行详细说明。

入口

如前所述线程池是生产者与消费者模型，生产者通过 ThreadPool::Run 函数触发 Task 派生子类型的创建，然后会调用到 ThreadPool::RunImpl 函数。这里 ThreadPool::RunImpl 便是线程池的真正「入口」。

  // ThreadPool 入口
  template <typename T, typename... Args>
  bool Run(Args&&... args) {
    return RunImpl(std::unique_ptr<Task>(new T(std::forward<Args>(args)...)));
  }

  bool ThreadPool::RunImpl(std::unique_ptr<Task> task) {
  Worker* new_worker = nullptr;
  {
    MonitorLocker ml(&pool_monitor_);
    if (shutting_down_) {
      return false;
    }
    // Task 与（潜在的）Worker 创建
    new_worker = ScheduleTaskLocked(&ml, std::move(task));
  }

  if (new_worker != nullptr) {
    // 如果创建了空闲 Worker 就启动它
    new_worker->StartThread();
  }
  return true;
}
  
  // ThreadPool 入口使用方式以 MessageHandler 中对线程池的使用做为示例
  bool MessageHandler::Run(ThreadPool* pool,
                         StartCallback start_callback,
                         EndCallback end_callback,
                         CallbackData data) {
  // ...省略
  // message_handle.cc 中对线程池入口的调用，this 参数是当前 MessageHandler 对象
  // MessageHandlerTask 泛型指定生成的 Task 类型为 MessageHandlerTask
  pool_->Run<MessageHandlerTask>(this);
  // ...省略
}

启动

ThreadPool::RunImpl 方法内部通过调用 ThreadPool::ScheduleTaskLocked 方法来负责 Worker 与 Task 的创建。Task 直接创建且只有一个显式的状态: Pending（通过 pending_tasks_ 变量维护），一个 Task 不是在 Pending 状态就是在运行状态。而 Worker 创建的过程中会判断当前存活的线程数量（也就是 Worker 队列数量）是否超过最大限制（max_pool_size_），如果超过限制则尝试唤醒空闲线程（Worker）。

ThreadPool::Worker* ThreadPool::ScheduleTaskLocked(MonitorLocker* ml,
                                                  std::unique_ptr<Task> task) {
 // 将 task 添加到队列并记录待运行 task 数量
 tasks_.Append(task.release());
 pending_tasks_++;

 // 如果空闲线程大于等于 pending 状态 task 数量，则优先唤醒空闲线程
 if (count_idle_ >= pending_tasks_) {
   ml->Notify();
   return nullptr;
 }

 // 正在运行与空闲的线程数超过最大线程数限制，则优先唤醒空闲线程
 if (max_pool_size_ > 0 && (count_idle_ + count_running_) >= max_pool_size_) {
   if (!idle_workers_.IsEmpty()) {
     ml->Notify();
   }
   return nullptr;
 }

 // 否则直接创建空闲的 Worker 并返回
 auto new_worker = new Worker(this);
 idle_workers_.Append(new_worker);
 count_idle_++;
 return new_worker;
}

正如你所想，Worker 除了有空闲状态（Idle）、还有运行状态(Running)、死亡状态(Dead)，三者之间的转换关系如下：

ThreadPool 中与之对应的状态变化方法分别是：

方法	作用
ThreadPool::IdleToRunningLocked	空闲转运行
ThreadPool::RunningToIdleLocked	运行转空闲
ThreadPool::IdleToDeadLocked	空闲转死亡

状态变化操作仅仅只是将 Worker 在不同的队列中移动并改变状态计数，以 ThreadPool::IdleToRunningLocked 方法为例：

// Worker 从空闲转移到运行状态
void ThreadPool::IdleToRunningLocked(Worker* worker) {
  // 从空闲队列中移除
  idle_workers_.Remove(worker);
  // 添加到运行状态队列
  running_workers_.Append(worker);
  // 维护状态变量
  count_idle_--;
  count_running_++;
}

Worker 创建出来后默认是 Idle 状态，ThreadPool::ScheduleTaskLocked 方法内被创建出来后立即执行了 ThreadPool::StartThread 方法来启动 Worker(即开启新线程调用 ThreadPool::Worker::Main 方法)，启动后其状态会变成 Running 状态。启动的详细过程如上节「Worker 类型」介绍所述。

// 在新线程内运行 Worker::Main 方法，this 参数为当前 ThreadPool 对象，最终会传入 Main 方法内
void ThreadPool::Worker::StartThread() {
  int result = OSThread::Start("DartWorker", &Worker::Main,
                               reinterpret_cast<uword>(this));
}

这里需要关注的是 ThreadPool::Worker::Main 函数内部实现，并且 ThreadPool::Worker::Main 函数的执行是在新的线程，新的线程，新的线程内，已与「入口」函数所在线程不同。它的核心是通过 ThreadPool::WorkerLoop 在新线程内消费当前 ThreadPool 内保存的 Pending 状态的 Task 队列。

// 源码略有简化
void ThreadPool::Worker::Main(uword args) {
  // 获取当前 OSThread 对象（OSThread 是 Runtime 对平台线程的抽象，保存在当前线程的 TLS）
  OSThread* os_thread = OSThread::Current();
  // 将传过来的参数转换为 ThreadPool 对象
  Worker* worker = reinterpret_cast<Worker*>(args);
  ThreadPool* pool = worker->pool_;
  // 将 Worker 与 OSThread 相互关联
  os_thread->owning_thread_pool_worker_ = worker;
  worker->os_thread_ = os_thread;
  // 保存 join_id 用于资源清理
  worker->join_id_ = OSThread::GetCurrentThreadJoinId(os_thread);
  
  // 开始消费 Task 循环
  pool->WorkerLoop(worker);

  // 退出循环清理绑定关系
  worker->os_thread_ = nullptr;
  os_thread->owning_thread_pool_worker_ = nullptr;
}

整个 ThreadPool 内核心中的核心便是 ThreadPool::WorkerLoop，它负责来消费处于 Pending 状态的 Task。整个方法主体只有一个 while 循环，注意循环作用域开头的 MonitorLocker ，它是一个条件变量互斥锁（详情参考上面 锁 小节的介绍），作用域内加锁，离开作用域后解锁。MonitorLocker 变量的存在保证了多个线程不会进入同时进入同一个 Task，即多个线程不会同时进入同一个 Isolate。

下面这段代码看起来长，但相信我它并不复杂请一定耐心看完。

void ThreadPool::WorkerLoop(Worker* worker) {
  // 在当前线程中收集 dead 状态的 worker
  WorkerList dead_workers_to_join;

  while (true) {
    // 声明线程锁并加锁，该锁在变量离开作用域后解锁
    MonitorLocker ml(&pool_monitor_);

    // Pending 任务队列不为空，进入内部循环
    if (!tasks_.IsEmpty()) {
      // 将当前 worker 的状态转移至 Running
      IdleToRunningLocked(worker);
      // 消费 task_ 直到列表为空
      while (!tasks_.IsEmpty()) {
        // 将 task_ 从队列中取出
        std::unique_ptr<Task> task(tasks_.RemoveFirst());
        // 减少 Pending task 数量
        pending_tasks_--;
        // 对上面声名的锁临时解锁，允许其它线程可以继续消费其它 task_
        MonitorLeaveScope mls(&ml);
        // 运行 task_，消费内部的 Message
        task->Run();
        ASSERT(Isolate::Current() == nullptr);
        // task 指针置空
        task.reset();
      }
      // task_ 队列为空后将当前 worker 的状态转移至 Idle
      RunningToIdleLocked(worker);
    }

    // 所有线程都空闲时整个线程池进入空闲状态
    if (running_workers_.IsEmpty()) {
      OnEnterIdleLocked(&ml);
      if (!tasks_.IsEmpty()) {
        continue;
      }
    }

    // 如果线程池关闭则将当前 worker 转移到 Dead 状态
    if (shutting_down_) {
      // 收集之前其它线程已经 Dead 的 worker
      ObtainDeadWorkersLocked(&dead_workers_to_join);
      IdleToDeadLocked(worker);
      break;
    }

    // 下面的代码核心逻辑是将当前线程挂起，在挂起的时间内等待被唤醒
    // 由于挂起前 worker 已进入 Idle 状态，如果等待超时，则当前线程进会进入 Dead 状态
    const int64_t idle_start = OS::GetCurrentMonotonicMicros();
    bool done = false;
    while (!done) {
      // 线程默认挂起时长为 5 秒
      const auto result = ml.WaitMicros(ComputeTimeout(idle_start));

      if (!tasks_.IsEmpty()) break;

      if (shutting_down_ || result == Monitor::kTimedOut) {
        done = true;
        break;
      }
    }
    // 如果超时或关闭
    if (done) {
      // 收集之前其它线程已经 Dead 的 worker
      ObtainDeadWorkersLocked(&dead_workers_to_join);
      // 将当前 worker 置于 Dead 状态
      IdleToDeadLocked(worker);
      break;
    }
  }
  // 如收集到的 Dead 状态 Worker 不为空，则等待它们结束后再结束当前线程
  JoinDeadWorkersLocked(&dead_workers_to_join);
}

注意 while (!tasks_.IsEmpty()) 循环的存在，它表明当前线程会消费 tasks_ 队列中所有 Task，同时通过 MonitorLeaveScope 将线程锁临时解锁，也给其它线程来消费 tasks_ 队列的机会。

所谓消费 Task 就是执行其 Run 方法，Run 方法处理完所有 Message 才结束

在消费 Task 前后会改变 Worker 的状态（Running/Idle），Worker 进入 Idle 状态后马上会被挂起，直到超时或被唤醒。超时后会进入 Dead 状态，唤醒则变成 Running 状态然后继续消费 Task 。正是由这套状态机制的保证了一个线程（Worker）不会同时进入两个 Isolate。

One More Thing

还有几个值得注意的细节是整个 Runtime 内并不是只有一个线程池实例，实际上 ThreadPool 还有一个派生类型 MutatorThreadPool，所以整个 Dart Runtime 只有两个线程池实例。MutatorThreadPool 类型的实例用来运行 Dart 代码，而 ThreadPool 类型的线程池用来做内存的 GC 操作或编译等辅助工作。并且在线程池数量限制上也有所不同，MutatorThreadPool 类型线程池默认最大线程数量是 8，而 ThreadPool 类型对线程数量没有限制。

另外线程挂起的默认超时时长是 5 秒，只要线程在 5 秒内被唤醒它仍然会苟活于世。至于为什么是 5 秒不是 10 秒，我也不知道，如果你知道勿必告诉我 🥰

ThreadPool 相关的知识点不复杂，理解起来不会有太多阻碍。核心思想仍然是传统的生产者与消费者模式，再加上针对不同平台的线程抽象结合生命周期定义组成了整程线池的核心逻辑。