1、协程调度器介绍

• 协程模块中，对于每个协程，都需要用户手动调用协程的resume方法将协程运行起来，然后等协程运行结束并返回，再运行下一个协程。这种运行协程的方式其实是用户自己在挑选协程执行，相当于用户在充当调度器，显然不够灵活。
• 协程调度器应该具有自动消耗协程的功能，程序员只需要把需要执行的协程喂给调度器，调度器就可以自己消耗掉这些协程。
• sylar框架中的协程调度器是一个N-M的协程调度器，N个线程上可以运行M个协程，协程可以在线程之间进行切换，也可以指定某个协程在哪个线程上执行。
• 该模块的难点是当caller线程也参与调度时调度协程和主线程切换的情况
  2、设计思路
• 调度器创建后，内部首先会创建一个调度线程池，调度开始后，所有调度线程按顺序从任务队列里取任务执行，调度线程数越多，能够同时调度的任务也就越多，当所有任务都调度完后，调度线程就停下来等新的任务进来。接下来是添加调度任务，添加调度任务的本质就是往调度器的任务队列里塞任务，但是，只添加调度任务是不够的，还应该有一种方式用于通知调度线程有新的任务加进来了，因为调度线程并不一定知道有新任务进来了。当然调度线程也可以不停地轮询有没有新任务，但是这样CPU占用率会很高。接下来是调度器的停止。调度器应该支持停止调度的功能，以便回收调度线程的资源，只有当所有的调度线程都结束后，调度器才算真正停止。
• 首先是协程调度器的初始化。sylar的协程调度器在初始化时支持传入线程数和一个布尔型的use_caller参数，表示是否使用caller线程。在使用caller线程的情况下，线程数自动减一，并且调度器内部会初始化一个属于caller线程的调度协程并保存起来（比如，在main函数中创建的调度器，如果use_caller为true，那调度器会初始化一个属于main函数线程的调度协程）。调度器创建好后，即可调用调度器的schedule方法向调度器添加调度任务，但此时调度器并不会立刻执行这些任务，而是将它们保存到内部的一个任务队列中。接下来是调用start方法启动调度。start方法调用后会创建调度线程池，线程数量由初始化时的线程数和use_caller确定。调度线程一旦创建，就会立刻从任务队列里取任务执行。比较特殊的一点是，如果初始化时指定线程数为1且use_caller为true，那么start方法什么也不做，因为不需要创建新线程用于调度。并且，由于没有创建新的调度线程，那只能由caller线程的调度协程来负责调度协程，而caller线程的调度协程的执行时机与start方法并不在同一个地方。caller线程的调度协程的执行时机在stop函数中。接下来是调度协程，对应run方法。调度协程负责从调度器的任务队列中取任务执行。取出的任务即子协程，这里调度协程和子协程的切换模型即为前一章介绍的非对称模型，每个子协程执行完后都必须返回调度协程，由调度协程重新从任务队列中取新的协程并执行。如果任务队列空了，那么调度协程会切换到一个idle协程，这个idle协程什么也不做，等有新任务进来时，idle协程才会退出并回到调度协程，重新开始下一轮调度。接下来是添加调度任务，对应schedule方法，这个方法支持传入协程或函数，并且支持一个线程号参数，表示是否将这个协程或函数绑定到一个具体的线程上执行。如果任务队列为空，那么在添加任务之后，要调用一次tickle方法以通知各调度线程的调度协程有新任务来了。在执行调度任务时，还可以通过调度器的GetThis()方法获取到当前调度器，再通过schedule方法继续添加新的任务，这就变相实现了在子协程中创建并运行新的子协程的功能。**接下来是调度器的停止。**调度器的停止行为要分两种情况讨论，首先是use_caller为false的情况，这种情况下，由于没有使用caller线程进行调度，那么只需要简单地等各个调度线程的调度协程退出就行了。如果use_caller为true，表示caller线程也要参于调度，这时，调度器初始化时记录的属于caller线程的调度协程就要起作用了，在调度器停止前，应该让这个caller线程的调度协程也运行一次，让caller线程完成调度工作后再退出。如果调度器只使用了caller线程进行调度，那么所有的调度任务要在调度器停止时才会被调度。
  3、代码实现
• 协程调度器成员变量

/// 协程调度器名称 std::string m_name; /// 互斥锁 MutexType m_mutex; /// 线程池 std::vector m_threads; /// 任务队列 std::list m_tasks; /// 线程池的线程ID数组 std::vector m_threadIds; /// 工作线程数量，不包含use_caller的主线程 size_t m_threadCount = 0; /// 活跃线程数 std::atomic m_activeThreadCount = {0}; /// idle线程数 std::atomic m_idleThreadCount = {0};   /// 是否use caller bool m_useCaller; /// use_caller为true时，调度器所在线程的调度协程 Fiber::ptr m_rootFiber; /// use_caller为true时，调度器所在线程的id int m_rootThread = 0;   /// 是否正在停止 bool m_stopping = false; 

• start() 开始调度

void Scheduler::start() {     SYLAR_LOG_DEBUG(g_logger) << "start";     MutexType::Lock lock(m_mutex);     if (m_stopping) {         SYLAR_LOG_ERROR(g_logger) << "Scheduler is stopped";         return;     }     SYLAR_ASSERT(m_threads.empty());     m_threads.resize(m_threadCount);     for (size_t i = 0; i < m_threadCount; i++) {         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());     } } 

• 调度方法

 void Scheduler::run() {         SYLAR_LOG_DEBUG(g_logger) << m_name << " run";         setThis();         set_hook_enable(true);         // 少用系统调用         // if(sylar::GetThreadId() != m_rootThread) {         // 第二个条件是将use_caller为true时，非caller线程中的t_scheduler_fiber赋值，caller线程中的t_scheduler_fiber在Scheduler初始化时已经赋值了         // if (!m_useCaller || (t_scheduler_fiber == nullptr && m_useCaller)) {             t_scheduler_fiber = Fiber::GetThis();         // }         //初始话一个没有任务时执行的协程         Fiber::ptr idle_fiber(new Fiber(std::bind(&Scheduler::idle, this)));         Fiber::ptr cb_fiber;         ScheduleTask task;         while(true) {             task.reset();             bool need_tickle = false;             {                 MutexType::Lock lock(m_lock);                 //这个for的作用是找到一个可以给当前线程执行的任务                 auto it = m_tasks.begin();                 for(; it != m_tasks.end(); ++it) {                     if (it ->threadid != -1 && it->threadid != sylar::GetThreadId()) {                         need_tickle = true;                         continue;                     }                      SYLAR_ASSERT(it->cb || it->fiber);                     // 状态可以再加一个assert                     task = *it;                     ++m_activeThreadCount;                     m_tasks.erase(it++);                     break;                 }                 // 通知其他线程                 need_tickle |= (it  != m_tasks.end());             }              if(need_tickle) {                 tickle();             }              if(task.fiber) {                 // 这个任务是fiber                 task.fiber->swapIn();                 --m_activeThreadCount;                  if(task.fiber->getState() == Fiber::READY) {                     schedule(task.fiber);                 } else if(task.fiber->getState() != Fiber::TERM                     && task.fiber->getState() != Fiber::EXCEPT) {                     task.fiber->setState(Fiber::HOLD);                 }                 // 让fiber资源尽早释放，不用等到下个fiber来的时候再释放                 task.reset();             } else if(task.cb) {                 // 这个任务是callback                 if(cb_fiber) {                     cb_fiber->reset(task.cb, m_useCaller);                 } else {                     cb_fiber.reset(new Fiber(task.cb));                 }                 task.reset();                 cb_fiber->swapIn();                 --m_activeThreadCount;                 if(cb_fiber->getState() == Fiber::READY) {                     schedule(task.fiber);                 } else if(cb_fiber->getState() != Fiber::TERM                     && cb_fiber->getState() != Fiber::EXCEPT) {                     cb_fiber->setState(Fiber::HOLD);                 }                 // 让fiber资源尽早释放，不用等到下个cb来的时候再释放                 cb_fiber.reset();             } else {                 // 进到这个分支情况一定是任务队列空了，调度idle协程即可                 if (idle_fiber->getState() == Fiber::TERM) {                     // 如果调度器没有调度任务，那么idle协程会不停地切换调度协程和idle协程，不会结束，                     //如果idle协程结束了，那一定是调度器停止了，在stop函数中控制                     SYLAR_LOG_DEBUG(g_logger) << "idle fiber term";                     break;                 }                  ++m_idleThreadCount;                 idle_fiber->swapIn();                 --m_idleThreadCount;                 if(idle_fiber->getState() != Fiber::TERM                     && idle_fiber->getState() != Fiber::EXCEPT) {                     idle_fiber->setState(Fiber::HOLD);                 }             }         }         // SYLAR_LOG_DEBUG(g_logger) << t_scheduler_fiber.use_count();         // t_scheduler_fiber = nullptr;     } 

• stop() 停止调度器

void Scheduler::stop() {         SYLAR_LOG_DEBUG(g_logger) << "stop";         if (stopping()) {             return;         }         // 这里控制会控制idle协程的退出         m_stopping = true;         /// 只能由创建线程发起scheduler的线程发起stop         // 当m_useCaller为true时，创建线程发起scheduler的线程中GetThis() == this         // 当m_useCaller为false时，创建线程发起scheduler的线程中GetThis() == nullptr         if (m_useCaller) {             SYLAR_ASSERT(GetThis() == this);         } else {             SYLAR_ASSERT(GetThis() != this);         }          for (size_t i = 0; i < m_thread_nums; i++) {             tickle();         }          // 这里的主要目的，use_caller是ture的时候，在这里让caller的调度协程进行任务调度         // 如果将这里删除，那么caller线程永远不会进行任务调度         if (m_scheduleFiber) {             tickle();             m_scheduleFiber->swapIn();         }          // 目的是防止调度线程先执行完，销毁了Scheduler对象         // 即使使用detach也是不可以的，这里涉及到多个线程使用Scheduler对象，如果线程还没退出对象销毁了，这个时候就会造成段错误         std::vector thrs;         {             MutexType::Lock lock(m_lock);             thrs.swap(m_threads);         }         for (auto &i : thrs) {             i->join();         }         SYLAR_LOG_DEBUG(g_logger) << t_scheduler_fiber.use_count();     }