// 命名空間 cbricks::pool
namespace cbricks{namespace pool{
// 協(xié)程調(diào)度池
classWorkerPool: base::Noncopyable{
public:
// 構(gòu)造函數(shù)  threads——使用的線程個數(shù). 默認為 8 個
WorkerPool(size_t threads =8);
// ...
public:
/**
        公開方法
    */
/**
        * submit: 向協(xié)程調(diào)度池中提交一個任務(wù) （仿 golang 協(xié)程池 ants 風格）
            - task 會被隨機分配到線程手中，保證負載均衡
            - task 被一個線程取到之后，會創(chuàng)建對應(yīng)協(xié)程實例，為其分配本地棧，此時該任務(wù)和協(xié)程就固定屬于一個線程了
        * param：task——提交的任務(wù)  nonblock——是否為阻塞模式
            - 阻塞模式：線程本地任務(wù)隊列滿時阻塞等待 
            - 非阻塞模式：線程本地隊列滿時直接返回 false
        * response：true——提交成功 false——提交失敗
    */
bool submit(task task, bool nonblock = false);

// 工作協(xié)程調(diào)度任務(wù)過程中，可以通過執(zhí)行次方法主動讓出線程的調(diào)度權(quán) （仿 golang runtime.Goched 風格）
void sched();
}}

#include <iostream>

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

void testWorkerPool();

int main(int argc, char** argv){
// 測試函數(shù)
testWorkerPool();
}

void testWorkerPool(){
// 協(xié)程調(diào)度框架類型別名定義
typedef cbricks::pool::WorkerPool workerPool;
// 信號量類型別名定義
typedef cbricks::sync::Semaphore semaphore;

// 初始化協(xié)程調(diào)度框架，設(shè)置并發(fā)的 threads 數(shù)量為 8
workerPool::ptr workerPoolPtr(new workerPool(8));

// 初始化一個原子計數(shù)器
    std::atomic<int> cnt{0};
// 初始化一個信號量實例
    semaphore sem;

// 投遞 10000 個異步任務(wù)到協(xié)程調(diào)度框架中，執(zhí)行邏輯就是對 cnt 加 1
for(int i =0; i <10000; i++){
// 執(zhí)行 submit 方法，將任務(wù)提交到協(xié)程調(diào)度框架中
        workerPoolPtr->submit([&cnt,&sem](){
            cnt++;
            sem.notify();
});
}

// 通過信號量等待 10000 個異步任務(wù)執(zhí)行完成
for(int i =0; i <10000; i++){
        sem.wait();
}

// 輸出 cnt 結(jié)果（預期結(jié)果為 10000）
    std::cout << cnt << std::endl;
}

// 保證頭文件內(nèi)容不被重復編譯
#pragma once 

/**
 依賴的標準庫頭文件   
*/
// 標準庫智能指針相關(guān)
#include <memory>
// 標準庫函數(shù)編程相關(guān)
#include <functional>
// 標準庫原子量相關(guān)
#include <atomic>
// 標準庫——動態(tài)數(shù)組，以此作為線程池的載體
#include <vector>

/**
    依賴的項目內(nèi)部頭文件
*/
// 線程 thread 實現(xiàn)
#include "../sync/thread.h"
// 協(xié)程 coroutine 實現(xiàn)
#include "../sync/coroutine.h"
// 阻塞隊列 channel 實現(xiàn) （一定程度上仿 golang channel 風格）
#include "../sync/channel.h"
// 信號量 semaphore 實現(xiàn)
#include "../sync/sem.h"
// 拷貝禁用工具，用于保證類實例無法被值拷貝和值傳遞
#include "../base/nocopy.h"

// 命名空間 cbricks::pool
namespace cbricks{namespace pool{
// 協(xié)程調(diào)度池 繼承 Noncopyable 保證禁用值拷貝和值傳遞功能
classWorkerPool: base::Noncopyable{
public:
// 協(xié)程池共享指針類型別名
typedef std::shared_ptr<WorkerPool> ptr;
// 一筆需要執(zhí)行的任務(wù)
typedef std::function<void()> task;
// 一個線程持有的本地任務(wù)隊列
typedef sync::Channel<task> localq;
// 本地任務(wù)隊列指針別名
typedef localq::ptr localqPtr;
// 線程指針別名
typedef sync::Thread* threadPtr;
// 一個分配了運行任務(wù)的協(xié)程
typedef sync::Coroutine worker;
// 協(xié)程智能指針別名
typedef sync::Coroutine::ptr workerPtr;
// 讀寫鎖別名
typedef sync::RWLock rwlock;
// 信號量類型別名
typedef sync::Semaphore semaphore;

public:
/**
      構(gòu)造/析構(gòu)函數(shù)
    */
// 構(gòu)造函數(shù)  threads——使用的線程個數(shù). 默認為 8 個
WorkerPool(size_t threads =8);
// 析構(gòu)函數(shù)  
~WorkerPool();

public:
/**
        公開方法
    */
/**
        * submit: 向協(xié)程調(diào)度池中提交一個任務(wù) （仿 golang 協(xié)程池 ants 風格）
            - task 會被隨機分配到線程手中，保證負載均衡
            - task 被一個線程取到之后，會創(chuàng)建對應(yīng)協(xié)程實例，為其分配本地棧，此時該任務(wù)和協(xié)程就固定屬于一個線程了
        * param：task——提交的任務(wù)  nonblock——是否為阻塞模式
            - 阻塞模式：線程本地任務(wù)隊列滿時阻塞等待 
            - 非阻塞模式：線程本地隊列滿時直接返回 false
        * response：true——提交成功 false——提交失敗
    */
bool submit(task task, bool nonblock = false);

// 工作協(xié)程調(diào)度任務(wù)過程中，可以通過執(zhí)行次方法主動讓出線程的調(diào)度權(quán) （仿 golang runtime.Goched 風格）
void sched();

private:
/**
     * thread——workerPool 中封裝的線程類
     * - index：線程在線程池中的 index
     * - thr：真正的線程實例，類型為 sync/thread.h 中的 Thread
     * - taskq：線程的本地任務(wù)隊列，其中數(shù)據(jù)類型為閉包函數(shù) void()
     * - lock：一把線程實例粒度的讀寫鎖. 用于隔離 submit 操作和 workstealing 操作，避免因任務(wù)隊列阻塞導致死鎖
     */
structthread{
typedef std::shared_ptr<thread> ptr;
int index;
        threadPtr thr;
        localqPtr taskq;
        rwlock lock;
/**
         *  構(gòu)造函數(shù)
         * param: index: 線程在線程池中的 index; thr: 底層真正的線程實例; taskq：線程持有的本地任務(wù)隊列
        */
thread(int index,threadPtr thr, localqPtr taskq):index(index),thr(thr),taskq(taskq){}
~thread()=default;
};

private:
/**
        私有方法
    */
// work：線程運行主函數(shù)，持續(xù)不斷地從本地任務(wù)隊列 taskq 或本地協(xié)程隊列 t_schedq 中獲取任務(wù)/協(xié)程進行調(diào)度. 倘若本地任務(wù)為空，會嘗試從其他線程本地任務(wù)隊列竊取任務(wù)執(zhí)行
void work();
/**
     * readAndGo：從指定的任務(wù)隊列中獲取任務(wù)并執(zhí)行
     * param：taskq——指定的任務(wù)隊列 nonblock——是否為阻塞模式
     * reponse：true——成功 false——失敗
    */
bool readAndGo(localqPtr taskq, bool nonblock);
/**
     * goTask: 為一筆任務(wù)創(chuàng)建一個協(xié)程實例，并調(diào)度該任務(wù)函數(shù)
     * param: cb——待執(zhí)行任務(wù)
     * tip：如果該任務(wù)未一次性執(zhí)行完成（途中使用了 sched 方法），則會在棧中封存好任務(wù)的執(zhí)行信息，然后將該協(xié)程實例追加到線程本地的協(xié)程隊列 t_schedq 中，等待后續(xù)再被線程調(diào)度
     */
void goTask(task cb);
/**
     * goWorker：調(diào)度某個協(xié)程實例，其中已經(jīng)分配好執(zhí)行的任務(wù)函數(shù)
     * param: worker——分配好執(zhí)行任務(wù)函數(shù)的協(xié)程實例
     * tip：如果該任務(wù)未一次性執(zhí)行完成（途中使用了 sched 方法），則會在棧中封存好任務(wù)的執(zhí)行信息，然后將該協(xié)程實例追加到線程本地的協(xié)程隊列 t_schedq 中，等待后續(xù)再被線程調(diào)度
    */
void goWorker(workerPtr worker);

/**
     * workStealing：當其他線程任務(wù)隊列 taskq 中竊取半數(shù)任務(wù)填充到本地隊列
     */
void workStealing();
/**
     * workStealing 重載：從線程 stealFrom 的任務(wù)隊列中竊取半數(shù)任務(wù)填充到線程 stealTo 本地隊列
     */
void workStealing(thread::ptr stealTo, thread::ptr stealFrom);
/**
     * getStealingTarget：隨機獲取一個線程作為竊取目標
     */
thread::ptr getStealingTarget();

/**
     * getThreadByThreadName 通過線程名稱獲取對應(yīng)的線程實例
     */
thread::ptr getThreadByThreadName(std::string threadName);
/**
     * getThread 獲取當前線程實例
     */
thread::ptr getThread();

private:
/**
     * 靜態(tài)私有方法
     */
// getThreadNameByIndex：通過線程 index 映射得到線程名稱
static const std::string getThreadNameByIndex(int index);
// getThreadIndex：獲取當前線程的 index
static const int getThreadIndex();
// getThreadName：獲取當前線程的名稱
static const std::string getThreadName();


private:
/**
     * 私有成員屬性
     */
// 基于 vector 實現(xiàn)的線程池，元素類型為 WorkerPool::thread 對應(yīng)共享指針
    std::vector<thread::ptr> m_threadPool;

// 基于原子變量標識 workerPool 是否已關(guān)閉
    std::atomic<bool> m_closed{false};
};

}}

// 標準庫隊列實現(xiàn). 依賴隊列作為線程本地協(xié)程隊列的存儲載體
#include <queue>

// workerpool 頭文件
#include "workerpool.h"
// 本項目定義的斷言頭文件
#include "../trace/assert.h"

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

/**
 * 全局變量 s_taskId：用于分配任務(wù) id 的全局遞增計數(shù)器，通過原子變量保證并發(fā)安全
 * 每個任務(wù)函數(shù)會根據(jù)分配到的 id，被均勻地分發(fā)給各個線程，以此實現(xiàn)負載均衡
*/
static std::atomic<int> s_taskId{0};

/**
 * 線程本地變量  t_schedq：線程私有的協(xié)程隊列
 * 當線程下某個協(xié)程沒有一次性將任務(wù)執(zhí)行完成時（任務(wù)調(diào)用了 sched 讓渡函數(shù)），則該協(xié)程會被暫存于此隊列中，等待后續(xù)被相同的線程繼續(xù)調(diào)度
 */
staticthread_local std::queue<WorkerPool::workerPtr> t_schedq;

// ...

}}

// namespace cbricks::pool
namespace cbricks{namespace pool{
// ...
/**
 * workerpool 構(gòu)造函數(shù)：
 * - 初始化好各個線程實例 thread
 * - 將各 thread 添加到線程池 m_threadPool 中
 */
WorkerPool::WorkerPool(size_t threads){
CBRICKS_ASSERT(threads >0,"worker pool init with nonpositive threads num");

// 為線程池預留好對應(yīng)的容量
this->m_threadPool.reserve(threads);

/**
     * 構(gòu)造好對應(yīng)于每個 thread 的信號量
     * 這是為了保證 thread 實例先被添加進入 m_threadPool，thread 中的調(diào)度函數(shù)才能往下執(zhí)行
     * 這樣做是因為 thread 調(diào)度函數(shù)有依賴于從 m_threadPool 獲取自身實例的操作
    */
std::vector<semaphore> sems(threads);
// 另一個信號量，用于保證所有 thread 調(diào)度函數(shù)都正常啟動后，當前構(gòu)造函數(shù)才能退出，避免 sems 被提前析構(gòu)
    semaphore waitGroup;

// 初始化好對應(yīng)數(shù)量的 thread 實例并添加進入 m_threadPool
for(int i =0; i < threads; i++){
// 根據(jù) 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](){
/**
                 * 此處 wait 操作是需要等待對應(yīng) thread 實例先被推送進入 m_threadPool
                 * 因為一旦后續(xù)的 work 函數(shù)運行，就會涉及從 m_threadPool 中獲取 thread 實例的操作
                 * 因此先后順序不能顛倒
                 */
                sems[getThreadIndex()].wait();
/**
                 * 此處 notify 操作是配合外層的 waitGroup.wait 操作
                 * 保證所有 thread 都正常啟動后，workerPool 構(gòu)造函數(shù)才能退出
                 * 這是為了防止 sems 被提前析構(gòu)
                 */
                waitGroup.notify();
// 異步啟動的 thread，最終運行的調(diào)度函數(shù)是 workerpool::work
this->work();

},
// 注入 thread 名稱，與 index 有映射關(guān)系
            threadName),
// 分配給 thread 的本地任務(wù)隊列
localqPtr(new localq))));
/**
         * 在 thread 實例被推送入 m_threadPool 后進行 notify
         * 這樣 thread 調(diào)度函數(shù)才會被向下放行
         */
        sems[i].notify();
}

/**
     * 等待所有 thread 實例正常啟動后，構(gòu)造函數(shù)再退出
     */
for(int i =0; i < threads; i++){
        waitGroup.wait();
}
}

// 析構(gòu)函數(shù)
WorkerPool::~WorkerPool(){
// 將 workpool 的關(guān)閉標識置為 true，后續(xù)運行中的線程感知到此標識后會主動退出
this->m_closed.store(true);
// 等待所有線程都退出后，再退出 workpool 的析構(gòu)函數(shù)
for(int i =0; i <this->m_threadPool.size(); i++){
// 關(guān)閉各 thread 的本地任務(wù)隊列
this->m_threadPool[i]->taskq->close();
// 等待各 thread 退出
this->m_threadPool[i]->thr->join();
}
}

// ...

}}

// namespace cbricks::pool
namespace cbricks{namespace pool{
// ...
/**
 * submit: 提交一個任務(wù)到協(xié)程調(diào)度池中，任務(wù)以閉包函數(shù) void() 的形式組裝
 * - 為任務(wù)分配全局遞增且唯一的 taskId
 * - 根據(jù) taskId 將任務(wù)均勻分發(fā)給指定 thread
 * - 將任務(wù)寫入到指定 thread 的本地任務(wù)隊列中
 */
bool WorkerPool::submit(task task, bool nonblock){
// 若 workerpool 已關(guān)閉，則提交失敗
if(this->m_closed.load()){
returnfalse;
}

// 基于任務(wù) id 對 m_threadPool 長度取模，將任務(wù)映射到指定 thread
int targetThreadId =(s_taskId++)%(this->m_threadPool.size());
    thread::ptr targetThr =this->m_threadPool[targetThreadId];

// 針對目標 thread 加讀鎖，這是為了防止和目標 thread 的 workstealing 操作并發(fā)最終因任務(wù)隊列 taskq 容量溢出而導致死鎖
rwlock::readLockGuard guard(targetThr->lock);

// 往對應(yīng) thread 的本地任務(wù)隊列中寫入任務(wù)
return targetThr->taskq->write(task, nonblock);
}

// ...

}}

// namespace cbricks::pool
namespace cbricks{ namespace pool{
// ...
// sched：讓渡函數(shù). 在任務(wù)執(zhí)行過程中，可以通過該方法主動讓出線程的執(zhí)行權(quán)，則此時任務(wù)所屬的協(xié)程會被添加到 thread 的本地協(xié)程隊列 t_schedq 中，等待后續(xù)再被調(diào)度執(zhí)行
void WorkerPool::sched(){
    worker::GetThis()->sched();
}

// ...

}}

// namespace cbricks::pool
namespace cbricks{namespace pool{
// ...
/**
 * work: 線程運行的主函數(shù)
 * 1） 獲取需要調(diào)度的協(xié)程（下述任意步驟執(zhí)行成功，則跳到步驟 2））
 *   - 從本地任務(wù)隊列 taskq 中取任務(wù)，獲取成功則為之初始化協(xié)程實例
 *   - 從本地協(xié)程隊列 schedq 中取協(xié)程
 *   - 從其他線程的任務(wù)隊列 taskq 中偷取一半任務(wù)到本地任務(wù)隊列
 * 2） 調(diào)度協(xié)程執(zhí)行任務(wù)
 * 3) 針對主動讓渡而退出的協(xié)程，添加到本地協(xié)程隊列
 * 4) 循環(huán)上述流程
 */
void WorkerPool::work(){
// 獲取到當前 thread 對應(yīng)的本地任務(wù)隊列 taskq
    localqPtr taskq =this->getThread()->taskq;

// main loop
while(true){
// 如果 workerpool 已關(guān)閉 則主動退出
if(this->m_closed.load()){
return;
}

/**  
         * 執(zhí)行優(yōu)先級為 本地任務(wù)隊列 taskq -> 本地協(xié)程隊列 t_t_schedq -> 竊取其他線程任務(wù)隊列 other_taskq
         * 為防止饑餓，至多調(diào)度 10 次的 taskq 后，必須嘗試處理一次 t_schedq 
        */

// 標識本地任務(wù)隊列 taskq 是否為空
bool taskqEmpty =false;
// 至多調(diào)度 10 次本地任務(wù)隊列 taskq
for(int i =0; i <10; i++){
// 以【非阻塞模式】從 taskq 獲取任務(wù)并為之分配協(xié)程實例和調(diào)度執(zhí)行
if(!this->readAndGo(taskq,false)){
// 如果 taskq 為空，將 taskqEmpty 置為 true 并直接退出循環(huán)
                taskqEmpty =true;
break;
}
}

// 嘗試從線程本地的協(xié)程隊列 t_schedq 中獲取協(xié)程并進行調(diào)度
if(!t_schedq.empty()){
// 從協(xié)程隊列中取出頭部的協(xié)程實例
            workerPtr worker = t_schedq.front();
            t_schedq.pop();
// 進行協(xié)程調(diào)度
this->goWorker(worker);
// 處理完成后直接進入下一輪循環(huán)
continue;
}

// 如果未發(fā)現(xiàn) taskq 為空，則無需 workstealing，直接進入下一輪循環(huán)
if(!taskqEmpty){
continue;
}

/** 
         * 走到這里意味著 taskq 和 schedq 都是空的，則要嘗試發(fā)起竊取操作
         * 隨機選擇一個目標線程竊取半數(shù)任務(wù)添加到本地隊列中
        */
this->workStealing();

/**  
         * 以【阻塞模式】嘗試從本地任務(wù)獲取任務(wù)并調(diào)度執(zhí)行
         * 若此時仍沒有可調(diào)度的任務(wù)，則當前 thread 陷入阻塞，讓出 cpu 執(zhí)行權(quán)
         * 直到有新任務(wù)分配給當前 thread 時，thread 才會被喚醒
        */
this->readAndGo(taskq,true);
}
}
// ...

}}

// namespace cbricks::pool
namespace cbricks{namespace pool{
// ...
/**
 * readAndGo：
 *   - 從指定任務(wù)隊列中獲取一個任務(wù)
 *   - 為之分配協(xié)程實例并調(diào)度執(zhí)行
 *   - 若協(xié)程實例未一次性執(zhí)行完成（執(zhí)行了讓渡 sched），則將協(xié)程添加到線程本地的協(xié)程隊列 schedq 中
 * param：taskq——任務(wù)隊列；nonblock——是否以非阻塞模式從任務(wù)隊列中獲取任務(wù)
 * response：true——成功；false，失?。ㄈ蝿?wù)隊列 taskq 為空）
 */
// 將一個任務(wù)包裝成協(xié)程并進行調(diào)度. 如果沒有一次性調(diào)度完成，則將協(xié)程實例添加到線程本地的協(xié)程隊列 t_schedq
bool WorkerPool::readAndGo(cbricks::pool::WorkerPool::localqPtr taskq, bool nonblock){
// 任務(wù)容器
    task cb;
// 從 taskq 中獲取任務(wù)
if(!taskq->read(cb,nonblock)){
returnfalse;
}

// 對任務(wù)進行調(diào)度
this->goTask(cb);
returntrue;
}

/**
 * goTask
 *   - 為指定任務(wù)分配協(xié)程實例
 *   - 執(zhí)行協(xié)程
 *   - 若協(xié)程實例未一次性執(zhí)行完成（執(zhí)行了讓渡 sched），則將協(xié)程添加到線程本地的協(xié)程隊列 schedq 中
 * param：cb——待執(zhí)行的任務(wù)
 */
void WorkerPool::goTask(task cb){
// 初始化協(xié)程實例
    workerPtr _worker(newworker(cb));
// 調(diào)度協(xié)程
this->goWorker(_worker);
}

/**
 * goWorker
 *   - 執(zhí)行協(xié)程
 *   - 若協(xié)程實例未一次性執(zhí)行完成（執(zhí)行了讓渡 sched），則將協(xié)程添加到線程本地的協(xié)程隊列 schedq 中
 * param：worker——待運行的協(xié)程
 */
void WorkerPool::goWorker(workerPtr worker){
// 調(diào)度協(xié)程，此時線程的執(zhí)行權(quán)會切換進入到協(xié)程對應(yīng)的方法棧中
    worker->go();
// 走到此處意味著線程執(zhí)行權(quán)已經(jīng)從協(xié)程切換回來
// 如果此時協(xié)程并非已完成的狀態(tài)，則需要將其添加到線程本地的協(xié)程隊列 schedq 中，等待后續(xù)繼續(xù)調(diào)度
if(worker->getState()!= sync::Coroutine::Dead){
        t_schedq.push(worker);
}
}
// ...

}}

// namespace cbricks::pool
namespace cbricks{namespace pool{
// ...
// 從某個 thread 中竊取一半任務(wù)給到本 thread 的 taskq
void WorkerPool::workStealing(){
// 選擇一個竊取的目標 thread 
    thread::ptr stealFrom =this->getStealingTarget();
if(!stealFrom){
return;
}
// 從目標 thread 中竊取半數(shù)任務(wù)添加到本 thread taskq 中
this->workStealing(this->getThread(),stealFrom);
}

// 從 thread:stealFrom 中竊取半數(shù)任務(wù)給到 thread:stealTo
void WorkerPool::workStealing(thread::ptr stealTo, thread::ptr stealFrom){
// 確定竊取任務(wù)數(shù)量：目標本地任務(wù)隊列 taskq 中任務(wù)總數(shù)的一半
int stealNum = stealFrom->taskq->size()/2;
if(stealNum <=0){
return;
}

// 針對 thread:stealTo 加寫鎖，防止因 workstealing 和 submit 行為并發(fā)，導致線程因 taskq 容量溢出而發(fā)生死鎖
rwlock::lockGuard guard(stealTo->lock);
// 檢查此時 stealTo 中的 taskq 如果剩余容量已不足以承載擬竊取的任務(wù)量，則直接退出
if(stealTo->taskq->size()+ stealNum > stealTo->taskq->cap()){
return;
}

// 創(chuàng)建任務(wù)容器，以非阻塞模式從 stealFrom 的 taskq 中竊取指定數(shù)量的任務(wù)
std::vector<task> containers(stealNum);
if(!stealFrom->taskq->readN(containers,true)){
return;
}

// 將竊取到的任務(wù)添加到 stealTo 的 taskq 中
    stealTo->taskq->writeN(containers,false);
}

// 隨機選擇本 thread 外的一個 thread 作為竊取的目標
WorkerPool::thread::ptr WorkerPool::getStealingTarget(){
// 如果線程池長度不足 2，直接返回
if(this->m_threadPool.size()<2){
returnnullptr;
}

// 通過隨機數(shù)，獲取本 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];
}
// ...

}}