高效能網路庫分析1——libevent
libevent事件迴圈分析
libevent庫事件迴圈封裝在event_base_loop()函式中，函式的主迴圈流程：
while(1){ if(base中有事件或loop需要阻塞){ timeout_next(base, &tv_p); // 從timer heap中取根節點的超時時間，作為IO複用的阻塞等待時間tv_p（tv_p指向tv） } else // 無需阻塞IO複用等待 evutil_timerclear(&tv) //IO複用阻塞等待時長清0- … clear_time_cache(base); // 清除base物件中系統時間快取tv_catch 1) update_time_catch(base); // 跟新base物件的系統時間快取 2) timeout_process(base); //定時器啟用事件檢測 … 3) event_process_active(base); // 處理啟用的事件 } |
看起來過程很複雜，實際上歸納一下，一輪迴圈也就分為兩個階段：
1）啟用事件檢測階段
其中，由於libevent時間管理的策略原因，啟用事件檢測階段又分為兩個子階段：
1.1）定時器啟用事件檢測，evsel->dispatch(
1.2）IO事件及signal事件檢測；timeout_process(base)。
2）啟用事件處理階段
遍歷各級啟用事件優先順序佇列，逐佇列逐事件呼叫事件回撥函式，處理事件。
首先，談下libevent庫中時間管理的策略
libevent採用monotonic時間，作為定時器是否啟用的依據。（CLOCK_MONOTONIC：以絕對時間為準，獲取的時間為系統重啟到現在的時間，更改系統時間對它沒有影響。）
使用者呼叫，
evtimer_set(&ev,timeer_cb, NULL)
構造一個timer事件，然後呼叫，
event_add(&ev,&tv);
為定時器事件設定超時值，在event_add()函式內部會呼叫event_add_nolock_()函式判斷插入事件的型別，然後將事件插入對因事件列表或者堆中，其中在處理定時器事件時，操作如下
int event_add_nolock_(struct event *ev, const struct timeval *tv, int tv_is_absolute){ … if (res != -1 && tv != NULL){ // 表明事件型別為定時器 … // 將當前MONOTONIC時間+使用者設定的定時器超時值作為定時器超時啟用的絕對時間 evutil_timeradd(&now, tv, &ev->ev_timeout); … // 將定時器事件插入到小根堆中 event_queue_insert_timeout(base, ev); } … } |
可以看到，實際上事件真正的超時值是一個絕對時間。所以，若將libevent事件迴圈看做一個生命週期，定時器事件則是生命週期中的固定點（同步點），而IO&signal事件由於是非同步發生的，所以是生命週期中的非同步點，如下圖：
libevent定時器與muduo定時器的區別：
libevent庫中定時器機採用一種弱定時器機制，利用select()等IO複用系統呼叫的阻塞時長來實現定時器功能，因此使定時器具備平臺無關性，但缺點使不太精確，且要將定時器事件的處理與其他事件分開處理。
而muduo庫針對linux平臺設計，定時器採用timerfd定時器檔案描述符，雖然在精準度上也沒比libevent提高多少，但由於使用檔案描述符，使得定時器事件的處理與IO，檔案等事件的處理統一，利於開發。
回到事件主迴圈，簡單分析下2個階段核心函式的操作
1）evsel->dispatch()
這個部分，實際上就是封裝了IO複用那一套東西，根據平臺對IO複用的支援，libevent可以使用select()，poll()，epoll()，kqueue()。網上有大量相關資料，不再贅述。
2）timerout_process(base)
首先，這個函式的命名方式非常容易引起歧義，一開始我以為這個函式是為了處理因超時被啟用的定時器事件，結果閱讀原始碼後發現這個原來是檢測timer heap中哪些定時器事件超時並激活他們的函式。和evsel->dispatch()同屬啟用事件檢測階段…，下面是關鍵程式碼：
timeout_process(base){ gettime(base, &now); // 獲取當前系統絕對時間 while(不斷從timer heap根取定時器事件用於檢測){ if(evutil_timercmp(ev->ev_timeout, now, >)) { // 這是一個巨集，實際就是比對一個定時器事件的超時時間與當前時間，若堆頂定時器超時時間晚於當前時間，說明堆中無能啟用事件，退出迴圈。 break; } // else // 存在能啟用的定時器事件,處理描述如下 { /*首先，將事件從timer heap中取下， 然後，檢測事件是否已在啟用佇列，若在就從啟用佇列中移除（重置事件）*/ event_del_nolock_(ev, EVENT_DEL_NOBLOCK); /*最後，將定時器事件插入到啟用佇列中等待處理*/ event_active_nolock_(ev,EV_TIMEOUT,1); } } } |
3）event_process_active(base)
event_process_active()函式處理經過evsel->dispatch()函式檢測過得的啟用事件。libevent庫將啟用事件按優先順序組織為多個優先順序連結串列。event_process_active函式的職責就是按優先順序從高到低處理啟用事件（呼叫其事件控制代碼）。
event_process_active(){ for(inti=0;i<base→nativequeues;i++){ activeq= &base->activequeues[i]; //引數二為回撥函式為某一優先順序梯隊的啟用事件回撥函式連結串列，引數三為陣列最大回調函式個數，函式處理一個優先順序梯隊下所有啟用事件 event_process_active_single_queue(base,activeq,maxcb,…) } } |
其中event_process_active_single_queue處理一個優先順序梯度下所有啟用事件，過程如下：
structevent_callback *evcb; for(/*遍歷回撥函式列表-> evcb*/){ /*先從啟用佇列中刪除事件*/ event_queue_remove_active(base,evcb); /*然後根據evcb物件的型別呼叫事件控制代碼*/ switch(evcb→evcb_closure){ //此處用到closure一詞，可能作者思路是事件關閉前呼叫的意思 caseEV_CLOSURE_EVENT_SIGNAL: event_signal_closure(base,ev); //處理signal break; caseEV_CLOSURE_EVENT_PERSIST: event_persist_closure(base,ev); //處理週期性定時器事件 break; caseEV_CLOSURE_EVENT: evcb_callback(ev->ev_fd,res, ev→ev_arg); // IO事件，timer事件 break; ... } } |