libevent 源碼解析-事件循環

時間  2019-11-06
標籤 libevent 源碼 解析 事件 循環 简体版
原文   原文鏈接

最近閱讀了 libevent 的源碼，寫一篇文章來總結本身學習到知識。使用libevent應該優先選用最新的穩定版本，而閱讀源碼爲了下降難度，我選擇了1.4的版本，也就是patches-1.4分支。讀這篇文章須要 Unix 網絡編程的基礎，知道 reactor 模式，若是對此還有疑問能夠看我這篇文章典型服務器模式原理分析與實踐html

libevent 的文件結構

關於 libevent 的文件結構這篇文章The Libevent Reference Manual: Preliminaries說明的比較清楚了，這裏簡要說明一下。react

event 和 event_basegit

event 和 event_base 是 libevent 的核心，也是咱們要探討的核心，主要圍繞兩個結構體類型 event 和 event_base 展開，event 定義了事件的結構，event_base 則是事件循環的框架，這兩個結構體分別定義在 event.h 和 event-internal.h 文件中。在 event.c 中定義了事件初始化，事件註冊，事件刪除等 API，還包含了事件循環框架 event base 相關的 API。github

evbuffer 和 buffereventredis

evbuffer 和 bufferevent 則處理了 libevent 中關於讀寫緩衝的問題，這兩個結構體也定義在 event.h 頭文件中，而相關 API 則分別在 buffer.c 文件和 evbuffer.c 文件中定義相關API。bufferevent 是一個緩衝區管理結構體，在其中包含了兩個 evbuffer 指針，一個是讀緩存區，一個是寫緩存區。evbuffer 則是與底層 IO 打交道的。另外不得不提的是 bufferevent 中爲讀緩存區和寫緩存區都設定了一個高水位和低水位，高水位是爲了不單個緩存區佔用過多的內存，低水位是爲了減小回調函數調用的次數，提升效率。編程

IO 多路複用系統緩存

libevent 是跨平臺的網絡庫，在不一樣平臺下實現 IO 多路的方式不同，即便在同一平臺下也可能有多種實現方式，libevent 支持 select，poll，epoll，kqueue 等方式。bash

util服務器

util 模塊就是一些公共方法了，好比日誌函數，時間處理函數等網絡

事件

libevent 將事件進一步抽象化了，除了讀和寫事件，還包括定時事件，甚至將信號也轉化成了事件來處理。首先看一下 event 的結構體。

  1. libevent 用鏈表來保存註冊事件和激活的事件，ev_next 是全部註冊事件的鏈表，ev_active_next 是激活事件的鏈表，ev_signal_next 是信號事件的鏈表。時間事件用最小堆來管理，用最小堆是很是高效的方式，每次只須要判斷堆頂的事件，若是堆頂的時間事件都沒有就緒，那麼後面的時間也必定沒有就緒。

  2. 每一個事件循環就有一個 event_base，用來調度事件，ev_base 指向該事件所在的事件循環。

  3. ev_events 則表示該事件所關心的事件類型，能夠是如下幾種狀況：

// 時間事件
#define EV_TIMEOUT 0x01
// 可讀事件
#define EV_READ 0x02
// 可寫事件
#define EV_WRITE 0x04
// 信號
#define EV_SIGNAL 0x08
// 標識是否爲永久事件。非永久事件激活一次後，就會從註冊隊列中刪除，若是想繼續監聽該事件，須要再次加入事件隊列。而永久事件則激活後不會從註冊事件中刪除，除非本身手動刪除。
#define EV_PERSIST 0x10 /* Persistant event */
複製代碼

  1. 若是該事件是個時間事件，那麼 ev_timeout 就是這個事件的超時時長。

  2. libevent 的事件可使用優先級，優先級高的事件老是先響應，ev_pri 就是該事件的優先級。

  3. ev_callback 是該事件對應的回調函數，當事件被觸發後會調用該回調函數進行處理。

struct event {
    /*
    ** libevent 用雙向鏈表來保存註冊的全部事件，包括IO事件，信號事件。
    ** ev_next 存儲了該事件在事件鏈表中的位置
    ** 另外，libevent 還用另外一個鏈表來存儲激活的事件，經過遍歷激活的事件鏈表來分發任務
    ** ev_active_next 存儲了該事件在激活事件鏈表中的位置
    ** 相似，ev_signal_next 就是該事件在信號事件鏈表中的位置
    */
	TAILQ_ENTRY (event) ev_next;
	TAILQ_ENTRY (event) ev_active_next;
	TAILQ_ENTRY (event) ev_signal_next;
    /* libevent 用最小堆來管理超時時間，min_heap_idx 保存堆頂的 index */
	unsigned int min_heap_idx;	/* for managing timeouts */

    /* event_base 是整個事件循環的核心，每一個 event 都處在一個 event_base 中，ev_base 保存這個結構體的指針 */
	struct event_base *ev_base;
    /* 對於 IO 事件，ev_fd 是綁定的文件描述符，對於 signal 事件，ev_fd 是綁定的信號 */
	int ev_fd;
    /* 要處理的事件類型， */
	short ev_events;
    /* 事件就緒執行時，調用ev_callback的次數，一般爲1 */
	short ev_ncalls;
	short *ev_pncalls;	/* Allows deletes in callback */
    /* 事件超時的時間長度 */
	struct timeval ev_timeout;
    /* 優先級 */
	int ev_pri;		/* smaller numbers are higher priority */
    /* 響應事件時調用的callback函數 */
	void (*ev_callback)(int, short, void *arg);
	void *ev_arg;

	int ev_res;		/* result passed to event callback */
    /* 表示事件所處的狀態 */
	int ev_flags;
};
複製代碼

對於事件的處理主要有三個 API：event_set，event_add，event_del

event_set event_set 用來初始化一個event對象

void event_set(struct event *ev, int fd, short events,
      void (*callback)(int, short, void *), void *arg)
{
    /* Take the current base - caller needs to set the real base later */
    /* current_base 是一個全局變量，ev_base 會默認指向這個變量，
    ** 以後 ev_base 也能夠經過 event_base_set 設置指向指定的 event_base 
    ** 特別是對於一個進程中有多個 event_base 的狀況下，須要綁定到指定的 event_base 上*/
    ev->ev_base = current_base;

    ev->ev_callback = callback;
    ev->ev_arg = arg;
    ev->ev_fd = fd;
    ev->ev_events = events;
    ev->ev_res = 0;
    ev->ev_flags = EVLIST_INIT;
    ev->ev_ncalls = 0;
    ev->ev_pncalls = NULL;

    min_heap_elem_init(ev);

    /* by default, we put new events into the middle priority */
    /* 設定默認優先級爲最大優先級的一半 */
    if(current_base)
        ev->ev_pri = current_base->nactivequeues/2;
}
複製代碼

event_add event_add 則是像事件隊列中添加註冊的事件，若是該事件監聽在讀事件、寫事件或者信號上，那麼就會將其先添加到IO多路複用系統中，而後再加入到註冊事件鏈表中。若是參數 tv 不爲 NULL，還會將該事件註冊到時間事件的最小堆上。

int event_add(struct event *ev, const struct timeval *tv)
{
    // 要註冊的evbase
    struct event_base *base = ev->ev_base;
    const struct eventop *evsel = base->evsel;
    void *evbase = base->evbase;
    int res = 0;

    event_debug((
         "event_add: event: %p, %s%s%scall %p",
         ev,
         ev->ev_events & EV_READ ? "EV_READ " : " ",
         ev->ev_events & EV_WRITE ? "EV_WRITE " : " ",
         tv ? "EV_TIMEOUT " : " ",
         ev->ev_callback));

    // 校驗沒有設其餘的標誌位
    assert(!(ev->ev_flags & ~EVLIST_ALL));

    /*
     * prepare for timeout insertion further below, if we get a
     * failure on any step, we should not change any state.
     */
    // 分配最小堆插入一個元素的內存，先分配內存是爲了保證時間事件
    if (tv != NULL && !(ev->ev_flags & EVLIST_TIMEOUT)) {
        if (min_heap_reserve(&base->timeheap,
            1 + min_heap_size(&base->timeheap)) == -1)
            return (-1);  /* ENOMEM == errno */
    }

    /* ev_events 監聽的事件類型爲讀寫或者信號 並且 該事件沒有被註冊過，也不在激活隊列裏 */
    if ((ev->ev_events & (EV_READ|EV_WRITE|EV_SIGNAL)) &&
        !(ev->ev_flags & (EVLIST_INSERTED|EVLIST_ACTIVE))) {
        /* 將事件註冊到 IO 多路複用中 */
        res = evsel->add(evbase, ev);
        if (res != -1)
            /* 註冊成功後將事件加入到 event_base 的事件鏈表中 */
            event_queue_insert(base, ev, EVLIST_INSERTED);
    }

    /* 
     * we should change the timout state only if the previous event
     * addition succeeded.
     */
    if (res != -1 && tv != NULL) {
        struct timeval now;

        /* 
         * we already reserved memory above for the case where we
         * are not replacing an exisiting timeout.
         */
        /* 若是事件已經在定時事件中了，則從時間事件鏈表中刪除該事件 */
        if (ev->ev_flags & EVLIST_TIMEOUT)
            event_queue_remove(base, ev, EVLIST_TIMEOUT);

        /* Check if it is active due to a timeout.  Rescheduling
         * this timeout before the callback can be executed
         * removes it from the active list. */
        /* 若是事件已經在激活隊列中，則從激活隊列中刪除該事件 */
        if ((ev->ev_flags & EVLIST_ACTIVE) &&
            (ev->ev_res & EV_TIMEOUT)) {
            /* See if we are just active executing this
             * event in a loop
             */
            if (ev->ev_ncalls && ev->ev_pncalls) {
                /* Abort loop */
                *ev->ev_pncalls = 0;
            }
            
            event_queue_remove(base, ev, EVLIST_ACTIVE);
        }
        // 獲取當前時間
        gettime(base, &now);
        // 計算超時時間
        evutil_timeradd(&now, tv, &ev->ev_timeout);

        event_debug((
             "event_add: timeout in %ld seconds, call %p",
             tv->tv_sec, ev->ev_callback));
        // 插入到定時時間事件隊列中
        event_queue_insert(base, ev, EVLIST_TIMEOUT);
    }

    return (res);
}
複製代碼

事件循環

介紹完事件則介紹一下事件調度的核心 event_base，event_base 定義在頭文件 event-internal.h 中。

首先在 event_base 中有一個成員 evsel，該成員保存了 IO 多路複用資源的函數指針，eventop 的結構以下：

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;
};
複製代碼

對於每一種 IO 多路複用都實現了 init, add, del, dispatch 幾種方法，init 就是初始化，add 就是添加事件，del 就是刪除事件，diapatch 的就等待事件被激活，並分別處理激活的事件。在 event_base 中還有一個成員 evbase，這個成員保存了 IO 多路複用的資源。好比 add 函數的第一個參數是 void*，這個 void* 就是要傳入 evbase 的。實際上這種作法就是經過 C 的函數指針來實現了多態，若是是面向對象的語言就不用搞這麼複雜了。不過這種 C 實現多態的方法仍是值得咱們學習的。

event_base 中有一個成員 activequeues 須要說明一下，這是一個指向指針的指針。以前說過激活隊列是有優先級的，同一優先級的激活事件放在一個鏈表中，那麼多個不一樣優先級的激活隊列的頭節點也就組成了一個隊列。所以，這裏是指向指針的指針。

struct event_base {
    /* eventop 對象指針，決定了使用哪一種IO多路複用資源 
    ** 可是 eventop 實際上只保存了函數指針，最後資源的句柄是保存在 evbase 中。
    ** 好比要使用 epoll，那麼就應該有一個 epoll 的文件描述符，eventop 中只保存了epoll相關的add，del等函數
    ** epoll 的文件描述符是保存在 evbase 中的，所以調用的形式就是 evsel->add(evbase, ev);
    */
    const struct eventop *evsel;
    void *evbase;
    /* event base 上全部事件的數量包括註冊事件和激活事件
    ** 在 event_queue_insert 函數中加 1 */
    int event_count;        /* counts number of total events */
    /* event base 上被激活的事件的數量 */
    int event_count_active; /* counts number of active events */

    int event_gotterm;      /* Set to terminate loop */
    int event_break;        /* Set to terminate loop immediately */

    /* active event management */
    /* libevent 支持事件的優先級，對於激活的事件，不一樣優先級的事件存儲在不一樣的鏈表中 
    ** 而後再用一個鏈表把這些鏈表串起來
    */
    struct event_list **activequeues;
    /* 事件能夠設定的最大優先級 */
    int nactivequeues;

    /* signal handling info */
    struct evsignal_info sig;
    /* 保存全部註冊事件的鏈表 */
    struct event_list eventqueue;
    /* 上一次進行事件循環的時間 */
    struct timeval event_tv;
    /* 管理時間事件的小頂堆 */
    struct min_heap timeheap;

    struct timeval tv_cache;
};
複製代碼

關於 event_base 主要相關的有如下幾個函數：event_base_new，event_base_free，event_base_loop

event_base_new 和 event_base_free 分別就是分配 event_base 資源和釋放 event_base 資源，比較好理解。經過 event_base_new 首先建立 event_base，而後建立不一樣的事件並註冊到 event_base 中，最後經過 event_base_loop 啓動事件循環。若是要退出事件循環，能夠調用 event_base_loopbreak 或 event_loopexit_cb。

事件循環的核心是調用 IO 複用的 dispatch 函數，須要注意的是在調用 dispatch 函數以前會先計算出最近一個時間事件距離如今還有多久，而後將這個時間差做爲 dispatch 阻塞的時間，這樣時間事件就能夠被及時響應，不會由於阻塞在 IO 多路複用上過久而等待太多時間。這是一種常見的作法，redis 的事件循環也是這樣實現的。

int
event_base_loop(struct event_base *base, int flags)
{
    const struct eventop *evsel = base->evsel;
    void *evbase = base->evbase;
    struct timeval tv;
    struct timeval *tv_p;
    int res, done;

    /* clear time cache */
    base->tv_cache.tv_sec = 0;

    if (base->sig.ev_signal_added)
        evsignal_base = base;
    done = 0;
    while (!done) {
        /* Terminate the loop if we have been asked to */
        /* 調用 event_loopexit_cb 跳出循環，爲何搞了兩個函數？ */
        if (base->event_gotterm) {
            base->event_gotterm = 0;
            break;
        }

        /* 調用 event_base_loopbreak 函數跳出循環 */
        if (base->event_break) {
            base->event_break = 0;
            break;
        }

        /* You cannot use this interface for multi-threaded apps */
        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;
        /* 若是沒有激活事件，且等待方式不是非阻塞，計算當前時間距離最小堆堆頂時間事件的時間差，做爲阻塞的時間 */
        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.
             */
            /* 若是有激活事件，將阻塞時間設置爲 0 */
            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;
        /* 調用 IO 多路複用函數等待事件就緒，就緒的信號事件和IO事件會被插入到激活鏈表中 */
        res = evsel->dispatch(base, evbase, tv_p);

        if (res == -1)
            return (-1);
        /* 寫時間緩存 */
        gettime(base, &base->tv_cache);
        /* 檢查heap中的時間事件，將就緒的事件從heap中刪除並插入到激活隊列中 */
        timeout_process(base);
        /* 若是有激活的信號事件和IO時間，則處理 */
        if (base->event_count_active) {
            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);
}
複製代碼

總結

libevent 的事件循環的核心就是以上描述的這些了，對於細節地方的實如今個人 github 上給出了相應的中文註釋。另外，網絡上有一副圖很直觀的描述了 libevent 的事件循環，我將其從新畫了一遍，稍微修改了一下，貼出來和你們交流。

libevent 事件循環
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程