libevent源码学习（二）事件循环event_base

向右看齐 2022-06-15 19:54 387阅读 0赞

Reactor模式中有一些基本组建，在libevent中，实现Reactor的核心就封装在event\_base中

struct event_base {
        const struct eventop *evsel;
        void *evbase;
        //...
        int event_count;
        int event_count_active;
        struct event_list **activequeues;
        //...
    };

这其中有一个重要的结构体`struct eventop`，来看看它的结构：

struct eventop
    {
        /** The name of this backend. */
        const char *name;
        void *(*init)(struct event_base *);
        int (*add)(struct event_base *, evutil_socket_t fd, short old, short events, void *fdinfo);
        /** As "add", except 'events' contains the events we mean to disable. */
        int (*del)(struct event_base *, evutil_socket_t fd, short old, short events, void *fdinfo);
        int (*dispatch)(struct event_base *, struct timeval *);
        void (*dealloc)(struct event_base *);
        int need_reinit;
        enum event_method_feature features;
        size_t fdinfo_len;
    };

这里定义了一系列的回调函数`init`、`add`、等，在初始化的时候，会将IO复用的函数封装在这些回调里面，  
这个结构体就起到了一个跨平台的作用：

static const struct eventop *eventops[] = {
    #ifdef _EVENT_HAVE_EVENT_PORTS
        &evportops,
    #endif
    #ifdef _EVENT_HAVE_WORKING_KQUEUE
        &kqops,
    #endif
    #ifdef _EVENT_HAVE_EPOLL
        &epollops,
    #endif
    #ifdef _EVENT_HAVE_DEVPOLL
        &devpollops,
    #endif
    #ifdef _EVENT_HAVE_POLL
        &pollops,
    #endif
    #ifdef _EVENT_HAVE_SELECT
        &selectops,
    #endif
    #ifdef WIN32
        &win32ops,
    #endif
        NULL
    };

以epoll为例，在`epoll.c`文件中，分别以回调函数要求的参数形式封装了底层的调用：

const struct eventop epollops = {
        "epoll",//name
        epoll_init,//init
        epoll_nochangelist_add,//add
        epoll_nochangelist_del,//del
        epoll_dispatch,//dispacth
        epoll_dealloc,//dealloc
        1, /* need reinit */
        EV_FEATURE_ET|EV_FEATURE_O1,
        0
    };

因此，在libevent中，实现IO复用机制，都必须实现这些回调函数:初始化、销毁、注册、分发、注销

另外一个重要的结构就是`void *evbase`

这个指针，有点类似于上下文`context`，如果底层使用`epoll`，那么它将指向一个`epollop`结构，表示epoll调用需要的一些内容：

struct epollop {
        struct epoll_event *events;
        int nevents;
        int epfd;
    };

如果使用`select`将指向：

struct selectop 
    {
        int event_fds;      /* Highest fd in fd set */
        int event_fdsz;
        int resize_out_sets;
        fd_set *event_readset_in;
        fd_set *event_writeset_in;
        fd_set *event_readset_out;
        fd_set *event_writeset_out;
    };

## 事件处理主循环 ##

Libevent的事件主循环通过`event_base_loop()`完成的，其主要流程如下：

Created with Raphaël 2.1.0开始如果发现系统时间向后调整了，就校正系统时间根据timer heap和event的最小时间计算io复用的最大等待时间更新last wait time，并清空time cache调用系统IO复用等待就绪IO event检查信号，检查激活signal event，并把event插入到激活链表中将就绪的I/O event 插入到激活链表中检查heap中的timer event，将就绪的timer event从heap上删除，并插入到激活链表中结束

在应用程序代码中，调用`event_base_dispatch`，实际是对`event_base_loop`的简单的封装：

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

根据流程图描述的内容，仔细看看event\_base\_loop函数：

int
    event_base_loop(struct event_base *base, int flags)
    {
        const struct eventop *evsel = base->evsel;
        struct timeval tv;
        struct timeval *tv_p;
        int res, done, retval = 0;
    
        //清空时间缓存
        clear_time_cache(base);//base->tv_cache.tv_sec=0
    
        if (base->sig.ev_signal_added && base->sig.ev_n_signals_added)
            evsig_set_base(base);//将base赋值给evsignal_base
    
        done = 0;
    
        base->event_gotterm = base->event_break = 0;
    
        while (!done) {
       //事件循环
            base->event_continue = 0;
            //查看是否需要跳出循环，程序可以调用event_loopexit_cb()设置event_gotterm标记
            //调用event_base_loopbreak()设置event_break标记
            if (base->event_gotterm) {
                break;
            }
            if (base->event_break) {
                break;
            }
    
            //矫正系统时间
            timeout_correct(base, &tv);
    
            //根据timer_heap中事件的最小超时事件，设置IO复用接口的最大等待时间
            tv_p = &tv;
            if (!N_ACTIVE_CALLBACKS(base) && !(flags & EVLOOP_NONBLOCK)) {
                timeout_next(base, &tv_p);
            } else {
                /*
                 * if we have active events, we just poll new events
                 * without waiting.
                 */
                evutil_timerclear(&tv);
            }
    
            //如果没有注册事件就退出循环
            /*
            static int
            event_haveevents(struct event_base *base)
            {
                return (base->virtual_event_count > 0 || base->event_count > 0);
            }
            */
    
            if (!event_haveevents(base) && !N_ACTIVE_CALLBACKS(base)) {
                event_debug(("%s: no events registered.", __func__));
                retval = 1;
                goto done;
            }
    
            /* update last old time */
            //更新last wait time 并清空time cache
            gettime(base, &base->event_tv);
            clear_time_cache(base);
    
            //上面提到，tv_p表示timer heap中的最小时间
            //evsel->dispath最终将调用epoll_dispatch回调，
            //在epoll_dispach对epoll_wait等系统调用进行封装
            //跟进函数，发现最终会将就绪时间插入到激活链表中:
            //event_queue_insert(base, ev, EVLIST_ACTIVE);
            res = evsel->dispatch(base, tv_p);
    
            if (res == -1) {
                event_debug(("%s: dispatch returned unsuccessfully.",
                    __func__));
                retval = -1;
                goto done;
            }
    
            update_time_cache(base);
    
            //将堆中就绪的timer event从heap删除，并插入到激活链表中
            //在event_del_internal -> event_active_nolock
            timeout_process(base);
    
            if (N_ACTIVE_CALLBACKS(base)) {
            //处理就绪链表中的事件，调用event中的回调函数
                int n = event_process_active(base);
                if ((flags & EVLOOP_ONCE)
                    && N_ACTIVE_CALLBACKS(base) == 0
                    && n != 0)
                    done = 1;
            } else if (flags & EVLOOP_NONBLOCK)
                done = 1;
        }
        event_debug(("%s: asked to terminate loop.", __func__));
    
    done:
        clear_time_cache(base);
        base->running_loop = 0;
    
        EVBASE_RELEASE_LOCK(base, th_base_lock);
    
        return (retval);
    }

## IO、Timer和Signal事件统一 ##

先说说Timer，在muduo中使用timerfd\_\*，而在libevent中用最小堆(nginx里面用红黑树)的方式实现，这样IO复用调用（epoll）等，  
就算没有IO事件激活，也会返回处理超时事件。

Signal也是异步事件，但是集成Signal就复杂了，因为进程不知道什么时候发生异步信号。

Libevet的做法是，当异步信号发生时，先想办法（socketpair）让IO 复用调用返回，然后再统一和I/O事件，Timer事件一起处理