Redis源码探究 事件驱动网络编程 Server
Redis源码探究-事件驱动网络编程-Server
本文使用的是 github 上 Redis 最早的源代码,Redis 1.3.6,发布于 2010 年。
Redis 使用了事件驱动网络编程,Reactor 模式,其核心是:注册事件,提供回调,非阻塞 IO。
EventLoop
事件驱动的核心是 EventLoop 结构,它代表了一个 Event Loop,也就是说,使用者向这个 EventLoop 注册事件,并提供回调函数,EventLoop 就不停地 “Loop” 着等待事件 Event 发生,然后调用回调函数。
/* State of an event based program */
typedef struct aeEventLoop {
int maxfd;
long long timeEventNextId;
aeFileEvent events[AE_SETSIZE]; /* Registered events */
aeFiredEvent fired[AE_SETSIZE]; /* Fired events */
aeTimeEvent *timeEventHead;
int stop;
void *apidata; /* This is used for polling API specific data */
aeBeforeSleepProc *beforesleep;
} aeEventLoop;
事件
抽象上来看,一个事件包含了三个部分,事件触发条件、事件处理方法以及存放事件处理结果的位置。
具体实现上来看,Redis 将事件分为了两类,FileEvent 和 TimeEvent:
-
FileEvent:
/* File event structure */ typedef struct aeFileEvent { int mask; /* one of AE_(READABLE|WRITABLE) */ aeFileProc *rfileProc; aeFileProc *wfileProc; void *clientData; } aeFileEvent;
mask:代表事件的触发条件,分为 READABLE 或 WRITABLE。
rfileProc/wfileProc:代表事件的处理方法,READABLE 的事件则将处理的函数存在 rfileProc,WRITABLE 的函数则将处理的函数存在 wfileProc。
clientData:处理结果存放的内存空间。
-
TimeEvent:
/* Time event structure */ typedef struct aeTimeEvent { long long id; /* time event identifier. */ long when_sec; /* seconds */ long when_ms; /* milliseconds */ aeTimeProc *timeProc; aeEventFinalizerProc *finalizerProc; void *clientData; struct aeTimeEvent *next; } aeTimeEvent;
timeProc:计时到后调用的处理函数。
clientData:处理结果存放的内存空间。
next:所有的 TimeEvent 形成一个链表。
事件触发
Redis 的事件触发机制是,先调用系统的 fd 监听 API,监听注册的事件 fd,当事件触发后,将触发事件统一存放到一处(fired 数组),然后集中地一个一个处理触发的事件。
Redis 支持多种 fd 监听 API,比如 select 和 epoll,而每个 API 对应的数据结构不同,所以 Redis 使用一个 void 指针来指向这些数据结构。
aeEventLoop
aeEventLoop 结构包含了一个 Event Loop 所需要的信息。
-
maxfd:所有事件都是由一个文件描述符 fd 表示的,在注册事件时,需要提供该事件的 fd,aeEventLoop.maxfd 代表的是所有 fd 中的最大值。
-
events:存放所有注册的 FileEvent 的信息。
-
fired:存放已发生的事件。
/* A fired event */ typedef struct aeFiredEvent { int fd; int mask; } aeFiredEvent;
fd:已发生事件的 fd。
mask:事件的触发条件,即 READABLE 或 WRITABLE。
-
timeEventHead:存放所有的 TimeEvent 的链表。
-
stop:EventLoop 运行状态标志位。
-
apidata:fd 监听 API 各自的具体数据。
Server 实际启动运行
main()
从 Server 的 main() 函数开始,只关注网络方面的代码。
在去掉一些错误处理和日志代码后,main() 函数非常简洁:
static struct redisServer server; /* server global state */
int main(int argc, char **argv) {
initServerConfig();//初始化 Server 默认配置
initServer();//初始化 Server
aeMain(server.el);//执行 EventLoop
aeDeleteEventLoop(server.el);//结束 EventLoop
return 0;
}
redisServer 结构
redisServer 是存放 Server 所有信息的结构体,从 redisServer 中网络相关的变量如下:
struct redisServer {
int port;
int fd;
char neterr[ANET_ERR_LEN];
aeEventLoop *el;
char *bindaddr;
}
initServer()
同样只提取出网络相关代码:
server.el = aeCreateEventLoop();
server.fd = anetTcpServer(server.neterr, server.port, server.bindaddr); //start listening on this fd.
aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);
aeCreateFileEvent(server.el, server.fd, AE_READABLE, acceptHandler, NULL);
-
aeCreateEventLoop()
创建 EventLoop:zai
/* Include the best multiplexing layer supported by this system. * The following should be ordered by performances, descending. */ #ifdef HAVE_EPOLL //优先选择 epoll,其次 kqueue,最次 select。 #include "ae_epoll.c" #else #ifdef HAVE_KQUEUE #include "ae_kqueue.c" #else #include "ae_select.c" #endif #endif aeEventLoop *aeCreateEventLoop(void) { aeEventLoop *eventLoop; int i; eventLoop = zmalloc(sizeof(*eventLoop)); if (!eventLoop) return NULL; eventLoop->timeEventHead = NULL; eventLoop->timeEventNextId = 0; eventLoop->stop = 0; eventLoop->maxfd = -1; eventLoop->beforesleep = NULL; if (aeApiCreate(eventLoop) == -1) { zfree(eventLoop); return NULL; } /* Events with mask == AE_NONE are not set. So let's initialize the * vector with it. */ for (i = 0; i < AE_SETSIZE; i++) eventLoop->events[i].mask = AE_NONE; return eventLoop; }
首先 malloc 一个 aeEventLoop 结构,然后把变量都设为初始值。
重点在于 aeApiCreate(eventLoop),该函数有三个实现,分别使用 epoll、kqueue 以及 select 实现,redis 使用宏来控制选择哪个版本,我主要关注性能最好的 epoll 实现:
#include <sys/epoll.h> typedef struct aeApiState { int epfd; struct epoll_event events[AE_SETSIZE]; } aeApiState; static int aeApiCreate(aeEventLoop *eventLoop) { aeApiState *state = zmalloc(sizeof(aeApiState)); if (!state) return -1; state->epfd = epoll_create(1024); /* 1024 is just an hint for the kernel */ if (state->epfd == -1) return -1; eventLoop->apidata = state; return 0; }
可以看到 eventLoop->apidata 中存放的是 epoll_create() 后生成的 epoll fd 以及 epoll_event。
-
anetTcpServer()
根据 server.neterr、server.port 和 server.bindaddr 这三个设置,建立 socket,绑定地址,并开始监听。
其中 port 和 bindaddr 在 initServerConfig() 中进行初始化赋值:
#define REDIS_SERVERPORT 6379 /* TCP port */ server.port = REDIS_SERVERPORT; server.saveparams = NULL;
值得注意的是,代码中也像我之前的文章中写到过的一样,设置了 SO_REUSEADDR 属性,使得一个地址即使在 time_wait 状态,也可以被新的 socket 绑定。
if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) == -1) { anetSetError(err, "setsockopt SO_REUSEADDR: %s\n", strerror(errno)); close(s); return ANET_ERR; }
-
aeCreateFileEvent()
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask, aeFileProc *proc, void *clientData);
这个函数首先要从逻辑上理解,它的功能是创建一个 FileEvent,需要提供的参数是:
- eventloop,创建的 FileEvent 要加入这个 eventloop;
- fd,这个 FileEvent 要监听这个 fd 上的事件;
- mask,这个 FileEvent 触发的条件,READABLE 或 WRITABLE;
- proc,这个 FileEvent 触发后的处理方法;
- clientData,这个 FileEvent 处理后数据存放的位置。
我们再来看实际的代码实现:
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask, aeFileProc *proc, void *clientData) { if (fd >= AE_SETSIZE) return AE_ERR; aeFileEvent *fe = &eventLoop->events[fd]; if (aeApiAddEvent(eventLoop, fd, mask) == -1)//将 fd 加入 eventloop 监听 return AE_ERR; fe->mask |= mask;//事件触发条件存入 aeFileEvent if (mask & AE_READABLE) fe->rfileProc = proc;//事件处理方法存入 aeFileEvent if (mask & AE_WRITABLE) fe->wfileProc = proc;//事件处理方法存入 aeFileEvent fe->clientData = clientData;//事件处理结果存入 aeFileEvent if (fd > eventLoop->maxfd) eventLoop->maxfd = fd; return AE_OK; } //以下为epoll 版本,还有 kqueue 和 select 版本。 static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) { aeApiState *state = eventLoop->apidata; struct epoll_event ee; /* If the fd was already monitored for some event, we need a MOD * operation. Otherwise we need an ADD operation. */ int op = eventLoop->events[fd].mask == AE_NONE ? EPOLL_CTL_ADD : EPOLL_CTL_MOD; ee.events = 0; mask |= eventLoop->events[fd].mask; /* Merge old events */ if (mask & AE_READABLE) ee.events |= EPOLLIN;//READABLE转换为epoll特有的EPOLLIN if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;//WRITABLE转换为epoll特有的EPOLLOUT ee.data.u64 = 0; /* avoid valgrind warning */ ee.data.fd = fd;//需要监听触发的 fd if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1; return 0; }
看到这里,我们可以总结出,Redis 1.3.6 事件触发和事件处理两项上,是割裂的:
- 事件触发由 eventloop->apidata 处理;
- 事件处理由 eventloop->events[] 处理;
- 它们之间的接触点则是事件的 fd;
- 而已触发的事件的 fd,则由 api 函数保存到 fired[] 中;
- 获取到已触发的事件的 fd 后,以此 fd 为 key,可以从 events[fd] 中取出事件的处理方法等信息,并进行处理。
这样的割裂,我想是为了支持多种事件触发 API 而设计的。
aeCreateFileEvent(server.el, server.fd, AE_READABLE, acceptHandler, NULL);
Redis 实际上是把刚刚创建的服务器监听 socket 加入了 eventloop,并指定触发条件为 READABLE 可读时,事件的处理方法则是 acceptHandler,在事件触发后进行事件处理时,就会调用 acceptHandler。
-
aeCreateTimeEvent()
long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds, aeTimeProc *proc, void *clientData, aeEventFinalizerProc *finalizerProc);
创建 TimeEvent 和 FileEvent 类似,但是其触发条件是时间,由 milliseconds 参数指定,也不需要指定 fd 和 mask 了。
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; te->next = eventLoop->timeEventHead; eventLoop->timeEventHead = te; return id; }
TimeEvent 的创建不需要调用 eventloop->apidata,而只是把要触发的时间保存在 aeTimeEvent 结构里了,这是可以理解的,因为 epoll 或 select 等 API 并没有提供以时间触发的条件,但是提供了时间限制,我们可以在之后看到 Redis 是如何实现的。
aeMain(server.el)
初始化完成后,就开始正式运行 eventloop 了。
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
while (!eventLoop->stop) {
if (eventLoop->beforesleep != NULL)
eventLoop->beforesleep(eventLoop);
aeProcessEvents(eventLoop, AE_ALL_EVENTS);
}
}
当 stop 标志位为 0 时,维持运行,进入 aeProcessEvents()。
#define AE_FILE_EVENTS 1
#define AE_TIME_EVENTS 2
#define AE_ALL_EVENTS (AE_FILE_EVENTS|AE_TIME_EVENTS)
#define AE_DONT_WAIT 4
int aeProcessEvents(aeEventLoop *eventLoop, int flags);
抽象地看,aeProcessEvents() 函数的功能是”执行指定 eventloop 中指定类型的 event”,flags 参数即用来指定类型,包括指定运行 FileEvent(AE_FILE_EVENTS),运行 TimeEvent(AE_TIME_EVENTS) 以及两种都运行(AE_ALL_EVENTS),AE_DONT_WAIT 则是目前有什么事件执行什么事件并尽快返回。
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
int processed = 0, numevents;
/* Nothing to do? return ASAP */
if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
/* Note that we want call select() even if there are no
* file events to process as long as we want to process time
* events, in order to sleep until the next time event is ready
* to fire. */
if (eventLoop->maxfd != -1 ||
((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
int j;
aeTimeEvent *shortest = NULL;
struct timeval tv, *tvp;
if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
shortest = aeSearchNearestTimer(eventLoop);
if (shortest) {
long now_sec, now_ms;
/* Calculate the time missing for the nearest
* timer to fire. */
aeGetTime(&now_sec, &now_ms);
tvp = &tv;
tvp->tv_sec = shortest->when_sec - now_sec;
if (shortest->when_ms < now_ms) {
tvp->tv_usec = ((shortest->when_ms+1000) - now_ms)*1000;
tvp->tv_sec --;
} else {
tvp->tv_usec = (shortest->when_ms - now_ms)*1000;
}
if (tvp->tv_sec < 0) tvp->tv_sec = 0;
if (tvp->tv_usec < 0) tvp->tv_usec = 0;
} else {
/* If we have to check for events but need to return
* ASAP because of AE_DONT_WAIT we need to se the timeout
* to zero */
if (flags & AE_DONT_WAIT) {
tv.tv_sec = tv.tv_usec = 0;
tvp = &tv;
} else {
/* Otherwise we can block */
tvp = NULL; /* wait forever */
}
}
......
从 TimeEvent 链表中找出触发时间最近的一个,并计算出距离此刻的时间,保存在 tvp 变量中,若没有 TimeEvent,则 tvp 置为 0。
......
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;
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++;
}
}
/* Check time events */
if (flags & AE_TIME_EVENTS)
processed += processTimeEvents(eventLoop);
return processed; /* return the number of processed file/time events */
}
aeApiPoll() 的作用是调用事件监听 API,等待事件触发返回:
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
aeApiState *state = eventLoop->apidata;
int retval, numevents = 0;
retval = epoll_wait(state->epfd,state->events,AE_SETSIZE,
tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);
if (retval > 0) {
int j;
numevents = retval;
for (j = 0; j < numevents; j++) {
int mask = 0;
struct epoll_event *e = state->events+j;
if (e->events & EPOLLIN) mask |= AE_READABLE;
if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
eventLoop->fired[j].fd = e->data.fd;
eventLoop->fired[j].mask = mask;
}
}
return numevents;
}
等待 epoll 的返回,返回的 epoll_event 结构都存放在 apidata->events。
然后根据这些 epoll_event,将触发的 FileEvent 从第 0 个开始逐个保存在 eventloop->fired[] 中,保存的信息包括 fd 和 触发条件。
由 aeApiPoll(eventLoop, tvp) 继续往下看。
这里的重点是,监听等待超时的时间设置为了 TimeEvent 的事件 tvp,也就是说,Redis 通过这种方式,实现了一个时间触发的事件,同时一并处理 FileEvent 事件,即:
- 事件监听函数的超时时间设置为最近的 TimeEvent 事件的触发时间;
- 函数返回后,先处理在这一段时间内触发的 FileEvent,这些已触发的事件的信息保存在 fired[] 中,从中取出 fd 和触发条件,以此 fd 为 key,从 events[fd] 中取出事件的处理函数,并调用进行处理。
- FileEvent 处理完后,调用 processTimeEvents(eventLoop) 处理 TimeEvent。
/* Process time events */
static int processTimeEvents(aeEventLoop *eventLoop) {
int processed = 0;
aeTimeEvent *te;
long long maxId;
te = eventLoop->timeEventHead;
maxId = eventLoop->timeEventNextId-1;
while(te) {
long now_sec, now_ms;
long long id;
if (te->id > maxId) {
te = te->next;
continue;
}
aeGetTime(&now_sec, &now_ms);
if (now_sec > te->when_sec ||
(now_sec == te->when_sec && now_ms >= te->when_ms))
{
int retval;
id = te->id;
retval = te->timeProc(eventLoop, id, te->clientData);
processed++;
/* After an event is processed our time event list may
* no longer be the same, so we restart from head.
* Still we make sure to don't process events registered
* by event handlers itself in order to don't loop forever.
* To do so we saved the max ID we want to handle.
*
* FUTURE OPTIMIZATIONS:
* Note that this is NOT great algorithmically. Redis uses
* a single time event so it's not a problem but the right
* way to do this is to add the new elements on head, and
* to flag deleted elements in a special way for later
* deletion (putting references to the nodes to delete into
* another linked list). */
if (retval != AE_NOMORE) {
aeAddMillisecondsToNow(retval,&te->when_sec,&te->when_ms);
} else {
aeDeleteTimeEvent(eventLoop, id);
}
te = eventLoop->timeEventHead;
} else {
te = te->next;
}
}
return processed;
}
可以看到, Redis 遍历了整个 TimeEvent 链表,获取当前的时间,与这些 TimeEvent 的触发时间作对比,所有已经超过触发时间的 TimeEvent 统统处理,调用它们的 timeProc 处理函数。
acceptHandler
到此我们已经了解到 Redis 怎么创建事件、等待事件触发并处理事件的了。
aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);
aeCreateFileEvent(server.el, server.fd, AE_READABLE, acceptHandler, NULL);
我们注意到,在初始化过程中 Redis 仅创建了两个事件,一个 FileEvent,一个 TimeEvent。
这个 TimeEvent 我们可以看出是每 1 毫秒执行一次 serverCron,应该是一些业务相关的东西,在此按下不表。
而这个 FileEvent 监听的是服务器端口的可读事件,是接收客户端连接并处理的地方:
static void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
int cport, cfd;
char cip[128];
redisClient *c;
REDIS_NOTUSED(el);
REDIS_NOTUSED(mask);
REDIS_NOTUSED(privdata);
cfd = anetAccept(server.neterr, fd, cip, &cport);
if (cfd == AE_ERR) {
redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr);
return;
}
redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
if ((c = createClient(cfd)) == NULL) {
redisLog(REDIS_WARNING,"Error allocating resoures for the client");
close(cfd); /* May be already closed, just ingore errors */
return;
}
......
}
首先 accept 新的连接,返回新连接的 socket fd。
static redisClient *createClient(int fd) {
redisClient *c = zmalloc(sizeof(*c));
anetNonBlock(NULL,fd);
anetTcpNoDelay(NULL,fd);
if (!c) return NULL;
selectDb(c,0);
c->fd = fd;
c->querybuf = sdsempty();
c->argc = 0;
c->argv = NULL;
c->bulklen = -1;
c->multibulk = 0;
c->mbargc = 0;
c->mbargv = NULL;
c->sentlen = 0;
c->flags = 0;
c->lastinteraction = time(NULL);
c->authenticated = 0;
c->replstate = REDIS_REPL_NONE;
c->reply = listCreate();
listSetFreeMethod(c->reply,decrRefCount);
listSetDupMethod(c->reply,dupClientReplyValue);
c->blockingkeys = NULL;
c->blockingkeysnum = 0;
c->io_keys = listCreate();
listSetFreeMethod(c->io_keys,decrRefCount);
if (aeCreateFileEvent(server.el, c->fd, AE_READABLE,
readQueryFromClient, c) == AE_ERR) {
freeClient(c);
return NULL;
}
listAddNodeTail(server.clients,c);
initClientMultiState(c);
return c;
}
使用新建的客户端连接创建 redisClient 结构,这里大多是 Client 相关的东西,留待以后研究,重点关注 Server 网络编程相关的事情:
- Redis 将新连接设置为了非阻塞 IO。
- Redis 将新连接的 socket 属性设置为了 TcpNoDelay,如我之前的博客写过,不这样设置的话,因为 Nagle 算法的缘故,会产生不必要的延迟。
- 调用 aeCreateFileEvent() 将新连接加入 EventLoop 进行监听,触发条件为 READABLE,处理方法为 readQueryFromClient(),处理结果存放位置为这个 redisClient 结构中。
#define REDIS_IOBUF_LEN 1024
static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
redisClient *c = (redisClient*) privdata;
char buf[REDIS_IOBUF_LEN];
int nread;
REDIS_NOTUSED(el);
REDIS_NOTUSED(mask);
nread = read(fd, buf, REDIS_IOBUF_LEN);
if (nread == -1) {
if (errno == EAGAIN) {
nread = 0;
} else {
redisLog(REDIS_VERBOSE, "Reading from client: %s",strerror(errno));
freeClient(c);
return;
}
} else if (nread == 0) {
redisLog(REDIS_VERBOSE, "Client closed connection");
freeClient(c);
return;
}
if (nread) {
c->querybuf = sdscatlen(c->querybuf, buf, nread);
c->lastinteraction = time(NULL);
} else {
return;
}
if (!(c->flags & REDIS_BLOCKED))
processInputBuffer(c);
}
当客户端的可读事件触发后,从客户端读取最多 1024 个字节,保存到 redisClient->querybuf 中,最后调用 processInputBuffer() 处理客户端的查询请求。
尾声
至此,Redis Server 的初始化流程就研究完了,果然写得很好啊,利用 C 完成了事件驱动的编程,并做了一层抽象,以适应不同的监听 API,学习了!
参考文章:https://redis.io/topics/internals-rediseventlib