redis7.x源码分析:(4) ae事件处理器（一）
ae模块是redis实现的Reactor模型的封装。它的主要代码实现集中在 ae.c 中，另外还提供了平台相关的io多路复用的封装，它们都实现了一套相同的poll接口，就类似于C++中提供了一个接口基类，由针对不同平台的派生类去实现。
// 创建平台相关的io模型实例 static int aeApiCreate(aeEventLoop *eventLoop) // 修改可侦听的fd数量 static int aeApiResize(aeEventLoop *eventLoop, int setsize) // 释放实例 static void aeApiFree(aeEventLoop *eventLoop) // 添加或修改fd的侦听事件 static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) // 删除或修改fd的侦听事件 static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask) // 侦听所有fd的事件 static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) // 返回使用的io模型名称 static char *aeApiName(void)
实际使用的io模型会根据编译时定义的宏在 ae.c 的代码头部直接引入。
/* Include the best multiplexing layer supported by this system.
* The following should be ordered by performances, descending. */
// 根据不同操作系统下不同的宏选择应该使用的io复用模型
#ifdef HAVE_EVPORT
#include "ae_evport.c"
#else
#ifdef HAVE_EPOLL
// 对应linux
#include "ae_epoll.c"
#else
#ifdef HAVE_KQUEUE
// 对应BSD（FreeBSD、MacOS等）
#include "ae_kqueue.c"
#else
// 对应Solaris
#include "ae_select.c"
#endif
#endif
#endif
以linux为例先来看一下ae_epoll.c的实现，它的代码也比较简单：
static int aeApiCreate(aeEventLoop *eventLoop) {
aeApiState *state = zmalloc(sizeof(aeApiState));
if (!state) return -1;
// 分配events数组大小
state->events = zmalloc(sizeof(struct epoll_event)*eventLoop->setsize);
if (!state->events) {
zfree(state);
return -1;
}
// 创建epoll, kernal 2.6.8 之后参数被忽略了
state->epfd = epoll_create(1024); /* 1024 is just a hint for the kernel */
if (state->epfd == -1) {
zfree(state->events);
zfree(state);
return -1;
}
anetCloexec(state->epfd);
eventLoop->apidata = state;
return 0;
}
......
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
aeApiState *state = eventLoop->apidata;
struct epoll_event ee = {0}; /* avoid valgrind warning */
/* If the fd was already monitored for some event, we need a MOD
* operation. Otherwise we need an ADD operation. */
// 没注册过, 认为是添加fd到epoll
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;
if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
ee.data.fd = fd;
if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
return 0;
}
static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask) {
aeApiState *state = eventLoop->apidata;
struct epoll_event ee = {0}; /* avoid valgrind warning */
// 删除指定事件
int mask = eventLoop->events[fd].mask & (~delmask);
ee.events = 0;
// 设置需要的事件
if (mask & AE_READABLE) ee.events |= EPOLLIN;
if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
ee.data.fd = fd;
// 没有事件执行删除，否则执行修改
if (mask != AE_NONE) {
epoll_ctl(state->epfd,EPOLL_CTL_MOD,fd,&ee);
} else {
/* Note, Kernel < 2.6.9 requires a non null event pointer even for
* EPOLL_CTL_DEL. */
epoll_ctl(state->epfd,EPOLL_CTL_DEL,fd,&ee);
}
}
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
aeApiState *state = eventLoop->apidata;
int retval, numevents = 0;
// 此处的 (tvp->tv_usec + 999)/1000 是为了避免当 tvp->tv_usec < 1000us 时转换成毫秒为 0ms, 导致epoll_wait不会等待
// 因此这么写之后能保证只要 tvp->tv_usec 不为0，那么至少都能等待 1ms 时长
retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,
tvp ? (tvp->tv_sec*1000 + (tvp->tv_usec + 999)/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;
if (e->events & EPOLLERR) mask |= AE_WRITABLE|AE_READABLE;
if (e->events & EPOLLHUP) mask |= AE_WRITABLE|AE_READABLE;
// 保存触发的事件
eventLoop->fired[j].fd = e->data.fd;
eventLoop->fired[j].mask = mask;
}
} else if (retval == -1 && errno != EINTR) {
panic("aeApiPoll: epoll_wait, %s", strerror(errno));
}
return numevents;
}
接下来再看一下ae相关接口，先看下fd相关实现。
主要结构体定义如下：
/* File event structure */
typedef struct aeFileEvent {
int mask; /* one of AE_(READABLE|WRITABLE|BARRIER) */ // 事件类型
aeFileProc *rfileProc; // 读事件回调
aeFileProc *wfileProc; // 写事件回调
void *clientData; // 上下文指针
} aeFileEvent;
/* Time event structure */
typedef struct aeTimeEvent {
long long id; /* time event identifier. */ // 定时器事件id
monotime when; // 触发时间
aeTimeProc *timeProc; // 定时器触发回调
aeEventFinalizerProc *finalizerProc; // 定时器销毁时执行的回调
void *clientData; // 上下文指针
struct aeTimeEvent *prev; // 前一个定时器事件
struct aeTimeEvent *next; // 后一个定时器事件
int refcount; /* refcount to prevent timer events from being
* freed in recursive time event calls. */ // 引用计数
} aeTimeEvent;
/* A fired event */
typedef struct aeFiredEvent {
int fd; // 触发事件的定时器
int mask; // 触发的事件类型
} aeFiredEvent;
/* State of an event based program */
typedef struct aeEventLoop {
int maxfd; /* highest file descriptor currently registered */ // 注册的最大fd值
int setsize; /* max number of file descriptors tracked */ // 注册的fd数量
long long timeEventNextId; // 自增用于生成定时器id
aeFileEvent *events; /* Registered events */ // 注册的fd事件
aeFiredEvent *fired; /* Fired events */ // 触发的fd事件
aeTimeEvent *timeEventHead; // 定时器事件链表头
int stop; // 结束标志
void *apidata; /* This is used for polling API specific data */ // 不同poll的上下文信息
aeBeforeSleepProc *beforesleep; // poll前调用
aeBeforeSleepProc *aftersleep; // poll后调用
int flags;
} aeEventLoop;
aeEventLoop中的events是用于保存fd注册事件的数组，它是以fd值作为索引来存取事件的。
创建eventloop事件管理器：
aeEventLoop *aeCreateEventLoop(int setsize) {
aeEventLoop *eventLoop;
int i;
// 初始化一下单调时钟使用哪种方式实现
monotonicInit(); /* just in case the calling app didn't initialize */
if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) goto err;
// events、fired数组以fd值为下标，直接存取对应fd的相关数据
eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);
eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);
if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err;
eventLoop->setsize = setsize;
eventLoop->timeEventHead = NULL;
eventLoop->timeEventNextId = 0;
eventLoop->stop = 0;
eventLoop->maxfd = -1;
eventLoop->beforesleep = NULL;
eventLoop->aftersleep = NULL;
eventLoop->flags = 0;
// 根据平台创建io模型实例
if (aeApiCreate(eventLoop) == -1) goto err;
/* Events with mask == AE_NONE are not set. So let's initialize the
* vector with it. */
for (i = 0; i < setsize; i++)
eventLoop->events[i].mask = AE_NONE;
return eventLoop;
err:
if (eventLoop) {
zfree(eventLoop->events);
zfree(eventLoop->fired);
zfree(eventLoop);
}
return NULL;
}
添加和删除事件：
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
aeFileProc *proc, void *clientData)
{
// 超出events数组有效范围直接报错
if (fd >= eventLoop->setsize) {
errno = ERANGE;
return AE_ERR;
}
aeFileEvent *fe = &eventLoop->events[fd];
// 添加或修改fd读写事件
if (aeApiAddEvent(eventLoop, fd, mask) == -1)
return AE_ERR;
fe->mask |= mask;
if (mask & AE_READABLE) fe->rfileProc = proc;
if (mask & AE_WRITABLE) fe->wfileProc = proc;
fe->clientData = clientData;
if (fd > eventLoop->maxfd)
eventLoop->maxfd = fd;
return AE_OK;
}
void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask)
{
if (fd >= eventLoop->setsize) return;
aeFileEvent *fe = &eventLoop->events[fd];
if (fe->mask == AE_NONE) return;
/* We want to always remove AE_BARRIER if set when AE_WRITABLE
* is removed. */
if (mask & AE_WRITABLE) mask |= AE_BARRIER;
aeApiDelEvent(eventLoop, fd, mask);
fe->mask = fe->mask & (~mask);
// 如果本次删除事件的fd是最大的fd，并且该fd已经没有事件了（可以认为该fd被清除了）
if (fd == eventLoop->maxfd && fe->mask == AE_NONE) {
/* Update the max fd */
int j;
// 从后往前遍历找到第一个有事件的fd赋给maxfd
for (j = eventLoop->maxfd-1; j >= 0; j--)
if (eventLoop->events[j].mask != AE_NONE) break;
eventLoop->maxfd = j;
}
}
事件处理函数，处理一轮事件：
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 to 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;
struct timeval tv, *tvp;
int64_t usUntilTimer = -1;
// 获取最近的一个定时器距当前时间的触发时长
if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
usUntilTimer = usUntilEarliestTimer(eventLoop);
if (usUntilTimer >= 0) {
tv.tv_sec = usUntilTimer / 1000000;
tv.tv_usec = usUntilTimer % 1000000;
tvp = &tv;
} else {
/* If we have to check for events but need to return
* ASAP because of AE_DONT_WAIT we need to set 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 */
}
}
// 不需要等待，超时时间设为0
if (eventLoop->flags & AE_DONT_WAIT) {
tv.tv_sec = tv.tv_usec = 0;
tvp = &tv;
}
// 执行poll前的回调(对fd读写以及命令处理都会在beforsleep中执行, 如果开启多线程读写也会在它内部一并处理)
if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP)
eventLoop->beforesleep(eventLoop);
/* Call the multiplexing API, will return only on timeout or when
* some event fires. */
// 等待事件
numevents = aeApiPoll(eventLoop, tvp);
/* After sleep callback. */
// 执行poll后的回调
if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)
eventLoop->aftersleep(eventLoop);
// 处理poll到的读写事件
for (j = 0; j < numevents; j++) {
int fd = eventLoop->fired[j].fd;
aeFileEvent *fe = &eventLoop->events[fd];
int mask = eventLoop->fired[j].mask;
int fired = 0; /* Number of events fired for current fd. */
/* Normally we execute the readable event first, and the writable
* event later. This is useful as sometimes we may be able
* to serve the reply of a query immediately after processing the
* query.
*
* However if AE_BARRIER is set in the mask, our application is
* asking us to do the reverse: never fire the writable event
* after the readable. In such a case, we invert the calls.
* This is useful when, for instance, we want to do things
* in the beforeSleep() hook, like fsyncing a file to disk,
* before replying to a client. */
// 是否反转读写事件的顺序（不设置 AE_BARRIER 时，先读后写）
int invert = fe->mask & AE_BARRIER;
/* 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.
*
* Fire the readable event if the call sequence is not
* inverted. */
if (!invert && fe->mask & mask & AE_READABLE) {
// 处理读事件回调
fe->rfileProc(eventLoop,fd,fe->clientData,mask);
fired++;
// 防止在事件处理函数中 resize 过数组大小，所以需要重新获取下地址
fe = &eventLoop->events[fd]; /* Refresh in case of resize. */
}
/* Fire the writable event. */
if (fe->mask & mask & AE_WRITABLE) {
// 处理写事件回调（读写回调函数相同时，每轮poll只执行一次读或者写回调）
if (!fired || fe->wfileProc != fe->rfileProc) {
fe->wfileProc(eventLoop,fd,fe->clientData,mask);
fired++;
}
}
/* If we have to invert the call, fire the readable event now
* after the writable one. */
if (invert) {
// 反转读写时先写，然后在这边后读（读写回调函数相同时，每轮poll只执行一次读或者写回调）
fe = &eventLoop->events[fd]; /* Refresh in case of resize. */
if ((fe->mask & mask & AE_READABLE) &&
(!fired || fe->wfileProc != fe->rfileProc))
{
fe->rfileProc(eventLoop,fd,fe->clientData,mask);
fired++;
}
}
processed++;
}
}
/* Check time events */
// 处理定时器事件
if (flags & AE_TIME_EVENTS)
processed += processTimeEvents(eventLoop);
return processed; /* return the number of processed file/time events */
}
执行主事件循环：
void aeMain(aeEventLoop *eventLoop) {
// 启动事件循环
eventLoop->stop = 0;
while (!eventLoop->stop) {
// 处理事件
aeProcessEvents(eventLoop, AE_ALL_EVENTS|
AE_CALL_BEFORE_SLEEP|
AE_CALL_AFTER_SLEEP);
}
}
#redis#
查看13道真题和解析