libevent源码学习（13）：事件主循环event

#libevent源码学习（13）：事件主循环event| 来源: 网络整理| 查看: 265

开启事件主循环

执行事件主循环

校对时间

阻塞/非阻塞

处理激活队列中的event

事件主循环的退出

event_base_loopexit

event_base_loopbreak开启事件主循环

在libevent中，事件主循环的作用就是执行一个循环，在循环中监听事件以及超时的事件并且将这些激活的事件进行处理。libevent提供了对用户开放了两种执行事件主循环的函数：

int event_base_dispatch(struct event_base *); int event_base_loop(struct event_base *, int);

在event_base_dispatch函数中，实际上调用的是event_base_loop(event_base, 0)；也就是如果使用event_base_dispatch函数执行事件主循环，那么会将event_base_loop的第二个参数设置为0去调用它，下面来看看event_base_loop函数的定义：执行事件主循环

int event_base_loop(struct event_base *base, int flags) { //设置为EVLOOP_NONBLOCK，那么event_loop只会处理当前已经激活的event，处理结束后就会退出event_loop //设置为EVLOOP_ONCE，那么event_loop就会等待到第一个事件超时，处理在这段时间内激活的event，直到所有激活的事件都处理完就退出event_loop //设置为其他值，那么就会一直循环监听，直到被动退出 const struct eventop *evsel = base->evsel; //获取使用的后端 struct timeval tv; struct timeval *tv_p; int res, done, retval = 0; /* Grab the lock. We will release it inside evsel.dispatch, and again * as we invoke user callbacks. */ EVBASE_ACQUIRE_LOCK(base, th_base_lock); //event_base初始化后默认分配的是递归锁 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 -1; } base->running_loop = 1; clear_time_cache(base); //清空缓存的超时 if (base->sig.ev_signal_added && base->sig.ev_n_signals_added) evsig_set_base(base); done = 0; #ifndef _EVENT_DISABLE_THREAD_SUPPORT base->th_owner_id = EVTHREAD_GET_ID(); //存放执行base主循环的线程id #endif base->event_gotterm = base->event_break = 0; while (!done) { base->event_continue = 0; /* Terminate the loop if we have been asked to */ if (base->event_gotterm) { break; } if (base->event_break) { break; } timeout_correct(base, &tv); //用来检查用户是否回调了系统时间，如果改变了就往回调整系统时间 tv_p = &tv; //回调后的时间 //如果当前没有事件激活，那么后面的dispatch就应该阻塞，如果不阻塞那么就会多次调用dispatch； //而对于flags来说，如果把flags设置为EVLOOP_NONBLOCK，那么说明调用者希望后面的dispatch是非阻塞的 //因此如果当前没有事件激活的，那么flags就不应该设置为EVLOOP_NONBLOCK； //除此之外的其他情况则会将dispatch设置为非阻塞。 //dispatch是否阻塞取决于传入的超时参数，超时参数描述了dispatch阻塞的时长，如果为0那么dispatch就立即返回，如果为-1就是一直阻塞，直到相应事件发生。 //因此这里如果需要阻塞，那么就设置dispatch的阻塞时长为从现在开始到第一个超时的event所需的时间 //否则则设置阻塞时长为0，相当于非阻塞。 if (!N_ACTIVE_CALLBACKS(base) && !(flags & EVLOOP_NONBLOCK)) { timeout_next(base, &tv_p); //计算min_heap中最先超时的event还有多长时间就要超时 } else {//如果有事件激活或者是event_loop为非阻塞，就不用等待 /* * if we have active events, we just poll new events * without waiting. */ evutil_timerclear(&tv); //清空超时 } /* If we have no events, we just exit */ if (!event_haveevents(base) && !N_ACTIVE_CALLBACKS(base)) {//如果base中的所有event都处理完了就退出，如果base中存在永久事件，那么event_loop是不会主动退出的 event_debug(("%s: no events registered.", __func__)); retval = 1; goto done; } /* update last old time */ gettime(base, &base->event_tv); //更新系统时间 clear_time_cache(base); //清空缓存的时间 //根据前面，如果希望dispatch是非阻塞的，那么这里的tv_p就是0，该函数就会立即返回； //如果希望dispatch是阻塞的，那么这里的tv_p就是距离第一个超时event的剩余时长 //这是为了处理在所有事件都没有超时的情况下可能发生的感兴趣事件 //如果在这段时间内没有感兴趣的事件发生，实际上这里的dispatch也并没有对事件做任何处理 res = evsel->dispatch(base, tv_p); //返回后，接下来就是处理超时的事件了 if (res == -1) { event_debug(("%s: dispatch returned unsuccessfully.", __func__)); retval = -1; goto done; } update_time_cache(base); timeout_process(base); //将base的min_heap中所有超时的事件以超时激活类型添加到激活队列中 if (N_ACTIVE_CALLBACKS(base)) { //如果激活队列中有事件 int n = event_process_active(base); //执行激活队列中的event相应的回调函数，返回的n是成功执行的非内部事件数目 if ((flags & EVLOOP_ONCE) //如果设置了EVLOOP_ONCE，并且所有激活的事件都处理完了，那么就退出event_loop && N_ACTIVE_CALLBACKS(base) == 0 && n != 0) done = 1; } else if (flags & EVLOOP_NONBLOCK)//如果设置了EVLOOP_NONBLOCK那么也会退出event_loop循环 done = 1; } event_debug(("%s: asked to terminate loop.", __func__)); //循环结束 done: clear_time_cache(base); base->running_loop = 0; //表示base的循环结束 EVBASE_RELEASE_LOCK(base, th_base_lock); return (retval); }

校对时间

在循环中，首先会调用timeout_correct来校对时间，之所以校对，是因为用户是可以改变系统时间的，比如说把现在的10点改成了8点，那么当前的系统时间就会变小，由于定时器中存放的时间都是按照之前设置的系统时间，因此这些超时event会更晚发生，所以在每次循环的开始应该先进行校对时间。timeout_correct函数定义如下：

static void timeout_correct(struct event_base *base, struct timeval *tv) { /* Caller must hold th_base_lock. */ struct event **pev; unsigned int size; struct timeval off; int i; if (use_monotonic) //如果平台支持use_monotonic，那么系统时间就不需要回调了，直接使用系统启动时间 return; /* Check if time is running backwards */ gettime(base, tv); //tv就是系统时间，也是进行回调纠正的时间 if (evutil_timercmp(tv, &base->event_tv, >=)) { //如果此时获取的系统时间tv大于之前保存的时间event_tv，那么说明没有进行回调 base->event_tv = *tv; return; } //执行到这里说明刚刚获得的tv小于保存的base->event_tv，说明用户往回修改了系统时间 event_debug(("%s: time is running backwards, corrected", __func__)); evutil_timersub(&base->event_tv, tv, &off); //计算二者之间的差值，保存到off中，反映了用户调小了多少时间 /* * We can modify the key element of the node without destroying * the minheap property, because we change every element. */ pev = base->timeheap.p; size = base->timeheap.n; for (; size-- > 0; ++pev) { struct timeval *ev_tv = &(**pev).ev_timeout; evutil_timersub(ev_tv, &off, ev_tv); } for (i=0; in_common_timeouts; ++i) { struct event *ev; struct common_timeout_list *ctl = base->common_timeout_queues[i]; TAILQ_FOREACH(ev, &ctl->events, ev_timeout_pos.ev_next_with_common_timeout) { struct timeval *ev_tv = &ev->ev_timeout; ev_tv->tv_usec &= MICROSECONDS_MASK; evutil_timersub(ev_tv, &off, ev_tv); ev_tv->tv_usec |= COMMON_TIMEOUT_MAGIC | (itimeheap); //获取最先超时的event if (ev == NULL) { //说明timeheap中没有event，也就是没有事件激活 /* if no time-based events are active wait for I/O */ *tv_p = NULL; goto out; } if (gettime(base, &now) == -1) { //获取时间到now中 res = -1; goto out; } //比较最先超时的event的超时时间与获取到的时间， //如果最先超时的时间都已经小于等于当前获取到的时间，说明此时可能已经超过了设置的时间，就清空tv，并且立即返回 if (evutil_timercmp(&ev->ev_timeout, &now, ev_timeout, &now, tv); //计算剩余的超时时间，保存到tv中 EVUTIL_ASSERT(tv->tv_sec >= 0); EVUTIL_ASSERT(tv->tv_usec >= 0); event_debug(("timeout_next: in %d seconds", (int)tv->tv_sec)); out: return (res); }

可以知道，该函数用来计算当前距离min_heap中最先超时的那个event超时还剩多长时间。如果最先超时的那个event的超时时间不小于当前的系统时间，那么就会清空传入的tv的超时值然后返回，否则就计算超时时间与系统时间的差值保存到tv的超时值中。

回到前面，N_ACTIVE_CALLBACKS(base)是一个宏定义，表示base中已激活的event数量。当base中没有激活的event，并且没有设置EVLOOP_NONBLOCK时，tv中保存的就是还有多长时间min_heap中第一个event会超时。否则tv中保存的超时值就是0。

接下来就继续执行了evsel->dispatch(base, tv_p); 语句，直接调用base后端方法中绑定的dispatch函数，可以参考epoll模型中该函数的作用。如果base所绑定的eventop为epoll，那么显然在dispatch中就会调用epoll_wait。而这里的tv_p参数也会作为超时参数传给epoll_wait。如果base中没有激活的event并且没有设置EVLOOP_NONBLOCK，说明需要按照阻塞的方式去执行dispatch，此时tv_p中保存的是还有多长时间第一个event超时，就让dispatch阻塞tv_p这么长的时间去监听所有已添加但未激活的event。除了这种情况，如果base中存在已激活的event，那么自然就应该马上到激活队列中去处理这些event，因此传递给dispatch的tv_p为0让其立刻返回；如果设置了EVLOOP_NONBLOCK，其含义是只处理当前已经激活的，不需要阻塞去监听其他event，因此这里依然会设置tv_p为0让dispatch立刻返回。值得一提的是，这里的dispatch函数的调用相当于是在第一个event超时发生之前去监听所有event感兴趣的非超时事件。

在dispatch函数返回后，说明此时监听非超时event结束，接下来就开始处理超时event。先调用timeout_process将min_heap中所有超时的event添加到激活队列中（可参考timeout_process），timeout_process函数返回后，如果base中有激活的event，那么就会调用event_process_active函数去处理这些event，该函数定义如下：处理激活队列中的event

static int event_process_active(struct event_base *base)//遍历base的激活队列中所有event，调用其回调函数 { /* Caller must hold th_base_lock */ struct event_list *activeq = NULL; int i, c = 0; for (i = 0; i < base->nactivequeues; ++i) { //遍历激活队列中的事件 if (TAILQ_FIRST(&base->activequeues[i]) != NULL) { //同一个优先级下可以有多个事件 base->event_running_priority = i; //设置当前的优先级 activeq = &base->activequeues[i]; //获取优先级i下的所有event组成的链表 c = event_process_active_single_queue(base, activeq); //遍历activeq链表，调用其中每个event的回调函数 if (c < 0) {//c是执行的非内部事件数目 base->event_running_priority = -1; return -1; } else if (c > 0) break; /* Processed a real event; do not * consider lower-priority events */ /* If we get here, all of the events we processed * were internal. Continue. */ } } event_process_deferred_callbacks(&base->defer_queue,&base->event_break); //处理延时激活队列中激活的event base->event_running_priority = -1; return c; } static int event_process_active_single_queue(struct event_base *base, struct event_list *activeq) { struct event *ev; int count = 0; EVUTIL_ASSERT(activeq != NULL); for (ev = TAILQ_FIRST(activeq); ev; ev = TAILQ_FIRST(activeq)) { if (ev->ev_events & EV_PERSIST) //如果是永久事件就从激活队列中删除，保留其在添加队列和定时队列 event_queue_remove(base, ev, EVLIST_ACTIVE); else //非永久事件从所有队列中删除 event_del_internal(ev); if (!(ev->ev_flags & EVLIST_INTERNAL))//非内部事件计数 ++count; ...... switch (ev->ev_closure) { //在调用回调函数是否进行其他行为 case EV_CLOSURE_SIGNAL: event_signal_closure(base, ev); break; case EV_CLOSURE_PERSIST: //对于永久事件，在调用回调函数之前会重新调用event_add来添加该事件到对应队列中 event_persist_closure(base, ev); break; default: case EV_CLOSURE_NONE: //对于一般事件，直接调用回调函数 EVBASE_RELEASE_LOCK(base, th_base_lock);//释放锁 (*ev->ev_callback)( ev->ev_fd, ev->ev_res, ev->ev_arg); //调用回调函数 break; } ...... }

到这里，就把所有激活的event都调用了相应的回调函数，如果是永久事件，那么它还会存在于队列中等待下一次激活，如果是普通事件则会在激活后彻底删除。激活后会进行一些判断：如果flags设置了EVLOOP_ONCE，按照前面所说的，必须要将当前所有已经激活的event都处理完才退出主循环，因此这里还需要再次确保激活队列中没有event，这里判断n的主要原因是确保刚才激活的event都是用户定义的event而不是内部event（如common_timeout event就属于内部event），如果只处理了内部event那也被认为是没有处理任何激活事件，当以上条件都满足，那么done就会设置为1，下一轮while(!done)就会直接退出，主循环结束；而如果flags设置了EVLOOP_NONBLOCK，根据其含义下一次循环也直接退出。

if (N_ACTIVE_CALLBACKS(base)) { //如果激活队列中有事件 int n = event_process_active(base); //执行激活队列中的event相应的回调函数，返回的n是成功执行的非内部事件数目 if ((flags & EVLOOP_ONCE) //如果设置了EVLOOP_ONCE，并且所有激活的事件都处理完了，那么就退出event_loop && N_ACTIVE_CALLBACKS(base) == 0 && n != 0) done = 1; } else if (flags & EVLOOP_NONBLOCK)//如果设置了EVLOOP_NONBLOCK那么也会退出event_loop循环 done = 1; }

这里提到了设置EVLOOP_NONBLOCK和EVLOOP_ONCE时退出主循环的方式，前面说过，如果设置的flags不是这二者（如调用event_base_dispatch），那么主循环就会一直执行下去，那么此时又该如何退出事件主循环呢？事件主循环的退出

对于event_base_loop的退出，实际上是通过设置base的event_gotterm或event_break来实现的，在base的英文注释中提到，设置前者表明会在当前所有激活的event都处理结束后退出主循环，而设置后者表明会立刻退出主循环，二者的注释如下所示：

/** Set if we should terminate the loop once we're done processing events. */ int event_gotterm; /** Set if we should terminate the loop immediately */ int event_break;

关于设置base中的这两个变量，libevent向用户提供了两种退出event_base_loop的方法:

int event_base_loopexit(struct event_base *, const struct timeval *); int event_base_loopbreak(struct event_base *);

event_base_loopexit

先来看看event_base_loopexit函数，该函数定义如下：

int event_base_loopexit(struct event_base *event_base, const struct timeval *tv) { return (event_base_once(event_base, -1, EV_TIMEOUT, event_loopexit_cb, event_base, tv)); //添加一个新的纯超时事件，在tv时超时，回调event_loopexit_cb设置event_gotterm位 } int event_base_once(struct event_base *base, evutil_socket_t fd, short events, void (*callback)(evutil_socket_t, short, void *), void *arg, const struct timeval *tv) { struct event_once *eonce; struct timeval etv; int res = 0; /* We cannot support signals that just fire once, or persistent * events. */ if (events & (EV_SIGNAL|EV_PERSIST)) return (-1); if ((eonce = mm_calloc(1, sizeof(struct event_once))) == NULL) return (-1); eonce->cb = callback; eonce->arg = arg; //注册一个新事件，回调函数为event_once_cb if (events == EV_TIMEOUT) { //如果是超时事件 if (tv == NULL) { evutil_timerclear(&etv); tv = &etv; } evtimer_assign(&eonce->ev, base, event_once_cb, eonce); //注册一个纯超时事件 } else if (events & (EV_READ|EV_WRITE)) { //如果是读写事件 events &= EV_READ|EV_WRITE; event_assign(&eonce->ev, base, fd, events, event_once_cb, eonce); } else { /* Bad event combination */ mm_free(eonce); return (-1); } if (res == 0) res = event_add(&eonce->ev, tv); //添加事件到相应队列中 if (res != 0) { mm_free(eonce); return (res); } return (0); }

也就是说，event_base_loopexit函数实际上就是重新注册了一个纯超时event，超时时直接调用event_loopexit_cb函数，event_loopexit_cb函数定义如下：

static void event_loopexit_cb(evutil_socket_t fd, short what, void *arg) { struct event_base *base = arg; base->event_gotterm = 1; }

可以看到，在event_loopexit_cb函数，将base的event_gotterm置为1。在此之后，当event_base_loop执行完当前循环，在下一次循环开始检测到event_gotterm非0，就会退出主循环。因此，event_base_loopexit实际上是传入一个timeval超时参数，它会根据超时参数注册并且添加一个纯超时event，当event超时就会回调event_loopexit_cb函数将base的event_gotterm置为1，从而让事件主循环退出。

接下来再看看event_base_loopbreak函数。event_base_loopbreak

event_base_loopbreak函数定义如下所示：

int event_base_loopbreak(struct event_base *event_base) //设置event_break位 { int r = 0; if (event_base == NULL) return (-1); EVBASE_ACQUIRE_LOCK(event_base, th_base_lock); event_base->event_break = 1; if (EVBASE_NEED_NOTIFY(event_base)) { r = evthread_notify_base(event_base); } else { r = (0); } EVBASE_RELEASE_LOCK(event_base, th_base_lock); return r; }

由此可见event_base_loopbreak的执行逻辑非常简单，就是直接将base的event_break置为1来退出主循环。

【本文地址】

公司简介

联系我们