libevent(瞭解)

1 前言

Libevent是一個輕量級的開源高性能網絡庫，使用者衆多，研究者更甚，相關文章也很多。寫這一系列文章的用意在於，一則分享心得；二則對libevent代碼和設計思想作系統的、更深層次的分析，寫出來，也可供後來者參考。

附帶一句：Libevent是用c語言編寫的（MS大牛們都偏心c語言哪），並且幾乎是無處不函數指針，學習其源代碼也須要至關的c語言基礎。

2 LIBEVENT簡介

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

Libevent已經被普遍的應用，做爲底層的網絡庫；好比memcached、Vomit、Nylon、Netchat等等。
Libevent當前的最新穩定版是1.4.13；這也是本文參照的版本。

3 學習的好處

學習libevent有助於提高程序設計功力，除了網絡程序設計方面外，Libevent的代碼裏有不少有用的設計技巧和基礎數據結構，好比信息隱藏、函數指針、c語言的多態支持、鏈表和堆等等，都有助於提高自身的程序功力。
程序設計不止要了解框架，不少細節之處偏偏也是事關整個系統成敗的關鍵。只對libevent自己的框架大概瞭解，那或許僅僅是隻知其一;不知其二，不深刻代碼分析，就難以瞭解其設計的精巧之處，也就難覺得本身所用。

事實上Libevent自己就是一個典型的Reactor模型，理解Reactor模式是理解libevent的基石；所以下一節將介紹典型的事件驅動設計模式——Reactor模式。

前面講到，整個libevent自己就是一個Reactor，所以本節將專門對Reactor模式進行必要的介紹，並列出libevnet中的幾個重要組件和Reactor的對應關係，在後面的章節中可能還會提到本節介紹的基本概念。

1 REACTOR的事件處理機制

首先來回想一下普通函數調用的機制：程序調用某函數?函數執行，程序等待?函數將結果和控制權返回給程序?程序繼續處理。
Reactor釋義「反應堆」，是一種事件驅動機制。和普通函數調用的不一樣之處在於：應用程序不是主動的調用某個API完成處理，而是偏偏相反，Reactor逆置了事件處理流程，應用程序須要提供相應的接口並註冊到Reactor上，若是相應的時間發生，Reactor將主動調用應用程序註冊的接口，這些接口又稱爲「回調函數」。使用Libevent也是想Libevent框架註冊相應的事件和回調函數；當這些時間發聲時，Libevent會調用這些回調函數處理相應的事件（I/O讀寫、定時和信號）。
用「好萊塢原則」來形容Reactor再合適不過了：不要打電話給咱們，咱們會打電話通知你。
舉個例子：你去應聘某xx公司，面試結束後。
「普通函數調用機制」公司HR比較懶，不會記你的聯繫方式，那怎麼辦呢，你只能面試完後本身打電話去問結果；有沒有被錄取啊，仍是被據了；

「Reactor」公司HR就記下了你的聯繫方式，結果出來後會主動打電話通知你：有沒有被錄取啊，仍是被據了；你不用本身打電話去問結果，事實上也不能，你沒有HR的留聯繫方式。

2 REACTOR模式的優勢

Reactor模式是編寫高性能網絡服務器的必備技術之一，它具備以下的優勢：
1）響應快，沒必要爲單個同步時間所阻塞，雖然Reactor自己依然是同步的；
2）編程相對簡單，能夠最大程度的避免複雜的多線程及同步問題，而且避免了多線程/進程的切換開銷；
3）可擴展性，能夠方便的經過增長Reactor實例個數來充分利用CPU資源；
4）可複用性，reactor框架自己與具體事件處理邏輯無關，具備很高的複用性；

3 REACTOR模式框架

使用Reactor模型，必備的幾個組件：事件源、Reactor框架、多路複用機制和事件處理程序，先來看看Reactor模型的總體框架，接下來再對每一個組件作逐一說明。

1） 事件源
Linux上是文件描述符，Windows上就是Socket或者Handle了，這裏統一稱爲「句柄集」；程序在指定的句柄上註冊關心的事件，好比I/O事件。

2） event demultiplexer——事件多路分發機制
由操做系統提供的I/O多路複用機制，好比select和epoll。
程序首先將其關心的句柄（事件源）及其事件註冊到event demultiplexer上；
當有事件到達時，event demultiplexer會發出通知「在已經註冊的句柄集中，一個或多個句柄的事件已經就緒」；
程序收到通知後，就能夠在非阻塞的狀況下對事件進行處理了。
對應到libevent中，依然是select、poll、epoll等，可是libevent使用結構體eventop進行了封裝，以統一的接口來支持這些I/O多路複用機制，達到了對外隱藏底層系統機制的目的。

3） Reactor——反應器
Reactor，是事件管理的接口，內部使用event demultiplexer註冊、註銷事件；並運行事件循環，當有事件進入「就緒」狀態時，調用註冊事件的回調函數處理事件。
對應到libevent中，就是event_base結構體。
一個典型的Reactor聲明方式

view plain

1. class Reactor
2. {
3. public:
4.     int register_handler(Event_Handler *pHandler, int event);
5.     int remove_handler(Event_Handler *pHandler, int event);
6.     void handle_events(timeval *ptv);
7.     // …
8. };

4） Event Handler——事件處理程序
事件處理程序提供了一組接口，每一個接口對應了一種類型的事件，供Reactor在相應的事件發生時調用，執行相應的事件處理。一般它會綁定一個有效的句柄。
對應到libevent中，就是event結構體。
下面是兩種典型的Event Handler類聲明方式，兩者互有優缺點。

view plain

1. class Event_Handler
2. {
3. public:
4.     virtual void handle_read() = 0;
5.     virtual void handle_write() = 0;
6.     virtual void handle_timeout() = 0;
7.     virtual void handle_close() = 0;
8.     virtual HANDLE get_handle() = 0;
9.     // …
10. };
11. class Event_Handler
12. {
13. public:
14.     // events maybe read/write/timeout/close .etc
15.     virtual void handle_events(int events) = 0;
16.     virtual HANDLE get_handle() = 0;
17.     // …
18. };

 

4 REACTOR事件處理流程

前面說過Reactor將事件流「逆置」了，那麼使用Reactor模式後，事件控制流是什麼樣子呢？
能夠參見下面的序列圖。

5 小結

上面講到了Reactor的基本概念、框架和處理流程，對Reactor有個基本清晰的瞭解後，再來對比看libevent就會更容易理解了，接下來就正式進入到libevent的代碼世界了，加油！

1 前言

學習源代碼該從哪裏入手？我以爲從程序的基本使用場景和代碼的總體處理流程入手是個不錯的方法，至少從我的的經驗上講，用此方法分析libevent是比較有效的。

2 基本應用場景

基本應用場景也是使用libevnet的基本流程，下面來考慮一個最簡單的場景，使用livevent設置定時器，應用程序只須要執行下面幾個簡單的步驟便可。
1）首先初始化libevent庫，並保存返回的指針
struct event_base * base = event_init();
實際上這一步至關於初始化一個Reactor實例；在初始化libevent後，就能夠註冊事件了。

2）初始化事件event，設置回調函數和關注的事件
evtimer_set(&ev, timer_cb, NULL);
事實上這等價於調用event_set(&ev, -1, 0, timer_cb, NULL);
event_set的函數原型是：
void event_set(struct event *ev, int fd, short event, void (*cb)(int, short, void *), void *arg)
ev：執行要初始化的event對象；
fd：該event綁定的「句柄」，對於信號事件，它就是關注的信號；
event：在該fd上關注的事件類型，它能夠是EV_READ, EV_WRITE, EV_SIGNAL；
cb：這是一個函數指針，當fd上的事件event發生時，調用該函數執行處理，它有三個參數，調用時由event_base負責傳入，按順序，實際上就是event_set時的fd, event和arg；
arg：傳遞給cb函數指針的參數；
因爲定時事件不須要fd，而且定時事件是根據添加時（event_add）的超時值設定的，所以這裏event也不須要設置。
這一步至關於初始化一個event handler，在libevent中事件類型保存在event結構體中。
注意：libevent並不會管理event事件集合，這須要應用程序自行管理；

3）設置event從屬的event_base
event_base_set(base, &ev);
這一步至關於指明event要註冊到哪一個event_base實例上；

4）是正式的添加事件的時候了
event_add(&ev, timeout);
基本信息都已設置完成，只要簡單的調用event_add()函數便可完成，其中timeout是定時值；
這一步至關於調用Reactor::register_handler()函數註冊事件。

5）程序進入無限循環，等待就緒事件並執行事件處理
event_base_dispatch(base);

3 實例代碼

上面例子的程序代碼以下所示

view plain

1. struct event ev;
2. struct timeval tv;
3. void time_cb(int fd, short event, void *argc)
4. {
5.     printf(「timer wakeup/n」);
6.     event_add(&ev, &tv); // reschedule timer
7. }
8. int main()
9. {
10.     struct event_base *base = event_init();
11.     tv.tv_sec = 10; // 10s period
12.     tv.tv_usec = 0;
13.     evtimer_set(&ev, time_cb, NULL);
14.     event_add(&ev, &tv);
15.     event_base_dispatch(base);
16. }

 

4 事件處理流程

當應用程序向libevent註冊一個事件後，libevent內部是怎麼樣進行處理的呢？下面的圖就給出了這一基本流程。
1）    首先應用程序準備並初始化event，設置好事件類型和回調函數；這對應於前面第步驟2和3；
2）    向libevent添加該事件event。對於定時事件，libevent使用一個小根堆管理，key爲超時時間；對於Signal和I/O事件，libevent將其放入到等待鏈表（wait list）中，這是一個雙向鏈表結構；
3）    程序調用event_base_dispatch()系列函數進入無限循環，等待事件，以select()函數爲例；每次循環前libevent會檢查定時事件的最小超時時間tv，根據tv設置select()的最大等待時間，以便於後面及時處理超時事件；
當select()返回後，首先檢查超時事件，而後檢查I/O事件；
Libevent將全部的就緒事件，放入到激活鏈表中；
而後對激活鏈表中的事件，調用事件的回調函數執行事件處理；

5 小結

本節介紹了libevent的簡單實用場景，並旋風般的介紹了libevent的事件處理流程，讀者應該對libevent有了基本的印象，下面將會詳細介紹libevent的事件管理框架（Reactor模式中的Reactor框架）作詳細的介紹，在此以前會對源代碼文件作簡單的分類。

——libevent源代碼文件組織

1 前言

詳細分析源代碼以前，若是能對其代碼文件的基本結構有個大概的認識和分類，對於代碼的分析將是大有裨益的。本節內容很少，我想並非說它不重要！

2 源代碼組織結構

Libevent的源代碼雖然都在一層文件夾下面，可是其代碼分類仍是至關清晰的，主要可分爲頭文件、內部使用的頭文件、輔助功能函數、日誌、libevent框架、對系統I/O多路複用機制的封裝、信號管理、定時事件管理、緩衝區管理、基本數據結構和基於libevent的兩個實用庫等幾個部分，有些部分可能就是一個源文件。
源代碼中的test部分就不在咱們關注的範疇了。
1）頭文件
主要就是event.h：事件宏定義、接口函數聲明，主要結構體event的聲明；
2）內部頭文件
xxx-internal.h：內部數據結構和函數，對外不可見，以達到信息隱藏的目的；
3）libevent框架
event.c：event總體框架的代碼實現；
4）對系統I/O多路複用機制的封裝
epoll.c：對epoll的封裝；
select.c：對select的封裝；
devpoll.c：對dev/poll的封裝;
kqueue.c：對kqueue的封裝；
5）定時事件管理
min-heap.h：其實就是一個以時間做爲key的小根堆結構；
6）信號管理
signal.c：對信號事件的處理；
7）輔助功能函數
evutil.h 和evutil.c：一些輔助功能函數，包括建立socket pair和一些時間操做函數：加、減和比較等。
8）日誌
log.h和log.c：log日誌函數
9）緩衝區管理
evbuffer.c和buffer.c：libevent對緩衝區的封裝；
10）基本數據結構
compat/sys下的兩個源文件：queue.h是libevent基本數據結構的實現，包括鏈表，雙向鏈表，隊列等；_libevent_time.h：一些用於時間操做的結構體定義、函數和宏定義；
11）實用網絡庫
http和evdns：是基於libevent實現的http服務器和異步dns查詢庫；

3 小結

本節介紹了libevent的組織和分類，下面將會詳細介紹libevent的核心部分event結構。

對事件處理流程有了高層的認識後，本節將詳細介紹libevent的核心結構event，以及libevent對event的管理。

1 LIBEVENT的核心-EVENT

Libevent是基於事件驅動（event-driven）的，從名字也能夠看到event是整個庫的核心。event就是Reactor框架中的事件處理程序組件；它提供了函數接口，供Reactor在事件發生時調用，以執行相應的事件處理，一般它會綁定一個有效的句柄。
首先給出event結構體的聲明，它位於event.h文件中：

view plain

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種類型：
I/O事件： EV_WRITE和EV_READ
定時事件：EV_TIMEOUT
信號：    EV_SIGNAL
輔助選項：EV_PERSIST，代表是一個永久事件
Libevent中的定義爲：

view plain

1. #define EV_TIMEOUT 0×01
2. #define EV_READ  0×02
3. #define EV_WRITE 0×04
4. #define EV_SIGNAL 0×08
5. #define EV_PERSIST 0×10 /* Persistant event */

能夠看出事件類型可使用「|」運算符進行組合，須要說明的是，信號和I/O事件不能同時設置；
還能夠看出libevent使用event結構體將這3種事件的處理統一塊兒來；
2）ev_next，ev_active_next和ev_signal_next都是雙向鏈表節點指針；它們是libevent對不一樣事件類型和在不一樣的時期，對事件的管理時使用到的字段。
libevent使用雙向鏈表保存全部註冊的I/O和Signal事件，ev_next就是該I/O事件在鏈表中的位置；稱此鏈表爲「已註冊事件鏈表」；
一樣ev_signal_next就是signal事件在signal事件鏈表中的位置；
ev_active_next：libevent將全部的激活事件放入到鏈表active list中，而後遍歷active list執行調度，ev_active_next就指明瞭event在active list中的位置；
2）min_heap_idx和ev_timeout，若是是timeout事件，它們是event在小根堆中的索引和超時值，libevent使用小根堆來管理定時事件，這將在後面定時事件處理時專門講解
3）ev_base該事件所屬的反應堆實例，這是一個event_base結構體，下一節將會詳細講解；
4）ev_fd，對於I/O事件，是綁定的文件描述符；對於signal事件，是綁定的信號；
5）ev_callback，event的回調函數，被ev_base調用，執行事件處理程序，這是一個函數指針，原型爲：
void (*ev_callback)(int fd, short events, void *arg)
其中參數fd對應於ev_fd；events對應於ev_events；arg對應於ev_arg；
6）ev_arg：void*，代表能夠是任意類型的數據，在設置event時指定；
7）eb_flags：libevent用於標記event信息的字段，代表其當前的狀態，可能的值有：

view plain

1. #define EVLIST_TIMEOUT 0×01 // event在time堆中
2. #define EVLIST_INSERTED 0×02 // event在已註冊事件鏈表中
3. #define EVLIST_SIGNAL 0×04 // 未見使用
4. #define EVLIST_ACTIVE 0×08 // event在激活鏈表中
5. #define EVLIST_INTERNAL 0×10 // 內部使用標記
6. #define EVLIST_INIT     0×80 // event已被初始化

8）ev_ncalls：事件就緒執行時，調用ev_callback的次數，一般爲1；
9）ev_pncalls：指針，一般指向ev_ncalls或者爲NULL；
10）ev_res：記錄了當前激活事件的類型；

 

 

2 LIBEVENT對EVENT的管理

從event結構體中的3個鏈表節點指針和一個堆索引出發，大致上也能窺出libevent對event的管理方法了，能夠參見下面的示意圖：

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

3 事件設置的接口函數

要向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)
1.設置事件ev綁定的文件描述符或者信號，對於定時事件，設爲-1便可；
2.設置事件類型，好比EV_READ|EV_PERSIST, EV_WRITE, EV_SIGNAL等；
3.設置事件的回調函數以及參數arg；
4.初始化其它字段，好比缺省的event_base和優先級；
int event_base_set(struct event_base *base, struct event *ev)
設置event ev將要註冊到的event_base；
libevent有一個全局event_base指針current_base，默認狀況下事件ev將被註冊到current_base上，使用該函數能夠指定不一樣的event_base；
若是一個進程中存在多個libevent實例，則必需要調用該函數爲event設置不一樣的event_base；

int event_priority_set(struct event *ev, int pri)
設置event ev的優先級，沒什麼可說的，注意的一點就是：當ev正處於就緒狀態時，不能設置，返回-1。

4 小結

本節講述了libevent的核心event結構，以及libevent支持的事件類型和libevent對event的管理模型；接下來將會描述libevent的事件處理框架，以及其中使用的重要的結構體event_base；

前面已經對libevent的事件處理框架和event結構體作了描述，如今是時候剖析libevent對事件的詳細處理流程了，本節將分析libevent的事件處理框架event_base和libevent註冊、刪除事件的具體流程，可結合前一節libevent對event的管理。

1 事件處理框架-EVENT_BASE

回想Reactor模式的幾個基本組件，本節講解的部分對應於Reactor框架組件。在libevent中，這就表現爲event_base結構體，結構體聲明以下，它位於event-internal.h文件中：

view plain

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的一個對象實例。
evsel指向了全局變量static const struct eventop *eventops[]中的一個；
前面也說過，libevent將系統提供的I/O demultiplex機制統一封裝成了eventop結構；所以eventops[]包含了select、poll、kequeue和epoll等等其中的若干個全局實例對象。
evbase其實是一個eventop實例對象；
先來看看eventop結構體，它的成員是一系列的函數指針, 在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中，每種I/O demultiplex機制的實現都必須提供這五個函數接口，來完成自身的初始化、銷燬釋放；對事件的註冊、註銷和分發。
好比對於epoll，libevent實現了5個對應的接口函數，並在初始化時並將eventop的5個函數指針指向這5個函數，那麼程序就可使用epoll做爲I/O demultiplex機制了，這個在後面會再次提到。
2）activequeues是一個二級指針，前面講過libevent支持事件優先級，所以你能夠把它看做是數組，其中的元素activequeues[priority]是一個鏈表，鏈表的每一個節點指向一個優先級爲priority的就緒事件event。
3）eventqueue，鏈表，保存了全部的註冊事件event的指針。
4）sig是由來管理信號的結構體，將在後面信號處理時專門講解；
5）timeheap是管理定時事件的小根堆，將在後面定時事件處理時專門講解；
6）event_tv和tv_cache是libevent用於時間管理的變量，將在後面講到；
其它各個變量都能因名知意，就再也不囉嗦了。

2 建立和初始化EVENT_BASE

建立一個event_base對象也既是建立了一個新的libevent實例，程序須要經過調用event_init()（內部調用event_base_new函數執行具體操做）函數來建立，該函數同時還對新生成的libevent實例進行了初始化。
該函數首先爲event_base實例申請空間，而後初始化timer mini-heap，選擇並初始化合適的系統I/O 的demultiplexer機制，初始化各事件鏈表；
函數還檢測了系統的時間設置，爲後面的時間管理打下基礎。

3 接口函數

前面提到Reactor框架的做用就是提供事件的註冊、註銷接口；根據系統提供的事件多路分發機制執行事件循環，當有事件進入「就緒」狀態時，調用註冊事件的回調函數來處理事件。
Libevent中對應的接口函數主要就是：

view plain

1. int  event_add(struct event *ev, const struct timeval *timeout);
2. int  event_del(struct event *ev);
3. int  event_base_loop(struct event_base *base, int loops);
4. void event_active(struct event *event, int res, short events);
5. void event_process_active(struct event_base *base);

 

本節將按介紹事件註冊和刪除的代碼流程，libevent的事件循環框架將在下一節再具體描述。
對於定時事件，這些函數將調用timer heap管理接口執行插入和刪除操做；對於I/O和Signal事件將調用eventopadd和delete接口函數執行插入和刪除操做（eventop會對Signal事件調用Signal處理接口執行操做）；這些組件將在後面的內容描述。
1）註冊事件
函數原型：
int event_add(struct event *ev, const struct timeval *tv)
參數：ev：指向要註冊的事件；
tv：超時時間；
函數將ev註冊到ev->ev_base上，事件類型由ev->ev_events指明，若是註冊成功，ev將被插入到已註冊鏈表中；若是tv不是NULL，則會同時註冊定時事件，將ev添加到timer堆上；
若是其中有一步操做失敗，那麼函數保證沒有事件會被註冊，能夠講這至關於一個原子操做。這個函數也體現了libevent細節之處的巧妙設計，且仔細看程序代碼，部分有省略，註釋直接附在代碼中。

view plain

1. int event_add(struct event *ev, const struct timeval *tv)
2. {
3.  struct event_base *base = ev->ev_base; // 要註冊到的event_base
4.  const struct eventop *evsel = base->evsel;
5.  void *evbase = base->evbase; // base使用的系統I/O策略
6.  // 新的timer事件，調用timer heap接口在堆上預留一個位置
7.  // 注：這樣能保證該操做的原子性：
8.  // 向系統I/O機制註冊可能會失敗，而當在堆上預留成功後，
9.  // 定時事件的添加將確定不會失敗；
10.  // 而預留位置的可能結果是堆擴充，可是內部元素並不會改變
11.  if (tv != NULL && !(ev->ev_flags & EVLIST_TIMEOUT)) {
12.   if (min_heap_reserve(&base->timeheap,
13.    1 + min_heap_size(&base->timeheap)) == -1)
14.    return (-1);  /* ENOMEM == errno */
15.  }
16.  // 若是事件ev不在已註冊或者激活鏈表中，則調用evbase註冊事件
17.  if ((ev->ev_events & (EV_READ|EV_WRITE|EV_SIGNAL)) &&
18.   !(ev->ev_flags & (EVLIST_INSERTED|EVLIST_ACTIVE))) {
19.    res = evsel->add(evbase, ev);
20.    if (res != -1) // 註冊成功，插入event到已註冊鏈表中
21.     event_queue_insert(base, ev, EVLIST_INSERTED);
22.  }
23.  // 準備添加定時事件
24.  if (res != -1 && tv != NULL) {
25.   struct timeval now;
26.   // EVLIST_TIMEOUT代表event已經在定時器堆中了，刪除舊的
27.   if (ev->ev_flags & EVLIST_TIMEOUT)
28.    event_queue_remove(base, ev, EVLIST_TIMEOUT);
29.   // 若是事件已是就緒狀態則從激活鏈表中刪除
30.   if ((ev->ev_flags & EVLIST_ACTIVE) &&
31.    (ev->ev_res & EV_TIMEOUT)) {
32.     // 將ev_callback調用次數設置爲0
33.     if (ev->ev_ncalls && ev->ev_pncalls) {
34.      *ev->ev_pncalls = 0;
35.     }
36.     event_queue_remove(base, ev, EVLIST_ACTIVE);
37.   }
38.   // 計算時間，並插入到timer小根堆中
39.   gettime(base, &now);
40.   evutil_timeradd(&now, tv, &ev->ev_timeout);
41.   event_queue_insert(base, ev, EVLIST_TIMEOUT);
42.  }
43.  return (res);
44. }
45. event_queue_insert()負責將事件插入到對應的鏈表中，下面是程序代碼；
46. event_queue_remove()負責將事件從對應的鏈表中刪除，這裏就再也不重複貼代碼了；
47. void event_queue_insert(struct event_base *base, struct event *ev, int queue)
48. {
49.  // ev可能已經在激活列表中了，避免重複插入
50.  if (ev->ev_flags & queue) {
51.   if (queue & EVLIST_ACTIVE)
52.    return;
53.  }
54.  // …
55.  ev->ev_flags |= queue; // 記錄queue標記
56.  switch (queue) {
57.  case EVLIST_INSERTED: // I/O或Signal事件，加入已註冊事件鏈表
58.   TAILQ_INSERT_TAIL(&base->eventqueue, ev, ev_next);
59.   break;
60.  case EVLIST_ACTIVE: // 就緒事件，加入激活鏈表
61.   base->event_count_active++;
62.   TAILQ_INSERT_TAIL(base->activequeues[ev->ev_pri], ev, ev_active_next);
63.   break;
64.  case EVLIST_TIMEOUT: // 定時事件，加入堆
65.   min_heap_push(&base->timeheap, ev);
66.   break;
67.  }
68. }

 

2）刪除事件：
函數原型爲：int  event_del(struct event *ev);
該函數將刪除事件ev，對於I/O事件，從I/O 的demultiplexer上將事件註銷；對於Signal事件，將從Signal事件鏈表中刪除；對於定時事件，將從堆上刪除；
一樣刪除事件的操做則不必定是原子的，好比刪除時間事件以後，有可能從系統I/O機制中註銷會失敗。

 

view plain

1. int event_del(struct event *ev)
2. {
3.  struct event_base *base;
4.  const struct eventop *evsel;
5.  void *evbase;
6.  // ev_base爲NULL，代表ev沒有被註冊
7.  if (ev->ev_base == NULL)
8.   return (-1);
9.  // 取得ev註冊的event_base和eventop指針
10.  base = ev->ev_base;
11.  evsel = base->evsel;
12.  evbase = base->evbase;
13.  // 將ev_callback調用次數設置爲
14.  if (ev->ev_ncalls && ev->ev_pncalls) {
15.   *ev->ev_pncalls = 0;
16.  }
17.  // 從對應的鏈表中刪除
18.  if (ev->ev_flags & EVLIST_TIMEOUT)
19.   event_queue_remove(base, ev, EVLIST_TIMEOUT);
20.  if (ev->ev_flags & EVLIST_ACTIVE)
21.   event_queue_remove(base, ev, EVLIST_ACTIVE);
22.  if (ev->ev_flags & EVLIST_INSERTED) {
23.   event_queue_remove(base, ev, EVLIST_INSERTED);
24.   // EVLIST_INSERTED代表是I/O或者Signal事件，
25.   // 須要調用I/O demultiplexer註銷事件
26.   return (evsel->del(evbase, ev));
27.  }
28.  return (0);
29. }

4 小節
分析了event_base這一重要結構體，初步看到了libevent對系統的I/O demultiplex機制的封裝event_op結構，並結合源代碼分析了事件的註冊和刪除處理，下面將會接着分析事件管理框架中的主事件循環部分。

如今咱們已經初步瞭解了libevent的Reactor組件——event_base和事件管理框架，接下來就是libevent事件處理的中心部分——事件主循環，根據系統提供的事件多路分發機制執行事件循環，對已註冊的就緒事件，調用註冊事件的回調函數來處理事件。

1 階段性的勝利

Libevent將I/O事件、定時器和信號事件處理很好的結合到了一塊兒，本節也會介紹libevent是如何作到這一點的。
在看完本節的內容後，讀者應該會對Libevent的基本框架：事件管理和主循環有比較清晰的認識了，並可以把libevent的事件控制流程清晰的串通起來，剩下的就是一些細節的內容了。

2 事件處理主循環

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

清楚了event_base_loop所做的主要操做，就能夠對比源代碼看個究竟了，代碼結構仍是至關清晰的。

view plain

1. int event_base_loop(struct event_base *base, int flags)
2. {
3.     const struct eventop *evsel = base->evsel;
4.     void *evbase = base->evbase;
5.     struct timeval tv;
6.     struct timeval *tv_p;
7.     int res, done;
8.     // 清空時間緩存
9.     base->tv_cache.tv_sec = 0;
10.     // evsignal_base是全局變量，在處理signal時，用於指名signal所屬的event_base實例
11.     if (base->sig.ev_signal_added)
12.         evsignal_base = base;
13.     done = 0;
14.     while (!done) { // 事件主循環
15.         // 查看是否須要跳出循環，程序能夠調用event_loopexit_cb()設置event_gotterm標記
16.         // 調用event_base_loopbreak()設置event_break標記
17.         if (base->event_gotterm) {
18.             base->event_gotterm = 0;
19.             break;
20.         }
21.         if (base->event_break) {
22.             base->event_break = 0;
23.             break;
24.         }
25.         // 校訂系統時間，若是系統使用的是非MONOTONIC時間，用戶可能會向後調整了系統時間
26.         // 在timeout_correct函數裏，比較last wait time和當前時間，若是當前時間< last wait time
27.         // 代表時間有問題，這是須要更新timer_heap中全部定時事件的超時時間。
28.         timeout_correct(base, &tv);
29.         // 根據timer heap中事件的最小超時時間，計算系統I/O demultiplexer的最大等待時間
30.         tv_p = &tv;
31.         if (!base->event_count_active && !(flags & EVLOOP_NONBLOCK)) {
32.             timeout_next(base, &tv_p);
33.         } else {
34.             // 依然有未處理的就緒時間，就讓I/O demultiplexer當即返回，沒必要等待
35.             // 下面會提到，在libevent中，低優先級的就緒事件可能不能當即被處理
36.             evutil_timerclear(&tv);
37.         }
38.         // 若是當前沒有註冊事件，就退出
39.         if (!event_haveevents(base)) {
40.             event_debug((「%s: no events registered.」, __func__));
41.             return (1);
42.         }
43.         // 更新last wait time，並清空time cache
44.         gettime(base, &base->event_tv);
45.         base->tv_cache.tv_sec = 0;
46.         // 調用系統I/O demultiplexer等待就緒I/O events，多是epoll_wait，或者select等；
47.         // 在evsel->dispatch()中，會把就緒signal event、I/O event插入到激活鏈表中
48.         res = evsel->dispatch(base, evbase, tv_p);
49.         if (res == -1)
50.             return (-1);
51.         // 將time cache賦值爲當前系統時間
52.         gettime(base, &base->tv_cache);
53.         // 檢查heap中的timer events，將就緒的timer event從heap上刪除，並插入到激活鏈表中
54.         timeout_process(base);
55.         // 調用event_process_active()處理激活鏈表中的就緒event，調用其回調函數執行事件處理
56.         // 該函數會尋找最高優先級（priority值越小優先級越高）的激活事件鏈表，
57.         // 而後處理鏈表中的全部就緒事件；
58.         // 所以低優先級的就緒事件可能得不到及時處理；
59.         if (base->event_count_active) {
60.             event_process_active(base);
61.             if (!base->event_count_active && (flags & EVLOOP_ONCE))
62.                 done = 1;
63.         } else if (flags & EVLOOP_NONBLOCK)
64.             done = 1;
65.     }
66.     // 循環結束，清空時間緩存
67.     base->tv_cache.tv_sec = 0;
68.     event_debug((「%s: asked to terminate loop.」, __func__));
69.     return (0);
70. }

 

3 I/O和TIMER事件的統一

Libevent將Timer和Signal事件都統一到了系統的I/O 的demultiplex機制中了，相信讀者從上面的流程和代碼中也能窺出一斑了，下面就再囉嗦一次了。
首先將Timer事件融合到系統I/O多路複用機制中，仍是至關清晰的，由於系統的I/O機制像select()和epoll_wait()都容許程序制定一個最大等待時間（也稱爲最大超時時間）timeout，即便沒有I/O事件發生，它們也保證能在timeout時間內返回。
那麼根據全部Timer事件的最小超時時間來設置系統I/O的timeout時間；當系統I/O返回時，再激活全部就緒的Timer事件就能夠了，這樣就能將Timer事件完美的融合到系統的I/O機制中了。
這是在Reactor和Proactor模式（主動器模式，好比Windows上的IOCP）中處理Timer事件的經典方法了，ACE採用的也是這種方法，你們能夠參考POSA vol2書中的Reactor模式一節。
堆是一種經典的數據結構，向堆中插入、刪除元素時間複雜度都是O(lgN)，N爲堆中元素的個數，而獲取最小key值（小根堆）的複雜度爲O(1)；所以變成了管理Timer事件的絕佳人選（固然是非惟一的），libevent就是採用的堆結構。

4 I/O和SIGNAL事件的統一

Signal是異步事件的經典事例，將Signal事件統一到系統的I/O多路複用中就不像Timer事件那麼天然了，Signal事件的出現對於進程來說是徹底隨機的，進程不能只是測試一個變量來判別是否發生了一個信號，而是必須告訴內核「在此信號發生時，請執行以下的操做」。
若是當Signal發生時，並不當即調用event的callback函數處理信號，而是設法通知系統的I/O機制，讓其返回，而後再統一和I/O事件以及Timer一塊兒處理，不就能夠了嘛。是的，這也是libevent中使用的方法。
問題的核心在於，當Signal發生時，如何通知系統的I/O多路複用機制，這裏先買個小關子，放到信號處理一節再詳細說明，我想讀者確定也能想出通知的方法，好比使用pipe。

5 小節

介紹了libevent的事件主循環，描述了libevent是如何處理就緒的I/O事件、定時器和信號事件，以及如何將它們無縫的融合到一塊兒。
加油！

 

如今咱們已經瞭解了libevent的基本框架：事件管理框架和事件主循環。上節提到了libevent中I/O事件和Signal以及Timer事件的集成，這一節將分析如何將Signal集成到事件主循環的框架中。

1 集成策略——使用SOCKET PAIR

前一節已經作了足夠多的介紹了，基本方法就是採用「消息機制」。在libevent中這是經過socket pair完成的，下面就來詳細分析一下。
Socket pair就是一個socket對，包含兩個socket，一個讀socket，一個寫socket。工做方式以下圖所示：

建立一個socket pair並非複雜的操做，能夠參見下面的流程圖，清晰起見，其中忽略了一些錯誤處理和檢查。

Libevent提供了輔助函數evutil_socketpair()來建立一個socket pair，能夠結合上面的建立流程來分析該函數。

2 集成到事件主循環——通知EVENT_BASE

Socket pair建立好了，但是libevent的事件主循環仍是不知道Signal是否發生了啊，看來咱們還差了最後一步，那就是：爲socket pair的讀socket在libevent的event_base實例上註冊一個persist的讀事件。
這樣當向寫socket寫入數據時，讀socket就會獲得通知，觸發讀事件，從而event_base就能相應的獲得通知了。
前面提到過，Libevent會在事件主循環中檢查標記，來肯定是否有觸發的signal，若是標記被設置就處理這些signal，這段代碼在各個具體的I/O機制中，以Epoll爲例，在epoll_dispatch()函數中，代碼片斷以下：

view plain

1. res = epoll_wait(epollop->epfd, events, epollop->nevents, timeout);
2. if (res == -1) {
3.     if (errno != EINTR) {
4.         event_warn(「epoll_wait」);
5.         return (-1);
6.     }
7.     evsignal_process(base);// 處理signal事件
8.     return (0);
9. } else if (base->sig.evsignal_caught) {
10.     evsignal_process(base);// 處理signal事件
11. }

完整的處理框架以下所示：

注1：libevent中，初始化階段並不註冊讀socket的讀事件，而是在註冊信號階段纔會測試並註冊；
注2：libevent中，檢查I/O事件是在各系統I/O機制的dispatch()函數中完成的，該dispatch()函數在event_base_loop()函數中被調用；

 

3 EVSIGNAL_INFO結構體

Libevent中Signal事件的管理是經過結構體evsignal_info完成的，結構體位於evsignal.h文件中，定義以下：

view plain

1. struct evsignal_info {
2.     struct event ev_signal;
3.     int ev_signal_pair[2];
4.     int ev_signal_added;
5.     volatile sig_atomic_t evsignal_caught;
6.     struct event_list evsigevents[NSIG];
7.     sig_atomic_t evsigcaught[NSIG];
8. #ifdef HAVE_SIGACTION
9.     struct sigaction **sh_old;
10. #else
11.     ev_sighandler_t **sh_old;
12. #endif
13.     int sh_old_max;
14. };

下面詳細介紹一下個字段的含義和做用：
1）ev_signal， 爲socket pair的讀socket向event_base註冊讀事件時使用的event結構體；
2）ev_signal_pair，socket pair對，做用見第一節的介紹；
3）ev_signal_added，記錄ev_signal事件是否已經註冊了；
4）evsignal_caught，是否有信號發生的標記；是volatile類型，由於它會在另外的線程中被修改；
5）evsigvents[NSIG]，數組，evsigevents[signo]表示註冊到信號signo的事件鏈表；
6）evsigcaught[NSIG]，具體記錄每一個信號觸發的次數，evsigcaught[signo]是記錄信號signo被觸發的次數；
7）sh_old記錄了原來的signal處理函數指針，當信號signo註冊的event被清空時，須要從新設置其處理函數；
evsignal_info的初始化包括，建立socket pair，設置ev_signal事件（但並無註冊，而是等到有信號註冊時才檢查並註冊），並將全部標記置零，初始化信號的註冊事件鏈表指針等。

 

4 註冊、註銷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鏈表中；
從signo上註銷一個已註冊的signal事件就更簡單了，直接從其已註冊事件的鏈表中移除便可。若是事件鏈表已空，那麼就恢復舊的處理函數；
下面的講解都以signal()函數爲例，sigaction()函數的處理和signal()類似。
處理函數evsignal_handler()函數作的事情很簡單，就是記錄信號的發生次數，並通知event_base有信號觸發，須要處理：

view plain

1. static void evsignal_handler(int sig)
2. {
3.     int save_errno = errno; // 不覆蓋原來的錯誤代碼
4.     if (evsignal_base == NULL) {
5.         event_warn(「%s: received signal %d, but have no base configured」, __func__, sig);
6.         return;
7.     }
8.     // 記錄信號sig的觸發次數，並設置event觸發標記
9.     evsignal_base->sig.evsigcaught[sig]++;
10.     evsignal_base->sig.evsignal_caught = 1;
11. #ifndef HAVE_SIGACTION
12.     signal(sig, evsignal_handler); // 從新註冊信號
13. #endif
14.     // 向寫socket寫一個字節數據，觸發event_base的I/O事件，從而通知其有信號觸發，須要處理
15.     send(evsignal_base->sig.ev_signal_pair[0], 」a」, 1, 0);
16.     errno = save_errno; // 錯誤代碼
17. }

5 小節
本節介紹了libevent對signal事件的具體處理框架，包括事件註冊、刪除和socket pair通知機制，以及是如何將Signal事件集成到事件主循環之中的。

如今再來詳細分析libevent中I/O事件和Timer事件的集成，與Signal相比，Timer事件的集成會直觀和簡單不少。Libevent對堆的調整操做作了一些優化，本節還會描述這些優化方法。

1 集成到事件主循環

由於系統的I/O機制像select()和epoll_wait()都容許程序制定一個最大等待時間（也稱爲最大超時時間）timeout，即便沒有I/O事件發生，它們也保證能在timeout時間內返回。
那麼根據全部Timer事件的最小超時時間來設置系統I/O的timeout時間；當系統I/O返回時，再激活全部就緒的Timer事件就能夠了，這樣就能將Timer事件完美的融合到系統的I/O機制中了。
具體的代碼在源文件event.c的event_base_loop()中，如今就對比代碼來看看這一處理方法：

view plain

1.       if (!base->event_count_active && !(flags & EVLOOP_NONBLOCK)) {
2.           // 根據Timer事件計算evsel->dispatch的最大等待時間
3.     timeout_next(base, &tv_p);
4. } else {
5.           // 若是還有活動事件，就不要等待，讓evsel->dispatch當即返回
6.     evutil_timerclear(&tv);
7. }
8. // …
9.       // 調用select() or epoll_wait() 等待就緒I/O事件
10. res = evsel->dispatch(base, evbase, tv_p);
11.       // …
12.       // 處理超時事件，將超時事件插入到激活鏈表中
13.       timeout_process(base);

timeout_next()函數根據堆中具備最小超時值的事件和當前時間來計算等待時間，下面看看代碼：

view plain

1. static int timeout_next(struct event_base *base, struct timeval **tv_p)
2. {
3.     struct timeval now;
4.     struct event *ev;
5.     struct timeval *tv = *tv_p;
6.     // 堆的首元素具備最小的超時值
7.     if ((ev = min_heap_top(&base->timeheap)) == NULL) {
8.         // 若是沒有定時事件，將等待時間設置爲NULL,表示一直阻塞直到有I/O事件發生
9.         *tv_p = NULL;
10.         return (0);
11.     }
12.     // 取得當前時間
13.     gettime(base, &now);
14.     // 若是超時時間<=當前值，不能等待，須要當即返回
15.     if (evutil_timercmp(&ev->ev_timeout, &now, <=)) {
16.         evutil_timerclear(tv);
17.         return (0);
18.     }
19.     // 計算等待的時間=當前時間-最小的超時時間
20.     evutil_timersub(&ev->ev_timeout, &now, tv);
21.     return (0);
22. }

 

2 TIMER小根堆

Libevent使用堆來管理Timer事件，其key值就是事件的超時時間，源代碼位於文件min_heap.h中。
全部的數據結構書中都有關於堆的詳細介紹，向堆中插入、刪除元素時間複雜度都是O(lgN)，N爲堆中元素的個數，而獲取最小key值（小根堆）的複雜度爲O(1)。堆是一個徹底二叉樹，基本存儲方式是一個數組。
Libevent實現的堆仍是比較輕巧的，雖然我不喜歡這種編碼方式（搞一些複雜的表達式）。輕巧到什麼地方呢，就以插入元素爲例，來對比說明，下面僞代碼中的size表示當前堆的元素個數：
典型的代碼邏輯以下：

view plain

1. Heap[size++] = new; // 先放到數組末尾，元素個數+1
2. // 下面就是shift_up()的代碼邏輯，不斷的將new向上調整
3. _child = size;
4. while(_child>0) // 循環
5. {
6.    _parent = (_child-1)/2; // 計算parent
7.    if(Heap[_parent].key < Heap[_child].key)
8.       break; // 調整結束，跳出循環
9.    swap(_parent, _child); // 交換parent和child
10. }

而libevent的heap代碼對這一過程作了優化，在插入新元素時，只是爲新元素預留了一個位置hole（初始時hole位於數組尾部），但並不馬上將新元素插入到hole上，而是不斷向上調整hole的值，將父節點向下調整，最後確認hole就是新元素的所在位置時，纔會真正的將新元素插入到hole上，所以在調整過程當中就比上面的代碼少了一次賦值的操做，代碼邏輯是：
下面就是shift_up()的代碼邏輯，不斷的將new的「預留位置」向上調整

view plain

1. // 下面就是shift_up()的代碼邏輯，不斷的將new的「預留位置」向上調整
2. _hole = size; // _hole就是爲new預留的位置，但並不馬上將new放上
3. while(_hole>0) // 循環
4. {
5.     _parent = (_hole-1)/2; // 計算parent
6.     if(Heap[_parent].key < new.key)
7.         break; // 調整結束，跳出循環
8.     Heap[_hole] = Heap[_parent]; // 將parent向下調整
9.     _hole = _parent; // 將_hole調整到_parent
10. }
11. Heap[_hole] = new; // 調整結束，將new插入到_hole指示的位置
12. size++; // 元素個數+1

因爲每次調整都少作一次賦值操做，在調整路徑比較長時，調整效率會比第一種有所提升。libevent中的min_heap_shift_up_()函數就是上面邏輯的具體實現，對應的向下調整函數是min_heap_shift_down_()。
舉個例子，向一個小根堆3, 5, 8, 7, 12中插入新元素2，使用第一中典型的代碼邏輯，其調整過程以下圖所示：

使用libevent中的堆調整邏輯，調整過程以下圖所示：

對於刪除和元素修改操做，也聽從相同的邏輯，就再也不羅嗦了。

 

3 小節

經過設置系統I/O機制的wait時間，從而簡捷的集成Timer事件；主要分析了libevent對堆調整操做的優化