Libevent源码分析—event_base_dispatch()

我们知道libevent是一个Reactor模式的事件驱动的网络库。

 

到目前为止，我们已经看了核心的event和event_base结构体的源码，看了初始化这两个结构体的源码，看了注册event的源码，也将event注册到I/O多路复用监听的事件上了。现在准备工作都做好了，下面就是看运行时的主循环了，在这个主循环中，是如何检测事件、分发事件、调用事件的回调函数的。这一步就是libevent的核心框架流程了。

 

Reactor模式中的Event、Event Handler、Reactor目前都完成了，下面就剩Event Demultiplexer了。

这一步通过event_base_dispatch()完成

int
event_base_dispatch(struct event_base *event_base)
{
  return (event_base_loop(event_base, 0));  //调用event_base_loop()
}

可以看到，该函数只是做了调用event_base_loop()这一个动作，所以工作实际是在函数event_base_loop()内完成的。

event_base_loop()

该函数完成以下工作：

1.信号标记被设置，则调用信号的回调函数

2.根据定时器最小时间，设置I/O多路复用的最大等待时间，这样即使没有I/O事件发生，也能在最小定时器超时时返回。

3.调用I/O多路复用，监听事件，将活跃事件添加到活跃事件链表中

4.检查定时事件，将就绪的定时事件从小根堆中删除，插入到活跃事件链表中

5.对event_base的活跃事件链表中的事件，调用event_process_active(）函数，在该函数内调用event的回调函数，优先级高的event先处理。

 

该函数内部调用了eventop.dispatch()监听事件，event_sigcb函数指针处理信号事件，timeout_process()将超时的定时事件加入到活跃事件链表中，event_process_active()处理活跃事件链表中的事件，调用相应的回调函数。

int
event_base_loop(struct event_base *base, int flags)
{
    const struct eventop *evsel = base->evsel;
    void *evbase = base->evbase;  //event_base的I/O多路复用
    struct timeval tv;
    struct timeval *tv_p;
    int res, done;
    /* clear time cache */
    //清空时间缓存
    base->tv_cache.tv_sec = 0;
    //处理Signal事件时，指定信号所属的event_base
    if (base->sig.ev_signal_added)
        evsignal_base = base;
    done = 0;
    while (!done) {  //进入事件主循环
        /* Terminate the loop if we have been asked to */
        //设置event_base的标记，以表明是否需要跳出循环
        if (base->event_gotterm) {  //event_loopexit_cb()可设置
            base->event_gotterm = 0;
            break;
        }
        if (base->event_break) {  //event_base_loopbreak()可设置
            base->event_break = 0;
            break;
        }
        /* You cannot use this interface for multi-threaded apps */
        //当event_gotsig被设置时，则event_sigcb就是信号处理的回调函数
        while (event_gotsig) {
            event_gotsig = 0;
            if (event_sigcb) {
                res = (*event_sigcb)();  //调用信号处理的回调函数
                if (res == -1) {
                    errno = EINTR;
                    return (-1);
                }
            }
        }
        timeout_correct(base, &tv);  //校准时间
        tv_p = &tv;
        //根据定时器堆中最小超时时间计算I/O多路复用的最大等待时间tv_p
        if (!base->event_count_active && !(flags & EVLOOP_NONBLOCK)) {
            timeout_next(base, &tv_p);
        } else {
            /* 
             * 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)) {
            event_debug(("%s: no events registered.", __func__));
            return (1);
        }
        /* update last old time */
        gettime(base, &base->event_tv);
        /* clear time cache */
        base->tv_cache.tv_sec = 0;
        //调用I/O多路复用，监听事件
        res = evsel->dispatch(base, evbase, tv_p);
        if (res == -1)
            return (-1);
        //将time cache赋值为当前系统时间
        gettime(base, &base->tv_cache);
        
        //检查定时事件，将就绪的定时事件从小根堆中删除，插入到活跃事件链表中
        timeout_process(base);
        if (base->event_count_active) {
            //处理event_base的活跃链表中的事件
            //调用event的回调函数，优先级高的event先处理
            event_process_active(base);  
            if (!base->event_count_active && (flags & EVLOOP_ONCE))
                done = 1;
        } else if (flags & EVLOOP_NONBLOCK)
            done = 1;
    }
    /* clear time cache */
    //循环结束，清空时间缓存
    base->tv_cache.tv_sec = 0;
    event_debug(("%s: asked to terminate loop.", __func__));
    return (0);
}

epoll_dispatch()

在上面我们看到，event_base_loop()中通过I/O多路复用的dispatch()函数完成监听事件功能。在之前的event_init()中我们看到，通过遍历eventops数组，从中选择一个I/O多路复用机制，所以不同的I/O多路复用机制有不同的eventop结构体，相应的也就有不同的dispatch()函数。下面，再次看下eventop结构体(event-internal.h)

struct eventop {
        const char *name;
        void *(*init)(struct event_base *);  //初始化
        int (*add)(void *, struct event *);  //注册事件
        int (*del)(void *, struct event *);  //删除事件
        int (*dispatch)(struct event_base *, void *, struct timeval *);  //事件分发
        void (*dealloc)(struct event_base *, void *);  //注销，释放资源
        /* set if we need to reinitialize the event base */
        int need_reinit;
};

在event_add()中通过add()成员函数注册event到监听事件中，现在在event_base_loop()中通过dispatch()成员函数监听事件。

libevent支持多种I/O多路复用机制，下面先看下epoll的eventop结构体(epoll.c)

const struct eventop epollops = {
    "epoll",
    epoll_init,
    epoll_add,
    epoll_del,
    epoll_dispatch,
    epoll_dealloc,
    1 /* need reinit */
};

然后看下epoll的dispatch()函数(epoll.c)

从下面源码可见，epoll_dispatch()的工作主要有：

1.调用epoll_wait()监听事件

2.如果有信号发生，调用evsignal_process()处理信号

3.将活跃的event根据其活跃的类型注册到活跃事件链表上

4.如果events数组大小不够，则重新分配为原来2倍大小

static int
epoll_dispatch(struct event_base *base, void *arg, struct timeval *tv)
{
    struct epollop *epollop = arg;
    struct epoll_event *events = epollop->events;
    struct evepoll *evep;
    int i, res, timeout = -1;
    if (tv != NULL)
        timeout = tv->tv_sec * 1000 + (tv->tv_usec + 999) / 1000;  //转换为微米
    if (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;
    }
    res = epoll_wait(epollop->epfd, events, epollop->nevents, timeout);  //监听事件发生
    if (res == -1) {
        if (errno != EINTR) {
            event_warn("epoll_wait");
            return (-1);
        }
        evsignal_process(base);  //由于Signal事件发生中断，处理Signal事件
        return (0);
    } else if (base->sig.evsignal_caught) {
        evsignal_process(base);  //有Signal事件发生，处理Signal事件
    }
    event_debug(("%s: epoll_wait reports %d", __func__, res));
    for (i = 0; i < res; i++) {  //处理活跃事件
        int what = events[i].events;  //活跃类型
        struct event *evread = NULL, *evwrite = NULL;
        int fd = events[i].data.fd;  //event的文件描述符
        if (fd < 0 || fd >= epollop->nfds)
            continue;
        evep = &epollop->fds[fd];
        if (what & (EPOLLHUP|EPOLLERR)) {  //判断epoll的events类型，并找到注册的event
            evread = evep->evread;
            evwrite = evep->evwrite;
        } else {
            if (what & EPOLLIN) {
                evread = evep->evread;
            }
            if (what & EPOLLOUT) {
                evwrite = evep->evwrite;
            }
        }
        if (!(evread||evwrite))
            continue;
        
        //添加event到活跃事件链表中
        if (evread != NULL)
            event_active(evread, EV_READ, 1);
        if (evwrite != NULL)
            event_active(evwrite, EV_WRITE, 1);
    }
    //如果注册的事件全部变为活跃，则增大events数组为原来两倍
    if (res == epollop->nevents && epollop->nevents < MAX_NEVENTS) {
        /* 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 = realloc(epollop->events,
            new_nevents * sizeof(struct epoll_event));
        if (new_events) {
            epollop->events = new_events;
            epollop->nevents = new_nevents;
        }
    }
    return (0);
}

event_process_active()

好了，现在活跃的I/O事件、定时器事件已经全部添加到活跃事件链表中了。下面就开始调用这些event的回调函数进行处理了，这步是在event_base_loop()中调用event_process_active()来完成的。

该函数从event_base的activequeueus链表数组上取出一个链表；对该链表上的event调用回调函数；优先调用优先级值最小的event

/*
 * Active events are stored in priority queues.  Lower priorities are always
 * process before higher priorities.  Low priority events can starve high
 * priority ones.
 */
static void
event_process_active(struct event_base *base)
{
    struct event *ev;
    struct event_list *activeq = NULL;
    int i;
    short ncalls;
    for (i = 0; i < base->nactivequeues; ++i) {  //取出第一个活跃链表
        if (TAILQ_FIRST(base->activequeues[i]) != NULL) {
            activeq = base->activequeues[i];
            break;
        }
    }
    assert(activeq != NULL);

    //优先处理优先级值最小的event
    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(ev);  //不是持久事件，则直接删除该事件
        
        /* Allows deletes to work */
        ncalls = ev->ev_ncalls;
        ev->ev_pncalls = &ncalls;
        while (ncalls) {
            ncalls--;
            ev->ev_ncalls = ncalls;
            //调用该event的回调函数，event.ev_res保存返回值
            (*ev->ev_callback)((int)ev->ev_fd, ev->ev_res, ev->ev_arg);  
            if (event_gotsig || base->event_break) {
                  ev->ev_pncalls = NULL;
                return;
            }
        }
        ev->ev_pncalls = NULL;
    }
}