1 /* 2 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> 3 * All rights reserved. 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions are met: 7 * 8 * * Redistributions of source code must retain the above copyright notice, 9 * this list of conditions and the following disclaimer. 10 * * Redistributions in binary form must reproduce the above copyright 11 * notice, this list of conditions and the following disclaimer in the 12 * documentation and/or other materials provided with the distribution. 13 * * Neither the name of Redis nor the names of its contributors may be used 14 * to endorse or promote products derived from this software without 15 * specific prior written permission. 16 * 17 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 18 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 19 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 20 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 21 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 22 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 23 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 24 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 25 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 26 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 27 * POSSIBILITY OF SUCH DAMAGE. 28 */ 29 30 #include "server.h" 31 #include "cluster.h" 32 #include "slowlog.h" 33 #include "bio.h" 34 #include "latency.h" 35 36 #include <time.h> 37 #include <signal.h> 38 #include <sys/wait.h> 39 #include <errno.h> 40 #include <assert.h> 41 #include <ctype.h> 42 #include <stdarg.h> 43 #include <arpa/inet.h> 44 #include <sys/stat.h> 45 #include <fcntl.h> 46 #include <sys/time.h> 47 #include <sys/resource.h> 48 #include <sys/uio.h> 49 #include <sys/un.h> 50 #include <limits.h> 51 #include <float.h> 52 #include <math.h> 53 #include <sys/resource.h> 54 #include <sys/utsname.h> 55 #include <locale.h> 56 #include <sys/socket.h> 57 58 /* Our shared "common" objects */ 59 60 struct sharedObjectsStruct shared; 61 62 /* Global vars that are actually used as constants. The following double 63 * values are used for double on-disk serialization, and are initialized 64 * at runtime to avoid strange compiler optimizations. */ 65 66 double R_Zero, R_PosInf, R_NegInf, R_Nan; 67 68 /*================================= Globals ================================= */ 69 70 /* Global vars */ 71 struct redisServer server; /* server global state */ 72 73 /* Our command table. 74 * 75 * Every entry is composed of the following fields: 76 * 77 * name: a string representing the command name. 78 * function: pointer to the C function implementing the command. 79 * arity: number of arguments, it is possible to use -N to say >= N 80 * sflags: command flags as string. See below for a table of flags. 81 * flags: flags as bitmask. Computed by Redis using the 'sflags' field. 82 * get_keys_proc: an optional function to get key arguments from a command. 83 * This is only used when the following three fields are not 84 * enough to specify what arguments are keys. 85 * first_key_index: first argument that is a key 86 * last_key_index: last argument that is a key 87 * key_step: step to get all the keys from first to last argument. For instance 88 * in MSET the step is two since arguments are key,val,key,val,... 89 * microseconds: microseconds of total execution time for this command. 90 * calls: total number of calls of this command. 91 * 92 * The flags, microseconds and calls fields are computed by Redis and should 93 * always be set to zero. 94 * 95 * Command flags are expressed using strings where every character represents 96 * a flag. Later the populateCommandTable() function will take care of 97 * populating the real 'flags' field using this characters. 98 * 99 * This is the meaning of the flags: 100 * 101 * w: write command (may modify the key space). 102 * r: read command (will never modify the key space). 103 * m: may increase memory usage once called. Don't allow if out of memory. 104 * a: admin command, like SAVE or SHUTDOWN. 105 * p: Pub/Sub related command. 106 * f: force replication of this command, regardless of server.dirty. 107 * s: command not allowed in scripts. 108 * R: random command. Command is not deterministic, that is, the same command 109 * with the same arguments, with the same key space, may have different 110 * results. For instance SPOP and RANDOMKEY are two random commands. 111 * S: Sort command output array if called from script, so that the output 112 * is deterministic. 113 * l: Allow command while loading the database. 114 * t: Allow command while a slave has stale data but is not allowed to 115 * server this data. Normally no command is accepted in this condition 116 * but just a few. 117 * M: Do not automatically propagate the command on MONITOR. 118 * k: Perform an implicit ASKING for this command, so the command will be 119 * accepted in cluster mode if the slot is marked as 'importing'. 120 * F: Fast command: O(1) or O(log(N)) command that should never delay 121 * its execution as long as the kernel scheduler is giving us time. 122 * Note that commands that may trigger a DEL as a side effect (like SET) 123 * are not fast commands. 124 */ 125 struct redisCommand redisCommandTable[] = { 126 {"get",getCommand,2,"rF",0,NULL,1,1,1,0,0}, 127 {"set",setCommand,-3,"wm",0,NULL,1,1,1,0,0}, 128 {"setnx",setnxCommand,3,"wmF",0,NULL,1,1,1,0,0}, 129 {"setex",setexCommand,4,"wm",0,NULL,1,1,1,0,0}, 130 {"psetex",psetexCommand,4,"wm",0,NULL,1,1,1,0,0}, 131 {"append",appendCommand,3,"wm",0,NULL,1,1,1,0,0}, 132 {"strlen",strlenCommand,2,"rF",0,NULL,1,1,1,0,0}, 133 {"del",delCommand,-2,"w",0,NULL,1,-1,1,0,0}, 134 {"exists",existsCommand,-2,"rF",0,NULL,1,-1,1,0,0}, 135 {"setbit",setbitCommand,4,"wm",0,NULL,1,1,1,0,0}, 136 {"getbit",getbitCommand,3,"rF",0,NULL,1,1,1,0,0}, 137 {"setrange",setrangeCommand,4,"wm",0,NULL,1,1,1,0,0}, 138 {"getrange",getrangeCommand,4,"r",0,NULL,1,1,1,0,0}, 139 {"substr",getrangeCommand,4,"r",0,NULL,1,1,1,0,0}, 140 {"incr",incrCommand,2,"wmF",0,NULL,1,1,1,0,0}, 141 {"decr",decrCommand,2,"wmF",0,NULL,1,1,1,0,0}, 142 {"mget",mgetCommand,-2,"r",0,NULL,1,-1,1,0,0}, 143 {"rpush",rpushCommand,-3,"wmF",0,NULL,1,1,1,0,0}, 144 {"lpush",lpushCommand,-3,"wmF",0,NULL,1,1,1,0,0}, 145 {"rpushx",rpushxCommand,3,"wmF",0,NULL,1,1,1,0,0}, 146 {"lpushx",lpushxCommand,3,"wmF",0,NULL,1,1,1,0,0}, 147 {"linsert",linsertCommand,5,"wm",0,NULL,1,1,1,0,0}, 148 {"rpop",rpopCommand,2,"wF",0,NULL,1,1,1,0,0}, 149 {"lpop",lpopCommand,2,"wF",0,NULL,1,1,1,0,0}, 150 {"brpop",brpopCommand,-3,"ws",0,NULL,1,1,1,0,0}, 151 {"brpoplpush",brpoplpushCommand,4,"wms",0,NULL,1,2,1,0,0}, 152 {"blpop",blpopCommand,-3,"ws",0,NULL,1,-2,1,0,0}, 153 {"llen",llenCommand,2,"rF",0,NULL,1,1,1,0,0}, 154 {"lindex",lindexCommand,3,"r",0,NULL,1,1,1,0,0}, 155 {"lset",lsetCommand,4,"wm",0,NULL,1,1,1,0,0}, 156 {"lrange",lrangeCommand,4,"r",0,NULL,1,1,1,0,0}, 157 {"ltrim",ltrimCommand,4,"w",0,NULL,1,1,1,0,0}, 158 {"lrem",lremCommand,4,"w",0,NULL,1,1,1,0,0}, 159 {"rpoplpush",rpoplpushCommand,3,"wm",0,NULL,1,2,1,0,0}, 160 {"sadd",saddCommand,-3,"wmF",0,NULL,1,1,1,0,0}, 161 {"srem",sremCommand,-3,"wF",0,NULL,1,1,1,0,0}, 162 {"smove",smoveCommand,4,"wF",0,NULL,1,2,1,0,0}, 163 {"sismember",sismemberCommand,3,"rF",0,NULL,1,1,1,0,0}, 164 {"scard",scardCommand,2,"rF",0,NULL,1,1,1,0,0}, 165 {"spop",spopCommand,-2,"wRsF",0,NULL,1,1,1,0,0}, 166 {"srandmember",srandmemberCommand,-2,"rR",0,NULL,1,1,1,0,0}, 167 {"sinter",sinterCommand,-2,"rS",0,NULL,1,-1,1,0,0}, 168 {"sinterstore",sinterstoreCommand,-3,"wm",0,NULL,1,-1,1,0,0}, 169 {"sunion",sunionCommand,-2,"rS",0,NULL,1,-1,1,0,0}, 170 {"sunionstore",sunionstoreCommand,-3,"wm",0,NULL,1,-1,1,0,0}, 171 {"sdiff",sdiffCommand,-2,"rS",0,NULL,1,-1,1,0,0}, 172 {"sdiffstore",sdiffstoreCommand,-3,"wm",0,NULL,1,-1,1,0,0}, 173 {"smembers",sinterCommand,2,"rS",0,NULL,1,1,1,0,0}, 174 {"sscan",sscanCommand,-3,"rR",0,NULL,1,1,1,0,0}, 175 {"zadd",zaddCommand,-4,"wmF",0,NULL,1,1,1,0,0}, 176 {"zincrby",zincrbyCommand,4,"wmF",0,NULL,1,1,1,0,0}, 177 {"zrem",zremCommand,-3,"wF",0,NULL,1,1,1,0,0}, 178 {"zremrangebyscore",zremrangebyscoreCommand,4,"w",0,NULL,1,1,1,0,0}, 179 {"zremrangebyrank",zremrangebyrankCommand,4,"w",0,NULL,1,1,1,0,0}, 180 {"zremrangebylex",zremrangebylexCommand,4,"w",0,NULL,1,1,1,0,0}, 181 {"zunionstore",zunionstoreCommand,-4,"wm",0,zunionInterGetKeys,0,0,0,0,0}, 182 {"zinterstore",zinterstoreCommand,-4,"wm",0,zunionInterGetKeys,0,0,0,0,0}, 183 {"zrange",zrangeCommand,-4,"r",0,NULL,1,1,1,0,0}, 184 {"zrangebyscore",zrangebyscoreCommand,-4,"r",0,NULL,1,1,1,0,0}, 185 {"zrevrangebyscore",zrevrangebyscoreCommand,-4,"r",0,NULL,1,1,1,0,0}, 186 {"zrangebylex",zrangebylexCommand,-4,"r",0,NULL,1,1,1,0,0}, 187 {"zrevrangebylex",zrevrangebylexCommand,-4,"r",0,NULL,1,1,1,0,0}, 188 {"zcount",zcountCommand,4,"rF",0,NULL,1,1,1,0,0}, 189 {"zlexcount",zlexcountCommand,4,"rF",0,NULL,1,1,1,0,0}, 190 {"zrevrange",zrevrangeCommand,-4,"r",0,NULL,1,1,1,0,0}, 191 {"zcard",zcardCommand,2,"rF",0,NULL,1,1,1,0,0}, 192 {"zscore",zscoreCommand,3,"rF",0,NULL,1,1,1,0,0}, 193 {"zrank",zrankCommand,3,"rF",0,NULL,1,1,1,0,0}, 194 {"zrevrank",zrevrankCommand,3,"rF",0,NULL,1,1,1,0,0}, 195 {"zscan",zscanCommand,-3,"rR",0,NULL,1,1,1,0,0}, 196 {"hset",hsetCommand,4,"wmF",0,NULL,1,1,1,0,0}, 197 {"hsetnx",hsetnxCommand,4,"wmF",0,NULL,1,1,1,0,0}, 198 {"hget",hgetCommand,3,"rF",0,NULL,1,1,1,0,0}, 199 {"hmset",hmsetCommand,-4,"wm",0,NULL,1,1,1,0,0}, 200 {"hmget",hmgetCommand,-3,"r",0,NULL,1,1,1,0,0}, 201 {"hincrby",hincrbyCommand,4,"wmF",0,NULL,1,1,1,0,0}, 202 {"hincrbyfloat",hincrbyfloatCommand,4,"wmF",0,NULL,1,1,1,0,0}, 203 {"hdel",hdelCommand,-3,"wF",0,NULL,1,1,1,0,0}, 204 {"hlen",hlenCommand,2,"rF",0,NULL,1,1,1,0,0}, 205 {"hstrlen",hstrlenCommand,3,"rF",0,NULL,1,1,1,0,0}, 206 {"hkeys",hkeysCommand,2,"rS",0,NULL,1,1,1,0,0}, 207 {"hvals",hvalsCommand,2,"rS",0,NULL,1,1,1,0,0}, 208 {"hgetall",hgetallCommand,2,"r",0,NULL,1,1,1,0,0}, 209 {"hexists",hexistsCommand,3,"rF",0,NULL,1,1,1,0,0}, 210 {"hscan",hscanCommand,-3,"rR",0,NULL,1,1,1,0,0}, 211 {"incrby",incrbyCommand,3,"wmF",0,NULL,1,1,1,0,0}, 212 {"decrby",decrbyCommand,3,"wmF",0,NULL,1,1,1,0,0}, 213 {"incrbyfloat",incrbyfloatCommand,3,"wmF",0,NULL,1,1,1,0,0}, 214 {"getset",getsetCommand,3,"wm",0,NULL,1,1,1,0,0}, 215 {"mset",msetCommand,-3,"wm",0,NULL,1,-1,2,0,0}, 216 {"msetnx",msetnxCommand,-3,"wm",0,NULL,1,-1,2,0,0}, 217 {"randomkey",randomkeyCommand,1,"rR",0,NULL,0,0,0,0,0}, 218 {"select",selectCommand,2,"rlF",0,NULL,0,0,0,0,0}, 219 {"move",moveCommand,3,"wF",0,NULL,1,1,1,0,0}, 220 {"rename",renameCommand,3,"w",0,NULL,1,2,1,0,0}, 221 {"renamenx",renamenxCommand,3,"wF",0,NULL,1,2,1,0,0}, 222 {"expire",expireCommand,3,"wF",0,NULL,1,1,1,0,0}, 223 {"expireat",expireatCommand,3,"wF",0,NULL,1,1,1,0,0}, 224 {"pexpire",pexpireCommand,3,"wF",0,NULL,1,1,1,0,0}, 225 {"pexpireat",pexpireatCommand,3,"wF",0,NULL,1,1,1,0,0}, 226 {"keys",keysCommand,2,"rS",0,NULL,0,0,0,0,0}, 227 {"scan",scanCommand,-2,"rR",0,NULL,0,0,0,0,0}, 228 {"dbsize",dbsizeCommand,1,"rF",0,NULL,0,0,0,0,0}, 229 {"auth",authCommand,2,"rsltF",0,NULL,0,0,0,0,0}, 230 {"ping",pingCommand,-1,"rtF",0,NULL,0,0,0,0,0}, 231 {"echo",echoCommand,2,"rF",0,NULL,0,0,0,0,0}, 232 {"save",saveCommand,1,"ars",0,NULL,0,0,0,0,0}, 233 {"bgsave",bgsaveCommand,1,"ar",0,NULL,0,0,0,0,0}, 234 {"bgrewriteaof",bgrewriteaofCommand,1,"ar",0,NULL,0,0,0,0,0}, 235 {"shutdown",shutdownCommand,-1,"arlt",0,NULL,0,0,0,0,0}, 236 {"lastsave",lastsaveCommand,1,"rRF",0,NULL,0,0,0,0,0}, 237 {"type",typeCommand,2,"rF",0,NULL,1,1,1,0,0}, 238 {"multi",multiCommand,1,"rsF",0,NULL,0,0,0,0,0}, 239 {"exec",execCommand,1,"sM",0,NULL,0,0,0,0,0}, 240 {"discard",discardCommand,1,"rsF",0,NULL,0,0,0,0,0}, 241 {"sync",syncCommand,1,"ars",0,NULL,0,0,0,0,0}, 242 {"psync",syncCommand,3,"ars",0,NULL,0,0,0,0,0}, 243 {"replconf",replconfCommand,-1,"arslt",0,NULL,0,0,0,0,0}, 244 {"flushdb",flushdbCommand,1,"w",0,NULL,0,0,0,0,0}, 245 {"flushall",flushallCommand,1,"w",0,NULL,0,0,0,0,0}, 246 {"sort",sortCommand,-2,"wm",0,sortGetKeys,1,1,1,0,0}, 247 {"info",infoCommand,-1,"rlt",0,NULL,0,0,0,0,0}, 248 {"monitor",monitorCommand,1,"ars",0,NULL,0,0,0,0,0}, 249 {"ttl",ttlCommand,2,"rF",0,NULL,1,1,1,0,0}, 250 {"pttl",pttlCommand,2,"rF",0,NULL,1,1,1,0,0}, 251 {"persist",persistCommand,2,"wF",0,NULL,1,1,1,0,0}, 252 {"slaveof",slaveofCommand,3,"ast",0,NULL,0,0,0,0,0}, 253 {"role",roleCommand,1,"lst",0,NULL,0,0,0,0,0}, 254 {"debug",debugCommand,-2,"as",0,NULL,0,0,0,0,0}, 255 {"config",configCommand,-2,"art",0,NULL,0,0,0,0,0}, 256 {"subscribe",subscribeCommand,-2,"rpslt",0,NULL,0,0,0,0,0}, 257 {"unsubscribe",unsubscribeCommand,-1,"rpslt",0,NULL,0,0,0,0,0}, 258 {"psubscribe",psubscribeCommand,-2,"rpslt",0,NULL,0,0,0,0,0}, 259 {"punsubscribe",punsubscribeCommand,-1,"rpslt",0,NULL,0,0,0,0,0}, 260 {"publish",publishCommand,3,"pltrF",0,NULL,0,0,0,0,0}, 261 {"pubsub",pubsubCommand,-2,"pltrR",0,NULL,0,0,0,0,0}, 262 {"watch",watchCommand,-2,"rsF",0,NULL,1,-1,1,0,0}, 263 {"unwatch",unwatchCommand,1,"rsF",0,NULL,0,0,0,0,0}, 264 {"cluster",clusterCommand,-2,"ar",0,NULL,0,0,0,0,0}, 265 {"restore",restoreCommand,-4,"wm",0,NULL,1,1,1,0,0}, 266 {"restore-asking",restoreCommand,-4,"wmk",0,NULL,1,1,1,0,0}, 267 {"migrate",migrateCommand,-6,"w",0,migrateGetKeys,0,0,0,0,0}, 268 {"asking",askingCommand,1,"r",0,NULL,0,0,0,0,0}, 269 {"readonly",readonlyCommand,1,"rF",0,NULL,0,0,0,0,0}, 270 {"readwrite",readwriteCommand,1,"rF",0,NULL,0,0,0,0,0}, 271 {"dump",dumpCommand,2,"r",0,NULL,1,1,1,0,0}, 272 {"object",objectCommand,3,"r",0,NULL,2,2,2,0,0}, 273 {"client",clientCommand,-2,"rs",0,NULL,0,0,0,0,0}, 274 {"eval",evalCommand,-3,"s",0,evalGetKeys,0,0,0,0,0}, 275 {"evalsha",evalShaCommand,-3,"s",0,evalGetKeys,0,0,0,0,0}, 276 {"slowlog",slowlogCommand,-2,"r",0,NULL,0,0,0,0,0}, 277 {"script",scriptCommand,-2,"rs",0,NULL,0,0,0,0,0}, 278 {"time",timeCommand,1,"rRF",0,NULL,0,0,0,0,0}, 279 {"bitop",bitopCommand,-4,"wm",0,NULL,2,-1,1,0,0}, 280 {"bitcount",bitcountCommand,-2,"r",0,NULL,1,1,1,0,0}, 281 {"bitpos",bitposCommand,-3,"r",0,NULL,1,1,1,0,0}, 282 {"wait",waitCommand,3,"rs",0,NULL,0,0,0,0,0}, 283 {"command",commandCommand,0,"rlt",0,NULL,0,0,0,0,0}, 284 {"geoadd",geoaddCommand,-5,"wm",0,NULL,1,1,1,0,0}, 285 {"georadius",georadiusCommand,-6,"r",0,NULL,1,1,1,0,0}, 286 {"georadiusbymember",georadiusByMemberCommand,-5,"r",0,NULL,1,1,1,0,0}, 287 {"geohash",geohashCommand,-2,"r",0,NULL,1,1,1,0,0}, 288 {"geopos",geoposCommand,-2,"r",0,NULL,1,1,1,0,0}, 289 {"geodist",geodistCommand,-4,"r",0,NULL,1,1,1,0,0}, 290 {"pfselftest",pfselftestCommand,1,"r",0,NULL,0,0,0,0,0}, 291 {"pfadd",pfaddCommand,-2,"wmF",0,NULL,1,1,1,0,0}, 292 {"pfcount",pfcountCommand,-2,"r",0,NULL,1,-1,1,0,0}, 293 {"pfmerge",pfmergeCommand,-2,"wm",0,NULL,1,-1,1,0,0}, 294 {"pfdebug",pfdebugCommand,-3,"w",0,NULL,0,0,0,0,0}, 295 {"latency",latencyCommand,-2,"arslt",0,NULL,0,0,0,0,0} 296 }; 297 298 struct evictionPoolEntry *evictionPoolAlloc(void); 299 300 /*============================ Utility functions ============================ */ 301 302 /* Low level logging. To use only for very big messages, otherwise 303 * serverLog() is to prefer. */ 304 void serverLogRaw(int level, const char *msg) { 305 const int syslogLevelMap[] = { LOG_DEBUG, LOG_INFO, LOG_NOTICE, LOG_WARNING }; 306 const char *c = ".-*#"; 307 FILE *fp; 308 char buf[64]; 309 int rawmode = (level & LL_RAW); 310 int log_to_stdout = server.logfile[0] == '\0'; 311 312 level &= 0xff; /* clear flags */ 313 if (level < server.verbosity) return; 314 315 fp = log_to_stdout ? stdout : fopen(server.logfile,"a"); 316 if (!fp) return; 317 318 if (rawmode) { 319 fprintf(fp,"%s",msg); 320 } else { 321 int off; 322 struct timeval tv; 323 int role_char; 324 pid_t pid = getpid(); 325 326 gettimeofday(&tv,NULL); 327 off = strftime(buf,sizeof(buf),"%d %b %H:%M:%S.",localtime(&tv.tv_sec)); 328 snprintf(buf+off,sizeof(buf)-off,"%03d",(int)tv.tv_usec/1000); 329 if (server.sentinel_mode) { 330 role_char = 'X'; /* Sentinel. */ 331 } else if (pid != server.pid) { 332 role_char = 'C'; /* RDB / AOF writing child. */ 333 } else { 334 role_char = (server.masterhost ? 'S':'M'); /* Slave or Master. */ 335 } 336 fprintf(fp,"%d:%c %s %c %s\n", 337 (int)getpid(),role_char, buf,c[level],msg); 338 } 339 fflush(fp); 340 341 if (!log_to_stdout) fclose(fp); 342 if (server.syslog_enabled) syslog(syslogLevelMap[level], "%s", msg); 343 } 344 345 /* Like serverLogRaw() but with printf-alike support. This is the function that 346 * is used across the code. The raw version is only used in order to dump 347 * the INFO output on crash. */ 348 void serverLog(int level, const char *fmt, ...) { 349 va_list ap; 350 char msg[LOG_MAX_LEN]; 351 352 if ((level&0xff) < server.verbosity) return; 353 354 va_start(ap, fmt); 355 vsnprintf(msg, sizeof(msg), fmt, ap); 356 va_end(ap); 357 358 serverLogRaw(level,msg); 359 } 360 361 /* Log a fixed message without printf-alike capabilities, in a way that is 362 * safe to call from a signal handler. 363 * 364 * We actually use this only for signals that are not fatal from the point 365 * of view of Redis. Signals that are going to kill the server anyway and 366 * where we need printf-alike features are served by serverLog(). */ 367 void serverLogFromHandler(int level, const char *msg) { 368 int fd; 369 int log_to_stdout = server.logfile[0] == '\0'; 370 char buf[64]; 371 372 if ((level&0xff) < server.verbosity || (log_to_stdout && server.daemonize)) 373 return; 374 fd = log_to_stdout ? STDOUT_FILENO : 375 open(server.logfile, O_APPEND|O_CREAT|O_WRONLY, 0644); 376 if (fd == -1) return; 377 ll2string(buf,sizeof(buf),getpid()); 378 if (write(fd,buf,strlen(buf)) == -1) goto err; 379 if (write(fd,":signal-handler (",17) == -1) goto err; 380 ll2string(buf,sizeof(buf),time(NULL)); 381 if (write(fd,buf,strlen(buf)) == -1) goto err; 382 if (write(fd,") ",2) == -1) goto err; 383 if (write(fd,msg,strlen(msg)) == -1) goto err; 384 if (write(fd,"\n",1) == -1) goto err; 385 err: 386 if (!log_to_stdout) close(fd); 387 } 388 389 /* Return the UNIX time in microseconds */ 390 long long ustime(void) { 391 struct timeval tv; 392 long long ust; 393 394 gettimeofday(&tv, NULL); 395 ust = ((long long)tv.tv_sec)*1000000; 396 ust += tv.tv_usec; 397 return ust; 398 } 399 400 /* Return the UNIX time in milliseconds */ 401 mstime_t mstime(void) { 402 return ustime()/1000; 403 } 404 405 /* After an RDB dump or AOF rewrite we exit from children using _exit() instead of 406 * exit(), because the latter may interact with the same file objects used by 407 * the parent process. However if we are testing the coverage normal exit() is 408 * used in order to obtain the right coverage information. */ 409 void exitFromChild(int retcode) { 410 #ifdef COVERAGE_TEST 411 exit(retcode); 412 #else 413 _exit(retcode); 414 #endif 415 } 416 417 /*====================== Hash table type implementation ==================== */ 418 419 /* This is a hash table type that uses the SDS dynamic strings library as 420 * keys and redis objects as values (objects can hold SDS strings, 421 * lists, sets). */ 422 423 void dictVanillaFree(void *privdata, void *val) 424 { 425 DICT_NOTUSED(privdata); 426 zfree(val); 427 } 428 429 void dictListDestructor(void *privdata, void *val) 430 { 431 DICT_NOTUSED(privdata); 432 listRelease((list*)val); 433 } 434 435 int dictSdsKeyCompare(void *privdata, const void *key1, 436 const void *key2) 437 { 438 int l1,l2; 439 DICT_NOTUSED(privdata); 440 441 l1 = sdslen((sds)key1); 442 l2 = sdslen((sds)key2); 443 if (l1 != l2) return 0; 444 return memcmp(key1, key2, l1) == 0; 445 } 446 447 /* A case insensitive version used for the command lookup table and other 448 * places where case insensitive non binary-safe comparison is needed. */ 449 int dictSdsKeyCaseCompare(void *privdata, const void *key1, 450 const void *key2) 451 { 452 DICT_NOTUSED(privdata); 453 454 return strcasecmp(key1, key2) == 0; 455 } 456 457 void dictObjectDestructor(void *privdata, void *val) 458 { 459 DICT_NOTUSED(privdata); 460 461 if (val == NULL) return; /* Values of swapped out keys as set to NULL */ 462 decrRefCount(val); 463 } 464 465 void dictSdsDestructor(void *privdata, void *val) 466 { 467 DICT_NOTUSED(privdata); 468 469 sdsfree(val); 470 } 471 472 int dictObjKeyCompare(void *privdata, const void *key1, 473 const void *key2) 474 { 475 const robj *o1 = key1, *o2 = key2; 476 return dictSdsKeyCompare(privdata,o1->ptr,o2->ptr); 477 } 478 479 unsigned int dictObjHash(const void *key) { 480 const robj *o = key; 481 return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr)); 482 } 483 484 unsigned int dictSdsHash(const void *key) { 485 return dictGenHashFunction((unsigned char*)key, sdslen((char*)key)); 486 } 487 488 unsigned int dictSdsCaseHash(const void *key) { 489 return dictGenCaseHashFunction((unsigned char*)key, sdslen((char*)key)); 490 } 491 492 int dictEncObjKeyCompare(void *privdata, const void *key1, 493 const void *key2) 494 { 495 robj *o1 = (robj*) key1, *o2 = (robj*) key2; 496 int cmp; 497 498 if (o1->encoding == OBJ_ENCODING_INT && 499 o2->encoding == OBJ_ENCODING_INT) 500 return o1->ptr == o2->ptr; 501 502 o1 = getDecodedObject(o1); 503 o2 = getDecodedObject(o2); 504 cmp = dictSdsKeyCompare(privdata,o1->ptr,o2->ptr); 505 decrRefCount(o1); 506 decrRefCount(o2); 507 return cmp; 508 } 509 510 unsigned int dictEncObjHash(const void *key) { 511 robj *o = (robj*) key; 512 513 if (sdsEncodedObject(o)) { 514 return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr)); 515 } else { 516 if (o->encoding == OBJ_ENCODING_INT) { 517 char buf[32]; 518 int len; 519 520 len = ll2string(buf,32,(long)o->ptr); 521 return dictGenHashFunction((unsigned char*)buf, len); 522 } else { 523 unsigned int hash; 524 525 o = getDecodedObject(o); 526 hash = dictGenHashFunction(o->ptr, sdslen((sds)o->ptr)); 527 decrRefCount(o); 528 return hash; 529 } 530 } 531 } 532 533 /* Sets type hash table */ 534 dictType setDictType = { 535 dictEncObjHash, /* hash function */ 536 NULL, /* key dup */ 537 NULL, /* val dup */ 538 dictEncObjKeyCompare, /* key compare */ 539 dictObjectDestructor, /* key destructor */ 540 NULL /* val destructor */ 541 }; 542 543 /* Sorted sets hash (note: a skiplist is used in addition to the hash table) */ 544 dictType zsetDictType = { 545 dictEncObjHash, /* hash function */ 546 NULL, /* key dup */ 547 NULL, /* val dup */ 548 dictEncObjKeyCompare, /* key compare */ 549 dictObjectDestructor, /* key destructor */ 550 NULL /* val destructor */ 551 }; 552 553 /* Db->dict, keys are sds strings, vals are Redis objects. */ 554 dictType dbDictType = { 555 dictSdsHash, /* hash function */ 556 NULL, /* key dup */ 557 NULL, /* val dup */ 558 dictSdsKeyCompare, /* key compare */ 559 dictSdsDestructor, /* key destructor */ 560 dictObjectDestructor /* val destructor */ 561 }; 562 563 /* server.lua_scripts sha (as sds string) -> scripts (as robj) cache. */ 564 dictType shaScriptObjectDictType = { 565 dictSdsCaseHash, /* hash function */ 566 NULL, /* key dup */ 567 NULL, /* val dup */ 568 dictSdsKeyCaseCompare, /* key compare */ 569 dictSdsDestructor, /* key destructor */ 570 dictObjectDestructor /* val destructor */ 571 }; 572 573 /* Db->expires */ 574 dictType keyptrDictType = { 575 dictSdsHash, /* hash function */ 576 NULL, /* key dup */ 577 NULL, /* val dup */ 578 dictSdsKeyCompare, /* key compare */ 579 NULL, /* key destructor */ 580 NULL /* val destructor */ 581 }; 582 583 /* Command table. sds string -> command struct pointer. */ 584 dictType commandTableDictType = { 585 dictSdsCaseHash, /* hash function */ 586 NULL, /* key dup */ 587 NULL, /* val dup */ 588 dictSdsKeyCaseCompare, /* key compare */ 589 dictSdsDestructor, /* key destructor */ 590 NULL /* val destructor */ 591 }; 592 593 /* Hash type hash table (note that small hashes are represented with ziplists) */ 594 dictType hashDictType = { 595 dictEncObjHash, /* hash function */ 596 NULL, /* key dup */ 597 NULL, /* val dup */ 598 dictEncObjKeyCompare, /* key compare */ 599 dictObjectDestructor, /* key destructor */ 600 dictObjectDestructor /* val destructor */ 601 }; 602 603 /* Keylist hash table type has unencoded redis objects as keys and 604 * lists as values. It's used for blocking operations (BLPOP) and to 605 * map swapped keys to a list of clients waiting for this keys to be loaded. */ 606 dictType keylistDictType = { 607 dictObjHash, /* hash function */ 608 NULL, /* key dup */ 609 NULL, /* val dup */ 610 dictObjKeyCompare, /* key compare */ 611 dictObjectDestructor, /* key destructor */ 612 dictListDestructor /* val destructor */ 613 }; 614 615 /* Cluster nodes hash table, mapping nodes addresses 1.2.3.4:6379 to 616 * clusterNode structures. */ 617 dictType clusterNodesDictType = { 618 dictSdsHash, /* hash function */ 619 NULL, /* key dup */ 620 NULL, /* val dup */ 621 dictSdsKeyCompare, /* key compare */ 622 dictSdsDestructor, /* key destructor */ 623 NULL /* val destructor */ 624 }; 625 626 /* Cluster re-addition blacklist. This maps node IDs to the time 627 * we can re-add this node. The goal is to avoid readding a removed 628 * node for some time. */ 629 dictType clusterNodesBlackListDictType = { 630 dictSdsCaseHash, /* hash function */ 631 NULL, /* key dup */ 632 NULL, /* val dup */ 633 dictSdsKeyCaseCompare, /* key compare */ 634 dictSdsDestructor, /* key destructor */ 635 NULL /* val destructor */ 636 }; 637 638 /* Migrate cache dict type. */ 639 dictType migrateCacheDictType = { 640 dictSdsHash, /* hash function */ 641 NULL, /* key dup */ 642 NULL, /* val dup */ 643 dictSdsKeyCompare, /* key compare */ 644 dictSdsDestructor, /* key destructor */ 645 NULL /* val destructor */ 646 }; 647 648 /* Replication cached script dict (server.repl_scriptcache_dict). 649 * Keys are sds SHA1 strings, while values are not used at all in the current 650 * implementation. */ 651 dictType replScriptCacheDictType = { 652 dictSdsCaseHash, /* hash function */ 653 NULL, /* key dup */ 654 NULL, /* val dup */ 655 dictSdsKeyCaseCompare, /* key compare */ 656 dictSdsDestructor, /* key destructor */ 657 NULL /* val destructor */ 658 }; 659 660 int htNeedsResize(dict *dict) { 661 long long size, used; 662 663 size = dictSlots(dict); 664 used = dictSize(dict); 665 return (size && used && size > DICT_HT_INITIAL_SIZE && 666 (used*100/size < HASHTABLE_MIN_FILL)); 667 } 668 669 /* If the percentage of used slots in the HT reaches HASHTABLE_MIN_FILL 670 * we resize the hash table to save memory */ 671 void tryResizeHashTables(int dbid) { 672 if (htNeedsResize(server.db[dbid].dict)) 673 dictResize(server.db[dbid].dict); 674 if (htNeedsResize(server.db[dbid].expires)) 675 dictResize(server.db[dbid].expires); 676 } 677 678 /* Our hash table implementation performs rehashing incrementally while 679 * we write/read from the hash table. Still if the server is idle, the hash 680 * table will use two tables for a long time. So we try to use 1 millisecond 681 * of CPU time at every call of this function to perform some rehahsing. 682 * 683 * The function returns 1 if some rehashing was performed, otherwise 0 684 * is returned. */ 685 int incrementallyRehash(int dbid) { 686 /* Keys dictionary */ 687 if (dictIsRehashing(server.db[dbid].dict)) { 688 dictRehashMilliseconds(server.db[dbid].dict,1); 689 return 1; /* already used our millisecond for this loop... */ 690 } 691 /* Expires */ 692 if (dictIsRehashing(server.db[dbid].expires)) { 693 dictRehashMilliseconds(server.db[dbid].expires,1); 694 return 1; /* already used our millisecond for this loop... */ 695 } 696 return 0; 697 } 698 699 /* This function is called once a background process of some kind terminates, 700 * as we want to avoid resizing the hash tables when there is a child in order 701 * to play well with copy-on-write (otherwise when a resize happens lots of 702 * memory pages are copied). The goal of this function is to update the ability 703 * for dict.c to resize the hash tables accordingly to the fact we have o not 704 * running childs. */ 705 void updateDictResizePolicy(void) { 706 if (server.rdb_child_pid == -1 && server.aof_child_pid == -1) 707 dictEnableResize(); 708 else 709 dictDisableResize(); 710 } 711 712 /* ======================= Cron: called every 100 ms ======================== */ 713 714 /* Helper function for the activeExpireCycle() function. 715 * This function will try to expire the key that is stored in the hash table 716 * entry 'de' of the 'expires' hash table of a Redis database. 717 * 718 * If the key is found to be expired, it is removed from the database and 719 * 1 is returned. Otherwise no operation is performed and 0 is returned. 720 * 721 * When a key is expired, server.stat_expiredkeys is incremented. 722 * 723 * The parameter 'now' is the current time in milliseconds as is passed 724 * to the function to avoid too many gettimeofday() syscalls. */ 725 int activeExpireCycleTryExpire(redisDb *db, dictEntry *de, long long now) { 726 long long t = dictGetSignedIntegerVal(de); 727 if (now > t) { 728 sds key = dictGetKey(de); 729 robj *keyobj = createStringObject(key,sdslen(key)); 730 731 propagateExpire(db,keyobj); 732 dbDelete(db,keyobj); 733 notifyKeyspaceEvent(NOTIFY_EXPIRED, 734 "expired",keyobj,db->id); 735 decrRefCount(keyobj); 736 server.stat_expiredkeys++; 737 return 1; 738 } else { 739 return 0; 740 } 741 } 742 743 /* Try to expire a few timed out keys. The algorithm used is adaptive and 744 * will use few CPU cycles if there are few expiring keys, otherwise 745 * it will get more aggressive to avoid that too much memory is used by 746 * keys that can be removed from the keyspace. 747 * 748 * No more than CRON_DBS_PER_CALL databases are tested at every 749 * iteration. 750 * 751 * This kind of call is used when Redis detects that timelimit_exit is 752 * true, so there is more work to do, and we do it more incrementally from 753 * the beforeSleep() function of the event loop. 754 * 755 * Expire cycle type: 756 * 757 * If type is ACTIVE_EXPIRE_CYCLE_FAST the function will try to run a 758 * "fast" expire cycle that takes no longer than EXPIRE_FAST_CYCLE_DURATION 759 * microseconds, and is not repeated again before the same amount of time. 760 * 761 * If type is ACTIVE_EXPIRE_CYCLE_SLOW, that normal expire cycle is 762 * executed, where the time limit is a percentage of the REDIS_HZ period 763 * as specified by the REDIS_EXPIRELOOKUPS_TIME_PERC define. */ 764 765 void activeExpireCycle(int type) { 766 /* This function has some global state in order to continue the work 767 * incrementally across calls. */ 768 static unsigned int current_db = 0; /* Last DB tested. */ 769 static int timelimit_exit = 0; /* Time limit hit in previous call? */ 770 static long long last_fast_cycle = 0; /* When last fast cycle ran. */ 771 772 int j, iteration = 0; 773 int dbs_per_call = CRON_DBS_PER_CALL; 774 long long start = ustime(), timelimit; 775 776 if (type == ACTIVE_EXPIRE_CYCLE_FAST) { 777 /* Don't start a fast cycle if the previous cycle did not exited 778 * for time limt. Also don't repeat a fast cycle for the same period 779 * as the fast cycle total duration itself. */ 780 if (!timelimit_exit) return; 781 if (start < last_fast_cycle + ACTIVE_EXPIRE_CYCLE_FAST_DURATION*2) return; 782 last_fast_cycle = start; 783 } 784 785 /* We usually should test CRON_DBS_PER_CALL per iteration, with 786 * two exceptions: 787 * 788 * 1) Don't test more DBs than we have. 789 * 2) If last time we hit the time limit, we want to scan all DBs 790 * in this iteration, as there is work to do in some DB and we don't want 791 * expired keys to use memory for too much time. */ 792 if (dbs_per_call > server.dbnum || timelimit_exit) 793 dbs_per_call = server.dbnum; 794 795 /* We can use at max ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC percentage of CPU time 796 * per iteration. Since this function gets called with a frequency of 797 * server.hz times per second, the following is the max amount of 798 * microseconds we can spend in this function. */ 799 timelimit = 1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/server.hz/100; 800 timelimit_exit = 0; 801 if (timelimit <= 0) timelimit = 1; 802 803 if (type == ACTIVE_EXPIRE_CYCLE_FAST) 804 timelimit = ACTIVE_EXPIRE_CYCLE_FAST_DURATION; /* in microseconds. */ 805 806 for (j = 0; j < dbs_per_call; j++) { 807 int expired; 808 redisDb *db = server.db+(current_db % server.dbnum); 809 810 /* Increment the DB now so we are sure if we run out of time 811 * in the current DB we'll restart from the next. This allows to 812 * distribute the time evenly across DBs. */ 813 current_db++; 814 815 /* Continue to expire if at the end of the cycle more than 25% 816 * of the keys were expired. */ 817 do { 818 unsigned long num, slots; 819 long long now, ttl_sum; 820 int ttl_samples; 821 822 /* If there is nothing to expire try next DB ASAP. */ 823 if ((num = dictSize(db->expires)) == 0) { 824 db->avg_ttl = 0; 825 break; 826 } 827 slots = dictSlots(db->expires); 828 now = mstime(); 829 830 /* When there are less than 1% filled slots getting random 831 * keys is expensive, so stop here waiting for better times... 832 * The dictionary will be resized asap. */ 833 if (num && slots > DICT_HT_INITIAL_SIZE && 834 (num*100/slots < 1)) break; 835 836 /* The main collection cycle. Sample random keys among keys 837 * with an expire set, checking for expired ones. */ 838 expired = 0; 839 ttl_sum = 0; 840 ttl_samples = 0; 841 842 if (num > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP) 843 num = ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP; 844 845 while (num--) { 846 dictEntry *de; 847 long long ttl; 848 849 if ((de = dictGetRandomKey(db->expires)) == NULL) break; 850 ttl = dictGetSignedIntegerVal(de)-now; 851 if (activeExpireCycleTryExpire(db,de,now)) expired++; 852 if (ttl < 0) ttl = 0; 853 ttl_sum += ttl; 854 ttl_samples++; 855 } 856 857 /* Update the average TTL stats for this database. */ 858 if (ttl_samples) { 859 long long avg_ttl = ttl_sum/ttl_samples; 860 861 if (db->avg_ttl == 0) db->avg_ttl = avg_ttl; 862 /* Smooth the value averaging with the previous one. */ 863 db->avg_ttl = (db->avg_ttl+avg_ttl)/2; 864 } 865 866 /* We can't block forever here even if there are many keys to 867 * expire. So after a given amount of milliseconds return to the 868 * caller waiting for the other active expire cycle. */ 869 iteration++; 870 if ((iteration & 0xf) == 0) { /* check once every 16 iterations. */ 871 long long elapsed = ustime()-start; 872 873 latencyAddSampleIfNeeded("expire-cycle",elapsed/1000); 874 if (elapsed > timelimit) timelimit_exit = 1; 875 } 876 if (timelimit_exit) return; 877 /* We don't repeat the cycle if there are less than 25% of keys 878 * found expired in the current DB. */ 879 } while (expired > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP/4); 880 } 881 } 882 883 unsigned int getLRUClock(void) { 884 return (mstime()/LRU_CLOCK_RESOLUTION) & LRU_CLOCK_MAX; 885 } 886 887 /* Add a sample to the operations per second array of samples. */ 888 void trackInstantaneousMetric(int metric, long long current_reading) { 889 long long t = mstime() - server.inst_metric[metric].last_sample_time; 890 long long ops = current_reading - 891 server.inst_metric[metric].last_sample_count; 892 long long ops_sec; 893 894 ops_sec = t > 0 ? (ops*1000/t) : 0; 895 896 server.inst_metric[metric].samples[server.inst_metric[metric].idx] = 897 ops_sec; 898 server.inst_metric[metric].idx++; 899 server.inst_metric[metric].idx %= STATS_METRIC_SAMPLES; 900 server.inst_metric[metric].last_sample_time = mstime(); 901 server.inst_metric[metric].last_sample_count = current_reading; 902 } 903 904 /* Return the mean of all the samples. */ 905 long long getInstantaneousMetric(int metric) { 906 int j; 907 long long sum = 0; 908 909 for (j = 0; j < STATS_METRIC_SAMPLES; j++) 910 sum += server.inst_metric[metric].samples[j]; 911 return sum / STATS_METRIC_SAMPLES; 912 } 913 914 /* Check for timeouts. Returns non-zero if the client was terminated. 915 * The function gets the current time in milliseconds as argument since 916 * it gets called multiple times in a loop, so calling gettimeofday() for 917 * each iteration would be costly without any actual gain. */ 918 int clientsCronHandleTimeout(client *c, mstime_t now_ms) { 919 time_t now = now_ms/1000; 920 921 if (server.maxidletime && 922 !(c->flags & CLIENT_SLAVE) && /* no timeout for slaves */ 923 !(c->flags & CLIENT_MASTER) && /* no timeout for masters */ 924 !(c->flags & CLIENT_BLOCKED) && /* no timeout for BLPOP */ 925 !(c->flags & CLIENT_PUBSUB) && /* no timeout for Pub/Sub clients */ 926 (now - c->lastinteraction > server.maxidletime)) 927 { 928 serverLog(LL_VERBOSE,"Closing idle client"); 929 freeClient(c); 930 return 1; 931 } else if (c->flags & CLIENT_BLOCKED) { 932 /* Blocked OPS timeout is handled with milliseconds resolution. 933 * However note that the actual resolution is limited by 934 * server.hz. */ 935 936 if (c->bpop.timeout != 0 && c->bpop.timeout < now_ms) { 937 /* Handle blocking operation specific timeout. */ 938 replyToBlockedClientTimedOut(c); 939 unblockClient(c); 940 } else if (server.cluster_enabled) { 941 /* Cluster: handle unblock & redirect of clients blocked 942 * into keys no longer served by this server. */ 943 if (clusterRedirectBlockedClientIfNeeded(c)) 944 unblockClient(c); 945 } 946 } 947 return 0; 948 } 949 950 /* The client query buffer is an sds.c string that can end with a lot of 951 * free space not used, this function reclaims space if needed. 952 * 953 * The function always returns 0 as it never terminates the client. */ 954 int clientsCronResizeQueryBuffer(client *c) { 955 size_t querybuf_size = sdsAllocSize(c->querybuf); 956 time_t idletime = server.unixtime - c->lastinteraction; 957 958 /* There are two conditions to resize the query buffer: 959 * 1) Query buffer is > BIG_ARG and too big for latest peak. 960 * 2) Client is inactive and the buffer is bigger than 1k. */ 961 if (((querybuf_size > PROTO_MBULK_BIG_ARG) && 962 (querybuf_size/(c->querybuf_peak+1)) > 2) || 963 (querybuf_size > 1024 && idletime > 2)) 964 { 965 /* Only resize the query buffer if it is actually wasting space. */ 966 if (sdsavail(c->querybuf) > 1024) { 967 c->querybuf = sdsRemoveFreeSpace(c->querybuf); 968 } 969 } 970 /* Reset the peak again to capture the peak memory usage in the next 971 * cycle. */ 972 c->querybuf_peak = 0; 973 return 0; 974 } 975 976 #define CLIENTS_CRON_MIN_ITERATIONS 5 977 void clientsCron(void) { 978 /* Make sure to process at least numclients/server.hz of clients 979 * per call. Since this function is called server.hz times per second 980 * we are sure that in the worst case we process all the clients in 1 981 * second. */ 982 int numclients = listLength(server.clients); 983 int iterations = numclients/server.hz; 984 mstime_t now = mstime(); 985 986 /* Process at least a few clients while we are at it, even if we need 987 * to process less than CLIENTS_CRON_MIN_ITERATIONS to meet our contract 988 * of processing each client once per second. */ 989 if (iterations < CLIENTS_CRON_MIN_ITERATIONS) 990 iterations = (numclients < CLIENTS_CRON_MIN_ITERATIONS) ? 991 numclients : CLIENTS_CRON_MIN_ITERATIONS; 992 993 while(listLength(server.clients) && iterations--) { 994 client *c; 995 listNode *head; 996 997 /* Rotate the list, take the current head, process. 998 * This way if the client must be removed from the list it's the 999 * first element and we don't incur into O(N) computation. */ 1000 listRotate(server.clients); 1001 head = listFirst(server.clients); 1002 c = listNodeValue(head); 1003 /* The following functions do different service checks on the client. 1004 * The protocol is that they return non-zero if the client was 1005 * terminated. */ 1006 if (clientsCronHandleTimeout(c,now)) continue; 1007 if (clientsCronResizeQueryBuffer(c)) continue; 1008 } 1009 } 1010 1011 /* This function handles 'background' operations we are required to do 1012 * incrementally in Redis databases, such as active key expiring, resizing, 1013 * rehashing. */ 1014 void databasesCron(void) { 1015 /* Expire keys by random sampling. Not required for slaves 1016 * as master will synthesize DELs for us. */ 1017 if (server.active_expire_enabled && server.masterhost == NULL) 1018 activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW); 1019 1020 /* Perform hash tables rehashing if needed, but only if there are no 1021 * other processes saving the DB on disk. Otherwise rehashing is bad 1022 * as will cause a lot of copy-on-write of memory pages. */ 1023 if (server.rdb_child_pid == -1 && server.aof_child_pid == -1) { 1024 /* We use global counters so if we stop the computation at a given 1025 * DB we'll be able to start from the successive in the next 1026 * cron loop iteration. */ 1027 static unsigned int resize_db = 0; 1028 static unsigned int rehash_db = 0; 1029 int dbs_per_call = CRON_DBS_PER_CALL; 1030 int j; 1031 1032 /* Don't test more DBs than we have. */ 1033 if (dbs_per_call > server.dbnum) dbs_per_call = server.dbnum; 1034 1035 /* Resize */ 1036 for (j = 0; j < dbs_per_call; j++) { 1037 tryResizeHashTables(resize_db % server.dbnum); 1038 resize_db++; 1039 } 1040 1041 /* Rehash */ 1042 if (server.activerehashing) { 1043 for (j = 0; j < dbs_per_call; j++) { 1044 int work_done = incrementallyRehash(rehash_db % server.dbnum); 1045 rehash_db++; 1046 if (work_done) { 1047 /* If the function did some work, stop here, we'll do 1048 * more at the next cron loop. */ 1049 break; 1050 } 1051 } 1052 } 1053 } 1054 } 1055 1056 /* We take a cached value of the unix time in the global state because with 1057 * virtual memory and aging there is to store the current time in objects at 1058 * every object access, and accuracy is not needed. To access a global var is 1059 * a lot faster than calling time(NULL) */ 1060 void updateCachedTime(void) { 1061 server.unixtime = time(NULL); 1062 server.mstime = mstime(); 1063 } 1064 1065 /* This is our timer interrupt, called server.hz times per second. 1066 * Here is where we do a number of things that need to be done asynchronously. 1067 * For instance: 1068 * 1069 * - Active expired keys collection (it is also performed in a lazy way on 1070 * lookup). 1071 * - Software watchdog. 1072 * - Update some statistic. 1073 * - Incremental rehashing of the DBs hash tables. 1074 * - Triggering BGSAVE / AOF rewrite, and handling of terminated children. 1075 * - Clients timeout of different kinds. 1076 * - Replication reconnection. 1077 * - Many more... 1078 * 1079 * Everything directly called here will be called server.hz times per second, 1080 * so in order to throttle execution of things we want to do less frequently 1081 * a macro is used: run_with_period(milliseconds) { .... } 1082 */ 1083 1084 int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { 1085 int j; 1086 UNUSED(eventLoop); 1087 UNUSED(id); 1088 UNUSED(clientData); 1089 1090 /* Software watchdog: deliver the SIGALRM that will reach the signal 1091 * handler if we don't return here fast enough. */ 1092 if (server.watchdog_period) watchdogScheduleSignal(server.watchdog_period); 1093 1094 /* Update the time cache. */ 1095 updateCachedTime(); 1096 1097 run_with_period(100) { 1098 trackInstantaneousMetric(STATS_METRIC_COMMAND,server.stat_numcommands); 1099 trackInstantaneousMetric(STATS_METRIC_NET_INPUT, 1100 server.stat_net_input_bytes); 1101 trackInstantaneousMetric(STATS_METRIC_NET_OUTPUT, 1102 server.stat_net_output_bytes); 1103 } 1104 1105 /* We have just LRU_BITS bits per object for LRU information. 1106 * So we use an (eventually wrapping) LRU clock. 1107 * 1108 * Note that even if the counter wraps it's not a big problem, 1109 * everything will still work but some object will appear younger 1110 * to Redis. However for this to happen a given object should never be 1111 * touched for all the time needed to the counter to wrap, which is 1112 * not likely. 1113 * 1114 * Note that you can change the resolution altering the 1115 * LRU_CLOCK_RESOLUTION define. */ 1116 server.lruclock = getLRUClock(); 1117 1118 /* Record the max memory used since the server was started. */ 1119 if (zmalloc_used_memory() > server.stat_peak_memory) 1120 server.stat_peak_memory = zmalloc_used_memory(); 1121 1122 /* Sample the RSS here since this is a relatively slow call. */ 1123 server.resident_set_size = zmalloc_get_rss(); 1124 1125 /* We received a SIGTERM, shutting down here in a safe way, as it is 1126 * not ok doing so inside the signal handler. */ 1127 if (server.shutdown_asap) { 1128 if (prepareForShutdown(SHUTDOWN_NOFLAGS) == C_OK) exit(0); 1129 serverLog(LL_WARNING,"SIGTERM received but errors trying to shut down the server, check the logs for more information"); 1130 server.shutdown_asap = 0; 1131 } 1132 1133 /* Show some info about non-empty databases */ 1134 run_with_period(5000) { 1135 for (j = 0; j < server.dbnum; j++) { 1136 long long size, used, vkeys; 1137 1138 size = dictSlots(server.db[j].dict); 1139 used = dictSize(server.db[j].dict); 1140 vkeys = dictSize(server.db[j].expires); 1141 if (used || vkeys) { 1142 serverLog(LL_VERBOSE,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size); 1143 /* dictPrintStats(server.dict); */ 1144 } 1145 } 1146 } 1147 1148 /* Show information about connected clients */ 1149 if (!server.sentinel_mode) { 1150 run_with_period(5000) { 1151 serverLog(LL_VERBOSE, 1152 "%lu clients connected (%lu slaves), %zu bytes in use", 1153 listLength(server.clients)-listLength(server.slaves), 1154 listLength(server.slaves), 1155 zmalloc_used_memory()); 1156 } 1157 } 1158 1159 /* We need to do a few operations on clients asynchronously. */ 1160 clientsCron(); 1161 1162 /* Handle background operations on Redis databases. */ 1163 databasesCron(); 1164 1165 /* Start a scheduled AOF rewrite if this was requested by the user while 1166 * a BGSAVE was in progress. */ 1167 if (server.rdb_child_pid == -1 && server.aof_child_pid == -1 && 1168 server.aof_rewrite_scheduled) 1169 { 1170 rewriteAppendOnlyFileBackground(); 1171 } 1172 1173 /* Check if a background saving or AOF rewrite in progress terminated. */ 1174 if (server.rdb_child_pid != -1 || server.aof_child_pid != -1 || 1175 ldbPendingChildren()) 1176 { 1177 int statloc; 1178 pid_t pid; 1179 1180 if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) { 1181 int exitcode = WEXITSTATUS(statloc); 1182 int bysignal = 0; 1183 1184 if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc); 1185 1186 if (pid == -1) { 1187 serverLog(LL_WARNING,"wait3() returned an error: %s. " 1188 "rdb_child_pid = %d, aof_child_pid = %d", 1189 strerror(errno), 1190 (int) server.rdb_child_pid, 1191 (int) server.aof_child_pid); 1192 } else if (pid == server.rdb_child_pid) { 1193 backgroundSaveDoneHandler(exitcode,bysignal); 1194 } else if (pid == server.aof_child_pid) { 1195 backgroundRewriteDoneHandler(exitcode,bysignal); 1196 } else { 1197 if (!ldbRemoveChild(pid)) { 1198 serverLog(LL_WARNING, 1199 "Warning, detected child with unmatched pid: %ld", 1200 (long)pid); 1201 } 1202 } 1203 updateDictResizePolicy(); 1204 } 1205 } else { 1206 /* If there is not a background saving/rewrite in progress check if 1207 * we have to save/rewrite now */ 1208 for (j = 0; j < server.saveparamslen; j++) { 1209 struct saveparam *sp = server.saveparams+j; 1210 1211 /* Save if we reached the given amount of changes, 1212 * the given amount of seconds, and if the latest bgsave was 1213 * successful or if, in case of an error, at least 1214 * CONFIG_BGSAVE_RETRY_DELAY seconds already elapsed. */ 1215 if (server.dirty >= sp->changes && 1216 server.unixtime-server.lastsave > sp->seconds && 1217 (server.unixtime-server.lastbgsave_try > 1218 CONFIG_BGSAVE_RETRY_DELAY || 1219 server.lastbgsave_status == C_OK)) 1220 { 1221 serverLog(LL_NOTICE,"%d changes in %d seconds. Saving...", 1222 sp->changes, (int)sp->seconds); 1223 rdbSaveBackground(server.rdb_filename); 1224 break; 1225 } 1226 } 1227 1228 /* Trigger an AOF rewrite if needed */ 1229 if (server.rdb_child_pid == -1 && 1230 server.aof_child_pid == -1 && 1231 server.aof_rewrite_perc && 1232 server.aof_current_size > server.aof_rewrite_min_size) 1233 { 1234 long long base = server.aof_rewrite_base_size ? 1235 server.aof_rewrite_base_size : 1; 1236 long long growth = (server.aof_current_size*100/base) - 100; 1237 if (growth >= server.aof_rewrite_perc) { 1238 serverLog(LL_NOTICE,"Starting automatic rewriting of AOF on %lld%% growth",growth); 1239 rewriteAppendOnlyFileBackground(); 1240 } 1241 } 1242 } 1243 1244 1245 /* AOF postponed flush: Try at every cron cycle if the slow fsync 1246 * completed. */ 1247 if (server.aof_flush_postponed_start) flushAppendOnlyFile(0); 1248 1249 /* AOF write errors: in this case we have a buffer to flush as well and 1250 * clear the AOF error in case of success to make the DB writable again, 1251 * however to try every second is enough in case of 'hz' is set to 1252 * an higher frequency. */ 1253 run_with_period(1000) { 1254 if (server.aof_last_write_status == C_ERR) 1255 flushAppendOnlyFile(0); 1256 } 1257 1258 /* Close clients that need to be closed asynchronous */ 1259 freeClientsInAsyncFreeQueue(); 1260 1261 /* Clear the paused clients flag if needed. */ 1262 clientsArePaused(); /* Don't check return value, just use the side effect. */ 1263 1264 /* Replication cron function -- used to reconnect to master and 1265 * to detect transfer failures. */ 1266 run_with_period(1000) replicationCron(); 1267 1268 /* Run the Redis Cluster cron. */ 1269 run_with_period(100) { 1270 if (server.cluster_enabled) clusterCron(); 1271 } 1272 1273 /* Run the Sentinel timer if we are in sentinel mode. */ 1274 run_with_period(100) { 1275 if (server.sentinel_mode) sentinelTimer(); 1276 } 1277 1278 /* Cleanup expired MIGRATE cached sockets. */ 1279 run_with_period(1000) { 1280 migrateCloseTimedoutSockets(); 1281 } 1282 1283 server.cronloops++; 1284 return 1000/server.hz; 1285 } 1286 1287 /* This function gets called every time Redis is entering the 1288 * main loop of the event driven library, that is, before to sleep 1289 * for ready file descriptors. */ 1290 void beforeSleep(struct aeEventLoop *eventLoop) { 1291 UNUSED(eventLoop); 1292 1293 /* Call the Redis Cluster before sleep function. Note that this function 1294 * may change the state of Redis Cluster (from ok to fail or vice versa), 1295 * so it's a good idea to call it before serving the unblocked clients 1296 * later in this function. */ 1297 if (server.cluster_enabled) clusterBeforeSleep(); 1298 1299 /* Run a fast expire cycle (the called function will return 1300 * ASAP if a fast cycle is not needed). */ 1301 if (server.active_expire_enabled && server.masterhost == NULL) 1302 activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST); 1303 1304 /* Send all the slaves an ACK request if at least one client blocked 1305 * during the previous event loop iteration. */ 1306 if (server.get_ack_from_slaves) { 1307 robj *argv[3]; 1308 1309 argv[0] = createStringObject("REPLCONF",8); 1310 argv[1] = createStringObject("GETACK",6); 1311 argv[2] = createStringObject("*",1); /* Not used argument. */ 1312 replicationFeedSlaves(server.slaves, server.slaveseldb, argv, 3); 1313 decrRefCount(argv[0]); 1314 decrRefCount(argv[1]); 1315 decrRefCount(argv[2]); 1316 server.get_ack_from_slaves = 0; 1317 } 1318 1319 /* Unblock all the clients blocked for synchronous replication 1320 * in WAIT. */ 1321 if (listLength(server.clients_waiting_acks)) 1322 processClientsWaitingReplicas(); 1323 1324 /* Try to process pending commands for clients that were just unblocked. */ 1325 if (listLength(server.unblocked_clients)) 1326 processUnblockedClients(); 1327 1328 /* Write the AOF buffer on disk */ 1329 flushAppendOnlyFile(0); 1330 1331 /* Handle writes with pending output buffers. */ 1332 handleClientsWithPendingWrites(); 1333 } 1334 1335 /* =========================== Server initialization ======================== */ 1336 1337 void createSharedObjects(void) { 1338 int j; 1339 1340 shared.crlf = createObject(OBJ_STRING,sdsnew("\r\n")); 1341 shared.ok = createObject(OBJ_STRING,sdsnew("+OK\r\n")); 1342 shared.err = createObject(OBJ_STRING,sdsnew("-ERR\r\n")); 1343 shared.emptybulk = createObject(OBJ_STRING,sdsnew("$0\r\n\r\n")); 1344 shared.czero = createObject(OBJ_STRING,sdsnew(":0\r\n")); 1345 shared.cone = createObject(OBJ_STRING,sdsnew(":1\r\n")); 1346 shared.cnegone = createObject(OBJ_STRING,sdsnew(":-1\r\n")); 1347 shared.nullbulk = createObject(OBJ_STRING,sdsnew("$-1\r\n")); 1348 shared.nullmultibulk = createObject(OBJ_STRING,sdsnew("*-1\r\n")); 1349 shared.emptymultibulk = createObject(OBJ_STRING,sdsnew("*0\r\n")); 1350 shared.pong = createObject(OBJ_STRING,sdsnew("+PONG\r\n")); 1351 shared.queued = createObject(OBJ_STRING,sdsnew("+QUEUED\r\n")); 1352 shared.emptyscan = createObject(OBJ_STRING,sdsnew("*2\r\n$1\r\n0\r\n*0\r\n")); 1353 shared.wrongtypeerr = createObject(OBJ_STRING,sdsnew( 1354 "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n")); 1355 shared.nokeyerr = createObject(OBJ_STRING,sdsnew( 1356 "-ERR no such key\r\n")); 1357 shared.syntaxerr = createObject(OBJ_STRING,sdsnew( 1358 "-ERR syntax error\r\n")); 1359 shared.sameobjecterr = createObject(OBJ_STRING,sdsnew( 1360 "-ERR source and destination objects are the same\r\n")); 1361 shared.outofrangeerr = createObject(OBJ_STRING,sdsnew( 1362 "-ERR index out of range\r\n")); 1363 shared.noscripterr = createObject(OBJ_STRING,sdsnew( 1364 "-NOSCRIPT No matching script. Please use EVAL.\r\n")); 1365 shared.loadingerr = createObject(OBJ_STRING,sdsnew( 1366 "-LOADING Redis is loading the dataset in memory\r\n")); 1367 shared.slowscripterr = createObject(OBJ_STRING,sdsnew( 1368 "-BUSY Redis is busy running a script. You can only call SCRIPT KILL or SHUTDOWN NOSAVE.\r\n")); 1369 shared.masterdownerr = createObject(OBJ_STRING,sdsnew( 1370 "-MASTERDOWN Link with MASTER is down and slave-serve-stale-data is set to 'no'.\r\n")); 1371 shared.bgsaveerr = createObject(OBJ_STRING,sdsnew( 1372 "-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")); 1373 shared.roslaveerr = createObject(OBJ_STRING,sdsnew( 1374 "-READONLY You can't write against a read only slave.\r\n")); 1375 shared.noautherr = createObject(OBJ_STRING,sdsnew( 1376 "-NOAUTH Authentication required.\r\n")); 1377 shared.oomerr = createObject(OBJ_STRING,sdsnew( 1378 "-OOM command not allowed when used memory > 'maxmemory'.\r\n")); 1379 shared.execaborterr = createObject(OBJ_STRING,sdsnew( 1380 "-EXECABORT Transaction discarded because of previous errors.\r\n")); 1381 shared.noreplicaserr = createObject(OBJ_STRING,sdsnew( 1382 "-NOREPLICAS Not enough good slaves to write.\r\n")); 1383 shared.busykeyerr = createObject(OBJ_STRING,sdsnew( 1384 "-BUSYKEY Target key name already exists.\r\n")); 1385 shared.space = createObject(OBJ_STRING,sdsnew(" ")); 1386 shared.colon = createObject(OBJ_STRING,sdsnew(":")); 1387 shared.plus = createObject(OBJ_STRING,sdsnew("+")); 1388 1389 for (j = 0; j < PROTO_SHARED_SELECT_CMDS; j++) { 1390 char dictid_str[64]; 1391 int dictid_len; 1392 1393 dictid_len = ll2string(dictid_str,sizeof(dictid_str),j); 1394 shared.select[j] = createObject(OBJ_STRING, 1395 sdscatprintf(sdsempty(), 1396 "*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n", 1397 dictid_len, dictid_str)); 1398 } 1399 shared.messagebulk = createStringObject("$7\r\nmessage\r\n",13); 1400 shared.pmessagebulk = createStringObject("$8\r\npmessage\r\n",14); 1401 shared.subscribebulk = createStringObject("$9\r\nsubscribe\r\n",15); 1402 shared.unsubscribebulk = createStringObject("$11\r\nunsubscribe\r\n",18); 1403 shared.psubscribebulk = createStringObject("$10\r\npsubscribe\r\n",17); 1404 shared.punsubscribebulk = createStringObject("$12\r\npunsubscribe\r\n",19); 1405 shared.del = createStringObject("DEL",3); 1406 shared.rpop = createStringObject("RPOP",4); 1407 shared.lpop = createStringObject("LPOP",4); 1408 shared.lpush = createStringObject("LPUSH",5); 1409 for (j = 0; j < OBJ_SHARED_INTEGERS; j++) { 1410 shared.integers[j] = createObject(OBJ_STRING,(void*)(long)j); 1411 shared.integers[j]->encoding = OBJ_ENCODING_INT; 1412 } 1413 for (j = 0; j < OBJ_SHARED_BULKHDR_LEN; j++) { 1414 shared.mbulkhdr[j] = createObject(OBJ_STRING, 1415 sdscatprintf(sdsempty(),"*%d\r\n",j)); 1416 shared.bulkhdr[j] = createObject(OBJ_STRING, 1417 sdscatprintf(sdsempty(),"$%d\r\n",j)); 1418 } 1419 /* The following two shared objects, minstring and maxstrings, are not 1420 * actually used for their value but as a special object meaning 1421 * respectively the minimum possible string and the maximum possible 1422 * string in string comparisons for the ZRANGEBYLEX command. */ 1423 shared.minstring = createStringObject("minstring",9); 1424 shared.maxstring = createStringObject("maxstring",9); 1425 } 1426 1427 void initServerConfig(void) { 1428 int j; 1429 1430 getRandomHexChars(server.runid,CONFIG_RUN_ID_SIZE); 1431 server.configfile = NULL; 1432 server.executable = NULL; 1433 server.hz = CONFIG_DEFAULT_HZ; 1434 server.runid[CONFIG_RUN_ID_SIZE] = '\0'; 1435 server.arch_bits = (sizeof(long) == 8) ? 64 : 32; 1436 server.port = CONFIG_DEFAULT_SERVER_PORT; 1437 server.tcp_backlog = CONFIG_DEFAULT_TCP_BACKLOG; 1438 server.bindaddr_count = 0; 1439 server.unixsocket = NULL; 1440 server.unixsocketperm = CONFIG_DEFAULT_UNIX_SOCKET_PERM; 1441 server.ipfd_count = 0; 1442 server.sofd = -1; 1443 server.dbnum = CONFIG_DEFAULT_DBNUM; 1444 server.verbosity = CONFIG_DEFAULT_VERBOSITY; 1445 server.maxidletime = CONFIG_DEFAULT_CLIENT_TIMEOUT; 1446 server.tcpkeepalive = CONFIG_DEFAULT_TCP_KEEPALIVE; 1447 server.active_expire_enabled = 1; 1448 server.client_max_querybuf_len = PROTO_MAX_QUERYBUF_LEN; 1449 server.saveparams = NULL; 1450 server.loading = 0; 1451 server.logfile = zstrdup(CONFIG_DEFAULT_LOGFILE); 1452 server.syslog_enabled = CONFIG_DEFAULT_SYSLOG_ENABLED; 1453 server.syslog_ident = zstrdup(CONFIG_DEFAULT_SYSLOG_IDENT); 1454 server.syslog_facility = LOG_LOCAL0; 1455 server.daemonize = CONFIG_DEFAULT_DAEMONIZE; 1456 server.supervised = 0; 1457 server.supervised_mode = SUPERVISED_NONE; 1458 server.aof_state = AOF_OFF; 1459 server.aof_fsync = CONFIG_DEFAULT_AOF_FSYNC; 1460 server.aof_no_fsync_on_rewrite = CONFIG_DEFAULT_AOF_NO_FSYNC_ON_REWRITE; 1461 server.aof_rewrite_perc = AOF_REWRITE_PERC; 1462 server.aof_rewrite_min_size = AOF_REWRITE_MIN_SIZE; 1463 server.aof_rewrite_base_size = 0; 1464 server.aof_rewrite_scheduled = 0; 1465 server.aof_last_fsync = time(NULL); 1466 server.aof_rewrite_time_last = -1; 1467 server.aof_rewrite_time_start = -1; 1468 server.aof_lastbgrewrite_status = C_OK; 1469 server.aof_delayed_fsync = 0; 1470 server.aof_fd = -1; 1471 server.aof_selected_db = -1; /* Make sure the first time will not match */ 1472 server.aof_flush_postponed_start = 0; 1473 server.aof_rewrite_incremental_fsync = CONFIG_DEFAULT_AOF_REWRITE_INCREMENTAL_FSYNC; 1474 server.aof_load_truncated = CONFIG_DEFAULT_AOF_LOAD_TRUNCATED; 1475 server.pidfile = NULL; 1476 server.rdb_filename = zstrdup(CONFIG_DEFAULT_RDB_FILENAME); 1477 server.aof_filename = zstrdup(CONFIG_DEFAULT_AOF_FILENAME); 1478 server.requirepass = NULL; 1479 server.rdb_compression = CONFIG_DEFAULT_RDB_COMPRESSION; 1480 server.rdb_checksum = CONFIG_DEFAULT_RDB_CHECKSUM; 1481 server.stop_writes_on_bgsave_err = CONFIG_DEFAULT_STOP_WRITES_ON_BGSAVE_ERROR; 1482 server.activerehashing = CONFIG_DEFAULT_ACTIVE_REHASHING; 1483 server.notify_keyspace_events = 0; 1484 server.maxclients = CONFIG_DEFAULT_MAX_CLIENTS; 1485 server.bpop_blocked_clients = 0; 1486 server.maxmemory = CONFIG_DEFAULT_MAXMEMORY; 1487 server.maxmemory_policy = CONFIG_DEFAULT_MAXMEMORY_POLICY; 1488 server.maxmemory_samples = CONFIG_DEFAULT_MAXMEMORY_SAMPLES; 1489 server.hash_max_ziplist_entries = OBJ_HASH_MAX_ZIPLIST_ENTRIES; 1490 server.hash_max_ziplist_value = OBJ_HASH_MAX_ZIPLIST_VALUE; 1491 server.list_max_ziplist_size = OBJ_LIST_MAX_ZIPLIST_SIZE; 1492 server.list_compress_depth = OBJ_LIST_COMPRESS_DEPTH; 1493 server.set_max_intset_entries = OBJ_SET_MAX_INTSET_ENTRIES; 1494 server.zset_max_ziplist_entries = OBJ_ZSET_MAX_ZIPLIST_ENTRIES; 1495 server.zset_max_ziplist_value = OBJ_ZSET_MAX_ZIPLIST_VALUE; 1496 server.hll_sparse_max_bytes = CONFIG_DEFAULT_HLL_SPARSE_MAX_BYTES; 1497 server.shutdown_asap = 0; 1498 server.repl_ping_slave_period = CONFIG_DEFAULT_REPL_PING_SLAVE_PERIOD; 1499 server.repl_timeout = CONFIG_DEFAULT_REPL_TIMEOUT; 1500 server.repl_min_slaves_to_write = CONFIG_DEFAULT_MIN_SLAVES_TO_WRITE; 1501 server.repl_min_slaves_max_lag = CONFIG_DEFAULT_MIN_SLAVES_MAX_LAG; 1502 server.cluster_enabled = 0; 1503 server.cluster_node_timeout = CLUSTER_DEFAULT_NODE_TIMEOUT; 1504 server.cluster_migration_barrier = CLUSTER_DEFAULT_MIGRATION_BARRIER; 1505 server.cluster_slave_validity_factor = CLUSTER_DEFAULT_SLAVE_VALIDITY; 1506 server.cluster_require_full_coverage = CLUSTER_DEFAULT_REQUIRE_FULL_COVERAGE; 1507 server.cluster_configfile = zstrdup(CONFIG_DEFAULT_CLUSTER_CONFIG_FILE); 1508 server.migrate_cached_sockets = dictCreate(&migrateCacheDictType,NULL); 1509 server.next_client_id = 1; /* Client IDs, start from 1 .*/ 1510 server.loading_process_events_interval_bytes = (1024*1024*2); 1511 1512 server.lruclock = getLRUClock(); 1513 resetServerSaveParams(); 1514 1515 appendServerSaveParams(60*60,1); /* save after 1 hour and 1 change */ 1516 appendServerSaveParams(300,100); /* save after 5 minutes and 100 changes */ 1517 appendServerSaveParams(60,10000); /* save after 1 minute and 10000 changes */ 1518 /* Replication related */ 1519 server.masterauth = NULL; 1520 server.masterhost = NULL; 1521 server.masterport = 6379; 1522 server.master = NULL; 1523 server.cached_master = NULL; 1524 server.repl_master_initial_offset = -1; 1525 server.repl_state = REPL_STATE_NONE; 1526 server.repl_syncio_timeout = CONFIG_REPL_SYNCIO_TIMEOUT; 1527 server.repl_serve_stale_data = CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA; 1528 server.repl_slave_ro = CONFIG_DEFAULT_SLAVE_READ_ONLY; 1529 server.repl_down_since = 0; /* Never connected, repl is down since EVER. */ 1530 server.repl_disable_tcp_nodelay = CONFIG_DEFAULT_REPL_DISABLE_TCP_NODELAY; 1531 server.repl_diskless_sync = CONFIG_DEFAULT_REPL_DISKLESS_SYNC; 1532 server.repl_diskless_sync_delay = CONFIG_DEFAULT_REPL_DISKLESS_SYNC_DELAY; 1533 server.slave_priority = CONFIG_DEFAULT_SLAVE_PRIORITY; 1534 server.master_repl_offset = 0; 1535 1536 /* Replication partial resync backlog */ 1537 server.repl_backlog = NULL; 1538 server.repl_backlog_size = CONFIG_DEFAULT_REPL_BACKLOG_SIZE; 1539 server.repl_backlog_histlen = 0; 1540 server.repl_backlog_idx = 0; 1541 server.repl_backlog_off = 0; 1542 server.repl_backlog_time_limit = CONFIG_DEFAULT_REPL_BACKLOG_TIME_LIMIT; 1543 server.repl_no_slaves_since = time(NULL); 1544 1545 /* Client output buffer limits */ 1546 for (j = 0; j < CLIENT_TYPE_OBUF_COUNT; j++) 1547 server.client_obuf_limits[j] = clientBufferLimitsDefaults[j]; 1548 1549 /* Double constants initialization */ 1550 R_Zero = 0.0; 1551 R_PosInf = 1.0/R_Zero; 1552 R_NegInf = -1.0/R_Zero; 1553 R_Nan = R_Zero/R_Zero; 1554 1555 /* Command table -- we initiialize it here as it is part of the 1556 * initial configuration, since command names may be changed via 1557 * redis.conf using the rename-command directive. */ 1558 server.commands = dictCreate(&commandTableDictType,NULL); 1559 server.orig_commands = dictCreate(&commandTableDictType,NULL); 1560 populateCommandTable(); 1561 server.delCommand = lookupCommandByCString("del"); 1562 server.multiCommand = lookupCommandByCString("multi"); 1563 server.lpushCommand = lookupCommandByCString("lpush"); 1564 server.lpopCommand = lookupCommandByCString("lpop"); 1565 server.rpopCommand = lookupCommandByCString("rpop"); 1566 server.sremCommand = lookupCommandByCString("srem"); 1567 server.execCommand = lookupCommandByCString("exec"); 1568 1569 /* Slow log */ 1570 server.slowlog_log_slower_than = CONFIG_DEFAULT_SLOWLOG_LOG_SLOWER_THAN; 1571 server.slowlog_max_len = CONFIG_DEFAULT_SLOWLOG_MAX_LEN; 1572 1573 /* Latency monitor */ 1574 server.latency_monitor_threshold = CONFIG_DEFAULT_LATENCY_MONITOR_THRESHOLD; 1575 1576 /* Debugging */ 1577 server.assert_failed = "<no assertion failed>"; 1578 server.assert_file = "<no file>"; 1579 server.assert_line = 0; 1580 server.bug_report_start = 0; 1581 server.watchdog_period = 0; 1582 } 1583 1584 extern char **environ; 1585 1586 /* Restart the server, executing the same executable that started this 1587 * instance, with the same arguments and configuration file. 1588 * 1589 * The function is designed to directly call execve() so that the new 1590 * server instance will retain the PID of the previous one. 1591 * 1592 * The list of flags, that may be bitwise ORed together, alter the 1593 * behavior of this function: 1594 * 1595 * RESTART_SERVER_NONE No flags. 1596 * RESTART_SERVER_GRACEFULLY Do a proper shutdown before restarting. 1597 * RESTART_SERVER_CONFIG_REWRITE Rewrite the config file before restarting. 1598 * 1599 * On success the function does not return, because the process turns into 1600 * a different process. On error C_ERR is returned. */ 1601 int restartServer(int flags, mstime_t delay) { 1602 int j; 1603 1604 /* Check if we still have accesses to the executable that started this 1605 * server instance. */ 1606 if (access(server.executable,X_OK) == -1) return C_ERR; 1607 1608 /* Config rewriting. */ 1609 if (flags & RESTART_SERVER_CONFIG_REWRITE && 1610 server.configfile && 1611 rewriteConfig(server.configfile) == -1) return C_ERR; 1612 1613 /* Perform a proper shutdown. */ 1614 if (flags & RESTART_SERVER_GRACEFULLY && 1615 prepareForShutdown(SHUTDOWN_NOFLAGS) != C_OK) return C_ERR; 1616 1617 /* Close all file descriptors, with the exception of stdin, stdout, strerr 1618 * which are useful if we restart a Redis server which is not daemonized. */ 1619 for (j = 3; j < (int)server.maxclients + 1024; j++) close(j); 1620 1621 /* Execute the server with the original command line. */ 1622 if (delay) usleep(delay*1000); 1623 execve(server.executable,server.exec_argv,environ); 1624 1625 /* If an error occurred here, there is nothing we can do, but exit. */ 1626 _exit(1); 1627 1628 return C_ERR; /* Never reached. */ 1629 } 1630 1631 /* This function will try to raise the max number of open files accordingly to 1632 * the configured max number of clients. It also reserves a number of file 1633 * descriptors (CONFIG_MIN_RESERVED_FDS) for extra operations of 1634 * persistence, listening sockets, log files and so forth. 1635 * 1636 * If it will not be possible to set the limit accordingly to the configured 1637 * max number of clients, the function will do the reverse setting 1638 * server.maxclients to the value that we can actually handle. */ 1639 void adjustOpenFilesLimit(void) { 1640 rlim_t maxfiles = server.maxclients+CONFIG_MIN_RESERVED_FDS; 1641 struct rlimit limit; 1642 1643 if (getrlimit(RLIMIT_NOFILE,&limit) == -1) { 1644 serverLog(LL_WARNING,"Unable to obtain the current NOFILE limit (%s), assuming 1024 and setting the max clients configuration accordingly.", 1645 strerror(errno)); 1646 server.maxclients = 1024-CONFIG_MIN_RESERVED_FDS; 1647 } else { 1648 rlim_t oldlimit = limit.rlim_cur; 1649 1650 /* Set the max number of files if the current limit is not enough 1651 * for our needs. */ 1652 if (oldlimit < maxfiles) { 1653 rlim_t bestlimit; 1654 int setrlimit_error = 0; 1655 1656 /* Try to set the file limit to match 'maxfiles' or at least 1657 * to the higher value supported less than maxfiles. */ 1658 bestlimit = maxfiles; 1659 while(bestlimit > oldlimit) { 1660 rlim_t decr_step = 16; 1661 1662 limit.rlim_cur = bestlimit; 1663 limit.rlim_max = bestlimit; 1664 if (setrlimit(RLIMIT_NOFILE,&limit) != -1) break; 1665 setrlimit_error = errno; 1666 1667 /* We failed to set file limit to 'bestlimit'. Try with a 1668 * smaller limit decrementing by a few FDs per iteration. */ 1669 if (bestlimit < decr_step) break; 1670 bestlimit -= decr_step; 1671 } 1672 1673 /* Assume that the limit we get initially is still valid if 1674 * our last try was even lower. */ 1675 if (bestlimit < oldlimit) bestlimit = oldlimit; 1676 1677 if (bestlimit < maxfiles) { 1678 int old_maxclients = server.maxclients; 1679 server.maxclients = bestlimit-CONFIG_MIN_RESERVED_FDS; 1680 if (server.maxclients < 1) { 1681 serverLog(LL_WARNING,"Your current 'ulimit -n' " 1682 "of %llu is not enough for the server to start. " 1683 "Please increase your open file limit to at least " 1684 "%llu. Exiting.", 1685 (unsigned long long) oldlimit, 1686 (unsigned long long) maxfiles); 1687 exit(1); 1688 } 1689 serverLog(LL_WARNING,"You requested maxclients of %d " 1690 "requiring at least %llu max file descriptors.", 1691 old_maxclients, 1692 (unsigned long long) maxfiles); 1693 serverLog(LL_WARNING,"Server can't set maximum open files " 1694 "to %llu because of OS error: %s.", 1695 (unsigned long long) maxfiles, strerror(setrlimit_error)); 1696 serverLog(LL_WARNING,"Current maximum open files is %llu. " 1697 "maxclients has been reduced to %d to compensate for " 1698 "low ulimit. " 1699 "If you need higher maxclients increase 'ulimit -n'.", 1700 (unsigned long long) bestlimit, server.maxclients); 1701 } else { 1702 serverLog(LL_NOTICE,"Increased maximum number of open files " 1703 "to %llu (it was originally set to %llu).", 1704 (unsigned long long) maxfiles, 1705 (unsigned long long) oldlimit); 1706 } 1707 } 1708 } 1709 } 1710 1711 /* Check that server.tcp_backlog can be actually enforced in Linux according 1712 * to the value of /proc/sys/net/core/somaxconn, or warn about it. */ 1713 void checkTcpBacklogSettings(void) { 1714 #ifdef HAVE_PROC_SOMAXCONN 1715 FILE *fp = fopen("/proc/sys/net/core/somaxconn","r"); 1716 char buf[1024]; 1717 if (!fp) return; 1718 if (fgets(buf,sizeof(buf),fp) != NULL) { 1719 int somaxconn = atoi(buf); 1720 if (somaxconn > 0 && somaxconn < server.tcp_backlog) { 1721 serverLog(LL_WARNING,"WARNING: The TCP backlog setting of %d cannot be enforced because /proc/sys/net/core/somaxconn is set to the lower value of %d.", server.tcp_backlog, somaxconn); 1722 } 1723 } 1724 fclose(fp); 1725 #endif 1726 } 1727 1728 /* Initialize a set of file descriptors to listen to the specified 'port' 1729 * binding the addresses specified in the Redis server configuration. 1730 * 1731 * The listening file descriptors are stored in the integer array 'fds' 1732 * and their number is set in '*count'. 1733 * 1734 * The addresses to bind are specified in the global server.bindaddr array 1735 * and their number is server.bindaddr_count. If the server configuration 1736 * contains no specific addresses to bind, this function will try to 1737 * bind * (all addresses) for both the IPv4 and IPv6 protocols. 1738 * 1739 * On success the function returns C_OK. 1740 * 1741 * On error the function returns C_ERR. For the function to be on 1742 * error, at least one of the server.bindaddr addresses was 1743 * impossible to bind, or no bind addresses were specified in the server 1744 * configuration but the function is not able to bind * for at least 1745 * one of the IPv4 or IPv6 protocols. */ 1746 int listenToPort(int port, int *fds, int *count) { 1747 int j; 1748 1749 /* Force binding of 0.0.0.0 if no bind address is specified, always 1750 * entering the loop if j == 0. */ 1751 if (server.bindaddr_count == 0) server.bindaddr[0] = NULL; 1752 for (j = 0; j < server.bindaddr_count || j == 0; j++) { 1753 if (server.bindaddr[j] == NULL) { 1754 /* Bind * for both IPv6 and IPv4, we enter here only if 1755 * server.bindaddr_count == 0. */ 1756 fds[*count] = anetTcp6Server(server.neterr,port,NULL, 1757 server.tcp_backlog); 1758 if (fds[*count] != ANET_ERR) { 1759 anetNonBlock(NULL,fds[*count]); 1760 (*count)++; 1761 } 1762 fds[*count] = anetTcpServer(server.neterr,port,NULL, 1763 server.tcp_backlog); 1764 if (fds[*count] != ANET_ERR) { 1765 anetNonBlock(NULL,fds[*count]); 1766 (*count)++; 1767 } 1768 /* Exit the loop if we were able to bind * on IPv4 or IPv6, 1769 * otherwise fds[*count] will be ANET_ERR and we'll print an 1770 * error and return to the caller with an error. */ 1771 if (*count) break; 1772 } else if (strchr(server.bindaddr[j],':')) { 1773 /* Bind IPv6 address. */ 1774 fds[*count] = anetTcp6Server(server.neterr,port,server.bindaddr[j], 1775 server.tcp_backlog); 1776 } else { 1777 /* Bind IPv4 address. */ 1778 fds[*count] = anetTcpServer(server.neterr,port,server.bindaddr[j], 1779 server.tcp_backlog); 1780 } 1781 if (fds[*count] == ANET_ERR) { 1782 serverLog(LL_WARNING, 1783 "Creating Server TCP listening socket %s:%d: %s", 1784 server.bindaddr[j] ? server.bindaddr[j] : "*", 1785 port, server.neterr); 1786 return C_ERR; 1787 } 1788 anetNonBlock(NULL,fds[*count]); 1789 (*count)++; 1790 } 1791 return C_OK; 1792 } 1793 1794 /* Resets the stats that we expose via INFO or other means that we want 1795 * to reset via CONFIG RESETSTAT. The function is also used in order to 1796 * initialize these fields in initServer() at server startup. */ 1797 void resetServerStats(void) { 1798 int j; 1799 1800 server.stat_numcommands = 0; 1801 server.stat_numconnections = 0; 1802 server.stat_expiredkeys = 0; 1803 server.stat_evictedkeys = 0; 1804 server.stat_keyspace_misses = 0; 1805 server.stat_keyspace_hits = 0; 1806 server.stat_fork_time = 0; 1807 server.stat_fork_rate = 0; 1808 server.stat_rejected_conn = 0; 1809 server.stat_sync_full = 0; 1810 server.stat_sync_partial_ok = 0; 1811 server.stat_sync_partial_err = 0; 1812 for (j = 0; j < STATS_METRIC_COUNT; j++) { 1813 server.inst_metric[j].idx = 0; 1814 server.inst_metric[j].last_sample_time = mstime(); 1815 server.inst_metric[j].last_sample_count = 0; 1816 memset(server.inst_metric[j].samples,0, 1817 sizeof(server.inst_metric[j].samples)); 1818 } 1819 server.stat_net_input_bytes = 0; 1820 server.stat_net_output_bytes = 0; 1821 server.aof_delayed_fsync = 0; 1822 } 1823 1824 void initServer(void) { 1825 int j; 1826 1827 signal(SIGHUP, SIG_IGN); 1828 signal(SIGPIPE, SIG_IGN); 1829 setupSignalHandlers(); 1830 1831 if (server.syslog_enabled) { 1832 openlog(server.syslog_ident, LOG_PID | LOG_NDELAY | LOG_NOWAIT, 1833 server.syslog_facility); 1834 } 1835 1836 server.pid = getpid(); 1837 server.current_client = NULL; 1838 server.clients = listCreate(); 1839 server.clients_to_close = listCreate(); 1840 server.slaves = listCreate(); 1841 server.monitors = listCreate(); 1842 server.clients_pending_write = listCreate(); 1843 server.slaveseldb = -1; /* Force to emit the first SELECT command. */ 1844 server.unblocked_clients = listCreate(); 1845 server.ready_keys = listCreate(); 1846 server.clients_waiting_acks = listCreate(); 1847 server.get_ack_from_slaves = 0; 1848 server.clients_paused = 0; 1849 server.system_memory_size = zmalloc_get_memory_size(); 1850 1851 createSharedObjects(); 1852 adjustOpenFilesLimit(); 1853 server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR); 1854 server.db = zmalloc(sizeof(redisDb)*server.dbnum); 1855 1856 /* Open the TCP listening socket for the user commands. */ 1857 if (server.port != 0 && 1858 listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR) 1859 exit(1); 1860 1861 /* Open the listening Unix domain socket. */ 1862 if (server.unixsocket != NULL) { 1863 unlink(server.unixsocket); /* don't care if this fails */ 1864 server.sofd = anetUnixServer(server.neterr,server.unixsocket, 1865 server.unixsocketperm, server.tcp_backlog); 1866 if (server.sofd == ANET_ERR) { 1867 serverLog(LL_WARNING, "Opening Unix socket: %s", server.neterr); 1868 exit(1); 1869 } 1870 anetNonBlock(NULL,server.sofd); 1871 } 1872 1873 /* Abort if there are no listening sockets at all. */ 1874 if (server.ipfd_count == 0 && server.sofd < 0) { 1875 serverLog(LL_WARNING, "Configured to not listen anywhere, exiting."); 1876 exit(1); 1877 } 1878 1879 /* Create the Redis databases, and initialize other internal state. */ 1880 for (j = 0; j < server.dbnum; j++) { 1881 server.db[j].dict = dictCreate(&dbDictType,NULL); 1882 server.db[j].expires = dictCreate(&keyptrDictType,NULL); 1883 server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL); 1884 server.db[j].ready_keys = dictCreate(&setDictType,NULL); 1885 server.db[j].watched_keys = dictCreate(&keylistDictType,NULL); 1886 server.db[j].eviction_pool = evictionPoolAlloc(); 1887 server.db[j].id = j; 1888 server.db[j].avg_ttl = 0; 1889 } 1890 server.pubsub_channels = dictCreate(&keylistDictType,NULL); 1891 server.pubsub_patterns = listCreate(); 1892 listSetFreeMethod(server.pubsub_patterns,freePubsubPattern); 1893 listSetMatchMethod(server.pubsub_patterns,listMatchPubsubPattern); 1894 server.cronloops = 0; 1895 server.rdb_child_pid = -1; 1896 server.aof_child_pid = -1; 1897 server.rdb_child_type = RDB_CHILD_TYPE_NONE; 1898 aofRewriteBufferReset(); 1899 server.aof_buf = sdsempty(); 1900 server.lastsave = time(NULL); /* At startup we consider the DB saved. */ 1901 server.lastbgsave_try = 0; /* At startup we never tried to BGSAVE. */ 1902 server.rdb_save_time_last = -1; 1903 server.rdb_save_time_start = -1; 1904 server.dirty = 0; 1905 resetServerStats(); 1906 /* A few stats we don't want to reset: server startup time, and peak mem. */ 1907 server.stat_starttime = time(NULL); 1908 server.stat_peak_memory = 0; 1909 server.resident_set_size = 0; 1910 server.lastbgsave_status = C_OK; 1911 server.aof_last_write_status = C_OK; 1912 server.aof_last_write_errno = 0; 1913 server.repl_good_slaves_count = 0; 1914 updateCachedTime(); 1915 1916 /* Create the serverCron() time event, that's our main way to process 1917 * background operations. */ 1918 if(aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) { 1919 serverPanic("Can't create the serverCron time event."); 1920 exit(1); 1921 } 1922 1923 /* Create an event handler for accepting new connections in TCP and Unix 1924 * domain sockets. */ 1925 for (j = 0; j < server.ipfd_count; j++) { 1926 if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE, 1927 acceptTcpHandler,NULL) == AE_ERR) 1928 { 1929 serverPanic( 1930 "Unrecoverable error creating server.ipfd file event."); 1931 } 1932 } 1933 if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE, 1934 acceptUnixHandler,NULL) == AE_ERR) serverPanic("Unrecoverable error creating server.sofd file event."); 1935 1936 /* Open the AOF file if needed. */ 1937 if (server.aof_state == AOF_ON) { 1938 server.aof_fd = open(server.aof_filename, 1939 O_WRONLY|O_APPEND|O_CREAT,0644); 1940 if (server.aof_fd == -1) { 1941 serverLog(LL_WARNING, "Can't open the append-only file: %s", 1942 strerror(errno)); 1943 exit(1); 1944 } 1945 } 1946 1947 /* 32 bit instances are limited to 4GB of address space, so if there is 1948 * no explicit limit in the user provided configuration we set a limit 1949 * at 3 GB using maxmemory with 'noeviction' policy'. This avoids 1950 * useless crashes of the Redis instance for out of memory. */ 1951 if (server.arch_bits == 32 && server.maxmemory == 0) { 1952 serverLog(LL_WARNING,"Warning: 32 bit instance detected but no memory limit set. Setting 3 GB maxmemory limit with 'noeviction' policy now."); 1953 server.maxmemory = 3072LL*(1024*1024); /* 3 GB */ 1954 server.maxmemory_policy = MAXMEMORY_NO_EVICTION; 1955 } 1956 1957 if (server.cluster_enabled) clusterInit(); 1958 replicationScriptCacheInit(); 1959 scriptingInit(1); 1960 slowlogInit(); 1961 latencyMonitorInit(); 1962 bioInit(); 1963 } 1964 1965 /* Populates the Redis Command Table starting from the hard coded list 1966 * we have on top of redis.c file. */ 1967 void populateCommandTable(void) { 1968 int j; 1969 int numcommands = sizeof(redisCommandTable)/sizeof(struct redisCommand); 1970 1971 for (j = 0; j < numcommands; j++) { 1972 struct redisCommand *c = redisCommandTable+j; 1973 char *f = c->sflags; 1974 int retval1, retval2; 1975 1976 while(*f != '\0') { 1977 switch(*f) { 1978 case 'w': c->flags |= CMD_WRITE; break; 1979 case 'r': c->flags |= CMD_READONLY; break; 1980 case 'm': c->flags |= CMD_DENYOOM; break; 1981 case 'a': c->flags |= CMD_ADMIN; break; 1982 case 'p': c->flags |= CMD_PUBSUB; break; 1983 case 's': c->flags |= CMD_NOSCRIPT; break; 1984 case 'R': c->flags |= CMD_RANDOM; break; 1985 case 'S': c->flags |= CMD_SORT_FOR_SCRIPT; break; 1986 case 'l': c->flags |= CMD_LOADING; break; 1987 case 't': c->flags |= CMD_STALE; break; 1988 case 'M': c->flags |= CMD_SKIP_MONITOR; break; 1989 case 'k': c->flags |= CMD_ASKING; break; 1990 case 'F': c->flags |= CMD_FAST; break; 1991 default: serverPanic("Unsupported command flag"); break; 1992 } 1993 f++; 1994 } 1995 1996 retval1 = dictAdd(server.commands, sdsnew(c->name), c); 1997 /* Populate an additional dictionary that will be unaffected 1998 * by rename-command statements in redis.conf. */ 1999 retval2 = dictAdd(server.orig_commands, sdsnew(c->name), c); 2000 serverAssert(retval1 == DICT_OK && retval2 == DICT_OK); 2001 } 2002 } 2003 2004 void resetCommandTableStats(void) { 2005 int numcommands = sizeof(redisCommandTable)/sizeof(struct redisCommand); 2006 int j; 2007 2008 for (j = 0; j < numcommands; j++) { 2009 struct redisCommand *c = redisCommandTable+j; 2010 2011 c->microseconds = 0; 2012 c->calls = 0; 2013 } 2014 } 2015 2016 /* ========================== Redis OP Array API ============================ */ 2017 2018 void redisOpArrayInit(redisOpArray *oa) { 2019 oa->ops = NULL; 2020 oa->numops = 0; 2021 } 2022 2023 int redisOpArrayAppend(redisOpArray *oa, struct redisCommand *cmd, int dbid, 2024 robj **argv, int argc, int target) 2025 { 2026 redisOp *op; 2027 2028 oa->ops = zrealloc(oa->ops,sizeof(redisOp)*(oa->numops+1)); 2029 op = oa->ops+oa->numops; 2030 op->cmd = cmd; 2031 op->dbid = dbid; 2032 op->argv = argv; 2033 op->argc = argc; 2034 op->target = target; 2035 oa->numops++; 2036 return oa->numops; 2037 } 2038 2039 void redisOpArrayFree(redisOpArray *oa) { 2040 while(oa->numops) { 2041 int j; 2042 redisOp *op; 2043 2044 oa->numops--; 2045 op = oa->ops+oa->numops; 2046 for (j = 0; j < op->argc; j++) 2047 decrRefCount(op->argv[j]); 2048 zfree(op->argv); 2049 } 2050 zfree(oa->ops); 2051 } 2052 2053 /* ====================== Commands lookup and execution ===================== */ 2054 2055 struct redisCommand *lookupCommand(sds name) { 2056 return dictFetchValue(server.commands, name); 2057 } 2058 2059 struct redisCommand *lookupCommandByCString(char *s) { 2060 struct redisCommand *cmd; 2061 sds name = sdsnew(s); 2062 2063 cmd = dictFetchValue(server.commands, name); 2064 sdsfree(name); 2065 return cmd; 2066 } 2067 2068 /* Lookup the command in the current table, if not found also check in 2069 * the original table containing the original command names unaffected by 2070 * redis.conf rename-command statement. 2071 * 2072 * This is used by functions rewriting the argument vector such as 2073 * rewriteClientCommandVector() in order to set client->cmd pointer 2074 * correctly even if the command was renamed. */ 2075 struct redisCommand *lookupCommandOrOriginal(sds name) { 2076 struct redisCommand *cmd = dictFetchValue(server.commands, name); 2077 2078 if (!cmd) cmd = dictFetchValue(server.orig_commands,name); 2079 return cmd; 2080 } 2081 2082 /* Propagate the specified command (in the context of the specified database id) 2083 * to AOF and Slaves. 2084 * 2085 * flags are an xor between: 2086 * + PROPAGATE_NONE (no propagation of command at all) 2087 * + PROPAGATE_AOF (propagate into the AOF file if is enabled) 2088 * + PROPAGATE_REPL (propagate into the replication link) 2089 * 2090 * This should not be used inside commands implementation. Use instead 2091 * alsoPropagate(), preventCommandPropagation(), forceCommandPropagation(). 2092 */ 2093 void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, 2094 int flags) 2095 { 2096 if (server.aof_state != AOF_OFF && flags & PROPAGATE_AOF) 2097 feedAppendOnlyFile(cmd,dbid,argv,argc); 2098 if (flags & PROPAGATE_REPL) 2099 replicationFeedSlaves(server.slaves,dbid,argv,argc); 2100 } 2101 2102 /* Used inside commands to schedule the propagation of additional commands 2103 * after the current command is propagated to AOF / Replication. 2104 * 2105 * 'cmd' must be a pointer to the Redis command to replicate, dbid is the 2106 * database ID the command should be propagated into. 2107 * Arguments of the command to propagte are passed as an array of redis 2108 * objects pointers of len 'argc', using the 'argv' vector. 2109 * 2110 * The function does not take a reference to the passed 'argv' vector, 2111 * so it is up to the caller to release the passed argv (but it is usually 2112 * stack allocated). The function autoamtically increments ref count of 2113 * passed objects, so the caller does not need to. */ 2114 void alsoPropagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, 2115 int target) 2116 { 2117 robj **argvcopy; 2118 int j; 2119 2120 if (server.loading) return; /* No propagation during loading. */ 2121 2122 argvcopy = zmalloc(sizeof(robj*)*argc); 2123 for (j = 0; j < argc; j++) { 2124 argvcopy[j] = argv[j]; 2125 incrRefCount(argv[j]); 2126 } 2127 redisOpArrayAppend(&server.also_propagate,cmd,dbid,argvcopy,argc,target); 2128 } 2129 2130 /* It is possible to call the function forceCommandPropagation() inside a 2131 * Redis command implementation in order to to force the propagation of a 2132 * specific command execution into AOF / Replication. */ 2133 void forceCommandPropagation(client *c, int flags) { 2134 if (flags & PROPAGATE_REPL) c->flags |= CLIENT_FORCE_REPL; 2135 if (flags & PROPAGATE_AOF) c->flags |= CLIENT_FORCE_AOF; 2136 } 2137 2138 /* Avoid that the executed command is propagated at all. This way we 2139 * are free to just propagate what we want using the alsoPropagate() 2140 * API. */ 2141 void preventCommandPropagation(client *c) { 2142 c->flags |= CLIENT_PREVENT_PROP; 2143 } 2144 2145 /* AOF specific version of preventCommandPropagation(). */ 2146 void preventCommandAOF(client *c) { 2147 c->flags |= CLIENT_PREVENT_AOF_PROP; 2148 } 2149 2150 /* Replication specific version of preventCommandPropagation(). */ 2151 void preventCommandReplication(client *c) { 2152 c->flags |= CLIENT_PREVENT_REPL_PROP; 2153 } 2154 2155 /* Call() is the core of Redis execution of a command. 2156 * 2157 * The following flags can be passed: 2158 * CMD_CALL_NONE No flags. 2159 * CMD_CALL_SLOWLOG Check command speed and log in the slow log if needed. 2160 * CMD_CALL_STATS Populate command stats. 2161 * CMD_CALL_PROPAGATE_AOF Append command to AOF if it modified the dataset 2162 * or if the client flags are forcing propagation. 2163 * CMD_CALL_PROPAGATE_REPL Send command to salves if it modified the dataset 2164 * or if the client flags are forcing propagation. 2165 * CMD_CALL_PROPAGATE Alias for PROPAGATE_AOF|PROPAGATE_REPL. 2166 * CMD_CALL_FULL Alias for SLOWLOG|STATS|PROPAGATE. 2167 * 2168 * The exact propagation behavior depends on the client flags. 2169 * Specifically: 2170 * 2171 * 1. If the client flags CLIENT_FORCE_AOF or CLIENT_FORCE_REPL are set 2172 * and assuming the corresponding CMD_CALL_PROPAGATE_AOF/REPL is set 2173 * in the call flags, then the command is propagated even if the 2174 * dataset was not affected by the command. 2175 * 2. If the client flags CLIENT_PREVENT_REPL_PROP or CLIENT_PREVENT_AOF_PROP 2176 * are set, the propagation into AOF or to slaves is not performed even 2177 * if the command modified the dataset. 2178 * 2179 * Note that regardless of the client flags, if CMD_CALL_PROPAGATE_AOF 2180 * or CMD_CALL_PROPAGATE_REPL are not set, then respectively AOF or 2181 * slaves propagation will never occur. 2182 * 2183 * Client flags are modified by the implementation of a given command 2184 * using the following API: 2185 * 2186 * forceCommandPropagation(client *c, int flags); 2187 * preventCommandPropagation(client *c); 2188 * preventCommandAOF(client *c); 2189 * preventCommandReplication(client *c); 2190 * 2191 */ 2192 void call(client *c, int flags) { 2193 long long dirty, start, duration; 2194 int client_old_flags = c->flags; 2195 2196 /* Sent the command to clients in MONITOR mode, only if the commands are 2197 * not generated from reading an AOF. */ 2198 if (listLength(server.monitors) && 2199 !server.loading && 2200 !(c->cmd->flags & (CMD_SKIP_MONITOR|CMD_ADMIN))) 2201 { 2202 replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc); 2203 } 2204 2205 /* Initialization: clear the flags that must be set by the command on 2206 * demand, and initialize the array for additional commands propagation. */ 2207 c->flags &= ~(CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP); 2208 redisOpArrayInit(&server.also_propagate); 2209 2210 /* Call the command. */ 2211 dirty = server.dirty; 2212 start = ustime(); 2213 c->cmd->proc(c); 2214 duration = ustime()-start; 2215 dirty = server.dirty-dirty; 2216 if (dirty < 0) dirty = 0; 2217 2218 /* When EVAL is called loading the AOF we don't want commands called 2219 * from Lua to go into the slowlog or to populate statistics. */ 2220 if (server.loading && c->flags & CLIENT_LUA) 2221 flags &= ~(CMD_CALL_SLOWLOG | CMD_CALL_STATS); 2222 2223 /* If the caller is Lua, we want to force the EVAL caller to propagate 2224 * the script if the command flag or client flag are forcing the 2225 * propagation. */ 2226 if (c->flags & CLIENT_LUA && server.lua_caller) { 2227 if (c->flags & CLIENT_FORCE_REPL) 2228 server.lua_caller->flags |= CLIENT_FORCE_REPL; 2229 if (c->flags & CLIENT_FORCE_AOF) 2230 server.lua_caller->flags |= CLIENT_FORCE_AOF; 2231 } 2232 2233 /* Log the command into the Slow log if needed, and populate the 2234 * per-command statistics that we show in INFO commandstats. */ 2235 if (flags & CMD_CALL_SLOWLOG && c->cmd->proc != execCommand) { 2236 char *latency_event = (c->cmd->flags & CMD_FAST) ? 2237 "fast-command" : "command"; 2238 latencyAddSampleIfNeeded(latency_event,duration/1000); 2239 slowlogPushEntryIfNeeded(c->argv,c->argc,duration); 2240 } 2241 if (flags & CMD_CALL_STATS) { 2242 c->cmd->microseconds += duration; 2243 c->cmd->calls++; 2244 } 2245 2246 /* Propagate the command into the AOF and replication link */ 2247 if (flags & CMD_CALL_PROPAGATE && 2248 (c->flags & CLIENT_PREVENT_PROP) != CLIENT_PREVENT_PROP) 2249 { 2250 int propagate_flags = PROPAGATE_NONE; 2251 2252 /* Check if the command operated changes in the data set. If so 2253 * set for replication / AOF propagation. */ 2254 if (dirty) propagate_flags |= (PROPAGATE_AOF|PROPAGATE_REPL); 2255 2256 /* If the client forced AOF / replication of the command, set 2257 * the flags regardless of the command effects on the data set. */ 2258 if (c->flags & CLIENT_FORCE_REPL) propagate_flags |= PROPAGATE_REPL; 2259 if (c->flags & CLIENT_FORCE_AOF) propagate_flags |= PROPAGATE_AOF; 2260 2261 /* However prevent AOF / replication propagation if the command 2262 * implementatino called preventCommandPropagation() or similar, 2263 * or if we don't have the call() flags to do so. */ 2264 if (c->flags & CLIENT_PREVENT_REPL_PROP || 2265 !(flags & CMD_CALL_PROPAGATE_REPL)) 2266 propagate_flags &= ~PROPAGATE_REPL; 2267 if (c->flags & CLIENT_PREVENT_AOF_PROP || 2268 !(flags & CMD_CALL_PROPAGATE_AOF)) 2269 propagate_flags &= ~PROPAGATE_AOF; 2270 2271 /* Call propagate() only if at least one of AOF / replication 2272 * propagation is needed. */ 2273 if (propagate_flags != PROPAGATE_NONE) 2274 propagate(c->cmd,c->db->id,c->argv,c->argc,propagate_flags); 2275 } 2276 2277 /* Restore the old replication flags, since call() can be executed 2278 * recursively. */ 2279 c->flags &= ~(CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP); 2280 c->flags |= client_old_flags & 2281 (CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP); 2282 2283 /* Handle the alsoPropagate() API to handle commands that want to propagate 2284 * multiple separated commands. Note that alsoPropagate() is not affected 2285 * by CLIENT_PREVENT_PROP flag. */ 2286 if (server.also_propagate.numops) { 2287 int j; 2288 redisOp *rop; 2289 2290 if (flags & CMD_CALL_PROPAGATE) { 2291 for (j = 0; j < server.also_propagate.numops; j++) { 2292 rop = &server.also_propagate.ops[j]; 2293 int target = rop->target; 2294 /* Whatever the command wish is, we honor the call() flags. */ 2295 if (!(flags&CMD_CALL_PROPAGATE_AOF)) target &= ~PROPAGATE_AOF; 2296 if (!(flags&CMD_CALL_PROPAGATE_REPL)) target &= ~PROPAGATE_REPL; 2297 if (target) 2298 propagate(rop->cmd,rop->dbid,rop->argv,rop->argc,target); 2299 } 2300 } 2301 redisOpArrayFree(&server.also_propagate); 2302 } 2303 server.stat_numcommands++; 2304 } 2305 2306 /* If this function gets called we already read a whole 2307 * command, arguments are in the client argv/argc fields. 2308 * processCommand() execute the command or prepare the 2309 * server for a bulk read from the client. 2310 * 2311 * If C_OK is returned the client is still alive and valid and 2312 * other operations can be performed by the caller. Otherwise 2313 * if C_ERR is returned the client was destroyed (i.e. after QUIT). */ 2314 int processCommand(client *c) { 2315 /* The QUIT command is handled separately. Normal command procs will 2316 * go through checking for replication and QUIT will cause trouble 2317 * when FORCE_REPLICATION is enabled and would be implemented in 2318 * a regular command proc. */ 2319 if (!strcasecmp(c->argv[0]->ptr,"quit")) { 2320 addReply(c,shared.ok); 2321 c->flags |= CLIENT_CLOSE_AFTER_REPLY; 2322 return C_ERR; 2323 } 2324 2325 /* Now lookup the command and check ASAP about trivial error conditions 2326 * such as wrong arity, bad command name and so forth. */ 2327 c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr); 2328 if (!c->cmd) { 2329 flagTransaction(c); 2330 addReplyErrorFormat(c,"unknown command '%s'", 2331 (char*)c->argv[0]->ptr); 2332 return C_OK; 2333 } else if ((c->cmd->arity > 0 && c->cmd->arity != c->argc) || 2334 (c->argc < -c->cmd->arity)) { 2335 flagTransaction(c); 2336 addReplyErrorFormat(c,"wrong number of arguments for '%s' command", 2337 c->cmd->name); 2338 return C_OK; 2339 } 2340 2341 /* Check if the user is authenticated */ 2342 if (server.requirepass && !c->authenticated && c->cmd->proc != authCommand) 2343 { 2344 flagTransaction(c); 2345 addReply(c,shared.noautherr); 2346 return C_OK; 2347 } 2348 2349 /* If cluster is enabled perform the cluster redirection here. 2350 * However we don't perform the redirection if: 2351 * 1) The sender of this command is our master. 2352 * 2) The command has no key arguments. */ 2353 if (server.cluster_enabled && 2354 !(c->flags & CLIENT_MASTER) && 2355 !(c->flags & CLIENT_LUA && 2356 server.lua_caller->flags & CLIENT_MASTER) && 2357 !(c->cmd->getkeys_proc == NULL && c->cmd->firstkey == 0)) 2358 { 2359 int hashslot; 2360 2361 if (server.cluster->state != CLUSTER_OK) { 2362 flagTransaction(c); 2363 clusterRedirectClient(c,NULL,0,CLUSTER_REDIR_DOWN_STATE); 2364 return C_OK; 2365 } else { 2366 int error_code; 2367 clusterNode *n = getNodeByQuery(c,c->cmd,c->argv,c->argc,&hashslot,&error_code); 2368 if (n == NULL || n != server.cluster->myself) { 2369 flagTransaction(c); 2370 clusterRedirectClient(c,n,hashslot,error_code); 2371 return C_OK; 2372 } 2373 } 2374 } 2375 2376 /* Handle the maxmemory directive. 2377 * 2378 * First we try to free some memory if possible (if there are volatile 2379 * keys in the dataset). If there are not the only thing we can do 2380 * is returning an error. */ 2381 if (server.maxmemory) { 2382 int retval = freeMemoryIfNeeded(); 2383 /* freeMemoryIfNeeded may flush slave output buffers. This may result 2384 * into a slave, that may be the active client, to be freed. */ 2385 if (server.current_client == NULL) return C_ERR; 2386 2387 /* It was impossible to free enough memory, and the command the client 2388 * is trying to execute is denied during OOM conditions? Error. */ 2389 if ((c->cmd->flags & CMD_DENYOOM) && retval == C_ERR) { 2390 flagTransaction(c); 2391 addReply(c, shared.oomerr); 2392 return C_OK; 2393 } 2394 } 2395 2396 /* Don't accept write commands if there are problems persisting on disk 2397 * and if this is a master instance. */ 2398 if (((server.stop_writes_on_bgsave_err && 2399 server.saveparamslen > 0 && 2400 server.lastbgsave_status == C_ERR) || 2401 server.aof_last_write_status == C_ERR) && 2402 server.masterhost == NULL && 2403 (c->cmd->flags & CMD_WRITE || 2404 c->cmd->proc == pingCommand)) 2405 { 2406 flagTransaction(c); 2407 if (server.aof_last_write_status == C_OK) 2408 addReply(c, shared.bgsaveerr); 2409 else 2410 addReplySds(c, 2411 sdscatprintf(sdsempty(), 2412 "-MISCONF Errors writing to the AOF file: %s\r\n", 2413 strerror(server.aof_last_write_errno))); 2414 return C_OK; 2415 } 2416 2417 /* Don't accept write commands if there are not enough good slaves and 2418 * user configured the min-slaves-to-write option. */ 2419 if (server.masterhost == NULL && 2420 server.repl_min_slaves_to_write && 2421 server.repl_min_slaves_max_lag && 2422 c->cmd->flags & CMD_WRITE && 2423 server.repl_good_slaves_count < server.repl_min_slaves_to_write) 2424 { 2425 flagTransaction(c); 2426 addReply(c, shared.noreplicaserr); 2427 return C_OK; 2428 } 2429 2430 /* Don't accept write commands if this is a read only slave. But 2431 * accept write commands if this is our master. */ 2432 if (server.masterhost && server.repl_slave_ro && 2433 !(c->flags & CLIENT_MASTER) && 2434 c->cmd->flags & CMD_WRITE) 2435 { 2436 addReply(c, shared.roslaveerr); 2437 return C_OK; 2438 } 2439 2440 /* Only allow SUBSCRIBE and UNSUBSCRIBE in the context of Pub/Sub */ 2441 if (c->flags & CLIENT_PUBSUB && 2442 c->cmd->proc != pingCommand && 2443 c->cmd->proc != subscribeCommand && 2444 c->cmd->proc != unsubscribeCommand && 2445 c->cmd->proc != psubscribeCommand && 2446 c->cmd->proc != punsubscribeCommand) { 2447 addReplyError(c,"only (P)SUBSCRIBE / (P)UNSUBSCRIBE / PING / QUIT allowed in this context"); 2448 return C_OK; 2449 } 2450 2451 /* Only allow INFO and SLAVEOF when slave-serve-stale-data is no and 2452 * we are a slave with a broken link with master. */ 2453 if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED && 2454 server.repl_serve_stale_data == 0 && 2455 !(c->cmd->flags & CMD_STALE)) 2456 { 2457 flagTransaction(c); 2458 addReply(c, shared.masterdownerr); 2459 return C_OK; 2460 } 2461 2462 /* Loading DB? Return an error if the command has not the 2463 * CMD_LOADING flag. */ 2464 if (server.loading && !(c->cmd->flags & CMD_LOADING)) { 2465 addReply(c, shared.loadingerr); 2466 return C_OK; 2467 } 2468 2469 /* Lua script too slow? Only allow a limited number of commands. */ 2470 if (server.lua_timedout && 2471 c->cmd->proc != authCommand && 2472 c->cmd->proc != replconfCommand && 2473 !(c->cmd->proc == shutdownCommand && 2474 c->argc == 2 && 2475 tolower(((char*)c->argv[1]->ptr)[0]) == 'n') && 2476 !(c->cmd->proc == scriptCommand && 2477 c->argc == 2 && 2478 tolower(((char*)c->argv[1]->ptr)[0]) == 'k')) 2479 { 2480 flagTransaction(c); 2481 addReply(c, shared.slowscripterr); 2482 return C_OK; 2483 } 2484 2485 /* Exec the command */ 2486 if (c->flags & CLIENT_MULTI && 2487 c->cmd->proc != execCommand && c->cmd->proc != discardCommand && 2488 c->cmd->proc != multiCommand && c->cmd->proc != watchCommand) 2489 { 2490 queueMultiCommand(c); 2491 addReply(c,shared.queued); 2492 } else { 2493 call(c,CMD_CALL_FULL); 2494 c->woff = server.master_repl_offset; 2495 if (listLength(server.ready_keys)) 2496 handleClientsBlockedOnLists(); 2497 } 2498 return C_OK; 2499 } 2500 2501 /*================================== Shutdown =============================== */ 2502 2503 /* Close listening sockets. Also unlink the unix domain socket if 2504 * unlink_unix_socket is non-zero. */ 2505 void closeListeningSockets(int unlink_unix_socket) { 2506 int j; 2507 2508 for (j = 0; j < server.ipfd_count; j++) close(server.ipfd[j]); 2509 if (server.sofd != -1) close(server.sofd); 2510 if (server.cluster_enabled) 2511 for (j = 0; j < server.cfd_count; j++) close(server.cfd[j]); 2512 if (unlink_unix_socket && server.unixsocket) { 2513 serverLog(LL_NOTICE,"Removing the unix socket file."); 2514 unlink(server.unixsocket); /* don't care if this fails */ 2515 } 2516 } 2517 2518 int prepareForShutdown(int flags) { 2519 int save = flags & SHUTDOWN_SAVE; 2520 int nosave = flags & SHUTDOWN_NOSAVE; 2521 2522 serverLog(LL_WARNING,"User requested shutdown..."); 2523 2524 /* Kill all the Lua debugger forked sessions. */ 2525 ldbKillForkedSessions(); 2526 2527 /* Kill the saving child if there is a background saving in progress. 2528 We want to avoid race conditions, for instance our saving child may 2529 overwrite the synchronous saving did by SHUTDOWN. */ 2530 if (server.rdb_child_pid != -1) { 2531 serverLog(LL_WARNING,"There is a child saving an .rdb. Killing it!"); 2532 kill(server.rdb_child_pid,SIGUSR1); 2533 rdbRemoveTempFile(server.rdb_child_pid); 2534 } 2535 2536 if (server.aof_state != AOF_OFF) { 2537 /* Kill the AOF saving child as the AOF we already have may be longer 2538 * but contains the full dataset anyway. */ 2539 if (server.aof_child_pid != -1) { 2540 /* If we have AOF enabled but haven't written the AOF yet, don't 2541 * shutdown or else the dataset will be lost. */ 2542 if (server.aof_state == AOF_WAIT_REWRITE) { 2543 serverLog(LL_WARNING, "Writing initial AOF, can't exit."); 2544 return C_ERR; 2545 } 2546 serverLog(LL_WARNING, 2547 "There is a child rewriting the AOF. Killing it!"); 2548 kill(server.aof_child_pid,SIGUSR1); 2549 } 2550 /* Append only file: fsync() the AOF and exit */ 2551 serverLog(LL_NOTICE,"Calling fsync() on the AOF file."); 2552 aof_fsync(server.aof_fd); 2553 } 2554 2555 /* Create a new RDB file before exiting. */ 2556 if ((server.saveparamslen > 0 && !nosave) || save) { 2557 serverLog(LL_NOTICE,"Saving the final RDB snapshot before exiting."); 2558 /* Snapshotting. Perform a SYNC SAVE and exit */ 2559 if (rdbSave(server.rdb_filename) != C_OK) { 2560 /* Ooops.. error saving! The best we can do is to continue 2561 * operating. Note that if there was a background saving process, 2562 * in the next cron() Redis will be notified that the background 2563 * saving aborted, handling special stuff like slaves pending for 2564 * synchronization... */ 2565 serverLog(LL_WARNING,"Error trying to save the DB, can't exit."); 2566 return C_ERR; 2567 } 2568 } 2569 2570 /* Remove the pid file if possible and needed. */ 2571 if (server.daemonize || server.pidfile) { 2572 serverLog(LL_NOTICE,"Removing the pid file."); 2573 unlink(server.pidfile); 2574 } 2575 2576 /* Best effort flush of slave output buffers, so that we hopefully 2577 * send them pending writes. */ 2578 flushSlavesOutputBuffers(); 2579 2580 /* Close the listening sockets. Apparently this allows faster restarts. */ 2581 closeListeningSockets(1); 2582 serverLog(LL_WARNING,"%s is now ready to exit, bye bye...", 2583 server.sentinel_mode ? "Sentinel" : "Redis"); 2584 return C_OK; 2585 } 2586 2587 /*================================== Commands =============================== */ 2588 2589 /* Return zero if strings are the same, non-zero if they are not. 2590 * The comparison is performed in a way that prevents an attacker to obtain 2591 * information about the nature of the strings just monitoring the execution 2592 * time of the function. 2593 * 2594 * Note that limiting the comparison length to strings up to 512 bytes we 2595 * can avoid leaking any information about the password length and any 2596 * possible branch misprediction related leak. 2597 */ 2598 int time_independent_strcmp(char *a, char *b) { 2599 char bufa[CONFIG_AUTHPASS_MAX_LEN], bufb[CONFIG_AUTHPASS_MAX_LEN]; 2600 /* The above two strlen perform len(a) + len(b) operations where either 2601 * a or b are fixed (our password) length, and the difference is only 2602 * relative to the length of the user provided string, so no information 2603 * leak is possible in the following two lines of code. */ 2604 unsigned int alen = strlen(a); 2605 unsigned int blen = strlen(b); 2606 unsigned int j; 2607 int diff = 0; 2608 2609 /* We can't compare strings longer than our static buffers. 2610 * Note that this will never pass the first test in practical circumstances 2611 * so there is no info leak. */ 2612 if (alen > sizeof(bufa) || blen > sizeof(bufb)) return 1; 2613 2614 memset(bufa,0,sizeof(bufa)); /* Constant time. */ 2615 memset(bufb,0,sizeof(bufb)); /* Constant time. */ 2616 /* Again the time of the following two copies is proportional to 2617 * len(a) + len(b) so no info is leaked. */ 2618 memcpy(bufa,a,alen); 2619 memcpy(bufb,b,blen); 2620 2621 /* Always compare all the chars in the two buffers without 2622 * conditional expressions. */ 2623 for (j = 0; j < sizeof(bufa); j++) { 2624 diff |= (bufa[j] ^ bufb[j]); 2625 } 2626 /* Length must be equal as well. */ 2627 diff |= alen ^ blen; 2628 return diff; /* If zero strings are the same. */ 2629 } 2630 2631 void authCommand(client *c) { 2632 if (!server.requirepass) { 2633 addReplyError(c,"Client sent AUTH, but no password is set"); 2634 } else if (!time_independent_strcmp(c->argv[1]->ptr, server.requirepass)) { 2635 c->authenticated = 1; 2636 addReply(c,shared.ok); 2637 } else { 2638 c->authenticated = 0; 2639 addReplyError(c,"invalid password"); 2640 } 2641 } 2642 2643 /* The PING command. It works in a different way if the client is in 2644 * in Pub/Sub mode. */ 2645 void pingCommand(client *c) { 2646 /* The command takes zero or one arguments. */ 2647 if (c->argc > 2) { 2648 addReplyErrorFormat(c,"wrong number of arguments for '%s' command", 2649 c->cmd->name); 2650 return; 2651 } 2652 2653 if (c->flags & CLIENT_PUBSUB) { 2654 addReply(c,shared.mbulkhdr[2]); 2655 addReplyBulkCBuffer(c,"pong",4); 2656 if (c->argc == 1) 2657 addReplyBulkCBuffer(c,"",0); 2658 else 2659 addReplyBulk(c,c->argv[1]); 2660 } else { 2661 if (c->argc == 1) 2662 addReply(c,shared.pong); 2663 else 2664 addReplyBulk(c,c->argv[1]); 2665 } 2666 } 2667 2668 void echoCommand(client *c) { 2669 addReplyBulk(c,c->argv[1]); 2670 } 2671 2672 void timeCommand(client *c) { 2673 struct timeval tv; 2674 2675 /* gettimeofday() can only fail if &tv is a bad address so we 2676 * don't check for errors. */ 2677 gettimeofday(&tv,NULL); 2678 addReplyMultiBulkLen(c,2); 2679 addReplyBulkLongLong(c,tv.tv_sec); 2680 addReplyBulkLongLong(c,tv.tv_usec); 2681 } 2682 2683 /* Helper function for addReplyCommand() to output flags. */ 2684 int addReplyCommandFlag(client *c, struct redisCommand *cmd, int f, char *reply) { 2685 if (cmd->flags & f) { 2686 addReplyStatus(c, reply); 2687 return 1; 2688 } 2689 return 0; 2690 } 2691 2692 /* Output the representation of a Redis command. Used by the COMMAND command. */ 2693 void addReplyCommand(client *c, struct redisCommand *cmd) { 2694 if (!cmd) { 2695 addReply(c, shared.nullbulk); 2696 } else { 2697 /* We are adding: command name, arg count, flags, first, last, offset */ 2698 addReplyMultiBulkLen(c, 6); 2699 addReplyBulkCString(c, cmd->name); 2700 addReplyLongLong(c, cmd->arity); 2701 2702 int flagcount = 0; 2703 void *flaglen = addDeferredMultiBulkLength(c); 2704 flagcount += addReplyCommandFlag(c,cmd,CMD_WRITE, "write"); 2705 flagcount += addReplyCommandFlag(c,cmd,CMD_READONLY, "readonly"); 2706 flagcount += addReplyCommandFlag(c,cmd,CMD_DENYOOM, "denyoom"); 2707 flagcount += addReplyCommandFlag(c,cmd,CMD_ADMIN, "admin"); 2708 flagcount += addReplyCommandFlag(c,cmd,CMD_PUBSUB, "pubsub"); 2709 flagcount += addReplyCommandFlag(c,cmd,CMD_NOSCRIPT, "noscript"); 2710 flagcount += addReplyCommandFlag(c,cmd,CMD_RANDOM, "random"); 2711 flagcount += addReplyCommandFlag(c,cmd,CMD_SORT_FOR_SCRIPT,"sort_for_script"); 2712 flagcount += addReplyCommandFlag(c,cmd,CMD_LOADING, "loading"); 2713 flagcount += addReplyCommandFlag(c,cmd,CMD_STALE, "stale"); 2714 flagcount += addReplyCommandFlag(c,cmd,CMD_SKIP_MONITOR, "skip_monitor"); 2715 flagcount += addReplyCommandFlag(c,cmd,CMD_ASKING, "asking"); 2716 flagcount += addReplyCommandFlag(c,cmd,CMD_FAST, "fast"); 2717 if (cmd->getkeys_proc) { 2718 addReplyStatus(c, "movablekeys"); 2719 flagcount += 1; 2720 } 2721 setDeferredMultiBulkLength(c, flaglen, flagcount); 2722 2723 addReplyLongLong(c, cmd->firstkey); 2724 addReplyLongLong(c, cmd->lastkey); 2725 addReplyLongLong(c, cmd->keystep); 2726 } 2727 } 2728 2729 /* COMMAND <subcommand> <args> */ 2730 void commandCommand(client *c) { 2731 dictIterator *di; 2732 dictEntry *de; 2733 2734 if (c->argc == 1) { 2735 addReplyMultiBulkLen(c, dictSize(server.commands)); 2736 di = dictGetIterator(server.commands); 2737 while ((de = dictNext(di)) != NULL) { 2738 addReplyCommand(c, dictGetVal(de)); 2739 } 2740 dictReleaseIterator(di); 2741 } else if (!strcasecmp(c->argv[1]->ptr, "info")) { 2742 int i; 2743 addReplyMultiBulkLen(c, c->argc-2); 2744 for (i = 2; i < c->argc; i++) { 2745 addReplyCommand(c, dictFetchValue(server.commands, c->argv[i]->ptr)); 2746 } 2747 } else if (!strcasecmp(c->argv[1]->ptr, "count") && c->argc == 2) { 2748 addReplyLongLong(c, dictSize(server.commands)); 2749 } else if (!strcasecmp(c->argv[1]->ptr,"getkeys") && c->argc >= 3) { 2750 struct redisCommand *cmd = lookupCommand(c->argv[2]->ptr); 2751 int *keys, numkeys, j; 2752 2753 if (!cmd) { 2754 addReplyErrorFormat(c,"Invalid command specified"); 2755 return; 2756 } else if ((cmd->arity > 0 && cmd->arity != c->argc-2) || 2757 ((c->argc-2) < -cmd->arity)) 2758 { 2759 addReplyError(c,"Invalid number of arguments specified for command"); 2760 return; 2761 } 2762 2763 keys = getKeysFromCommand(cmd,c->argv+2,c->argc-2,&numkeys); 2764 addReplyMultiBulkLen(c,numkeys); 2765 for (j = 0; j < numkeys; j++) addReplyBulk(c,c->argv[keys[j]+2]); 2766 getKeysFreeResult(keys); 2767 } else { 2768 addReplyError(c, "Unknown subcommand or wrong number of arguments."); 2769 return; 2770 } 2771 } 2772 2773 /* Convert an amount of bytes into a human readable string in the form 2774 * of 100B, 2G, 100M, 4K, and so forth. */ 2775 void bytesToHuman(char *s, unsigned long long n) { 2776 double d; 2777 2778 if (n < 1024) { 2779 /* Bytes */ 2780 sprintf(s,"%lluB",n); 2781 return; 2782 } else if (n < (1024*1024)) { 2783 d = (double)n/(1024); 2784 sprintf(s,"%.2fK",d); 2785 } else if (n < (1024LL*1024*1024)) { 2786 d = (double)n/(1024*1024); 2787 sprintf(s,"%.2fM",d); 2788 } else if (n < (1024LL*1024*1024*1024)) { 2789 d = (double)n/(1024LL*1024*1024); 2790 sprintf(s,"%.2fG",d); 2791 } else if (n < (1024LL*1024*1024*1024*1024)) { 2792 d = (double)n/(1024LL*1024*1024*1024); 2793 sprintf(s,"%.2fT",d); 2794 } else if (n < (1024LL*1024*1024*1024*1024*1024)) { 2795 d = (double)n/(1024LL*1024*1024*1024*1024); 2796 sprintf(s,"%.2fP",d); 2797 } else { 2798 /* Let's hope we never need this */ 2799 sprintf(s,"%lluB",n); 2800 } 2801 } 2802 2803 /* Create the string returned by the INFO command. This is decoupled 2804 * by the INFO command itself as we need to report the same information 2805 * on memory corruption problems. */ 2806 sds genRedisInfoString(char *section) { 2807 sds info = sdsempty(); 2808 time_t uptime = server.unixtime-server.stat_starttime; 2809 int j, numcommands; 2810 struct rusage self_ru, c_ru; 2811 unsigned long lol, bib; 2812 int allsections = 0, defsections = 0; 2813 int sections = 0; 2814 2815 if (section == NULL) section = "default"; 2816 allsections = strcasecmp(section,"all") == 0; 2817 defsections = strcasecmp(section,"default") == 0; 2818 2819 getrusage(RUSAGE_SELF, &self_ru); 2820 getrusage(RUSAGE_CHILDREN, &c_ru); 2821 getClientsMaxBuffers(&lol,&bib); 2822 2823 /* Server */ 2824 if (allsections || defsections || !strcasecmp(section,"server")) { 2825 static int call_uname = 1; 2826 static struct utsname name; 2827 char *mode; 2828 2829 if (server.cluster_enabled) mode = "cluster"; 2830 else if (server.sentinel_mode) mode = "sentinel"; 2831 else mode = "standalone"; 2832 2833 if (sections++) info = sdscat(info,"\r\n"); 2834 2835 if (call_uname) { 2836 /* Uname can be slow and is always the same output. Cache it. */ 2837 uname(&name); 2838 call_uname = 0; 2839 } 2840 2841 info = sdscatprintf(info, 2842 "# Server\r\n" 2843 "redis_version:%s\r\n" 2844 "redis_git_sha1:%s\r\n" 2845 "redis_git_dirty:%d\r\n" 2846 "redis_build_id:%llx\r\n" 2847 "redis_mode:%s\r\n" 2848 "os:%s %s %s\r\n" 2849 "arch_bits:%d\r\n" 2850 "multiplexing_api:%s\r\n" 2851 "gcc_version:%d.%d.%d\r\n" 2852 "process_id:%ld\r\n" 2853 "run_id:%s\r\n" 2854 "tcp_port:%d\r\n" 2855 "uptime_in_seconds:%jd\r\n" 2856 "uptime_in_days:%jd\r\n" 2857 "hz:%d\r\n" 2858 "lru_clock:%ld\r\n" 2859 "executable:%s\r\n" 2860 "config_file:%s\r\n", 2861 REDIS_VERSION, 2862 redisGitSHA1(), 2863 strtol(redisGitDirty(),NULL,10) > 0, 2864 (unsigned long long) redisBuildId(), 2865 mode, 2866 name.sysname, name.release, name.machine, 2867 server.arch_bits, 2868 aeGetApiName(), 2869 #ifdef __GNUC__ 2870 __GNUC__,__GNUC_MINOR__,__GNUC_PATCHLEVEL__, 2871 #else 2872 0,0,0, 2873 #endif 2874 (long) getpid(), 2875 server.runid, 2876 server.port, 2877 (intmax_t)uptime, 2878 (intmax_t)(uptime/(3600*24)), 2879 server.hz, 2880 (unsigned long) server.lruclock, 2881 server.executable ? server.executable : "", 2882 server.configfile ? server.configfile : ""); 2883 } 2884 2885 /* Clients */ 2886 if (allsections || defsections || !strcasecmp(section,"clients")) { 2887 if (sections++) info = sdscat(info,"\r\n"); 2888 info = sdscatprintf(info, 2889 "# Clients\r\n" 2890 "connected_clients:%lu\r\n" 2891 "client_longest_output_list:%lu\r\n" 2892 "client_biggest_input_buf:%lu\r\n" 2893 "blocked_clients:%d\r\n", 2894 listLength(server.clients)-listLength(server.slaves), 2895 lol, bib, 2896 server.bpop_blocked_clients); 2897 } 2898 2899 /* Memory */ 2900 if (allsections || defsections || !strcasecmp(section,"memory")) { 2901 char hmem[64]; 2902 char peak_hmem[64]; 2903 char total_system_hmem[64]; 2904 char used_memory_lua_hmem[64]; 2905 char used_memory_rss_hmem[64]; 2906 char maxmemory_hmem[64]; 2907 size_t zmalloc_used = zmalloc_used_memory(); 2908 size_t total_system_mem = server.system_memory_size; 2909 const char *evict_policy = maxmemoryToString(); 2910 long long memory_lua = (long long)lua_gc(server.lua,LUA_GCCOUNT,0)*1024; 2911 2912 /* Peak memory is updated from time to time by serverCron() so it 2913 * may happen that the instantaneous value is slightly bigger than 2914 * the peak value. This may confuse users, so we update the peak 2915 * if found smaller than the current memory usage. */ 2916 if (zmalloc_used > server.stat_peak_memory) 2917 server.stat_peak_memory = zmalloc_used; 2918 2919 bytesToHuman(hmem,zmalloc_used); 2920 bytesToHuman(peak_hmem,server.stat_peak_memory); 2921 bytesToHuman(total_system_hmem,total_system_mem); 2922 bytesToHuman(used_memory_lua_hmem,memory_lua); 2923 bytesToHuman(used_memory_rss_hmem,server.resident_set_size); 2924 bytesToHuman(maxmemory_hmem,server.maxmemory); 2925 2926 if (sections++) info = sdscat(info,"\r\n"); 2927 info = sdscatprintf(info, 2928 "# Memory\r\n" 2929 "used_memory:%zu\r\n" 2930 "used_memory_human:%s\r\n" 2931 "used_memory_rss:%zu\r\n" 2932 "used_memory_rss_human:%s\r\n" 2933 "used_memory_peak:%zu\r\n" 2934 "used_memory_peak_human:%s\r\n" 2935 "total_system_memory:%lu\r\n" 2936 "total_system_memory_human:%s\r\n" 2937 "used_memory_lua:%lld\r\n" 2938 "used_memory_lua_human:%s\r\n" 2939 "maxmemory:%lld\r\n" 2940 "maxmemory_human:%s\r\n" 2941 "maxmemory_policy:%s\r\n" 2942 "mem_fragmentation_ratio:%.2f\r\n" 2943 "mem_allocator:%s\r\n", 2944 zmalloc_used, 2945 hmem, 2946 server.resident_set_size, 2947 used_memory_rss_hmem, 2948 server.stat_peak_memory, 2949 peak_hmem, 2950 (unsigned long)total_system_mem, 2951 total_system_hmem, 2952 memory_lua, 2953 used_memory_lua_hmem, 2954 server.maxmemory, 2955 maxmemory_hmem, 2956 evict_policy, 2957 zmalloc_get_fragmentation_ratio(server.resident_set_size), 2958 ZMALLOC_LIB 2959 ); 2960 } 2961 2962 /* Persistence */ 2963 if (allsections || defsections || !strcasecmp(section,"persistence")) { 2964 if (sections++) info = sdscat(info,"\r\n"); 2965 info = sdscatprintf(info, 2966 "# Persistence\r\n" 2967 "loading:%d\r\n" 2968 "rdb_changes_since_last_save:%lld\r\n" 2969 "rdb_bgsave_in_progress:%d\r\n" 2970 "rdb_last_save_time:%jd\r\n" 2971 "rdb_last_bgsave_status:%s\r\n" 2972 "rdb_last_bgsave_time_sec:%jd\r\n" 2973 "rdb_current_bgsave_time_sec:%jd\r\n" 2974 "aof_enabled:%d\r\n" 2975 "aof_rewrite_in_progress:%d\r\n" 2976 "aof_rewrite_scheduled:%d\r\n" 2977 "aof_last_rewrite_time_sec:%jd\r\n" 2978 "aof_current_rewrite_time_sec:%jd\r\n" 2979 "aof_last_bgrewrite_status:%s\r\n" 2980 "aof_last_write_status:%s\r\n", 2981 server.loading, 2982 server.dirty, 2983 server.rdb_child_pid != -1, 2984 (intmax_t)server.lastsave, 2985 (server.lastbgsave_status == C_OK) ? "ok" : "err", 2986 (intmax_t)server.rdb_save_time_last, 2987 (intmax_t)((server.rdb_child_pid == -1) ? 2988 -1 : time(NULL)-server.rdb_save_time_start), 2989 server.aof_state != AOF_OFF, 2990 server.aof_child_pid != -1, 2991 server.aof_rewrite_scheduled, 2992 (intmax_t)server.aof_rewrite_time_last, 2993 (intmax_t)((server.aof_child_pid == -1) ? 2994 -1 : time(NULL)-server.aof_rewrite_time_start), 2995 (server.aof_lastbgrewrite_status == C_OK) ? "ok" : "err", 2996 (server.aof_last_write_status == C_OK) ? "ok" : "err"); 2997 2998 if (server.aof_state != AOF_OFF) { 2999 info = sdscatprintf(info, 3000 "aof_current_size:%lld\r\n" 3001 "aof_base_size:%lld\r\n" 3002 "aof_pending_rewrite:%d\r\n" 3003 "aof_buffer_length:%zu\r\n" 3004 "aof_rewrite_buffer_length:%lu\r\n" 3005 "aof_pending_bio_fsync:%llu\r\n" 3006 "aof_delayed_fsync:%lu\r\n", 3007 (long long) server.aof_current_size, 3008 (long long) server.aof_rewrite_base_size, 3009 server.aof_rewrite_scheduled, 3010 sdslen(server.aof_buf), 3011 aofRewriteBufferSize(), 3012 bioPendingJobsOfType(BIO_AOF_FSYNC), 3013 server.aof_delayed_fsync); 3014 } 3015 3016 if (server.loading) { 3017 double perc; 3018 time_t eta, elapsed; 3019 off_t remaining_bytes = server.loading_total_bytes- 3020 server.loading_loaded_bytes; 3021 3022 perc = ((double)server.loading_loaded_bytes / 3023 (server.loading_total_bytes+1)) * 100; 3024 3025 elapsed = time(NULL)-server.loading_start_time; 3026 if (elapsed == 0) { 3027 eta = 1; /* A fake 1 second figure if we don't have 3028 enough info */ 3029 } else { 3030 eta = (elapsed*remaining_bytes)/(server.loading_loaded_bytes+1); 3031 } 3032 3033 info = sdscatprintf(info, 3034 "loading_start_time:%jd\r\n" 3035 "loading_total_bytes:%llu\r\n" 3036 "loading_loaded_bytes:%llu\r\n" 3037 "loading_loaded_perc:%.2f\r\n" 3038 "loading_eta_seconds:%jd\r\n", 3039 (intmax_t) server.loading_start_time, 3040 (unsigned long long) server.loading_total_bytes, 3041 (unsigned long long) server.loading_loaded_bytes, 3042 perc, 3043 (intmax_t)eta 3044 ); 3045 } 3046 } 3047 3048 /* Stats */ 3049 if (allsections || defsections || !strcasecmp(section,"stats")) { 3050 if (sections++) info = sdscat(info,"\r\n"); 3051 info = sdscatprintf(info, 3052 "# Stats\r\n" 3053 "total_connections_received:%lld\r\n" 3054 "total_commands_processed:%lld\r\n" 3055 "instantaneous_ops_per_sec:%lld\r\n" 3056 "total_net_input_bytes:%lld\r\n" 3057 "total_net_output_bytes:%lld\r\n" 3058 "instantaneous_input_kbps:%.2f\r\n" 3059 "instantaneous_output_kbps:%.2f\r\n" 3060 "rejected_connections:%lld\r\n" 3061 "sync_full:%lld\r\n" 3062 "sync_partial_ok:%lld\r\n" 3063 "sync_partial_err:%lld\r\n" 3064 "expired_keys:%lld\r\n" 3065 "evicted_keys:%lld\r\n" 3066 "keyspace_hits:%lld\r\n" 3067 "keyspace_misses:%lld\r\n" 3068 "pubsub_channels:%ld\r\n" 3069 "pubsub_patterns:%lu\r\n" 3070 "latest_fork_usec:%lld\r\n" 3071 "migrate_cached_sockets:%ld\r\n", 3072 server.stat_numconnections, 3073 server.stat_numcommands, 3074 getInstantaneousMetric(STATS_METRIC_COMMAND), 3075 server.stat_net_input_bytes, 3076 server.stat_net_output_bytes, 3077 (float)getInstantaneousMetric(STATS_METRIC_NET_INPUT)/1024, 3078 (float)getInstantaneousMetric(STATS_METRIC_NET_OUTPUT)/1024, 3079 server.stat_rejected_conn, 3080 server.stat_sync_full, 3081 server.stat_sync_partial_ok, 3082 server.stat_sync_partial_err, 3083 server.stat_expiredkeys, 3084 server.stat_evictedkeys, 3085 server.stat_keyspace_hits, 3086 server.stat_keyspace_misses, 3087 dictSize(server.pubsub_channels), 3088 listLength(server.pubsub_patterns), 3089 server.stat_fork_time, 3090 dictSize(server.migrate_cached_sockets)); 3091 } 3092 3093 /* Replication */ 3094 if (allsections || defsections || !strcasecmp(section,"replication")) { 3095 if (sections++) info = sdscat(info,"\r\n"); 3096 info = sdscatprintf(info, 3097 "# Replication\r\n" 3098 "role:%s\r\n", 3099 server.masterhost == NULL ? "master" : "slave"); 3100 if (server.masterhost) { 3101 long long slave_repl_offset = 1; 3102 3103 if (server.master) 3104 slave_repl_offset = server.master->reploff; 3105 else if (server.cached_master) 3106 slave_repl_offset = server.cached_master->reploff; 3107 3108 info = sdscatprintf(info, 3109 "master_host:%s\r\n" 3110 "master_port:%d\r\n" 3111 "master_link_status:%s\r\n" 3112 "master_last_io_seconds_ago:%d\r\n" 3113 "master_sync_in_progress:%d\r\n" 3114 "slave_repl_offset:%lld\r\n" 3115 ,server.masterhost, 3116 server.masterport, 3117 (server.repl_state == REPL_STATE_CONNECTED) ? 3118 "up" : "down", 3119 server.master ? 3120 ((int)(server.unixtime-server.master->lastinteraction)) : -1, 3121 server.repl_state == REPL_STATE_TRANSFER, 3122 slave_repl_offset 3123 ); 3124 3125 if (server.repl_state == REPL_STATE_TRANSFER) { 3126 info = sdscatprintf(info, 3127 "master_sync_left_bytes:%lld\r\n" 3128 "master_sync_last_io_seconds_ago:%d\r\n" 3129 , (long long) 3130 (server.repl_transfer_size - server.repl_transfer_read), 3131 (int)(server.unixtime-server.repl_transfer_lastio) 3132 ); 3133 } 3134 3135 if (server.repl_state != REPL_STATE_CONNECTED) { 3136 info = sdscatprintf(info, 3137 "master_link_down_since_seconds:%jd\r\n", 3138 (intmax_t)server.unixtime-server.repl_down_since); 3139 } 3140 info = sdscatprintf(info, 3141 "slave_priority:%d\r\n" 3142 "slave_read_only:%d\r\n", 3143 server.slave_priority, 3144 server.repl_slave_ro); 3145 } 3146 3147 info = sdscatprintf(info, 3148 "connected_slaves:%lu\r\n", 3149 listLength(server.slaves)); 3150 3151 /* If min-slaves-to-write is active, write the number of slaves 3152 * currently considered 'good'. */ 3153 if (server.repl_min_slaves_to_write && 3154 server.repl_min_slaves_max_lag) { 3155 info = sdscatprintf(info, 3156 "min_slaves_good_slaves:%d\r\n", 3157 server.repl_good_slaves_count); 3158 } 3159 3160 if (listLength(server.slaves)) { 3161 int slaveid = 0; 3162 listNode *ln; 3163 listIter li; 3164 3165 listRewind(server.slaves,&li); 3166 while((ln = listNext(&li))) { 3167 client *slave = listNodeValue(ln); 3168 char *state = NULL; 3169 char ip[NET_IP_STR_LEN]; 3170 int port; 3171 long lag = 0; 3172 3173 if (anetPeerToString(slave->fd,ip,sizeof(ip),&port) == -1) continue; 3174 switch(slave->replstate) { 3175 case SLAVE_STATE_WAIT_BGSAVE_START: 3176 case SLAVE_STATE_WAIT_BGSAVE_END: 3177 state = "wait_bgsave"; 3178 break; 3179 case SLAVE_STATE_SEND_BULK: 3180 state = "send_bulk"; 3181 break; 3182 case SLAVE_STATE_ONLINE: 3183 state = "online"; 3184 break; 3185 } 3186 if (state == NULL) continue; 3187 if (slave->replstate == SLAVE_STATE_ONLINE) 3188 lag = time(NULL) - slave->repl_ack_time; 3189 3190 info = sdscatprintf(info, 3191 "slave%d:ip=%s,port=%d,state=%s," 3192 "offset=%lld,lag=%ld\r\n", 3193 slaveid,ip,slave->slave_listening_port,state, 3194 slave->repl_ack_off, lag); 3195 slaveid++; 3196 } 3197 } 3198 info = sdscatprintf(info, 3199 "master_repl_offset:%lld\r\n" 3200 "repl_backlog_active:%d\r\n" 3201 "repl_backlog_size:%lld\r\n" 3202 "repl_backlog_first_byte_offset:%lld\r\n" 3203 "repl_backlog_histlen:%lld\r\n", 3204 server.master_repl_offset, 3205 server.repl_backlog != NULL, 3206 server.repl_backlog_size, 3207 server.repl_backlog_off, 3208 server.repl_backlog_histlen); 3209 } 3210 3211 /* CPU */ 3212 if (allsections || defsections || !strcasecmp(section,"cpu")) { 3213 if (sections++) info = sdscat(info,"\r\n"); 3214 info = sdscatprintf(info, 3215 "# CPU\r\n" 3216 "used_cpu_sys:%.2f\r\n" 3217 "used_cpu_user:%.2f\r\n" 3218 "used_cpu_sys_children:%.2f\r\n" 3219 "used_cpu_user_children:%.2f\r\n", 3220 (float)self_ru.ru_stime.tv_sec+(float)self_ru.ru_stime.tv_usec/1000000, 3221 (float)self_ru.ru_utime.tv_sec+(float)self_ru.ru_utime.tv_usec/1000000, 3222 (float)c_ru.ru_stime.tv_sec+(float)c_ru.ru_stime.tv_usec/1000000, 3223 (float)c_ru.ru_utime.tv_sec+(float)c_ru.ru_utime.tv_usec/1000000); 3224 } 3225 3226 /* cmdtime */ 3227 if (allsections || !strcasecmp(section,"commandstats")) { 3228 if (sections++) info = sdscat(info,"\r\n"); 3229 info = sdscatprintf(info, "# Commandstats\r\n"); 3230 numcommands = sizeof(redisCommandTable)/sizeof(struct redisCommand); 3231 for (j = 0; j < numcommands; j++) { 3232 struct redisCommand *c = redisCommandTable+j; 3233 3234 if (!c->calls) continue; 3235 info = sdscatprintf(info, 3236 "cmdstat_%s:calls=%lld,usec=%lld,usec_per_call=%.2f\r\n", 3237 c->name, c->calls, c->microseconds, 3238 (c->calls == 0) ? 0 : ((float)c->microseconds/c->calls)); 3239 } 3240 } 3241 3242 /* Cluster */ 3243 if (allsections || defsections || !strcasecmp(section,"cluster")) { 3244 if (sections++) info = sdscat(info,"\r\n"); 3245 info = sdscatprintf(info, 3246 "# Cluster\r\n" 3247 "cluster_enabled:%d\r\n", 3248 server.cluster_enabled); 3249 } 3250 3251 /* Key space */ 3252 if (allsections || defsections || !strcasecmp(section,"keyspace")) { 3253 if (sections++) info = sdscat(info,"\r\n"); 3254 info = sdscatprintf(info, "# Keyspace\r\n"); 3255 for (j = 0; j < server.dbnum; j++) { 3256 long long keys, vkeys; 3257 3258 keys = dictSize(server.db[j].dict); 3259 vkeys = dictSize(server.db[j].expires); 3260 if (keys || vkeys) { 3261 info = sdscatprintf(info, 3262 "db%d:keys=%lld,expires=%lld,avg_ttl=%lld\r\n", 3263 j, keys, vkeys, server.db[j].avg_ttl); 3264 } 3265 } 3266 } 3267 return info; 3268 } 3269 3270 void infoCommand(client *c) { 3271 char *section = c->argc == 2 ? c->argv[1]->ptr : "default"; 3272 3273 if (c->argc > 2) { 3274 addReply(c,shared.syntaxerr); 3275 return; 3276 } 3277 addReplyBulkSds(c, genRedisInfoString(section)); 3278 } 3279 3280 void monitorCommand(client *c) { 3281 /* ignore MONITOR if already slave or in monitor mode */ 3282 if (c->flags & CLIENT_SLAVE) return; 3283 3284 c->flags |= (CLIENT_SLAVE|CLIENT_MONITOR); 3285 listAddNodeTail(server.monitors,c); 3286 addReply(c,shared.ok); 3287 } 3288 3289 /* ============================ Maxmemory directive ======================== */ 3290 3291 /* freeMemoryIfNeeded() gets called when 'maxmemory' is set on the config 3292 * file to limit the max memory used by the server, before processing a 3293 * command. 3294 * 3295 * The goal of the function is to free enough memory to keep Redis under the 3296 * configured memory limit. 3297 * 3298 * The function starts calculating how many bytes should be freed to keep 3299 * Redis under the limit, and enters a loop selecting the best keys to 3300 * evict accordingly to the configured policy. 3301 * 3302 * If all the bytes needed to return back under the limit were freed the 3303 * function returns C_OK, otherwise C_ERR is returned, and the caller 3304 * should block the execution of commands that will result in more memory 3305 * used by the server. 3306 * 3307 * ------------------------------------------------------------------------ 3308 * 3309 * LRU approximation algorithm 3310 * 3311 * Redis uses an approximation of the LRU algorithm that runs in constant 3312 * memory. Every time there is a key to expire, we sample N keys (with 3313 * N very small, usually in around 5) to populate a pool of best keys to 3314 * evict of M keys (the pool size is defined by MAXMEMORY_EVICTION_POOL_SIZE). 3315 * 3316 * The N keys sampled are added in the pool of good keys to expire (the one 3317 * with an old access time) if they are better than one of the current keys 3318 * in the pool. 3319 * 3320 * After the pool is populated, the best key we have in the pool is expired. 3321 * However note that we don't remove keys from the pool when they are deleted 3322 * so the pool may contain keys that no longer exist. 3323 * 3324 * When we try to evict a key, and all the entries in the pool don't exist 3325 * we populate it again. This time we'll be sure that the pool has at least 3326 * one key that can be evicted, if there is at least one key that can be 3327 * evicted in the whole database. */ 3328 3329 /* Create a new eviction pool. */ 3330 struct evictionPoolEntry *evictionPoolAlloc(void) { 3331 struct evictionPoolEntry *ep; 3332 int j; 3333 3334 ep = zmalloc(sizeof(*ep)*MAXMEMORY_EVICTION_POOL_SIZE); 3335 for (j = 0; j < MAXMEMORY_EVICTION_POOL_SIZE; j++) { 3336 ep[j].idle = 0; 3337 ep[j].key = NULL; 3338 } 3339 return ep; 3340 } 3341 3342 /* This is an helper function for freeMemoryIfNeeded(), it is used in order 3343 * to populate the evictionPool with a few entries every time we want to 3344 * expire a key. Keys with idle time smaller than one of the current 3345 * keys are added. Keys are always added if there are free entries. 3346 * 3347 * We insert keys on place in ascending order, so keys with the smaller 3348 * idle time are on the left, and keys with the higher idle time on the 3349 * right. */ 3350 3351 #define EVICTION_SAMPLES_ARRAY_SIZE 16 3352 void evictionPoolPopulate(dict *sampledict, dict *keydict, struct evictionPoolEntry *pool) { 3353 int j, k, count; 3354 dictEntry *_samples[EVICTION_SAMPLES_ARRAY_SIZE]; 3355 dictEntry **samples; 3356 3357 /* Try to use a static buffer: this function is a big hit... 3358 * Note: it was actually measured that this helps. */ 3359 if (server.maxmemory_samples <= EVICTION_SAMPLES_ARRAY_SIZE) { 3360 samples = _samples; 3361 } else { 3362 samples = zmalloc(sizeof(samples[0])*server.maxmemory_samples); 3363 } 3364 3365 count = dictGetSomeKeys(sampledict,samples,server.maxmemory_samples); 3366 for (j = 0; j < count; j++) { 3367 unsigned long long idle; 3368 sds key; 3369 robj *o; 3370 dictEntry *de; 3371 3372 de = samples[j]; 3373 key = dictGetKey(de); 3374 /* If the dictionary we are sampling from is not the main 3375 * dictionary (but the expires one) we need to lookup the key 3376 * again in the key dictionary to obtain the value object. */ 3377 if (sampledict != keydict) de = dictFind(keydict, key); 3378 o = dictGetVal(de); 3379 idle = estimateObjectIdleTime(o); 3380 3381 /* Insert the element inside the pool. 3382 * First, find the first empty bucket or the first populated 3383 * bucket that has an idle time smaller than our idle time. */ 3384 k = 0; 3385 while (k < MAXMEMORY_EVICTION_POOL_SIZE && 3386 pool[k].key && 3387 pool[k].idle < idle) k++; 3388 if (k == 0 && pool[MAXMEMORY_EVICTION_POOL_SIZE-1].key != NULL) { 3389 /* Can't insert if the element is < the worst element we have 3390 * and there are no empty buckets. */ 3391 continue; 3392 } else if (k < MAXMEMORY_EVICTION_POOL_SIZE && pool[k].key == NULL) { 3393 /* Inserting into empty position. No setup needed before insert. */ 3394 } else { 3395 /* Inserting in the middle. Now k points to the first element 3396 * greater than the element to insert. */ 3397 if (pool[MAXMEMORY_EVICTION_POOL_SIZE-1].key == NULL) { 3398 /* Free space on the right? Insert at k shifting 3399 * all the elements from k to end to the right. */ 3400 memmove(pool+k+1,pool+k, 3401 sizeof(pool[0])*(MAXMEMORY_EVICTION_POOL_SIZE-k-1)); 3402 } else { 3403 /* No free space on right? Insert at k-1 */ 3404 k--; 3405 /* Shift all elements on the left of k (included) to the 3406 * left, so we discard the element with smaller idle time. */ 3407 sdsfree(pool[0].key); 3408 memmove(pool,pool+1,sizeof(pool[0])*k); 3409 } 3410 } 3411 pool[k].key = sdsdup(key); 3412 pool[k].idle = idle; 3413 } 3414 if (samples != _samples) zfree(samples); 3415 } 3416 3417 int freeMemoryIfNeeded(void) { 3418 size_t mem_used, mem_tofree, mem_freed; 3419 int slaves = listLength(server.slaves); 3420 mstime_t latency, eviction_latency; 3421 3422 /* Remove the size of slaves output buffers and AOF buffer from the 3423 * count of used memory. */ 3424 mem_used = zmalloc_used_memory(); 3425 if (slaves) { 3426 listIter li; 3427 listNode *ln; 3428 3429 listRewind(server.slaves,&li); 3430 while((ln = listNext(&li))) { 3431 client *slave = listNodeValue(ln); 3432 unsigned long obuf_bytes = getClientOutputBufferMemoryUsage(slave); 3433 if (obuf_bytes > mem_used) 3434 mem_used = 0; 3435 else 3436 mem_used -= obuf_bytes; 3437 } 3438 } 3439 if (server.aof_state != AOF_OFF) { 3440 mem_used -= sdslen(server.aof_buf); 3441 mem_used -= aofRewriteBufferSize(); 3442 } 3443 3444 /* Check if we are over the memory limit. */ 3445 if (mem_used <= server.maxmemory) return C_OK; 3446 3447 if (server.maxmemory_policy == MAXMEMORY_NO_EVICTION) 3448 return C_ERR; /* We need to free memory, but policy forbids. */ 3449 3450 /* Compute how much memory we need to free. */ 3451 mem_tofree = mem_used - server.maxmemory; 3452 mem_freed = 0; 3453 latencyStartMonitor(latency); 3454 while (mem_freed < mem_tofree) { 3455 int j, k, keys_freed = 0; 3456 3457 for (j = 0; j < server.dbnum; j++) { 3458 long bestval = 0; /* just to prevent warning */ 3459 sds bestkey = NULL; 3460 dictEntry *de; 3461 redisDb *db = server.db+j; 3462 dict *dict; 3463 3464 if (server.maxmemory_policy == MAXMEMORY_ALLKEYS_LRU || 3465 server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM) 3466 { 3467 dict = server.db[j].dict; 3468 } else { 3469 dict = server.db[j].expires; 3470 } 3471 if (dictSize(dict) == 0) continue; 3472 3473 /* volatile-random and allkeys-random policy */ 3474 if (server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM || 3475 server.maxmemory_policy == MAXMEMORY_VOLATILE_RANDOM) 3476 { 3477 de = dictGetRandomKey(dict); 3478 bestkey = dictGetKey(de); 3479 } 3480 3481 /* volatile-lru and allkeys-lru policy */ 3482 else if (server.maxmemory_policy == MAXMEMORY_ALLKEYS_LRU || 3483 server.maxmemory_policy == MAXMEMORY_VOLATILE_LRU) 3484 { 3485 struct evictionPoolEntry *pool = db->eviction_pool; 3486 3487 while(bestkey == NULL) { 3488 evictionPoolPopulate(dict, db->dict, db->eviction_pool); 3489 /* Go backward from best to worst element to evict. */ 3490 for (k = MAXMEMORY_EVICTION_POOL_SIZE-1; k >= 0; k--) { 3491 if (pool[k].key == NULL) continue; 3492 de = dictFind(dict,pool[k].key); 3493 3494 /* Remove the entry from the pool. */ 3495 sdsfree(pool[k].key); 3496 /* Shift all elements on its right to left. */ 3497 memmove(pool+k,pool+k+1, 3498 sizeof(pool[0])*(MAXMEMORY_EVICTION_POOL_SIZE-k-1)); 3499 /* Clear the element on the right which is empty 3500 * since we shifted one position to the left. */ 3501 pool[MAXMEMORY_EVICTION_POOL_SIZE-1].key = NULL; 3502 pool[MAXMEMORY_EVICTION_POOL_SIZE-1].idle = 0; 3503 3504 /* If the key exists, is our pick. Otherwise it is 3505 * a ghost and we need to try the next element. */ 3506 if (de) { 3507 bestkey = dictGetKey(de); 3508 break; 3509 } else { 3510 /* Ghost... */ 3511 continue; 3512 } 3513 } 3514 } 3515 } 3516 3517 /* volatile-ttl */ 3518 else if (server.maxmemory_policy == MAXMEMORY_VOLATILE_TTL) { 3519 for (k = 0; k < server.maxmemory_samples; k++) { 3520 sds thiskey; 3521 long thisval; 3522 3523 de = dictGetRandomKey(dict); 3524 thiskey = dictGetKey(de); 3525 thisval = (long) dictGetVal(de); 3526 3527 /* Expire sooner (minor expire unix timestamp) is better 3528 * candidate for deletion */ 3529 if (bestkey == NULL || thisval < bestval) { 3530 bestkey = thiskey; 3531 bestval = thisval; 3532 } 3533 } 3534 } 3535 3536 /* Finally remove the selected key. */ 3537 if (bestkey) { 3538 long long delta; 3539 3540 robj *keyobj = createStringObject(bestkey,sdslen(bestkey)); 3541 propagateExpire(db,keyobj); 3542 /* We compute the amount of memory freed by dbDelete() alone. 3543 * It is possible that actually the memory needed to propagate 3544 * the DEL in AOF and replication link is greater than the one 3545 * we are freeing removing the key, but we can't account for 3546 * that otherwise we would never exit the loop. 3547 * 3548 * AOF and Output buffer memory will be freed eventually so 3549 * we only care about memory used by the key space. */ 3550 delta = (long long) zmalloc_used_memory(); 3551 latencyStartMonitor(eviction_latency); 3552 dbDelete(db,keyobj); 3553 latencyEndMonitor(eviction_latency); 3554 latencyAddSampleIfNeeded("eviction-del",eviction_latency); 3555 latencyRemoveNestedEvent(latency,eviction_latency); 3556 delta -= (long long) zmalloc_used_memory(); 3557 mem_freed += delta; 3558 server.stat_evictedkeys++; 3559 notifyKeyspaceEvent(NOTIFY_EVICTED, "evicted", 3560 keyobj, db->id); 3561 decrRefCount(keyobj); 3562 keys_freed++; 3563 3564 /* When the memory to free starts to be big enough, we may 3565 * start spending so much time here that is impossible to 3566 * deliver data to the slaves fast enough, so we force the 3567 * transmission here inside the loop. */ 3568 if (slaves) flushSlavesOutputBuffers(); 3569 } 3570 } 3571 if (!keys_freed) { 3572 latencyEndMonitor(latency); 3573 latencyAddSampleIfNeeded("eviction-cycle",latency); 3574 return C_ERR; /* nothing to free... */ 3575 } 3576 } 3577 latencyEndMonitor(latency); 3578 latencyAddSampleIfNeeded("eviction-cycle",latency); 3579 return C_OK; 3580 } 3581 3582 /* =================================== Main! ================================ */ 3583 3584 #ifdef __linux__ 3585 int linuxOvercommitMemoryValue(void) { 3586 FILE *fp = fopen("/proc/sys/vm/overcommit_memory","r"); 3587 char buf[64]; 3588 3589 if (!fp) return -1; 3590 if (fgets(buf,64,fp) == NULL) { 3591 fclose(fp); 3592 return -1; 3593 } 3594 fclose(fp); 3595 3596 return atoi(buf); 3597 } 3598 3599 void linuxMemoryWarnings(void) { 3600 if (linuxOvercommitMemoryValue() == 0) { 3601 serverLog(LL_WARNING,"WARNING overcommit_memory is set to 0! Background save may fail under low memory condition. To fix this issue add 'vm.overcommit_memory = 1' to /etc/sysctl.conf and then reboot or run the command 'sysctl vm.overcommit_memory=1' for this to take effect."); 3602 } 3603 if (THPIsEnabled()) { 3604 serverLog(LL_WARNING,"WARNING you have Transparent Huge Pages (THP) support enabled in your kernel. This will create latency and memory usage issues with Redis. To fix this issue run the command 'echo never > /sys/kernel/mm/transparent_hugepage/enabled' as root, and add it to your /etc/rc.local in order to retain the setting after a reboot. Redis must be restarted after THP is disabled."); 3605 } 3606 } 3607 #endif /* __linux__ */ 3608 3609 void createPidFile(void) { 3610 /* If pidfile requested, but no pidfile defined, use 3611 * default pidfile path */ 3612 if (!server.pidfile) server.pidfile = zstrdup(CONFIG_DEFAULT_PID_FILE); 3613 3614 /* Try to write the pid file in a best-effort way. */ 3615 FILE *fp = fopen(server.pidfile,"w"); 3616 if (fp) { 3617 fprintf(fp,"%d\n",(int)getpid()); 3618 fclose(fp); 3619 } 3620 } 3621 3622 void daemonize(void) { 3623 int fd; 3624 3625 if (fork() != 0) exit(0); /* parent exits */ 3626 setsid(); /* create a new session */ 3627 3628 /* Every output goes to /dev/null. If Redis is daemonized but 3629 * the 'logfile' is set to 'stdout' in the configuration file 3630 * it will not log at all. */ 3631 if ((fd = open("/dev/null", O_RDWR, 0)) != -1) { 3632 dup2(fd, STDIN_FILENO); 3633 dup2(fd, STDOUT_FILENO); 3634 dup2(fd, STDERR_FILENO); 3635 if (fd > STDERR_FILENO) close(fd); 3636 } 3637 } 3638 3639 void version(void) { 3640 printf("Redis server v=%s sha=%s:%d malloc=%s bits=%d build=%llx\n", 3641 REDIS_VERSION, 3642 redisGitSHA1(), 3643 atoi(redisGitDirty()) > 0, 3644 ZMALLOC_LIB, 3645 sizeof(long) == 4 ? 32 : 64, 3646 (unsigned long long) redisBuildId()); 3647 exit(0); 3648 } 3649 3650 void usage(void) { 3651 fprintf(stderr,"Usage: ./redis-server [/path/to/redis.conf] [options]\n"); 3652 fprintf(stderr," ./redis-server - (read config from stdin)\n"); 3653 fprintf(stderr," ./redis-server -v or --version\n"); 3654 fprintf(stderr," ./redis-server -h or --help\n"); 3655 fprintf(stderr," ./redis-server --test-memory <megabytes>\n\n"); 3656 fprintf(stderr,"Examples:\n"); 3657 fprintf(stderr," ./redis-server (run the server with default conf)\n"); 3658 fprintf(stderr," ./redis-server /etc/redis/6379.conf\n"); 3659 fprintf(stderr," ./redis-server --port 7777\n"); 3660 fprintf(stderr," ./redis-server --port 7777 --slaveof 127.0.0.1 8888\n"); 3661 fprintf(stderr," ./redis-server /etc/myredis.conf --loglevel verbose\n\n"); 3662 fprintf(stderr,"Sentinel mode:\n"); 3663 fprintf(stderr," ./redis-server /etc/sentinel.conf --sentinel\n"); 3664 exit(1); 3665 } 3666 3667 void redisAsciiArt(void) { 3668 #include "asciilogo.h" 3669 char *buf = zmalloc(1024*16); 3670 char *mode; 3671 3672 if (server.cluster_enabled) mode = "cluster"; 3673 else if (server.sentinel_mode) mode = "sentinel"; 3674 else mode = "standalone"; 3675 3676 if (server.syslog_enabled) { 3677 serverLog(LL_NOTICE, 3678 "Redis %s (%s/%d) %s bit, %s mode, port %d, pid %ld ready to start.", 3679 REDIS_VERSION, 3680 redisGitSHA1(), 3681 strtol(redisGitDirty(),NULL,10) > 0, 3682 (sizeof(long) == 8) ? "64" : "32", 3683 mode, server.port, 3684 (long) getpid() 3685 ); 3686 } else { 3687 snprintf(buf,1024*16,ascii_logo, 3688 REDIS_VERSION, 3689 redisGitSHA1(), 3690 strtol(redisGitDirty(),NULL,10) > 0, 3691 (sizeof(long) == 8) ? "64" : "32", 3692 mode, server.port, 3693 (long) getpid() 3694 ); 3695 serverLogRaw(LL_NOTICE|LL_RAW,buf); 3696 } 3697 zfree(buf); 3698 } 3699 3700 static void sigShutdownHandler(int sig) { 3701 char *msg; 3702 3703 switch (sig) { 3704 case SIGINT: 3705 msg = "Received SIGINT scheduling shutdown..."; 3706 break; 3707 case SIGTERM: 3708 msg = "Received SIGTERM scheduling shutdown..."; 3709 break; 3710 default: 3711 msg = "Received shutdown signal, scheduling shutdown..."; 3712 }; 3713 3714 /* SIGINT is often delivered via Ctrl+C in an interactive session. 3715 * If we receive the signal the second time, we interpret this as 3716 * the user really wanting to quit ASAP without waiting to persist 3717 * on disk. */ 3718 if (server.shutdown_asap && sig == SIGINT) { 3719 serverLogFromHandler(LL_WARNING, "You insist... exiting now."); 3720 rdbRemoveTempFile(getpid()); 3721 exit(1); /* Exit with an error since this was not a clean shutdown. */ 3722 } else if (server.loading) { 3723 exit(0); 3724 } 3725 3726 serverLogFromHandler(LL_WARNING, msg); 3727 server.shutdown_asap = 1; 3728 } 3729 3730 void setupSignalHandlers(void) { 3731 struct sigaction act; 3732 3733 /* When the SA_SIGINFO flag is set in sa_flags then sa_sigaction is used. 3734 * Otherwise, sa_handler is used. */ 3735 sigemptyset(&act.sa_mask); 3736 act.sa_flags = 0; 3737 act.sa_handler = sigShutdownHandler; 3738 sigaction(SIGTERM, &act, NULL); 3739 sigaction(SIGINT, &act, NULL); 3740 3741 #ifdef HAVE_BACKTRACE 3742 sigemptyset(&act.sa_mask); 3743 act.sa_flags = SA_NODEFER | SA_RESETHAND | SA_SIGINFO; 3744 act.sa_sigaction = sigsegvHandler; 3745 sigaction(SIGSEGV, &act, NULL); 3746 sigaction(SIGBUS, &act, NULL); 3747 sigaction(SIGFPE, &act, NULL); 3748 sigaction(SIGILL, &act, NULL); 3749 #endif 3750 return; 3751 } 3752 3753 void memtest(size_t megabytes, int passes); 3754 3755 /* Returns 1 if there is --sentinel among the arguments or if 3756 * argv[0] is exactly "redis-sentinel". */ 3757 int checkForSentinelMode(int argc, char **argv) { 3758 int j; 3759 3760 if (strstr(argv[0],"redis-sentinel") != NULL) return 1; 3761 for (j = 1; j < argc; j++) 3762 if (!strcmp(argv[j],"--sentinel")) return 1; 3763 return 0; 3764 } 3765 3766 /* Function called at startup to load RDB or AOF file in memory. */ 3767 void loadDataFromDisk(void) { 3768 long long start = ustime(); 3769 if (server.aof_state == AOF_ON) { 3770 if (loadAppendOnlyFile(server.aof_filename) == C_OK) 3771 serverLog(LL_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000); 3772 } else { 3773 if (rdbLoad(server.rdb_filename) == C_OK) { 3774 serverLog(LL_NOTICE,"DB loaded from disk: %.3f seconds", 3775 (float)(ustime()-start)/1000000); 3776 } else if (errno != ENOENT) { 3777 serverLog(LL_WARNING,"Fatal error loading the DB: %s. Exiting.",strerror(errno)); 3778 exit(1); 3779 } 3780 } 3781 } 3782 3783 void redisOutOfMemoryHandler(size_t allocation_size) { 3784 serverLog(LL_WARNING,"Out Of Memory allocating %zu bytes!", 3785 allocation_size); 3786 serverPanic("Redis aborting for OUT OF MEMORY"); 3787 } 3788 3789 void redisSetProcTitle(char *title) { 3790 #ifdef USE_SETPROCTITLE 3791 char *server_mode = ""; 3792 if (server.cluster_enabled) server_mode = " [cluster]"; 3793 else if (server.sentinel_mode) server_mode = " [sentinel]"; 3794 3795 setproctitle("%s %s:%d%s", 3796 title, 3797 server.bindaddr_count ? server.bindaddr[0] : "*", 3798 server.port, 3799 server_mode); 3800 #else 3801 UNUSED(title); 3802 #endif 3803 } 3804 3805 /* 3806 * Check whether systemd or upstart have been used to start redis. 3807 */ 3808 3809 int redisSupervisedUpstart(void) { 3810 const char *upstart_job = getenv("UPSTART_JOB"); 3811 3812 if (!upstart_job) { 3813 serverLog(LL_WARNING, 3814 "upstart supervision requested, but UPSTART_JOB not found"); 3815 return 0; 3816 } 3817 3818 serverLog(LL_NOTICE, "supervised by upstart, will stop to signal readyness"); 3819 raise(SIGSTOP); 3820 unsetenv("UPSTART_JOB"); 3821 return 1; 3822 } 3823 3824 int redisSupervisedSystemd(void) { 3825 const char *notify_socket = getenv("NOTIFY_SOCKET"); 3826 int fd = 1; 3827 struct sockaddr_un su; 3828 struct iovec iov; 3829 struct msghdr hdr; 3830 int sendto_flags = 0; 3831 3832 if (!notify_socket) { 3833 serverLog(LL_WARNING, 3834 "systemd supervision requested, but NOTIFY_SOCKET not found"); 3835 return 0; 3836 } 3837 3838 if ((strchr("@/", notify_socket[0])) == NULL || strlen(notify_socket) < 2) { 3839 return 0; 3840 } 3841 3842 serverLog(LL_NOTICE, "supervised by systemd, will signal readyness"); 3843 if ((fd = socket(AF_UNIX, SOCK_DGRAM, 0)) == -1) { 3844 serverLog(LL_WARNING, 3845 "Can't connect to systemd socket %s", notify_socket); 3846 return 0; 3847 } 3848 3849 memset(&su, 0, sizeof(su)); 3850 su.sun_family = AF_UNIX; 3851 strncpy (su.sun_path, notify_socket, sizeof(su.sun_path) -1); 3852 su.sun_path[sizeof(su.sun_path) - 1] = '\0'; 3853 3854 if (notify_socket[0] == '@') 3855 su.sun_path[0] = '\0'; 3856 3857 memset(&iov, 0, sizeof(iov)); 3858 iov.iov_base = "READY=1"; 3859 iov.iov_len = strlen("READY=1"); 3860 3861 memset(&hdr, 0, sizeof(hdr)); 3862 hdr.msg_name = &su; 3863 hdr.msg_namelen = offsetof(struct sockaddr_un, sun_path) + 3864 strlen(notify_socket); 3865 hdr.msg_iov = &iov; 3866 hdr.msg_iovlen = 1; 3867 3868 unsetenv("NOTIFY_SOCKET"); 3869 #ifdef HAVE_MSG_NOSIGNAL 3870 sendto_flags |= MSG_NOSIGNAL; 3871 #endif 3872 if (sendmsg(fd, &hdr, sendto_flags) < 0) { 3873 serverLog(LL_WARNING, "Can't send notification to systemd"); 3874 close(fd); 3875 return 0; 3876 } 3877 close(fd); 3878 return 1; 3879 } 3880 3881 int redisIsSupervised(int mode) { 3882 if (mode == SUPERVISED_AUTODETECT) { 3883 const char *upstart_job = getenv("UPSTART_JOB"); 3884 const char *notify_socket = getenv("NOTIFY_SOCKET"); 3885 3886 if (upstart_job) { 3887 redisSupervisedUpstart(); 3888 } else if (notify_socket) { 3889 redisSupervisedSystemd(); 3890 } 3891 } else if (mode == SUPERVISED_UPSTART) { 3892 return redisSupervisedUpstart(); 3893 } else if (mode == SUPERVISED_SYSTEMD) { 3894 return redisSupervisedSystemd(); 3895 } 3896 3897 return 0; 3898 } 3899 3900 3901 int main(int argc, char **argv) { 3902 struct timeval tv; 3903 int j; 3904 3905 #ifdef REDIS_TEST 3906 if (argc == 3 && !strcasecmp(argv[1], "test")) { 3907 if (!strcasecmp(argv[2], "ziplist")) { 3908 return ziplistTest(argc, argv); 3909 } else if (!strcasecmp(argv[2], "quicklist")) { 3910 quicklistTest(argc, argv); 3911 } else if (!strcasecmp(argv[2], "intset")) { 3912 return intsetTest(argc, argv); 3913 } else if (!strcasecmp(argv[2], "zipmap")) { 3914 return zipmapTest(argc, argv); 3915 } else if (!strcasecmp(argv[2], "sha1test")) { 3916 return sha1Test(argc, argv); 3917 } else if (!strcasecmp(argv[2], "util")) { 3918 return utilTest(argc, argv); 3919 } else if (!strcasecmp(argv[2], "sds")) { 3920 return sdsTest(argc, argv); 3921 } else if (!strcasecmp(argv[2], "endianconv")) { 3922 return endianconvTest(argc, argv); 3923 } else if (!strcasecmp(argv[2], "crc64")) { 3924 return crc64Test(argc, argv); 3925 } 3926 3927 return -1; /* test not found */ 3928 } 3929 #endif 3930 3931 /* We need to initialize our libraries, and the server configuration. */ 3932 #ifdef INIT_SETPROCTITLE_REPLACEMENT 3933 spt_init(argc, argv); 3934 #endif 3935 setlocale(LC_COLLATE,""); 3936 zmalloc_enable_thread_safeness(); 3937 zmalloc_set_oom_handler(redisOutOfMemoryHandler); 3938 srand(time(NULL)^getpid()); 3939 gettimeofday(&tv,NULL); 3940 dictSetHashFunctionSeed(tv.tv_sec^tv.tv_usec^getpid()); 3941 server.sentinel_mode = checkForSentinelMode(argc,argv); 3942 initServerConfig(); 3943 3944 /* Store the executable path and arguments in a safe place in order 3945 * to be able to restart the server later. */ 3946 server.executable = getAbsolutePath(argv[0]); 3947 server.exec_argv = zmalloc(sizeof(char*)*(argc+1)); 3948 server.exec_argv[argc] = NULL; 3949 for (j = 0; j < argc; j++) server.exec_argv[j] = zstrdup(argv[j]); 3950 3951 /* We need to init sentinel right now as parsing the configuration file 3952 * in sentinel mode will have the effect of populating the sentinel 3953 * data structures with master nodes to monitor. */ 3954 if (server.sentinel_mode) { 3955 initSentinelConfig(); 3956 initSentinel(); 3957 } 3958 3959 /* Check if we need to start in redis-check-rdb mode. We just execute 3960 * the program main. However the program is part of the Redis executable 3961 * so that we can easily execute an RDB check on loading errors. */ 3962 if (strstr(argv[0],"redis-check-rdb") != NULL) 3963 exit(redis_check_rdb_main(argv,argc)); 3964 3965 if (argc >= 2) { 3966 j = 1; /* First option to parse in argv[] */ 3967 sds options = sdsempty(); 3968 char *configfile = NULL; 3969 3970 /* Handle special options --help and --version */ 3971 if (strcmp(argv[1], "-v") == 0 || 3972 strcmp(argv[1], "--version") == 0) version(); 3973 if (strcmp(argv[1], "--help") == 0 || 3974 strcmp(argv[1], "-h") == 0) usage(); 3975 if (strcmp(argv[1], "--test-memory") == 0) { 3976 if (argc == 3) { 3977 memtest(atoi(argv[2]),50); 3978 exit(0); 3979 } else { 3980 fprintf(stderr,"Please specify the amount of memory to test in megabytes.\n"); 3981 fprintf(stderr,"Example: ./redis-server --test-memory 4096\n\n"); 3982 exit(1); 3983 } 3984 } 3985 3986 /* First argument is the config file name? */ 3987 if (argv[j][0] != '-' || argv[j][1] != '-') { 3988 configfile = argv[j]; 3989 server.configfile = getAbsolutePath(configfile); 3990 /* Replace the config file in server.exec_argv with 3991 * its absoulte path. */ 3992 zfree(server.exec_argv[j]); 3993 server.exec_argv[j] = zstrdup(server.configfile); 3994 j++; 3995 } 3996 3997 /* All the other options are parsed and conceptually appended to the 3998 * configuration file. For instance --port 6380 will generate the 3999 * string "port 6380\n" to be parsed after the actual file name 4000 * is parsed, if any. */ 4001 while(j != argc) { 4002 if (argv[j][0] == '-' && argv[j][1] == '-') { 4003 /* Option name */ 4004 if (!strcmp(argv[j], "--check-rdb")) { 4005 /* Argument has no options, need to skip for parsing. */ 4006 j++; 4007 continue; 4008 } 4009 if (sdslen(options)) options = sdscat(options,"\n"); 4010 options = sdscat(options,argv[j]+2); 4011 options = sdscat(options," "); 4012 } else { 4013 /* Option argument */ 4014 options = sdscatrepr(options,argv[j],strlen(argv[j])); 4015 options = sdscat(options," "); 4016 } 4017 j++; 4018 } 4019 if (server.sentinel_mode && configfile && *configfile == '-') { 4020 serverLog(LL_WARNING, 4021 "Sentinel config from STDIN not allowed."); 4022 serverLog(LL_WARNING, 4023 "Sentinel needs config file on disk to save state. Exiting..."); 4024 exit(1); 4025 } 4026 resetServerSaveParams(); 4027 loadServerConfig(configfile,options); 4028 sdsfree(options); 4029 } else { 4030 serverLog(LL_WARNING, "Warning: no config file specified, using the default config. In order to specify a config file use %s /path/to/%s.conf", argv[0], server.sentinel_mode ? "sentinel" : "redis"); 4031 } 4032 4033 server.supervised = redisIsSupervised(server.supervised_mode); 4034 int background = server.daemonize && !server.supervised; 4035 if (background) daemonize(); 4036 4037 initServer(); 4038 if (background || server.pidfile) createPidFile(); 4039 redisSetProcTitle(argv[0]); 4040 redisAsciiArt(); 4041 checkTcpBacklogSettings(); 4042 4043 if (!server.sentinel_mode) { 4044 /* Things not needed when running in Sentinel mode. */ 4045 serverLog(LL_WARNING,"Server started, Redis version " REDIS_VERSION); 4046 #ifdef __linux__ 4047 linuxMemoryWarnings(); 4048 #endif 4049 loadDataFromDisk(); 4050 if (server.cluster_enabled) { 4051 if (verifyClusterConfigWithData() == C_ERR) { 4052 serverLog(LL_WARNING, 4053 "You can't have keys in a DB different than DB 0 when in " 4054 "Cluster mode. Exiting."); 4055 exit(1); 4056 } 4057 } 4058 if (server.ipfd_count > 0) 4059 serverLog(LL_NOTICE,"The server is now ready to accept connections on port %d", server.port); 4060 if (server.sofd > 0) 4061 serverLog(LL_NOTICE,"The server is now ready to accept connections at %s", server.unixsocket); 4062 } else { 4063 sentinelIsRunning(); 4064 } 4065 4066 /* Warning the user about suspicious maxmemory setting. */ 4067 if (server.maxmemory > 0 && server.maxmemory < 1024*1024) { 4068 serverLog(LL_WARNING,"WARNING: You specified a maxmemory value that is less than 1MB (current value is %llu bytes). Are you sure this is what you really want?", server.maxmemory); 4069 } 4070 4071 aeSetBeforeSleepProc(server.el,beforeSleep); 4072 aeMain(server.el); 4073 aeDeleteEventLoop(server.el); 4074 return 0; 4075 } 4076 4077 /* The End */ 4078