Redis 单线程事件循环 #1
Assignees
Labels
Comments
概览
有了这写概念，我们可以试着描绘一下
|
事件库 aeEventLoop 实现细节先来看核心数据结构： /* State of an event based program */
typedef struct aeEventLoop {
int maxfd; // 当前已经注册在此的最大文件描述符
int setsize; // 可“关心”的文件描述符数量
long long timeEventNextId; // 下一个 timer 的id
time_t lastTime; // 上一轮事件循环时的系统事件，用来诊断系统时间偏差
aeFileEvent *events; // 注册的文件事件
aeTimeEvent *timeEventHead; // 注册的时间事件
aeFiredEvent *fired; // 就绪的事件
int stop; // 事件轮询是否停止
void *apidata; /* This is used for polling API specific data */
aeBeforeSleepProc *beforesleep; // 下一次事件轮训之前的钩子函数
aeBeforeSleepProc *aftersleep; // 事件轮询结束后的钩子函数
} aeEventLoop;
/* File event structure */
typedef struct aeFileEvent {
int mask; /* one of AE_(READABLE|WRITABLE) */
aeFileProc *rfileProc; // 读事件就绪时的回调函数
aeFileProc *wfileProc; // 写事件就绪时的回调函数
void *clientData; // fd 对应的 client 实例
} aeFileEvent;
/* 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;
/* A fired event */
typedef struct aeFiredEvent {
int fd;
int mask;
} aeFiredEvent;
1.
|
Redis 单线程源码详解在这个 1. server 启动，创建 EventLoop在 void initServer(void)
{
...
// 创建全局唯一的 EventLoop 实例
server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
if (server.el == NULL) {
serverLog(LL_WARNING,
"Failed creating the event loop. Error message: '%s'",
strerror(errno));
exit(1);
}
...
/* Create an event handler for accepting new connections in TCP and Unix
* domain sockets. */
// ipfd 表示服务启动是监听的 socket 对应的 fd，epoll 监听此 fd，有读事件发生(新连接到来)时调用回调函数 acceptTcpHandler
for (j = 0; j < server.ipfd_count; j++) {
if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
acceptTcpHandler,NULL) == AE_ERR)
{
serverPanic(
"Unrecoverable error creating server.ipfd file event.");
}
}
}
....2. 新连接到来时创建连接以及 client 实例在前面我们将 server 对应的 socket 添加到 epoll 的监听队列，当有新的连接到来时，会触发读事件就绪，此时回调函数 void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
...
// 创建 connect fd，代表 Redis Server 和客户端的一个连接(socket)
cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK)
serverLog(LL_WARNING,
"Accepting client connection: %s", server.neterr);
return;
}
serverLog(LL_VERBOSE, "Accepted %s:%d", cip, cport);
acceptCommonHandler(cfd, 0, cip);
}
static void acceptCommonHandler(int fd, int flags, char *ip) {
client *c;
// 1. 为 connect fd 创建一个 Client 对象
if ((c = createClient(fd)) == NULL) {
serverLog(LL_WARNING,
"Error registering fd event for the new client: %s (fd=%d)",
strerror(errno), fd);
close(fd); /* May be already closed, just ignore errors */
return;
}
// 2. 检查是否超过了最大连接数
if (listLength(server.clients) > server.maxclients) {
char *err = "-ERR max number of clients reached\r\n";
/* That's a best effort error message, don't check write errors */
if (write(c->fd, err, strlen(err)) == -1) {
/* Nothing to do, Just to avoid the warning... */
}
server.stat_rejected_conn++;
freeClient(c);
return;
}
// 3. 检查 protect mode 是否开启，如果开启，不允许远程登录
if (server.protected_mode && server.bindaddr_count == 0 && server.requirepass == NULL && !(flags & CLIENT_UNIX_SOCKET) && ip != NULL) {
...
}
server.stat_numconnections++;
c->flags |= flags;
}
client *createClient(int fd) {
client *c = zmalloc(sizeof(client));
...
// 1. 标记 fd 为非阻塞
anetNonBlock(NULL, fd);
// 2. 设置不开启 Nagle 算法
anetEnableTcpNoDelay(NULL, fd);
// 3. 设置 KeepAlive
if (server.tcpkeepalive)
anetKeepAlive(NULL, fd, server.tcpkeepalive);
// 4. 为 fd 创建对应的文件事件监听对应 socket 的读事件，并指定对应事件发生之后的回调函数为 readQueryFromClient
if (aeCreateFileEvent(server.el, fd, AE_READABLE,
readQueryFromClient, c) == AE_ERR) {
close(fd);
zfree(c);
return NULL;
}
// 5. 默认使用 0 号 db
selectDb(c, 0);
uint64_t client_id;
// 6. 设置 client 其他默认属性
atomicGetIncr(server.next_client_id, client_id, 1);
c->id = client_id;
c->fd = fd;
...
return c;
}在这个方法中，主要做了以下几件事：
3. 客户端请求到来，执行具体的 handler在 void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
...
// 调用 read 从 socket 中读取客户端请求数据到 client->querybuf
c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);
nread = read(fd, c->querybuf+qblen, readlen);
...
// 如果 client->querybuf 的大小超过 client_max_querybuf_len，直接返回错误，并关闭连接
if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty();
bytes = sdscatrepr(bytes,c->querybuf,64);
serverLog(LL_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes);
sdsfree(ci);
sdsfree(bytes);
freeClient(c);
return;
}
// 处理客户端请求
processInputBufferAndReplicate(c);
}再来看 void processInputBufferAndReplicate(client *c) {
if (!(c->flags & CLIENT_MASTER)) {
processInputBuffer(c);
} else {
size_t prev_offset = c->reploff;
processInputBuffer(c);
size_t applied = c->reploff - prev_offset;
if (applied) {
replicationFeedSlavesFromMasterStream(server.slaves,
c->pending_querybuf, applied);
sdsrange(c->pending_querybuf,applied,-1);
}
}
}
void processInputBuffer(client *c) {
// 设置 server 的当前处理 client 为c，可以理解为获得了 server 这把锁
server.current_client = c;
// 不断从 querybuf 中取出数据解析成成对的命令，直到 querybuf 为空
while(c->qb_pos < sdslen(c->querybuf)) {
// 进行一些 flags 的判断
...
// 根据命令类型判断是 单条指令 还是 多条指令一起执行
if (c->reqtype == PROTO_REQ_INLINE) {
if (processInlineBuffer(c) != C_OK) break;
} else if (c->reqtype == PROTO_REQ_MULTIBULK) {
if (processMultibulkBuffer(c) != C_OK) break;
} else {
serverPanic("Unknown request type");
}
// 参数个数为 0 时重置客户端，可以接收下一个命令
if (c->argc == 0) {
resetClient(c);
} else {
// 执行命令
if (processCommand(c) == C_OK) {
// 集群信息同步
if (c->flags & CLIENT_MASTER && !(c->flags & CLIENT_MULTI)) {
/* Update the applied replication offset of our master. */
c->reploff = c->read_reploff - sdslen(c->querybuf) + c->qb_pos;
}
// 如果不是阻塞状态，则重置client，可以接受下一个命令
if (!(c->flags & CLIENT_BLOCKED) || c->btype != BLOCKED_MODULE)
resetClient(c);
}
// 释放“锁”
if (server.current_client == NULL) break;
}
}
// 重置 querybuf
if (c->qb_pos) {
sdsrange(c->querybuf,c->qb_pos,-1);
c->qb_pos = 0;
}
server.current_client = NULL;
}我们再来看 int processCommand(client *c) {
// 1. 如果命令是 quit，则直接退出
if (!strcasecmp(c->argv[0]->ptr, "quit")) {
addReply(c, shared.ok);
c->flags |= CLIENT_CLOSE_AFTER_REPLY;
return C_ERR;
}
// 2. 在 command table 寻找对应命令的处理函数，
c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);
...
// 3. 用户权限校验
if (server.requirepass && !c->authenticated && c->cmd->proc != authCommand) {
flagTransaction(c);
addReply(c, shared.noautherr);
return C_OK;
}
// 4. 如果是集群模式，还需要处理集群 node 重定向
if (server.cluster_enabled && !(c->flags & CLIENT_MASTER) && !(c->flags & CLIENT_LUA && server.lua_caller->flags & CLIENT_MASTER) &&
!(c->cmd->getkeys_proc == NULL && c->cmd->firstkey == 0 && c->cmd->proc != execCommand)) {
...
}
// 5. 处理 maxmemory 情形
if (server.maxmemory && !server.lua_timedout) {
...
}
// 6. 非 master 或者 磁盘有问题是，不要进行 AOF 等持久化操作
int deny_write_type = writeCommandsDeniedByDiskError();
if (deny_write_type != DISK_ERROR_TYPE_NONE &&
server.masterhost == NULL &&
(c->cmd->flags & CMD_WRITE ||
c->cmd->proc == pingCommand)) {
flagTransaction(c);
if (deny_write_type == DISK_ERROR_TYPE_RDB)
addReply(c, shared.bgsaveerr);
else
addReplySds(c,
sdscatprintf(sdsempty(),
"-MISCONF Errors writing to the AOF file: %s\r\n",
strerror(server.aof_last_write_errno)));
return C_OK;
}
// 7. 当此服务器时master时：如果配置了 repl_min_slaves_to_write，当slave数目小于时，禁止执行写命令
if (server.masterhost == NULL &&
server.repl_min_slaves_to_write &&
server.repl_min_slaves_max_lag &&
c->cmd->flags & CMD_WRITE &&
server.repl_good_slaves_count < server.repl_min_slaves_to_write) {
flagTransaction(c);
addReply(c, shared.noreplicaserr);
return C_OK;
}
// 8. 当只读时，除了 master 的命令，不执行任何其他指令
if (server.masterhost && server.repl_slave_ro &&
!(c->flags & CLIENT_MASTER) &&
c->cmd->flags & CMD_WRITE) {
addReply(c, shared.roslaveerr);
return C_OK;
}
// 9. 当客户端处于 Pub/Sub 时，只处理部分命令
if (c->flags & CLIENT_PUBSUB &&
c->cmd->proc != pingCommand &&
c->cmd->proc != subscribeCommand &&
c->cmd->proc != unsubscribeCommand &&
c->cmd->proc != psubscribeCommand &&
c->cmd->proc != punsubscribeCommand) {
addReplyError(c, "only (P)SUBSCRIBE / (P)UNSUBSCRIBE / PING / QUIT allowed in this context");
return C_OK;
}
// 10. 服务器为slave，但是没有连接 master 时，只会执行带有 CMD_STALE 标志的命令，如 info 等
if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED &&
server.repl_serve_stale_data == 0 &&
!(c->cmd->flags & CMD_STALE)) {
flagTransaction(c);
addReply(c, shared.masterdownerr);
return C_OK;
}
// 11. 正在加载数据库时，只会执行带有 CMD_LOADING 标志的命令，其余都会被拒绝
if (server.loading && !(c->cmd->flags & CMD_LOADING)) {
addReply(c, shared.loadingerr);
return C_OK;
}
// 12. 当服务器因为执行lua脚本阻塞时，只会执行部分命令，其余都会拒绝
if (server.lua_timedout &&
c->cmd->proc != authCommand &&
c->cmd->proc != replconfCommand &&
!(c->cmd->proc == shutdownCommand &&
c->argc == 2 &&
tolower(((char *) c->argv[1]->ptr)[0]) == 'n') &&
!(c->cmd->proc == scriptCommand &&
c->argc == 2 &&
tolower(((char *) c->argv[1]->ptr)[0]) == 'k')) {
flagTransaction(c);
addReply(c, shared.slowscripterr);
return C_OK;
}
// 13. 真正执行命令
if (c->flags & CLIENT_MULTI &&
c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
c->cmd->proc != multiCommand && c->cmd->proc != watchCommand) {
// 如果是事务命令，则开启事务，命令进入等待队列
queueMultiCommand(c);
addReply(c, shared.queued);
} else {
// 否则调用 call 直接执行
call(c, CMD_CALL_FULL);
c->woff = server.master_repl_offset;
if (listLength(server.ready_keys))
handleClientsBlockedOnKeys();
}
return C_OK;
}最后就是
void call(client *c, int flags)
{
....
start = ustime();
c->cmd->proc(c);
duration = ustime() - start;
....
}经过上面的过程，命令执行结束，对应的结果已经写在了 4. 在下一次事件循环之前，将写缓冲区中的数据发送给客户端这个过程在主事件循环之前的钩子函数 int main(int argc, char **argv)
{
...
aeSetBeforeSleepProc(server.el, beforeSleep);
aeSetAfterSleepProc(server.el, afterSleep);
aeMain(server.el); // 启动单线程网络模型
....
}
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
// 这是一个死循环，一直到 redis-server 停止
while (!eventLoop->stop) {
if (eventLoop->beforesleep != NULL)
eventLoop->beforesleep(eventLoop);
aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP); // 处理三个事件：time file call_after_sleep
}
}再具体的实现中，我们只关注如何将写缓冲区的数据写回给客户端： void beforeSleep(struct aeEventLoop *eventLoop) {
...
/* Handle writes with pending output buffers. */
handleClientsWithPendingWrites();
....
}
int handleClientsWithPendingWrites(void) {
listIter li;
listNode *ln;
int processed = listLength(server.clients_pending_write);
// clients_pending_write 是一个 client 队列，listRewind 获取一个用于迭代的游标
listRewind(server.clients_pending_write,&li);
// 当队列不为空时，持续进行下面的逻辑处理
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
c->flags &= ~CLIENT_PENDING_WRITE;
// 将遍历过 client 从队列中删除
listDelNode(server.clients_pending_write,ln);
/* If a client is protected, don't do anything,
* that may trigger write error or recreate handler. */
if (c->flags & CLIENT_PROTECTED) continue;
// 将 client 的数据写回 client 对应的s ocket
if (writeToClient(c->fd,c,0) == C_ERR) continue;
// 这次一次性没发完，那就给对应 socket 创建额外的写事件
if (clientHasPendingReplies(c)) {
int ae_flags = AE_WRITABLE;
/* For the fsync=always policy, we want that a given FD is never
* served for reading and writing in the same event loop iteration,
* so that in the middle of receiving the query, and serving it
* to the client, we'll call beforeSleep() that will do the
* actual fsync of AOF to disk. AE_BARRIER ensures that. */
if (server.aof_state == AOF_ON &&
server.aof_fsync == AOF_FSYNC_ALWAYS)
{
ae_flags |= AE_BARRIER;
}
if (aeCreateFileEvent(server.el, c->fd, ae_flags,
sendReplyToClient, c) == AE_ERR)
{
freeClientAsync(c);
}
}
}
return processed;
}对 /* Write data in output buffers to client. Return C_OK if the client
* is still valid after the call, C_ERR if it was freed. */
int writeToClient(int fd, client *c, int handler_installed) {
ssize_t nwritten = 0, totwritten = 0;
size_t objlen;
clientReplyBlock *o;
while(clientHasPendingReplies(c)) {
// 优先处理 buf，先发送一批。在执行之前会判断如果 client->buf 中有数据，则发送 client->buf 中的
if (c->bufpos > 0) {
nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen);
if (nwritten <= 0) break;
c->sentlen += nwritten;
totwritten += nwritten;
/* If the buffer was sent, set bufpos to zero to continue with
* the remainder of the reply. */
if ((int)c->sentlen == c->bufpos) {
c->bufpos = 0;
c->sentlen = 0;
}
} else {
// client->buf 中没数据了，则处理 client->reply 链表中剩下的
o = listNodeValue(listFirst(c->reply));
objlen = o->used;
if (objlen == 0) {
c->reply_bytes -= o->size;
listDelNode(c->reply,listFirst(c->reply));
continue;
}
nwritten = write(fd, o->buf + c->sentlen, objlen - c->sentlen);
if (nwritten <= 0) break;
c->sentlen += nwritten;
totwritten += nwritten;
/* If we fully sent the object on head go to the next one */
if (c->sentlen == objlen) {
c->reply_bytes -= o->size;
listDelNode(c->reply,listFirst(c->reply));
c->sentlen = 0;
/* If there are no longer objects in the list, we expect
* the count of reply bytes to be exactly zero. */
if (listLength(c->reply) == 0)
serverAssert(c->reply_bytes == 0);
}
}
if (totwritten > NET_MAX_WRITES_PER_EVENT &&
(server.maxmemory == 0 ||
zmalloc_used_memory() < server.maxmemory) &&
!(c->flags & CLIENT_SLAVE)) break;
}
server.stat_net_output_bytes += totwritten;
if (nwritten == -1) {
if (errno == EAGAIN) {
nwritten = 0;
} else {
serverLog(LL_VERBOSE,
"Error writing to client: %s", strerror(errno));
freeClient(c);
return C_ERR;
}
}
if (totwritten > 0) {
/* For clients representing masters we don't count sending data
* as an interaction, since we always send REPLCONF ACK commands
* that take some time to just fill the socket output buffer.
* We just rely on data / pings received for timeout detection. */
if (!(c->flags & CLIENT_MASTER)) c->lastinteraction = server.unixtime;
}
// 数据全部发送完毕了，那么前一步因为没发完而创建的文件监听事件可以从 EventLoop 中删除了
if (!clientHasPendingReplies(c)) {
c->sentlen = 0;
if (handler_installed) aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
/* Close connection after entire reply has been sent. */
if (c->flags & CLIENT_CLOSE_AFTER_REPLY) {
freeClient(c);
return C_ERR;
}
}
return C_OK;
} |
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
在关注 redis 单线程/多线程 时，有几个重要的时间节点：
从
Redis v1.0到Redis v6.0以前，Redis 的核心网络模型一直都是一个典型的 单Reactor模型，所有的事件都在这个线程内处理完成。本 issue 旨在解释清楚这个 单Reactor模型 的所有运作细节，为以后更好地理解新的 Multi-Reactors/Master-Workers 模型做准备。注：本文基于
Redis v5.0.0版本分析。The text was updated successfully, but these errors were encountered: