Gentoo中文社区

batsom

管理团队

注册时间： 2022-08-03

帖子： 594

个人网站

linux源码解读（二十八）：通过epoll实现协程（二）

1、协程只是一种思路，并且没有操作系统层面的参与，所以全靠3环的应用开发人员自己实现。市面上有各种协程框架，这里以微信的libco库为例，看看协程到底是怎么落地实现的！libco 是微信后台开发和使用的协程库，号称可以调度千万级协程；从使用上来说，libco 不仅提供了一套类 pthread 的协程通信机制，同时可以零改造地将三方库的阻塞 IO 调用进行协程化；正式介绍libco的源码前，先直观感受一下libco的效果，demo代码如下：

void A() {
   cout << 1 << " ";
   cout << 2 << " ";
   cout << 3 << " ";
}

void B() {
   cout << "x" << " ";
   cout << "y" << " ";
   cout << "z" << " ";
}

int main(void) {
  A();
  B();
}

这个代码很简单，刚开始学编程的人都能看懂，结果如下：

1 2 3 x y z

  如果用libco的协程api在A和B函数之间切换（注意这是简化后的伪代码，目的是抓住主干，避免被细枝末节的代码干扰），如下：

void A() {
   cout << 1 << " ";
   cout << 2 << " ";
   co_yield_ct();  // 切出到主协程
   cout << 3 << " ";
}

void B() {
   cout << "x" << " ";
   co_yield_ct();  // 切出到主协程
   cout << "y" << " ";
   cout << "z" << " ";
}

int main(void) {
  ...  // 主协程
  co_resume(A);  // 启动协程 A
  co_resume(B);  // 启动协程 B
  co_resume(A);  // 从协程 A 切出处继续执行
  co_resume(B);  // 从协程 B 切出处继续执行
}

  这时的结果就变了：

1 2 x 3 y z

  可以看到代码在A和B函数之间来回切换执行，整个切换的顺序完全依靠co_yield_ct和co_resume两个函数人为控制！这样就实现了A函数阻塞时人为切换到B函数执行；B函数阻塞时再切换到A函数继续执行，不浪费一点CPU时间片！

     2、之前解读linux源码时，遇到某些功能时都是先看结构体，再阅读函数功能，原因很简单：重要的字段和数据都会放在结构体中统一管理（本质是能快速寻址，利于读写），函数的所有代码都是围绕结构体中这些数据读写展开的！libco重要的结构体之一就是stCoRoutine_t（从名称包含routine就能大概才出来结构体和执行的函数相关），定义如下：

/*协程跳转（也即是yield、resume）执行的函数结构体，
  包含了协程运行最重要的3要素：
  协程运行环境、执行函数的入口、函数参数
*/
struct stCoRoutine_t
{
    stCoRoutineEnv_t *env;//协程运行环境
    pfn_co_routine_t pfn;// 协程跳转执行的函数
    void *arg;//函数的参数
    coctx_t ctx;//保存协程的下文环境 

    char cStart;//协程是否开始
    char cEnd;//协程是否结束
    char cIsMain;//当前是main函数吗
    char cEnableSysHook; //是否运行系统 hook，即非侵入式逻辑
    char cIsShareStack;//多个协程之间是否共享栈
    void *pvEnv;

    //char sRunStack[ 1024 * 128 ];
    stStackMem_t* stack_mem;// 协程运行时的栈空间

    //save stack buffer while confilct on same stack_buffer;
    char* stack_sp; 
    unsigned int save_size;
    char* save_buffer;
    stCoSpec_t aSpec[1024];
};

   看吧，这个结构体几乎包含了协程子重要的几个元素：协程的调度环境、协程要运行的函数及参数，协程切换时要保存的上下文等！这些结构体之间的关系如下：

   结构体有了，接下来就是初始化这个结构体了，再co_create_env函数中做的，核心是把传入的函数入口、参数、env等变量纳入stCoRoutine_t统一管理，同时初始化栈和其他变量！

/*创建协程
env：协程跳转执行函数的结构体，也可以理解为函数的环境
pfn：协程跳转执行的函数入口
初始化stCoRoutine_t *lp结构体
*/
struct stCoRoutine_t *co_create_env( stCoRoutineEnv_t * env, const stCoRoutineAttr_t* attr,
        pfn_co_routine_t pfn,void *arg )
{

    stCoRoutineAttr_t at;
    if( attr )
    {
        memcpy( &at,attr,sizeof(at) );
    }
    if( at.stack_size <= 0 )
    {
        at.stack_size = 128 * 1024;//协程栈128k
    }
    else if( at.stack_size > 1024 * 1024 * 8 )//不能超过8M
    {
        at.stack_size = 1024 * 1024 * 8;
    }

    if( at.stack_size & 0xFFF ) 
    {
        at.stack_size &= ~0xFFF;//栈大小的低12bit清零，也就是页对齐
        at.stack_size += 0x1000;
    }

    stCoRoutine_t *lp = (stCoRoutine_t*)malloc( sizeof(stCoRoutine_t) );
    
    memset( lp,0,(long)(sizeof(stCoRoutine_t))); 

    /*stCoRoutine_t包含了协程运行最重要的3要素：环境、函数入口和参数*/
    lp->env = env;
    lp->pfn = pfn;
    lp->arg = arg;
    
    stStackMem_t* stack_mem = NULL;
    if( at.share_stack )//如果用共享内存
    {
        stack_mem = co_get_stackmem( at.share_stack);
        at.stack_size = at.share_stack->stack_size;
    }
    else//否则重新分配协程栈
    {
        stack_mem = co_alloc_stackmem(at.stack_size);
    }
    lp->stack_mem = stack_mem;

    lp->ctx.ss_sp = stack_mem->stack_buffer;
    lp->ctx.ss_size = at.stack_size;

    lp->cStart = 0;
    lp->cEnd = 0;
    lp->cIsMain = 0;
    lp->cEnableSysHook = 0;
    lp->cIsShareStack = at.share_stack != NULL;

    lp->save_size = 0;
    lp->save_buffer = NULL;

    return lp;
}

  协程结构体初始化完成后就该使用了吧，还记得文章开头的那个demo案例么？main里面直接调用的co_resume函数，如下：

void co_resume( stCoRoutine_t *co )
{
    stCoRoutineEnv_t *env = co->env;
    // 获取当前正在运行的协程的结构
    // 每次有新协程产生就放入数组管理
    stCoRoutine_t *lpCurrRoutine = env->pCallStack[ env->iCallStackSize - 1 ];
    if( !co->cStart )
    {
        // 为将要运行的 co 布置上下文环境：初始化协程的栈，并把函数参数、返回地址入栈
        coctx_make( &co->ctx,(coctx_pfn_t)CoRoutineFunc,co,0 );
        co->cStart = 1;
    }
    //需要恢复的协程加入数组，表示正在运行
    env->pCallStack[ env->iCallStackSize++ ] = co;
    co_swap( lpCurrRoutine, co );
}

   唯一的参数就是协程结构体；前面做的都是各种准备工作，最关键的就是最后一个co_swap函数了，从名字和参数看就知道是协程结构体（本质上就是函数）互相切换！

void co_swap(stCoRoutine_t* curr, stCoRoutine_t* pending_co)
{
     stCoRoutineEnv_t* env = co_get_curr_thread_env();

    //get curr stack sp
    char c;
    /*记录curr协程栈位置，后续切换回curr时可用于恢复栈内容*/
    curr->stack_sp= &c;

    if (!pending_co->cIsShareStack)
    {
        env->pending_co = NULL;
        env->occupy_co = NULL;
    }
    else //共享栈模式
    {
        env->pending_co = pending_co;
        //get last occupy co on the same stack mem
        //取出共享栈中已有的协程
        stCoRoutine_t* occupy_co = pending_co->stack_mem->occupy_co;
        //set pending co to occupy thest stack mem;
        //在共享栈中记录挂起的协程
        pending_co->stack_mem->occupy_co = pending_co;
        /*env记录pending和occupy两个协程，便于切换后仍然能找到*/
        env->occupy_co = occupy_co;
        if (occupy_co && occupy_co != pending_co)
        {
            /*换个地方保存协程*/
            save_stack_buffer(occupy_co);
        }
    }

    //swap context
    /*协程切换最核心的函数：切换通用寄存器、esp+ebp、eip;*/
    coctx_swap(&(curr->ctx),&(pending_co->ctx) );

    //stack buffer may be overwrite, so get again;
    stCoRoutineEnv_t* curr_env = co_get_curr_thread_env();
    stCoRoutine_t* update_occupy_co =  curr_env->occupy_co;
    stCoRoutine_t* update_pending_co = curr_env->pending_co;
    
    if (update_occupy_co && update_pending_co && update_occupy_co != update_pending_co)
    {
        //resume stack buffer
        if (update_pending_co->save_buffer && update_pending_co->save_size > 0)
        {
            memcpy(update_pending_co->stack_sp, update_pending_co->save_buffer, update_pending_co->save_size);
        }
    }
}

  co_swap最核心的莫过于coctx_swap了，由于涉及到寄存器操作，C语言已经无能为力，这里直接用汇编简单粗暴来做：先把curr的上下文push到curr的栈里，然后通过movq %rsi, %rsp把rsp切换到pending的栈，最后通过一系列的pop把pending的context赋值给寄存器，最后一句ret把此时栈顶的地址赋值给eip，由此完成切换！

.globl coctx_swap
.type  coctx_swap, @function
coctx_swap:
    leaq 8(%rsp),%rax  # rax=(*rsp) + 8；
                       # 此时栈顶元素是当前的%rip(即当前协程挂起后被再次唤醒时，需要执行的下一条指令
                       # 的地址),后面会把栈顶的这个地址保存到curr->ctx->regs[9]中，所以保存rsp的
                       # 时候就跳过这8个字节了
    leaq 112(%rdi),%rsp#rsp=(*rdi) + (8*14);
                       # %rdi存放的是函数第一个参数的地址，即curr->ctx的地址
                       # 然后加上需要保存的14个寄存器的长度，使rsp指向curr->ctx->regs[13]
    pushq %rax         # curr->ctx->regs[13] = rax;
                       # 保存rsp，看第一行代码的注释
    pushq %rbx         # curr->ctx->regs[12] = rbx;
    pushq %rcx         # curr->ctx->regs[11] = rcx;
    pushq %rdx         # curr->ctx->regs[10] = rcx;
    pushq -8(%rax)     # curr->ctx->regs[9] = (*rax) - 8;
                       # 把协程挂起后被再次唤醒时，需要执行的下一条指令的地址保存起来
    pushq %rsi         # curr->ctx->regs[8] = rsi;
    pushq %rdi         # curr->ctx->regs[7] = rdi;
    pushq %rbp         # curr->ctx->regs[6] = rbp;
    pushq %r8          # curr->ctx->regs[5] = r8;
    pushq %r9          # curr->ctx->regs[4] = r9;
    pushq %r12         # curr->ctx->regs[3] = r12;
    pushq %r13         # curr->ctx->regs[2] = r13;
    pushq %r14         # curr->ctx->regs[1] = r14;
    pushq %r15         # curr->ctx->regs[0] = r15;

    movq %rsi, %rsp    # rsp = rsi; 
                       # rsi中存放的是函数的第二个参数的地址，即使rsp指向pending_co->ctx->regs[0]
    popq %r15          # r15 = pending_co->ctx->regs[0];
    popq %r14          # r14 = pending_co->ctx->regs[1];
    popq %r13          # r13 = pending_co->ctx->regs[2];
    popq %r12          # r12 = pending_co->ctx->regs[3];
    popq %r9           # r9 = pending_co->ctx->regs[4];
    popq %r8           # r8 = pending_co->ctx->regs[5];
    popq %rbp          # rbp = pending_co->ctx->regs[6];
    popq %rdi          # rdi = pending_co->ctx->regs[7];
    popq %rsi          # rsi = pending_co->ctx->regs[8];
    popq %rax          # rax = pending_co->ctx->regs[9];
                       # 对照前面，ctx->regs[9]中存放的是协程被唤醒后需要执行的下一条指令的地址
    popq %rdx          # rdx = pending_co->ctx->regs[10]; 
    popq %rcx          # rcx = pending_co->ctx->regs[11];
    popq %rbx          # rbx = pending_co->ctx->regs[12];
    popq %rsp          # rsp = pending_co->ctx->regs[13]; rsp += 8；
                       # 这句代码是理解整个过程的关键。和coctx_make函数中保存rsp时减8再保存相对应。
    pushq %rax         # rsp -= 8;*rsp = rax;
                       # 此时栈顶元素就是协程被唤醒后需要执行的下一条指令的地址了

    xorl %eax, %eax    # eax = 0;
                       # 使eax清零，eax中的内容作为函数的返回值 
    ret                # 相当于popq %rip 这样就可以唤醒上次挂起的协程，接着运行

  自此，调用resume完成了函数执行的切换！resume的代码分析完了，轮到另一个yield了！其实这两个函数核心功能都是切换函数，所以底层调用的都是co_swap，如下：

void co_yield_env( stCoRoutineEnv_t *env )
{
    //从数组从分别取出两个协程用于交换
    stCoRoutine_t *last = env->pCallStack[ env->iCallStackSize - 2 ];
    stCoRoutine_t *curr = env->pCallStack[ env->iCallStackSize - 1 ];

    env->iCallStackSize--;

    co_swap( curr, last);
}

  看吧，代码是不是超级简单了？

  3、上述代码完美地在一个线程内部完成了不同代码之间的切换，不过都是人工手动掌控切换时机的，如果是网络IO了？开发人员是无法精准预测收发数据时机的，所以也没法人为精准“埋点”resume、yield做协程切换，这种业务场景该怎么处理了？之前用的是epoll来监听socket是否有事件到达，这里该怎么复用epoll了？在example_echosrv.c文件中，微信官方提供了服务端协程的用例，代码不多，但是涉及到main、readwrite、accept等多个方法之间的来回切换，并且方法内部还有for死循环，整个流程比较复杂， 我画了个简单的草图，如下：

  可以看出：整个过程只有一个线程；里面有专门负责接受客户端连接的accept_co协程，也有监听事件的mian主协程，也有添加事件和负责读写的readwrite_so！epoll还是用于网络IO的事件监听和触发，接着就是通过yield、resume切换导到合适的协程处理这些io事件！整个过程逻辑上不算难，就是很繁杂，需要静下心来慢慢捋！为了实现整个流程，有几个关键的函数需要着重说明。

       （1）libco为了对统一网络IO，条件变量需要超时管理的事件，实现了基于时间轮(timing wheel)的超时管理器； 时间轮为图中深红色的轮状数组，数组的每一个单元我们称为一个槽(slot)。单个slot里存储一定时间内注册的事件列表（图中黄色链表）。在libco中，单个slot的精度为1毫秒（刚好是jiffies），整个时间轮由60000个slot组成，对应的整个时间轮覆盖60秒的时间；

   为了实现时间轮，两个核心的方法如下：

/* 在时间轮中插入新项
 * @param 
 * apTimeout :时间轮结构
 * apItem    :新的超时项
 * allNow    :当前事件(timestamp in ms)
 * @return   :0成功, else失败行数
 */
int AddTimeout( stTimeout_t *apTimeout,stTimeoutItem_t *apItem ,unsigned long long allNow )
{
    if( apTimeout->ullStart == 0 )
    {
        apTimeout->ullStart = allNow;// 设置时间轮的最早时间是当前时间
        apTimeout->llStartIdx = 0;
    }
    /* 当前时间小于初始时间出错返回 */
    if( allNow < apTimeout->ullStart )
    {
        co_log_err("CO_ERR: AddTimeout line %d allNow %llu apTimeout->ullStart %llu",
                    __LINE__,allNow,apTimeout->ullStart);

        return __LINE__;
    }
    /* 当前时间大于超时时间出错返回 */
    if( apItem->ullExpireTime < allNow )
    {
        co_log_err("CO_ERR: AddTimeout line %d apItem->ullExpireTime %llu allNow %llu apTimeout->ullStart %llu",
                    __LINE__,apItem->ullExpireTime,allNow,apTimeout->ullStart);

        return __LINE__;
    }
    // 计算超时时间
    unsigned long long diff = apItem->ullExpireTime - apTimeout->ullStart;
    /* 超时时间大于时间轮的最长时间出错返回 */
    if( diff >= (unsigned long long)apTimeout->iItemSize )
    {
        diff = apTimeout->iItemSize - 1;
        co_log_err("CO_ERR: AddTimeout line %d diff %d",
                    __LINE__,diff);

        //return __LINE__;
    }
    /* 将时间加入到时间轮中 */
    AddTail( apTimeout->pItems + ( apTimeout->llStartIdx + diff ) % apTimeout->iItemSize , apItem );

    return 0;
}
/* 在时间轮中取出所有超时项
 * @param 
 * apTimeout:时间轮结构
 * allNow   :当前时间(timestamp in ms)
 * apResult :超时事件结果链表
 */
inline void TakeAllTimeout( stTimeout_t *apTimeout,unsigned long long allNow,stTimeoutItemLink_t *apResult )
{
    if( apTimeout->ullStart == 0 )
    {
        apTimeout->ullStart = allNow;
        apTimeout->llStartIdx = 0;
    }

    if( allNow < apTimeout->ullStart )
    {
        return ;
    }
    int cnt = allNow - apTimeout->ullStart + 1;
    if( cnt > apTimeout->iItemSize )
    {
        cnt = apTimeout->iItemSize;
    }
    if( cnt < 0 )
    {
        return;
    }
    for( int i = 0;i<cnt;i++)
    {
        int idx = ( apTimeout->llStartIdx + i) % apTimeout->iItemSize;
        Join<stTimeoutItem_t,stTimeoutItemLink_t>( apResult,apTimeout->pItems + idx  );
    }
    apTimeout->ullStart = allNow;
    apTimeout->llStartIdx += cnt - 1;


}

  （2）每个main函数都需要调用的方法，用于不停的监听是否有事件发生，如下：

/* 事件循环：不停的监听是否有事件发生
 * @param 
 * ctx:epoll句柄
 * pfn:退出事件循环检查函数
 * arg:pfn参数
 */
void co_eventloop( stCoEpoll_t *ctx,pfn_co_eventloop_t pfn,void *arg )
{
    if( !ctx->result )
    {
        ctx->result =  co_epoll_res_alloc( stCoEpoll_t::_EPOLL_SIZE );
    }
    co_epoll_res *result = ctx->result;
    for(;;)
    {
        int ret = co_epoll_wait( ctx->iEpollFd,result,stCoEpoll_t::_EPOLL_SIZE, 1 );
        stTimeoutItemLink_t *active = (ctx->pstActiveList);
        stTimeoutItemLink_t *timeout = (ctx->pstTimeoutList);
        memset( timeout,0,sizeof(stTimeoutItemLink_t) ); //清空超时队列
        for(int i=0;i<ret;i++)//遍历有事件的fd
        {
            //获取event里数据指向的stTimeoutItem_t
            stTimeoutItem_t *item = (stTimeoutItem_t*)result->events[i].data.ptr;
            if( item->pfnPrepare )//如果有预处理函数，执行，由其加入就绪列表
            {
                item->pfnPrepare( item,result->events[i],active );
            }
            else//手动加入就绪列表
            {
                AddTail( active,item );
            }
        }
        unsigned long long now = GetTickMS();
        /*时间轮中取出所有的超时项，并插入超时列表*/
        TakeAllTimeout( ctx->pTimeout,now,timeout );

        stTimeoutItem_t *lp = timeout->head;
        while( lp )
        {
            //printf("raise timeout %p\n",lp);
            lp->bTimeout = true;//设置为超时
            lp = lp->pNext;
        }
        //将超时列表合并入就绪列表
        Join<stTimeoutItem_t,stTimeoutItemLink_t>( active,timeout );

        lp = active->head;
        while( lp )
        {
            PopHead<stTimeoutItem_t,stTimeoutItemLink_t>( active );
            if (lp->bTimeout && now < lp->ullExpireTime) 
            {    //还未达到超时时间但已经标记为超时的，加回时间轮 
                int ret = AddTimeout(ctx->pTimeout, lp, now);
                if (!ret) 
                {
                    lp->bTimeout = false;
                    lp = active->head;
                    continue;
                }
            }
            /*调用stTimeoutItem_t项的执行函数，也就是OnPollProcessEvent
            里面有co_resume*/
            if( lp->pfnProcess )
            {
                lp->pfnProcess( lp );
            }

            lp = active->head;
        }
        if( pfn )//用于用户控制跳出事件循环
        {
            if( -1 == pfn( arg ) )
            {
                break;
            }
        }
    }
}

  （3）将 fd 交由 Epoll 管理，待 Epoll 的相应的事件触发时，再切换回来执行 read 或者 write 操作，从而实现由 Epoll 管理协程，如下：

/* poll内核：将 fd 交由 Epoll 管理，待 Epoll 的相应的事件触发时，
             再切换回来执行 read 或者 write 操作，
             从而实现由 Epoll 管理协程的功能
 * @param 
 * ctx:epoll句柄
 * fds:fd数组
 * nfds:fd数组长度
 * timeout:超时时间ms
 * pollfunc:默认poll 
 */
typedef int (*poll_pfn_t)(struct pollfd fds[], nfds_t nfds, int timeout);
int co_poll_inner( stCoEpoll_t *ctx,struct pollfd fds[], nfds_t nfds, int timeout, poll_pfn_t pollfunc)
{
    if (timeout == 0)
    {
        //调用系统原生poll(其实上层poll已经做过检查了，此处无需再做)
        return pollfunc(fds, nfds, timeout);
    }
    if (timeout < 0)
    {
        timeout = INT_MAX;
    }
    int epfd = ctx->iEpollFd;
    stCoRoutine_t* self = co_self();

    //1.struct change
    /* 1. 初始化poll相关的数据结构 */
    stPoll_t& arg = *((stPoll_t*)malloc(sizeof(stPoll_t)));
    memset( &arg,0,sizeof(arg) );

    arg.iEpollFd = epfd;
    arg.fds = (pollfd*)calloc(nfds, sizeof(pollfd));//分配nfds个pollfd
    arg.nfds = nfds;

    stPollItem_t arr[2];
    //nfds少于2且未使用共享栈的情况下
    if( nfds < sizeof(arr) / sizeof(arr[0]) && !self->cIsShareStack)
    {    // 栈中分配
        arg.pPollItems = arr;
    }    
    else
    {
        arg.pPollItems = (stPollItem_t*)malloc( nfds * sizeof( stPollItem_t ) );
    }
    memset( arg.pPollItems,0,nfds * sizeof(stPollItem_t) );
    //调用co_resume(arg.pArg), 唤醒参数arg.pArg所指协程
    arg.pfnProcess = OnPollProcessEvent; 
    arg.pArg = GetCurrCo( co_get_curr_thread_env() );//得到当前运行的协程
    
    
    //2. add epoll把事件加入到epoll中监控
    for(nfds_t i=0;i<nfds;i++)
    {
        arg.pPollItems[i].pSelf = arg.fds + i;
        arg.pPollItems[i].pPoll = &arg;

        arg.pPollItems[i].pfnPrepare = OnPollPreparePfn;// 预处理回调函数
        struct epoll_event &ev = arg.pPollItems[i].stEvent;

        if( fds[i].fd > -1 )
        {
            ev.data.ptr = arg.pPollItems + i;
            ev.events = PollEvent2Epoll( fds[i].events );

            int ret = co_epoll_ctl( epfd,EPOLL_CTL_ADD, fds[i].fd, &ev ); //事件加入epoll的监听
            if (ret < 0 && errno == EPERM && nfds == 1 && pollfunc != NULL)
            {
                if( arg.pPollItems != arr )
                {
                    free( arg.pPollItems );
                    arg.pPollItems = NULL;
                }
                free(arg.fds);
                free(&arg);
                return pollfunc(fds, nfds, timeout);
            }
        }
        //if fail,the timeout would work
    }

    //3.add timeout 给时间轮添加超时时间
    unsigned long long now = GetTickMS();
    arg.ullExpireTime = now + timeout;
    int ret = AddTimeout( ctx->pTimeout,&arg,now );
    int iRaiseCnt = 0;
    if( ret != 0 )
    {
        co_log_err("CO_ERR: AddTimeout ret %d now %lld timeout %d arg.ullExpireTime %lld",
                ret,now,timeout,arg.ullExpireTime);
        errno = EINVAL;
        iRaiseCnt = -1;

    }
    else
    {    // 把执行权交给调用此协程的协程，也就是main线程
        co_yield_env( co_get_curr_thread_env() );
        iRaiseCnt = arg.iRaiseCnt;
    }
    /*当 main 协程的事件循环 co_eventloop 中触发了对应的监听事件时，会恢复执行
    此时，将开始执行下半段，即将上半段添加的句柄 fds 从 epoll 中移除，
    清理残留的数据结构*/
    {
        //clear epoll status and memory 
        // 将该项从时间轮中删除
        RemoveFromLink<stTimeoutItem_t,stTimeoutItemLink_t>( &arg );
        for(nfds_t i = 0;i < nfds;i++)
        {
            int fd = fds[i].fd;
            if( fd > -1 )
            {
                co_epoll_ctl( epfd,EPOLL_CTL_DEL,fd,&arg.pPollItems[i].stEvent );
            }
            fds[i].revents = arg.fds[i].revents;
        }


        if( arg.pPollItems != arr )
        {
            free( arg.pPollItems );
            arg.pPollItems = NULL;
        }

        free(arg.fds);
        free(&arg);
    }

    return iRaiseCnt;
}

总结：

1、协程：在各个不同的方法之间切换，从汇编层面看，就是jmp到不同的代码执行

参考：

1、https://www.cyhone.com/articles/analysis-of-libco/  微信 libco 协程库源码分析

2、https://github.com/tencent/libco  libco源码

3、https://www.infoq.cn/article/CplusStyleCorourtine-At-Wechat    C/C++ 协程库 libco：微信怎样漂亮地完成异步化改造

4、https://zhuanlan.zhihu.com/p/27409164  libco协程上下文切换原理

5、https://cloud.tencent.com/developer/article/1459729  libco的设计与实现

6、https://nifengz.com/libco_context_swap/

7、http://kaiyuan.me/2017/07/10/libco/  libco分析

8、https://www.changliu.me/post/libco-auto/ 自动切换

9、https://blog.csdn.net/MOU_IT/article/details/115033799 事件注册poll

10、http://kaiyuan.me/2017/10/20/libco2/  协程的管理