redis启动过程源码解析

//整个程序的入口函数
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[]);
server.exec_argv = zmalloc(sizeof(char*)*(argc+));
server.exec_argv[argc] = NULL;
for (j = ; j < argc; j++) server.exec_argv[j] = zstrdup(argv[j]); if (argc >= ) {
j = ; /* First option to parse in argv[] */
sds options = sdsempty();
char *configfile = NULL; //命令指定了配置文件,对配置文件做处理,配置文件跟在程序名后第一位
/* First argument is the config file name? */
if (argv[j][] != '-' || argv[j][] != '-') {
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][] == '-' && argv[j][] == '-') {
/* 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]+);//去掉参数前面的--
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 ;
}

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 = -; /* 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 = ;
server.clients_paused = ;
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 != &&
listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)
exit(); /* Create the Redis databases, and initialize other internal state. */
for (j = ; 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 = ;
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 = ;
server.rdb_child_pid = -;
server.aof_child_pid = -;
server.rdb_child_type = RDB_CHILD_TYPE_NONE;
server.rdb_bgsave_scheduled = ;
server.child_info_pipe[] = -;
server.child_info_pipe[] = -;
server.child_info_data.magic = ;
aofRewriteBufferReset();
server.aof_buf = sdsempty();
server.lastsave = time(NULL); /* At startup we consider the DB saved. */
server.lastbgsave_try = ; /* At startup we never tried to BGSAVE. */
server.rdb_save_time_last = -;
server.rdb_save_time_start = -;
server.dirty = ;
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 = ;
server.stat_rdb_cow_bytes = ;
server.stat_aof_cow_bytes = ;
server.cron_malloc_stats.zmalloc_used = ;
server.cron_malloc_stats.process_rss = ;
server.cron_malloc_stats.allocator_allocated = ;
server.cron_malloc_stats.allocator_active = ;
server.cron_malloc_stats.allocator_resident = ;
server.lastbgsave_status = C_OK;
server.aof_last_write_status = C_OK;
server.aof_last_write_errno = ;
server.repl_good_slaves_count = ; /* 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, , serverCron, NULL, NULL) == AE_ERR) {
serverPanic("Can't create event loop timers.");
exit();
} /* Create an event handler for accepting new connections in TCP and Unix
* domain sockets. */
//设定套接字可读可写状态的处理函数
for (j = ; 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.");
}
}
}

static int _anetTcpServer(char *err, int port, char *bindaddr, int af, int backlog)
{
int s = -, rv;
char _port[]; /* strlen("65535") */
struct addrinfo hints, *servinfo, *p;
snprintf(_port,,"%d",port);
memset(&hints,,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)) != ) {
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)) == -)
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 != -) close(s);
s = ANET_ERR;
end:
freeaddrinfo(servinfo);
return s;
}

static int anetListen(char *err, int s, struct sockaddr *sa, socklen_t len, int backlog) {
if (bind(s,sa,len) == -) {
anetSetError(err, "bind: %s", strerror(errno));
close(s);
return ANET_ERR;
} if (listen(s, backlog) == -) {
anetSetError(err, "listen: %s", strerror(errno));
close(s);
return ANET_ERR;
}
return ANET_OK;
}

/* 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;

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;
}

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 ;
}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) == -)
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;
}

//处理客户端命令的主函数，由main函数调用,aeEventLoop定义在(ae.h),beforesleep和aftersleep均在服务器启动初始化时设定
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = ;
while (!eventLoop->stop) {
if (eventLoop->beforesleep != NULL)
eventLoop->beforesleep(eventLoop);
//处理文件时间及时间事件
aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);
}
}