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

時間 2019-12-01

標籤 libevent 技術定時事件原理简体版

原文原文鏈接

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

1 Libevent中的epoll

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

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

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

　　Libevent爲了封裝IO複用技術，定義了一個統一的事件操做結構體eventop：github

/** 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了。windows

/* 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;
}