Libevent的IO復用技術和定時事件原理

Libevent 是一個用C語言編寫的、輕量級的開源高性能網絡庫，主要有以下幾個亮點：事件驅動（ event-driven），高性能;輕量級，專注於網絡，不如 ACE 那麼臃腫龐大；源代碼相當精煉、易讀；跨平台，支持 Windows、 Linux、 *BSD 和 Mac Os；支持多種 I/O 多路復用技術， epoll、 poll、 dev/poll、 select 和 kqueue 等；支持 I/O，定時器和信號等事件；注冊事件優先級。

1 Libevent中的epoll

　　Libevent重要的底層技術之一就是IO復用函數，比如Linux的epoll、Windows下的select。Libevent的epoll相關的函數在epoll.c文件中，為了方便使用epoll對事件的操作，定義了一個epollop結構體。

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

　　其中，events指針用於存放就緒的事件，也就是內核會拷貝就緒的事件到這個events指向的內存中；nevents表示events指向的內存為多大，也就是可以存放多少個epoll_event類型的數據；epfd也就是調用epoll_create()返回的內核事件表對應的描述符。

　　Libevent為了封裝IO復用技術，定義了一個統一的事件操作結構體eventop：

/** Structure to define the backend of a given event_base. */
struct eventop {
    /* 後端IO復用技術的名稱 */
    const char *name;
    
    void *(*init)(struct event_base *);
    int (*add)(struct event_base *, evutil_socket_t fd, short old, short events, void *fdinfo);
    int (*del)(struct event_base *, evutil_socket_t fd, short old, short events, void *fdinfo);
    int (*dispatch)(struct event_base *, struct timeval *);
    
    /* 釋放IO復用機制使用的資源 */
    void (*dealloc)(struct event_base *);
    
    int need_reinit;
    
    /* IO復用機制支持的一些特性，可選如下3個值的按位或:EV_FEATURE_ET(支持邊沿觸發事件EV_ET)、
     * EV_FEATURE_O1(事件監測算法的復雜度為O(1))和EV_FEATURE_FDS(不僅能監聽socket上的事件，還能
     * 監聽其他類型的文件描述符上的事件) */
    enum event_method_feature features;
    
    /* 有些IO復用機制需要為每個IO事件隊列和信號事件隊列分配額外的內存，以避免同一個文件描述符被重復
     * 插入IO復用機制的事件表中。evmap_io_add(或evmap_io_del)函數最親愛調用eventop的add(或del)方法時，
     * 將這段內存的起始地址作為第5個參數傳遞給add(或del)方法。fdinfo_len指定了該段內存的長度 */ 
    size_t fdinfo_len;
};

　　對於epoll來說，封裝的事件操作為：

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

　　結構體中的函數都是在epoll.c中定義好的，並且都是static的，但是只需要並且也只能通過epollops變量來調用這些函數了，epoll相關的函數就不在贅述，詳情可以參考源代碼。那麼Libevent是什麼時候來獲取這個變量的值呢？秘密就在event.c文件中，其中定義的eventops數組包含了支持的所有IO復用技術，當然包括我們講的epoll了。

/* Libevent通過遍歷eventops數組來選擇其後端IO復用技術，遍歷的順序是從數組的第一個元素開始，
 * 到最後一個元組結束。Linux系統下，默認選擇的後端IO復用技術是epoll。*/
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
};

　　在event_base的初始化函數event_base_new_with_config中，會遍歷eventops數組，選擇其中符合要求的IO復用機制，然後退出遍歷過程，這樣event_base就選擇了一個後端的IO復用機制，比如Libevent在Linux下默認是使用epoll的。

for (i = 0; eventops[i] && !base->evbase; i++) {
    // ...

    /* also obey the environment variables */
    if (should_check_environment &&
        event_is_method_disabled(eventops[i]->name))
        continue;

    /* base->evsel記錄後端IO復用機制 */
    base->evsel = eventops[i];

    /* 指向IO復用機制真正存儲的數據，它通過evsel成員的init函數來進行初始化 */
    /* 比如epoll時，evbase指向的是epollop */
    base->evbase = base->evsel->init(base);
}

 　　到這裡為止，Libevent已經初始化好了一種後台IO復用機制技術，這裡以epoll為例，其他IO復用技術流程也類似。

2 Libevent的定時事件原理

　　Libevent的定時事件也是要"加入"到Libevent中的IO復用框架中的，比如我們需要定時5秒鐘，那麼等到5秒鐘之後就可以執行對應設置的回調函數了。以下是使用Libevent實現一個簡單的定時器應用：

#include <iostream>

#include <event.h>
#include <event2/http.h>

using namespace std;

// Time callback function
void onTime(int sock, short event, void *arg)
{
    static int cnt = 0;
    cout << "Game Over! " << cnt++ << endl;

    struct timeval tv;
    tv.tv_sec = 1;
    tv.tv_usec = 0;
    if (cnt < 5) {
        // Add timer event
        event_add((struct event *) arg, &tv);
    }
    else {
        cout << "onTime is over" << endl;
    }
}

int main(int argc, char **argv)
{
    cout << event_get_version() << endl;

    struct event_base *base = event_init();
    struct event ev;

    evtimer_set(&ev, onTime, &ev);

    struct timeval timeevent;
    timeevent.tv_sec = 1;
    timeevent.tv_usec = 0;

    event_add(&ev, &timeevent);

    // Start event loop
    event_base_dispatch(base);
    event_base_free(base);

    return 0;
}

　　定時器事件會被加入到一個時間堆（小堆結構）中，每次執行事件等待函數時，對於epoll來說就是epoll_wait函數了，把時間堆上最小節點的時間值賦值給該函數，這樣如果有事件來臨或者是時間超時了，都會返回。然後判斷當前時間和調用事件等待函數之前的時間差是否大於或等於時間堆上最小節點的時間值，如果條件成立就執行對應的時間回調函數，這樣就完成了一個定時事件。下面代碼就是在事件監聽循環中的部分代碼。

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;
    if (!N_ACTIVE_CALLBACKS(base) && !(flags & EVLOOP_NONBLOCK)) {
        /* 獲取時間堆上堆頂元素的超時值，即IO復用系統調用本次應該設置的超時值 */
        timeout_next(base, &tv_p);
    } else {
        /*
         * if we have active events, we just poll new events
         * without waiting.
         */
        /* 如果有就緒事件尚未處理，則將IO復用系統調用的超時時間置0
         * 這樣IO復用系統調用就直接返回，程序也就可以直接處理就緒事件了 */
        evutil_timerclear(&tv);
    }

    /* If we have no events, we just exit */
    /* 如果event_base中沒有注冊任何事件，則直接退出事件循環 */
    if (!event_haveevents(base) && !N_ACTIVE_CALLBACKS(base)) {
        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方法等待事件 */
    res = evsel->dispatch(base, tv_p);

    // 超時時間返回值為0
    if (res == -1) {
        event_debug(("%s: dispatch returned unsuccessfully.",
            __func__));
        retval = -1;
        goto done;
    }

    update_time_cache(base);    /* 將系統緩存更新為當前系統時間 */

    timeout_process(base);    /* 檢查時間堆上的到期事件並以此執行之 */

    if (N_ACTIVE_CALLBACKS(base)) {
        /* 調用event_process_active函數依次處理就緒的信號事件和IO事件 */
        /* 這裡也可能有定時事件 */
        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;
} 

參考：

　　1、Libevent初探

　　2、Libevent源碼