深入redis内部--初始化服务器
初始化服务器代码如下：void initServer() {
int j;
signal(SIGHUP, SIG_IGN);
signal(SIGPIPE, SIG_IGN);
setupSignalHandlers();
if (server.syslog_enabled) {
openlog(server.syslog_ident, LOG_PID | LOG_NDELAY | LOG_NOWAIT,
server.syslog_facility);
}
server.current_client = NULL;
server.clients = listCreate(); //创建客户队列
server.clients_to_close = listCreate(); //创建将关闭的客户队列
server.slaves = listCreate(); //创建从机队列
server.monitors = listCreate(); //创建监控队列
server.slaveseldb = -1; /* Force to emit the first SELECT command. */
server.unblocked_clients = listCreate();//创建非堵塞客户队列
server.ready_keys = listCreate(); //创建可读key队列
createSharedObjects(); // 创建共享对象
adjustOpenFilesLimit(); //改变可打开文件的最大数量
server.el = aeCreateEventLoop(server.maxclients+REDIS_EVENTLOOP_FDSET_INCR);//创建事件处理
server.db = zmalloc(sizeof(redisDb)*server.dbnum);//分别db内存
/* Open the TCP listening socket for the user commands. */
if (listenToPort(server.port,server.ipfd,&server.ipfd_count) == REDIS_ERR)//监听端口
exit(1);
/* Open the listening Unix domain socket. */
if (server.unixsocket != NULL) {
unlink(server.unixsocket); /* don't care if this fails */
server.sofd = anetUnixServer(server.neterr,server.unixsocket,server.unixsocketperm);
if (server.sofd == ANET_ERR) {
redisLog(REDIS_WARNING, "Opening socket: %s", server.neterr);
exit(1);
}
}
/* Abort if there are no listening sockets at all. */
if (server.ipfd_count == 0 && server.sofd < 0) {
redisLog(REDIS_WARNING, "Configured to not listen anywhere, exiting.");
exit(1);
}
/* Create the Redis databases, and initialize other internal state. */
for (j = 0; j < server.dbnum; j++) {
server.db.dict = dictCreate(&dbDictType,NULL);
server.db.expires = dictCreate(&keyptrDictType,NULL);
server.db.blocking_keys = dictCreate(&keylistDictType,NULL);
server.db.ready_keys = dictCreate(&setDictType,NULL);
server.db.watched_keys = dictCreate(&keylistDictType,NULL);
server.db.id = j;
server.db.avg_ttl = 0;
}
server.pubsub_channels = dictCreate(&keylistDictType,NULL);
server.pubsub_patterns = listCreate();
listSetFreeMethod(server.pubsub_patterns,freePubsubPattern);
listSetMatchMethod(server.pubsub_patterns,listMatchPubsubPattern);
server.cronloops = 0;
server.rdb_child_pid = -1;
server.aof_child_pid = -1;
aofRewriteBufferReset();
server.aof_buf = sdsempty();
server.lastsave = time(NULL); /* At startup we consider the DB saved. */
server.lastbgsave_try = 0; /* At startup we never tried to BGSAVE. */
server.rdb_save_time_last = -1;
server.rdb_save_time_start = -1;
server.dirty = 0;
server.stat_numcommands = 0;
server.stat_numconnections = 0;
server.stat_expiredkeys = 0;
server.stat_evictedkeys = 0;
server.stat_starttime = time(NULL);
server.stat_keyspace_misses = 0;
server.stat_keyspace_hits = 0;
server.stat_peak_memory = 0;
server.stat_fork_time = 0;
server.stat_rejected_conn = 0;
server.stat_sync_full = 0;
server.stat_sync_partial_ok = 0;
server.stat_sync_partial_err = 0;
memset(server.ops_sec_samples,0,sizeof(server.ops_sec_samples));
server.ops_sec_idx = 0;
server.ops_sec_last_sample_time = mstime();
server.ops_sec_last_sample_ops = 0;
server.unixtime = time(NULL);
server.mstime = mstime();
server.lastbgsave_status = REDIS_OK;
server.repl_good_slaves_count = 0;
/* Create the serverCron() time event, that's our main way to process
* background operations. */
if(aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
redisPanic("Can't create the serverCron time event.");
exit(1);
}
/* Create an event handler for accepting new connections in TCP and Unix
* domain sockets. */
for (j = 0; j < server.ipfd_count; j++) {
if (aeCreateFileEvent(server.el, server.ipfd, AE_READABLE,
acceptTcpHandler,NULL) == AE_ERR)
{
redisPanic(
"Unrecoverable error creating server.ipfd file event.");
}
}
if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
acceptUnixHandler,NULL) == AE_ERR) redisPanic("Unrecoverable error creating server.sofd file event.");
/* Open the AOF file if needed. */
if (server.aof_state == REDIS_AOF_ON) {
server.aof_fd = open(server.aof_filename,
O_WRONLY|O_APPEND|O_CREAT,0644);
if (server.aof_fd == -1) {
redisLog(REDIS_WARNING, "Can't open the append-only file: %s",
strerror(errno));
exit(1);
}
}
/* 32 bit instances are limited to 4GB of address space, so if there is
* no explicit limit in the user provided configuration we set a limit
* at 3 GB using maxmemory with 'noeviction' policy'. This avoids
* useless crashes of the Redis instance for out of memory. */
if (server.arch_bits == 32 && server.maxmemory == 0) {
redisLog(REDIS_WARNING,"Warning: 32 bit instance detected but no memory limit set. Setting 3 GB maxmemory limit with 'noeviction' policy now.");
server.maxmemory = 3072LL*(1024*1024); /* 3 GB */
server.maxmemory_policy = REDIS_MAXMEMORY_NO_EVICTION;
}
replicationScriptCacheInit();
scriptingInit();
slowlogInit();
bioInit();
}
1.1 信号处理
signal(SIGHUP, SIG_IGN);
signal(SIGPIPE, SIG_IGN);
setupSignalHandlers();
signal语法：
#include
void (*signal(int sig, void (*func)(int)))(int);
int sighold(int sig);
int sigignore(int sig);
int sigpause(int sig);
int sigrelse(int sig);
void (*sigset(int sig, void (*disp)(int)))(int);
signal变量定义在signal.h文件中，其中：
1.信号
Signal
Description
SIGABRT
由调用abort函数产生，进程非正常退出
SIGALRM
用alarm函数设置的timer超时或setitimer函数设置的interval timer超时
SIGBUS
某种特定的硬件异常，通常由内存访问引起
SIGCANCEL
由Solaris Thread Library内部使用，通常不会使用
SIGCHLD
进程Terminate或Stop的时候，SIGCHLD会发送给它的父进程。缺省情况下该Signal会被忽略
SIGCONT
当被stop的进程恢复运行的时候，自动发送
SIGEMT
和实现相关的硬件异常
SIGFPE
数学相关的异常，如被0除，浮点溢出，等等
SIGFREEZE
Solaris专用，Hiberate或者Suspended时候发送
SIGHUP
发送给具有Terminal的Controlling Process，当terminal被disconnect时候发送
SIGILL
非法指令异常
SIGINFO
BSD signal。由Status Key产生，通常是CTRL+T。发送给所有Foreground Group的进程
SIGINT
由Interrupt Key产生，通常是CTRL+C或者DELETE。发送给所有ForeGround Group的进程
SIGIO
异步IO事件
SIGIOT
实现相关的硬件异常，一般对应SIGABRT
SIGKILL
无法处理和忽略。中止某个进程
SIGLWP
由Solaris Thread Libray内部使用
SIGPIPE
在reader中止之后写Pipe的时候发送
SIGPOLL
当某个事件发送给Pollable Device的时候发送
SIGPROF
Setitimer指定的Profiling Interval Timer所产生
SIGPWR
和系统相关。和UPS相关。
SIGQUIT
输入Quit Key的时候（CTRL+/）发送给所有Foreground Group的进程
SIGSEGV
非法内存访问
SIGSTKFLT
Linux专用，数学协处理器的栈异常
SIGSTOP
中止进程。无法处理和忽略。
SIGSYS
非法系统调用
SIGTERM
请求中止进程，kill命令缺省发送
SIGTHAW
Solaris专用，从Suspend恢复时候发送
SIGTRAP
实现相关的硬件异常。一般是调试异常
SIGTSTP
Suspend Key，一般是Ctrl+Z。发送给所有Foreground Group的进程
SIGTTIN
当Background Group的进程尝试读取Terminal的时候发送
SIGTTOU
当Background Group的进程尝试写Terminal的时候发送
SIGURG
当out-of-band data接收的时候可能发送
SIGUSR1
用户自定义signal 1
SIGUSR2
用户自定义signal 2
SIGVTALRM
setitimer函数设置的Virtual Interval Timer超时的时候
SIGWAITING
Solaris Thread Library内部实现专用
SIGWINCH
当Terminal的窗口大小改变的时候，发送给Foreground Group的所有进程
SIGXCPU
当CPU时间限制超时的时候
SIGXFSZ
进程超过文件大小限制
SIGXRES
Solaris专用，进程超过资源限制的时候发送
================================================================
2.处理函数
SIG_DFL Request for default signal handling. SIG_ERR Return value from signal() in case of error. SIG_HOLD Request that signal be held. SIG_IGN Request that signal be ignored. 3. 建立信号处理函数
void setupSignalHandlers(void) {
struct sigaction act;
/* When the SA_SIGINFO flag is set in sa_flags then sa_sigaction is used.
* Otherwise, sa_handler is used. */
sigemptyset(&act.sa_mask);
act.sa_flags = 0;
act.sa_handler = sigtermHandler; //信号中断处理函数，使用 sigaction(SIGTERM, &act, NULL);
sigaction(SIGTERM, &act, NULL);
#ifdef HAVE_BACKTRACE
sigemptyset(&act.sa_mask);
act.sa_flags = SA_NODEFER | SA_RESETHAND | SA_SIGINFO;
act.sa_sigaction = sigsegvHandler;
sigaction(SIGSEGV, &act, NULL);
sigaction(SIGBUS, &act, NULL);
sigaction(SIGFPE, &act, NULL);
sigaction(SIGILL, &act, NULL);
#endif
return;
}
1.2 创建共享对象
void createSharedObjects(void) {
int j;
shared.crlf = createObject(REDIS_STRING,sdsnew("\r\n"));
shared.ok = createObject(REDIS_STRING,sdsnew("+OK\r\n"));
shared.err = createObject(REDIS_STRING,sdsnew("-ERR\r\n"));
shared.emptybulk = createObject(REDIS_STRING,sdsnew("$0\r\n\r\n"));
shared.czero = createObject(REDIS_STRING,sdsnew(":0\r\n"));
shared.cone = createObject(REDIS_STRING,sdsnew(":1\r\n"));
shared.cnegone = createObject(REDIS_STRING,sdsnew(":-1\r\n"));
shared.nullbulk = createObject(REDIS_STRING,sdsnew("$-1\r\n"));
shared.nullmultibulk = createObject(REDIS_STRING,sdsnew("*-1\r\n"));
shared.emptymultibulk = createObject(REDIS_STRING,sdsnew("*0\r\n"));
shared.pong = createObject(REDIS_STRING,sdsnew("+PONG\r\n"));
shared.queued = createObject(REDIS_STRING,sdsnew("+QUEUED\r\n"));
shared.emptyscan = createObject(REDIS_STRING,sdsnew("*2\r\n$1\r\n0\r\n*0\r\n"));
shared.wrongtypeerr = createObject(REDIS_STRING,sdsnew(
"-WRONGTYPE Operation against a key holding the wrong kind of value\r\n"));
shared.nokeyerr = createObject(REDIS_STRING,sdsnew(
"-ERR no such key\r\n"));
shared.syntaxerr = createObject(REDIS_STRING,sdsnew(
"-ERR syntax error\r\n"));
shared.sameobjecterr = createObject(REDIS_STRING,sdsnew(
"-ERR source and destination objects are the same\r\n"));
shared.outofrangeerr = createObject(REDIS_STRING,sdsnew(
"-ERR index out of range\r\n"));
shared.noscripterr = createObject(REDIS_STRING,sdsnew(
"-NOSCRIPT No matching script. Please use EVAL.\r\n"));
shared.loadingerr = createObject(REDIS_STRING,sdsnew(
"-LOADING Redis is loading the dataset in memory\r\n"));
shared.slowscripterr = createObject(REDIS_STRING,sdsnew(
"-BUSY Redis is busy running a script. You can only call SCRIPT KILL or SHUTDOWN NOSAVE.\r\n"));
shared.masterdownerr = createObject(REDIS_STRING,sdsnew(
"-MASTERDOWN Link with MASTER is down and slave-serve-stale-data is set to 'no'.\r\n"));
shared.bgsaveerr = createObject(REDIS_STRING,sdsnew(
"-MISCONF Redis is configured to save RDB snapshots, but is currently not able to persist on disk. Commands that may modify the data set are disabled. Please check Redis logs for details about the error.\r\n"));
shared.roslaveerr = createObject(REDIS_STRING,sdsnew(
"-READONLY You can't write against a read only slave.\r\n"));
shared.noautherr = createObject(REDIS_STRING,sdsnew(
"-NOAUTH Authentication required.\r\n"));
shared.oomerr = createObject(REDIS_STRING,sdsnew(
"-OOM command not allowed when used memory > 'maxmemory'.\r\n"));
shared.execaborterr = createObject(REDIS_STRING,sdsnew(
"-EXECABORT Transaction discarded because of previous errors.\r\n"));
shared.noreplicaserr = createObject(REDIS_STRING,sdsnew(
"-NOREPLICAS Not enough good slaves to write.\r\n"));
shared.space = createObject(REDIS_STRING,sdsnew(" "));
shared.colon = createObject(REDIS_STRING,sdsnew(":"));
shared.plus = createObject(REDIS_STRING,sdsnew("+"));
for (j = 0; j < REDIS_SHARED_SELECT_CMDS; j++) {
char dictid_str;
int dictid_len;
dictid_len = ll2string(dictid_str,sizeof(dictid_str),j);
shared.select = createObject(REDIS_STRING,
sdscatprintf(sdsempty(),
"*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n",
dictid_len, dictid_str));
}
shared.messagebulk = createStringObject("$7\r\nmessage\r\n",13);
shared.pmessagebulk = createStringObject("$8\r\npmessage\r\n",14);
shared.subscribebulk = createStringObject("$9\r\nsubscribe\r\n",15);
shared.unsubscribebulk = createStringObject("$11\r\nunsubscribe\r\n",18);
shared.psubscribebulk = createStringObject("$10\r\npsubscribe\r\n",17);
shared.punsubscribebulk = createStringObject("$12\r\npunsubscribe\r\n",19);
shared.del = createStringObject("DEL",3);
shared.rpop = createStringObject("RPOP",4);
shared.lpop = createStringObject("LPOP",4);
shared.lpush = createStringObject("LPUSH",5);
for (j = 0; j < REDIS_SHARED_INTEGERS; j++) {
shared.integers = createObject(REDIS_STRING,(void*)(long)j);
shared.integers->encoding = REDIS_ENCODING_INT;
}
for (j = 0; j < REDIS_SHARED_BULKHDR_LEN; j++) {
shared.mbulkhdr = createObject(REDIS_STRING,
sdscatprintf(sdsempty(),"*%d\r\n",j));
shared.bulkhdr = createObject(REDIS_STRING,
sdscatprintf(sdsempty(),"$%d\r\n",j));
}
}
1.3 改变打开的最大文件数量
/* This function will try to raise the max number of open files accordingly to
* the configured max number of clients. It will also account for 32 additional
* file descriptors as we need a few more for persistence, listening
* sockets, log files and so forth.
*
* If it will not be possible to set the limit accordingly to the configured
* max number of clients, the function will do the reverse setting
* server.maxclients to the value that we can actually handle. */
void adjustOpenFilesLimit(void) {
rlim_t maxfiles = server.maxclients+32;
struct rlimit limit;
if (getrlimit(RLIMIT_NOFILE,&limit) == -1) {
redisLog(REDIS_WARNING,"Unable to obtain the current NOFILE limit (%s), assuming 1024 and setting the max clients configuration accordingly.",
strerror(errno));
server.maxclients = 1024-32;
} else {
rlim_t oldlimit = limit.rlim_cur;
/* Set the max number of files if the current limit is not enough
* for our needs. */
if (oldlimit < maxfiles) {
rlim_t f;
f = maxfiles;
while(f > oldlimit) {
limit.rlim_cur = f;
limit.rlim_max = f;
if (setrlimit(RLIMIT_NOFILE,&limit) != -1) break;
f -= 128;
}
if (f < oldlimit) f = oldlimit;
if (f != maxfiles) {
server.maxclients = f-32;
redisLog(REDIS_WARNING,"Unable to set the max number of files limit to %d (%s), setting the max clients configuration to %d.",
(int) maxfiles, strerror(errno), (int) server.maxclients);
} else {
redisLog(REDIS_NOTICE,"Max number of open files set to %d",
(int) maxfiles);
}
}
}
}
1.4 创建事件处理
aeEventLoop *aeCreateEventLoop(int setsize) {
aeEventLoop *eventLoop;
int i;
if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) goto err;
eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);
eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);
if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err;
eventLoop->setsize = setsize;
eventLoop->lastTime = time(NULL);
eventLoop->timeEventHead = NULL;
eventLoop->timeEventNextId = 0;
eventLoop->stop = 0;
eventLoop->maxfd = -1;
eventLoop->beforesleep = NULL;
if (aeApiCreate(eventLoop) == -1) goto err;
/* Events with mask == AE_NONE are not set. So let's initialize the
* vector with it. */
for (i = 0; i < setsize; i++)
eventLoop->events.mask = AE_NONE;
return eventLoop;
err:
if (eventLoop) {
zfree(eventLoop->events);
zfree(eventLoop->fired);
zfree(eventLoop);
}
return NULL;
}
1.5 绑定监听端口
/* Initialize a set of file descriptors to listen to the specified 'port'
* binding the addresses specified in the Redis server configuration.
*
* The listening file descriptors are stored in the integer array 'fds'
* and their number is set in '*count'.
*
* The addresses to bind are specified in the global server.bindaddr array
* and their number is server.bindaddr_count. If the server configuration
* contains no specific addresses to bind, this function will try to
* bind * (all addresses) for both the IPv4 and IPv6 protocols.
*
* On success the function returns REDIS_OK.
*
* On error the function returns REDIS_ERR. For the function to be on
* error, at least one of the server.bindaddr addresses was
* impossible to bind, or no bind addresses were specified in the server
* configuration but the function is not able to bind * for at least
* one of the IPv4 or IPv6 protocols. */
int listenToPort(int port, int *fds, int *count) {
int j;
/* Force binding of 0.0.0.0 if no bind address is specified, always
* entering the loop if j == 0. */
if (server.bindaddr_count == 0) server.bindaddr = NULL;
for (j = 0; j < server.bindaddr_count || j == 0; j++) {
if (server.bindaddr == NULL) {
/* Bind * for both IPv6 and IPv4, we enter here only if
* server.bindaddr_count == 0. */
fds[*count] = anetTcp6Server(server.neterr,port,NULL);
if (fds[*count] != ANET_ERR) (*count)++;
fds[*count] = anetTcpServer(server.neterr,port,NULL);
if (fds[*count] != ANET_ERR) (*count)++;
/* Exit the loop if we were able to bind * on IPv4 or IPv6,
* otherwise fds[*count] will be ANET_ERR and we'll print an
* error and return to the caller with an error. */
if (*count) break;
} else if (strchr(server.bindaddr,':')) {
/* Bind IPv6 address. */
fds[*count] = anetTcp6Server(server.neterr,port,server.bindaddr);
} else {
/* Bind IPv4 address. */
fds[*count] = anetTcpServer(server.neterr,port,server.bindaddr);
}
if (fds[*count] == ANET_ERR) {
redisLog(REDIS_WARNING,
"Creating Server TCP listening socket %s:%d: %s",
server.bindaddr ? server.bindaddr : "*",
server.port, server.neterr);
return REDIS_ERR;
}
(*count)++;
}
return REDIS_OK;
}
1.6 创建redis数据库，并初始化一些中间状态
/* Create the Redis databases, and initialize other internal state. */
for (j = 0; j < server.dbnum; j++) {
server.db.dict = dictCreate(&dbDictType,NULL);
server.db.expires = dictCreate(&keyptrDictType,NULL);
server.db.blocking_keys = dictCreate(&keylistDictType,NULL);
server.db.ready_keys = dictCreate(&setDictType,NULL);
server.db.watched_keys = dictCreate(&keylistDictType,NULL);
server.db.id = j;
server.db.avg_ttl = 0;
}
server.pubsub_channels = dictCreate(&keylistDictType,NULL);
server.pubsub_patterns = listCreate();
listSetFreeMethod(server.pubsub_patterns,freePubsubPattern);
listSetMatchMethod(server.pubsub_patterns,listMatchPubsubPattern);
server.cronloops = 0;
server.rdb_child_pid = -1;
server.aof_child_pid = -1;
aofRewriteBufferReset();
server.aof_buf = sdsempty();
server.lastsave = time(NULL); /* At startup we consider the DB saved. */
server.lastbgsave_try = 0; /* At startup we never tried to BGSAVE. */
server.rdb_save_time_last = -1;
server.rdb_save_time_start = -1;
server.dirty = 0;
server.stat_numcommands = 0;
server.stat_numconnections = 0;
server.stat_expiredkeys = 0;
server.stat_evictedkeys = 0;
server.stat_starttime = time(NULL);
server.stat_keyspace_misses = 0;
server.stat_keyspace_hits = 0;
server.stat_peak_memory = 0;
server.stat_fork_time = 0;
server.stat_rejected_conn = 0;
server.stat_sync_full = 0;
server.stat_sync_partial_ok = 0;
server.stat_sync_partial_err = 0;
memset(server.ops_sec_samples,0,sizeof(server.ops_sec_samples));
server.ops_sec_idx = 0;
server.ops_sec_last_sample_time = mstime();
server.ops_sec_last_sample_ops = 0;
server.unixtime = time(NULL);
server.mstime = mstime();
server.lastbgsave_status = REDIS_OK;
server.repl_good_slaves_count = 0;
1.7 创建服务器时间事件
/* Create the serverCron() time event, that's our main way to process
* background operations. */
if(aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
redisPanic("Can't create the serverCron time event.");
exit(1);
}
-----------------------------------------------------------------------------
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->next = eventLoop->timeEventHead;
eventLoop->timeEventHead = te;
return id;
}
补充：servercron
/* This is our timer interrupt, called server.hz times per second.
* Here is where we do a number of things that need to be done asynchronously.
* For instance:
*
* - Active expired keys collection (it is also performed in a lazy way on
* lookup).
* - Software watchdog.
* - Update some statistic.
* - Incremental rehashing of the DBs hash tables.
* - Triggering BGSAVE / AOF rewrite, and handling of terminated children.
* - Clients timeout of different kinds.
* - Replication reconnection.
* - Many more...
*
* Everything directly called here will be called server.hz times per second,
* so in order to throttle execution of things we want to do less frequently
* a macro is used: run_with_period(milliseconds) { .... }
*/
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
int j;
REDIS_NOTUSED(eventLoop);
REDIS_NOTUSED(id);
REDIS_NOTUSED(clientData);
/* Software watchdog: deliver the SIGALRM that will reach the signal
* handler if we don't return here fast enough. */
if (server.watchdog_period) watchdogScheduleSignal(server.watchdog_period);
/* We take a cached value of the unix time in the global state because
* with virtual memory and aging there is to store the current time
* in objects at every object access, and accuracy is not needed.
* To access a global var is faster than calling time(NULL) */
server.unixtime = time(NULL);
server.mstime = mstime();
run_with_period(100) trackOperationsPerSecond();
/* We have just 22 bits per object for LRU information.
* So we use an (eventually wrapping) LRU clock with 10 seconds resolution.
* 2^22 bits with 10 seconds resolution is more or less 1.5 years.
*
* Note that even if this will wrap after 1.5 years it's not a problem,
* everything will still work but just some object will appear younger
* to Redis. But for this to happen a given object should never be touched
* for 1.5 years.
*
* Note that you can change the resolution altering the
* REDIS_LRU_CLOCK_RESOLUTION define.
*/
updateLRUClock();
/* Record the max memory used since the server was started. */
if (zmalloc_used_memory() > server.stat_peak_memory)
server.stat_peak_memory = zmalloc_used_memory();
/* We received a SIGTERM, shutting down here in a safe way, as it is
* not ok doing so inside the signal handler. */
if (server.shutdown_asap) {
if (prepareForShutdown(0) == REDIS_OK) exit(0);
redisLog(REDIS_WARNING,"SIGTERM received but errors trying to shut down the server, check the logs for more information");
server.shutdown_asap = 0;
}
/* Show some info about non-empty databases */
run_with_period(5000) {
for (j = 0; j < server.dbnum; j++) {
long long size, used, vkeys;
size = dictSlots(server.db.dict);
used = dictSize(server.db.dict);
vkeys = dictSize(server.db.expires);
if (used || vkeys) {
redisLog(REDIS_VERBOSE,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size);
/* dictPrintStats(server.dict); */
}
}
}
/* Show information about connected clients */
if (!server.sentinel_mode) {
run_with_period(5000) {
redisLog(REDIS_VERBOSE,
"%lu clients connected (%lu slaves), %zu bytes in use",
listLength(server.clients)-listLength(server.slaves),
listLength(server.slaves),
zmalloc_used_memory());
}
}
/* We need to do a few operations on clients asynchronously. */
clientsCron();
/* Handle background operations on Redis databases. */
databasesCron();
/* Start a scheduled AOF rewrite if this was requested by the user while
* a BGSAVE was in progress. */
if (server.rdb_child_pid == -1 && server.aof_child_pid == -1 &&
server.aof_rewrite_scheduled)
{
rewriteAppendOnlyFileBackground();
}
/* Check if a background saving or AOF rewrite in progress terminated. */
if (server.rdb_child_pid != -1 || server.aof_child_pid != -1) {
int statloc;
pid_t pid;
if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
int exitcode = WEXITSTATUS(statloc);
int bysignal = 0;
if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc);
if (pid == server.rdb_child_pid) {
backgroundSaveDoneHandler(exitcode,bysignal);
} else if (pid == server.aof_child_pid) {
backgroundRewriteDoneHandler(exitcode,bysignal);
} else {
redisLog(REDIS_WARNING,
"Warning, detected child with unmatched pid: %ld",
(long)pid);
}
updateDictResizePolicy();
}
} else {
/* If there is not a background saving/rewrite in progress check if
* we have to save/rewrite now */
for (j = 0; j < server.saveparamslen; j++) {
struct saveparam *sp = server.saveparams+j;
/* Save if we reached the given amount of changes,
* the given amount of seconds, and if the latest bgsave was
* successful or if, in case of an error, at least
* REDIS_BGSAVE_RETRY_DELAY seconds already elapsed. */
if (server.dirty >= sp->changes &&
server.unixtime-server.lastsave > sp->seconds &&
(server.unixtime-server.lastbgsave_try >
REDIS_BGSAVE_RETRY_DELAY ||
server.lastbgsave_status == REDIS_OK))
{
redisLog(REDIS_NOTICE,"%d changes in %d seconds. Saving...",
sp->changes, (int)sp->seconds);
rdbSaveBackground(server.rdb_filename);
break;
}
}
/* Trigger an AOF rewrite if needed */
if (server.rdb_child_pid == -1 &&
server.aof_child_pid == -1 &&
server.aof_rewrite_perc &&
server.aof_current_size > server.aof_rewrite_min_size)
{
long long base = server.aof_rewrite_base_size ?
server.aof_rewrite_base_size : 1;
long long growth = (server.aof_current_size*100/base) - 100;
if (growth >= server.aof_rewrite_perc) {
redisLog(REDIS_NOTICE,"Starting automatic rewriting of AOF on %lld%% growth",growth);
rewriteAppendOnlyFileBackground();
}
}
}
/* If we postponed an AOF buffer flush, let's try to do it every time the
* cron function is called. */
if (server.aof_flush_postponed_start) flushAppendOnlyFile(0);
/* Close clients that need to be closed asynchronous */
freeClientsInAsyncFreeQueue();
/* Replication cron function -- used to reconnect to master and
* to detect transfer failures. */
run_with_period(1000) replicationCron();
/* Run the sentinel timer if we are in sentinel mode. */
run_with_period(100) {
if (server.sentinel_mode) sentinelTimer();
}
server.cronloops++;
return 1000/server.hz;
}
1.8 创建事件处理
/* Create an event handler for accepting new connections in TCP and Unix
* domain sockets. */
for (j = 0; j < server.ipfd_count; j++) {
if (aeCreateFileEvent(server.el, server.ipfd, AE_READABLE,
acceptTcpHandler,NULL) == AE_ERR)
{
redisPanic(
"Unrecoverable error creating server.ipfd file event.");
}
}
if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
acceptUnixHandler,NULL) == AE_ERR) redisPanic("Unrecoverable error creating server.sofd file event.");
1.9 打开aof文件
/* Open the AOF file if needed. */
if (server.aof_state == REDIS_AOF_ON) {
server.aof_fd = open(server.aof_filename,
O_WRONLY|O_APPEND|O_CREAT,0644);
if (server.aof_fd == -1) {
redisLog(REDIS_WARNING, "Can't open the append-only file: %s",
strerror(errno));
exit(1);
}
}
1.10 限制32位机器内存，防止crashed
/* 32 bit instances are limited to 4GB of address space, so if there is
* no explicit limit in the user provided configuration we set a limit
* at 3 GB using maxmemory with 'noeviction' policy'. This avoids
* useless crashes of the Redis instance for out of memory. */
if (server.arch_bits == 32 && server.maxmemory == 0) {
redisLog(REDIS_WARNING,"Warning: 32 bit instance detected but no memory limit set. Setting 3 GB maxmemory limit with 'noeviction' policy now.");
server.maxmemory = 3072LL*(1024*1024); /* 3 GB */
server.maxmemory_policy = REDIS_MAXMEMORY_NO_EVICTION;
}
1.11 复制初始化
/* ----------------------- REPLICATION SCRIPT CACHE --------------------------
* The goal of this code is to keep track of scripts already sent to every
* connected slave, in order to be able to replicate EVALSHA as it is without
* translating it to EVAL every time it is possible.
*
* We use a capped collection implemented by an hash table for fast lookup
* of scripts we can send as EVALSHA, plus a linked list that is used for
* eviction of the oldest entry when the max number of items is reached.
*
* We don't care about taking a different cache for every different slave
* since to fill the cache again is not very costly, the goal of this code
* is to avoid that the same big script is trasmitted a big number of times
* per second wasting bandwidth and processor speed, but it is not a problem
* if we need to rebuild the cache from scratch from time to time, every used
* script will need to be transmitted a single time to reappear in the cache.
*
* This is how the system works:
*
* 1) Every time a new slave connects, we flush the whole script cache.
* 2) We only send as EVALSHA what was sent to the master as EVALSHA, without
* trying to convert EVAL into EVALSHA specifically for slaves.
* 3) Every time we trasmit a script as EVAL to the slaves, we also add the
* corresponding SHA1 of the script into the cache as we are sure every
* slave knows about the script starting from now.
* 4) On SCRIPT FLUSH command, we replicate the command to all the slaves
* and at the same time flush the script cache.
* 5) When the last slave disconnects, flush the cache.
* 6) We handle SCRIPT LOAD as well since that's how scripts are loaded
* in the master sometimes.
*/
/* Initialize the script cache, only called at startup. */
void replicationScriptCacheInit(void) {
server.repl_scriptcache_size = 10000;
server.repl_scriptcache_dict = dictCreate(&replScriptCacheDictType,NULL);
server.repl_scriptcache_fifo = listCreate();
}
1.12 初始化lua脚本环境
/* Initialize the scripting environment.
* It is possible to call this function to reset the scripting environment
* assuming that we call scriptingRelease() before.
* See scriptingReset() for more information. */
void scriptingInit(void) {
lua_State *lua = lua_open();
luaLoadLibraries(lua);
luaRemoveUnsupportedFunctions(lua);
/* Initialize a dictionary we use to map SHAs to scripts.
* This is useful for replication, as we need to replicate EVALSHA
* as EVAL, so we need to remember the associated script. */
server.lua_scripts = dictCreate(&shaScriptObjectDictType,NULL);
/* Register the redis commands table and fields */
lua_newtable(lua);
/* redis.call */
lua_pushstring(lua,"call");
lua_pushcfunction(lua,luaRedisCallCommand);
lua_settable(lua,-3);
/* redis.pcall */
lua_pushstring(lua,"pcall");
lua_pushcfunction(lua,luaRedisPCallCommand);
lua_settable(lua,-3);
/* redis.log and log levels. */
lua_pushstring(lua,"log");
lua_pushcfunction(lua,luaLogCommand);
lua_settable(lua,-3);
lua_pushstring(lua,"LOG_DEBUG");
lua_pushnumber(lua,REDIS_DEBUG);
lua_settable(lua,-3);
lua_pushstring(lua,"LOG_VERBOSE");
lua_pushnumber(lua,REDIS_VERBOSE);
lua_settable(lua,-3);
lua_pushstring(lua,"LOG_NOTICE");
lua_pushnumber(lua,REDIS_NOTICE);
lua_settable(lua,-3);
lua_pushstring(lua,"LOG_WARNING");
lua_pushnumber(lua,REDIS_WARNING);
lua_settable(lua,-3);
/* redis.sha1hex */
lua_pushstring(lua, "sha1hex");
lua_pushcfunction(lua, luaRedisSha1hexCommand);
lua_settable(lua, -3);
/* redis.error_reply and redis.status_reply */
lua_pushstring(lua, "error_reply");
lua_pushcfunction(lua, luaRedisErrorReplyCommand);
lua_settable(lua, -3);
lua_pushstring(lua, "status_reply");
lua_pushcfunction(lua, luaRedisStatusReplyCommand);
lua_settable(lua, -3);
/* Finally set the table as 'redis' global var. */
lua_setglobal(lua,"redis");
/* Replace math.random and math.randomseed with our implementations. */
lua_getglobal(lua,"math");
lua_pushstring(lua,"random");
lua_pushcfunction(lua,redis_math_random);
lua_settable(lua,-3);
lua_pushstring(lua,"randomseed");
lua_pushcfunction(lua,redis_math_randomseed);
lua_settable(lua,-3);
lua_setglobal(lua,"math");
/* Add a helper function that we use to sort the multi bulk output of non
* deterministic commands, when containing 'false' elements. */
{
char *compare_func = "function __redis__compare_helper(a,b)\n"
"if a == false then a = '' end\n"
"if b == false then b = '' end\n"
"return a
页:
[1]