C++ 从零实现协程调度框架

// 命名空间 cbricks::pool
namespace cbricks{namespace pool{
// 协程调度池
classWorkerPool: base::Noncopyable{
public:
// 构造函数  threads——使用的线程个数. 默认为 8 个
WorkerPool(size_t threads =8);
// ...
public:


bool submit(task task, bool nonblock = false);

// 工作协程调度任务过程中，可以通过执行次方法主动让出线程的调度权 （仿 golang runtime.Goched 风格）
void sched();
}}

#include 

#include "sync/sem.h"
#include "pool/workerpool.h"

void testWorkerPool();

int main(int argc, char** argv){
// 测试函数
testWorkerPool();
}

void testWorkerPool(){
// 协程调度框架类型别名定义
typedef cbricks::pool::WorkerPool workerPool;
// 信号量类型别名定义
typedef cbricks::sync::Semaphore semaphore;

// 初始化协程调度框架，设置并发的 threads 数量为 8
workerPool::ptr workerPoolPtr(new workerPool(8));

// 初始化一个原子计数器
    std::atomic cnt{0};
// 初始化一个信号量实例
    semaphore sem;

// 投递 10000 个异步任务到协程调度框架中，执行逻辑就是对 cnt 加 1
for(int i =0; i <10000; i++){
// 执行 submit 方法，将任务提交到协程调度框架中
        workerPoolPtr->submit([&cnt,&sem](){
            cnt++;
            sem.notify();
});
}

// 通过信号量等待 10000 个异步任务执行完成
for(int i =0; i <10000; i++){
        sem.wait();
}

// 输出 cnt 结果（预期结果为 10000）
    std::cout << cnt << std::endl;
}

// 保证头文件内容不被重复编译
#pragma once 


// 标准库智能指针相关
#include 
// 标准库函数编程相关
#include 
// 标准库原子量相关
#include 
// 标准库——动态数组，以此作为线程池的载体
#include 


// 线程 thread 实现
#include "../sync/thread.h"
// 协程 coroutine 实现
#include "../sync/coroutine.h"
// 阻塞队列 channel 实现 （一定程度上仿 golang channel 风格）
#include "../sync/channel.h"
// 信号量 semaphore 实现
#include "../sync/sem.h"
// 拷贝禁用工具，用于保证类实例无法被值拷贝和值传递
#include "../base/nocopy.h"

// 命名空间 cbricks::pool
namespace cbricks{namespace pool{
// 协程调度池 继承 Noncopyable 保证禁用值拷贝和值传递功能
classWorkerPool: base::Noncopyable{
public:
// 协程池共享指针类型别名
typedef std::shared_ptr ptr;
// 一笔需要执行的任务
typedef std::function task;
// 一个线程持有的本地任务队列
typedef sync::Channel localq;
// 本地任务队列指针别名
typedef localq::ptr localqPtr;
// 线程指针别名
typedef sync::Thread* threadPtr;
// 一个分配了运行任务的协程
typedef sync::Coroutine worker;
// 协程智能指针别名
typedef sync::Coroutine::ptr workerPtr;
// 读写锁别名
typedef sync::RWLock rwlock;
// 信号量类型别名
typedef sync::Semaphore semaphore;

public:

// 构造函数  threads——使用的线程个数. 默认为 8 个
WorkerPool(size_t threads =8);
// 析构函数  
~WorkerPool();

public:


bool submit(task task, bool nonblock = false);

// 工作协程调度任务过程中，可以通过执行次方法主动让出线程的调度权 （仿 golang runtime.Goched 风格）
void sched();

private:

structthread{
typedef std::shared_ptr ptr;
int index;
        threadPtr thr;
        localqPtr taskq;
        rwlock lock;

thread(int index,threadPtr thr, localqPtr taskq):index(index),thr(thr),taskq(taskq){}
~thread()=default;
};

private:

// work：线程运行主函数，持续不断地从本地任务队列 taskq 或本地协程队列 t_schedq 中获取任务/协程进行调度. 倘若本地任务为空，会尝试从其他线程本地任务队列窃取任务执行
void work();

bool readAndGo(localqPtr taskq, bool nonblock);

void goTask(task cb);

void goWorker(workerPtr worker);


void workStealing();

void workStealing(thread::ptr stealTo, thread::ptr stealFrom);

thread::ptr getStealingTarget();


thread::ptr getThreadByThreadName(std::string threadName);

thread::ptr getThread();

private:

// getThreadNameByIndex：通过线程 index 映射得到线程名称
static const std::string getThreadNameByIndex(int index);
// getThreadIndex：获取当前线程的 index
static const int getThreadIndex();
// getThreadName：获取当前线程的名称
static const std::string getThreadName();


private:

// 基于 vector 实现的线程池，元素类型为 WorkerPool::thread 对应共享指针
    std::vector m_threadPool;

// 基于原子变量标识 workerPool 是否已关闭
    std::atomic m_closed{false};
};

}}

// 标准库队列实现. 依赖队列作为线程本地协程队列的存储载体
#include 

// workerpool 头文件
#include "workerpool.h"
// 本项目定义的断言头文件
#include "../trace/assert.h"

// namespace cbricks::pool
namespace cbricks{namespace pool{


static std::atomic s_taskId{0};


staticthread_local std::queue t_schedq;

// ...

}}

// namespace cbricks::pool
namespace cbricks{namespace pool{
// ...

WorkerPool::WorkerPool(size_t threads){
CBRICKS_ASSERT(threads >0,"worker pool init with nonpositive threads num");

// 为线程池预留好对应的容量
this->m_threadPool.reserve(threads);


std::vector sems(threads);
// 另一个信号量，用于保证所有 thread 调度函数都正常启动后，当前构造函数才能退出，避免 sems 被提前析构
    semaphore waitGroup;

// 初始化好对应数量的 thread 实例并添加进入 m_threadPool
for(int i =0; i < threads; i++){
// 根据 index 映射得到 thread 名称
        std::string threadName =WorkerPool::getThreadNameByIndex(i);
// 将 thread 实例添加进入 m_threadPool
this->m_threadPool.push_back(thread::ptr(
// thread 实例初始化
newthread(
            i,
// 
new sync::Thread([this,&sems,&waitGroup](){

                sems[getThreadIndex()].wait();

                waitGroup.notify();
// 异步启动的 thread，最终运行的调度函数是 workerpool::work
this->work();

},
// 注入 thread 名称，与 index 有映射关系
            threadName),
// 分配给 thread 的本地任务队列
localqPtr(new localq))));

        sems[i].notify();
}


for(int i =0; i < threads; i++){
        waitGroup.wait();
}
}

// 析构函数
WorkerPool::~WorkerPool(){
// 将 workpool 的关闭标识置为 true，后续运行中的线程感知到此标识后会主动退出
this->m_closed.store(true);
// 等待所有线程都退出后，再退出 workpool 的析构函数
for(int i =0; i m_threadPool.size(); i++){
// 关闭各 thread 的本地任务队列
this->m_threadPool[i]->taskq->close();
// 等待各 thread 退出
this->m_threadPool[i]->thr->join();
}
}

// ...

}}

// namespace cbricks::pool
namespace cbricks{namespace pool{
// ...

bool WorkerPool::submit(task task, bool nonblock){
// 若 workerpool 已关闭，则提交失败
if(this->m_closed.load()){
returnfalse;
}

// 基于任务 id 对 m_threadPool 长度取模，将任务映射到指定 thread
int targetThreadId =(s_taskId++)%(this->m_threadPool.size());
    thread::ptr targetThr =this->m_threadPool[targetThreadId];

// 针对目标 thread 加读锁，这是为了防止和目标 thread 的 workstealing 操作并发最终因任务队列 taskq 容量溢出而导致死锁
rwlock::readLockGuard guard(targetThr->lock);

// 往对应 thread 的本地任务队列中写入任务
return targetThr->taskq->write(task, nonblock);
}

// ...

}}

// namespace cbricks::pool
namespace cbricks{ namespace pool{
// ...
// sched：让渡函数. 在任务执行过程中，可以通过该方法主动让出线程的执行权，则此时任务所属的协程会被添加到 thread 的本地协程队列 t_schedq 中，等待后续再被调度执行
void WorkerPool::sched(){
    worker::GetThis()->sched();
}

// ...

}}

// namespace cbricks::pool
namespace cbricks{namespace pool{
// ...

void WorkerPool::work(){
// 获取到当前 thread 对应的本地任务队列 taskq
    localqPtr taskq =this->getThread()->taskq;

// main loop
while(true){
// 如果 workerpool 已关闭 则主动退出
if(this->m_closed.load()){
return;
}



// 标识本地任务队列 taskq 是否为空
bool taskqEmpty =false;
// 至多调度 10 次本地任务队列 taskq
for(int i =0; i <10; i++){
// 以【非阻塞模式】从 taskq 获取任务并为之分配协程实例和调度执行
if(!this->readAndGo(taskq,false)){
// 如果 taskq 为空，将 taskqEmpty 置为 true 并直接退出循环
                taskqEmpty =true;
break;
}
}

// 尝试从线程本地的协程队列 t_schedq 中获取协程并进行调度
if(!t_schedq.empty()){
// 从协程队列中取出头部的协程实例
            workerPtr worker = t_schedq.front();
            t_schedq.pop();
// 进行协程调度
this->goWorker(worker);
// 处理完成后直接进入下一轮循环
continue;
}

// 如果未发现 taskq 为空，则无需 workstealing，直接进入下一轮循环
if(!taskqEmpty){
continue;
}


this->workStealing();


this->readAndGo(taskq,true);
}
}
// ...

}}

// namespace cbricks::pool
namespace cbricks{namespace pool{
// ...

// 将一个任务包装成协程并进行调度. 如果没有一次性调度完成，则将协程实例添加到线程本地的协程队列 t_schedq
bool WorkerPool::readAndGo(cbricks::pool::WorkerPool::localqPtr taskq, bool nonblock){
// 任务容器
    task cb;
// 从 taskq 中获取任务
if(!taskq->read(cb,nonblock)){
returnfalse;
}

// 对任务进行调度
this->goTask(cb);
returntrue;
}


void WorkerPool::goTask(task cb){
// 初始化协程实例
    workerPtr _worker(newworker(cb));
// 调度协程
this->goWorker(_worker);
}


void WorkerPool::goWorker(workerPtr worker){
// 调度协程，此时线程的执行权会切换进入到协程对应的方法栈中
    worker->go();
// 走到此处意味着线程执行权已经从协程切换回来
// 如果此时协程并非已完成的状态，则需要将其添加到线程本地的协程队列 schedq 中，等待后续继续调度
if(worker->getState()!= sync::Coroutine::Dead){
        t_schedq.push(worker);
}
}
// ...

}}

// namespace cbricks::pool
namespace cbricks{namespace pool{
// ...
// 从某个 thread 中窃取一半任务给到本 thread 的 taskq
void WorkerPool::workStealing(){
// 选择一个窃取的目标 thread 
    thread::ptr stealFrom =this->getStealingTarget();
if(!stealFrom){
return;
}
// 从目标 thread 中窃取半数任务添加到本 thread taskq 中
this->workStealing(this->getThread(),stealFrom);
}

// 从 thread:stealFrom 中窃取半数任务给到 thread:stealTo
void WorkerPool::workStealing(thread::ptr stealTo, thread::ptr stealFrom){
// 确定窃取任务数量：目标本地任务队列 taskq 中任务总数的一半
int stealNum = stealFrom->taskq->size()/2;
if(stealNum <=0){
return;
}

// 针对 thread:stealTo 加写锁，防止因 workstealing 和 submit 行为并发，导致线程因 taskq 容量溢出而发生死锁
rwlock::lockGuard guard(stealTo->lock);
// 检查此时 stealTo 中的 taskq 如果剩余容量已不足以承载拟窃取的任务量，则直接退出
if(stealTo->taskq->size()+ stealNum > stealTo->taskq->cap()){
return;
}

// 创建任务容器，以非阻塞模式从 stealFrom 的 taskq 中窃取指定数量的任务
std::vector containers(stealNum);
if(!stealFrom->taskq->readN(containers,true)){
return;
}

// 将窃取到的任务添加到 stealTo 的 taskq 中
    stealTo->taskq->writeN(containers,false);
}

// 随机选择本 thread 外的一个 thread 作为窃取的目标
WorkerPool::thread::ptr WorkerPool::getStealingTarget(){
// 如果线程池长度不足 2，直接返回
if(this->m_threadPool.size()<2){
returnnullptr;
}

// 通过随机数，获取本 thread 之外的一个目标 thread index
int threadIndex =WorkerPool::getThreadIndex();
int targetIndex =rand()%this->m_threadPool.size();
while( targetIndex == threadIndex){
        targetIndex =rand()%this->m_threadPool.size();
}

// 返回目标 thread
returnthis->m_threadPool[targetIndex];
}
// ...

}}