redis启动过程源码解析
redis整个程序的入口函数在server.c中的main函数,函数调用关系如下图1,调用顺序为从上到下,从左至右。
图1 redis启动函数调用图
main函数源码如下,1-55行根据配置文件和启动命令参数设置全局对象server ,57-59设置redis的服务器端为后台进程, initserver主要提前创建一些经常用到的对象用于节约内存,根据设置的ip地址和端口创建监听套接字用于客户端连接,并初始化时间事件,64行用于设置server->el ->beforesleep = beforesleep,aemain函数是个循环函数,用于监听客户端连接,接收客户端命令并进行处理等。
1 //整个程序的入口函数 2 int main(int argc, char **argv) { 3 //初始化服务器配置,设置全局对象server的状态 4 initserverconfig(); 5 6 /* store the executable path and arguments in a safe place in order 7 * to be able to restart the server later. */ 8 //存储运行命令的绝对路径及运行参数 9 server.executable = getabsolutepath(argv[0]); 10 server.exec_argv = zmalloc(sizeof(char*)*(argc+1)); 11 server.exec_argv[argc] = null; 12 for (j = 0; j < argc; j++) server.exec_argv[j] = zstrdup(argv[j]); 13 14 if (argc >= 2) { 15 j = 1; /* first option to parse in argv[] */ 16 sds options = sdsempty(); 17 char *configfile = null; 18 19 //命令指定了配置文件,对配置文件做处理,配置文件跟在程序名后第一位 20 /* first argument is the config file name? */ 21 if (argv[j][0] != '-' || argv[j][1] != '-') { 22 configfile = argv[j]; 23 server.configfile = getabsolutepath(configfile); 24 /* replace the config file in server.exec_argv with 25 * its absolute path. */ 26 zfree(server.exec_argv[j]); 27 server.exec_argv[j] = zstrdup(server.configfile); 28 j++; 29 } 30 31 /* all the other options are parsed and conceptually appended to the 32 * configuration file. for instance --port 6380 will generate the 33 * string "port 6380\n" to be parsed after the actual file name 34 * is parsed, if any. */ 35 while(j != argc) {//解析除配置文件外的其它参数 36 if (argv[j][0] == '-' && argv[j][1] == '-') { 37 /* option name */ 38 if (!strcmp(argv[j], "--check-rdb")) { 39 /* argument has no options, need to skip for parsing. */ 40 j++; 41 continue; 42 } 43 if (sdslen(options)) options = sdscat(options,"\n"); 44 options = sdscat(options,argv[j]+2);//去掉参数前面的-- 45 options = sdscat(options," ");//参数对应的值和参数名应" "分隔 46 } else { 47 /* option argument */ 48 options = sdscatrepr(options,argv[j],strlen(argv[j])); 49 options = sdscat(options," "); 50 } 51 j++; 52 } 53 //从指定配置文件和命令选项设置服务器对象server参数,覆盖默认配置 54 loadserverconfig(configfile,options); 55 } 56 57 server.supervised = redisissupervised(server.supervised_mode); 58 int background = server.daemonize && !server.supervised; 59 if (background) daemonize();//后台进程模式 60 61 //初始化服务器功能,包括时间事件1ms调用servercron,文件事件(套接字可读可写时的处理函数),集群初始化等 62 initserver(); 63 aesetbeforesleepproc(server.el,beforesleep);//设置beforesleep事件处理函数 64 aesetaftersleepproc(server.el,aftersleep);//设置aftersleep事件处理函数 65 aemain(server.el);//循环,接受客户端连接,处理命令等 66 aedeleteeventloop(server.el);//退出循环,删除事件处理 67 return 0; 68 }
initserver函数源码如下,主要对server中的变量进行初始化,其中listentoport根据监听的地址和端口, 设置server.ipfd(监听套接字数组)和server.ipfd_count(监听套接字数目),85行aecreatetimeevent设置定时器事件,每1ms执行servercron函数,94行aecreatefileevent函数设置监听套接字有客户端连接时执行的事件处理函数accepttcphandler。
1 void initserver(void) { 2 int j; 3 server.hz = server.config_hz; 4 server.pid = getpid(); 5 server.current_client = null; 6 server.clients = listcreate(); 7 server.clients_index = raxnew(); 8 server.clients_to_close = listcreate(); 9 server.slaves = listcreate(); 10 server.monitors = listcreate(); 11 server.clients_pending_write = listcreate(); 12 server.slaveseldb = -1; /* force to emit the first select command. */ 13 server.unblocked_clients = listcreate(); 14 server.ready_keys = listcreate(); 15 server.clients_waiting_acks = listcreate(); 16 server.get_ack_from_slaves = 0; 17 server.clients_paused = 0; 18 server.system_memory_size = zmalloc_get_memory_size(); 19 //创建常用字符串,比如命令回复pong及命令字符串等 20 createsharedobjects(); 21 //根据配置的最大连接客户端数和config_min_reserved_fds(用于监听、日志文件等文件描述符)设置最大允许打开的文件描述符数 22 adjustopenfileslimit(); 23 //初始化aeeventloop对象并赋值给server中的指针变量el 24 server.el = aecreateeventloop(server.maxclients+config_fdset_incr); 25 server.db = zmalloc(sizeof(redisdb)*server.dbnum); 26 27 /* open the tcp listening socket for the user commands. */ 28 //用于创建监听文件描述符 29 if (server.port != 0 && 30 listentoport(server.port,server.ipfd,&server.ipfd_count) == c_err) 31 exit(1); 32 33 /* create the redis databases, and initialize other internal state. */ 34 for (j = 0; j < server.dbnum; j++) { 35 server.db[j].dict = dictcreate(&dbdicttype,null); 36 server.db[j].expires = dictcreate(&keyptrdicttype,null); 37 server.db[j].blocking_keys = dictcreate(&keylistdicttype,null); 38 server.db[j].ready_keys = dictcreate(&objectkeypointervaluedicttype,null); 39 server.db[j].watched_keys = dictcreate(&keylistdicttype,null); 40 server.db[j].id = j; 41 server.db[j].avg_ttl = 0; 42 server.db[j].defrag_later = listcreate(); 43 } 44 //创建lru池,内存超过设置的值时,根据淘汰策略,清除部分key 45 evictionpoolalloc(); /* initialize the lru keys pool. */ 46 server.pubsub_channels = dictcreate(&keylistdicttype,null); 47 server.pubsub_patterns = listcreate(); 48 listsetfreemethod(server.pubsub_patterns,freepubsubpattern); 49 listsetmatchmethod(server.pubsub_patterns,listmatchpubsubpattern); 50 server.cronloops = 0; 51 server.rdb_child_pid = -1; 52 server.aof_child_pid = -1; 53 server.rdb_child_type = rdb_child_type_none; 54 server.rdb_bgsave_scheduled = 0; 55 server.child_info_pipe[0] = -1; 56 server.child_info_pipe[1] = -1; 57 server.child_info_data.magic = 0; 58 aofrewritebufferreset(); 59 server.aof_buf = sdsempty(); 60 server.lastsave = time(null); /* at startup we consider the db saved. */ 61 server.lastbgsave_try = 0; /* at startup we never tried to bgsave. */ 62 server.rdb_save_time_last = -1; 63 server.rdb_save_time_start = -1; 64 server.dirty = 0; 65 resetserverstats(); 66 /* a few stats we don't want to reset: server startup time, and peak mem. */ 67 server.stat_starttime = time(null); 68 server.stat_peak_memory = 0; 69 server.stat_rdb_cow_bytes = 0; 70 server.stat_aof_cow_bytes = 0; 71 server.cron_malloc_stats.zmalloc_used = 0; 72 server.cron_malloc_stats.process_rss = 0; 73 server.cron_malloc_stats.allocator_allocated = 0; 74 server.cron_malloc_stats.allocator_active = 0; 75 server.cron_malloc_stats.allocator_resident = 0; 76 server.lastbgsave_status = c_ok; 77 server.aof_last_write_status = c_ok; 78 server.aof_last_write_errno = 0; 79 server.repl_good_slaves_count = 0; 80 81 /* create the timer callback, this is our way to process many background 82 * operations incrementally, like clients timeout, eviction of unaccessed 83 * expired keys and so forth. */ 84 //创建时间事件,1ms调用servercron 85 if (aecreatetimeevent(server.el, 1, servercron, null, null) == ae_err) { 86 serverpanic("can't create event loop timers."); 87 exit(1); 88 } 89 90 /* create an event handler for accepting new connections in tcp and unix 91 * domain sockets. */ 92 //设定套接字可读可写状态的处理函数 93 for (j = 0; j < server.ipfd_count; j++) { 94 if (aecreatefileevent(server.el, server.ipfd[j], ae_readable, 95 accepttcphandler,null) == ae_err) 96 { 97 serverpanic( 98 "unrecoverable error creating server.ipfd file event."); 99 } 100 } 101 }
listentoport函数会对所有配置需要监听的ip地址和端口调用_anettcpserver函数(anet.c), _anettcpserver函数源码如下,1-17行建立socket套接字s,23行调用函数anetlisten完成服务端socket的bind和listen。
1 static int _anettcpserver(char *err, int port, char *bindaddr, int af, int backlog) 2 { 3 int s = -1, rv; 4 char _port[6]; /* strlen("65535") */ 5 struct addrinfo hints, *servinfo, *p; 6 snprintf(_port,6,"%d",port); 7 memset(&hints,0,sizeof(hints)); 8 hints.ai_family = af; 9 hints.ai_socktype = sock_stream; 10 hints.ai_flags = ai_passive; /* no effect if bindaddr != null */ 11 12 if ((rv = getaddrinfo(bindaddr,_port,&hints,&servinfo)) != 0) { 13 anetseterror(err, "%s", gai_strerror(rv)); 14 return anet_err; 15 } 16 for (p = servinfo; p != null; p = p->ai_next) { 17 if ((s = socket(p->ai_family,p->ai_socktype,p->ai_protocol)) == -1) 18 continue; 19 20 if (af == af_inet6 && anetv6only(err,s) == anet_err) goto error; 21 if (anetsetreuseaddr(err,s) == anet_err) goto error; 22 //anetlisten调用listen和bind 23 if (anetlisten(err,s,p->ai_addr,p->ai_addrlen,backlog) == anet_err) s = anet_err; 24 goto end; 25 } 26 if (p == null) { 27 anetseterror(err, "unable to bind socket, errno: %d", errno); 28 goto error; 29 } 30 31 error: 32 if (s != -1) close(s); 33 s = anet_err; 34 end: 35 freeaddrinfo(servinfo); 36 return s; 37 }
anetlisten函数源码如下:
1 static int anetlisten(char *err, int s, struct sockaddr *sa, socklen_t len, int backlog) { 2 if (bind(s,sa,len) == -1) { 3 anetseterror(err, "bind: %s", strerror(errno)); 4 close(s); 5 return anet_err; 6 } 7 8 if (listen(s, backlog) == -1) { 9 anetseterror(err, "listen: %s", strerror(errno)); 10 close(s); 11 return anet_err; 12 } 13 return anet_ok; 14 }
在分析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时间事件队列添加元素,源码如下:
1 long long aecreatetimeevent(aeeventloop *eventloop, long long milliseconds, 2 aetimeproc *proc, void *clientdata, 3 aeeventfinalizerproc *finalizerproc) 4 { 5 long long id = eventloop->timeeventnextid++; 6 aetimeevent *te; 7 8 te = zmalloc(sizeof(*te)); 9 if (te == null) return ae_err; 10 te->id = id; 11 aeaddmillisecondstonow(milliseconds,&te->when_sec,&te->when_ms); 12 te->timeproc = proc; 13 te->finalizerproc = finalizerproc; 14 te->clientdata = clientdata; 15 te->prev = null; 16 te->next = eventloop->timeeventhead; 17 if (te->next) 18 te->next->prev = te; 19 eventloop->timeeventhead = te; 20 return id; 21 }
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)函数源码如下:
1 //处理客户端命令的主函数,由main函数调用,aeeventloop定义在(ae.h),beforesleep和aftersleep均在服务器启动初始化时设定 2 void aemain(aeeventloop *eventloop) { 3 eventloop->stop = 0; 4 while (!eventloop->stop) { 5 if (eventloop->beforesleep != null) 6 eventloop->beforesleep(eventloop); 7 //处理文件时间及时间事件 8 aeprocessevents(eventloop, ae_all_events|ae_call_after_sleep); 9 } 10 }
推荐阅读
-
微信小程序之侧边栏滑动实现过程解析(附完整源码)
-
深入解析vue 源码目录及构建过程分析
-
基于Spring注解的上下文初始化过程源码解析(一)
-
Tomcat源码分析三:Tomcat启动加载过程(一)的源码解析
-
spring5 源码深度解析-----ApplicationContext容器refresh过程
-
Spring MVC源码(一) ----- 启动过程与组件初始化
-
Mybaits 源码解析 (六)----- 全网最详细:Select 语句的执行过程分析(上篇)(Mapper方法是如何调用到XML中的SQL的?)
-
jQuery选择器源码解读(五):tokenize的解析过程
-
微信小程序之侧边栏滑动实现过程解析(附完整源码)
-
Tomcat源码分析 (六)----- Tomcat 启动过程(一)