// 事件的dispatch
int
event_base_loop(struct event_base *base, int flags)
{
//得到采用的事件模型 epoll/epoll/select
const struct eventop *evsel = base->evsel;
struct timeval tv;
struct timeval *tv_p;
int res, done, retval = ; /* Grab the lock. We will release it inside evsel.dispatch, and again
* as we invoke user callbacks. */
EVBASE_ACQUIRE_LOCK(base, th_base_lock);
//判断是否loop正在running 如果由则退出
if (base->running_loop) {
event_warnx("%s: reentrant invocation. Only one event_base_loop"
" can run on each event_base at once.", __func__);
EVBASE_RELEASE_LOCK(base, th_base_lock);
return -;
} base->running_loop = ;
//清理时间缓存
clear_time_cache(base); if (base->sig.ev_signal_added && base->sig.ev_n_signals_added)
evsig_set_base_(base); done = ; #ifndef EVENT__DISABLE_THREAD_SUPPORT
base->th_owner_id = EVTHREAD_GET_ID();
#endif
//终止和中断标志至0
base->event_gotterm = base->event_break = ;
//事件循环 与 平时我们自己写的epoll_wait select 等待事件一样 在一个死循环中
while (!done) {
base->event_continue = ;
base->n_deferreds_queued = ; /* Terminate the loop if we have been asked to */
//被其他线程中断
if (base->event_gotterm) {
break;
} if (base->event_break) {
break;
} tv_p = &tv;
//当前没有激活的事件
if (!N_ACTIVE_CALLBACKS(base) && !(flags & EVLOOP_NONBLOCK)) {
timeout_next(base, &tv_p);//判断小根堆中的root是否已经超时， 如果超时 就将tv清0. 如果没有的话就将root的时间减去现在时间的结果赋值给tv(定时器触发的剩余
的时间长度)
} else {
/*
* if we have active events, we just poll new events
* without waiting.
*/
//如果有激活事件 就将tv清空
evutil_timerclear(&tv);
} /* If we have no events, we just exit */
//未注册事件就退出循环
if (==(flags&EVLOOP_NO_EXIT_ON_EMPTY) &&
!event_haveevents(base) && !N_ACTIVE_CALLBACKS(base)) {
event_debug(("%s: no events registered.", __func__));
retval = ;
goto done;
}
event_queue_make_later_events_active(base);
//清理时间缓存
clear_time_cache(base);
// 调用模型的epoll_wait/select/poll 等 tv_p 是刚才计算的最小时间间隔
res = evsel->dispatch(base, tv_p); //以epoll 为例详细说明 if (res == -) {
event_debug(("%s: dispatch returned unsuccessfully.",
__func__));
retval = -;
goto done;
}
//更新base中的时间， 下面就是调用定时事件和IO事件。
update_time_cache(base);
//判断定时器事件是否发生了，若发生就将事件加入激活队列
timeout_process(base); //见下文 if (N_ACTIVE_CALLBACKS(base)) {
//执行激活队列中的事件
int n = event_process_active(base); //见下文
if ((flags & EVLOOP_ONCE)
&& N_ACTIVE_CALLBACKS(base) ==
&& n != )
done = ;
} else if (flags & EVLOOP_NONBLOCK)
done = ;
}
event_debug(("%s: asked to terminate loop.", __func__)); done:
clear_time_cache(base);
base->running_loop = ; EVBASE_RELEASE_LOCK(base, th_base_lock); return (retval);
}
以 epoll 模型的dispatch 看一下evsel->dispatch(base, tv_p);
static int
epoll_dispatch(struct event_base *base, struct timeval *tv)
{
struct epollop *epollop = base->evbase;
struct epoll_event *events = epollop->events;
int i, res;
long timeout = -;
if (tv != NULL) {
timeout = evutil_tv_to_msec_(tv); //将tv 话为 msec传递给epoll_wait
if (timeout < 0 || timeout > MAX_EPOLL_TIMEOUT_MSEC) {
/* Linux kernels can wait forever if the timeout is
* too big; see comment on MAX_EPOLL_TIMEOUT_MSEC. */
timeout = MAX_EPOLL_TIMEOUT_MSEC;
}
} epoll_apply_changes(base);
event_changelist_remove_all_(&base->changelist, base); EVBASE_RELEASE_LOCK(base, th_base_lock);//多线程下防止惊群而加锁 res = epoll_wait(epollop->epfd, events, epollop->nevents, timeout);//等待事件 EVBASE_ACQUIRE_LOCK(base, th_base_lock); if (res == -1) {
if (errno != EINTR) {
event_warn("epoll_wait");
return (-1);
}
return (0);
} event_debug(("%s: epoll_wait reports %d", __func__, res));
EVUTIL_ASSERT(res <= epollop->nevents);
//依次处理事件
for (i = 0; i < res; i++) {
int what = events[i].events;
short ev = 0; if (what & (EPOLLHUP|EPOLLERR)) {
ev = EV_READ | EV_WRITE;
} else {
if (what & EPOLLIN)
ev |= EV_READ;
if (what & EPOLLOUT)
ev |= EV_WRITE;
if (what & EPOLLRDHUP)
ev |= EV_CLOSED;
} if (!ev)
continue;
//根据fd找到相应的位置 将event_callback加入到激活queue等待被调用
evmap_io_active_(base, events[i].data.fd, ev | EV_ET);
}
// 拓展事件容量
if (res == epollop->nevents && epollop->nevents < MAX_NEVENT) {
/* We used all of the event space this time. We should
be ready for more events next time. */
int new_nevents = epollop->nevents * 2;
struct epoll_event *new_events; new_events = mm_realloc(epollop->events,
new_nevents * sizeof(struct epoll_event));
if (new_events) {
epollop->events = new_events;
epollop->nevents = new_nevents;
}
} return (0);
}
再看一下timeout_process.
static void
timeout_process(struct event_base *base)
{
/* Caller must hold lock. */
struct timeval now;
struct event *ev;
//没有定时事件直接退出
if (min_heap_empty_(&base->timeheap)) {
return;
} gettime(base, &now);
//取堆顶最小
while ((ev = min_heap_top_(&base->timeheap))) {
//没有超时就直接退出
if (evutil_timercmp(&ev->ev_timeout, &now, >))
break; /* delete this event from the I/O queues */
//有超时事件发生就将事件从io中删除
event_del_nolock_(ev, EVENT_DEL_NOBLOCK); event_debug(("timeout_process: event: %p, call %p",
ev, ev->ev_callback));
//加入激活队列中 最终会调用 event_queue_insert_timeout，被激活的计时和IO都放在同一个queue中
event_active_nolock_(ev, EV_TIMEOUT, );
}
}