前面的博文已经说到,如果要对多个超时event同时进行监听,就要对这些超时event进行集中管理,能够方便地(时间复杂度小)获取、加入、删除一个event。
在之前的Libevent版本,Libevent使用小根堆管理这些超时event。小根堆的插入和删除时间复杂度都是O(logN)。在2.0.4-alpha版本时,Libevent引入了一个叫common-timeout的东西来管理超时event,要注意的是,它并不是替代小根堆,而是和小根堆配合使用的。事实上,common-timeout的实现要用到小根堆。
Libevent的小根堆和数据结构教科书上的小根堆几乎是一样的。看一下数据结构和Libevent的小根堆源码,很容易就懂的。这样就不多讲了。
本文主要讲一下common-timeout。从common的字面意思和它的实际使用来说,可以把它翻译成“公用超时”。
毫无疑问,如果有相同超时时长的大量超时event都放到小根堆上,那么效率比较低的。虽然小根堆的插入和删除的时间复杂度都是O(logN),但是如果有大量的N,效率也是会下降很多。
common-timeout的思想是,既然有大量的超时event具有相同的超时时长,那么就它们必定依次激活。如果把它们按照超时时间升序地放到一个队列中(在Libevent中就是这样做的),那么每次只需检查队列的第一个超时event即可。因为其他超时event肯定在第一个超时之后才超时的。
前面说到common-timeout和小根堆是配合使用的。从common-timeout中选出最早超时的那个event,将之插入到小根堆中。然后通过小根堆对这个event进行超时监控。超时后再从common-timeout中选出下一个最早超时的event。具体的超时监控处理过程可以参考《超时event的处理》一文。通过这样处理后,就不用把大量的超时event都插入到小根堆中。
下面看一下Libevent的具体实现吧。
首先看一下event_base为common-timeout提供了什么成员变量。
[cpp] view plain copy //event-internal.h文件 struct event_base { //因为可以有多个不同时长的超时event组。故得是数组 //因为数组元素是common_timeout_list指针,所以得是二级指针 struct common_timeout_list **common_timeout_queues; //数组元素个数 int n_common_timeouts; //已分配的数组元素个数 int n_common_timeouts_allocated; }; struct common_timeout_list { //超时event队列。将所有具有相同超时时长的超时event放到一个队列里面 struct event_list events; struct timeval duration;//超时时长 struct event timeout_event;//具有相同超时时长的超时event代表 struct event_base *base; };在实际应用时,可能超时时长为10秒的有1k个超时event,时长为20秒的也有1k个,这就需要一个数组。数组的每一个元素是common_timeout_list结构体指针。每一个common_timeout_list结构体就会处理所有具有相同超时时长的超时event。
common_timeout_list结构体里面有一个event结构体成员,所以并不是从多个具有相同超时时长的超时event中选择一个作为代表,而是在内部有一个event。
common_timeout_list是使用struct event_list结构体队列来管理event,它是一种TAILQ_QUEUE队列,可以参考博文《 TAILQ_QUEUE队列》。
现在来看看怎么使用common-timeout。从上面的代码可以想到,如果要使用common-timeout,就必须把超时event插入到common_timeout_list的events队列中。又因为其要求具有相同的超时时长,所以要插入的超时event要和某个common_timeout_list结构体有相同的超时时长。所以,我们还是来看一下怎么设置common_timeout_list结构体的超时时长。
实际上,并不是设置。而是向event_base申请一个具有特定时长的common_timeout_list。每申请一个,就会在common_timeout_queues数组中加入一个common_timeout_list元素。可以通过event_base_init_common_timeout申请。申请后,就可以直接调用event_add把超时event插入到common-timeout中。但问题是,common-timeout和小根堆是共存的,event_add又没有第三个参数作为说明,要插入到common-timeout还是小根堆。
其实,event_add是根据第二个参数,即超时时长值进行区分的。
首先有一个基本事实,对一个struct timeval结构体,成员tv_usec的单位是微秒,所以最大也就是999999,只需低20比特位就能存储了。但成员tv_usec的类型是int或者long,肯定有32比特位。所以,就有高12比特位是空闲的。
Libevent就是利用那空闲的12个比特位做文章的。这12比特位是高比特位。Libevent使用最高的4比特位作为标志位,标志它是一个专门用于common-timeout的时间,下文将这个标志称为common-timeout标志。次8比特位用来记录该超时时长在common_timeout_queues数组中的位置,即下标值。这也限制了common_timeout_queues数组的长度,最大为2的8次方,即256。
为了方便地处理这些比特位,Libevent定义了下面这些宏定义和一个判断函数。
[cpp] view plain copy //event.c文件 #define COMMON_TIMEOUT_MICROSECONDS_MASK 0x000fffff #define MICROSECONDS_MASK COMMON_TIMEOUT_MICROSECONDS_MASK #define COMMON_TIMEOUT_IDX_MASK 0x0ff00000 #define COMMON_TIMEOUT_IDX_SHIFT 20 #define COMMON_TIMEOUT_MASK 0xf0000000 #define COMMON_TIMEOUT_MAGIC 0x50000000 #define COMMON_TIMEOUT_IDX(tv) \ (((tv)->tv_usec & COMMON_TIMEOUT_IDX_MASK)>>COMMON_TIMEOUT_IDX_SHIFT) #define MAX_COMMON_TIMEOUTS 256 static inline int is_common_timeout(const struct timeval *tv, const struct event_base *base) { int idx; //不具有common-timeout标志位,那么就肯定不是commont-timeout时间了 if ((tv->tv_usec & COMMON_TIMEOUT_MASK) != COMMON_TIMEOUT_MAGIC) return 0; idx = COMMON_TIMEOUT_IDX(tv);//获取数组下标 return idx < base->n_common_timeouts; } 代码最后面的那个判断函数,是用来判断一个给定的struct timeval时间,是否为common-timeout时间。在event_add_internal函数中会用之作为判断,然后根据判断结果来决定是插入小根堆还是common-timeout,这也就完成了区分。
那么怎么得到一个具有common-timeout标志的时间呢?其实,还是通过前面说到的event_base_init_common_timeout函数。该函数将返回一个具有common-timeout标志的时间。
[cpp] view plain copy //event.c文件 //申请一个时长为duration的common_timeout_list const struct timeval * event_base_init_common_timeout(struct event_base *base, const struct timeval *duration) { int i; struct timeval tv; const struct timeval *result=NULL; struct common_timeout_list *new_ctl; //这个时间的微秒位应该进位。用户没有将之进位。比如二进制的103,个位的3应该进位 if (duration->tv_usec > 1000000) { //将之进位,因为下面会用到高位 memcpy(&tv, duration, sizeof(struct timeval)); if (is_common_timeout(duration, base)) tv.tv_usec &= MICROSECONDS_MASK;//去除common-timeout标志 tv.tv_sec += tv.tv_usec / 1000000; //进位 tv.tv_usec %= 1000000; duration = &tv; } for (i = 0; i < base->n_common_timeouts; ++i) { const struct common_timeout_list *ctl = base->common_timeout_queues[i]; //具有相同的duration, 即之前有申请过这个超时时长。那么就不用分配空间。 if (duration->tv_sec == ctl->duration.tv_sec && duration->tv_usec == (ctl->duration.tv_usec & MICROSECONDS_MASK)) {//要&这个宏,才能是正确的时间 result = &ctl->duration; goto done; } } //达到了最大申请个数,不能再分配了 if (base->n_common_timeouts == MAX_COMMON_TIMEOUTS) { goto done; } //新的超时时长,需要分配一个common_timeout_list结构体。 //之前分配的空间已经用完了,要重新申请空间 if (base->n_common_timeouts_allocated == base->n_common_timeouts) { int n = base->n_common_timeouts < 16 ? 16 : base->n_common_timeouts*2; struct common_timeout_list **newqueues = mm_realloc(base->common_timeout_queues, n*sizeof(struct common_timeout_queue *)); if (!newqueues) { goto done; } base->n_common_timeouts_allocated = n; base->common_timeout_queues = newqueues; } //为该超时时长分配一个common_timeout_list结构体 new_ctl = mm_calloc(1, sizeof(struct common_timeout_list)); if (!new_ctl) { goto done; } //为这个结构体进行一些设置 TAILQ_INIT(&new_ctl->events); new_ctl->duration.tv_sec = duration->tv_sec; new_ctl->duration.tv_usec = duration->tv_usec | COMMON_TIMEOUT_MAGIC | //为这个时间加入common-timeout标志 (base->n_common_timeouts << COMMON_TIMEOUT_IDX_SHIFT);//加入下标值 //对timeout_event这个内部event进行赋值。设置回调函数和回调参数。 evtimer_assign(&new_ctl->timeout_event, base, common_timeout_callback, new_ctl); new_ctl->timeout_event.ev_flags |= EVLIST_INTERNAL; //标志成内部event event_priority_set(&new_ctl->timeout_event, 0); //优先级为最高级 new_ctl->base = base; //放到数组对应的位置上 base->common_timeout_queues[base->n_common_timeouts++] = new_ctl; result = &new_ctl->duration; done: return result; }该函数只是在event_base的common_timeout_queues数组中申请一个特定超时时长的位置。同时该函数也会返回一个struct timeval结构体指针变量,该结构体已经被赋予了common-timeout标志。以后使用该变量作为event_add的第二个参数,就可以把超时event插入到common-timeout中了。不应该也不能自己手动为struct timeval变量加入common-timeout标志。
该函数中,也给内部的event进行了赋值,设置了回调函数和回调参数。要注意的是回调参数是这个common_timeout_list结构体变量指针。在回调函数中,有了这个指针,就可以访问events变量,即访问到该结构体上的所有超时event。于是就能手动激活这些超时event。
在Libevent的官方例子中,得到event_base_init_common_timeout的返回值后,就把它存放到另外一个struct timeval结构体中。而不是直接使用返回值作为event_add的参数。
现在已经向event_base申请了一个特定的超时时长,并得到了具有common-timeout标志的时间。那么,就调用event_add看看。
[cpp] view plain copy //event.c文件 static inline int event_add_internal(struct event *ev, const struct timeval *tv, int tv_is_absolute) { struct event_base *base = ev->ev_base; int res = 0; int notify = 0; ...//加入到IO队列或者信号队列 if (res != -1 && tv != NULL) { struct timeval now; int common_timeout; gettime(base, &now); //判断这个时间是否为common-timeout标志 common_timeout = is_common_timeout(tv, base); if (common_timeout) { struct timeval tmp = *tv; //只取真正的时间部分,common-timeout标志位和下标位不要 tmp.tv_usec &= MICROSECONDS_MASK; //转换成绝对时间 evutil_timeradd(&now, &tmp, &ev->ev_timeout); ev->ev_timeout.tv_usec |= (tv->tv_usec & ~MICROSECONDS_MASK); //加入标志位 } event_queue_insert(base, ev, EVLIST_TIMEOUT); if (common_timeout) { struct common_timeout_list *ctl = get_common_timeout_list(base, &ev->ev_timeout); if (ev == TAILQ_FIRST(&ctl->events)) { common_timeout_schedule(ctl, &now, ev); } } } return (res); }由于在《超时event的处理》一文中已经对这个函数进行了一部分讲解,现在只讲有关common-timeout部分。
虽然上面的代码省略了很多东西,但是有一点要说明,当超时event被加入common-timeout时并不会设置notify变量的,即不需要通知主线程。
从上面的代码可以看到,首先是为超时event内部时间ev_timeout加入common-timeout标志。然后调用event_queue_insert进行插入。但此时调用event_queue_insert插入,并不是插入到小根堆。它只是插入到event_base的common_timeout_list数组的一个队列中。下面代码可以看到这一点。
[cpp] view plain copy //event.c文件 static void event_queue_insert(struct event_base *base, struct event *ev, int queue) { ev->ev_flags |= queue; switch (queue) { case EVLIST_TIMEOUT: { if (is_common_timeout(&ev->ev_timeout, base)) { //根据时间向event_base获取对应的common_timeout_list struct common_timeout_list *ctl = get_common_timeout_list(base, &ev->ev_timeout); insert_common_timeout_inorder(ctl, ev); } break; } } } static void //in order说明是有序的。 insert_common_timeout_inorder(struct common_timeout_list *ctl, struct event *ev) { struct event *e; //虽然有相同超时时长,但超时时间却是 超时时长 + 调用event_add的时间。 //所以是在不同的时间触发超时的。它们根据绝对超时时间,升序排在队列中。 //一般来说,直接插入队尾即可。因为后插入的,绝对超时时间肯定大。 //但由于线程抢占的原因,可能一个线程在evutil_timeradd(&now, &tmp, &ev->ev_timeout); //执行完,还没来得及插入,就被另外一个线程抢占了。而这个线程也是要插入一个 //common-timeout的超时event。这样就会发生:超时时间小的反而后插入。 //所以要从后面开始遍历队列,寻找一个合适的地方。 TAILQ_FOREACH_REVERSE(e, &ctl->events, event_list, ev_timeout_pos.ev_next_with_common_timeout) { if (evutil_timercmp(&ev->ev_timeout, &e->ev_timeout, >=)) { TAILQ_INSERT_AFTER(&ctl->events, e, ev, //从队列后面插入 ev_timeout_pos.ev_next_with_common_timeout); return; //插入后就返回 } } //在队列头插入,只会发生在前面的寻找都没有寻找到的情况下 TAILQ_INSERT_HEAD(&ctl->events, ev, ev_timeout_pos.ev_next_with_common_timeout); } 既然event_queue_insert函数并没有完成插入到小根堆。那么就看event_add_internal的最后面的那个if判断。读者可能会问,为什么要插入到小根堆。其实,前面已经说到了。common-timeout是采用一个代表的方式进行工作的。所以肯定要有一个代表被插入小根堆中,这也是common-timeout和小根堆的相互配合。 [cpp] view plain copy //event.c文件 if (common_timeout) { struct common_timeout_list *ctl = get_common_timeout_list(base, &ev->ev_timeout); if (ev == TAILQ_FIRST(&ctl->events)) { common_timeout_schedule(ctl, &now, ev); } } static void common_timeout_schedule(struct common_timeout_list *ctl, const struct timeval *now, struct event *head) { struct timeval timeout = head->ev_timeout; timeout.tv_usec &= MICROSECONDS_MASK; //清除common-timeout标志 //用common_timeout_list结构体的一个event成员作为超时event调用event_add_internal //由于已经清除了common-timeout标志,所以这次将插入到小根堆中。 event_add_internal(&ctl->timeout_event, &timeout, 1); }从判断可以看到,它判断要插入的这个超时event是否为这个队列的第一个元素。如果是的话,就说这个特定超时时长队列第一次有超时event要插入。这就要进行一些处理。
在common_timeout_schedule函数中,我们可以看到,它将一个event插入到小根堆中了。并且也可以看到,代表者不是用户给出的超时event中的一个,而是common_timeout_list结构体的一个event成员。
现在来看一下当common_timeout_list的内部event成员被激活时怎么处理。它的回调函数为common_timeout_callback。
[cpp] view plain copy //event.c文件 static void common_timeout_callback(evutil_socket_t fd, short what, void *arg) { struct timeval now; struct common_timeout_list *ctl = arg; struct event_base *base = ctl->base; struct event *ev = NULL; EVBASE_ACQUIRE_LOCK(base, th_base_lock); gettime(base, &now); while (1) { ev = TAILQ_FIRST(&ctl->events); //该超时event还没到超时时间。不要检查其他了。因为是升序的 if (!ev || ev->ev_timeout.tv_sec > now.tv_sec || (ev->ev_timeout.tv_sec == now.tv_sec && (ev->ev_timeout.tv_usec&MICROSECONDS_MASK) > now.tv_usec)) break; //一系列的删除操作。包括从这个超时event队列中删除 event_del_internal(ev); //手动激活超时event。注意,这个ev是用户的超时event event_active_nolock(ev, EV_TIMEOUT, 1); } //不是NULL,说明该队列还有超时event。那么需要再次common_timeout_schedule,进行监听 if (ev) common_timeout_schedule(ctl, &now, ev); EVBASE_RELEASE_LOCK(base, th_base_lock); }在回调函数中,会手动把用户的超时event激活。于是,用户的超时event就能被处理了。
由于Libevent这个内部超时event的优先级是最高的,所以在接下来就会处理用户的超时event,而无需等到下一轮多路IO复用函数调用返回后。这一点同信号event是一样的,在《 信号event的处理》博文的最后有一些论证。