从 example_echosvr.cpp 认识 Libco 事件循环

桃扇骨 2022-03-09 06:10 189阅读 0赞

### 主函数： ###

int main(int argc, char *argv[])
    { 
    
        g_listen_fd = CreateTcpSocket(port, ip, true);
        listen(g_listen_fd, 1024);
        SetNonBlock(g_listen_fd);
    
        for (int i = 0; i < cnt; i++)
        { 
            task_t *task = (task_t *)calloc(1, sizeof(task_t));
            task->fd = -1;
    
            co_create(&(task->co), NULL, readwrite_routine, task);
            /* 使用函数 readwrite_coroutine 创建多个读写协程， 调用 co_resume 启动协程运行直到其挂起 */
            co_resume(task->co);
        }
        stCoRoutine_t *accept_co = NULL;
        co_create(&accept_co, NULL, accept_routine, 0);
        co_resume(accept_co);
    
        co_eventloop(co_get_epoll_ct(), 0, 0);
        /* co_eventloop 实现事件的监听和协程的循环切换；*/
        exit(0);
        return 0;
    }

### readwrite\_routine 函数 ###

static void *readwrite_routine(void *arg)
    { 
        co_enable_hook_sys();
        
        task_t *co = (task_t *)arg;
        char buf[1024 * 16];
        for (;;)
        { 
            if (-1 == co->fd)
            { 
                g_readwrite.push(co);
                //将新创建的读写协程都加入到队列 g_readwrite 中
                co_yield_ct();
                //让出 CPU,libco是对称协程，所以会回到调用它的地方（main 主线程）
                //下次回来还是到这里
                continue;
            }
            int fd = co->fd;
            co->fd = -1;
    
            for (;;)
            { 
                struct pollfd pf = { 0};
                pf.fd = fd;
                pf.events = (POLLIN | POLLERR | POLLHUP);
                co_poll(co_get_epoll_ct(), &pf, 1, 1000);
                //内部就是：注册fd -> 加入定时器 -> 切换CPU->删除fd和定时器结构
                //为什么要删除?
                //因为不应该让某一个协程与某个fd绑定啊...(暂时是这样理解的呐！！)
    
                printf("get a message \n");
                //之后进行读写
    
                int ret = read(fd, buf, sizeof(buf));
                if (ret > 0)
                { 
                    printf("massege == %s\n", buf);
                    ret = write(fd, buf, ret);
                }
                if (ret > 0 || (-1 == ret && EAGAIN == errno))
                { 
                    continue;
                }
                close(fd);
                break;
            }
        }
        return 0;
    }

### accept\_routine 函数 ###

**主线程中的函数 co\_eventloop 监听网络事件，将来自于客户端新进的连接交由协程 accept\_co 处理**，关于 co\_eventloop 如何唤醒 accept\_co 的细节我们将在后续介绍

static void *accept_routine(void *)
    { 
        co_enable_hook_sys();
        printf("accept_routine\n");
        fflush(stdout);
        for (;;)
        { 
            //printf("pid %ld g_readwrite.size %ld\n",getpid(),g_readwrite.size());
            /*　检查队列 g_readwrite 是否有空闲的读写 coroutine　*/
            if (g_readwrite.empty())
            { 
                printf("empty\n"); //sleep
                struct pollfd pf = { 0};
                pf.fd = -1;
                /* 会调用　co_poll_inner　函数 这里是只是构造了一个空事件，满足参数要求，后续的　co_poll_inner　 会检测 fd > -1 并将该协程加入到 Epoll 管理的定时器队列中,时间到达，就会唤醒该协程 时间设置　1000 ms */
                poll(&pf, 1, 1000);
    
                continue;
            }
            struct sockaddr_in addr; //maybe sockaddr_un;
            memset(&addr, 0, sizeof(addr));
            socklen_t len = sizeof(addr);
    
            int fd = co_accept(g_listen_fd, (struct sockaddr *)&addr, &len); // accept
            if (fd < 0)
            { 
                printf(" co_accept() fd < 0 \n");
                struct pollfd pf = { 0};
                pf.fd = g_listen_fd;
                pf.events = (POLLIN | POLLERR | POLLHUP);
                /* 还是会调用　co_poll_inner　函数 如果接收连接失败，那么调用 co_poll 将服务端的 listen_fd 加入到 Epoll 中 并加入其对应的定时器，唤醒的条件有两个：(1)事件到达　(2)定时器超时 之后切换出CPU,交给主线程，主线程中通过eventloop->epoll调度协程， 如果连接套接字有事件发生，就回到co_poll_inne函数， 然后这个函数就会将该套接字的结构从epoll中删除，将其定时器也删除掉， 之后就到了这里 */
                co_poll(co_get_epoll_ct(), &pf, 1, 1000);
                continue;
            }
            if (g_readwrite.empty())
            { 
                close(fd);
                continue;
            }
            SetNonBlock(fd);
    
            printf("a new connection\n");
    
            task_t *co = g_readwrite.top();
            co->fd = fd;
            g_readwrite.pop();
            co_resume(co->co);
        }
        return 0;
    }

### eventloop函数 ###

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 (;;)
        { 
            //epoll_wait() 等待 I/O 就绪事件，最大等待时长设置为 1 毫秒,
            //为了配合时间轮工作，滴答时钟
            int ret = co_epoll_wait(ctx->iEpollFd, result, stCoEpoll_t::_EPOLL_SIZE, 1);
    
            //active 指针指向当前执行环境的 pstActiveList 队列，注意这里面可能已经有“活跃”的待处理事件
            stTimeoutItemLink_t *active = (ctx->pstActiveList);
            stTimeoutItemLink_t *timeout = (ctx->pstTimeoutList);
    
            memset(timeout, 0, sizeof(stTimeoutItemLink_t));
    
            //循环处理 epoll_wait() 得到的 I/O 就绪文件描述符
            for (int i = 0; i < ret; i++)
            { 
                stTimeoutItem_t *item = (stTimeoutItem_t *)result->events[i].data.ptr;
                /* 处理就绪的文件描述符。如果用户设置了预处理回调，则调用 pfnPrepare 做预处理（15行）； 否则直接将就绪事件 item 加入 active 队列。实际上，pfnPrepare() 预处理函数内部也是会将就绪 item 加入 active 队列， 最终都是加入到 active 队列等待统一处理。*/
                //执行预处理函数
                if (item->pfnPrepare)
                { 
                    item->pfnPrepare(item, result->events[i], active);
                }
                else
                { 
                    AddTail(active, item);
                }
            }
    
            unsigned long long now = GetTickMS();
            TakeAllTimeout(ctx->pTimeout, now, timeout); //从时间轮上取出已超时的事件，放到 timeout 队列，时间轮转动
    
            stTimeoutItem_t *lp = timeout->head;
            // 遍历 timeout 队列，设置事件已超时标志（bTimeout 设为 true）。
            while (lp)
            { 
                //printf("raise timeout %p\n",lp);
                lp->bTimeout = true; 标记为超时
                lp = lp->pNext;
            }
            // 将 timeout 队列中事件合并到 active 队列。
            Join<stTimeoutItem_t, stTimeoutItemLink_t>(active, timeout);
    
            //才开始真正 处理 active 中的事件 1.epoll就绪事件、2.时间轮超时事件。
            /* 遍历 active 队列，调用工作协程设置的 pfnProcess() 回调函数 resume 挂起的工作协程，处理对应的 I/O 或超时事件。 */
            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;
                    }
                }
                if (lp->pfnProcess)
                { 
                    lp->pfnProcess(lp); //注意这里，
                    /* co_poll_inner: arg.pfnProcess = OnPollProcessEvent( ==> resume ); */
                }
    
                lp = active->head;
            }
            if (pfn)
            { 
                if (-1 == pfn(arg))
                { 
                    break;
                }
            }
        }
    }

### co\_poll\_inner 注册事件的函数 ###

/* { struct pollfd pf = {0}; pf.fd = g_listen_fd; pf.events = (POLLIN | POLLERR | POLLHUP); co_poll(co_get_epoll_ct(), &pf, 1, 1000); } int co_poll(stCoEpoll_t *ctx, struct pollfd fds[], nfds_t nfds, int timeout_ms) { return co_poll_inner(ctx, fds, nfds, timeout_ms, NULL); } 所以　fds[].fd = g_listen_fd */
    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)
        { 
            return pollfunc(fds, nfds, timeout);
        }
        if (timeout < 0)
        { 
            timeout = INT_MAX;
        }
        int epfd = ctx->iEpollFd;
        stCoRoutine_t *self = co_self();
    
        //1.struct change
        stPoll_t &arg = *((stPoll_t *)malloc(sizeof(stPoll_t)));
        memset(&arg, 0, sizeof(arg));
    
        arg.iEpollFd = epfd;
        arg.fds = (pollfd *)calloc(nfds, sizeof(pollfd));
        arg.nfds = nfds;
    
        stPollItem_t arr[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));
    
        arg.pfnProcess = OnPollProcessEvent;
        /* void OnPollProcessEvent(stTimeoutItem_t *ap) { stCoRoutine_t *co = (stCoRoutine_t *)ap->pArg; co_resume(co); } 事件到来是直接调用这个函数就行了，而这个函数就会唤醒结构对应的协程开始进行工作 */
        arg.pArg = GetCurrCo(co_get_curr_thread_env());
    
        //2. add 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) //只有 fd >-1 才会添加进去,所以请看：
            /*accept_routine(): 　　struct pollfd pf = {0}; pf.fd = -1; poll(&pf, 1, 1000); */
            { 
                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);
                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); //添加操作，该函数执行成功返回值为　0
        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
        { 
            printf("co_poll_inner 切换出 CPU\n");
            co_yield_env(co_get_curr_thread_env());
    
            //从 accept 协程，切回到主协程中，回到 main 里面
    
            //切换回来不是到这里了吗？下面删除掉干嘛啊？
    
            iRaiseCnt = arg.iRaiseCnt;
        }
    
        { 
            //clear epoll status and memory，
            //这里将定时器也删除了
            //这是为什么?实在是有点想不通啊！！！
            RemoveFromLink<stTimeoutItem_t, stTimeoutItemLink_t>(&arg);
            for (nfds_t i = 0; i < nfds; i++)
            { 
                printf("清除相关信息\n");
                int fd = fds[i].fd; //假如是　accept_routine g_listen_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;
    }

OK,以上就是 Libco 的事件驱动模型了．

### 总结一下：主要清楚了两点就行了 ###

##### 1. 注册 #####

1.  构造对应的结构呢加入epoll
2.  构造定时器加入（注意会有一个回调函数）
3.  让出CPU
4.  删除对应fd的结构与对应的定时器
5.  执行read/write/accept操作

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdXNoZW5neGlfcm9vdA_size_16_color_FFFFFF_t_70]

##### 2. 主事件循环 #####

1.  调用 epoll\_wait() 等待 I/O 就绪事件，为了配合时间轮工作，这里的 timeout 设置为 1 毫秒。
2.  active 指针指向当前执行环境的 pstActiveList 队列，**注意这里面可能已经有“活跃”的待处理事件**。timeout 临时链表
3.  处理就绪的文件描述符。有预处理回调就调用 pfnPrepare 做预处理，否则直接将就绪事件 item 加入 active 队列。实际上，pfnPrepare() 预处理函数内部也是会将就绪item加入 active 队列，最终都是加入到 active 队列等到后边统一处理。
4.  从时间轮上取出已超时的事件，放到 timeout 队列。
5.  遍历 timeout 队列，设置事件已超时标志（bTimeout 设为 true）。
6.  将 timeout 队列中事件合并到 active 队列。
7.  遍历 active 队列，调用工作协程设置的 pfnProcess() 回调函数 resume 挂起的工作协程，处理对应的 I/O 或超时事件(在注册时会注册一个回调函数)。

这就是主协程的事件循环工作过程，我们看到它周而复始地 epoll\_wait()，唤醒挂起的工作协程去处理定时器与 I/O 事件。这里的逻辑看起来跟所有基于 epoll 实现的事件驱动网络框架并没有什么特别之处，更没有涉及到任何协程调度算法，由此也可以看到 libco 其实是一个很典型的非对称协程机制。

参考：  
[https://blog.didiyun.com/index.php/2018/11/23/libco/\#主协程事件循环源码分析][https_blog.didiyun.com_index.php_2018_11_23_libco]

[http://kaiyuan.me/2017/10/20/libco2/][http_kaiyuan.me_2017_10_20_libco2]

[https://segmentfault.com/a/1190000012834756][https_segmentfault.com_a_1190000012834756]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdXNoZW5neGlfcm9vdA_size_16_color_FFFFFF_t_70]: /images/20220309/803559e5b87c4332946bcce8e6d5b47b.png
[https_blog.didiyun.com_index.php_2018_11_23_libco]: https://blog.didiyun.com/index.php/2018/11/23/libco/#%E4%B8%BB%E5%8D%8F%E7%A8%8B%E4%BA%8B%E4%BB%B6%E5%BE%AA%E7%8E%AF%E6%BA%90%E7%A0%81%E5%88%86%E6%9E%90
[http_kaiyuan.me_2017_10_20_libco2]: http://kaiyuan.me/2017/10/20/libco2/
[https_segmentfault.com_a_1190000012834756]: https://segmentfault.com/a/1190000012834756