Redis 整个程序的入口函数在 server.c 中的 main 函数, 函数调用关系如下图 1, 调用顺序为从上到下, 从左至右.
图 1 Redis 启动函数调用图
main 函数源码如下, 1-55 行根据配置文件和启动命令参数设置全局对象 server ,57-59 设置 Redis 的服务器端为后台进程, initServer 主要提前创建一些经常用到的对象用于节约内存, 根据设置的 ip 地址和端口创建监听套接字用于客户端连接, 并初始化时间事件, 64 行用于设置 server->el ->beforesleep = beforesleep,aeMain 函数是个循环函数, 用于监听客户端连接, 接收客户端命令并进行处理等.
- // 整个程序的入口函数
- int main(int argc, char **argv) {
- // 初始化服务器配置, 设置全局对象 server 的状态
- initServerConfig();
- /* Store the executable path and arguments in a safe place in order
- * to be able to restart the server later. */
- // 存储运行命令的绝对路径及运行参数
- server.executable = getAbsolutePath(argv[0]);
- server.exec_argv = zmalloc(sizeof(char*)*(argc+1));
- server.exec_argv[argc] = NULL;
- for (j = 0; j <argc; j++) server.exec_argv[j] = zstrdup(argv[j]);
- if (argc>= 2) {
- j = 1; /* First option to parse in argv[] */
- sds options = sdsempty();
- char *configfile = NULL;
- // 命令指定了配置文件, 对配置文件做处理, 配置文件跟在程序名后第一位
- /* First argument is the config file name? */
- if (argv[j][0] != '-' || argv[j][1] != '-') {
- configfile = argv[j];
- server.configfile = getAbsolutePath(configfile);
- /* Replace the config file in server.exec_argv with
- * its absolute path. */
- zfree(server.exec_argv[j]);
- server.exec_argv[j] = zstrdup(server.configfile);
- j++;
- }
- /* All the other options are parsed and conceptually appended to the
- * configuration file. For instance --port 6380 will generate the
- * string "port 6380\n" to be parsed after the actual file name
- * is parsed, if any. */
- while(j != argc) {// 解析除配置文件外的其它参数
- if (argv[j][0] == '-' && argv[j][1] == '-') {
- /* Option name */
- if (!strcmp(argv[j], "--check-rdb")) {
- /* Argument has no options, need to skip for parsing. */
- j++;
- continue;
- }
- if (sdslen(options)) options = sdscat(options,"\n");
- options = sdscat(options,argv[j]+2);// 去掉参数前面的 --
- options = sdscat(options,"");// 参数对应的值和参数名应" " 分隔
- } else {
- /* Option argument */
- options = sdscatrepr(options,argv[j],strlen(argv[j]));
- options = sdscat(options," ");
- }
- j++;
- }
- // 从指定配置文件和命令选项设置服务器对象 server 参数, 覆盖默认配置
- loadServerConfig(configfile,options);
- }
- server.supervised = redisIsSupervised(server.supervised_mode);
- int background = server.daemonize && !server.supervised;
- if (background) daemonize();// 后台进程模式
- // 初始化服务器功能, 包括时间事件 1ms 调用 serverCron, 文件事件 (套接字可读可写时的处理函数), 集群初始化等
- initServer();
- aeSetBeforeSleepProc(server.el,beforeSleep);// 设置 beforeSleep 事件处理函数
- aeSetAfterSleepProc(server.el,afterSleep);// 设置 aftersleep 事件处理函数
- aeMain(server.el);// 循环, 接受客户端连接, 处理命令等
- aeDeleteEventLoop(server.el);// 退出循环, 删除事件处理
- return 0;
- }
initServer 函数源码如下, 主要对 server 中的变量进行初始化, 其中 listenToPort 根据监听的地址和端口, 设置 server.ipfd(监听套接字数组) 和 server.ipfd_count(监听套接字数目),85 行 aeCreateTimeEvent 设置定时器事件, 每 1ms 执行 serverCron 函数, 94 行 aeCreateFileEvent 函数设置监听套接字有客户端连接时执行的事件处理函数 acceptTcpHandler.
- void initServer(void) {
- int j;
- server.hz = server.config_hz;
- server.pid = getpid();
- server.current_client = NULL;
- server.clients = listCreate();
- server.clients_index = raxNew();
- server.clients_to_close = listCreate();
- server.slaves = listCreate();
- server.monitors = listCreate();
- server.clients_pending_write = listCreate();
- server.slaveseldb = -1; /* Force to emit the first SELECT command. */
- server.unblocked_clients = listCreate();
- server.ready_keys = listCreate();
- server.clients_waiting_acks = listCreate();
- server.get_ack_from_slaves = 0;
- server.clients_paused = 0;
- server.system_memory_size = zmalloc_get_memory_size();
- // 创建常用字符串, 比如命令回复 pong 及命令字符串等
- createSharedObjects();
- // 根据配置的最大连接客户端数和 CONFIG_MIN_RESERVED_FDS(用于监听, 日志文件等文件描述符) 设置最大允许打开的文件描述符数
- adjustOpenFilesLimit();
- // 初始化 aeEventLoop 对象并赋值给 server 中的指针变量 el
- server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
- server.db = zmalloc(sizeof(redisDb)*server.dbnum);
- /* Open the TCP listening socket for the user commands. */
- // 用于创建监听文件描述符
- if (server.port != 0 &&
- listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)
- exit(1);
- /* Create the Redis databases, and initialize other internal state. */
- for (j = 0; j <server.dbnum; j++) {
- server.db[j].dict = dictCreate(&dbDictType,NULL);
- server.db[j].expires = dictCreate(&keyptrDictType,NULL);
- server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
- server.db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType,NULL);
- server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);
- server.db[j].id = j;
- server.db[j].avg_ttl = 0;
- server.db[j].defrag_later = listCreate();
- }
- // 创建 lru 池, 内存超过设置的值时, 根据淘汰策略, 清除部分 key
- evictionPoolAlloc(); /* Initialize the LRU keys pool. */
- server.pubsub_channels = dictCreate(&keylistDictType,NULL);
- server.pubsub_patterns = listCreate();
- listSetFreeMethod(server.pubsub_patterns,freePubsubPattern);
- listSetMatchMethod(server.pubsub_patterns,listMatchPubsubPattern);
- server.cronloops = 0;
- server.rdb_child_pid = -1;
- server.aof_child_pid = -1;
- server.rdb_child_type = RDB_CHILD_TYPE_NONE;
- server.rdb_bgsave_scheduled = 0;
- server.child_info_pipe[0] = -1;
- server.child_info_pipe[1] = -1;
- server.child_info_data.magic = 0;
- aofRewriteBufferReset();
- server.aof_buf = sdsempty();
- server.lastsave = time(NULL); /* At startup we consider the DB saved. */
- server.lastbgsave_try = 0; /* At startup we never tried to BGSAVE. */
- server.rdb_save_time_last = -1;
- server.rdb_save_time_start = -1;
- server.dirty = 0;
- resetServerStats();
- /* A few stats we don't want to reset: server startup time, and peak mem. */
- server.stat_starttime = time(NULL);
- server.stat_peak_memory = 0;
- server.stat_rdb_cow_bytes = 0;
- server.stat_aof_cow_bytes = 0;
- server.cron_malloc_stats.zmalloc_used = 0;
- server.cron_malloc_stats.process_rss = 0;
- server.cron_malloc_stats.allocator_allocated = 0;
- server.cron_malloc_stats.allocator_active = 0;
- server.cron_malloc_stats.allocator_resident = 0;
- server.lastbgsave_status = C_OK;
- server.aof_last_write_status = C_OK;
- server.aof_last_write_errno = 0;
- server.repl_good_slaves_count = 0;
- /* Create the timer callback, this is our way to process many background
- * operations incrementally, like clients timeout, eviction of unaccessed
- * expired keys and so forth. */
- // 创建时间事件, 1ms 调用 serverCron
- if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
- serverPanic("Can't create event loop timers.");
- exit(1);
- }
- /* Create an event handler for accepting new connections in TCP and Unix
- * domain sockets. */
- // 设定套接字可读可写状态的处理函数
- 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.");
- }
- }
- }
listenToPort 函数会对所有配置需要监听的 ip 地址和端口调用_anetTcpServer 函数 (.NET.c), _anetTcpServer 函数源码如下, 1-17 行建立 socket 套接字 s,23 行调用函数 anetListen 完成服务端 socket 的 bind 和 listen.
- 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;
- snprintf(_port,6,"%d",port);
- memset(&hints,0,sizeof(hints));
- hints.ai_family = af;
- hints.ai_socktype = SOCK_STREAM;
- hints.ai_flags = AI_PASSIVE; /* No effect if bindaddr != NULL */
- if ((rv = getaddrinfo(bindaddr,_port,&hints,&servinfo)) != 0) {
- anetSetError(err, "%s", gai_strerror(rv));
- return ANET_ERR;
- }
- for (p = servinfo; p != NULL; p = p->ai_next) {
- if ((s = socket(p->ai_family,p->ai_socktype,p->ai_protocol)) == -1)
- continue;
- if (af == AF_INET6 && anetV6Only(err,s) == ANET_ERR) goto error;
- if (anetSetReuseAddr(err,s) == ANET_ERR) goto error;
- //anetListen 调用 listen 和 bind
- if (anetListen(err,s,p->ai_addr,p->ai_addrlen,backlog) == ANET_ERR) s = ANET_ERR;
- goto end;
- }
- if (p == NULL) {
- anetSetError(err, "unable to bind socket, errno: %d", errno);
- goto error;
- }
- error:
- if (s != -1) close(s);
- s = ANET_ERR;
- end:
- freeaddrinfo(servinfo);
- return s;
- }
anetListen 函数源码如下:
- static int anetListen(char *err, int s, struct sockaddr *sa, socklen_t len, int backlog) {
- if (bind(s,sa,len) == -1) {
- anetSetError(err, "bind: %s", strerror(errno));
- close(s);
- return ANET_ERR;
- }
- if (listen(s, backlog) == -1) {
- anetSetError(err, "listen: %s", strerror(errno));
- close(s);
- return ANET_ERR;
- }
- return ANET_OK;
- }
在分析 aeCreateTimeEvent 和 aeCreateFileEvent 函数之前, 先给出 server 中的 aeEventLoop 类型变量 el 结构体定义 (ae.h):
- /* 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. */
- long when_sec; /* seconds */
- long when_ms; /* milliseconds */
- // 事件对应的处理程序
- aeTimeProc *timeProc;
- // 时间事件的最后一次处理程序, 若已设置, 则删除时间事件时会被调用
- aeEventFinalizerProc *finalizerProc;
- void *clientData;
- struct aeTimeEvent *prev;
- struct aeTimeEvent *next;
- } 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 */
- // 文件描述符的最大监听数
- int setsize; /* max number of file descriptors tracked */
- // 生成时间事件的唯一标识
- long long timeEventNextId;
- // 检测系统时钟偏差
- time_t lastTime; /* Used to detect system clock skew */
- // 双向链表, 注册文件事件
- aeFileEvent *events; /* Registered events */
- // 已触发的文件事件
- aeFiredEvent *fired; /* Fired events */
- // 注册的时间事件
- aeTimeEvent *timeEventHead;
- int stop;
- // 处理底层特定 API 的数据, 对于 epoll 来说, 该结构体包含了 epoll fd 和 epoll_event
- void *apidata; /* This is used for polling API specific data */
- aeBeforeSleepProc *beforesleep;
- aeBeforeSleepProc *aftersleep;
- } aeEventLoop;
函数 aeCreateTimeEvent(ae.c) 用于往 server 中的变量 el 时间事件队列添加元素, 源码如下:
- 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->prev = NULL;
- te->next = eventLoop->timeEventHead;
- if (te->next)
- te->next->prev = te;
- eventLoop->timeEventHead = te;
- return id;
- }
aeCreateFileEvent(ae.c) 函数用于 server.el 增加文件事件队列添加元素, 源码如下:
- static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
- aeApiState *state = eventLoop->apidata;
- if (mask & AE_READABLE) FD_SET(fd,&state->rfds);
- if (mask & AE_WRITABLE) FD_SET(fd,&state->wfds);
- return 0;
- }
- int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
- aeFileProc *proc, void *clientData)
- {
- if (fd>= eventLoop->setsize) {
- errno = ERANGE;
- return AE_ERR;
- }
- aeFileEvent *fe = &eventLoop->events[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;
- }
执行完 initServer 及 aeSetBeforeSleepProc 和 aeSetAfterSleepProc 函数设置好全局变量 server 后, 进程调用 aeMain 函数进入循环, 开始接收客户端连接和命令交互. aeMain(ae.c) 函数源码如下:
- // 处理客户端命令的主函数, 由 main 函数调用, aeEventLoop 定义在 (ae.h),beforesleep 和 aftersleep 均在服务器启动初始化时设定
- void aeMain(aeEventLoop *eventLoop) {
- eventLoop->stop = 0;
- while (!eventLoop->stop) {
- if (eventLoop->beforesleep != NULL)
- eventLoop->beforesleep(eventLoop);
- // 处理文件时间及时间事件
- aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);
- }
- }
来源: https://www.cnblogs.com/killianxu/p/10556076.html