欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

redis启动过程源码解析

程序员文章站 2022-05-03 11:57:39
大部分人在用开源软件的时候,并不知道一个优秀的开源软件内部是如何启动起来的,本文通过分析redis启动过程的源码,来学习优秀软件的启动过程 ......

  redis整个程序的入口函数在server.c中的main函数,函数调用关系如下图1,调用顺序为从上到下,从左至右。

redis启动过程源码解析

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