• Linux 64位 C++协程池原理分析及代码实现

    导语

    本文介绍了协程的作用、结构、原理，并使用C++和汇编实现了64位系统下的协程池。文章内容避免了协程晦涩难懂的部分，用大量图文来分析原理，适合新手阅读学习。

    GitHub源码

    1. Web服务器问题

    现代分布式Web后台服务逻辑通常由一系列RPC请求组成，若串行则耗时比较长。

    此时一般都会使用线程池并行运行RPC请求，如图中GetData函数

    假设请求数据包不大，那么可假设GetData耗时组成如下图所示。在非阻塞读情况下，CPU将在Wait环节空转浪费资源(不断地read，得到返回码-1)。

    2. 协程的引入

    有没有办法只用一个线程并行执行GetData呢？答案是：可以！我们假设有3个并行的GetData任务，下图线程1通过跳转控制流，减少CPU资源浪费。执行流为①~⑦，在Wait阶段则跳到其他任务如①~⑤。运行结束后也跳到其他任务如⑥~⑦。通过这种方式，3个GetData能用一个线程以52ms的耗时并行执行。

    如果GetData任务可以被这样分配，则可以减少线程切换的消耗。因为协程的调度是线程内用户态执行的，CPU消耗非常小。

    相关视频推荐

    协程！协程！协程！给你一个吊打面试官的机会！

    用ntyco来解决，大块数据传输，连续包处理接收

    协程在 reactor 网络模型中的应

    免费学习地址：c/c++ linux服务器开发/后台架构师

    需要C/C++ Linux服务器架构师学习资料加qun812855908获取（资料包括C/C++，Linux，golang技术，Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，ffmpeg等），免费分享

    3. 协程的原理

    从上文可知，协程之间的切换本质是函数的跳转，即如何让正在执行的函数跳转到另一个新的函数上，以及下次如何又跳转回来。如下面代码所示:

    1. void func1() {
    2.    printf("① 跳转到func2");
    3.    Coroutine::CoYield(); // 通过该函数跳到func2
    4.    printf("③ func2跳转回func1");
    5. }
    6.  
    7. void func2() {
    8.    printf("② func2执行完毕");
    9. }

    要实现这种能力，需要结合汇编知识。首先研究如下简单函数的汇编语言

    1. #include <iostream>
    2. using namespace std;
    3.  
    4. class Object {
    5. public:    
    6.    int val[12];
    7. };
    8.  
    9. int func(Object *pObj1, Object *pObj2) {     
    10.    pObj1->val[0] = 1;
    11.    pObj1->val[11] = 11;
    12.    pObj2->val[0] = 2;
    13.    pObj2->val[11] = 12;
    14.    int arr[100];
    15.    arr[0] = 3;
    16.    arr[99] = 99;
    17.    return pObj1->val[0];
    18. }
    19.  
    20. int main() {
    21.    Object obj, obj2;
    22.    int a = func(&obj, &obj2);
    23.    return a;
    24. }

    下面看看在64位系统汇编中，func函数是如何执行的。

    push %rbp是进入func函数执行的第一个指令，作用是把rbp的地址压到栈顶。因为rsp始终指向栈顶，所以压栈后，rsp的地址下移8字节。rdi和rsi相差48个字节，该空间被class Object内的int val[12]占用。

    前两个指令让rbp指向rsp往下296字节的位置。后面两个指令把rdi和rsi地址保存在最下面。

    为什么rsp下移296字节？首先，上述代码使用了临时变量int arr[100]，需要有400个字节的栈空间；其次，x64系统存有128字节的红色区域可使用；最后，rdi和rsi地址共占16字节。因此，rbp到红色区域底部的空间一共是 288 + 8 + 104 + 8 + 8 = 416字节。

    接下来才开始执行func函数第一行代码，给val[0]赋值。

    然后分别给pObj1和pObj2的成员变量赋值

    接下来给临时变量arr赋值

    最后让eax指向返回值，恢复函数栈的栈底和栈顶。

    4. 协程的结构

    从前面我们知道，每个函数在内存中都有栈顶rsp和栈底rbp。这两个值决定了函数可操作的内存范围，如下图所示

    既然协程切换是从一个函数切换到另一个函数，那么就需要知道两个函数的rbp和rsp。然而，函数的rbp和rsp是执行时设定的，代码层面难以获得。
    既然如此，我们可以实现腾出空间，让函数在预期的rbp和rsp内。定义一个类如下：

    1. class Coroutine {
    2.     void* m_pRegister[14];
    3.     char m_pStack[1024];
    4.     std::function<void()> m_func;
    5. };

    那么在内存模型中，该类的布局如下所示

    这样的协程在能被使用前需要做初始化，如下图所示

    在其他协程切换过来时，cpu寄存器可按m_pRegister预设的地址赋值，开始执行DoWork函数，函数代码如下：

    1. static void Coroutine::DoWork(Coroutine *pThis) {
    2.     pThis->m_func();
    3.     pThis->Yield(); // 转让控制流给同线程的其他协程
    4. }

    由于是静态函数，需令参数pThis为协程地址。所以，初始化时需要设置m_pRegister中的rdi为this。上述第二行代码执行时，rbp会设为this。所以执行m_func时，如下图所示：

    5. 协程间的切换

    下面以Coroutine1切换到Coroutine2为例。主要分为两步：
    1. 保存Coroutine1的上下文

    2. 加载Coroutine2的上下文

    切换代码可见源代码Coroutine::Switch

    6. 协程池的实现

    本文实现协程池比较简单，初始化创建线程并设置thread_local变量以保存协程队列状态。并且，每个线程额外创建一个main协程用作Guard。

    在执行时，每个线程通过轮询的方式切换协程，若协程无任务则尝试CAS获取Job，否则直接执行已有Job。当Job执行完或主动CoYield时，切换到下一个协程。

    为了避免CAS空转，在没有任务时会阻塞休眠。当任务来临时则Notify所有线程的协程。

    7. 源代码

    example.cpp

    1. /**
    2.  * @file example.cpp
    3.  * @author souma
    4.  * @brief 使用协程池的示例，编译命令如下
    5.  * g++ example.cpp coroutine.cpp -lpthread -O3
    6.  * @version 0.1
    7.  * @date 2023-06-06
    8.  * 
    9.  * @copyright Copyright (c) 2023
    10.  * 
    11.  */
    12. #include <iostream>
    13. #include <array>
    14. #include "coroutine.h"
    15.  
    16. using namespace std;
    17. using namespace comm;
    18.  
    19. void func(const string &sTaskName, uint32_t uWaitSeconds) {
    20.     printf("[%ld] [%s start], wait seconds[%u]\n", time(nullptr), sTaskName.c_str(), uWaitSeconds);
    21.     time_t iStartSec = time(nullptr);
    22.     // 默认可用65535字节的栈内存，具体可看CO_STACK_SIZE
    23.     uint32_t uArrSize = 65535/4;
    24.     int arr[uArrSize];
    25.     while (time(nullptr) - iStartSec < uWaitSeconds) {
    26.         // 操作栈内存
    27.         for (int i = 0; i < uArrSize; ++i) {
    28.             arr[i] = i;
    29.         }
    30.  
    31.         // 切换控制流
    32.         printf("[%ld] [%s] -> [协程池]\n", time(nullptr), sTaskName.c_str());
    33.         usleep(100);
    34.         Coroutine::CoYield(); // 只需这一个函数即可切换控制流
    35.         printf("[%ld] [协程池] -> [%s]\n", time(nullptr), sTaskName.c_str());
    36.     }
    37.  
    38.     // 检查栈内存是否正确
    39.     for (int i = 0; i < uArrSize; ++i) {
    40.         if (arr[i] != i) {
    41.             printf("栈内存错误\n");
    42.             exit(-1);
    43.         }
    44.     }
    45.     printf("[%ld] [%s end], expect_timecost[%d], real_timecost[%ld]\n", time(nullptr), sTaskName.c_str(), uWaitSeconds, time(nullptr) - iStartSec);
    46. }
    47.  
    48. int main() {
    49.     // 如果想当线程池用，可以令第一个参数为线程数，第二个参数为1。
    50.     // 在该场景下，使用小线程大协程不仅CPU消耗低，整体耗时也很低，可以自行测试。
    51.     CoroutinePool oPool(2, 300);
    52.     oPool.Run();
    53.  
    54.     time_t iStartTime = time(nullptr);
    55.     const int iTaskCnt = 400;
    56.     vector<shared_ptr<Future>> vecFuture;
    57.     for (int i = 0; i < iTaskCnt; ++i) {
    58.         // 模拟GetData中的Wait环节, 1 ~ 5秒等待
    59.         shared_ptr<Future> pFuture = oPool.Submit([i](){func("Task" + to_string(i), random() % 5 + 1);});
    60.         if (pFuture != nullptr) {
    61.             vecFuture.emplace_back(pFuture);
    62.         }
    63.     }
    64.     
    65.     // 阻塞等待所有Task完成
    66.     for (auto it = vecFuture.begin(); it != vecFuture.end(); ++it) {
    67.         (*it)->Get();
    68.     }
    69.  
    70.     printf("demo's finished, time cost[%ld]\n", time(nullptr) - iStartTime);
    71.     return 0;
    72. }

    coroutine.h

    1. /**
    2.  * @file coroutine.h
    3.  * @author souma
    4.  * @brief 多线程无栈式协程池，请不要用-O0编译否则会产生coredump
    5.  * @version 0.1
    6.  * @date 2023-06-06
    7.  * 
    8.  * @copyright Copyright (c) 2023
    9.  * 
    10.  */
    11. #pragma once
    12. #include <functional>
    13. #include <memory>
    14. #include <vector>
    15. #include <queue>
    16. #include <thread>
    17. #include <mutex>
    18. #include <signal.h>
    19. #include <pthread.h>
    20. #include <condition_variable>
    21. #include <unistd.h>
    22.  
    23. namespace comm {
    24.     class Future;
    25.     class CoroutinePool;
    26.     class Coroutine;
    27.     struct CoroutinePoolCtx;
    28.     struct CoroutineTaskCtx;
    29.  
    30.  
    31.     struct CoroutinePoolCtx {
    32.         std::vector<std::shared_ptr<Coroutine>> m_vecCoroutine;
    33.         std::shared_ptr<Coroutine> m_pMainCoroutine;
    34.         uint32_t m_uCursor;
    35.         uint32_t m_uWorkCnt;
    36.     };
    37.  
    38.     struct CoroutineTaskCtx {
    39.         std::function<void()> m_userFunc;
    40.         std::shared_ptr<Future> m_pFuture;
    41.     };
    42.  
    43.     // class ArraySyncQueue start
    44.     template <class T>
    45.     class ArraySyncQueue {
    46.     public:
    47.         ArraySyncQueue(uint32_t uCapacity, uint32_t uSleepUs = 100, uint32_t uRetryTimes = 3);
    48.         bool Push(T *pObj);
    49.         T* Pop();
    50.         inline bool IsFull() const { return m_uPushCursor == m_uPopCursor - 1 || (m_uPopCursor == 0 && m_uPushCursor == m_vecQueue.size() - 1); }
    51.         bool IsEmpty() const { return m_uPopCursor == m_uPushCursor; }
    52.  
    53.         ~ArraySyncQueue();
    54.  
    55.     private:
    56.         uint32_t GetNextCursor(uint32_t uCursor);
    57.     private:
    58.         std::vector<T*> m_vecQueue;
    59.         uint32_t m_uPushCursor = 0;
    60.         uint32_t m_uPopCursor = 0;
    61.         uint32_t m_uSleepUs;
    62.         uint32_t m_uRetryTimes;
    63.     };
    64.     // class ArraySyncQueue end
    65.  
    66.     // class Coroutine start
    67.     class Coroutine {
    68.     public:
    69.         
    70.         friend class CoroutinePool;
    71.  
    72.         /**
    73.          * @brief 调用该函数将执行流交给其他协程，仅在协程池环境下有效
    74.          * 
    75.          * @return true:协程切换成功, false:不在协程池环境中运行
    76.          */
    77.         static bool CoYield();
    78.         
    79.         Coroutine(const Coroutine &) = delete;
    80.         Coroutine(Coroutine &&) = delete;
    81.         Coroutine & operator=(const Coroutine &) = delete;
    82.         Coroutine & operator=(Coroutine &&) = delete;
    83.  
    84.     private:
    85.         // 4096是预留给库使用的栈内存大小，后者是留给用户使用的栈内存大小
    86.         constexpr static uint32_t CO_STACK_SIZE = 4096 + 65535; 
    87.  
    88.         Coroutine();
    89.  
    90.         /**
    91.          * @brief 当前协程是否绑定了任务
    92.          *
    93.          * @return true:是
    94.          */
    95.         inline bool HasTask() const { return m_pTaskCtx != nullptr; }
    96.  
    97.         /**
    98.          * @brief 两个协程切换，从pPrev切换到pNext
    99.          */
    100.         static void Switch(Coroutine *pPrev, Coroutine *pNext);
    101.  
    102.         /**
    103.          * @brief 将控制流转给同线程的其他协程
    104.          */
    105.         void Yield();
    106.  
    107.         /**
    108.          * @brief 这个是给main协程用的
    109.          */
    110.         void Register();
    111.  
    112.         /**
    113.          * @brief 这个是给执行用户任务的协程用的
    114.          */
    115.         void Register(std::shared_ptr<CoroutineTaskCtx> pTaskCtx);
    116.  
    117.         /**
    118.          * @return CoroutinePoolCtx& 当前线程的协程上下文
    119.          */
    120.         static CoroutinePoolCtx & GetCtx();
    121.  
    122.         /**
    123.          * @brief 让当前线程的cursor往后移，轮询协程
    124.          */
    125.         static void MoveCursor();
    126.         
    127.         /**
    128.          * @brief 协程包一层的函数
    129.          */
    130.         static void DoWork(Coroutine *pThis);
    131.  
    132.         /**
    133.          * 
    134.          * @return void* 获得自建rsp地址
    135.          */
    136.         void* GetRsp();
    137.  
    138.         /**
    139.          * 保存寄存器的值到m_pStack中
    140.          */
    141.         void SaveReg();
    142.  
    143.     private:
    144.         void* m_pRegister[14];
    145.         char m_pStack[CO_STACK_SIZE];
    146.         std::shared_ptr<CoroutineTaskCtx> m_pTaskCtx;
    147.     };
    148.     // class Coroutine end
    149.  
    150.     // class CoroutinePool start
    151.     class CoroutinePool {
    152.     public:
    153.         friend class Coroutine;
    154.         /**
    155.          * @brief 建立一个多线程协程池，即创建uThreadCnt个线程，每个线程含有uCoroutineCnt个协程
    156.                   调用Run开始运行，调用Stop或直接析构结束
    157.          * @param uThreadCnt 线程数，小于1则为1
    158.          * @param uCoroutineCnt 每个线程的协程数，小于1则为1
    159.          * @param uJobQueueSize 总任务队列大小，小于1则为1
    160.          */
    161.         CoroutinePool(uint32_t uThreadCnt, uint32_t uCoroutineCnt, uint32_t uJobQueueSize = 1024000);
    162.  
    163.         /**
    164.          * @brief 线程安全，可重入
    165.          * @return true:正常
    166.          */
    167.         bool Run();
    168.  
    169.         /**
    170.          * @brief 停止协程池 (会先保证池中任务完成再停止)，线程安全可重入
    171.          * 
    172.          */
    173.         void Stop();
    174.  
    175.         /**
    176.          * @param userFunc 用户函数
    177.          * @return std::shared_ptr<Future>  nullptr:协程池队列满了，提交不了
    178.          */
    179.         std::shared_ptr<Future> Submit(const std::function<void()> &userFunc);
    180.  
    181.         ~CoroutinePool();
    182.         CoroutinePool(const CoroutinePool &) = delete;
    183.         CoroutinePool(CoroutinePool &&) = delete;
    184.         CoroutinePool & operator=(const CoroutinePool &) = delete;
    185.         CoroutinePool & operator=(CoroutinePool &&) = delete;
    186.  
    187.     private:
    188.         static void LoopWork(CoroutinePool &oPool);
    189.  
    190.     private:
    191.         bool m_bStarted;
    192.         uint32_t m_uThreadCnt;
    193.         uint32_t m_uRoutineCnt;
    194.         ArraySyncQueue<CoroutineTaskCtx> m_queueJob;
    195.         std::vector<std::shared_ptr<std::thread>> m_vecThread;
    196.         std::mutex m_oMutex;
    197.         std::condition_variable m_oCondition;
    198.     };
    199.     // class CoroutinePool end
    200.  
    201.     // class Future start
    202.     class Future {
    203.     public:
    204.         /**
    205.         * @brief 阻塞获得结果
    206.         * 
    207.         * @param uTimeoutMs 超时时间
    208.         * @return true:成功， false:超时
    209.         */
    210.         bool Get(uint32_t uTimeoutMs = -1);
    211.  
    212.         /**
    213.         * @brief 设置状态为完成
    214.         */
    215.         void SetFinished();
    216.  
    217.         Future();
    218.  
    219.         Future(const Future&) = delete;
    220.         Future(Future&&) = delete;
    221.  
    222.         Future & operator=(const Future&) = delete;
    223.         Future & operator=(Future&&) = delete;
    224.  
    225.     private:
    226.         std::mutex m_oMutex;
    227.         std::condition_variable m_oCondition;
    228.         bool m_bFinished;
    229.     };
    230.     // class Future end
    231. }

    coroutine.cpp

    1. /**
    2.  * @file coroutine.cpp
    3.  * @author souma
    4.  * @brief 协程池的具体实现
    5.  * @version 0.1
    6.  * @date 2023-06-06
    7.  * 
    8.  * @copyright Copyright (c) 2023
    9.  * 
    10.  */
    12. #include "coroutine.h"
    13. #include <cstring>
    14.  
    15. using namespace std;
    16. namespace comm {
    17.  
    18.     // class Coroutine start
    19.     Coroutine::Coroutine() {
    20.         m_pTaskCtx = nullptr;
    21.     }
    22.  
    23.     void Coroutine::Register() {
    24.         m_pTaskCtx = make_shared<CoroutineTaskCtx>();
    25.         m_pTaskCtx->m_userFunc = [](){};
    26.         m_pTaskCtx->m_pFuture = nullptr;
    27.         SaveReg();
    28.     }
    29.  
    30.     void Coroutine::Register(shared_ptr<CoroutineTaskCtx> pTaskCtx) {
    31.         m_pTaskCtx = pTaskCtx;
    32.         SaveReg();
    33.     }
    34.  
    35.     inline void Coroutine::Yield() {
    36.         Coroutine::Switch(this, Coroutine::GetCtx().m_pMainCoroutine.get());
    37.     }
    38.  
    39.     bool Coroutine::CoYield() {
    40.         if (GetCtx().m_vecCoroutine.size() == 0) {
    41.             return false;
    42.         }
    43.         GetCtx().m_vecCoroutine[GetCtx().m_uCursor]->Yield();
    44.         return true;
    45.     }
    46.  
    47.     CoroutinePoolCtx & Coroutine::GetCtx() {
    48.         thread_local CoroutinePoolCtx coroutinePoolCtx;
    49.         return coroutinePoolCtx;
    50.     }
    51.  
    52.     void Coroutine::MoveCursor() {
    53.         GetCtx().m_uCursor = GetCtx().m_uCursor == GetCtx().m_vecCoroutine.size() - 1 ? 0 : GetCtx().m_uCursor + 1;
    54.     }
    55.  
    56.     extern "C" __attribute__((noinline, weak))
    57.     void Coroutine::Switch(Coroutine *pPrev, Coroutine *pNext) {
    58.         // 1.保存pPrev协程的上下文, rdi和pPrev同指向
    59.         // 2.加载pNext协程的上下文, rsi和pNext同指向
    60.         asm volatile(R"(
    61.             movq %rsp, %rax
    62.             movq %rbp, 104(%rdi)
    63.             movq %rax, 96(%rdi)
    64.             movq %rbx, 88(%rdi)
    65.             movq %rcx, 80(%rdi)
    66.             movq %rdx, 72(%rdi)
    67.             movq 0(%rax), %rax
    68.             movq %rax, 64(%rdi)
    69.             movq %rsi, 56(%rdi)
    70.             movq %rdi, 48(%rdi)
    71.             movq %r8, 40(%rdi)
    72.             movq %r9, 32(%rdi)
    73.             movq %r12, 24(%rdi)
    74.             movq %r13, 16(%rdi)
    75.             movq %r14, 8(%rdi)
    76.             movq %r15, (%rdi)
    78.             movq (%rsi), %r15
    79.             movq 8(%rsi), %r14
    80.             movq 16(%rsi), %r13
    81.             movq 24(%rsi), %r12
    82.             movq 32(%rsi), %r9
    83.             movq 40(%rsi), %r8
    84.             movq 48(%rsi), %rdi
    85.             movq 64(%rsi), %rax
    86.             movq 72(%rsi), %rdx
    87.             movq 80(%rsi), %rcx
    88.             movq 88(%rsi), %rbx
    89.             movq 96(%rsi), %rsp
    90.             movq 104(%rsi), %rbp
    91.             movq 56(%rsi), %rsi
    92.             movq %rax, (%rsp)
    93.             xorq %rax, %rax
    94.         )");
    95.     }
    96.  
    97.     void Coroutine::DoWork(Coroutine *pThis) {
    98.         pThis->m_pTaskCtx->m_userFunc();
    99.         pThis->m_pTaskCtx->m_pFuture->SetFinished();
    100.         pThis->m_pTaskCtx.reset();
    101.         Coroutine::GetCtx().m_uWorkCnt--;
    102.         pThis->Yield();
    103.     }
    104.  
    105.     void* Coroutine::GetRsp() {
    106.         // m_pRegister和m_pStack中间预留一个指针空间
    107.         auto sp = std::end(m_pStack) - sizeof(void*);
    108.         // 预定Rsp的地址保证能够整除8字节
    109.         sp = decltype(sp)(reinterpret_cast<size_t>(sp) & (~0xF));
    110.         return sp;
    111.     }
    112.  
    113.     void Coroutine::SaveReg() {
    114.         void *pStack = GetRsp();
    115.         memset(m_pRegister, 0, sizeof m_pRegister);
    116.         void **pRax = (void**)pStack;
    117.         *pRax = (void*) DoWork;
    118.         // rsp
    119.         m_pRegister[12] = pStack;
    120.         // rax
    121.         m_pRegister[8] = *pRax;
    122.         // rdi
    123.         m_pRegister[6] = this;
    124.     }
    125.     // class Coroutine end
    126.  
    127.     // class CoroutinePool start
    128.     CoroutinePool::CoroutinePool(uint32_t uThreadCnt, uint32_t uCoroutineCnt, uint32_t uJobQueueSize) : m_queueJob(uJobQueueSize) {
    129.         m_bStarted = false;
    130.         m_uThreadCnt = max(uThreadCnt, 1u);
    131.         m_uRoutineCnt = max(uCoroutineCnt, 1u);
    132.     }
    133.  
    134.     bool CoroutinePool::Run() {
    135.         if (!__sync_bool_compare_and_swap(&m_bStarted, false, true)) {
    136.             return false;
    137.         }
    138.         
    139.         for (decltype(m_uThreadCnt) i = 0; i < m_uThreadCnt; ++i) {
    140.             m_vecThread.emplace_back(make_shared<thread>(CoroutinePool::LoopWork, ref(*this)));   
    141.         }
    142.         return true;
    143.     }
    144.  
    145.     void CoroutinePool::Stop() {
    146.         if (!__sync_bool_compare_and_swap(&m_bStarted, true, false)) {
    147.             return;
    148.         }
    149.         
    150.         m_oCondition.notify_all();
    151.         for (auto it = m_vecThread.begin(); it != m_vecThread.end(); ++it) {
    152.             (*it)->join();
    153.         }
    154.         m_vecThread.clear();
    155.     }
    156.  
    157.     shared_ptr<Future> CoroutinePool::Submit(const function<void()> &userFunc) {
    158.         shared_ptr<Future> pNewFuture = make_shared<Future>();
    159.         CoroutineTaskCtx *pTaskCtx = new CoroutineTaskCtx;
    160.         pTaskCtx->m_pFuture = pNewFuture;
    161.         pTaskCtx->m_userFunc = userFunc;
    162.  
    163.         if (!m_queueJob.Push(pTaskCtx)) {
    164.             delete pTaskCtx, pTaskCtx = nullptr;
    165.             return nullptr;
    166.         }
    167.         m_oCondition.notify_all();
    168.         return pNewFuture;
    169.     }
    170.  
    171.     CoroutinePool::~CoroutinePool() {
    172.         Stop();
    173.     }
    174.  
    175.     void CoroutinePool::LoopWork(CoroutinePool &oPool) {
    176.         Coroutine::GetCtx().m_uCursor = 0;
    177.         Coroutine::GetCtx().m_uWorkCnt = 0;
    178.         Coroutine::GetCtx().m_pMainCoroutine = shared_ptr<Coroutine>(new Coroutine);
    179.         Coroutine::GetCtx().m_pMainCoroutine->Register();
    180.  
    181.         Coroutine::GetCtx().m_vecCoroutine.clear();
    182.         for (decltype(oPool.m_uRoutineCnt) i = 0; i < oPool.m_uRoutineCnt; ++i) {
    183.             Coroutine::GetCtx().m_vecCoroutine.emplace_back(shared_ptr<Coroutine>(new Coroutine));
    184.         }
    185.  
    186.         Coroutine *pMainCoroutine, *pCurCoroutine;
    187.         while (oPool.m_bStarted || Coroutine::GetCtx().m_uWorkCnt > 0 || !oPool.m_queueJob.IsEmpty()) {
    188.             
    189.             pMainCoroutine = Coroutine::GetCtx().m_pMainCoroutine.get();
    190.             pCurCoroutine = Coroutine::GetCtx().m_vecCoroutine[Coroutine::GetCtx().m_uCursor].get();
    191.             
    192.             if (pCurCoroutine->HasTask()) {
    193.                 Coroutine::Switch(pMainCoroutine, pCurCoroutine);
    194.                 Coroutine::MoveCursor();
    195.                 continue;
    196.             }
    197.  
    198.             CoroutineTaskCtx *pTaskCtx = oPool.m_queueJob.Pop();
    199.             if (pTaskCtx == nullptr) {
    200.                 if (Coroutine::GetCtx().m_uWorkCnt > 0) {
    201.                     Coroutine::MoveCursor();
    202.                     continue;
    203.                 }
    204.                 unique_lock<mutex> oLock(oPool.m_oMutex);
    205.                 oPool.m_oCondition.wait(oLock);
    206.                 continue;
    207.             }
    208.  
    209.             pCurCoroutine->Register(shared_ptr<CoroutineTaskCtx>(pTaskCtx));
    210.             ++Coroutine::GetCtx().m_uWorkCnt;
    211.             Coroutine::Switch(pMainCoroutine, pCurCoroutine);
    212.             Coroutine::MoveCursor();
    213.         }
    214.     }
    215.     // class CoroutinePool end
    216.  
    217.     // class Future start
    218.     Future::Future() {
    219.         m_bFinished = false;
    220.     }
    221.  
    222.     bool Future::Get(uint32_t uTimeoutMs) {
    223.         unique_lock<mutex> oLock(m_oMutex);
    224.         if (m_bFinished) {
    225.             return true;
    226.         }
    227.         return m_oCondition.wait_for(oLock, chrono::milliseconds(uTimeoutMs)) == cv_status::no_timeout;
    228.     }
    229.  
    230.     void Future::SetFinished() {
    231.         {
    232.             unique_lock<mutex> oLock(m_oMutex);
    233.             m_bFinished = true;
    234.         }
    235.         m_oCondition.notify_all();
    236.     }
    237.     // class Future end
    238.  
    239.     // class ArraySyncQueue start
    240.     template <class T>
    241.     ArraySyncQueue<T>::ArraySyncQueue(uint32_t uCapacity, uint32_t uSleepUs, uint32_t uRetryTimes) {
    242.         for (uint32_t i = 0; i < std::max(uCapacity, 1u); ++i) {
    243.             m_vecQueue.emplace_back(nullptr);
    244.         }
    245.         m_uSleepUs = uSleepUs;
    246.         m_uRetryTimes = uRetryTimes;
    247.     }
    248.  
    249.     template <class T>
    250.     bool ArraySyncQueue<T>::Push(T *pObj) {
    251.         if (pObj == nullptr) {
    252.             return false;
    253.         }
    254.         uint32_t uRetryTimes = 0;
    255.         while (uRetryTimes <= m_uRetryTimes) {
    256.             uint32_t uPushCursor = m_uPushCursor;
    257.             if (uPushCursor == m_uPopCursor - 1 || (m_uPopCursor == 0 && uPushCursor == m_vecQueue.size() - 1)) {
    258.                 // 队列满了
    259.                 return false;
    260.             }
    261.  
    262.             if (!__sync_bool_compare_and_swap(&m_vecQueue[uPushCursor], nullptr, pObj)) {
    263.                 uRetryTimes++;
    264.                 usleep(m_uSleepUs);
    265.                 continue;
    266.             }
    267.  
    268.             m_uPushCursor = GetNextCursor(uPushCursor);
    269.             return true;
    270.         }
    271.         // 竞争失败
    272.         return false;
    273.     }
    274.  
    275.     template <class T>
    276.     T* ArraySyncQueue::Pop() {
    277.         uint32_t uRetryTimes = 0;
    278.         while (uRetryTimes <= m_uRetryTimes) {
    279.             uint32_t uPopCursor = m_uPopCursor;
    280.             if (uPopCursor == m_uPushCursor) {
    281.                 return nullptr;
    282.             }
    283.  
    284.             T* pToReturn = m_vecQueue[uPopCursor];
    285.             if (pToReturn == nullptr || !__sync_bool_compare_and_swap(&m_vecQueue[uPopCursor], pToReturn, nullptr)) {
    286.                 usleep(m_uSleepUs);
    287.                 uRetryTimes++;
    288.                 continue;
    289.             }
    290.             m_uPopCursor = GetNextCursor(uPopCursor);
    291.             return pToReturn;
    292.         }
    293.         return nullptr;
    294.     }
    295.  
    296.     template <class T>
    297.     uint32_t ArraySyncQueue<T>::GetNextCursor(uint32_t uCursor) {
    298.         if (uCursor == m_vecQueue.size() - 1) {
    299.             return 0;
    300.         }
    301.         return uCursor + 1;
    302.     }
    303.  
    304.     template <class T>
    305.     ArraySyncQueue<T>::~ArraySyncQueue() {
    306.         m_uRetryTimes = -1;
    307.         do {
    308.             T *pObj = Pop();
    309.             if (pObj == nullptr) {
    310.                 return;
    311.             }
    312.             delete pObj, pObj = nullptr;
    313.         } while (true);
    314.     }
    315.     // class ArraySyncQueue end
    316. }

    8. 补充说明

    8.1. 为什么不能-O0编译?

    在-O0的情况下，编译器会给函数(coroutine.cpp:57)Coroutine::Switch包一层汇编指令，导致实际执行汇编指令不是期望的。具体可以分别用-O0和-O3在GDB下disassemble看到差异。

    8.2. 如果函数使用栈很大怎么办?

    源码中定义的协程栈为CO_STACK_SIZE=4096 + 65535KB，若用户函数使用的栈超过该范围会产生coredump。简单可行的解法是：1.尽量使用堆变量；2.改大CO_STACK_SIZE。

  • 相关阅读:
    SpringCloud微服务实战——搭建企业级开发框架（三十六）：使用Spring Cloud Stream实现可灵活配置消息中间件的功能
    检查OpenGL的版本
    14：00面试，14：06就出来了，问的问题有点变态。。。
    用C#实现简单的线性回归
    业内专业人士揭秘：双11即将来临，挑选SSD硬盘避坑指南
    基于云的 LDAP 目录服务和本地 LDAP 相比有哪些优势？
    Numpy-01（安装、Ndarray对象介绍创建及使用）
    无法更新chrome
    js书写规范
    项目实战-智慧监督下的合同预付款控制策略-物料价格下行-智慧监督-合同预付款预警推送大数据
  • 原文地址：https://blog.csdn.net/qq_40989769/article/details/133827752