? ? ? ? 事件驅動這個名詞出現的越來越頻繁了，聽起來非常高大上，今天本人把Redis內部的驅動模型研究了一番，感覺收獲頗豐啊。一個ae.c主程序，加上4個事件類型的文件，讓你徹底弄清楚，Redis是如何處理這些事件的。在Redis的事件處理中，用到了epoll，select，kqueue和evport，evport可能大家會陌生許多。前面3個都是非常常見的事件，在libevent的事件網絡庫中也都有出現。作者在寫這個事件驅動模型的時候，也說了，這只是為了簡單的復用了，設計的一個小小的處理模型: ~~~ /* A simple event-driven programming library. Originally I wrote this code * for the Jim's event-loop (Jim is a Tcl interpreter) but later translated * it in form of a library for easy reuse. * * ae是作者寫的一個簡單的事件驅動庫，后面進行了轉化，變得更為簡單的復用 ~~~ 所以不是很復雜。在了解整個事件驅動的模型前，有先了解一些定義的事件結構體，事件類型總共2個一個FileEvent，TimeEvent: ~~~ /* File event structure */ /* 文件事件結構體 */ typedef struct aeFileEvent { //只為讀事件或者寫事件中的1種 int mask; /* one of AE_(READABLE|WRITABLE) */ //讀方法 aeFileProc *rfileProc; //寫方法 aeFileProc *wfileProc; //客戶端數據 void *clientData; } aeFileEvent; /* Time event structure */ /* 時間事件結構體 */ typedef struct aeTimeEvent { //時間事件id long long id; /* time event identifier. */ //時間秒數 long when_sec; /* seconds */ //時間毫秒 long when_ms; /* milliseconds */ //時間事件中的處理函數 aeTimeProc *timeProc; //被刪除的時候將會調用的方法 aeEventFinalizerProc *finalizerProc; //客戶端數據 void *clientData; //時間結構體內的下一個結構體 struct aeTimeEvent *next; } aeTimeEvent; /* A fired event */ /* fired結構體，用來表示將要被處理的文件事件 */ typedef struct aeFiredEvent { //文件描述符id int fd; int mask; } aeFiredEvent; ~~~ FireEvent只是用來標記要處理的文件Event。 這些事件都存在于一個aeEventLoop的結構體內: ~~~ /* State of an event based program */ typedef struct aeEventLoop { //目前創建的最高的文件描述符 int maxfd; /* highest file descriptor currently registered */ int setsize; /* max number of file descriptors tracked */ //下一個時間事件id long long timeEventNextId; time_t lastTime; /* Used to detect system clock skew */ //3種事件類型 aeFileEvent *events; /* Registered events */ aeFiredEvent *fired; /* Fired events */ aeTimeEvent *timeEventHead; //事件停止標志符 int stop; //這里存放的是event API的數據，包括epoll，select等事件 void *apidata; /* This is used for polling API specific data */ aeBeforeSleepProc *beforesleep; } aeEventLoop; ~~~ 在每種事件內部，都有定義相應的處理函數，把函數當做變量一樣存在結構體中。下面看下ae.c中的一些API的組成: ~~~ /* Prototypes */ aeEventLoop *aeCreateEventLoop(int setsize); /* 創建aeEventLoop，內部的fileEvent和Fired事件的個數為setSize個 */ void aeDeleteEventLoop(aeEventLoop *eventLoop); /* 刪除EventLoop，釋放相應的事件所占的空間 */ void aeStop(aeEventLoop *eventLoop); /* 設置eventLoop中的停止屬性為1 */ int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask, aeFileProc *proc, void *clientData); /* 在eventLoop中創建文件事件 */ void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask); /* 刪除文件事件 */ int aeGetFileEvents(aeEventLoop *eventLoop, int fd); //根據文件描述符id，找出文件的屬性，是讀事件還是寫事件 long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds, aeTimeProc *proc, void *clientData, aeEventFinalizerProc *finalizerProc); /* 在eventLoop中添加時間事件，創建的時間為當前時間加上自己傳入的時間 */ int aeDeleteTimeEvent(aeEventLoop *eventLoop, long long id); //根據時間id，刪除時間事件，涉及鏈表的操作 int aeProcessEvents(aeEventLoop *eventLoop, int flags); /* 處理eventLoop中的所有類型事件 */ int aeWait(int fd, int mask, long long milliseconds); /* 讓某事件等待 */ void aeMain(aeEventLoop *eventLoop); /* ae事件執行主程序 */ char *aeGetApiName(void); void aeSetBeforeSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *beforesleep); /* 每次eventLoop事件執行完后又重新開始執行時調用 */ int aeGetSetSize(aeEventLoop *eventLoop); /* 獲取eventLoop的大小 */ int aeResizeSetSize(aeEventLoop *eventLoop, int setsize); /* EventLoop重新調整大小 */ ~~~ 無非涉及一些文件，時間事件的添加，修改等，都是在eventLoop內部的修改，我們來看下最主要，最核心的方法: ~~~ /* ae事件執行主程序 */ void aeMain(aeEventLoop *eventLoop) { eventLoop->stop = 0; //如果eventLoop中的stop標志位不為1，就循環處理 while (!eventLoop->stop) { //每次eventLoop事件執行完后又重新開始執行時調用 if (eventLoop->beforesleep != NULL) eventLoop->beforesleep(eventLoop); //while循環處理所有的evetLoop的事件 aeProcessEvents(eventLoop, AE_ALL_EVENTS); } } ~~~ 道理很簡單通過，while循環，處理eventLoop中的所有類型事件，截取部分processEvents()代碼: ~~~ numevents = aeApiPoll(eventLoop, tvp); for (j = 0; j < numevents; j++) { aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd]; int mask = eventLoop->fired[j].mask; int fd = eventLoop->fired[j].fd; int rfired = 0; /* note the fe->mask & mask & ... code: maybe an already processed * event removed an element that fired and we still didn't * processed, so we check if the event is still valid. */ if (fe->mask & mask & AE_READABLE) { rfired = 1; //根據掩碼計算判斷是否為ae讀事件，調用時間中的讀的處理方法 fe->rfileProc(eventLoop,fd,fe->clientData,mask); } if (fe->mask & mask & AE_WRITABLE) { if (!rfired || fe->wfileProc != fe->rfileProc) fe->wfileProc(eventLoop,fd,fe->clientData,mask); } processed++; } } ~~~ ae中創建時間事件都是以當前時間為基準創建的； ~~~ /* 在eventLoop中添加時間事件，創建的時間為當前時間加上自己傳入的時間 */ long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds, aeTimeProc *proc, void *clientData, aeEventFinalizerProc *finalizerProc) { long long id = eventLoop->timeEventNextId++; aeTimeEvent *te; te = zmalloc(sizeof(*te)); if (te == NULL) return AE_ERR; te->id = id; aeAddMillisecondsToNow(milliseconds,&te->when_sec,&te->when_ms); te->timeProc = proc; te->finalizerProc = finalizerProc; te->clientData = clientData; //新加的變為timeEvent的頭部 te->next = eventLoop->timeEventHead; eventLoop->timeEventHead = te; //返回新創建的時間事件的id return id; } ~~~ ? 下面說說如何調用事件API庫里的方法呢。首先隆重介紹什么是epoll,poll，select，kqueu和evport。這些都是一種事件模型。 select事件的模型? （1）創建所關注的事件的描述符集合（fd_set)，對于一個描述符，可以關注其上面的讀(read)、寫(write)、異常(exception)事件,所以通常，要創建三個fd_set， 一個用來收集關注讀事件的描述符，一個用來收集關注寫事件的描述符，另外一個用來收集關注?異常事件的描述符集合。 （2）輪詢所有fd_set中的每一個fd ，檢查是否有相應的事件發生，如果有，就進行處理。 ? poll和上面的區別是可以復用文件描述符，上面對一個文件需要輪詢3個文件描述符集合，而poll只需要一個，效率更高 epoll是poll的升級版本，把描述符列表交給內核，一旦有事件發生，內核把發生事件的描述符列表通知給進程，這樣就避免了輪詢整個描述符列表。效率極大提高? evport這個出現的比較少，大致意思是evport將某一個對象的特定 event 與 Event port 相關聯： 在了解了3種事件模型的原理之后，我們看看ae.c在Redis中是如何調用的呢, ~~~ //這里存放的是event API的數據，包括epoll，select等事件 void *apidata; /* This is used for polling API specific data */ ~~~ 就是上面這個屬性，在上面的4種事件中，分別對應著3個文件，分別為ae_poll.c,ae_select.c，但是他們的API結構是類似的，我舉其中一個例子，epoll的例子,首先都會有此事件特定的結構體: ~~~ typedef struct aeApiState { int epfd; struct epoll_event *events; } aeApiState; ~~~ 還有共同套路的模板方法: ~~~ static int aeApiCreate(aeEventLoop *eventLoop) static int aeApiResize(aeEventLoop *eventLoop, int setsize) static void aeApiFree(aeEventLoop *eventLoop) static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask) static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) static char *aeApiName(void) ~~~ 在創建的時候賦值到eventloop的API data里面去: ~~~ state->epfd = epoll_create(1024); /* 1024 is just a hint for the kernel */ if (state->epfd == -1) { zfree(state->events); zfree(state); return -1; } //最后將state的數據賦值到eventLoop的API data中 eventLoop->apidata = state; return 0; ~~~ 在取出事件的poll方法的時候是這些方法的一個區分點： ~~~ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) { aeApiState *state = eventLoop->apidata; int retval, numevents = 0; retval = epoll_wait(state->epfd,state->events,eventLoop->setsize, tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1); if (retval > 0) { ..... ~~~ 而在select中的poll方法是這樣的: ~~~ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) { aeApiState *state = eventLoop->apidata; int retval, j, numevents = 0; memcpy(&state->_rfds,&state->rfds,sizeof(fd_set)); memcpy(&state->_wfds,&state->wfds,sizeof(fd_set)); retval = select(eventLoop->maxfd+1, &state->_rfds,&state->_wfds,NULL,tvp); ...... ~~~ 最后都是基于state中的事件和eventLoop之間的轉化實現操作。傳入eventLoop中的信息，傳入state的信息，經過內部的處理得出終的事件結果。調用就這么簡單。