Redis IO模型
RedisIO模型
Redis服务器端单线程可以达到每秒可以达到数万QPS的处理能力。如此高性能的其中一个原因就是运用了Linux提供的IO多路复用机制epoll。
源码地址 https://github.com/redis/redis/blob/6.0/src/server.c
(1) Redis源码概览
//file: src/server.c
// #L5297 5297行
int main(int argc, char **argv) {
// 省略部分代码...
// 1. 启动RedisServer时 初始化配置及资源
initServer();
// 2. 循环处理请求及事件，直到服务器关闭为止
aeMain(server.el);
}(1.1) 启动RedisServer初始化配置及资源
在 initServer 这个函数内，Redis 做了这么三件重要的事情。
1、创建一个epoll对象
2、对配置的监听端口进行listen
3、把 listen socket 让 epoll 给管理起来
void initServer(void) {
// 1. 创建epoll
server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
// 2. 监听端口
listenToPort(server.port,server.ipfd,&server.ipfd_count)
// 3. 注册accept事件处理器 (只是注册 后面会用到)
for (j = 0; j < server.ipfd_count; j++) {
aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
acceptTcpHandler,NULL)
}
}(1.2) 循环处理请求
// file: src/ae.c
/*
* 循环处理事件
*
* @param *eventLoop
*/
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
// 循环处理事件
while (!eventLoop->stop) {
// 处理事件
aeProcessEvents(eventLoop, AE_ALL_EVENTS|
AE_CALL_BEFORE_SLEEP|
AE_CALL_AFTER_SLEEP);
}
}// file: src/ae.c
/*
* 处理事件
*
* @param *eventLoop
* @param flags
*/
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
// 省略部分细节...
// 调用多路复用API获取就绪事件
numevents = aeApiPoll(eventLoop, tvp);
// 处理写事件
fe->wfileProc(eventLoop,fd,fe->clientData,mask);
// 处理读事件
fe->rfileProc(eventLoop,fd,fe->clientData,mask);
} 其实整个 Redis 的工作过程，就只需要理解清楚 main 函数中调用的 initServer 和 aeMain 这两个函数就足够了。
(2) RedisServer里IO多路复用代码详解
在 initServer 这个函数内，Redis 做了这么三件重要的事情。
1、创建一个 epoll 对象
2、对配置的端口进行监听(listen)
3、把 listen socket 让 epoll 给管理起来
//file: src/server.c
void initServer(void) {
// 2.1 创建 epoll
server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
// 2.2 绑定监听服务端口
// Open the TCP listening socket for the user commands.
listenToPort(server.port,server.ipfd,&server.ipfd_count)
// 2.3 注册accept事件处理器
// Create an event handler for accepting new connections in TCP and Unix domain sockets.
for (j = 0; j < server.ipfd_count; j++) {
aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
acceptTcpHandler,NULL)
}
}(2.1) 创建epoll-aeCreateEventLoop
redisServer结构
// file: src/server.h
// redisServer结构体
struct redisServer {
// ... 省略部分代码
aeEventLoop *el;
// ... 省略部分代码
}aeEventLoop结构
// file: src/ae.h
// 基于事件的程序的状态
/* State of an event based program */
typedef struct aeEventLoop {
int maxfd; // 当前注册的最大文件描述符
int setsize; // 跟踪的最大文件描述符数
long long timeEventNextId;
time_t lastTime; /* Used to detect system clock skew */
aeFileEvent *events; // 注册事件数组的指针 指向aeFileEvent数组
aeFiredEvent *fired; // 就绪事件数组的指针 指向aeFiredEvent数组
aeTimeEvent *timeEventHead; // 时间事件
int stop;
void *apidata; // 指向aeApiState结构体 创建的epoll对象就在aeApiState->epfd
aeBeforeSleepProc *beforesleep; // 在事件处理前执行的函数
aeBeforeSleepProc *aftersleep; // 在事件处理后执行的函数
int flags;
} aeEventLoop;
// file: src/ae.h
// 文件事件结构
/* File event structure */
typedef struct aeFileEvent {
int mask; // 标记 可读/可写/屏障
aeFileProc *rfileProc; // 写事件回调
aeFileProc *wfileProc; // 读事件回调
void *clientData; // 扩展数据
} aeFileEvent;Redis 在操作系统提供的 epoll 对象基础上又封装了一个 eventLoop 出来，所以创建的时候是先申请和创建 eventLoop。
// file: src/ae.c
/*
* 创建aeEventLoop结构体
*
* @param setsize
*/
aeEventLoop *aeCreateEventLoop(int setsize) {
aeEventLoop *eventLoop;
// ... 省略部分代码
eventLoop = zmalloc(sizeof(*eventLoop))
// 将来的各种回调事件就都会存在这里
// eventLoop->events是一个指针 指向数组 元素类型:aeFileEvent 大小:setsize
eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);
eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);
// ... 省略部分代码
// 创建epoll
aeApiCreate(eventLoop)
}// file: src/ae_epoll.c
static int aeApiCreate(aeEventLoop *eventLoop) {
aeApiState *state = zmalloc(sizeof(aeApiState));
// ... 省略部分代码
// 真正创建epoll
// 调linux epoll_create()函数 创建epoll
state->epfd = epoll_create(2024); /* 1024 is just a hint for the kernel */
// ... 省略部分代码
eventLoop->apidata = state;
return 0;
}(2.2) 注册事件及回调函数-aeCreateFileEvent
// file: src/ae.c
/*
* @param *eventLoop
* @param fd
* @param mask 0:未注册事件 1:描述符可读时触发 2:描述符可写时触发 3:
* @param *proc aeFileProc类型 入参传的是 acceptTcpHandler函数 回调时会用到这个函数
* @param *clientData
*/
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
aeFileProc *proc, void *clientData)
{
if (fd >= eventLoop->setsize) {
errno = ERANGE;
return AE_ERR;
}
// 从aeFileEvent事件数组里取出一个文件事件结构
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;
}//file: src/ae_epoll.c
// 添加事件
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. */
int op = eventLoop->events[fd].mask == AE_NONE ?
EPOLL_CTL_ADD : EPOLL_CTL_MOD;
// ...
// epoll_ctl 添加事件
epoll_ctl(state->epfd,op,fd,&ee);
return 0;
}这个函数其实就是对 epoll_ctl 的一个封装。主要就是实际执行 epoll_ctl EPOLL_CTL_ADD。
每一个 eventLoop->events 元素都指向一个 aeFileEvent 对象。
在这个对象上，设置了三个关键东西
rfileProc：读事件回调
wfileProc：写事件回调
clientData：一些额外的扩展数据
将来 当 epoll_wait 发现某个 fd 上有事件发生的时候，这样 redis 首先根据 fd 到 eventLoop->events 中查找 aeFileEvent 对象，然后再看 rfileProc、wfileProc 就可以找到读、写回调处理函数。
listen fd 对应的读回调函数 rfileProc 事实上就被设置成了 acceptTcpHandler，私有数据 client_data 也为 null。
(2.3) 获取就绪socket并处理事件-aeMain()
// file: src/ae.c
/*
* 循环接收请求
*
* @param *eventLoop
*/
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
// 循环
while (!eventLoop->stop) {
// 处理事件
aeProcessEvents(eventLoop, AE_ALL_EVENTS|
AE_CALL_BEFORE_SLEEP|
AE_CALL_AFTER_SLEEP);
}
}// 处理事件 返回处理完的事件个数
0 不做任何处理
1 AE_FILE_EVENTS 处理文件事件
2 AE_TIME_EVENTS 处理时间事件
3 AE_ALL_EVENTS 所有事件
4 AE_DONT_WAIT
8 AE_CALL_BEFORE_SLEEP
16 AE_CALL_AFTER_SLEEP
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
int processed = 0, numevents;
struct timeval tv, *tvp;
// 如果eventLoop处理前的函数不为空，就执行
if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP)
eventLoop->beforesleep(eventLoop);
// 调用多路复用 API，仅在超时或某些事件触发时返回
// 处理文件事件，阻塞时间由tvp决定
numevents = aeApiPoll(eventLoop, tvp);
// 处理后的函数不为空
/* After sleep callback. */
if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)
eventLoop->aftersleep(eventLoop);
for (j = 0; j < numevents; j++) {
// 先从eventLoop->fired[j]获取已就绪事件结构体(aeFiredEvent) 获取fd后 再从eventLoop->events注册事件里获取对应的事件结构体(aeFileEvent)
aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
// ...
// 如果可读
if (!invert && fe->mask & mask & AE_READABLE) {
// 调用读事件回调函数 对应 acceptTcpHandler
fe->rfileProc(eventLoop,fd,fe->clientData,mask);
fired++;
fe = &eventLoop->events[fd]; /* Refresh in case of resize. */
}
// 如果可写 触发写事件
if (fe->mask & mask & AE_WRITABLE) {
if (!fired || fe->wfileProc != fe->rfileProc) {
// 调用写事件回调函数 对应 acceptTcpHandler
fe->wfileProc(eventLoop,fd,fe->clientData,mask);
fired++;
}
}
processed++;
}
return processed; /* return the number of processed file/time events */
}// file: src/ae_poll.c
/*
* 获取就绪事件
*
* @param *eventLoop
* @param *tvp
*/
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
// 等待事件
aeApiState *state = eventLoop->apidata;
int retval, numevents = 0;
// 调linux epoll_wait函数来获取已就绪socket
retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,
tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);
// ...
return numevents;
} aeProcessEvents 就是调用 epoll_wait 来获取就绪socket 。
当发现有某个socket上数据就绪以后，则调用事先注册的事件处理器函数 rfileProc 和 wfileProc。
(3) RedisServer网络请求处理详细流程
(3.1) 绑定地址并监听套接字(bind listen)
// file: src/server.c
// 监听端口
/*
* @param port
* @param *fds
* @param *count
*/
int listenToPort(int port, int *fds, int *count) {
// ...
for (j = 0; j < server.bindaddr_count || j == 0; j++) {
// ... 省略 绑定IPV6 IPV4的细节 anetTcp6Server anetTcpServer
// 绑定
fds[*count] = anetTcpServer(server.neterr,port,NULL,
server.tcp_backlog);
// ...
}
return C_OK;
} Redis 是支持开启多个端口的，所以在 listenToPort 中我们看到是启用一个循环来调用 anetTcpServer。
在 anetTcpServer 中，逐步会展开调用，直到执行到 bind 和 listen 系统调用。
// file: src/anet.c
/*
* @param *err
* @param port
* @param *bindaddr
* @param backlog
*/
int anetTcpServer(char *err, int port, char *bindaddr, int backlog)
{
return _anetTcpServer(err, port, bindaddr, AF_INET, backlog);
}
/*
* @param *err
* @param port
* @param *bindaddr
* @param af
* @param backlog
*/
static int _anetTcpServer(char *err, int port, char *bindaddr, int af, int backlog)
{
int s = -1, rv;
char _port[6]; /* strlen("65535") */
struct addrinfo hints, *servinfo, *p;
// 创建套接字
s = socket(p->ai_family,p->ai_socktype,p->ai_protocol)
// 设置端口重用
anetSetReuseAddr(err,s)
// 监听
anetListen(err,s,p->ai_addr,p->ai_addrlen,backlog)
/*
* 监听
*
* @param *err
* @param s 对应创建的套接字fd
* @param *sa socket地址信息 (协议 地址)
* @param len
* @param backlog
*/
static int anetListen(char *err, int s, struct sockaddr *sa, socklen_t len, int backlog) {
// 绑定ip端口
bind(s,sa,len)
// 监听套接字
listen(s, backlog)
return ANET_OK;
}(3.2) 和客户端建立连接(accept)
// file: src/networking.c
/*
* 接收tcp处理器
*
* @param *el
* @param fd
* @param *privdata
* @param mask
*/
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
// ...
// 接收tcp请求
cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
// ...
// 接收通用处理
acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip);
}(3.2.1) 接收tcp请求-anetTcpAccept
// file: src/anet.c
/*
* 接收tcp请求
*
* @param *err
* @param s fd
* @param *ip
* @param ip_len
* @param *port
*/
int anetTcpAccept(char *err, int s, char *ip, size_t ip_len, int *port) {
int fd;
struct sockaddr_storage sa; // 套接字地址存储结构体
socklen_t salen = sizeof(sa);
// 接收请求
fd = anetGenericAccept(err,s,(struct sockaddr*)&sa,&salen)
// ...
return fd;
}// file: src/anet.c
/*
* @param *err
* @param s
* @param *sa
* @param *len
*/
static int anetGenericAccept(char *err, int s, struct sockaddr *sa, socklen_t *len) {
int fd;
while(1) {
// 接收socket数据
// fd是socket返回的socket，指向的定义的SOCKADDR_IN 结构体指针，指针的大小
fd = accept(s,sa,len);
// ...
break;
}
return fd;
}(3.2.2) 接收公共处理器-acceptCommonHandler
/*
*
*/
acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip);// file: src/networking.c
/*
* @param *conn
* @param flags
* @param *ip
*/
static void acceptCommonHandler(connection *conn, int flags, char *ip) {
client *c;
// 创建redisClient对象
c = createClient(conn)
// 建立连接
connAccept(conn, clientAcceptHandler)
} (3.2.2.1) connCreateAcceptedSocket()
conn是怎么创建的
// file: src/connection.c
/*
* @param fd
*/
connection *connCreateAcceptedSocket(int fd) {
// 创建连接
connection *conn = connCreateSocket();
//
conn->fd = fd;
// 设置连接状态为接收中
conn->state = CONN_STATE_ACCEPTING;
return conn;
}// file: src/connection.h
/*
* 连接结构体
*/
struct connection {
ConnectionType *type; // 连接类型 在后面会用到
ConnectionState state; // 连接状态
short int flags;
short int refs;
int last_errno;
void *private_data; // 私有数据
ConnectionCallbackFunc conn_handler; // 连接处理器
ConnectionCallbackFunc write_handler; // 写处理器
ConnectionCallbackFunc read_handler; // 读处理器
int fd; //
};(3.2.2.2) 创建redisClient对象-createClient()
// file: src/networking.c
/*
* @param *conn
*/
client *createClient(connection *conn) {
// 为用户连接创建client结构体
client *c = zmalloc(sizeof(client));
if (conn) {
// ... 处理连接
// 注册读事件处理器，等连接可读时调用 回调函数是readQueryFromClient
connSetReadHandler(conn, readQueryFromClient);
// 会把新创建的client结构体放到 conn结构体的private_data字段里
connSetPrivateData(conn, c);
}
// 设置client的一些参数
selectDb(c,0);
uint64_t client_id = ++server.next_client_id;
c->id = client_id;
c->resp = 2;
c->conn = conn;
// ...
return c;
}client结构体主要属性
// file: src/server.h
typedef struct client {
uint64_t id; /* 客户端增量唯一ID */
connection *conn; /* 连接 */
int resp; /* 响应协议版本 可能是2或3 */
redisDb *db; /* 指向当前选中的Db */
robj *name; /* 由客户端设置的名称 */
sds querybuf; /* 用来累积客户端查询的缓冲区 */
size_t qb_pos; /* querybuf中读取到的位置 */
// ... 省略部分字段
/* Response buffer */
int bufpos; //
char buf[PROTO_REPLY_CHUNK_BYTES]; //
} client;// file: src/connection.h
/*
* 注册读事件处理器，等连接可读时调用
*
* @param *conn
* @param func ConnectionCallbackFunc类型的回调函数
*/
static inline int connSetReadHandler(connection *conn, ConnectionCallbackFunc func) {
return conn->type->set_read_handler(conn, func);
}(3.2.2.3) 接收请求-connAccept()
/* The connection module does not deal with listening and accepting sockets,
* so we assume we have a socket when an incoming connection is created.
*
* The fd supplied should therefore be associated with an already accept()ed
* socket.
*
* connAccept() may directly call accept_handler(), or return and call it
* at a later time. This behavior is a bit awkward but aims to reduce the need
* to wait for the next event loop, if no additional handshake is required.
*
* IMPORTANT: accept_handler may decide to close the connection, calling connClose().
* To make this safe, the connection is only marked with CONN_FLAG_CLOSE_SCHEDULED
* in this case, and connAccept() returns with an error.
*
* connAccept() callers must always check the return value and on error (C_ERR)
* a connClose() must be called.
*
* @param *conn 连接
* @param accept_handler 接收处理器 是一个回调函数 ConnectionCallbackFunc
*/
static inline int connAccept(connection *conn, ConnectionCallbackFunc accept_handler) {
return conn->type->accept(conn, accept_handler);
}conn->type 对应 ConnectionType结构体
// file: src/connection.h
typedef struct ConnectionType {
void (*ae_handler)(struct aeEventLoop *el, int fd, void *clientData, int mask);
int (*connect)(struct connection *conn, const char *addr, int port, const char *source_addr, ConnectionCallbackFunc connect_handler);
int (*write)(struct connection *conn, const void *data, size_t data_len);
int (*read)(struct connection *conn, void *buf, size_t buf_len);
void (*close)(struct connection *conn);
int (*accept)(struct connection *conn, ConnectionCallbackFunc accept_handler);
int (*set_write_handler)(struct connection *conn, ConnectionCallbackFunc handler, int barrier);
int (*set_read_handler)(struct connection *conn, ConnectionCallbackFunc handler);
const char *(*get_last_error)(struct connection *conn);
int (*blocking_connect)(struct connection *conn, const char *addr, int port, long long timeout);
ssize_t (*sync_write)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
ssize_t (*sync_read)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
ssize_t (*sync_readline)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
int (*get_type)(struct connection *conn);
} ConnectionType;file: src/networking.c
/*
* @param *conn
*/
void clientAcceptHandler(connection *conn) {
//
client *c = connGetPrivateData(conn);
//
moduleFireServerEvent(REDISMODULE_EVENT_CLIENT_CHANGE,
REDISMODULE_SUBEVENT_CLIENT_CHANGE_CONNECTED,
c);
}(3.3) 从连接的socket读取客户端发送的数据(recv/read)
上一步监听后，等到client向redisServer发送的数据到达，会触发设置的回调方法readQueryFromClient，redisServer会调用readQueryFromClient()方法
// file: src/networking.c
/*
* @param *conn
*/
void readQueryFromClient(connection *conn) {
// 从连接的私有数据获取client // 在创建连接时把client放到了connection的private_data字段
client *c = connGetPrivateData(conn);
//
// 客户端输入缓冲区中有更多的数据，请继续解析它，以防检查是否有要执行的完整命令。
processInputBuffer(c);
}(3.4) 处理读到的数据
从输入缓冲区读到数据后，下面就开始处理数据
// file: src/networking.c
/* This function is called every time, in the client structure 'c', there is
* more query buffer to process, because we read more data from the socket
* or because a client was blocked and later reactivated, so there could be
* pending query buffer, already representing a full command, to process. */
void processInputBuffer(client *c) {
// ...
processCommandAndResetClient(c);
}
/* This function calls processCommand(), but also performs a few sub tasks
* for the client that are useful in that context:
*
* 1. It sets the current client to the client 'c'.
* 2. calls commandProcessed() if the command was handled.
*
* The function returns C_ERR in case the client was freed as a side effect
* of processing the command, otherwise C_OK is returned. */
int processCommandAndResetClient(client *c) {
int deadclient = 0;
server.current_client = c;
// 处理命令
if (processCommand(c) == C_OK) {
commandProcessed(c);
}
return deadclient ? C_ERR : C_OK;
}
// file: src/server.c
/*
* 处理各种命令 get set del exits quit lpush sadd 等
*
* @param *c
*/
int processCommand(client *c) {
// 查找命令，并进行命令合法性检查，以及命令参数个数检查
/* Now lookup the command and check ASAP about trivial error conditions
* such as wrong arity, bad command name and so forth. */
c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);
// ... 省略其他命令处理逻辑
// 处理命令
/* Exec the command */
if (c->flags & CLIENT_MULTI &&
c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
{ // 如果是 MULTI 事务，则入队
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;
}// file: src/server.c
/*
* 根据key查找值
*
* @param name
*/
struct redisCommand *lookupCommand(sds name) {
//
return dictFetchValue(server.commands, name);
}
/*
* call() 是 Redis 执行命令的核心。
*
* @param *c
* @param flags
*/
void call(client *c, int flags) {
// 要执行的redis命令
struct redisCommand *real_cmd = c->cmd;
// 调用命令处理函数
c->cmd->proc(c);
}proc对应的command有以下几种
/*
*
*/
struct redisCommand redisCommandTable[] = {
{"module",moduleCommand,-2,
"admin no-script",
0,NULL,0,0,0,0,0,0},
{"get",getCommand,2,
"read-only fast @string",
0,NULL,1,1,1,0,0,0},
/* Note that we can't flag set as fast, since it may perform an
* implicit DEL of a large key. */
{"set",setCommand,-3,
"write use-memory @string",
0,NULL,1,1,1,0,0,0},
{"setnx",setnxCommand,3,
"write use-memory fast @string",
0,NULL,1,1,1,0,0,0},
{"setex",setexCommand,4,
"write use-memory @string",
0,NULL,1,1,1,0,0,0},
// ..
{"rpush",rpushCommand,-3,
"write use-memory fast @list",
0,NULL,1,1,1,0,0,0},
{"lpush",lpushCommand,-3,
"write use-memory fast @list",
0,NULL,1,1,1,0,0,0},
// ...
{"sadd",saddCommand,-3,
"write use-memory fast @set",
0,NULL,1,1,1,0,0,0},
// ...
{"zadd",zaddCommand,-4,
"write use-memory fast @sortedset",
0,NULL,1,1,1,0,0,0},
// ...
{"stralgo",stralgoCommand,-2,
"read-only @string",
0,lcsGetKeys,0,0,0,0,0,0}
};
如果命令是get，其对应的命令处理函数就是 getCommand
(3.4.1) getCommand
// file: t_string.c
/*
* @param *c
*/
void getCommand(client *c) {
getGenericCommand(c);
}
int getGenericCommand(client *c) {
robj *o;
// 查找key
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp])) == NULL)
return C_OK;
// 找到key对应的值了
if (o->type != OBJ_STRING) { // key的对象类型不是string类型 返回错误
addReply(c,shared.wrongtypeerr);
return C_ERR;
} else { // key的对象类型是string类型
// 将结果添加到输出缓冲区中
addReplyBulk(c,o);
return C_OK;
}
}
// file: src/networking.c
/* Add a Redis Object as a bulk reply */
void addReplyBulk(client *c, robj *obj) {
addReplyBulkLen(c,obj);
addReply(c,obj);
addReply(c,shared.crlf);
}(3.4.2) setCommand
// file: src.t_string.c
/*
*
* @param *c
* @param flags
* @param *key
* @param *val
* @param *expire
* @param unit
* @param *ok_reply
* @param *abort_reply
*/
void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {
// 64位精度整数
long long milliseconds = 0; /* initialized to avoid any harmness warning */
if (expire) {
if (getLongLongFromObjectOrReply(c, expire, &milliseconds, NULL) != C_OK)
return;
if (milliseconds <= 0) {
addReplyErrorFormat(c,"invalid expire time in %s",c->cmd->name);
return;
}
if (unit == UNIT_SECONDS) milliseconds *= 1000;
}
if ((flags & OBJ_SET_NX && lookupKeyWrite(c->db,key) != NULL) ||
(flags & OBJ_SET_XX && lookupKeyWrite(c->db,key) == NULL))
{
addReply(c, abort_reply ? abort_reply : shared.null[c->resp]);
return;
}
genericSetKey(c,c->db,key,val,flags & OBJ_SET_KEEPTTL,1);
server.dirty++;
if (expire) setExpire(c,c->db,key,mstime()+milliseconds);
notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);
if (expire) notifyKeyspaceEvent(NOTIFY_GENERIC,
"expire",key,c->db->id);
addReply(c, ok_reply ? ok_reply : shared.ok);
}(3.5) 给客户端返回结果(send)
无论是执行get命令还是set命令，最后执行完命令都会调用addReply()方法
在addReply方法中做了两件事情:
1、prepareClientToWrite 判断是否需要返回数据，并且将当前 client 添加到等待写返回数据队列中。
2、调用 _addReplyToBuffer 和 _addReplyObjectToList 方法将返回值写入到输出缓冲区中，等待写入 socekt
// file: src/networking.c
/* -----------------------------------------------------------------------------
* 更高级别的函数用于在客户端输出缓冲区上对数据进行排队。
* 以下函数是命令实现将调用的函数。
* -------------------------------------------------------------------------- */
/*
* 将对象“obj”字符串表示添加到客户端输出缓冲区。
*
* @param *c redis client
* @param *obj 命令执行的结果 类型是redisObject
*/
void addReply(client *c, robj *obj) {
// 判断client是否可以接收新数据 (假客户端不能接收)
if (prepareClientToWrite(c) != C_OK) return;
// 根据redisobject格式把数据写入缓存
if (sdsEncodedObject(obj)) { // obj如果是row或者embstr格式
// 尝试将应答添加到客户端结构中的静态缓冲区。
if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != C_OK)
// 将回复添加到回复列表中。
_addReplyProtoToList(c,obj->ptr,sdslen(obj->ptr));
} else if (obj->encoding == OBJ_ENCODING_INT) { // obj 是数字格式
/* 对于整数编码字符串，我们只需使用优化函数将其转换为字符串，并将结果字符串附加到输出缓冲区。 */
char buf[32];
// 数字转为字符串
size_t len = ll2string(buf,sizeof(buf),(long)obj->ptr);
if (_addReplyToBuffer(c,buf,len) != C_OK)
_addReplyProtoToList(c,buf,len);
} else {
serverPanic("Wrong obj->encoding in addReply()");
}
}// file: src/networking.c
/* -----------------------------------------------------------------------------
* 低级函数用于向输出缓冲区添加更多数据。
* -------------------------------------------------------------------------- */
/*
* 尝试将应答添加到客户端结构中的静态缓冲区。
* 如果缓冲区已满或回复列表不为空，则返回C_ERR，在这种情况下，必须将回复添加到回复列表中。
*
* @param *c
* @param *s 要写入的数据
* @param len 数据长度
*/
int _addReplyToBuffer(client *c, const char *s, size_t len) {
// 剩余缓冲区大小
size_t available = sizeof(c->buf)-c->bufpos;
if (c->flags & CLIENT_CLOSE_AFTER_REPLY) return C_OK;
/* 如果回复列表中已经有内容，则无法向静态缓冲区添加更多内容。 */
if (listLength(c->reply) > 0) return C_ERR;
/* 检查缓冲区是否有足够的空间用于此字符串。 */
if (len > available) return C_ERR;
// 把数据*s(char[]类型) 拷贝到 client对象的Response buffer中
memcpy(c->buf+c->bufpos,s,len);
// 更新已使用缓冲区大小
c->bufpos+=len;
return C_OK;
}/*
* 将回复添加到回复列表中。
* 注意：对该函数的一些编辑需要转发到AddReplyFromClient。
*
* @param *c
* @param *s 要写入的数据
* @param len 数据长度
*/
void _addReplyProtoToList(client *c, const char *s, size_t len) {
// 写入回复后关闭
if (c->flags & CLIENT_CLOSE_AFTER_REPLY) return;
// 双向链表尾部
listNode *ln = listLast(c->reply);
// 链表里存的数据是 clientReplyBlock类型 是一个buf数组，有大小限制
clientReplyBlock *tail = ln? listNodeValue(ln): NULL;
/* Note that 'tail' may be NULL even if we have a tail node, because when
* addReplyDeferredLen() is used, it sets a dummy node to NULL just
* fo fill it later, when the size of the bulk length is set. */
/* 尽可能追加到尾部字符串。*/
if (tail) {
/* 复制我们可以放入尾部的部分，并将其余部分留给新节点 */
size_t avail = tail->size - tail->used;
// *s要复制的部分 (可能是全部，可能是部分)
size_t copy = avail >= len? len: avail;
// 复制到buf数组里
memcpy(tail->buf + tail->used, s, copy);
tail->used += copy;
s += copy;
len -= copy;
}
// len>0
if (len) {
/* 创建一个新节点，确保至少为其分配了 16K */
size_t size = len < PROTO_REPLY_CHUNK_BYTES? PROTO_REPLY_CHUNK_BYTES: len;
tail = zmalloc(size + sizeof(clientReplyBlock));
/* take over the allocation's internal fragmentation */
tail->size = zmalloc_usable(tail) - sizeof(clientReplyBlock);
tail->used = len;
memcpy(tail->buf, s, len);
listAddNodeTail(c->reply, tail);
c->reply_bytes += tail->size;
}
// 缓冲区达到限制后异步关闭客户端
asyncCloseClientOnOutputBufferLimitReached(c);
}(3.5.1) 回复列表中的数据什么时候写入到输出缓冲区呢？
// file: src/ae.c
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
// 循环
while (!eventLoop->stop) {
// 处理事件
aeProcessEvents(eventLoop, AE_ALL_EVENTS|
AE_CALL_BEFORE_SLEEP|
AE_CALL_AFTER_SLEEP);
}
}
/*
* 处理事件
*/
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
//
if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP)
eventLoop->beforesleep(eventLoop);
}// file: src/server.c
/* This function gets called every time Redis is entering the
* main loop of the event driven library, that is, before to sleep
* for ready file descriptors.
*
* Note: This function is (currently) called from two functions:
* 1. aeMain - The main server loop
* 2. processEventsWhileBlocked - Process clients during RDB/AOF load
*
* If it was called from processEventsWhileBlocked we don't want
* to perform all actions (For example, we don't want to expire
* keys), but we do need to perform some actions.
*
* The most important is freeClientsInAsyncFreeQueue but we also
* call some other low-risk functions. */
void beforeSleep(struct aeEventLoop *eventLoop) {
/* 处理具有挂起的输出缓冲区的写入。 */
handleClientsWithPendingWritesUsingThreads();
}// file: src/networking.c
int handleClientsWithPendingWritesUsingThreads(void) {
int processed = listLength(server.clients_pending_write);
if (processed == 0) return 0; /* Return ASAP if there are no clients. */
/* If I/O threads are disabled or we have few clients to serve, don't
* use I/O threads, but thejboring synchronous code. */
if (server.io_threads_num == 1 || stopThreadedIOIfNeeded()) {
return handleClientsWithPendingWrites();
}
/* Start threads if needed. */
if (!server.io_threads_active) startThreadedIO();
if (tio_debug) printf("%d TOTAL WRITE pending clients\n", processed);
/* Distribute the clients across N different lists. */
listIter li;
listNode *ln;
/* 把server.clients_pending_write链表 赋值 给迭代器&li */
listRewind(server.clients_pending_write,&li);
int item_id = 0;
// 遍历链表 server.clients_pending_write
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
c->flags &= ~CLIENT_PENDING_WRITE;
// & 状态=尽快关闭
if (c->flags & CLIENT_CLOSE_ASAP) {
// 删除双向链表里的当前节点
listDelNode(server.clients_pending_write, ln);
continue;
}
int target_id = item_id % server.io_threads_num;
// 把c添加到io_threads_list[target_id]链表尾部 后面会用到
listAddNodeTail(io_threads_list[target_id],c);
item_id++;
}
/* 把io_threads_list[0]链表 赋值 给迭代器&li */
listRewind(io_threads_list[0],&li);
// 遍历链表 io_threads_list[0]
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
// 将client的数据发送出去
writeToClient(c,0);
}
listEmpty(io_threads_list[0]);
/* 把server.clients_pending_write链表 赋值 给迭代器&li */
listRewind(server.clients_pending_write,&li);
// 遍历链表 server.clients_pending_write
while((ln = listNext(&li))) {
// 获取节点
client *c = listNodeValue(ln);
/* 如果某些客户端中存在挂起的写入，安装写入处理程序。*/
// 如果一次发送不完则准备下一次发送
if (clientHasPendingReplies(c) &&
connSetWriteHandler(c->conn, sendReplyToClient) == AE_ERR)
{
freeClientAsync(c);
}
}
listEmpty(server.clients_pending_write);
return processed;
}(3.5.2) 写事件处理器-sendReplyToClient
// file: src/networking.c
/*
* 写事件处理器
* 仅仅发送数据到client
*
* Write event handler. Just send data to the client.
*/
void sendReplyToClient(connection *conn) {
client *c = connGetPrivateData(conn);
//
writeToClient(c,1);
}(3.5.3) 把数据写入客户端的输出缓冲区
// file: src/networking.c
/*
* 把数据写入客户端的输出缓冲区
*
* 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 because of some
* error. If handler_installed is set, it will attempt to clear the
* write event.
*
* This function is called by threads, but always with handler_installed
* set to 0. So when handler_installed is set to 0 the function must be
* thread safe. */
int writeToClient(client *c, int handler_installed) {
/* Update total number of writes on server */
server.stat_total_writes_processed++;
ssize_t nwritten = 0, totwritten = 0;
size_t objlen;
clientReplyBlock *o;
while(clientHasPendingReplies(c)) {
if (c->bufpos > 0) { // 缓冲区有数据
// 把缓冲区数据写入socket
nwritten = connWrite(c->conn,c->buf+c->sentlen,c->bufpos-c->sentlen);
} else { // 处理待发送链表
o = listNodeValue(listFirst(c->reply));
objlen = o->used;
if (objlen == 0) {
c->reply_bytes -= o->size;
listDelNode(c->reply,listFirst(c->reply));
continue;
}
// 把链表节点里的数据写入socket
nwritten = connWrite(c->conn, o->buf + c->sentlen, objlen - c->sentlen);
}
}
return C_OK;
}// file: src/connection.h
/*
* 把数据写入到连接里
*/
static inline int connWrite(connection *conn, const void *data, size_t data_len) {
return conn->type->write(conn, data, data_len);
}(4) RedisServer-IO模型流程概览
(5) 思考
(5.1) Redis有哪些潜在的性能瓶颈
Redis单线程处理IO请求性能瓶颈主要包括2个方面：
1、任意一个请求在server中一旦发生耗时，都会影响整个server的性能，也就是说后面的请求都要等前面这个耗时请求处理完成，自己才能被处理到。
2、并发量非常大时，单线程读写客户端IO数据存在性能瓶颈，虽然采用IO多路复用机制，但是读写客户端数据时从内核缓冲区中拷贝数据的操作是同步操作，只能单线程依次读取客户端的数据，无法利用到CPU多核。
耗时的操作包括以下几种：
a、操作bigkey：写入一个bigkey在分配内存时需要消耗更多的时间，同样，删除bigkey释放内存同样会产生耗时；
b、使用复杂度过高的命令：例如SORT/SUNION/ZUNIONSTORE，或者O(N)命令，但是N很大，例如lrange key 0 -1一次查询全量数据；
c、大量key集中过期：Redis的过期机制也是在主线程中执行的，大量key集中过期会导致处理一个请求时，耗时都在删除过期key，耗时变长；
d、淘汰策略：淘汰策略也是在主线程执行的，当内存超过Redis内存上限后，每次写入都需要淘汰一些key，也会造成耗时变长；
e、AOF刷盘开启always机制：每次写入都需要把这个操作刷到磁盘，写磁盘的速度远比写内存慢，会拖慢Redis的性能；
f、主从全量同步生成RDB：虽然采用fork子进程生成数据快照，但fork这一瞬间也是会阻塞整个线程的，实例越大，阻塞时间越久；
针对问题1，一方面需要业务人员去规避，一方面Redis在4.0推出了lazy-free机制，把bigkey释放内存的耗时操作放在了异步线程中执行，降低对主线程的影响。
针对问题2，Redis在6.0推出了多线程，可以在高并发场景下利用CPU多核多线程读写客户端数据，进一步提升server性能，当然，只是针对客户端的读写是并行的，每个命令的真正操作依旧是单线程的。
(5.2) Redis 6.0版本为什么又引入了多线程
Redis 的瓶颈不在 CPU ，而在内存和网络，内存不够可以增加内存或通过数据结构等进行优化
但 Redis 的网络 IO 的读写占用了发部分 CPU 的时间，如果可以把网络处理改成多线程的方式，性能会有很大提升
所以总结下 Redis 6.0 版本引入多线程有两个原因
1.充分利用服务器的多核资源
2.多线程分摊 Redis 同步 IO 读写负荷
执行命令还是由单线程顺序执行，只是处理网络数据读写采用了多线程，而且 IO 线程要么同时读 Socket ，要么同时写 Socket ，不会同时读写
参考资料
[1] 深度解析单线程的 Redis 如何做到每秒数万 QPS 的超高处理能力！
[2] epoll源码分析以及在Redis中的实现
[3] Redis5.0.5源码解析 – 多路复用模型
[4] 03 | 高性能IO模型：为什么单线程Redis能那么快？
[5] Redis多线程架构的演进
[6] 性能突出的Redis是咋使用epoll的？