大部分人在用開源軟體的時候,並不知道一個優秀的開源軟體內部是如何啟動起來的,本文通過分析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 }
