libevent源碼剖析

時間 2019-11-21

標籤 libevent 源碼剖析简体版

原文原文鏈接

　　libevent是一個使用C語言編寫的，輕量級的開源高性能網絡庫，使用者不少，研究者也不少。因爲代碼簡潔，設計思想簡明巧妙，所以很適合用來學習，提高本身C語言的能力。數組

　　libevent有這樣顯著地幾個亮點：緩存

　　　　1.事件驅動，高性能安全

　　　　2.輕量級，專一於網絡，不如ACE那麼龐大臃腫網絡

　　　　3.代碼精煉易讀多線程

　　　　4.跨平臺，支持Windows,Linux,*BSD和Mac Os;框架

　　　　5.支持多種IO多路複用技術，epoll,poll，dev/poll、select和kqueue等異步

　　　　6.支持IO，定時器和信號等事件socket

　　　　7.註冊事件優先級ide

　　基於以上優勢，libevent已經被普遍的應用，做爲底層的網絡庫；好比memcached、Vomit、Nylon、Netchat等。memcached

　　下面咱們就來從程序的基本使用場景和代碼的總體處理流程入手來對libevent庫進行學習。

　　　　　當應用程序向libevent註冊一個事件後,libevent內部是怎樣處理的呢，如下是基本流程：

　　　　　　　　1)首先應用程序準備並初始化event，設置好事件類型和回調函數

　　　　　　　　2)向libevent添加該事件event。

　　　　　　　　3)程序調用event_base_dispatch()系列函數進入無線循環，等待事件.

　　　　　這只是大概的流程，如下是對於流程中的一些基本概念的講解：

　　　　　1、事件event

　　　　　　libevent是基於事件驅動的，從名字上能夠看出event是整個庫的核心。首先給出event結構體的聲明,它位於event.h文件中：

 1 struct event {
 2  TAILQ_ENTRY (event) ev_next;
 3  TAILQ_ENTRY (event) ev_active_next;
 4  TAILQ_ENTRY (event) ev_signal_next;
 5  unsigned int min_heap_idx; /* for managing timeouts */
 6  struct event_base *ev_base;
 7  int ev_fd;
 8  short ev_events;
 9  short ev_ncalls;
10  short *ev_pncalls; /* Allows deletes in callback */
11  struct timeval ev_timeout;
12  int ev_pri;  /* smaller numbers are higher priority */
13  void (*ev_callback)(int, short, void *arg);
14  void *ev_arg;
15  int ev_res;  /* result passed to event callback */
16  int ev_flags;
17 };

　　　　　　　　1) ev_events：event關注的事件類型，它能夠是如下3種類型：

　　　　　　　　　　IO事件：EV_WRITE和EV_READ

　　　　　　　　　　定時事件：EV_TIMEOUT

　　　　　　　　　　信號：EV_SIGNAL

　　　　　　　　　　輔助選項：EV_PERSIST，代表是一個永久事件

　　　　　　　　2) ev_next,ev_active_next,ev_signal_next都是雙向鏈表節點指針；他們是libevent對不一樣事件類型和在不一樣的時期，對事件的管理時使用到的字段。

　　　　　　　　　　libevent使用雙向鏈表保存全部註冊的IO和Signal事件，ev_next就是該IO事件在鏈表中的位置，稱此鏈表爲「已註冊事件鏈表」；

　　　　　　　　　　一樣ev_signal_next就是signal事件的signal事件鏈表中的位置。

　　　　　　　　　　ev_active_next:libevent將全部激活事件放入到鏈表active list中，而後遍歷active list執行調度。

　　　　　　　　　　每當事件event轉變爲就緒狀態時，libevent就會把它移入到active event list[priority]中，其中priority是event的優先級；接着libevent會根據本身的調度策略選擇就緒事件，調用其callback()函數執行事件處理；並根據就緒的句柄和時間類型填充callback函數的參數。

　　　　　　　　3)min_heap_idx和ev_timeout：若是事件是timeout事件，他們是event在小根堆中的索引和超時值，libevent使用小根堆來管理定時事件

　　　　　　　　4)ev_base：該事件所屬反應堆實例，這是一個event_base結構體。

　　　　　　　　5)ev_fd：對於IO事件，這是綁定的文件描述符，對於signal事件，是綁定的信號

　　　　　　　　6)ev_callback：event的回調函數，被ev_base調用，執行事件處理程序，這是一個函數指針，原型爲：

　　　　　　　　　　void (*ev_callback)(int fd,short events,void* arg)

　　　　　　　　　其中參數fd對應於ev_fd；events對應於events；arg對應於ev_arg；

　　　　　　　　7)ev_arg：void*，代表能夠是任意類型的數據，在設置event時指定；

　　　　　　　　8)eb_flags：libevent用於標記event信息的字段，代表其當前的狀態，可能的值有：

　　　　　　　　　　#define EVLIST_TIMEOUT 0x01 // event在time堆中
　　　　　　　　　　#define EVLIST_INSERTED 0x02 // event在已註冊事件鏈表中
　　　　　　　　　　#define EVLIST_SIGNAL 0x04 // 未見使用
　　　　　　　　　　#define EVLIST_ACTIVE 0x08 // event在激活鏈表中
　　　　　　　　　　#define EVLIST_INTERNAL 0x10 // 內部使用標記
　　　　　　　　　　#define EVLIST_INIT 0x80 // event已被初始化

　　　　　　　　9)ev_ncalls：事件就緒執行時，調用ev_callback的次數，一般爲1

　　　　　　　　10)ev_res：記錄了當前激活事件的類型

　　　　　　要想向libevent添加一個事件，首先須要設置event對象，這經過調用libevent提供的函數有：event_set()，event_base_set()，event_priority_set()來完成；下面分別講解：

　　　　　　void event_set(struct event *ev,int fd,short events,void (*callback)(int,short,void*),void *arg)

　　　　　　　　ev:執行要初始化的event對象

　　　　　　　　fd:對於信號來講是綁定的文件描述符，對於信號來講是綁定的signal信號

　　　　　　　　events:在該fd上關注的事件類型，它能夠是EV_READ,EV_WRITE,EV_SIGNAL；

　　　　　　　　callback:這是一個函數指針，當fd上的事件event發生時，調用該函數執行處理

　　　　　　　　arg:傳遞給callback函數指針的參數

　　　　　　int event_base_set(struct event_base* base,struct event *ev)

　　　　　　　　設置event ev將要註冊到的evnet_base;

　　　　　　int event_priority_set(struct event* ev,int pri)

　　　　　　　　設置event ev的優先級，注意，當ev正處於就緒狀態時，不能設置，返回-1.

　　　　2、事件處理框架event_base

　　　　　　如下是base_event結構體的聲明,它位於event-internal.h文件中：

 1 struct event_base {
 2  const struct eventop *evsel;
 3  void *evbase;　
 4  int event_count;  /* counts number of total events */
 5  int event_count_active; /* counts number of active events */
 6  int event_gotterm;  /* Set to terminate loop */
 7  int event_break;  /* Set to terminate loop immediately */
 8  /* active event management */
 9  struct event_list **activequeues;
10  int nactivequeues;
11  /* signal handling info */
12  struct evsignal_info sig;
13  struct event_list eventqueue;
14  struct timeval event_tv;
15  struct min_heap timeheap;
16  struct timeval tv_cache;
17 };

　　　　　　　　如下是結構體各字段的含義：

　　　　　　　　1)evsel和evbase這兩個字段的設置可能會讓人有些迷惑，這裏咱們能夠把evsel和evbase看作是類和靜態函數的關係，好比添加事件時的調用行爲：evsel->add(evbase,ev)，實際執行操做的是evbase，這至關於class::add(instance,ev)，instance就是class的一個對象實例。

　　　　　　　　2)activequeues是一個二級指針，前面講過libevent支持事件優先級，所以你能夠你把它看作是數組，其中的元素activequeues[priority]是一個鏈表，鏈表的每一個節點指向一個優先級爲priority的就緒事件event。

　　　　　　　　3)eventqueue：鏈表，保存了全部的註冊事件event的指針。

　　　　　　　　4)sig是用來管理信號的結構體

　　　　　　　　5)timeheap是管理定時事件的小根堆

　　　　　　　　6)event_tv和tv_cache是libevent用於事件管理的變量

　　　　　　咱們已經對event_base有了一個初步的瞭解，那麼event_base如何建立和初始化的呢？

　　　　　　　　建立一個event_base對象也便是建立了一個新的libevent實例，程序須要經過調用event_init()函數來建立，該函數首先爲event_base實例申請空間，而後初始化timer mini-heap，選擇並初始化合適的系統多路複用機制，初始化各事件的鏈表；函數還檢測了系統的時間設置，爲後面的事件管理打下了基礎

　　　　　3、事件主循環

　　　　　　　　libevent將IO事件、定時器和信號事件處理很好的結合到了一塊兒，那麼它是如何作到的呢？

　　　　　　　　libevent的事件主循環主要是經過event_base_loop()函數完成的，主要操做流程以下圖所示，event_base_loop所作的就是持續執行下面的循環

　　　　　　　　下面是源碼，能夠參考

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;
    // 清空時間緩存
    base->tv_cache.tv_sec = 0;
    // evsignal_base是全局變量，在處理signal時，用於指名signal所屬的event_base實例
    if (base->sig.ev_signal_added)
        evsignal_base = base;
    done = 0;
    while (!done) { // 事件主循環
        // 查看是否須要跳出循環，程序能夠調用event_loopexit_cb()設置event_gotterm標記
        // 調用event_base_loopbreak()設置event_break標記
        if (base->event_gotterm) {
            base->event_gotterm = 0;
            break;
        }
        if (base->event_break) {
            base->event_break = 0;
            break;
        }
        // 校訂系統時間，若是系統使用的是非MONOTONIC時間，用戶可能會向後調整了系統時間
        // 在timeout_correct函數裏，比較last wait time和當前時間，若是當前時間< last wait time
        // 代表時間有問題，這是須要更新timer_heap中全部定時事件的超時時間。
        timeout_correct(base, &tv);
   
        // 根據timer heap中事件的最小超時時間，計算系統I/O demultiplexer的最大等待時間
        tv_p = &tv;
        if (!base->event_count_active && !(flags & EVLOOP_NONBLOCK)) {
            timeout_next(base, &tv_p);
        } else {
            // 依然有未處理的就緒時間，就讓I/O demultiplexer當即返回，沒必要等待
            // 下面會提到，在libevent中，低優先級的就緒事件可能不能當即被處理
            evutil_timerclear(&tv);
        }
        // 若是當前沒有註冊事件，就退出
        if (!event_haveevents(base)) {
            event_debug(("%s: no events registered.", __func__));
            return (1);
        }
        // 更新last wait time，並清空time cache
        gettime(base, &base->event_tv);
        base->tv_cache.tv_sec = 0;
        // 調用系統I/O demultiplexer等待就緒I/O events，多是epoll_wait，或者select等；
        // 在evsel->dispatch()中，會把就緒signal event、I/O event插入到激活鏈表中
        res = evsel->dispatch(base, evbase, tv_p);
        if (res == -1)
            return (-1);
        // 將time cache賦值爲當前系統時間
        gettime(base, &base->tv_cache);
        // 檢查heap中的timer events，將就緒的timer event從heap上刪除，並插入到激活鏈表中
        timeout_process(base);
        // 調用event_process_active()處理激活鏈表中的就緒event，調用其回調函數執行事件處理
        // 該函數會尋找最高優先級（priority值越小優先級越高）的激活事件鏈表，
        // 而後處理鏈表中的全部就緒事件；
        // 所以低優先級的就緒事件可能得不到及時處理；
        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;
    }
    // 循環結束，清空時間緩存
    base->tv_cache.tv_sec = 0;
    event_debug(("%s: asked to terminate loop.", __func__));
    return (0);
}

View Code

　　　　　　　I/O和Timer事件的統一

　　　　　　　　　libevent將Timer和Signal事件都統一到了系統的IO多路複用機制中了從上面的流程圖中咱們能夠看出一點端倪，那麼libevent是如何作到的呢？

　　　　　　　　　首先是I/O和Timer事件的統一。系統的I/O機制向select()和epoll_wait()都容許程序制定一個最大等待時間timeout,即便沒有事件發生，他們也能保證在timeout時間內返回。那麼根據全部Timer事件的最小超時時間來設置系統I/O的timeout時間；當系統I/O返回時，再激活全部就緒的Timer事件就能夠了，這樣就能將Timer事件完美的融合到系統的I/O機制中了。

　　　　　　　　　libevent使用堆來管理Timer事件，其key值就是事件的超時時間，根據堆中具備最小超時值的事件和當前時間來計算等待時間。

　　　　　　　I/O和Signal事件的統一

　　　　　　　　　Signal是異步事件的經典，將Signal事件統一到系統的I/O多路複用中就不像Timer事件那麼天然了。Signal事件的出現對於進程來說徹底是隨機的，進程不能只是測試一個變量來判別是否發生了一個信號，而是必須告訴內核「在此信號發生時，請執行以下操做」。當Signal發生時，系統並不當即調用event的callback函數處理信號，而是設法通知系統的I/O機制，讓其返回，而後再統一和I/O事件以及Timer一塊兒處理。

　　　　　　　　　　那麼,系統是如何設法通知系統的I/O機制呢?基本的方法就是採用「消息機制」。在libevent中是經過socket pair完成的。socket pair就是一個socket對，一個讀socket，一個寫socket。Socket pair建立好了以後，讀socket在libevent的event_base實例上註冊了一個persist的讀事件。這樣當寫socket寫入數據時，讀socket就會相應的獲得通知了。前面提到過,libevent會在事件主循環中檢查標記，來肯定是否有觸發的Signal，若是標記被設置就處理這些signal。這段代碼在各個具體的I/O機制中，以epoll爲例，在epoll_dispatch()函數中，代碼片斷以下：

　　　　　　　　　　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事件
       　　　　　　　　 return (0);
   　　　　　　　　 } else if (base->sig.evsignal_caught) {
      　　　　　　　　 evsignal_process(base);// 處理signal事件
   　　　　　　　　 }

　　　　　　註冊、註銷signal事件

　　　　　　　　註冊signal事件是經過evsignal_add(struct event *ev)函數完成的,libevent對全部的信號註冊同一個處理函數evsignal_handler()，該函數註冊過程以下：

　　　　　　　　1.取得ev要註冊到的信號signo；

　　　　　　　　2.若是信號signo未被註冊，那麼就爲signo註冊信號處理函數evsignal_handler();

　　　　　　　　3.若是事件ev_signal尚未註冊，就註冊ev_signal事件；

　　　　　　　　4.將事件ev添加到signo的event鏈表中

　　　　　　　　註銷一個已註冊的signal事件就更簡單了，直接從其已註冊事件的鏈表中移除便可。若是鏈表已空，那麼就恢復舊有的處理函數；處理函數的evsignal_handler()函數就是記錄信號的發生次數，並通知event_base有信號觸發。

　　　　　3、支持多路複用

　　　　　　　　libevent的核心是事件驅動、同步非阻塞，爲了達到這一目標必須採用系統提供的I/O多路複用技術，而這些複用技術在不一樣的平臺上卻各有不一樣，如何能提供統一的支持方式呢？

　　　　　　　　libevent支持多種I/O多路複用的關鍵就在於結構體evnetop，這個結構體前面也提到過，他的成員是一系列函數指針，定義在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;
　　　　　　　　};

　　　　　　　在libevent中，每種IO多路複用技術的實現都必須提供這五種函數接口來完成自身的初始化、銷燬釋放；對事件的註冊、註銷和分發。好比對於epoll，libevent實現了5個對應的接口函數，並在初始化時經eventop的5個函數指針指向這5個函數，那麼程序就可使用epoll做爲IO多路複用機制了。

　　　　　4、時間管理

　　　　　　　　爲了支持定時器，libevent必須需和系統時間打交道主要涉及到時間的加減輔助函數、時間緩存、時間校訂和定時器堆的時間值調整等。

　　　　　　　　libevent在初始化時會檢測系統時間的類型，經過調用detect_monotonic()完成，它經過clock_gettime()來檢測系統是否支持monotonic時鐘類型

　　　　　　　　　　static void detect_monotonic(void)
　　　　　　　　　　{
　　　　　　　　　　　　#if defined(HAVE_CLOCK_GETTIME) && defined(CLOCK_MONOTONIC)
   　　　　　　　　　　　 struct timespec    ts;
　　　　　　　　　　　   if (clock_gettime(CLOCK_MONOTONIC, &ts) == 0)
     　　　　　　　　　　   use_monotonic = 1; // 系統支持monotonic時間
　　　　　　　　　　　　#endif
　　　　　　　　　　}

　　　　　　　　monotonic事件值得是系統從boot後到如今的時間，若是系統支持monotonic事件就將全局變量use_monotonic設置爲1。

　　　　　　　　1.時間緩存

　　　　　　　　　　結構體event_base中的tv_cache，用來記錄時間緩存。若是tv_cache已經設置，那麼就直接使用緩存的時間，不然須要再次

執行系統調用獲取系統時間

　　　　　　　　2.時間校訂

　　　　　　　　　　若是系統支持monotonic時間，該時間是從boot後到如今所通過的時間，所以不須要執行校訂。若是系統不支持monotonic時間，用戶可能會手動調整時間，校訂由函數timeout_correct()完成。

 1 static void timeout_correct(struct event_base *base, struct timeval *tv)
 2 {
 3     struct event **pev;
 4     unsigned int size;
 5     struct timeval off;
 6     if (use_monotonic) // monotonic時間就直接返回，無需調整
 7         return;
 8     gettime(base, tv); // tv <---tv_cache
 9     // 根據前面的分析能夠知道event_tv應該小於tv_cache
10     // 若是tv < event_tv代表用戶向前調整時間了，須要校訂時間
11     if (evutil_timercmp(tv, &base->event_tv, >=)) {
12         base->event_tv = *tv;
13         return;
14     }
15     // 計算時間差值
16     evutil_timersub(&base->event_tv, tv, &off);
17     // 調整定時事件小根堆
18     pev = base->timeheap.p;
19     size = base->timeheap.n;
20     for (; size-- > 0; ++pev) {
21         struct timeval *ev_tv = &(**pev).ev_timeout;
22         evutil_timersub(ev_tv, &off, ev_tv);
23     }
24     base->event_tv = *tv; // 更新event_tv爲tv_cache
25 }

　　　　　　　　在調整小根堆時，由於全部定時事件的時間值都會被減去相同的值，所以雖然堆中元素的時間鍵值被改變了，可是相對關係沒有改變，不會改變堆的總體結構。所以只要遍歷堆中的全部元素，將每一個元素的時間鍵值減去相同的值便可完成調整，不須要從新調整堆的結構。調整完後，要將event_tv值從新設置爲tv_cache值。

　　　　5、libevent支持多線程

　　　　　　libevent不是線程安全的，可是這並不表示libevent不支持多線程模式，其實方法在前面已經將signal事件處理時就接觸到了，那就是消息通知機制。如下：

　　　　　　　　1)暴力搶佔

　　　　　　　　　　中止正在執行的任務，立刻去執行新來的任務。好處是消息能夠當即獲得處理，須要注意的是必須處理好線程切換問題。

　　　　　　　　2)純粹的消息通知機制

　　　　　　　　　　執行完正在執行的任務，再執行新任務。經過消息通知，切換問題省心了，不過消息是不能當即處理的，並且全部的內容都是經過消息發送，增長了通訊的開銷。

　　　　　　　　3)消息通知+同步層

　　　　　　　　　　有個折中的辦法能夠減小消息通訊的開銷，就是提取一個同步層，把工做安排都存放在一個工做隊列中，並且可以保證「任何人把新任務扔到這個隊列」和「本身取出當前第一個任務」等這些操做都可以保證不會把隊列搞亂

　　　　　　　　　　工做隊列實際上就是一個加鎖的容器（隊列，鏈表），這個很容易實現，而消息通知僅需一個字節，具體的任務都push到了工做隊列中，減小了開銷