libco源码解析(1) 协程运行与基本结构
libco源码解析(2) 创建协程，co_create
libco源码解析(3) 协程执行，co_resume
libco源码解析(4) 协程切换，coctx_make与coctx_swap
libco源码解析(5) poll
libco源码解析(6) co_eventloop
libco源码解析(7) read，write与条件变量
libco源码解析(8) hook机制探究
libco源码解析(9) closure实现
引言
poll是libco中所有hook后的函数中可以说是最重要的一个，因为我们不但可以这个函数来隐式的转移CPU执行权，而且其他hook后的函数还可以使用这个hook后的poll在不切换线程的情况下去监听套接字，并在超时或者套接字有事件到来的时候唤醒这个调用poll的协程。
在example_cond.cpp中我们也可以看到使用poll去切换执行权的操作，也正因为此，我们才得以完成使用一个线程去完成一个完整的生产者消费者模型。
正文
关于libco的hook技术原理，我们会在后续文章中去讲解，现在知道它的作用是使用自己的函数去替换库函数就可以了。我们先来看看hook后的poll函数的实现：
typedef int (*poll_pfn_t)(struct pollfd fds[], nfds_t nfds, int timeout);
static poll_pfn_t g_sys_poll_func = (poll_pfn_t)dlsym(RTLD_NEXT,"poll");
// 在hook以后的poll中应该执行协程的调度
int poll(struct pollfd fds[], nfds_t nfds, int timeout)
{
HOOK_SYS_FUNC( poll );
// 如果本线程不存在协程或者超时时间为零的话调用hook前的poll
if (!co_is_enable_sys_hook() || timeout == 0) {
return g_sys_poll_func(fds, nfds, timeout);
}
pollfd *fds_merge = NULL;
nfds_t nfds_merge = 0; // 相当于一个单调递增的标志位
std::map<int, int> m; // fd --> idx
std::map<int, int>::iterator it;
if (nfds > 1) {
fds_merge = (pollfd *)malloc(sizeof(pollfd) * nfds);
for (size_t i = 0; i < nfds; i++) {
if ((it = m.find(fds[i].fd)) == m.end()) { // 在mp中没有找到
fds_merge[nfds_merge] = fds[i]; // 放入merge链表
m[fds[i].fd] = nfds_merge; // 没找到就放进去
nfds_merge++; // 游标递增
} else {
int j = it->second;
fds_merge[j].events |= fds[i].events; // merge in j slot
}
}
}
// 以上就相当于是一个小优化,就是查看此次poll中是否有fd相同的事件,有的话合并一下,仅此而已
int ret = 0;
if (nfds_merge == nfds || nfds == 1) {// 没有执行合并
// fds为poll的事件;nfds为事件数;timeout为超时时间;g_sys_poll_func为未hook的poll函数
// 返回值为此次就绪的事件数
// 在co_poll_inner中有一个协程的切换
ret = co_poll_inner(co_get_epoll_ct(), fds, nfds, timeout, g_sys_poll_func);
} else {
ret = co_poll_inner(co_get_epoll_ct(), fds_merge, nfds_merge, timeout,
g_sys_poll_func);
if (ret > 0) {
// 把merge的事件还原一下
for (size_t i = 0; i < nfds; i++) {
it = m.find(fds[i].fd);
if (it != m.end()) {
int j = it->second;
fds[i].revents = fds_merge[j].revents & fds[i].events;
}
}
}
}
free(fds_merge);
return ret;
}
其实hook后的poll函数最重要的一步就是调用了co_poll_inner，其他的只不过是一些检查罢了。
co_is_enable_sys_hook函数其实就是获取当前线程正在执行的协程，并查看其cEnableSysHook，cEnableSysHook在我们的函数中执行co_enable_hook_sys时设置为1。你可能会问，如果没有执行cEnableSysHook的话根本就不会使用hook后的poll啊，因为这个文件没有被链接进入项目，那么想想看两个函数，一个有cEnableSysHook，一个没有，没有的那个我们显然不希望使用hook后的函数。这也是这段检查的目的所在。
这里还蕴藏着一点，就是co_is_enable_sys_hook中判断co是否为空，而GetCurrThreadCo会在当前env没有被初始化的时候返回空。GetCurrThreadCo为什么不在判断为空时创建一个呢？因为在函数中可能主协程去运行一个带有cEnableSysHook的函数，这个时候我们显然不希望执行hook后的poll。具体可参见example_poll.cpp
bool co_is_enable_sys_hook()
{
// co_enable_sys_hook 会把cEnableSysHook设置成1
stCoRoutine_t *co = GetCurrThreadCo();
return ( co && co->cEnableSysHook );
}
stCoRoutine_t *GetCurrThreadCo( )
{
stCoRoutineEnv_t *env = co_get_curr_thread_env();
// 为什么返回0而不就地初始化呢?看似简单的一笔,与example_poll相关
if( !env ) return 0;
return GetCurrCo(env);
}
注意一点，就是所有经由hook后的socket创建的套接字都是非阻塞的，可能用户想要创建一个阻塞的，但是在libco看来它们都是非阻塞的，而显示给用户的时候还是阻塞的。所以在timeout为0的时候直接执行原poll就可以了，因为非阻塞，这里也不会发生线程的切换，始终在一个线程内运行。
然后是一大段看起来复杂，实际没啥难度的代码，意义就是合并传入的poll事件，至于合并，也就是把不同项但是同一fd的各个poll的项和起来，仅此而已。当然在执行完co_poll_inner之后还需要把合并后的事件再展开，使得整个过程对于用户而言是无感知的。
co_poll_inner
我们接下来看看co_poll_inner函数的实现，这个还是比较麻烦的，其中涉及到了协程的切换，其实也就是这个正在执行的协程让出自己的执行权给调用它的哪一个协程，从这里也可以看出libco是一个典型的非对称协程。
int co_poll_inner( stCoEpoll_t *ctx,struct pollfd fds[], nfds_t nfds, int timeout, poll_pfn_t pollfunc)
{
// 超时时间为零 直接执行系统调用 感觉这直接在hook的poll中判断就好了
if (timeout == 0)
{
return pollfunc(fds, nfds, timeout);
}
if (timeout < 0) // 搞不懂这是什么意思,小于零就看做无限阻塞?
{
timeout = INT_MAX;
}
// epoll fd
int epfd = ctx->iEpollFd;
// 获取当前线程正在运行的协程
stCoRoutine_t* self = co_self();
//1.struct change
// 一定要把这stPoll_t, stPollItem_t之间的关系看清楚
stPoll_t& arg = *((stPoll_t*)malloc(sizeof(stPoll_t)));
// 指针的初始化,非常关键,不加的话在addtail的条件判断中会出现问题
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)
{
// 如果poll中监听的描述符只有1个或者0个， 并且目前的不是共享栈模型
arg.pPollItems = arr;
}
else
{
arg.pPollItems = (stPollItem_t*)malloc( nfds * sizeof( stPollItem_t ) );
}
memset( arg.pPollItems,0,nfds * sizeof(stPollItem_t) );
// 在eventloop中调用的处理函数,功能是唤醒pArg中的协程,也就是这个调用poll的协程
arg.pfnProcess = OnPollProcessEvent;
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; // 第i个poll事件
arg.pPollItems[i].pPoll = &arg;
// 设置一个预处理回调 这个回调做的事情是把此事件从超时队列转到就绪队列
arg.pPollItems[i].pfnPrepare = OnPollPreparePfn;
// ev是arg.pPollItems[i].stEvent的一个引用,这里就相当于是缩写了
// epoll_event 就是epoll需要的事件类型
// 这个结构直接插在红黑树中,时间到来或超时我们可以拿到其中的data
// 一般我用的时候枚举中只使用fd,这里使用了一个指针
struct epoll_event &ev = arg.pPollItems[i].stEvent;
if( fds[i].fd > -1 ) // fd有效
{
ev.data.ptr = arg.pPollItems + i;
ev.events = PollEvent2Epoll( fds[i].events );
// 把事件加入poll中的事件进行封装以后加入epoll
int ret = co_epoll_ctl( epfd,EPOLL_CTL_ADD, fds[i].fd, &ev );
if (ret < 0 && errno == EPERM && nfds == 1 && pollfunc != NULL)
{ //加入epoll失败 且nfds只有一个
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;
// 正常返回return 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
{
// 让出CPU, 切换到其他协程, 当事件到来的时候就会调用callback,那里会唤醒此协程
co_yield_env( co_get_curr_thread_env() );
// --------------我是分割线---------------
// 在预处理中执行+1, 也就是此次阻塞等待的事件中有几个是实际发生了
iRaiseCnt = arg.iRaiseCnt;
}
{
// clear epoll status and memory
// 将该项从时间轮中删除
RemoveFromLink<stTimeoutItem_t,stTimeoutItemLink_t>( &arg );
// 将此次poll中涉及到的时间全部从epoll中删除
// 这意味着有一些事件没有发生就被终止了
// 比如poll中3个事件,实际触发了两个,最后一个在这里就被移出epoll了
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;
}
libco中本来的注释把这个函数分为三个部分：
struct change：把poll的结构改为epoll的结构。add epoll：把改变后的结构加入epoll。add timeout：把这个事件加入超时链表。这里面涉及到了stPoll_t，stPollItem_t，stTimeoutItem_t，三个结构的转化，前两者都继承于stTimeoutItem_t。搞清楚三个结构的转换是理解这个函数的关键之处。总的来说stPoll_t是一个单独的事件，可以放入到时间轮中，其中包含着stPollItem_t。每一个stPollItem_t是一个poll中的事件，都在转化为epoll事件后插入到epoll中。stTimeoutItem_t是前两者的基类，一般用于函数接口，使得可以接收上面两种参数，这里配套了一些基于模板的链表的操作。
我们一一来看一看，当然我们挑一些值得一提的地方说一说，一般的步骤我都写在了注释中：
struct change
我们先来看看stPoll_t结构：
struct stPoll_t : public stTimeoutItem_t
{
struct pollfd *fds; // 描述poll中的事件
nfds_t nfds; // typedef unsigned long int nfds_t;
stPollItem_t *pPollItems; // 要加入epoll的事件 长度为nfds
int iAllEventDetach; // 标识是否已经处理过了这个对象了
int iEpollFd; // epoll fd
int iRaiseCnt; // 此次触发的事件数
};
struct stPollItem_t : public stTimeoutItem_t
{
struct pollfd *pSelf;// 对应的poll结构
stPoll_t *pPoll; // 所属的stPoll_t
struct epoll_event stEvent; // poll结构所转换的epoll结构
};
我们注意到在第一部分设置了一个回调OnPollProcessEvent，它的作用是epoll中检测到事件超时或者事件就绪的时候执行的一个回调的，其实也就是执行了co_resume，从这里我们可以看到执行协程切换我们实际上只需要一个co结构而已。
void OnPollProcessEvent( stTimeoutItem_t * ap )
{
stCoRoutine_t *co = (stCoRoutine_t*)ap->pArg;
co_resume( co );
}
add epoll
这里我们注意到设置了一个预处理回调，算上前面那个回调，它们两个都是在eventloop中被使用，我们下一篇文章会说说eventloop，这部分与Eventloop部分粘连较多，就先不说了，只贴上注释的代码。
void OnPollPreparePfn( stTimeoutItem_t * ap,struct epoll_event &e,stTimeoutItemLink_t *active )
{
stPollItem_t *lp = (stPollItem_t *)ap;
// 把epoll此次触发的事件转换成poll中的事件
lp->pSelf->revents = EpollEvent2Poll( e.events );
stPoll_t *pPoll = lp->pPoll;
// 已经触发的事件数加一
pPoll->iRaiseCnt++;
// 若此事件还未被触发过
if( !pPoll->iAllEventDetach )
{
// 设置已经被触发的标志
pPoll->iAllEventDetach = 1;
// 将该事件从时间轮中移除
// 因为事件已经触发了，肯定不能再超时了
RemoveFromLink<stTimeoutItem_t,stTimeoutItemLink_t>( pPoll );
// 将该事件添加到active列表中
AddTail( active,pPoll );
}
}
add timeout
上一部分中我们已经把所有poll中的事件转化成epoll事件，并插入到epoll中了。并且给每一个poll事件分配了一个stPollItem_t结构，这些stPollItem_t都是属于结构为stPoll_t的arg的，这里我们再次重申一遍，时间轮的链表中存的实际是stPoll_t。
我们来看看如何把一个stPoll_t插入一个时间轮：
/*
* 将事件添加到定时器中
* @param apTimeout - (ref) 超时管理器
* @param apItem - (in) 即将插入的超时事件
* @param allNow - (in) 当前时间
*/
int AddTimeout( stTimeout_t *apTimeout,stTimeoutItem_t *apItem ,unsigned long long allNow )
{
// 当前时间管理器的最早超时时间
if( apTimeout->ullStart == 0 )
{
// 设置时间轮的最早时间是当前时间
apTimeout->ullStart = allNow;
// 设置最早时间对应的index 为 0
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;
// 预期时间到现在不能超过时间轮的大小
// 其实是可以的，只需要取余放进去并加上一个圈数的成员就可以了
// 遍历时圈数不为零就说明实际超时时间还有一个时间轮的长度，
// 遍历完一项以后圈数不为零就减1即可
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__;
}
// 时间轮粒度为1毫秒，即一项代表一毫秒，说实话精度确实算是比较高了，我以前写的是秒。不过毫秒必然会有很多项是空闲的吧。
AddTail( apTimeout->pItems + ( apTimeout->llStartIdx + diff ) % apTimeout->iItemSize , apItem );
return 0;
}
基本的逻辑都已经在注释中啦。AddTail实际执行了把stPoll_t插入到时间轮中。
我们可以看到在把时间加入到时间轮中成功以后会调用co_yield_env，这个函数会使得当前协程让出CPU，把执行权交给调用此协程的协程。想想其实是非常合理的，已经把所有的事件加入到epoll了，这个函数也没有什么执行的必要了，静静等待事件完成或超时就可以了。
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);
}
我们可以看到其实就是在线程独有的调用栈中拿最新的两个协程，正在执行的协程和调用正在执行的协程的协程（有点绕），然后把它们的上下文切换，即切换协程。
然后就是等待协程被切换回来了。
被切换回来以后可能只有一部分事件被触发，这个时候libco的做法是把没有被触发的全部移出epoll，这样避免了协程已经被epoll 中的回调调用，后面仍会触发，并执行回调，就core dump了。
然后就是一般的资源释放啦。
poll到这里就结束了，可以看出它与eventloop的关系还是比较密切的。