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 }
原文地址：https://www.cnblogs.com/killianxu/p/10556076.html
时间： 2024-10-29 16:25:33