C++高性能服務(wù)器框架之協(xié)程調(diào)度模塊
協(xié)程調(diào)度模塊概述
封裝了一個(gè)N : M協(xié)程調(diào)度器,創(chuàng)建M個(gè)協(xié)程在N個(gè)線程上運(yùn)行。通過(guò)schedule()
方法將cb或fiber
重新加到任務(wù)隊(duì)列中執(zhí)行任務(wù),協(xié)程可以在線程上自由切換,也可以在指定線程上執(zhí)行。
1 —— N 1 —— M
scheduler ---> thread ---> fiber
?
N : M 協(xié)程可以在線程間自由切換
1. 線程池, 分配一組線程
2. 協(xié)程調(diào)度器,將協(xié)程指定到相應(yīng)線程上執(zhí)行
a)隨機(jī)選擇空閑的協(xié)程執(zhí)行
b)協(xié)程指定必須在某個(gè)線程上執(zhí)行
協(xié)程調(diào)度器調(diào)度主要思想為:先查看任務(wù)隊(duì)列中有沒(méi)有任務(wù)需要執(zhí)行,若沒(méi)有任務(wù)需要執(zhí)行則執(zhí)行idel()
,其思想主要在run()
中體現(xiàn)。
其次,在設(shè)計(jì)協(xié)程調(diào)度器時(shí),設(shè)置了一個(gè)use_caller
來(lái)決定是否將當(dāng)前調(diào)度線程也納入調(diào)度中,這樣可以少創(chuàng)建一個(gè)線程執(zhí)行任務(wù),效率更高。
詳解
class Scheduler
兩個(gè)局部線程變量保存當(dāng)前線程的協(xié)程調(diào)度器和主協(xié)程
// 當(dāng)前協(xié)程調(diào)度器 static thread_local Scheduler* t_secheduler = nullptr; // 線程主協(xié)程 static thread_local Fiber* t_fiber = nullptr;
FiberAndThread(任務(wù)結(jié)構(gòu)體)
通過(guò)FiberAndThread
結(jié)構(gòu)體在存儲(chǔ)協(xié)程,回調(diào)函數(shù),線程的信息
struct FiberAndThread { ? ? ? ?// 協(xié)程 ? ? ? ?Fiber::ptr fiber; ? ? ? ?// 協(xié)程執(zhí)行函數(shù) ? ? ? ?std::function<void()> cb; ? ? ? ?// 線程id 協(xié)程在哪個(gè)線程上 ? ? ? ?int thread; ? // 確定協(xié)程在哪個(gè)線程上跑 ? ? ? ?FiberAndThread(Fiber::ptr f, int thr) ? ? ? ? ? :fiber(f), thread(thr) { ? ? ? } ? // 通過(guò)swap將傳入的 fiber 置空,使其引用計(jì)數(shù)-1 ? ? ? ?FiberAndThread(Fiber::ptr* f, int thr) ? ? ? ? ? :thread(thr) { ? ? ? ? ? ?fiber.swap(*f); ? ? ? } ? // 確定回調(diào)在哪個(gè)線程上跑 ? FiberAndThread(std::function<void()> f, int thr) ? ? ? ? ? :cb(f), thread(thr) { ? ? ? } ? // 通過(guò)swap將傳入的 cb 置空,使其引用計(jì)數(shù)-1 ? FiberAndThread(std::function<void()>* f, int thr) ? ? ? ? ? :thread(thr) { ? ? ? ? ? ?cb.swap(*f); ? ? ? } ? ? ? ? ?// 默認(rèn)構(gòu)造 ? ? ? ?FiberAndThread() { ? ? ? ? ? ?thread = -1; ? ? ? } ? // 重置 ? ? ? ?void reset() { ? ? ? ? ? ?fiber = nullptr; ? ? ? ? ? ?cb = nullptr; ? ? ? ? ? ?thread = -1; ? ? ? } ? ? };
mumber(成員變量)
private: ? ?// Mutex ? ?MutexType m_mutex; ? ?// 線程池 ? ?std::vector<sylar::Thread::ptr> m_threads; ? ?// 待執(zhí)行的協(xié)程隊(duì)列 ? ?std::list<FiberAndThread> m_fibers; ? ?// use_caller為true時(shí)有效,調(diào)度協(xié)程 ? ?Fiber::ptr m_rootFiber; ? ?// 協(xié)程調(diào)度器名稱 ? ?std::string m_name; ? protected: ? ?// 協(xié)程下的線程id數(shù)組 ? ?std::vector<int> m_threadIds; ? ?// 線程數(shù)量 ? ?size_t m_threadCount = 0; ? ?// 工作線程數(shù)量 ? ?std::atomic<size_t> m_activateThreadCount = {0}; ? ?// 空閑線程數(shù)量 ? ?std::atomic<size_t> m_idleThreadCount = {0}; ? ?// 是否正在停止 ? ?bool m_stopping = true; ? ?// 是否自動(dòng)停止 ? ?bool m_autoStop = false; ? ?// 主線程Id(use_caller) ? ?int m_rootThread = 0;
scheduler(調(diào)度協(xié)程)
// 調(diào)度協(xié)程 template<class FiberOrCb> void schedule(FiberOrCb fc, int thread = -1) { ? ?bool need_tickle = false; ? { ? ? ? ?MutexType::Lock lock(m_mutex); ? ? ? ?// 將任務(wù)加入到隊(duì)列中,若任務(wù)隊(duì)列中已經(jīng)有任務(wù)了,則tickle() ? ? ? ?need_tickle = scheduleNoLock(fc, thread); ? } ? ? ?if (need_tickle) { ? ? ? ?tickle(); ? } } // 批量調(diào)度協(xié)程 template<class InputIterator> void schedule(InputIterator begin, InputIterator end) { ? ?bool need_tickle = false; ? { ? ? ? ?MutexType::Lock lock(m_mutex); ? ? ? ?while (begin != end) { ? ? ? ? ? ?need_tickle = scheduleNoLock(&*begin, -1) || need_tickle; ? ? ? ? ? ?++begin; ? ? ? } ? } ? ?if (need_tickle) { ? ? ? ?tickle(); ? } }
檢查任務(wù)隊(duì)列中有無(wú)任務(wù),將任務(wù)加入到任務(wù)隊(duì)列中,若任務(wù)隊(duì)列中本來(lái)就已經(jīng)有任務(wù)了,就tickle
以下
/** * @brief 協(xié)程調(diào)度啟動(dòng)(無(wú)鎖) */ template<class FiberOrCb> bool scheduleNoLock(FiberOrCb fc, int thread) { ? ?bool need_tickle = m_fibers.empty(); ? ?FiberAndThread ft(fc, thread); ? ?if (ft.fiber || ft.cb) { ? ? ? ?m_fibers.push_back(ft); ? } ? ?return need_tickle; }
Scheduler(構(gòu)造函數(shù))
Scheduler::Scheduler(size_t threads, bool use_caller, const std::string &name) ? :m_name(name) { // 確定線程數(shù)量要正確 ? ?SYLAR_ASSERT(threads > 0); ? ?// 是否將協(xié)程調(diào)度線程也納入調(diào)度器 ? ?if (use_caller) { ? ? ? ?// 設(shè)置線程名稱 ? ? ? ?sylar::Thread::SetName(m_name); ? ? ? ?// 獲得主協(xié)程 ? ? ? ?sylar::Fiber::GetThis(); ? ? ? ?// 線程數(shù)量-1 ? ? ? ?--threads; ? ? ? ?SYLAR_ASSERT(GetThis() == nullptr); ? ? ? ?// 設(shè)置當(dāng)前協(xié)程調(diào)度器 ? ? ? ?t_secheduler = this; ? ? ? ? ?// 將此fiber設(shè)置為 use_caller,協(xié)程則會(huì)與 Fiber::CallerMainFunc() 綁定 ? ? ? ?// 非靜態(tài)成員函數(shù)需要傳遞this指針作為第一個(gè)參數(shù),用 std::bind()進(jìn)行綁定 ? ? ? ?m_rootFiber.reset(new Fiber(std::bind(&Scheduler::run, this), 0, true)); ? ? ? ?// 設(shè)置當(dāng)前線程的主協(xié)程為m_rootFiber ? ? ? ?// 這里的m_rootFiber是該線程的主協(xié)程(執(zhí)行run任務(wù)的協(xié)程),只有默認(rèn)構(gòu)造出來(lái)的fiber才是主協(xié)程 ? ? ? ?t_fiber = m_rootFiber.get(); ? ? ? ?// 獲得當(dāng)前線程id ? ? ? ?m_rootThread = sylar::GetThreadId(); ? ? ? ?m_threadIds.push_back(m_rootThread); ? } ? ?// 不將當(dāng)前線程納入調(diào)度器 ? ?else { ? ? ? ?m_rootThread = -1; ? } ? ?m_threadCount = threads; }
~Scheduler(析構(gòu)函數(shù))
Scheduler::~Scheduler() { ? ?// 必須達(dá)到停止條件 ? ?SYLAR_ASSERT(m_stopping); ? ?if (GetThis() == this) { ? ? ? ?t_secheduler = nullptr; ? } }
start(啟動(dòng)調(diào)度器)
void Scheduler::start() { SYLAR_LOG_INFO(g_logger) << "start()"; MutexType::Lock lock(m_mutex); // 已經(jīng)啟動(dòng)了 if (!m_stopping) { return; } // 將停止?fàn)顟B(tài)設(shè)置為false m_stopping = false; // 線程池為空 SYLAR_ASSERT(m_threads.empty()); // 創(chuàng)建線程池 m_threads.resize(m_threadCount); for (size_t i = 0; i < m_threadCount; ++i) { // 線程執(zhí)行 run() 任務(wù) m_threads[i].reset(new Thread(std::bind(&Scheduler::run, this) , m_name + "_" + std::to_string(i))); m_threadIds.push_back(m_threads[i]->getId()); } lock.unlock(); /* 在這里切換線程時(shí),swap的話會(huì)將線程的主協(xié)程與當(dāng)前協(xié)程交換,當(dāng)使用use_caller時(shí),t_fiber = m_rootFiber,call是將當(dāng)前協(xié)程與主協(xié)程交換 * 為了確保在啟動(dòng)之后仍有任務(wù)加入任務(wù)隊(duì)列中,所以在stop()中做該線程的啟動(dòng),這樣就不會(huì)漏掉任務(wù)隊(duì)列中的任務(wù) */ /* if(m_rootFiber) { // t_fiber = m_rootFiber.get(), 從自己切換到自己了屬于是 // m_rootFiber->swapIn(); m_rootFiber->call(); } */ SYLAR_LOG_INFO(g_logger) << "start() end"; }
stop(停止調(diào)度器)
void Scheduler::stop() { SYLAR_LOG_INFO(g_logger) << "stop()"; // 進(jìn)入stop將自動(dòng)停止設(shè)為true m_autoStop = true; // 使用use_caller,并且只有一個(gè)線程,并且主協(xié)程的狀態(tài)為結(jié)束或者初始化 if (m_rootFiber && m_threadCount == 0 && (m_rootFiber->getState() == Fiber::TERM || m_rootFiber->getState() == Fiber::INIT)) { SYLAR_LOG_INFO(g_logger) << this->m_name << " sheduler stopped"; // 停止?fàn)顟B(tài)為true m_stopping = true; // 若達(dá)到停止條件則直接return if (stopping()) { return; } } // use_caller線程 // 當(dāng)前調(diào)度器和t_secheduler相同 if (m_rootThread != -1) { SYLAR_ASSERT(GetThis() == this); } // 非use_caller,此時(shí)的t_secheduler應(yīng)該為nullptr else { SYLAR_ASSERT(GetThis() != this); } // 停止?fàn)顟B(tài)為true m_stopping = true; // 每個(gè)線程都tickle一下 for (size_t i = 0; i < m_threadCount; ++i) { tickle(); } // 使用use_caller多tickle一下 if (m_rootFiber) { tickle(); } // 使用use_caller,只要沒(méi)達(dá)到停止條件,調(diào)度器主協(xié)程交出執(zhí)行權(quán),執(zhí)行run if (m_rootFiber) { if (!stopping()) { m_rootFiber->call(); } } std::vector<Thread::ptr> thrs; { MutexType::Lock lock(m_mutex); thrs.swap(m_threads); } // 等待線程執(zhí)行完成 for (auto& i : thrs) { i->join(); } }
run(協(xié)程調(diào)度函數(shù))
void Scheduler::run() { SYLAR_LOG_INFO(g_logger) << "run()"; // hook set_hook_enable(true); // 設(shè)置當(dāng)前調(diào)度器 setThis(); // 非user_caller線程,設(shè)置主協(xié)程為線程主協(xié)程 if (sylar::GetThreadId() != m_rootThread) { t_fiber = Fiber::GetThis().get(); } SYLAR_LOG_DEBUG(g_logger) << "new idle_fiber"; // 定義dile_fiber,當(dāng)任務(wù)隊(duì)列中的任務(wù)執(zhí)行完之后,執(zhí)行idle() Fiber::ptr idle_fiber(new Fiber(std::bind(&Scheduler::idle, this))); Fiber::ptr cb_fiber; FiberAndThread ft; while (true) { ft.reset(); bool tickle_me = false; bool is_active = false; { // 從任務(wù)隊(duì)列中拿fiber和cb MutexType::Lock lock(m_mutex); auto it = m_fibers.begin(); while (it != m_fibers.end()) { // 如果當(dāng)前任務(wù)指定的線程不是當(dāng)前線程,則跳過(guò),并且tickle一下 if (it->thread != -1 && it->thread != sylar::GetThreadId()) { ++it; tickle_me = true; continue; } // 確保fiber或cb存在 SYLAR_ASSERT(it->fiber || it->cb); // 如果該fiber正在執(zhí)行則跳過(guò) if (it->fiber && it->fiber->getState() == Fiber::EXEC) { ++it; continue; } // 取出該任務(wù) ft = *it; // 從任務(wù)隊(duì)列中清除 m_fibers.erase(it); // 正在執(zhí)行任務(wù)的線程數(shù)量+1 ++m_activateThreadCount; // 正在執(zhí)行任務(wù) is_active = true; break; } } // 取到任務(wù)tickle一下 if (tickle_me) { tickle(); } // 如果任務(wù)是fiber,并且任務(wù)處于可執(zhí)行狀態(tài) if (ft.fiber && (ft.fiber->getState() != Fiber::TERM || ft.fiber->getState() != Fiber::EXCEPT)) { // 執(zhí)行任務(wù) ft.fiber->swapIn(); // 執(zhí)行完成,活躍的線程數(shù)量減-1 --m_activateThreadCount; // 如果線程的狀態(tài)被置為了READY if (ft.fiber->getState() == Fiber::READY) { // 將fiber重新加入到任務(wù)隊(duì)列中 schedule(ft.fiber); // INIT或HOLD狀態(tài) } else if (ft.fiber->getState() != Fiber::TERM && ft.fiber->getState() != Fiber::EXCEPT) { // 設(shè)置fiber狀態(tài)為HOLD ft.fiber->setState(Fiber::HOLD); } // 執(zhí)行完畢重置數(shù)據(jù)ft ft.reset(); // 如果任務(wù)是cb } else if(ft.cb) { // cb_fiber存在,重置該fiber if (cb_fiber) { cb_fiber->reset(ft.cb); // cb_fiber不存在,new新的fiber } else { SYLAR_LOG_DEBUG(g_logger) << "new ft.cb"; cb_fiber.reset(new Fiber(ft.cb)); } // 重置數(shù)據(jù)ft ft.reset(); // 執(zhí)行cb任務(wù) cb_fiber->swapIn(); // 執(zhí)行完,執(zhí)行任務(wù)線程數(shù)量-1 --m_activateThreadCount; // 若cb_fiber狀態(tài)為READY if (cb_fiber->getState() == Fiber::READY) { // 重新放入任務(wù)隊(duì)列中 schedule(cb_fiber); // 釋放智能指針 cb_fiber.reset(); // cb_fiber異?;蚪Y(jié)束,就重置狀態(tài),可以再次使用該cb_fiber } else if (cb_fiber->getState() == Fiber::EXCEPT || cb_fiber->getState() == Fiber::TERM) { // cb_fiber的執(zhí)行任務(wù)置空 cb_fiber->reset(nullptr); } else { // 設(shè)置狀態(tài)為HOLD,此任務(wù)后面還會(huì)通過(guò)ft.fiber被拉起 cb_fiber->setState(Fiber::HOLD); // 釋放該智能指針,調(diào)用下一個(gè)任務(wù)時(shí)要重新new一個(gè)新的cb_fiber cb_fiber.reset(); } // 沒(méi)有任務(wù)執(zhí)行 } else { // 我感覺(jué)這里判斷么啥用 if (is_active) { --m_activateThreadCount; continue; } // 如果idle_fiber的狀態(tài)為TERM則結(jié)束循環(huán),真正的結(jié)束 if (idle_fiber->getState() == Fiber::TERM) { SYLAR_LOG_INFO(g_logger) << "idle_fiber term"; break; } // 正在執(zhí)行idle的線程數(shù)量+1 ++m_idleThreadCount; // 執(zhí)行idle() idle_fiber->swapIn(); // 正在執(zhí)行idle的線程數(shù)量-1 --m_idleThreadCount; // idle_fiber狀態(tài)置為HOLD if (idle_fiber->getState() != Fiber::TERM && idle_fiber->getState() != Fiber::EXCEPT) { idle_fiber->setState(Fiber::HOLD); } } } }
stopping(判斷停止條件)
bool Scheduler::stopping() { MutexType::Lock lock(m_mutex); // 當(dāng)自動(dòng)停止 && 正在停止 && 任務(wù)隊(duì)列為空 && 活躍的線程數(shù)量為0 return m_autoStop && m_stopping && m_fibers.empty() && m_activateThreadCount == 0; }
總結(jié)
舉個(gè)具體的例子。
#include "../sylar/sylar.h" static sylar::Logger::ptr g_logger = SYLAR_LOG_ROOT(); void test_fiber() { static int count = 5; SYLAR_LOG_INFO(g_logger) << "---test in fiber---" << count; sylar::set_hook_enable(false); sleep(1); // 循環(huán)將test_fiber加入到任務(wù)隊(duì)列中,并且指定第一個(gè)拿到該任務(wù)的線程一直執(zhí)行 if (--count > 0) { sylar::Scheduler::GetThis()->schedule(&test_fiber, sylar::GetThreadId()); } } int main(int argc, char** argv) { g_logger->setLevel(sylar::LogLevel::INFO); sylar::Thread::SetName("main"); SYLAR_LOG_INFO(g_logger) << "main start"; sylar::Scheduler sc(2, false, "work"); sc.start(); SYLAR_LOG_INFO(g_logger) << "schedule"; sc.schedule(&test_fiber); sc.stop(); SYLAR_LOG_INFO(g_logger) << "main end"; return 0; }
// 設(shè)置2個(gè)線程, 并且將use_caller設(shè)為false, 設(shè)置名稱為"work", 指定線程 // 這里可以看到有3個(gè)線程 1684 1685 1686 // 1684為調(diào)度線程, 1685和1686為執(zhí)行任務(wù)的線程 // 可以看到任務(wù)都是在1686線程上執(zhí)行的,因?yàn)樵趕hceduler()時(shí)指定了任務(wù)在第一個(gè)拿到該任務(wù)的線程上一直執(zhí)行 1684 main 0 [INFO] [root] tests/test_scheduler.cc:20 main start 1684 main 0 [INFO] [root] tests/test_scheduler.cc:23 schedule 1686 work_1 4 [INFO] [root] tests/test_scheduler.cc:7 ---test in fiber---5 1686 work_1 4 [INFO] [root] tests/test_scheduler.cc:7 ---test in fiber---4 1686 work_1 4 [INFO] [root] tests/test_scheduler.cc:7 ---test in fiber---3 1686 work_1 4 [INFO] [root] tests/test_scheduler.cc:7 ---test in fiber---2 1686 work_1 4 [INFO] [root] tests/test_scheduler.cc:7 ---test in fiber---1 1685 work_0 0 [INFO] [system] sylar/scheduler.cc:237 idle_fiber term 1686 work_1 0 [INFO] [system] sylar/scheduler.cc:237 idle_fiber term 1684 main 0 [INFO] [root] tests/test_scheduler.cc:27 main end
// 設(shè)置2個(gè)線程, 并且將use_caller設(shè)為true, 設(shè)置名稱為"work",不指定線程 // 這里可以看到有2個(gè)線程 2841 2842 // 2841為調(diào)度線程,他也將自己納入調(diào)度器中執(zhí)行任務(wù) 2841 work 0 [INFO] [root] tests/test_scheduler.cc:20 main start 2841 work 0 [INFO] [root] tests/test_scheduler.cc:23 schedule 2842 work_0 4 [INFO] [root] tests/test_scheduler.cc:7 ---test in fiber---5 2841 work 6 [INFO] [root] tests/test_scheduler.cc:7 ---test in fiber---4 2842 work_0 4 [INFO] [root] tests/test_scheduler.cc:7 ---test in fiber---3 2841 work 6 [INFO] [root] tests/test_scheduler.cc:7 ---test in fiber---2 2842 work_0 4 [INFO] [root] tests/test_scheduler.cc:7 ---test in fiber---1 2842 work_0 0 [INFO] [system] sylar/scheduler.cc:237 idle_fiber term 2841 work 1 [INFO] [system] sylar/scheduler.cc:237 idle_fiber term 2841 work 0 [INFO] [root] tests/test_scheduler.cc:27 main end
當(dāng)我們?cè)O(shè)置use_caller
時(shí),將調(diào)度線程也納入管理中執(zhí)行任務(wù),在構(gòu)造函數(shù)中通過(guò)Fiber::GetThis()
獲得主協(xié)程,然后new
一個(gè)子協(xié)程作為該線程的主協(xié)程并與run()
綁定,在stop()
時(shí)使用call()
將執(zhí)行權(quán)交給線程主協(xié)程執(zhí)行run()
。
在start()
時(shí)創(chuàng)建其他線程并與run()
綁定,此時(shí)線程就開(kāi)始執(zhí)行run()
。
在run()
中,使用Fiber::GetThis()
獲得主協(xié)程并設(shè)置為線程主協(xié)程。在協(xié)程切換時(shí),都是執(zhí)行任務(wù)協(xié)程與線程主協(xié)程之間的切換。當(dāng)達(dá)到停止條件,idle()
執(zhí)行完畢時(shí),run()
也執(zhí)行完畢。
此時(shí)use_caller
線程使用back()
將線程主協(xié)程切換到主協(xié)程繼續(xù)執(zhí)行stop()
等待其他線程執(zhí)行完畢。
以上就是C++高性能服務(wù)器框架之協(xié)程調(diào)度模塊的詳細(xì)內(nèi)容,更多關(guān)于C++協(xié)程調(diào)度模塊的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
詳解C語(yǔ)言數(shù)據(jù)結(jié)構(gòu)之棧
這篇文章主要為大家介紹了C語(yǔ)言數(shù)據(jù)結(jié)構(gòu)之棧,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下,希望能夠給你帶來(lái)幫助2022-01-01c++ std::invalid_argument應(yīng)用
想研究std::invalid_argument的朋友可以參考下2013-01-01C語(yǔ)言單雙線性及循環(huán)鏈表與實(shí)例
這篇文章主要介紹了C語(yǔ)言的單鏈表、雙鏈表和循環(huán)鏈表,還有一些相關(guān)的實(shí)例,感興趣的同學(xué)可以借鑒一下2023-03-03C++文件關(guān)鍵詞快速定位出現(xiàn)的行號(hào)實(shí)現(xiàn)高效搜索
這篇文章主要為大家介紹了C++文件關(guān)鍵詞快速定位出現(xiàn)的行號(hào)實(shí)現(xiàn)高效搜索,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-10-10C++11 學(xué)習(xí)筆記之std::function和bind綁定器
這篇文章主要介紹了C++11 學(xué)習(xí)筆記之std::function和bind綁定器,本文給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2021-07-07