事件驅動模型Libev（二）

Libev設計思路

理清了Libev的代碼結構和主要的數據結構，就能夠跟着示例中接口進入到Libev中，跟着代碼瞭解其設計的思路。這裏咱們管struct ev_loop稱做爲事件循環驅動器而將各類watcher稱爲事件監控器。

1.分析例子中的IO事件

這裏在前面的例子中咱們先把定時器和信號事件的使用註釋掉，只看IO事件監控器，從而瞭解Libev最基本的邏輯。能夠結合Gdb設斷點一步一步的跟看看代碼的邏輯是怎樣的。

咱們從main開始一步步走。首先執行 struct ev_loop *main_loop = ev_default_loop(0); 經過跟進代碼能夠跟到函數 ev_default_loop 裏面去，其主要邏輯，就是全局對象指針ev_default_loop_ptr若爲空，也就是未曾使用預製的驅動器時，就讓他指向全局對象default_loop_struct，同時在本函數裏面統一用名字"loop"來表示該預製驅動器的指針。從而與函數參數爲 EV_P 以及 EV_A的寫法配合。接着對該指針作 loop_init操做，即初始化預製的事件驅動器。這裏函數的調用了就是用到了 EV_A_ 這樣的寫法進行簡化。初始化以後若是配置中Libev支持子進程，那麼經過信號監控器實現了子進程監控器。這裏能夠先不用去管他，知道這段代碼做用便可。 這裏再Libev的函數定義的時候，會看到 「EV_THROW」 這個東西，這裏能夠不用管它，他是對CPP中"try … throw"的支持，和 EV_CPP(extern "C" {)這樣不一樣尋常的 extern 「C」 同樣是一種編碼技巧。如今咱們以分析設計思路爲主。在瞭解了整體後，能夠再對其編碼技巧進行梳理。不然的話看一份代碼會很是吃力，並且速度慢。甚至有的時候這些「hacker」並不必定是有益的。

1.1驅動器的初始化

下面看下驅動器的初始化過程當中都作了哪些事情。首先最開始的一段代碼判斷系統的clock_gettime是否支持CLOCK_REALTIME和CLOCK_MONOTONIC。這兩種時間的區別在於後者不會由於系統時間被修改而被修改，詳細解釋能夠參考man page 。接着判斷環境變量對驅動器的影響，這個在官方的Manual中有提到，主要就是影響默認支持的IO複用機制。接着是一連串的初始值的賦值，開始不用瞭解其做用。在後面的分析過程當中即可以知道。接着是根據系統支持的IO複用機制，對其進行初始化操做。這裏能夠去"ev_epoll.c」 和"ev_select.c"中看一下。 最後是判斷若是系統須要信號事件，那麼經過一個PIPE的IO事件來實現，這裏暫且不用管他，在理解了IO事件的實現後，天然就知道這裏他作了什麼操做。

對於"ev_epoll.c」 和"ev_select.c"中的 xxx_init 其本質是一致的，就像插件同樣，遵循一個格式，而後能夠靈活的擴展。對於epoll主要就是作了一個 epoll_create*的操做（epoll_create1能夠支持EPOLL_CLOEXEC）。

backend_mintime = 1e-3; /* epoll does sometimes return early, this is just to avoid the worst */ backend_modify = epoll_modify; backend_poll = epoll_poll; 

這裏就能夠當作是插件的模板了，在後面會修改的時候調用backend_modify在poll的時候調用backend_poll.從而統一了操做。

epoll_eventmax = 64; /* initial number of events receivable per poll */ epoll_events = (struct epoll_event *)ev_malloc (sizeof (struct epoll_event) * epoll_eventmax) 

這個就看作爲是每一個機制特有的部分。熟悉epoll的話，這個就不用說了。

對於select (Linux平臺上的)

backend_mintime = 1e-6; backend_modify = select_modify; backend_poll = select_poll; 

這個和上面同樣，是至關於插件接口

vec_ri  = ev_malloc (sizeof (fd_set)); FD_ZERO ((fd_set *)vec_ri); vec_ro = ev_malloc (sizeof (fd_set)); vec_wi = ev_malloc (sizeof (fd_set)); FD_ZERO ((fd_set *)vec_wi); vec_wo = ev_malloc (sizeof (fd_set)); 

一樣，這個是select特有的，表示讀和寫的fd_set的vector，ri用來裝select返回後符合條件的部分。其餘的如poll、kqueue、Solaris port都是相似的，能夠自行閱讀。

1.2IO監控器的初始化

上面的過程執行完了ev_default_loop過程，而後到後面的ev_init(&io_w,io_action);,他不是一個函數，而是一個宏定義：

((ev_watcher *)(void *)(ev))->active = ((ev_watcher *)(void *)(ev))->pending = 0; ev_set_priority ((ev), 0); ev_set_cb ((ev), cb_); 

這裏雖然還有兩個函數的調用，可是很好理解，就是設置了以前介紹的基類中 「active"表示是否激活該watcher，「pending」該監控器是否處於pending狀態，「priority"其優先級以及觸發後執行的動做的回調函數。

1.3 設置IO事件監控器的觸發條件

在初始化監控器後，還要設置其監控監控的條件。當該條件知足時便觸發該監控器上註冊的觸發動做。ev_io_set(&io_w,STDIN_FILENO,EV_READ);從參數邊能夠猜出他幹了什麼事情。就是設置該監控器監控標準輸入上的讀事件。該調用也是一個宏定義：

(ev)->fd = (fd_); (ev)->events = (events_) | EV__IOFDSET; 

就是設置派生類IO監控器特有的變量fd和events，表示監控那個文件fd已經其上的可讀仍是可寫事件。 %TODO:補上EV_IOFDSET的做用

1.4註冊IO監控器到事件驅動器上

準備好了監控器後就要將其註冊到事件驅動器上，這樣就造成了一個完整的事件驅動模型。 ev_io_start(main_loop,&io_w); 。這個函數裏面會第一次見到一個一個宏 「EV_FREQUENT_CHECK」,是對函數 「ev_verify"的調用，那麼ev_verify是幹什麼的呢？用文檔的話「This can be used to catch bugs inside libev itself」，若是看其代碼的話，就是去檢測Libev的內部數據結構，判斷各邊界值是否合理，不合理的時候assert掉。在生產環境下，我以爲根據性格來對待。若是以爲他消耗資源（要檢測不少東西跑不少循環）能夠編譯的時候關掉該定義。若是須要assert，能夠在編譯的時候加上選項。

而後看到 ev_start 調用，該函數實際上就是給驅動器的loop->activecnt增一併置loop->active爲真(這裏統一用loop表示全局對象的預製驅動器對象default_loop_struct)，他們分別表示事件驅動器上正監控的監控器數目以及是否在爲監控器服務。

array_needsize (ANFD, anfds, anfdmax, fd + 1, array_init_zero); wlist_add (&anfds[fd].head, (WL)w); 

感興趣的能夠去看下Libev裏麼動態調整數組的實現。這裏咱們主要看總體邏輯。他的工做過程是先判斷數組anfds是否還有空間再加對文件描述符fd的監控，，沒有的話則調整數組的內存大小，使其大小足以容下。

這裏要介紹下以前沒有介紹的一個數據結構，這個沒有上下文比較難理解，所以放在這裏介紹。

typedef struct { WL head; unsigned char events; /* the events watched for */ unsigned char reify; /* flag set when this ANFD needs reification (EV_ANFD_REIFY, EV__IOFDSET) */ unsigned char emask; /* the epoll backend stores the actual kernel mask in here */ unsigned char unused; unsigned int egen; /* generation counter to counter epoll bugs */ } ANFD; /* 這裏去掉了對epoll的判斷和windows的IOCP*/ 

這裏首先只用關注一個 「head」 ，他是以前說過的wather的基類鏈表。這裏一個ANFD就表示對一個文件描述符的監控，那麼對該文件描述的可讀仍是可寫監控，監控的動做是如何定義的，就是經過這個鏈表，把對該文件描述法的監控器都掛上去，這樣就能夠經過文件描述符找到了。而前面的說的anfds就是這個對象的數組，下標經過文件描述符fd進行索引。在Redis-ae那篇文章中已經討論過這樣的能夠達到O(1)的索引速度並且空間佔用也是合理的。

接着的「fd_change」與「fd_reify」是呼應的。前者將fd添加到一個fdchanges的數組中，後者則依次遍歷這個數組中的fd上的watcher與anfds裏面對飲的watcher進行對比，判斷監控條件是否改變了，若是改變了則調用backend_modify也就是epoll_ctl等調整系統對該fd的監控。這個fdchanges數組的做用就在於此，他記錄了anfds數組中的watcher監控條件可能被修改的文件描述符，並在適當的時候將調用系統的epoll_ctl或則其餘文件複用機制修改系統監控的條件。這裏咱們把這兩個主要的物理結構梳理下:

總結一下注冊過程就是經過以前設置了監控條件IO watcher得到監控的文件描述符fd，找到其在anfds中對應的ANFD結構，將該watcher掛到該結構的head鏈上。因爲對應該fd的監控條件有改動了，所以在fdchanges數組中記錄下該fd，在後續的步驟中調用系統的接口修改對該fd監控的條件。

1.5 啓動事件驅動器

一切準備就緒了就能夠開始啓動事情驅動器了。就是 ev_run。 其邏輯很清晰。就是

do{ xxxx; backend_poll(); xxxx }while(condition_is_ok) 

循環中開始一段和fork 、 prepare相關這先直接跳過，到分析與之相關的監控事件纔去看他。直接到 /* calculate blocking time */ 這裏。熟悉事件模型的話，這裏仍是比較常規的。就是從定時器堆中取得最近的時間（固然這裏分析的時候沒有定時器）與loop->timeout_blocktime比較獲得阻塞時間。這裏若是設置了驅動器的io_blocktime，那麼在進入到poll以前會先sleep io_blocktime時間從而等待IO或者其餘要監控的事件準備。這裏進入到backend_poll中的阻塞時間是包括了io_blocktime的時間。而後進入到backend_poll中。對於epoll就是進入到epoll_wait裏面。

epoll(或者select、kqueue等)返回後，將監控中的文件描述符fd以及其pending（知足監控）的條件經過 fd_event作一個監控條件是否改變的判斷後到fd_event_nocheck裏面對anfds[fd]數組中的fd上的掛的監控器依次作檢測，若是pending條件符合，便經過ev_feed_event將該監控器加入到pendings數組中pendings[pri]上的pendings[pri][old_lenght+1]的位置上。這裏要介紹一個新的數據結構，他表示pending中的wather也就是監控條件知足了，可是尚未觸發動做的狀態。

typedef struct { W w; int events; /* the pending event set for the given watcher */ } ANPENDING; 

這裏 W w應該知道是以前說的基類指針。pendings就是這個類型的一個二維數組數組。其以watcher的優先級爲一級下標。再以該優先級上pengding的監控器數目爲二級下標，對應的監控器中的pending值就是該下標加一的結果。其定義爲 ANPENDING *pendings [NUMPRI]。同anfds同樣，二維數組的第二維 ANPENDING *是一個動態調整大小的數組。這樣操做以後。這個一系列的操做能夠認爲是fd_feed的後續操做，xxx_reify目的最後都是將pending的watcher加入到這個pengdings二維數組中。後續的幾個xxx_reify也是同樣，等分析到那個類型的監控器類型時在做展開。 這裏用個圖梳理下結構。

最後在循環中執行宏EV_INVOKE_PENDING，實際上是調用loop->invoke_cb，若是沒有自定義修改的話（通常不會修改）就是調用ev_invoke_pending。該函數會依次遍歷二維數組pendings，執行pending的每個watcher上的觸發動做回調函數。

至此一次IO觸發過程就完成了。

2總結出Libev的設計思路

在Libev中watcher要算最關鍵的數據結構了，整個邏輯都是圍繞着watcher作操做。Libev內部維護一個基類ev_wathcer和若干個特定監控器的派生類ev_xxx。在使用的時候首先生成一個特定watcher的實例。並經過該派生對象私有的成員設置其觸發條件。而後用anfds或者最小堆管理這些watchers。而後Libev經過backend_poll以及時間堆管理運算出pending的watcher。而後將他們加入到一個以優先級爲一維下標的二維數組。在合適的時間依次調用這些pengding的watcher上註冊的觸發動做回調函數，這樣即可以按優先級前後順序實現「only-for-ordering」的優先級模型。

 

 

 

另外兩個重要的監控器

前面經過IO監控器將Libev的整個工做流程過了一遍。中間濾過了不少與其餘事件監控器相關的部分，可是總體思路以及很明晰了，只要針對其餘類型的watcher看下其初始化和註冊過程以及在ev_run中的安排便可。這裏咱們再分析另兩個經常使用的watcher

1.分析定時器監控器

定時器在程序中能夠作固定週期tick操做，也能夠作一次性的定時操做。Libev中與定時器相似的還有個週期事件watcher。其本質都是同樣的，只是在時間的計算方法上略有不一樣，並有他本身的一個事件管理的堆。對於定時器事件，咱們按照以前說的順序從ev_init開始看起。

1.1定時器監控器的初始化

定時器初始化使用 ev_init(&timer_w,timer_action);,這個過程和以前的IO相似，主要就是設置基類的active、pending、priority以及觸發動做回調函數cb。

1.2設置定時器監控器的觸發條件

經過 ev_timer_set(&timer_w,2,0);能夠設置定時器在2秒鐘後被觸發。若是第三個參數不是0而是一個大於0的正整數n時，那麼在第一次觸發（2秒後），每隔n秒會再次觸發定時器事件。

其爲一個宏定義 do { ((ev_watcher_time *)(ev))->at = (after_); (ev)->repeat = (repeat_); } while (0) 也就是設置派生類定時器watcher的「at」爲觸發事件，以及重複條件「repeat」。

1.3將定時器註冊到事件驅動器上

ev_timer_start(main_loop,&timer_w);會將定時器監控器註冊到事件驅動器上。其首先 ev_at (w) += mn_now; 獲得將來的時間，這樣放到時間管理的堆「timers」中做爲權重。而後經過以前說過的「ev_start」修改驅動器loop的狀態。這裏咱們又看到了動態大小的數組了。Libev的堆的內存管理也是經過這樣的關係的。具體這裏堆的實現，感興趣的能夠仔細看下實現。這裏的操做就是將這個時間權重放到堆中合適的位置。這裏堆單元的結構爲：

typedef struct { ev_tstamp at; WT w; } ANHE; 

其實質就是一個時刻at上掛一個放定時器watcher的list。當超時時會依次執行這些定時器watcher上的觸發回調函數。

1.4定時器監控器的觸發

最後看下在一個事件驅動器循環中是如何處理定時器監控器的。這裏咱們依然拋開其餘的部分，只找定時器相關的看。在「/ calculate blocking time /」塊裏面，咱們看到計算blocking time的時候會先：

if (timercnt) { ev_tstamp to = ANHE_at (timers [HEAP0]) - mn_now; if (waittime > to) waittime = to; } 

若是有定時器，那麼就從定時器堆（一個最小堆）timers中取得堆頂上最小的一個時間。這樣就保證了在這個時間前能夠從backend_poll中出來。出來後執行timers_reify處理將pengding的定時器。

在timers_reify中依次取最小堆的堆頂，若是其上的ANHE.at小於當前時間，表示該定時器watcher超時了，那麼將其壓入一個數組中，因爲在實際執行pendings二維數組上對應優先級上的watcher是從尾往頭方向的，所以這裏先用一個數組依時間前後次存下到一箇中間數組loop->rfeeds中。而後將其逆序調用ev_invoke_pending插入到pendings二維數組中。這樣在執行pending事件的觸發動做的時候就能夠保證，時間靠前的定時器優先執行。函數 feed_reverse和 feed_reverse_done就是將超時的定時器加入到loop->rfeeds暫存數組以及將暫存數組中的pending的watcher插入到pengdings數組的操做。把pending的watcher加入到pendings數組，後續的操做就和以前的同樣了。回依次執行相應的回調函數。

這個過程當中還判判定時器的 w->repeat 的值，若是不爲0，那麼會重置該定時器的時間，並將其壓入堆中正確的位置，這樣在指定的時間事後又會被執行。若是其爲0，那麼調用ev_timer_stop關閉該定時器。 其首先經過clear_pending置pendings數組中記錄的該watcher上的回調函數爲一個不執行任何動做的啞動做。

總結一下定時器就是在backend_poll以前經過定時器堆頂的超時時間，保證blocking的時間不超過最近的定時器時間，在backend_poll返回後，從定時器堆中取得超時的watcher放入到pendings二維數組中，從而在後續處理中能夠執行其上註冊的觸發動做。而後從定時器管理堆上刪除該定時器。最後調用和ev_start呼應的ev_stop修改驅動器loop的狀態，即loop->activecnt減小一。並將該watcher的active置零。

對於週期性的事件監控器是一樣的處理過程。只是將timers_reify換成了periodics_reify。其內部會對週期性事件監控器派生類的作相似定時器裏面是否repeat的判斷操做。判斷是否從新調整時間，或者是否重複等邏輯，這些看下代碼比較容易理解，這裏再也不贅述。·

2.分析信號監控器

分析完了定時器的部分，再看下另外一個比較經常使用的信號事件的處理。Libev裏面的信號事件和Tornado.IOLoop是同樣的，經過一個pipe的IO事件來處理。直白的說就是註冊一個雙向的pipe文件對象，而後監控上面的讀事件，待相應的信號到來時，就往這個pipe中寫入一個值然他的讀端的讀事件觸發，這樣就能夠執行相應註冊的觸發動做回調函數了。

咱們仍是從初始化-》設置觸發條件-》註冊到驅動器-》觸發過程這樣的順序介紹。

2.1信號監控器的初始化

ev_init(&signal_w,signal_action);這個函數和上面的同樣不用說了

2.2設置信號監控器的觸發條件

ev_signal_set(&signal_w,SIGINT);該函數設置了Libev收到SIGINT信號是觸發註冊的觸發動做回調函數。其操做和上面的同樣，就是設置了信號監控器私有的(ev)->signum爲標記。

2.3將信號監控器註冊到驅動器上

這裏首先介紹一個數據結構：

typedef struct { EV_ATOMIC_T pending; EV_P; WL head; } ANSIG; static ANSIG signals [EV_NSIG - 1]; 

EV_ATOMIC_T pending;能夠認爲是一個原子對象，對他的讀寫是原子的。一個表示事件驅動器的loop，以及一個watcher的鏈表。

在ev_signal_start中，經過signals數組存儲信號監控單元。該數組和anfds數組相似，只是他以信號值爲索引。這樣能夠立馬找到信號所在的位置。從 Linux 2.6.27之後，Kernel提供了signalfd來爲信號產生一個文件描述符從而能夠用文件複用機制epoll、select等來管理信號。Libev就是用這樣的方式來管理信號的。 這裏的代碼用宏控制了。其邏輯大致是這樣的

#if EV_USE_SIGNALFD res = invoke_signalfd # if EV_USE_SIGNALFD if (res is not valied) # endif { use evpipe to instead } 

這個是框架。其具體的實現能夠參考使用signalfd和evpipe_init實現。其實質就是經過一個相似於管道的文件描述符fd，設置對該fd的讀事件監聽，當收到信號時經過signal註冊的回調函數往該fd裏面寫入，使其讀事件觸發，這樣經過backend_poll返回後就能夠處理ev_init爲該信號上註冊的觸發回調函數了。

在函數evpipe_init裏面也用了一個能夠學習的技巧，和上面的#if XXX if() #endif {} 同樣，處理了不支持eventfd的狀況。eventfd是Kernel 2.6.22之後才支持的系統調用，用來建立一個事件對象實現，進程(線程)間的等待/通知機制。他維護了一個能夠讀寫的文件描述符，可是隻能寫入8byte的內容。可是對於咱們的使用以及夠了，由於這裏主要是得到其可讀的狀態。對於不支持eventfd的狀況，則使用上面說過的，用系統的pipe調用產生的兩個文件描述符分別作讀寫對象，來完成。

2.4信號事件監控器的觸發

在上面設置信號的pipe的IO事件是，根據使用的機制不一樣，其實現和觸發有點不一樣。對於signalfd。

ev_io_init (&sigfd_w, sigfdcb, sigfd, EV_READ);  /*  for signalfd */ ev_set_priority (&sigfd_w, EV_MAXPRI); ev_io_start (EV_A_ &sigfd_w); 

也就是註冊了sigfdcb函數。該函數：

ssize_t res = read (sigfd, si, sizeof (si)); for (sip = si; (char *)sip < (char *)si + res; ++sip) ev_feed_signal_event (EV_A_ sip->ssi_signo); 

首先將pipe內容讀光，讓後續的能夠pengding在該fd上。而後對該signalfd上的全部信號弟阿勇ev_feed_signal_event吧每一個信號上的ANSIG->head上掛的watcher都用ev_feed_event加入到pendings二維數組中。這個過程和IO的徹底同樣。

而對於eventfd和pipe則是:

ev_init (&pipe_w, pipecb); ev_set_priority (&pipe_w, EV_MAXPRI); ev_io_set (&pipe_w, evpipe [0] < 0 ? evpipe [1] : evpipe [0], EV_READ); ev_io_start (EV_A_ &pipe_w); 

pipe_w是驅動器自身的loop->pipe_w。併爲其設置了回調函數pipecb：

#if EV_USE_EVENTFD if (evpipe [0] < 0) { uint64_t counter; read (evpipe [1], &counter, sizeof (uint64_t)); } else #endif { char dummy[4]; read (evpipe [0], &dummy, sizeof (dummy)); } ... xxx ... for (i = EV_NSIG - 1; i--; ) if (expect_false (signals [i].pending)) ev_feed_signal_event (EV_A_ i + 1); 

這裏將上面的技巧#if XXX if() #endif {}拓展爲了#if XXX if() {} else #endif {} 。這裏和上面的操做實際上是同樣的。後續操做和signalfd裏面同樣，就是讀光pipe裏面的內容，而後依次將watcher加入到pendings數組中。

 

 

其餘監控器

最主要的幾個監控器搞定了。其餘的我以爲比較能夠看的還有ev_child和ev_stat。其實和以前的三個基本原理的是同樣。暫不贅述。將來可能補充。

Libev中的Tips

若是將Libev當成組件去用的話。官方文檔是一份很好的選擇。這裏說下看Libev過程當中的感覺。

若是使用Libev但又以爲它沒有提供必要的功能而要去該其代碼。可能Libuv爲咱們作了一個很好的示例。Libuv以前是用Libev做爲其底層事件庫。後來做者重寫了本身的一套網絡庫Libuv。嚴格意義上說，Libev僅僅是一個事件模型框架，並不能算上是一個完整的網絡庫，正由於如此他才提供瞭如此多的事件類型。而對於網絡庫可能最重要的就是定時器、IO、以及信號事件。固然網絡還包括了socket、收發控制等內容。所以，個人感受是能夠將Libev當成一個很好的學習對象，不管是其設計思想、仍是代碼中個各類小tips、還有其對跨平臺支持的方法都是很好的示例。雖然用宏包裹的比較嚴密，只要稍加分析，理清其思路仍是比較容易的。

將Libev和以前的Redis-ae進行對比。能夠發現Libev在設計思想上更完整，提供的服務也更全，可是作的檢測多了，邏輯複雜了，消耗的資源也一定比簡單的封裝更多。從這個兩個模型能夠看出事件模型的框架都是：

取得一個合適的時間，用這個時間去poll。而後標記poll以後pending的文件對象。poll出來後判判定時器而後統一處理pending對象

這裏繪製一個總體的結構圖，不是很規範UML或者其餘什麼學術的圖，只是一個幫助理解的過程：

至此Libev的分析差很少完成了，主要去了解實現的思路。具體如何實現以及從什麼樣的角度去設計。其結果須要在生產環境中去檢驗。

 

 

轉載：http://my.oschina.net/u/917596/blog/177573