Redis事件驅(qū)動(dòng)(aeEventLoop)原理分析
關(guān)于Redis事件驅(qū)動(dòng)
眾所周知，Redis是高性能的、基于內(nèi)存的、k-v數(shù)據(jù)庫。其強(qiáng)大的功能背后，存在著2種不同類型的事件驅(qū)動(dòng)，包括：
- 文件事件(File event)
- 時(shí)間事件(Time event)
文件事件是對(duì)相關(guān)的 fd 相關(guān)操作的封裝，時(shí)間事件則是對(duì)定時(shí)任務(wù)相關(guān)操作的封裝。Redis server通過文件事件來進(jìn)行外部請(qǐng)求的處理與操作，通過時(shí)間事件來對(duì)系統(tǒng)內(nèi)部產(chǎn)生的定時(shí)任務(wù)進(jìn)行處理。(本文重點(diǎn)講解文件事件相關(guān)的操作流程以及原理)
文中探討的原理及源碼基于Redis官方 v7.0 版本
Redis事件驅(qū)動(dòng)的相關(guān)源碼
在Redis源碼中，涉及事件驅(qū)動(dòng)相關(guān)的源碼文件主要有以下幾個(gè)(以ae作為文件名稱前綴)：
src
├── ae.c
├── ae.h
├── ae_epoll.c
├── ae_evport.c
├── ae_kqueue.c
└── ae_select.c- ae.c 文件事件驅(qū)動(dòng)/時(shí)間事件驅(qū)動(dòng)的核心處理邏輯
- ae.h文件事件驅(qū)動(dòng)/時(shí)間事件驅(qū)動(dòng)結(jié)構(gòu)體、方法簽名的定義
- ae_epoll.c linux os 文件事件驅(qū)動(dòng)涉及的i/o多路復(fù)用實(shí)現(xiàn)
- ae_evport.c sun os 文件事件驅(qū)動(dòng)涉及的i/o多路復(fù)用實(shí)現(xiàn)
- ae_kqueue.c mac/BSD os 文件事件驅(qū)動(dòng)涉及的os i/o多路復(fù)用實(shí)現(xiàn)
- ae_select.c 其他 os 文件事件驅(qū)動(dòng)涉及的i/o多路復(fù)用實(shí)現(xiàn)(或者說是通用型的，包括Windows)
根據(jù)源碼中注釋(ae.c)可知 ae 的含義為 A simple event-driven。
/* 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.
*/一個(gè)簡(jiǎn)單的事件驅(qū)動(dòng)編程庫。最初我(作者:antirez)為Jim的事件循環(huán)（Jim是Tcl解釋器）編寫了這段代碼，但后來將其轉(zhuǎn)化為庫形式以便于重用。
多種i/o多路復(fù)用方法的選擇
在Redis源碼中存在多種i/o多路復(fù)用實(shí)現(xiàn)方式，如何選擇使用哪種i/o多路復(fù)用實(shí)現(xiàn)呢？源碼編譯時(shí)選擇不同的實(shí)現(xiàn)方式，即：Redis源碼編譯成二進(jìn)制文件的時(shí)候來選擇對(duì)應(yīng)的實(shí)現(xiàn)方式，在源碼可以看到蛛絲馬跡。
代碼文件: ae.c
#ifdef HAVE_EVPORT
#include "ae_evport.c"
#else
#ifdef HAVE_EPOLL
#include "ae_epoll.c"
#else
#ifdef HAVE_KQUEUE
#include "ae_kqueue.c"
#else
#include "ae_select.c"
#endif
#endif
#endif從上面代碼可知，在編譯源碼的預(yù)處理階段，根據(jù)不同的編譯條件(#ifdef/#else/#endif)來判斷對(duì)應(yīng)的宏是否定義(#define定義的常量)來加載實(shí)現(xiàn)邏輯。以epoll為例，若定義了 HAVE_EPOLL 宏，則加載 "ae_epoll.c" 文件。宏 "HAVE_EVPORT/HAVE_EPOLL/HAVE_KQUEUE" 分別對(duì)應(yīng)不同的系統(tǒng)(或者說是對(duì)應(yīng)的編譯器)。
代碼文件: config.h
#ifdef __sun
#include <sys/feature_tests.h>
#ifdef _DTRACE_VERSION
#define HAVE_EVPORT 1
#define HAVE_PSINFO 1
#endif
#endif
#ifdef __linux__
#define HAVE_EPOLL 1
#endif
#if (defined(__APPLE__) && defined(MAC_OS_X_VERSION_10_6)) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined (__NetBSD__)
#define HAVE_KQUEUE 1
#endif假設(shè)，當(dāng)前是linux系統(tǒng)，那么 宏__linux__ 又是從哪里來的呢？Linux環(huán)境下主要用gcc編譯，借助 gcc -dM -E - < /dev/null 命令從獲得相應(yīng)的變量中可以看到其定義。
root@ivansli ~# gcc -dM -E - < /dev/null | grep __linux
#define __linux 1
#define __linux__ 1即：Redis源碼會(huì)根據(jù)編譯器來判斷應(yīng)該把源碼編譯成對(duì)應(yīng)平臺(tái)(或者是通用平臺(tái)，性能會(huì)有所下降)運(yùn)行的二進(jìn)制可執(zhí)行程序。
核心結(jié)構(gòu)體 aeEventLoop
aeEventLoop 結(jié)構(gòu)體如下所示：
/* State of an event based program 事件驅(qū)動(dòng)程序的狀態(tài) */
typedefstruct aeEventLoop {
int maxfd; /* highest file descriptor currently registered. 當(dāng)前已注冊(cè)的最高文件描述符 */
int setsize; /* max number of file descriptors tracked. [events/fired數(shù)組的大小] */
longlong timeEventNextId; /* 時(shí)間事件的下一個(gè)ID */
/* events/fired 都是數(shù)組 */
/* events 數(shù)組，下標(biāo)含義：為某個(gè)fd。fd=>aeFileEvent，即 文件描述符=>文件事件 */
/* fired 為 io多路復(fù)用返回的數(shù)組，每一個(gè)值為就緒的fd */
/* 通過 fired 中的 fd 去 events 查找對(duì)應(yīng)的事件信息(事件信息包含conn) */
aeFileEvent *events; /* Registered events 已注冊(cè)事件，數(shù)組 */
aeFiredEvent *fired; /* Fired events 觸發(fā)的事件，數(shù)組 */
aeTimeEvent *timeEventHead; /* 時(shí)間事件，鏈表 */
int stop; /* 停止事件循環(huán) */
void *apidata; /* This is used for polling API specific data. 這用于獲取特定的API數(shù)據(jù)，aeApiState *state 包含io多路復(fù)用fd等字段 */
aeBeforeSleepProc *beforesleep;
aeBeforeSleepProc *aftersleep;
int flags;
} aeEventLoop;aeEventLoop 結(jié)構(gòu)體核心字段以及相關(guān)交互如下圖所示：
- setsize 文件事件數(shù)組大小，等于 server.maxclients+CONFIG_FDSET_INCR
- events 文件事件數(shù)組，大小等于setsize
- fired 文件事件就緒的fd數(shù)組，大小等于setsize
- timeEventHead 時(shí)間事件數(shù)組，雙向鏈表
- apidata 這用于獲取特定的API數(shù)據(jù)，指向 aeApiState結(jié)構(gòu)體，不同的i/o多路復(fù)用實(shí)現(xiàn)包含不同的字段。
// ae_epoll.c
typedefstruct aeApiState {/* 在 aeApiCreate 中初始化，linux則在 ae_linux.c 文件 */
int epfd; /* io多路復(fù)用fd */
struct epoll_event *events;/* 就緒的事件數(shù)組 */
} aeApiState;
// ae_kqueue.c
typedefstruct aeApiState {
int kqfd;
struct kevent *events;
/* Events mask for merge read and write event.
* To reduce memory consumption, we use 2 bits to store the mask
* of an event, so that 1 byte will store the mask of 4 events. */
char *eventsMask;
} aeApiState;
// ae_evport.c
typedefstruct aeApiState {
int portfd; /* event port */
uint_t npending; /* # of pending fds */
int pending_fds[MAX_EVENT_BATCHSZ]; /* pending fds */
int pending_masks[MAX_EVENT_BATCHSZ]; /* pending fds' masks */
} aeApiState;
// ae_select.c
typedefstruct aeApiState {
fd_set rfds, wfds;
/* We need to have a copy of the fd sets as it's not safe to reuse
* FD sets after select(). */
fd_set _rfds, _wfds;
} aeApiState;
aeEventLoop 相關(guān)操作方法簽名如下所示(文件ae.h)：
aeEventLoop *aeCreateEventLoop(int setsize);
void aeDeleteEventLoop(aeEventLoop *eventLoop);
void aeStop(aeEventLoop *eventLoop);
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask, aeFileProc *proc, void *clientData);
void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask);
int aeGetFileEvents(aeEventLoop *eventLoop, int fd);
void *aeGetFileClientData(aeEventLoop *eventLoop, int fd);
long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
aeTimeProc *proc, void *clientData, aeEventFinalizerProc *finalizerProc);
int aeDeleteTimeEvent(aeEventLoop *eventLoop, long long id);
int aeProcessEvents(aeEventLoop *eventLoop, int flags);
int aeWait(int fd, int mask, long long milliseconds);
void aeMain(aeEventLoop *eventLoop);
char *aeGetApiName(void);
void aeSetBeforeSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *beforesleep);
void aeSetAfterSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *aftersleep);
int aeGetSetSize(aeEventLoop *eventLoop);
int aeResizeSetSize(aeEventLoop *eventLoop, int setsize);
void aeSetDontWait(aeEventLoop *eventLoop, int noWait);aeEventLoop事件處理核心方法 | 用途 | 調(diào)用i/o多路復(fù)用方法 | epoll為例,調(diào)用方法 |
aeCreateEventLoop | 創(chuàng)建并初始化事件循環(huán) | aeApiCreate | epoll_create() 默認(rèn)水平觸發(fā) |
aeDeleteEventLoop | 刪除事件循環(huán) | aeApiFree | - |
aeCreateFileEvent | 創(chuàng)建文件事件 | aeApiAddEvent | epoll_ctl() EPOLL_CTL_ADD EPOLL_CTL_MOD |
aeDeleteFileEvent | 刪除文件事件 | aeApiDelEvent | epoll_ctl() EPOLL_CTL_MOD EPOLL_CTL_DEL |
aeProcessEvents | 處理文件事件 | aeApiPoll | epoll_wait() |
aeGetApiName | 獲取i/o多路復(fù)用的實(shí)現(xiàn)名稱 | aeApiName | - |
基于epoll的i/o多路復(fù)用
客戶端與服務(wù)端的連接建立過程，如下圖所示：
TCP三次握手時(shí)，Linux內(nèi)核會(huì)維護(hù)兩個(gè)隊(duì)列：
- 半連接隊(duì)列，被稱為SYN隊(duì)列
- 全連接隊(duì)列，被稱為 accept隊(duì)列
epoll相關(guān)處理方法與邏輯如下圖所示：
基于epoll的i/o多路復(fù)用偽代碼框架：
int main(){
lfd = socket(AF_INET,SOCK_STREAM,0); // 創(chuàng)建socket
bind(lfd, ...); // 綁定IP地址與端口
listen(lfd, ...); // 監(jiān)聽
// 創(chuàng)建epoll對(duì)象
efd = epoll_create(...);
// 把 listen socket 的事件管理起來
epoll_ctl(efd, EPOLL_CTL_ADD, lfd, ...);
//事件循環(huán)
for (;;) {
size_t nready = epoll_wait(efd, ep, ...);
for (int i = 0; i < nready; ++i){
if(ep[i].data.fd == lfd){
fd = accept(listenfd, ...); //lfd上發(fā)生事件表示都連接到達(dá)，accept接收它
epoll_ctl(efd, EPOLL_CTL_ADD, fd, ...);
}else{
//其它socket發(fā)生的事件都是讀寫請(qǐng)求、或者關(guān)閉連接
...
}
}
}
}
從上可知，Redis作為Server服務(wù)端在啟動(dòng)之后隨時(shí)隨刻監(jiān)聽著相關(guān)事件的發(fā)生。以linux為例，其處理過程與基于epoll的i/o多路復(fù)用偽代碼框架基本相似，Redis源碼中更多的是通過封裝使其得到一個(gè)方便使用的庫，庫的底層包含了多種i/o多路復(fù)用實(shí)現(xiàn)方式。
aeEventLoop 的執(zhí)行過程
以epoll為例，簡(jiǎn)化版的Redis事件驅(qū)動(dòng)交互過程。
圖中僅列出了核心方法，如有錯(cuò)誤歡迎指正
Red括： 針對(duì)不同的 fd 注冊(cè) AE_READABLE/AE_WRITABLE 類型的回調(diào)方法，同時(shí)把 fd 添加到 epoll 中。當(dāng) fd 關(guān)心的事件觸發(fā)之后，執(zhí)行對(duì)應(yīng)回調(diào)方法（主要針對(duì) 可讀/可寫/時(shí)間事件 3種類型的事件進(jìn)行處理）。Redis 中 epoll 使用的觸發(fā)方式為 LT 水平觸發(fā)，意味著數(shù)據(jù)一次性沒有處理完，下次 epoll_wait() 方法還會(huì)返回對(duì)應(yīng)fd，直到處理完畢，對(duì)于客戶端一次性發(fā)起批量處理多條命令的操作非常有益，減少對(duì)其他指令的阻塞時(shí)間。
