redis.c

   1 /*
   2  * Copyright (c) 2006-2009, 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 #define REDIS_VERSION "0.900"
  31
  32 #include "fmacros.h"
  33 #include "config.h"
  34
  35 #include <stdio.h>
  36 #include <stdlib.h>
  37 #include <string.h>
  38 #include <time.h>
  39 #include <unistd.h>
  40 #define __USE_POSIX199309
  41 #include <signal.h>
  42
  43 #ifdef HAVE_BACKTRACE
  44 #include <execinfo.h>
  45 #include <ucontext.h>
  46 #endif /* HAVE_BACKTRACE */
  47
  48 #include <sys/wait.h>
  49 #include <errno.h>
  50 #include <assert.h>
  51 #include <ctype.h>
  52 #include <stdarg.h>
  53 #include <inttypes.h>
  54 #include <arpa/inet.h>
  55 #include <sys/stat.h>
  56 #include <fcntl.h>
  57 #include <sys/time.h>
  58 #include <sys/resource.h>
  59 #include <limits.h>
  60
  61 #include "redis.h"
  62 #include "ae.h"     /* Event driven programming library */
  63 #include "sds.h"    /* Dynamic safe strings */
  64 #include "anet.h"   /* Networking the easy way */
  65 #include "dict.h"   /* Hash tables */
  66 #include "adlist.h" /* Linked lists */
  67 #include "zmalloc.h" /* total memory usage aware version of malloc/free */
  68 #include "lzf.h"    /* LZF compression library */
  69 #include "pqsort.h" /* Partial qsort for SORT+LIMIT */
  70
  71 /* Error codes */
  72 #define REDIS_OK                0
  73 #define REDIS_ERR               -1
  74
  75 /* Static server configuration */
  76 #define REDIS_SERVERPORT        6379    /* TCP port */
  77 #define REDIS_MAXIDLETIME       (60*5)  /* default client timeout */
  78 #define REDIS_IOBUF_LEN         1024
  79 #define REDIS_LOADBUF_LEN       1024
  80 #define REDIS_STATIC_ARGS       4
  81 #define REDIS_DEFAULT_DBNUM     16
  82 #define REDIS_CONFIGLINE_MAX    1024
  83 #define REDIS_OBJFREELIST_MAX   1000000 /* Max number of objects to cache */
  84 #define REDIS_MAX_SYNC_TIME     60      /* Slave can't take more to sync */
  85 #define REDIS_EXPIRELOOKUPS_PER_CRON    100 /* try to expire 100 keys/second */
  86 #define REDIS_MAX_WRITE_PER_EVENT (1024*64)
  87 #define REDIS_REQUEST_MAX_SIZE  (1024*1024*256) /* max bytes in inline command */
  88
  89 /* Hash table parameters */
  90 #define REDIS_HT_MINFILL        10      /* Minimal hash table fill 10% */
  91
  92 /* Command flags */
  93 #define REDIS_CMD_BULK          1       /* Bulk write command */
  94 #define REDIS_CMD_INLINE        2       /* Inline command */
  95 /* REDIS_CMD_DENYOOM reserves a longer comment: all the commands marked with
  96    this flags will return an error when the 'maxmemory' option is set in the
  97    config file and the server is using more than maxmemory bytes of memory.
  98    In short this commands are denied on low memory conditions. */
  99 #define REDIS_CMD_DENYOOM       4
 100
 101 /* Object types */
 102 #define REDIS_STRING 0
 103 #define REDIS_LIST 1
 104 #define REDIS_SET 2
 105 #define REDIS_HASH 3
 106
 107 /* Object types only used for dumping to disk */
 108 #define REDIS_EXPIRETIME 253
 109 #define REDIS_SELECTDB 254
 110 #define REDIS_EOF 255
 111
 112 /* Defines related to the dump file format. To store 32 bits lengths for short
 113  * keys requires a lot of space, so we check the most significant 2 bits of
 114  * the first byte to interpreter the length:
 115  *
 116  * 00|000000 => if the two MSB are 00 the len is the 6 bits of this byte
 117  * 01|000000 00000000 =>  01, the len is 14 byes, 6 bits + 8 bits of next byte
 118  * 10|000000 [32 bit integer] => if it's 01, a full 32 bit len will follow
 119  * 11|000000 this means: specially encoded object will follow. The six bits
 120  *           number specify the kind of object that follows.
 121  *           See the REDIS_RDB_ENC_* defines.
 122  *
 123  * Lenghts up to 63 are stored using a single byte, most DB keys, and may
 124  * values, will fit inside. */
 125 #define REDIS_RDB_6BITLEN 0
 126 #define REDIS_RDB_14BITLEN 1
 127 #define REDIS_RDB_32BITLEN 2
 128 #define REDIS_RDB_ENCVAL 3
 129 #define REDIS_RDB_LENERR UINT_MAX
 130
 131 /* When a length of a string object stored on disk has the first two bits
 132  * set, the remaining two bits specify a special encoding for the object
 133  * accordingly to the following defines: */
 134 #define REDIS_RDB_ENC_INT8 0        /* 8 bit signed integer */
 135 #define REDIS_RDB_ENC_INT16 1       /* 16 bit signed integer */
 136 #define REDIS_RDB_ENC_INT32 2       /* 32 bit signed integer */
 137 #define REDIS_RDB_ENC_LZF 3         /* string compressed with FASTLZ */
 138
 139 /* Client flags */
 140 #define REDIS_CLOSE 1       /* This client connection should be closed ASAP */
 141 #define REDIS_SLAVE 2       /* This client is a slave server */
 142 #define REDIS_MASTER 4      /* This client is a master server */
 143 #define REDIS_MONITOR 8      /* This client is a slave monitor, see MONITOR */
 144
 145 /* Slave replication state - slave side */
 146 #define REDIS_REPL_NONE 0   /* No active replication */
 147 #define REDIS_REPL_CONNECT 1    /* Must connect to master */
 148 #define REDIS_REPL_CONNECTED 2  /* Connected to master */
 149
 150 /* Slave replication state - from the point of view of master
 151  * Note that in SEND_BULK and ONLINE state the slave receives new updates
 152  * in its output queue. In the WAIT_BGSAVE state instead the server is waiting
 153  * to start the next background saving in order to send updates to it. */
 154 #define REDIS_REPL_WAIT_BGSAVE_START 3 /* master waits bgsave to start feeding it */
 155 #define REDIS_REPL_WAIT_BGSAVE_END 4 /* master waits bgsave to start bulk DB transmission */
 156 #define REDIS_REPL_SEND_BULK 5 /* master is sending the bulk DB */
 157 #define REDIS_REPL_ONLINE 6 /* bulk DB already transmitted, receive updates */
 158
 159 /* List related stuff */
 160 #define REDIS_HEAD 0
 161 #define REDIS_TAIL 1
 162
 163 /* Sort operations */
 164 #define REDIS_SORT_GET 0
 165 #define REDIS_SORT_DEL 1
 166 #define REDIS_SORT_INCR 2
 167 #define REDIS_SORT_DECR 3
 168 #define REDIS_SORT_ASC 4
 169 #define REDIS_SORT_DESC 5
 170 #define REDIS_SORTKEY_MAX 1024
 171
 172 /* Log levels */
 173 #define REDIS_DEBUG 0
 174 #define REDIS_NOTICE 1
 175 #define REDIS_WARNING 2
 176
 177 /* Anti-warning macro... */
 178 #define REDIS_NOTUSED(V) ((void) V)
 179
 180
 181 /*================================= Data types ============================== */
 182
 183 /* A redis object, that is a type able to hold a string / list / set */
 184 typedef struct redisObject {
 185     void *ptr;
 186     int type;
 187     int refcount;
 188 } robj;
 189
 190 typedef struct redisDb {
 191     dict *dict;
 192     dict *expires;
 193     int id;
 194 } redisDb;
 195
 196 /* With multiplexing we need to take per-clinet state.
 197  * Clients are taken in a liked list. */
 198 typedef struct redisClient {
 199     int fd;
 200     redisDb *db;
 201     int dictid;
 202     sds querybuf;
 203     robj **argv;
 204     int argc;
 205     int bulklen;            /* bulk read len. -1 if not in bulk read mode */
 206     list *reply;
 207     int sentlen;
 208     time_t lastinteraction; /* time of the last interaction, used for timeout */
 209     int flags;              /* REDIS_CLOSE | REDIS_SLAVE | REDIS_MONITOR */
 210     int slaveseldb;         /* slave selected db, if this client is a slave */
 211     int authenticated;      /* when requirepass is non-NULL */
 212     int replstate;          /* replication state if this is a slave */
 213     int repldbfd;           /* replication DB file descriptor */
 214     long repldboff;          /* replication DB file offset */
 215     off_t repldbsize;       /* replication DB file size */
 216 } redisClient;
 217
 218 struct saveparam {
 219     time_t seconds;
 220     int changes;
 221 };
 222
 223 /* Global server state structure */
 224 struct redisServer {
 225     int port;
 226     int fd;
 227     redisDb *db;
 228     dict *sharingpool;
 229     unsigned int sharingpoolsize;
 230     long long dirty;            /* changes to DB from the last save */
 231     list *clients;
 232     list *slaves, *monitors;
 233     char neterr[ANET_ERR_LEN];
 234     aeEventLoop *el;
 235     int cronloops;              /* number of times the cron function run */
 236     list *objfreelist;          /* A list of freed objects to avoid malloc() */
 237     time_t lastsave;            /* Unix time of last save succeeede */
 238     size_t usedmemory;             /* Used memory in megabytes */
 239     /* Fields used only for stats */
 240     time_t stat_starttime;         /* server start time */
 241     long long stat_numcommands;    /* number of processed commands */
 242     long long stat_numconnections; /* number of connections received */
 243     /* Configuration */
 244     int verbosity;
 245     int glueoutputbuf;
 246     int maxidletime;
 247     int dbnum;
 248     int daemonize;
 249     char *pidfile;
 250     int bgsaveinprogress;
 251     pid_t bgsavechildpid;
 252     struct saveparam *saveparams;
 253     int saveparamslen;
 254     char *logfile;
 255     char *bindaddr;
 256     char *dbfilename;
 257     char *requirepass;
 258     int shareobjects;
 259     /* Replication related */
 260     int isslave;
 261     char *masterhost;
 262     int masterport;
 263     redisClient *master;    /* client that is master for this slave */
 264     int replstate;
 265     unsigned int maxclients;
 266     unsigned int maxmemory;
 267     /* Sort parameters - qsort_r() is only available under BSD so we
 268      * have to take this state global, in order to pass it to sortCompare() */
 269     int sort_desc;
 270     int sort_alpha;
 271     int sort_bypattern;
 272 };
 273
 274 typedef void redisCommandProc(redisClient *c);
 275 struct redisCommand {
 276     char *name;
 277     redisCommandProc *proc;
 278     int arity;
 279     int flags;
 280 };
 281
 282 struct redisFunctionSym {
 283     char *name;
 284     unsigned long pointer;
 285 };
 286
 287 typedef struct _redisSortObject {
 288     robj *obj;
 289     union {
 290         double score;
 291         robj *cmpobj;
 292     } u;
 293 } redisSortObject;
 294
 295 typedef struct _redisSortOperation {
 296     int type;
 297     robj *pattern;
 298 } redisSortOperation;
 299
 300 struct sharedObjectsStruct {
 301     robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *pong, *space,
 302     *colon, *nullbulk, *nullmultibulk,
 303     *emptymultibulk, *wrongtypeerr, *nokeyerr, *syntaxerr, *sameobjecterr,
 304     *outofrangeerr, *plus,
 305     *select0, *select1, *select2, *select3, *select4,
 306     *select5, *select6, *select7, *select8, *select9;
 307 } shared;
 308
 309 /*================================ Prototypes =============================== */
 310
 311 static void freeStringObject(robj *o);
 312 static void freeListObject(robj *o);
 313 static void freeSetObject(robj *o);
 314 static void decrRefCount(void *o);
 315 static robj *createObject(int type, void *ptr);
 316 static void freeClient(redisClient *c);
 317 static int rdbLoad(char *filename);
 318 static void addReply(redisClient *c, robj *obj);
 319 static void addReplySds(redisClient *c, sds s);
 320 static void incrRefCount(robj *o);
 321 static int rdbSaveBackground(char *filename);
 322 static robj *createStringObject(char *ptr, size_t len);
 323 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc);
 324 static int syncWithMaster(void);
 325 static robj *tryObjectSharing(robj *o);
 326 static int removeExpire(redisDb *db, robj *key);
 327 static int expireIfNeeded(redisDb *db, robj *key);
 328 static int deleteIfVolatile(redisDb *db, robj *key);
 329 static int deleteKey(redisDb *db, robj *key);
 330 static time_t getExpire(redisDb *db, robj *key);
 331 static int setExpire(redisDb *db, robj *key, time_t when);
 332 static void updateSlavesWaitingBgsave(int bgsaveerr);
 333 static void freeMemoryIfNeeded(void);
 334 static int processCommand(redisClient *c);
 335 static void setupSigSegvAction(void);
 336 static void rdbRemoveTempFile(pid_t childpid);
 337
 338 static void authCommand(redisClient *c);
 339 static void pingCommand(redisClient *c);
 340 static void echoCommand(redisClient *c);
 341 static void setCommand(redisClient *c);
 342 static void setnxCommand(redisClient *c);
 343 static void getCommand(redisClient *c);
 344 static void delCommand(redisClient *c);
 345 static void existsCommand(redisClient *c);
 346 static void incrCommand(redisClient *c);
 347 static void decrCommand(redisClient *c);
 348 static void incrbyCommand(redisClient *c);
 349 static void decrbyCommand(redisClient *c);
 350 static void selectCommand(redisClient *c);
 351 static void randomkeyCommand(redisClient *c);
 352 static void keysCommand(redisClient *c);
 353 static void dbsizeCommand(redisClient *c);
 354 static void lastsaveCommand(redisClient *c);
 355 static void saveCommand(redisClient *c);
 356 static void bgsaveCommand(redisClient *c);
 357 static void shutdownCommand(redisClient *c);
 358 static void moveCommand(redisClient *c);
 359 static void renameCommand(redisClient *c);
 360 static void renamenxCommand(redisClient *c);
 361 static void lpushCommand(redisClient *c);
 362 static void rpushCommand(redisClient *c);
 363 static void lpopCommand(redisClient *c);
 364 static void rpopCommand(redisClient *c);
 365 static void llenCommand(redisClient *c);
 366 static void lindexCommand(redisClient *c);
 367 static void lrangeCommand(redisClient *c);
 368 static void ltrimCommand(redisClient *c);
 369 static void typeCommand(redisClient *c);
 370 static void lsetCommand(redisClient *c);
 371 static void saddCommand(redisClient *c);
 372 static void sremCommand(redisClient *c);
 373 static void smoveCommand(redisClient *c);
 374 static void sismemberCommand(redisClient *c);
 375 static void scardCommand(redisClient *c);
 376 static void spopCommand(redisClient *c);
 377 static void sinterCommand(redisClient *c);
 378 static void sinterstoreCommand(redisClient *c);
 379 static void sunionCommand(redisClient *c);
 380 static void sunionstoreCommand(redisClient *c);
 381 static void sdiffCommand(redisClient *c);
 382 static void sdiffstoreCommand(redisClient *c);
 383 static void syncCommand(redisClient *c);
 384 static void flushdbCommand(redisClient *c);
 385 static void flushallCommand(redisClient *c);
 386 static void sortCommand(redisClient *c);
 387 static void lremCommand(redisClient *c);
 388 static void infoCommand(redisClient *c);
 389 static void mgetCommand(redisClient *c);
 390 static void monitorCommand(redisClient *c);
 391 static void expireCommand(redisClient *c);
 392 static void getSetCommand(redisClient *c);
 393 static void ttlCommand(redisClient *c);
 394 static void slaveofCommand(redisClient *c);
 395 static void debugCommand(redisClient *c);
 396 /*================================= Globals ================================= */
 397
 398 /* Global vars */
 399 static struct redisServer server; /* server global state */
 400 static struct redisCommand cmdTable[] = {
 401     {"get",getCommand,2,REDIS_CMD_INLINE},
 402     {"set",setCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
 403     {"setnx",setnxCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
 404     {"del",delCommand,-2,REDIS_CMD_INLINE},
 405     {"exists",existsCommand,2,REDIS_CMD_INLINE},
 406     {"incr",incrCommand,2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
 407     {"decr",decrCommand,2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
 408     {"mget",mgetCommand,-2,REDIS_CMD_INLINE},
 409     {"rpush",rpushCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
 410     {"lpush",lpushCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
 411     {"rpop",rpopCommand,2,REDIS_CMD_INLINE},
 412     {"lpop",lpopCommand,2,REDIS_CMD_INLINE},
 413     {"llen",llenCommand,2,REDIS_CMD_INLINE},
 414     {"lindex",lindexCommand,3,REDIS_CMD_INLINE},
 415     {"lset",lsetCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
 416     {"lrange",lrangeCommand,4,REDIS_CMD_INLINE},
 417     {"ltrim",ltrimCommand,4,REDIS_CMD_INLINE},
 418     {"lrem",lremCommand,4,REDIS_CMD_BULK},
 419     {"sadd",saddCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
 420     {"srem",sremCommand,3,REDIS_CMD_BULK},
 421     {"smove",smoveCommand,4,REDIS_CMD_BULK},
 422     {"sismember",sismemberCommand,3,REDIS_CMD_BULK},
 423     {"scard",scardCommand,2,REDIS_CMD_INLINE},
 424     {"spop",spopCommand,2,REDIS_CMD_INLINE},
 425     {"sinter",sinterCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
 426     {"sinterstore",sinterstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
 427     {"sunion",sunionCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
 428     {"sunionstore",sunionstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
 429     {"sdiff",sdiffCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
 430     {"sdiffstore",sdiffstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
 431     {"smembers",sinterCommand,2,REDIS_CMD_INLINE},
 432     {"incrby",incrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
 433     {"decrby",decrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
 434     {"getset",getSetCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
 435     {"randomkey",randomkeyCommand,1,REDIS_CMD_INLINE},
 436     {"select",selectCommand,2,REDIS_CMD_INLINE},
 437     {"move",moveCommand,3,REDIS_CMD_INLINE},
 438     {"rename",renameCommand,3,REDIS_CMD_INLINE},
 439     {"renamenx",renamenxCommand,3,REDIS_CMD_INLINE},
 440     {"expire",expireCommand,3,REDIS_CMD_INLINE},
 441     {"keys",keysCommand,2,REDIS_CMD_INLINE},
 442     {"dbsize",dbsizeCommand,1,REDIS_CMD_INLINE},
 443     {"auth",authCommand,2,REDIS_CMD_INLINE},
 444     {"ping",pingCommand,1,REDIS_CMD_INLINE},
 445     {"echo",echoCommand,2,REDIS_CMD_BULK},
 446     {"save",saveCommand,1,REDIS_CMD_INLINE},
 447     {"bgsave",bgsaveCommand,1,REDIS_CMD_INLINE},
 448     {"shutdown",shutdownCommand,1,REDIS_CMD_INLINE},
 449     {"lastsave",lastsaveCommand,1,REDIS_CMD_INLINE},
 450     {"type",typeCommand,2,REDIS_CMD_INLINE},
 451     {"sync",syncCommand,1,REDIS_CMD_INLINE},
 452     {"flushdb",flushdbCommand,1,REDIS_CMD_INLINE},
 453     {"flushall",flushallCommand,1,REDIS_CMD_INLINE},
 454     {"sort",sortCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
 455     {"info",infoCommand,1,REDIS_CMD_INLINE},
 456     {"monitor",monitorCommand,1,REDIS_CMD_INLINE},
 457     {"ttl",ttlCommand,2,REDIS_CMD_INLINE},
 458     {"slaveof",slaveofCommand,3,REDIS_CMD_INLINE},
 459     {"debug",debugCommand,-2,REDIS_CMD_INLINE},
 460     {NULL,NULL,0,0}
 461 };
 462 /*============================ Utility functions ============================ */
 463
 464 /* Glob-style pattern matching. */
 465 int stringmatchlen(const char *pattern, int patternLen,
 466         const char *string, int stringLen, int nocase)
 467 {
 468     while(patternLen) {
 469         switch(pattern[0]) {
 470         case '*':
 471             while (pattern[1] == '*') {
 472                 pattern++;
 473                 patternLen--;
 474             }
 475             if (patternLen == 1)
 476                 return 1; /* match */
 477             while(stringLen) {
 478                 if (stringmatchlen(pattern+1, patternLen-1,
 479                             string, stringLen, nocase))
 480                     return 1; /* match */
 481                 string++;
 482                 stringLen--;
 483             }
 484             return 0; /* no match */
 485             break;
 486         case '?':
 487             if (stringLen == 0)
 488                 return 0; /* no match */
 489             string++;
 490             stringLen--;
 491             break;
 492         case '[':
 493         {
 494             int not, match;
 495
 496             pattern++;
 497             patternLen--;
 498             not = pattern[0] == '^';
 499             if (not) {
 500                 pattern++;
 501                 patternLen--;
 502             }
 503             match = 0;
 504             while(1) {
 505                 if (pattern[0] == '\\') {
 506                     pattern++;
 507                     patternLen--;
 508                     if (pattern[0] == string[0])
 509                         match = 1;
 510                 } else if (pattern[0] == ']') {
 511                     break;
 512                 } else if (patternLen == 0) {
 513                     pattern--;
 514                     patternLen++;
 515                     break;
 516                 } else if (pattern[1] == '-' && patternLen >= 3) {
 517                     int start = pattern[0];
 518                     int end = pattern[2];
 519                     int c = string[0];
 520                     if (start > end) {
 521                         int t = start;
 522                         start = end;
 523                         end = t;
 524                     }
 525                     if (nocase) {
 526                         start = tolower(start);
 527                         end = tolower(end);
 528                         c = tolower(c);
 529                     }
 530                     pattern += 2;
 531                     patternLen -= 2;
 532                     if (c >= start && c <= end)
 533                         match = 1;
 534                 } else {
 535                     if (!nocase) {
 536                         if (pattern[0] == string[0])
 537                             match = 1;
 538                     } else {
 539                         if (tolower((int)pattern[0]) == tolower((int)string[0]))
 540                             match = 1;
 541                     }
 542                 }
 543                 pattern++;
 544                 patternLen--;
 545             }
 546             if (not)
 547                 match = !match;
 548             if (!match)
 549                 return 0; /* no match */
 550             string++;
 551             stringLen--;
 552             break;
 553         }
 554         case '\\':
 555             if (patternLen >= 2) {
 556                 pattern++;
 557                 patternLen--;
 558             }
 559             /* fall through */
 560         default:
 561             if (!nocase) {
 562                 if (pattern[0] != string[0])
 563                     return 0; /* no match */
 564             } else {
 565                 if (tolower((int)pattern[0]) != tolower((int)string[0]))
 566                     return 0; /* no match */
 567             }
 568             string++;
 569             stringLen--;
 570             break;
 571         }
 572         pattern++;
 573         patternLen--;
 574         if (stringLen == 0) {
 575             while(*pattern == '*') {
 576                 pattern++;
 577                 patternLen--;
 578             }
 579             break;
 580         }
 581     }
 582     if (patternLen == 0 && stringLen == 0)
 583         return 1;
 584     return 0;
 585 }
 586
 587 static void redisLog(int level, const char *fmt, ...) {
 588     va_list ap;
 589     FILE *fp;
 590
 591     fp = (server.logfile == NULL) ? stdout : fopen(server.logfile,"a");
 592     if (!fp) return;
 593
 594     va_start(ap, fmt);
 595     if (level >= server.verbosity) {
 596         char *c = ".-*";
 597         char buf[64];
 598         time_t now;
 599
 600         now = time(NULL);
 601         strftime(buf,64,"%d %b %H:%M:%S",gmtime(&now));
 602         fprintf(fp,"%s %c ",buf,c[level]);
 603         vfprintf(fp, fmt, ap);
 604         fprintf(fp,"\n");
 605         fflush(fp);
 606     }
 607     va_end(ap);
 608
 609     if (server.logfile) fclose(fp);
 610 }
 611
 612 /*====================== Hash table type implementation  ==================== */
 613
 614 /* This is an hash table type that uses the SDS dynamic strings libary as
 615  * keys and radis objects as values (objects can hold SDS strings,
 616  * lists, sets). */
 617
 618 static int sdsDictKeyCompare(void *privdata, const void *key1,
 619         const void *key2)
 620 {
 621     int l1,l2;
 622     DICT_NOTUSED(privdata);
 623
 624     l1 = sdslen((sds)key1);
 625     l2 = sdslen((sds)key2);
 626     if (l1 != l2) return 0;
 627     return memcmp(key1, key2, l1) == 0;
 628 }
 629
 630 static void dictRedisObjectDestructor(void *privdata, void *val)
 631 {
 632     DICT_NOTUSED(privdata);
 633
 634     decrRefCount(val);
 635 }
 636
 637 static int dictSdsKeyCompare(void *privdata, const void *key1,
 638         const void *key2)
 639 {
 640     const robj *o1 = key1, *o2 = key2;
 641     return sdsDictKeyCompare(privdata,o1->ptr,o2->ptr);
 642 }
 643
 644 static unsigned int dictSdsHash(const void *key) {
 645     const robj *o = key;
 646     return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
 647 }
 648
 649 static dictType setDictType = {
 650     dictSdsHash,               /* hash function */
 651     NULL,                      /* key dup */
 652     NULL,                      /* val dup */
 653     dictSdsKeyCompare,         /* key compare */
 654     dictRedisObjectDestructor, /* key destructor */
 655     NULL                       /* val destructor */
 656 };
 657
 658 static dictType hashDictType = {
 659     dictSdsHash,                /* hash function */
 660     NULL,                       /* key dup */
 661     NULL,                       /* val dup */
 662     dictSdsKeyCompare,          /* key compare */
 663     dictRedisObjectDestructor,  /* key destructor */
 664     dictRedisObjectDestructor   /* val destructor */
 665 };
 666
 667 /* ========================= Random utility functions ======================= */
 668
 669 /* Redis generally does not try to recover from out of memory conditions
 670  * when allocating objects or strings, it is not clear if it will be possible
 671  * to report this condition to the client since the networking layer itself
 672  * is based on heap allocation for send buffers, so we simply abort.
 673  * At least the code will be simpler to read... */
 674 static void oom(const char *msg) {
 675     fprintf(stderr, "%s: Out of memory\n",msg);
 676     fflush(stderr);
 677     sleep(1);
 678     abort();
 679 }
 680
 681 /* ====================== Redis server networking stuff ===================== */
 682 static void closeTimedoutClients(void) {
 683     redisClient *c;
 684     listNode *ln;
 685     time_t now = time(NULL);
 686
 687     listRewind(server.clients);
 688     while ((ln = listYield(server.clients)) != NULL) {
 689         c = listNodeValue(ln);
 690         if (!(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
 691             !(c->flags & REDIS_MASTER) &&   /* no timeout for masters */
 692              (now - c->lastinteraction > server.maxidletime)) {
 693             redisLog(REDIS_DEBUG,"Closing idle client");
 694             freeClient(c);
 695         }
 696     }
 697 }
 698
 699 static int htNeedsResize(dict *dict) {
 700     long long size, used;
 701
 702     size = dictSlots(dict);
 703     used = dictSize(dict);
 704     return (size && used && size > DICT_HT_INITIAL_SIZE &&
 705             (used*100/size < REDIS_HT_MINFILL));
 706 }
 707
 708 /* If the percentage of used slots in the HT reaches REDIS_HT_MINFILL
 709  * we resize the hash table to save memory */
 710 static void tryResizeHashTables(void) {
 711     int j;
 712
 713     for (j = 0; j < server.dbnum; j++) {
 714         if (htNeedsResize(server.db[j].dict)) {
 715             redisLog(REDIS_DEBUG,"The hash table %d is too sparse, resize it...",j);
 716             dictResize(server.db[j].dict);
 717             redisLog(REDIS_DEBUG,"Hash table %d resized.",j);
 718         }
 719         if (htNeedsResize(server.db[j].expires))
 720             dictResize(server.db[j].expires);
 721     }
 722 }
 723
 724 static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
 725     int j, loops = server.cronloops++;
 726     REDIS_NOTUSED(eventLoop);
 727     REDIS_NOTUSED(id);
 728     REDIS_NOTUSED(clientData);
 729
 730     /* Update the global state with the amount of used memory */
 731     server.usedmemory = zmalloc_used_memory();
 732
 733     /* Show some info about non-empty databases */
 734     for (j = 0; j < server.dbnum; j++) {
 735         long long size, used, vkeys;
 736
 737         size = dictSlots(server.db[j].dict);
 738         used = dictSize(server.db[j].dict);
 739         vkeys = dictSize(server.db[j].expires);
 740         if (!(loops % 5) && (used || vkeys)) {
 741             redisLog(REDIS_DEBUG,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size);
 742             /* dictPrintStats(server.dict); */
 743         }
 744     }
 745
 746     /* We don't want to resize the hash tables while a bacground saving
 747      * is in progress: the saving child is created using fork() that is
 748      * implemented with a copy-on-write semantic in most modern systems, so
 749      * if we resize the HT while there is the saving child at work actually
 750      * a lot of memory movements in the parent will cause a lot of pages
 751      * copied. */
 752     if (!server.bgsaveinprogress) tryResizeHashTables();
 753
 754     /* Show information about connected clients */
 755     if (!(loops % 5)) {
 756         redisLog(REDIS_DEBUG,"%d clients connected (%d slaves), %zu bytes in use, %d shared objects",
 757             listLength(server.clients)-listLength(server.slaves),
 758             listLength(server.slaves),
 759             server.usedmemory,
 760             dictSize(server.sharingpool));
 761     }
 762
 763     /* Close connections of timedout clients */
 764     if (server.maxidletime && !(loops % 10))
 765         closeTimedoutClients();
 766
 767     /* Check if a background saving in progress terminated */
 768     if (server.bgsaveinprogress) {
 769         int statloc;
 770         if (wait4(-1,&statloc,WNOHANG,NULL)) {
 771             int exitcode = WEXITSTATUS(statloc);
 772             int bysignal = WIFSIGNALED(statloc);
 773
 774             if (!bysignal && exitcode == 0) {
 775                 redisLog(REDIS_NOTICE,
 776                     "Background saving terminated with success");
 777                 server.dirty = 0;
 778                 server.lastsave = time(NULL);
 779             } else if (!bysignal && exitcode != 0) {
 780                 redisLog(REDIS_WARNING, "Background saving error");
 781             } else {
 782                 redisLog(REDIS_WARNING,
 783                     "Background saving terminated by signal");
 784                 rdbRemoveTempFile(server.bgsavechildpid);
 785             }
 786             server.bgsaveinprogress = 0;
 787             server.bgsavechildpid = -1;
 788             updateSlavesWaitingBgsave(exitcode == 0 ? REDIS_OK : REDIS_ERR);
 789         }
 790     } else {
 791         /* If there is not a background saving in progress check if
 792          * we have to save now */
 793          time_t now = time(NULL);
 794          for (j = 0; j < server.saveparamslen; j++) {
 795             struct saveparam *sp = server.saveparams+j;
 796
 797             if (server.dirty >= sp->changes &&
 798                 now-server.lastsave > sp->seconds) {
 799                 redisLog(REDIS_NOTICE,"%d changes in %d seconds. Saving...",
 800                     sp->changes, sp->seconds);
 801                 rdbSaveBackground(server.dbfilename);
 802                 break;
 803             }
 804          }
 805     }
 806
 807     /* Try to expire a few timed out keys */
 808     for (j = 0; j < server.dbnum; j++) {
 809         redisDb *db = server.db+j;
 810         int num = dictSize(db->expires);
 811
 812         if (num) {
 813             time_t now = time(NULL);
 814
 815             if (num > REDIS_EXPIRELOOKUPS_PER_CRON)
 816                 num = REDIS_EXPIRELOOKUPS_PER_CRON;
 817             while (num--) {
 818                 dictEntry *de;
 819                 time_t t;
 820
 821                 if ((de = dictGetRandomKey(db->expires)) == NULL) break;
 822                 t = (time_t) dictGetEntryVal(de);
 823                 if (now > t) {
 824                     deleteKey(db,dictGetEntryKey(de));
 825                 }
 826             }
 827         }
 828     }
 829
 830     /* Check if we should connect to a MASTER */
 831     if (server.replstate == REDIS_REPL_CONNECT) {
 832         redisLog(REDIS_NOTICE,"Connecting to MASTER...");
 833         if (syncWithMaster() == REDIS_OK) {
 834             redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync succeeded");
 835         }
 836     }
 837     return 1000;
 838 }
 839
 840 static void createSharedObjects(void) {
 841     shared.crlf = createObject(REDIS_STRING,sdsnew("\r\n"));
 842     shared.ok = createObject(REDIS_STRING,sdsnew("+OK\r\n"));
 843     shared.err = createObject(REDIS_STRING,sdsnew("-ERR\r\n"));
 844     shared.emptybulk = createObject(REDIS_STRING,sdsnew("$0\r\n\r\n"));
 845     shared.czero = createObject(REDIS_STRING,sdsnew(":0\r\n"));
 846     shared.cone = createObject(REDIS_STRING,sdsnew(":1\r\n"));
 847     shared.nullbulk = createObject(REDIS_STRING,sdsnew("$-1\r\n"));
 848     shared.nullmultibulk = createObject(REDIS_STRING,sdsnew("*-1\r\n"));
 849     shared.emptymultibulk = createObject(REDIS_STRING,sdsnew("*0\r\n"));
 850     /* no such key */
 851     shared.pong = createObject(REDIS_STRING,sdsnew("+PONG\r\n"));
 852     shared.wrongtypeerr = createObject(REDIS_STRING,sdsnew(
 853         "-ERR Operation against a key holding the wrong kind of value\r\n"));
 854     shared.nokeyerr = createObject(REDIS_STRING,sdsnew(
 855         "-ERR no such key\r\n"));
 856     shared.syntaxerr = createObject(REDIS_STRING,sdsnew(
 857         "-ERR syntax error\r\n"));
 858     shared.sameobjecterr = createObject(REDIS_STRING,sdsnew(
 859         "-ERR source and destination objects are the same\r\n"));
 860     shared.outofrangeerr = createObject(REDIS_STRING,sdsnew(
 861         "-ERR index out of range\r\n"));
 862     shared.space = createObject(REDIS_STRING,sdsnew(" "));
 863     shared.colon = createObject(REDIS_STRING,sdsnew(":"));
 864     shared.plus = createObject(REDIS_STRING,sdsnew("+"));
 865     shared.select0 = createStringObject("select 0\r\n",10);
 866     shared.select1 = createStringObject("select 1\r\n",10);
 867     shared.select2 = createStringObject("select 2\r\n",10);
 868     shared.select3 = createStringObject("select 3\r\n",10);
 869     shared.select4 = createStringObject("select 4\r\n",10);
 870     shared.select5 = createStringObject("select 5\r\n",10);
 871     shared.select6 = createStringObject("select 6\r\n",10);
 872     shared.select7 = createStringObject("select 7\r\n",10);
 873     shared.select8 = createStringObject("select 8\r\n",10);
 874     shared.select9 = createStringObject("select 9\r\n",10);
 875 }
 876
 877 static void appendServerSaveParams(time_t seconds, int changes) {
 878     server.saveparams = zrealloc(server.saveparams,sizeof(struct saveparam)*(server.saveparamslen+1));
 879     if (server.saveparams == NULL) oom("appendServerSaveParams");
 880     server.saveparams[server.saveparamslen].seconds = seconds;
 881     server.saveparams[server.saveparamslen].changes = changes;
 882     server.saveparamslen++;
 883 }
 884
 885 static void ResetServerSaveParams() {
 886     zfree(server.saveparams);
 887     server.saveparams = NULL;
 888     server.saveparamslen = 0;
 889 }
 890
 891 static void initServerConfig() {
 892     server.dbnum = REDIS_DEFAULT_DBNUM;
 893     server.port = REDIS_SERVERPORT;
 894     server.verbosity = REDIS_DEBUG;
 895     server.maxidletime = REDIS_MAXIDLETIME;
 896     server.saveparams = NULL;
 897     server.logfile = NULL; /* NULL = log on standard output */
 898     server.bindaddr = NULL;
 899     server.glueoutputbuf = 1;
 900     server.daemonize = 0;
 901     server.pidfile = "/var/run/redis.pid";
 902     server.dbfilename = "dump.rdb";
 903     server.requirepass = NULL;
 904     server.shareobjects = 0;
 905     server.sharingpoolsize = 1024;
 906     server.maxclients = 0;
 907     server.maxmemory = 0;
 908     ResetServerSaveParams();
 909
 910     appendServerSaveParams(60*60,1);  /* save after 1 hour and 1 change */
 911     appendServerSaveParams(300,100);  /* save after 5 minutes and 100 changes */
 912     appendServerSaveParams(60,10000); /* save after 1 minute and 10000 changes */
 913     /* Replication related */
 914     server.isslave = 0;
 915     server.masterhost = NULL;
 916     server.masterport = 6379;
 917     server.master = NULL;
 918     server.replstate = REDIS_REPL_NONE;
 919 }
 920
 921 static void initServer() {
 922     int j;
 923
 924     signal(SIGHUP, SIG_IGN);
 925     signal(SIGPIPE, SIG_IGN);
 926     setupSigSegvAction();
 927
 928     server.clients = listCreate();
 929     server.slaves = listCreate();
 930     server.monitors = listCreate();
 931     server.objfreelist = listCreate();
 932     createSharedObjects();
 933     server.el = aeCreateEventLoop();
 934     server.db = zmalloc(sizeof(redisDb)*server.dbnum);
 935     server.sharingpool = dictCreate(&setDictType,NULL);
 936     if (!server.db || !server.clients || !server.slaves || !server.monitors || !server.el || !server.objfreelist)
 937         oom("server initialization"); /* Fatal OOM */
 938     server.fd = anetTcpServer(server.neterr, server.port, server.bindaddr);
 939     if (server.fd == -1) {
 940         redisLog(REDIS_WARNING, "Opening TCP port: %s", server.neterr);
 941         exit(1);
 942     }
 943     for (j = 0; j < server.dbnum; j++) {
 944         server.db[j].dict = dictCreate(&hashDictType,NULL);
 945         server.db[j].expires = dictCreate(&setDictType,NULL);
 946         server.db[j].id = j;
 947     }
 948     server.cronloops = 0;
 949     server.bgsaveinprogress = 0;
 950     server.bgsavechildpid = -1;
 951     server.lastsave = time(NULL);
 952     server.dirty = 0;
 953     server.usedmemory = 0;
 954     server.stat_numcommands = 0;
 955     server.stat_numconnections = 0;
 956     server.stat_starttime = time(NULL);
 957     aeCreateTimeEvent(server.el, 1000, serverCron, NULL, NULL);
 958 }
 959
 960 /* Empty the whole database */
 961 static long long emptyDb() {
 962     int j;
 963     long long removed = 0;
 964
 965     for (j = 0; j < server.dbnum; j++) {
 966         removed += dictSize(server.db[j].dict);
 967         dictEmpty(server.db[j].dict);
 968         dictEmpty(server.db[j].expires);
 969     }
 970     return removed;
 971 }
 972
 973 static int yesnotoi(char *s) {
 974     if (!strcasecmp(s,"yes")) return 1;
 975     else if (!strcasecmp(s,"no")) return 0;
 976     else return -1;
 977 }
 978
 979 /* I agree, this is a very rudimental way to load a configuration...
 980    will improve later if the config gets more complex */
 981 static void loadServerConfig(char *filename) {
 982     FILE *fp;
 983     char buf[REDIS_CONFIGLINE_MAX+1], *err = NULL;
 984     int linenum = 0;
 985     sds line = NULL;
 986
 987     if (filename[0] == '-' && filename[1] == '\0')
 988         fp = stdin;
 989     else {
 990         if ((fp = fopen(filename,"r")) == NULL) {
 991             redisLog(REDIS_WARNING,"Fatal error, can't open config file");
 992             exit(1);
 993         }
 994     }
 995
 996     while(fgets(buf,REDIS_CONFIGLINE_MAX+1,fp) != NULL) {
 997         sds *argv;
 998         int argc, j;
 999
1000         linenum++;
1001         line = sdsnew(buf);
1002         line = sdstrim(line," \t\r\n");
1003
1004         /* Skip comments and blank lines*/
1005         if (line[0] == '#' || line[0] == '\0') {
1006             sdsfree(line);
1007             continue;
1008         }
1009
1010         /* Split into arguments */
1011         argv = sdssplitlen(line,sdslen(line)," ",1,&argc);
1012         sdstolower(argv[0]);
1013
1014         /* Execute config directives */
1015         if (!strcasecmp(argv[0],"timeout") && argc == 2) {
1016             server.maxidletime = atoi(argv[1]);
1017             if (server.maxidletime < 0) {
1018                 err = "Invalid timeout value"; goto loaderr;
1019             }
1020         } else if (!strcasecmp(argv[0],"port") && argc == 2) {
1021             server.port = atoi(argv[1]);
1022             if (server.port < 1 || server.port > 65535) {
1023                 err = "Invalid port"; goto loaderr;
1024             }
1025         } else if (!strcasecmp(argv[0],"bind") && argc == 2) {
1026             server.bindaddr = zstrdup(argv[1]);
1027         } else if (!strcasecmp(argv[0],"save") && argc == 3) {
1028             int seconds = atoi(argv[1]);
1029             int changes = atoi(argv[2]);
1030             if (seconds < 1 || changes < 0) {
1031                 err = "Invalid save parameters"; goto loaderr;
1032             }
1033             appendServerSaveParams(seconds,changes);
1034         } else if (!strcasecmp(argv[0],"dir") && argc == 2) {
1035             if (chdir(argv[1]) == -1) {
1036                 redisLog(REDIS_WARNING,"Can't chdir to '%s': %s",
1037                     argv[1], strerror(errno));
1038                 exit(1);
1039             }
1040         } else if (!strcasecmp(argv[0],"loglevel") && argc == 2) {
1041             if (!strcasecmp(argv[1],"debug")) server.verbosity = REDIS_DEBUG;
1042             else if (!strcasecmp(argv[1],"notice")) server.verbosity = REDIS_NOTICE;
1043             else if (!strcasecmp(argv[1],"warning")) server.verbosity = REDIS_WARNING;
1044             else {
1045                 err = "Invalid log level. Must be one of debug, notice, warning";
1046                 goto loaderr;
1047             }
1048         } else if (!strcasecmp(argv[0],"logfile") && argc == 2) {
1049             FILE *logfp;
1050
1051             server.logfile = zstrdup(argv[1]);
1052             if (!strcasecmp(server.logfile,"stdout")) {
1053                 zfree(server.logfile);
1054                 server.logfile = NULL;
1055             }
1056             if (server.logfile) {
1057                 /* Test if we are able to open the file. The server will not
1058                  * be able to abort just for this problem later... */
1059                 logfp = fopen(server.logfile,"a");
1060                 if (logfp == NULL) {
1061                     err = sdscatprintf(sdsempty(),
1062                         "Can't open the log file: %s", strerror(errno));
1063                     goto loaderr;
1064                 }
1065                 fclose(logfp);
1066             }
1067         } else if (!strcasecmp(argv[0],"databases") && argc == 2) {
1068             server.dbnum = atoi(argv[1]);
1069             if (server.dbnum < 1) {
1070                 err = "Invalid number of databases"; goto loaderr;
1071             }
1072         } else if (!strcasecmp(argv[0],"maxclients") && argc == 2) {
1073             server.maxclients = atoi(argv[1]);
1074         } else if (!strcasecmp(argv[0],"maxmemory") && argc == 2) {
1075             server.maxmemory = atoi(argv[1]);
1076         } else if (!strcasecmp(argv[0],"slaveof") && argc == 3) {
1077             server.masterhost = sdsnew(argv[1]);
1078             server.masterport = atoi(argv[2]);
1079             server.replstate = REDIS_REPL_CONNECT;
1080         } else if (!strcasecmp(argv[0],"glueoutputbuf") && argc == 2) {
1081             if ((server.glueoutputbuf = yesnotoi(argv[1])) == -1) {
1082                 err = "argument must be 'yes' or 'no'"; goto loaderr;
1083             }
1084         } else if (!strcasecmp(argv[0],"shareobjects") && argc == 2) {
1085             if ((server.shareobjects = yesnotoi(argv[1])) == -1) {
1086                 err = "argument must be 'yes' or 'no'"; goto loaderr;
1087             }
1088         } else if (!strcasecmp(argv[0],"shareobjectspoolsize") && argc == 2) {
1089             server.sharingpoolsize = atoi(argv[1]);
1090             if (server.sharingpoolsize < 1) {
1091                 err = "invalid object sharing pool size"; goto loaderr;
1092             }
1093         } else if (!strcasecmp(argv[0],"daemonize") && argc == 2) {
1094             if ((server.daemonize = yesnotoi(argv[1])) == -1) {
1095                 err = "argument must be 'yes' or 'no'"; goto loaderr;
1096             }
1097         } else if (!strcasecmp(argv[0],"requirepass") && argc == 2) {
1098           server.requirepass = zstrdup(argv[1]);
1099         } else if (!strcasecmp(argv[0],"pidfile") && argc == 2) {
1100           server.pidfile = zstrdup(argv[1]);
1101         } else if (!strcasecmp(argv[0],"dbfilename") && argc == 2) {
1102           server.dbfilename = zstrdup(argv[1]);
1103         } else {
1104             err = "Bad directive or wrong number of arguments"; goto loaderr;
1105         }
1106         for (j = 0; j < argc; j++)
1107             sdsfree(argv[j]);
1108         zfree(argv);
1109         sdsfree(line);
1110     }
1111     if (fp != stdin) fclose(fp);
1112     return;
1113
1114 loaderr:
1115     fprintf(stderr, "\n*** FATAL CONFIG FILE ERROR ***\n");
1116     fprintf(stderr, "Reading the configuration file, at line %d\n", linenum);
1117     fprintf(stderr, ">>> '%s'\n", line);
1118     fprintf(stderr, "%s\n", err);
1119     exit(1);
1120 }
1121
1122 static void freeClientArgv(redisClient *c) {
1123     int j;
1124
1125     for (j = 0; j < c->argc; j++)
1126         decrRefCount(c->argv[j]);
1127     c->argc = 0;
1128 }
1129
1130 static void freeClient(redisClient *c) {
1131     listNode *ln;
1132
1133     aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
1134     aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
1135     sdsfree(c->querybuf);
1136     listRelease(c->reply);
1137     freeClientArgv(c);
1138     close(c->fd);
1139     ln = listSearchKey(server.clients,c);
1140     assert(ln != NULL);
1141     listDelNode(server.clients,ln);
1142     if (c->flags & REDIS_SLAVE) {
1143         if (c->replstate == REDIS_REPL_SEND_BULK && c->repldbfd != -1)
1144             close(c->repldbfd);
1145         list *l = (c->flags & REDIS_MONITOR) ? server.monitors : server.slaves;
1146         ln = listSearchKey(l,c);
1147         assert(ln != NULL);
1148         listDelNode(l,ln);
1149     }
1150     if (c->flags & REDIS_MASTER) {
1151         server.master = NULL;
1152         server.replstate = REDIS_REPL_CONNECT;
1153     }
1154     zfree(c->argv);
1155     zfree(c);
1156 }
1157
1158 static void glueReplyBuffersIfNeeded(redisClient *c) {
1159     int totlen = 0;
1160     listNode *ln;
1161     robj *o;
1162
1163     listRewind(c->reply);
1164     while((ln = listYield(c->reply))) {
1165         o = ln->value;
1166         totlen += sdslen(o->ptr);
1167         /* This optimization makes more sense if we don't have to copy
1168          * too much data */
1169         if (totlen > 1024) return;
1170     }
1171     if (totlen > 0) {
1172         char buf[1024];
1173         int copylen = 0;
1174
1175         listRewind(c->reply);
1176         while((ln = listYield(c->reply))) {
1177             o = ln->value;
1178             memcpy(buf+copylen,o->ptr,sdslen(o->ptr));
1179             copylen += sdslen(o->ptr);
1180             listDelNode(c->reply,ln);
1181         }
1182         /* Now the output buffer is empty, add the new single element */
1183         o = createObject(REDIS_STRING,sdsnewlen(buf,totlen));
1184         if (!listAddNodeTail(c->reply,o)) oom("listAddNodeTail");
1185     }
1186 }
1187
1188 static void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
1189     redisClient *c = privdata;
1190     int nwritten = 0, totwritten = 0, objlen;
1191     robj *o;
1192     REDIS_NOTUSED(el);
1193     REDIS_NOTUSED(mask);
1194
1195     if (server.glueoutputbuf && listLength(c->reply) > 1)
1196         glueReplyBuffersIfNeeded(c);
1197     while(listLength(c->reply)) {
1198         o = listNodeValue(listFirst(c->reply));
1199         objlen = sdslen(o->ptr);
1200
1201         if (objlen == 0) {
1202             listDelNode(c->reply,listFirst(c->reply));
1203             continue;
1204         }
1205
1206         if (c->flags & REDIS_MASTER) {
1207             /* Don't reply to a master */
1208             nwritten = objlen - c->sentlen;
1209         } else {
1210             nwritten = write(fd, ((char*)o->ptr)+c->sentlen, objlen - c->sentlen);
1211             if (nwritten <= 0) break;
1212         }
1213         c->sentlen += nwritten;
1214         totwritten += nwritten;
1215         /* If we fully sent the object on head go to the next one */
1216         if (c->sentlen == objlen) {
1217             listDelNode(c->reply,listFirst(c->reply));
1218             c->sentlen = 0;
1219         }
1220         /* Note that we avoid to send more thank REDIS_MAX_WRITE_PER_EVENT
1221          * bytes, in a single threaded server it's a good idea to server
1222          * other clients as well, even if a very large request comes from
1223          * super fast link that is always able to accept data (in real world
1224          * terms think to 'KEYS *' against the loopback interfae) */
1225         if (totwritten > REDIS_MAX_WRITE_PER_EVENT) break;
1226     }
1227     if (nwritten == -1) {
1228         if (errno == EAGAIN) {
1229             nwritten = 0;
1230         } else {
1231             redisLog(REDIS_DEBUG,
1232                 "Error writing to client: %s", strerror(errno));
1233             freeClient(c);
1234             return;
1235         }
1236     }
1237     if (totwritten > 0) c->lastinteraction = time(NULL);
1238     if (listLength(c->reply) == 0) {
1239         c->sentlen = 0;
1240         aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
1241     }
1242 }
1243
1244 static struct redisCommand *lookupCommand(char *name) {
1245     int j = 0;
1246     while(cmdTable[j].name != NULL) {
1247         if (!strcasecmp(name,cmdTable[j].name)) return &cmdTable[j];
1248         j++;
1249     }
1250     return NULL;
1251 }
1252
1253 /* resetClient prepare the client to process the next command */
1254 static void resetClient(redisClient *c) {
1255     freeClientArgv(c);
1256     c->bulklen = -1;
1257 }
1258
1259 /* If this function gets called we already read a whole
1260  * command, argments are in the client argv/argc fields.
1261  * processCommand() execute the command or prepare the
1262  * server for a bulk read from the client.
1263  *
1264  * If 1 is returned the client is still alive and valid and
1265  * and other operations can be performed by the caller. Otherwise
1266  * if 0 is returned the client was destroied (i.e. after QUIT). */
1267 static int processCommand(redisClient *c) {
1268     struct redisCommand *cmd;
1269     long long dirty;
1270
1271     /* Free some memory if needed (maxmemory setting) */
1272     if (server.maxmemory) freeMemoryIfNeeded();
1273
1274     /* The QUIT command is handled as a special case. Normal command
1275      * procs are unable to close the client connection safely */
1276     if (!strcasecmp(c->argv[0]->ptr,"quit")) {
1277         freeClient(c);
1278         return 0;
1279     }
1280     cmd = lookupCommand(c->argv[0]->ptr);
1281     if (!cmd) {
1282         addReplySds(c,sdsnew("-ERR unknown command\r\n"));
1283         resetClient(c);
1284         return 1;
1285     } else if ((cmd->arity > 0 && cmd->arity != c->argc) ||
1286                (c->argc < -cmd->arity)) {
1287         addReplySds(c,sdsnew("-ERR wrong number of arguments\r\n"));
1288         resetClient(c);
1289         return 1;
1290     } else if (server.maxmemory && cmd->flags & REDIS_CMD_DENYOOM && zmalloc_used_memory() > server.maxmemory) {
1291         addReplySds(c,sdsnew("-ERR command not allowed when used memory > 'maxmemory'\r\n"));
1292         resetClient(c);
1293         return 1;
1294     } else if (cmd->flags & REDIS_CMD_BULK && c->bulklen == -1) {
1295         int bulklen = atoi(c->argv[c->argc-1]->ptr);
1296
1297         decrRefCount(c->argv[c->argc-1]);
1298         if (bulklen < 0 || bulklen > 1024*1024*1024) {
1299             c->argc--;
1300             addReplySds(c,sdsnew("-ERR invalid bulk write count\r\n"));
1301             resetClient(c);
1302             return 1;
1303         }
1304         c->argc--;
1305         c->bulklen = bulklen+2; /* add two bytes for CR+LF */
1306         /* It is possible that the bulk read is already in the
1307          * buffer. Check this condition and handle it accordingly */
1308         if ((signed)sdslen(c->querybuf) >= c->bulklen) {
1309             c->argv[c->argc] = createStringObject(c->querybuf,c->bulklen-2);
1310             c->argc++;
1311             c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
1312         } else {
1313             return 1;
1314         }
1315     }
1316     /* Let's try to share objects on the command arguments vector */
1317     if (server.shareobjects) {
1318         int j;
1319         for(j = 1; j < c->argc; j++)
1320             c->argv[j] = tryObjectSharing(c->argv[j]);
1321     }
1322     /* Check if the user is authenticated */
1323     if (server.requirepass && !c->authenticated && cmd->proc != authCommand) {
1324         addReplySds(c,sdsnew("-ERR operation not permitted\r\n"));
1325         resetClient(c);
1326         return 1;
1327     }
1328
1329     /* Exec the command */
1330     dirty = server.dirty;
1331     cmd->proc(c);
1332     if (server.dirty-dirty != 0 && listLength(server.slaves))
1333         replicationFeedSlaves(server.slaves,cmd,c->db->id,c->argv,c->argc);
1334     if (listLength(server.monitors))
1335         replicationFeedSlaves(server.monitors,cmd,c->db->id,c->argv,c->argc);
1336     server.stat_numcommands++;
1337
1338     /* Prepare the client for the next command */
1339     if (c->flags & REDIS_CLOSE) {
1340         freeClient(c);
1341         return 0;
1342     }
1343     resetClient(c);
1344     return 1;
1345 }
1346
1347 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc) {
1348     listNode *ln;
1349     int outc = 0, j;
1350     robj **outv;
1351     /* (args*2)+1 is enough room for args, spaces, newlines */
1352     robj *static_outv[REDIS_STATIC_ARGS*2+1];
1353
1354     if (argc <= REDIS_STATIC_ARGS) {
1355         outv = static_outv;
1356     } else {
1357         outv = zmalloc(sizeof(robj*)*(argc*2+1));
1358         if (!outv) oom("replicationFeedSlaves");
1359     }
1360
1361     for (j = 0; j < argc; j++) {
1362         if (j != 0) outv[outc++] = shared.space;
1363         if ((cmd->flags & REDIS_CMD_BULK) && j == argc-1) {
1364             robj *lenobj;
1365
1366             lenobj = createObject(REDIS_STRING,
1367                 sdscatprintf(sdsempty(),"%d\r\n",sdslen(argv[j]->ptr)));
1368             lenobj->refcount = 0;
1369             outv[outc++] = lenobj;
1370         }
1371         outv[outc++] = argv[j];
1372     }
1373     outv[outc++] = shared.crlf;
1374
1375     /* Increment all the refcounts at start and decrement at end in order to
1376      * be sure to free objects if there is no slave in a replication state
1377      * able to be feed with commands */
1378     for (j = 0; j < outc; j++) incrRefCount(outv[j]);
1379     listRewind(slaves);
1380     while((ln = listYield(slaves))) {
1381         redisClient *slave = ln->value;
1382
1383         /* Don't feed slaves that are still waiting for BGSAVE to start */
1384         if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) continue;
1385
1386         /* Feed all the other slaves, MONITORs and so on */
1387         if (slave->slaveseldb != dictid) {
1388             robj *selectcmd;
1389
1390             switch(dictid) {
1391             case 0: selectcmd = shared.select0; break;
1392             case 1: selectcmd = shared.select1; break;
1393             case 2: selectcmd = shared.select2; break;
1394             case 3: selectcmd = shared.select3; break;
1395             case 4: selectcmd = shared.select4; break;
1396             case 5: selectcmd = shared.select5; break;
1397             case 6: selectcmd = shared.select6; break;
1398             case 7: selectcmd = shared.select7; break;
1399             case 8: selectcmd = shared.select8; break;
1400             case 9: selectcmd = shared.select9; break;
1401             default:
1402                 selectcmd = createObject(REDIS_STRING,
1403                     sdscatprintf(sdsempty(),"select %d\r\n",dictid));
1404                 selectcmd->refcount = 0;
1405                 break;
1406             }
1407             addReply(slave,selectcmd);
1408             slave->slaveseldb = dictid;
1409         }
1410         for (j = 0; j < outc; j++) addReply(slave,outv[j]);
1411     }
1412     for (j = 0; j < outc; j++) decrRefCount(outv[j]);
1413     if (outv != static_outv) zfree(outv);
1414 }
1415
1416 static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
1417     redisClient *c = (redisClient*) privdata;
1418     char buf[REDIS_IOBUF_LEN];
1419     int nread;
1420     REDIS_NOTUSED(el);
1421     REDIS_NOTUSED(mask);
1422
1423     nread = read(fd, buf, REDIS_IOBUF_LEN);
1424     if (nread == -1) {
1425         if (errno == EAGAIN) {
1426             nread = 0;
1427         } else {
1428             redisLog(REDIS_DEBUG, "Reading from client: %s",strerror(errno));
1429             freeClient(c);
1430             return;
1431         }
1432     } else if (nread == 0) {
1433         redisLog(REDIS_DEBUG, "Client closed connection");
1434         freeClient(c);
1435         return;
1436     }
1437     if (nread) {
1438         c->querybuf = sdscatlen(c->querybuf, buf, nread);
1439         c->lastinteraction = time(NULL);
1440     } else {
1441         return;
1442     }
1443
1444 again:
1445     if (c->bulklen == -1) {
1446         /* Read the first line of the query */
1447         char *p = strchr(c->querybuf,'\n');
1448         size_t querylen;
1449
1450         if (p) {
1451             sds query, *argv;
1452             int argc, j;
1453
1454             query = c->querybuf;
1455             c->querybuf = sdsempty();
1456             querylen = 1+(p-(query));
1457             if (sdslen(query) > querylen) {
1458                 /* leave data after the first line of the query in the buffer */
1459                 c->querybuf = sdscatlen(c->querybuf,query+querylen,sdslen(query)-querylen);
1460             }
1461             *p = '\0'; /* remove "\n" */
1462             if (*(p-1) == '\r') *(p-1) = '\0'; /* and "\r" if any */
1463             sdsupdatelen(query);
1464
1465             /* Now we can split the query in arguments */
1466             if (sdslen(query) == 0) {
1467                 /* Ignore empty query */
1468                 sdsfree(query);
1469                 return;
1470             }
1471             argv = sdssplitlen(query,sdslen(query)," ",1,&argc);
1472             if (argv == NULL) oom("sdssplitlen");
1473             sdsfree(query);
1474
1475             if (c->argv) zfree(c->argv);
1476             c->argv = zmalloc(sizeof(robj*)*argc);
1477             if (c->argv == NULL) oom("allocating arguments list for client");
1478
1479             for (j = 0; j < argc; j++) {
1480                 if (sdslen(argv[j])) {
1481                     c->argv[c->argc] = createObject(REDIS_STRING,argv[j]);
1482                     c->argc++;
1483                 } else {
1484                     sdsfree(argv[j]);
1485                 }
1486             }
1487             zfree(argv);
1488             /* Execute the command. If the client is still valid
1489              * after processCommand() return and there is something
1490              * on the query buffer try to process the next command. */
1491             if (c->argc && processCommand(c) && sdslen(c->querybuf)) goto again;
1492             return;
1493         } else if (sdslen(c->querybuf) >= REDIS_REQUEST_MAX_SIZE) {
1494             redisLog(REDIS_DEBUG, "Client protocol error");
1495             freeClient(c);
1496             return;
1497         }
1498     } else {
1499         /* Bulk read handling. Note that if we are at this point
1500            the client already sent a command terminated with a newline,
1501            we are reading the bulk data that is actually the last
1502            argument of the command. */
1503         int qbl = sdslen(c->querybuf);
1504
1505         if (c->bulklen <= qbl) {
1506             /* Copy everything but the final CRLF as final argument */
1507             c->argv[c->argc] = createStringObject(c->querybuf,c->bulklen-2);
1508             c->argc++;
1509             c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
1510             processCommand(c);
1511             return;
1512         }
1513     }
1514 }
1515
1516 static int selectDb(redisClient *c, int id) {
1517     if (id < 0 || id >= server.dbnum)
1518         return REDIS_ERR;
1519     c->db = &server.db[id];
1520     return REDIS_OK;
1521 }
1522
1523 static void *dupClientReplyValue(void *o) {
1524     incrRefCount((robj*)o);
1525     return 0;
1526 }
1527
1528 static redisClient *createClient(int fd) {
1529     redisClient *c = zmalloc(sizeof(*c));
1530
1531     anetNonBlock(NULL,fd);
1532     anetTcpNoDelay(NULL,fd);
1533     if (!c) return NULL;
1534     selectDb(c,0);
1535     c->fd = fd;
1536     c->querybuf = sdsempty();
1537     c->argc = 0;
1538     c->argv = NULL;
1539     c->bulklen = -1;
1540     c->sentlen = 0;
1541     c->flags = 0;
1542     c->lastinteraction = time(NULL);
1543     c->authenticated = 0;
1544     c->replstate = REDIS_REPL_NONE;
1545     if ((c->reply = listCreate()) == NULL) oom("listCreate");
1546     listSetFreeMethod(c->reply,decrRefCount);
1547     listSetDupMethod(c->reply,dupClientReplyValue);
1548     if (aeCreateFileEvent(server.el, c->fd, AE_READABLE,
1549         readQueryFromClient, c, NULL) == AE_ERR) {
1550         freeClient(c);
1551         return NULL;
1552     }
1553     if (!listAddNodeTail(server.clients,c)) oom("listAddNodeTail");
1554     return c;
1555 }
1556
1557 static void addReply(redisClient *c, robj *obj) {
1558     if (listLength(c->reply) == 0 &&
1559         (c->replstate == REDIS_REPL_NONE ||
1560          c->replstate == REDIS_REPL_ONLINE) &&
1561         aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
1562         sendReplyToClient, c, NULL) == AE_ERR) return;
1563     if (!listAddNodeTail(c->reply,obj)) oom("listAddNodeTail");
1564     incrRefCount(obj);
1565 }
1566
1567 static void addReplySds(redisClient *c, sds s) {
1568     robj *o = createObject(REDIS_STRING,s);
1569     addReply(c,o);
1570     decrRefCount(o);
1571 }
1572
1573 static void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
1574     int cport, cfd;
1575     char cip[128];
1576     redisClient *c;
1577     REDIS_NOTUSED(el);
1578     REDIS_NOTUSED(mask);
1579     REDIS_NOTUSED(privdata);
1580
1581     cfd = anetAccept(server.neterr, fd, cip, &cport);
1582     if (cfd == AE_ERR) {
1583         redisLog(REDIS_DEBUG,"Accepting client connection: %s", server.neterr);
1584         return;
1585     }
1586     redisLog(REDIS_DEBUG,"Accepted %s:%d", cip, cport);
1587     if ((c = createClient(cfd)) == NULL) {
1588         redisLog(REDIS_WARNING,"Error allocating resoures for the client");
1589         close(cfd); /* May be already closed, just ingore errors */
1590         return;
1591     }
1592     /* If maxclient directive is set and this is one client more... close the
1593      * connection. Note that we create the client instead to check before
1594      * for this condition, since now the socket is already set in nonblocking
1595      * mode and we can send an error for free using the Kernel I/O */
1596     if (server.maxclients && listLength(server.clients) > server.maxclients) {
1597         char *err = "-ERR max number of clients reached\r\n";
1598
1599         /* That's a best effort error message, don't check write errors */
1600         (void) write(c->fd,err,strlen(err));
1601         freeClient(c);
1602         return;
1603     }
1604     server.stat_numconnections++;
1605 }
1606
1607 /* ======================= Redis objects implementation ===================== */
1608
1609 static robj *createObject(int type, void *ptr) {
1610     robj *o;
1611
1612     if (listLength(server.objfreelist)) {
1613         listNode *head = listFirst(server.objfreelist);
1614         o = listNodeValue(head);
1615         listDelNode(server.objfreelist,head);
1616     } else {
1617         o = zmalloc(sizeof(*o));
1618     }
1619     if (!o) oom("createObject");
1620     o->type = type;
1621     o->ptr = ptr;
1622     o->refcount = 1;
1623     return o;
1624 }
1625
1626 static robj *createStringObject(char *ptr, size_t len) {
1627     return createObject(REDIS_STRING,sdsnewlen(ptr,len));
1628 }
1629
1630 static robj *createListObject(void) {
1631     list *l = listCreate();
1632
1633     if (!l) oom("listCreate");
1634     listSetFreeMethod(l,decrRefCount);
1635     return createObject(REDIS_LIST,l);
1636 }
1637
1638 static robj *createSetObject(void) {
1639     dict *d = dictCreate(&setDictType,NULL);
1640     if (!d) oom("dictCreate");
1641     return createObject(REDIS_SET,d);
1642 }
1643
1644 static void freeStringObject(robj *o) {
1645     sdsfree(o->ptr);
1646 }
1647
1648 static void freeListObject(robj *o) {
1649     listRelease((list*) o->ptr);
1650 }
1651
1652 static void freeSetObject(robj *o) {
1653     dictRelease((dict*) o->ptr);
1654 }
1655
1656 static void freeHashObject(robj *o) {
1657     dictRelease((dict*) o->ptr);
1658 }
1659
1660 static void incrRefCount(robj *o) {
1661     o->refcount++;
1662 #ifdef DEBUG_REFCOUNT
1663     if (o->type == REDIS_STRING)
1664         printf("Increment '%s'(%p), now is: %d\n",o->ptr,o,o->refcount);
1665 #endif
1666 }
1667
1668 static void decrRefCount(void *obj) {
1669     robj *o = obj;
1670
1671 #ifdef DEBUG_REFCOUNT
1672     if (o->type == REDIS_STRING)
1673         printf("Decrement '%s'(%p), now is: %d\n",o->ptr,o,o->refcount-1);
1674 #endif
1675     if (--(o->refcount) == 0) {
1676         switch(o->type) {
1677         case REDIS_STRING: freeStringObject(o); break;
1678         case REDIS_LIST: freeListObject(o); break;
1679         case REDIS_SET: freeSetObject(o); break;
1680         case REDIS_HASH: freeHashObject(o); break;
1681         default: assert(0 != 0); break;
1682         }
1683         if (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX ||
1684             !listAddNodeHead(server.objfreelist,o))
1685             zfree(o);
1686     }
1687 }
1688
1689 /* Try to share an object against the shared objects pool */
1690 static robj *tryObjectSharing(robj *o) {
1691     struct dictEntry *de;
1692     unsigned long c;
1693
1694     if (o == NULL || server.shareobjects == 0) return o;
1695
1696     assert(o->type == REDIS_STRING);
1697     de = dictFind(server.sharingpool,o);
1698     if (de) {
1699         robj *shared = dictGetEntryKey(de);
1700
1701         c = ((unsigned long) dictGetEntryVal(de))+1;
1702         dictGetEntryVal(de) = (void*) c;
1703         incrRefCount(shared);
1704         decrRefCount(o);
1705         return shared;
1706     } else {
1707         /* Here we are using a stream algorihtm: Every time an object is
1708          * shared we increment its count, everytime there is a miss we
1709          * recrement the counter of a random object. If this object reaches
1710          * zero we remove the object and put the current object instead. */
1711         if (dictSize(server.sharingpool) >=
1712                 server.sharingpoolsize) {
1713             de = dictGetRandomKey(server.sharingpool);
1714             assert(de != NULL);
1715             c = ((unsigned long) dictGetEntryVal(de))-1;
1716             dictGetEntryVal(de) = (void*) c;
1717             if (c == 0) {
1718                 dictDelete(server.sharingpool,de->key);
1719             }
1720         } else {
1721             c = 0; /* If the pool is empty we want to add this object */
1722         }
1723         if (c == 0) {
1724             int retval;
1725
1726             retval = dictAdd(server.sharingpool,o,(void*)1);
1727             assert(retval == DICT_OK);
1728             incrRefCount(o);
1729         }
1730         return o;
1731     }
1732 }
1733
1734 static robj *lookupKey(redisDb *db, robj *key) {
1735     dictEntry *de = dictFind(db->dict,key);
1736     return de ? dictGetEntryVal(de) : NULL;
1737 }
1738
1739 static robj *lookupKeyRead(redisDb *db, robj *key) {
1740     expireIfNeeded(db,key);
1741     return lookupKey(db,key);
1742 }
1743
1744 static robj *lookupKeyWrite(redisDb *db, robj *key) {
1745     deleteIfVolatile(db,key);
1746     return lookupKey(db,key);
1747 }
1748
1749 static int deleteKey(redisDb *db, robj *key) {
1750     int retval;
1751
1752     /* We need to protect key from destruction: after the first dictDelete()
1753      * it may happen that 'key' is no longer valid if we don't increment
1754      * it's count. This may happen when we get the object reference directly
1755      * from the hash table with dictRandomKey() or dict iterators */
1756     incrRefCount(key);
1757     if (dictSize(db->expires)) dictDelete(db->expires,key);
1758     retval = dictDelete(db->dict,key);
1759     decrRefCount(key);
1760
1761     return retval == DICT_OK;
1762 }
1763
1764 /*============================ DB saving/loading ============================ */
1765
1766 static int rdbSaveType(FILE *fp, unsigned char type) {
1767     if (fwrite(&type,1,1,fp) == 0) return -1;
1768     return 0;
1769 }
1770
1771 static int rdbSaveTime(FILE *fp, time_t t) {
1772     int32_t t32 = (int32_t) t;
1773     if (fwrite(&t32,4,1,fp) == 0) return -1;
1774     return 0;
1775 }
1776
1777 /* check rdbLoadLen() comments for more info */
1778 static int rdbSaveLen(FILE *fp, uint32_t len) {
1779     unsigned char buf[2];
1780
1781     if (len < (1<<6)) {
1782         /* Save a 6 bit len */
1783         buf[0] = (len&0xFF)|(REDIS_RDB_6BITLEN<<6);
1784         if (fwrite(buf,1,1,fp) == 0) return -1;
1785     } else if (len < (1<<14)) {
1786         /* Save a 14 bit len */
1787         buf[0] = ((len>>8)&0xFF)|(REDIS_RDB_14BITLEN<<6);
1788         buf[1] = len&0xFF;
1789         if (fwrite(buf,2,1,fp) == 0) return -1;
1790     } else {
1791         /* Save a 32 bit len */
1792         buf[0] = (REDIS_RDB_32BITLEN<<6);
1793         if (fwrite(buf,1,1,fp) == 0) return -1;
1794         len = htonl(len);
1795         if (fwrite(&len,4,1,fp) == 0) return -1;
1796     }
1797     return 0;
1798 }
1799
1800 /* String objects in the form "2391" "-100" without any space and with a
1801  * range of values that can fit in an 8, 16 or 32 bit signed value can be
1802  * encoded as integers to save space */
1803 static int rdbTryIntegerEncoding(sds s, unsigned char *enc) {
1804     long long value;
1805     char *endptr, buf[32];
1806
1807     /* Check if it's possible to encode this value as a number */
1808     value = strtoll(s, &endptr, 10);
1809     if (endptr[0] != '\0') return 0;
1810     snprintf(buf,32,"%lld",value);
1811
1812     /* If the number converted back into a string is not identical
1813      * then it's not possible to encode the string as integer */
1814     if (strlen(buf) != sdslen(s) || memcmp(buf,s,sdslen(s))) return 0;
1815
1816     /* Finally check if it fits in our ranges */
1817     if (value >= -(1<<7) && value <= (1<<7)-1) {
1818         enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT8;
1819         enc[1] = value&0xFF;
1820         return 2;
1821     } else if (value >= -(1<<15) && value <= (1<<15)-1) {
1822         enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT16;
1823         enc[1] = value&0xFF;
1824         enc[2] = (value>>8)&0xFF;
1825         return 3;
1826     } else if (value >= -((long long)1<<31) && value <= ((long long)1<<31)-1) {
1827         enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT32;
1828         enc[1] = value&0xFF;
1829         enc[2] = (value>>8)&0xFF;
1830         enc[3] = (value>>16)&0xFF;
1831         enc[4] = (value>>24)&0xFF;
1832         return 5;
1833     } else {
1834         return 0;
1835     }
1836 }
1837
1838 static int rdbSaveLzfStringObject(FILE *fp, robj *obj) {
1839     unsigned int comprlen, outlen;
1840     unsigned char byte;
1841     void *out;
1842
1843     /* We require at least four bytes compression for this to be worth it */
1844     outlen = sdslen(obj->ptr)-4;
1845     if (outlen <= 0) return 0;
1846     if ((out = zmalloc(outlen+1)) == NULL) return 0;
1847     comprlen = lzf_compress(obj->ptr, sdslen(obj->ptr), out, outlen);
1848     if (comprlen == 0) {
1849         zfree(out);
1850         return 0;
1851     }
1852     /* Data compressed! Let's save it on disk */
1853     byte = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_LZF;
1854     if (fwrite(&byte,1,1,fp) == 0) goto writeerr;
1855     if (rdbSaveLen(fp,comprlen) == -1) goto writeerr;
1856     if (rdbSaveLen(fp,sdslen(obj->ptr)) == -1) goto writeerr;
1857     if (fwrite(out,comprlen,1,fp) == 0) goto writeerr;
1858     zfree(out);
1859     return comprlen;
1860
1861 writeerr:
1862     zfree(out);
1863     return -1;
1864 }
1865
1866 /* Save a string objet as [len][data] on disk. If the object is a string
1867  * representation of an integer value we try to safe it in a special form */
1868 static int rdbSaveStringObject(FILE *fp, robj *obj) {
1869     size_t len = sdslen(obj->ptr);
1870     int enclen;
1871
1872     /* Try integer encoding */
1873     if (len <= 11) {
1874         unsigned char buf[5];
1875         if ((enclen = rdbTryIntegerEncoding(obj->ptr,buf)) > 0) {
1876             if (fwrite(buf,enclen,1,fp) == 0) return -1;
1877             return 0;
1878         }
1879     }
1880
1881     /* Try LZF compression - under 20 bytes it's unable to compress even
1882      * aaaaaaaaaaaaaaaaaa so skip it */
1883     if (1 && len > 20) {
1884         int retval;
1885
1886         retval = rdbSaveLzfStringObject(fp,obj);
1887         if (retval == -1) return -1;
1888         if (retval > 0) return 0;
1889         /* retval == 0 means data can't be compressed, save the old way */
1890     }
1891
1892     /* Store verbatim */
1893     if (rdbSaveLen(fp,len) == -1) return -1;
1894     if (len && fwrite(obj->ptr,len,1,fp) == 0) return -1;
1895     return 0;
1896 }
1897
1898 /* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success */
1899 static int rdbSave(char *filename) {
1900     dictIterator *di = NULL;
1901     dictEntry *de;
1902     FILE *fp;
1903     char tmpfile[256];
1904     int j;
1905     time_t now = time(NULL);
1906
1907     snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
1908     fp = fopen(tmpfile,"w");
1909     if (!fp) {
1910         redisLog(REDIS_WARNING, "Failed saving the DB: %s", strerror(errno));
1911         return REDIS_ERR;
1912     }
1913     if (fwrite("REDIS0001",9,1,fp) == 0) goto werr;
1914     for (j = 0; j < server.dbnum; j++) {
1915         redisDb *db = server.db+j;
1916         dict *d = db->dict;
1917         if (dictSize(d) == 0) continue;
1918         di = dictGetIterator(d);
1919         if (!di) {
1920             fclose(fp);
1921             return REDIS_ERR;
1922         }
1923
1924         /* Write the SELECT DB opcode */
1925         if (rdbSaveType(fp,REDIS_SELECTDB) == -1) goto werr;
1926         if (rdbSaveLen(fp,j) == -1) goto werr;
1927
1928         /* Iterate this DB writing every entry */
1929         while((de = dictNext(di)) != NULL) {
1930             robj *key = dictGetEntryKey(de);
1931             robj *o = dictGetEntryVal(de);
1932             time_t expiretime = getExpire(db,key);
1933
1934             /* Save the expire time */
1935             if (expiretime != -1) {
1936                 /* If this key is already expired skip it */
1937                 if (expiretime < now) continue;
1938                 if (rdbSaveType(fp,REDIS_EXPIRETIME) == -1) goto werr;
1939                 if (rdbSaveTime(fp,expiretime) == -1) goto werr;
1940             }
1941             /* Save the key and associated value */
1942             if (rdbSaveType(fp,o->type) == -1) goto werr;
1943             if (rdbSaveStringObject(fp,key) == -1) goto werr;
1944             if (o->type == REDIS_STRING) {
1945                 /* Save a string value */
1946                 if (rdbSaveStringObject(fp,o) == -1) goto werr;
1947             } else if (o->type == REDIS_LIST) {
1948                 /* Save a list value */
1949                 list *list = o->ptr;
1950                 listNode *ln;
1951
1952                 listRewind(list);
1953                 if (rdbSaveLen(fp,listLength(list)) == -1) goto werr;
1954                 while((ln = listYield(list))) {
1955                     robj *eleobj = listNodeValue(ln);
1956
1957                     if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
1958                 }
1959             } else if (o->type == REDIS_SET) {
1960                 /* Save a set value */
1961                 dict *set = o->ptr;
1962                 dictIterator *di = dictGetIterator(set);
1963                 dictEntry *de;
1964
1965                 if (!set) oom("dictGetIteraotr");
1966                 if (rdbSaveLen(fp,dictSize(set)) == -1) goto werr;
1967                 while((de = dictNext(di)) != NULL) {
1968                     robj *eleobj = dictGetEntryKey(de);
1969
1970                     if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
1971                 }
1972                 dictReleaseIterator(di);
1973             } else {
1974                 assert(0 != 0);
1975             }
1976         }
1977         dictReleaseIterator(di);
1978     }
1979     /* EOF opcode */
1980     if (rdbSaveType(fp,REDIS_EOF) == -1) goto werr;
1981
1982     /* Make sure data will not remain on the OS's output buffers */
1983     fflush(fp);
1984     fsync(fileno(fp));
1985     fclose(fp);
1986
1987     /* Use RENAME to make sure the DB file is changed atomically only
1988      * if the generate DB file is ok. */
1989     if (rename(tmpfile,filename) == -1) {
1990         redisLog(REDIS_WARNING,"Error moving temp DB file on the final destionation: %s", strerror(errno));
1991         unlink(tmpfile);
1992         return REDIS_ERR;
1993     }
1994     redisLog(REDIS_NOTICE,"DB saved on disk");
1995     server.dirty = 0;
1996     server.lastsave = time(NULL);
1997     return REDIS_OK;
1998
1999 werr:
2000     fclose(fp);
2001     unlink(tmpfile);
2002     redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
2003     if (di) dictReleaseIterator(di);
2004     return REDIS_ERR;
2005 }
2006
2007 static int rdbSaveBackground(char *filename) {
2008     pid_t childpid;
2009
2010     if (server.bgsaveinprogress) return REDIS_ERR;
2011     if ((childpid = fork()) == 0) {
2012         /* Child */
2013         close(server.fd);
2014         if (rdbSave(filename) == REDIS_OK) {
2015             exit(0);
2016         } else {
2017             exit(1);
2018         }
2019     } else {
2020         /* Parent */
2021         if (childpid == -1) {
2022             redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
2023                 strerror(errno));
2024             return REDIS_ERR;
2025         }
2026         redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
2027         server.bgsaveinprogress = 1;
2028         server.bgsavechildpid = childpid;
2029         return REDIS_OK;
2030     }
2031     return REDIS_OK; /* unreached */
2032 }
2033
2034 static void rdbRemoveTempFile(pid_t childpid) {
2035     char tmpfile[256];
2036
2037     snprintf(tmpfile,256,"temp-%d.rdb", (int) childpid);
2038     unlink(tmpfile);
2039 }
2040
2041 static int rdbLoadType(FILE *fp) {
2042     unsigned char type;
2043     if (fread(&type,1,1,fp) == 0) return -1;
2044     return type;
2045 }
2046
2047 static time_t rdbLoadTime(FILE *fp) {
2048     int32_t t32;
2049     if (fread(&t32,4,1,fp) == 0) return -1;
2050     return (time_t) t32;
2051 }
2052
2053 /* Load an encoded length from the DB, see the REDIS_RDB_* defines on the top
2054  * of this file for a description of how this are stored on disk.
2055  *
2056  * isencoded is set to 1 if the readed length is not actually a length but
2057  * an "encoding type", check the above comments for more info */
2058 static uint32_t rdbLoadLen(FILE *fp, int rdbver, int *isencoded) {
2059     unsigned char buf[2];
2060     uint32_t len;
2061
2062     if (isencoded) *isencoded = 0;
2063     if (rdbver == 0) {
2064         if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
2065         return ntohl(len);
2066     } else {
2067         int type;
2068
2069         if (fread(buf,1,1,fp) == 0) return REDIS_RDB_LENERR;
2070         type = (buf[0]&0xC0)>>6;
2071         if (type == REDIS_RDB_6BITLEN) {
2072             /* Read a 6 bit len */
2073             return buf[0]&0x3F;
2074         } else if (type == REDIS_RDB_ENCVAL) {
2075             /* Read a 6 bit len encoding type */
2076             if (isencoded) *isencoded = 1;
2077             return buf[0]&0x3F;
2078         } else if (type == REDIS_RDB_14BITLEN) {
2079             /* Read a 14 bit len */
2080             if (fread(buf+1,1,1,fp) == 0) return REDIS_RDB_LENERR;
2081             return ((buf[0]&0x3F)<<8)|buf[1];
2082         } else {
2083             /* Read a 32 bit len */
2084             if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
2085             return ntohl(len);
2086         }
2087     }
2088 }
2089
2090 static robj *rdbLoadIntegerObject(FILE *fp, int enctype) {
2091     unsigned char enc[4];
2092     long long val;
2093
2094     if (enctype == REDIS_RDB_ENC_INT8) {
2095         if (fread(enc,1,1,fp) == 0) return NULL;
2096         val = (signed char)enc[0];
2097     } else if (enctype == REDIS_RDB_ENC_INT16) {
2098         uint16_t v;
2099         if (fread(enc,2,1,fp) == 0) return NULL;
2100         v = enc[0]|(enc[1]<<8);
2101         val = (int16_t)v;
2102     } else if (enctype == REDIS_RDB_ENC_INT32) {
2103         uint32_t v;
2104         if (fread(enc,4,1,fp) == 0) return NULL;
2105         v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24);
2106         val = (int32_t)v;
2107     } else {
2108         val = 0; /* anti-warning */
2109         assert(0!=0);
2110     }
2111     return createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",val));
2112 }
2113
2114 static robj *rdbLoadLzfStringObject(FILE*fp, int rdbver) {
2115     unsigned int len, clen;
2116     unsigned char *c = NULL;
2117     sds val = NULL;
2118
2119     if ((clen = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR) return NULL;
2120     if ((len = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR) return NULL;
2121     if ((c = zmalloc(clen)) == NULL) goto err;
2122     if ((val = sdsnewlen(NULL,len)) == NULL) goto err;
2123     if (fread(c,clen,1,fp) == 0) goto err;
2124     if (lzf_decompress(c,clen,val,len) == 0) goto err;
2125     zfree(c);
2126     return createObject(REDIS_STRING,val);
2127 err:
2128     zfree(c);
2129     sdsfree(val);
2130     return NULL;
2131 }
2132
2133 static robj *rdbLoadStringObject(FILE*fp, int rdbver) {
2134     int isencoded;
2135     uint32_t len;
2136     sds val;
2137
2138     len = rdbLoadLen(fp,rdbver,&isencoded);
2139     if (isencoded) {
2140         switch(len) {
2141         case REDIS_RDB_ENC_INT8:
2142         case REDIS_RDB_ENC_INT16:
2143         case REDIS_RDB_ENC_INT32:
2144             return tryObjectSharing(rdbLoadIntegerObject(fp,len));
2145         case REDIS_RDB_ENC_LZF:
2146             return tryObjectSharing(rdbLoadLzfStringObject(fp,rdbver));
2147         default:
2148             assert(0!=0);
2149         }
2150     }
2151
2152     if (len == REDIS_RDB_LENERR) return NULL;
2153     val = sdsnewlen(NULL,len);
2154     if (len && fread(val,len,1,fp) == 0) {
2155         sdsfree(val);
2156         return NULL;
2157     }
2158     return tryObjectSharing(createObject(REDIS_STRING,val));
2159 }
2160
2161 static int rdbLoad(char *filename) {
2162     FILE *fp;
2163     robj *keyobj = NULL;
2164     uint32_t dbid;
2165     int type, retval, rdbver;
2166     dict *d = server.db[0].dict;
2167     redisDb *db = server.db+0;
2168     char buf[1024];
2169     time_t expiretime = -1, now = time(NULL);
2170
2171     fp = fopen(filename,"r");
2172     if (!fp) return REDIS_ERR;
2173     if (fread(buf,9,1,fp) == 0) goto eoferr;
2174     buf[9] = '\0';
2175     if (memcmp(buf,"REDIS",5) != 0) {
2176         fclose(fp);
2177         redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
2178         return REDIS_ERR;
2179     }
2180     rdbver = atoi(buf+5);
2181     if (rdbver > 1) {
2182         fclose(fp);
2183         redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
2184         return REDIS_ERR;
2185     }
2186     while(1) {
2187         robj *o;
2188
2189         /* Read type. */
2190         if ((type = rdbLoadType(fp)) == -1) goto eoferr;
2191         if (type == REDIS_EXPIRETIME) {
2192             if ((expiretime = rdbLoadTime(fp)) == -1) goto eoferr;
2193             /* We read the time so we need to read the object type again */
2194             if ((type = rdbLoadType(fp)) == -1) goto eoferr;
2195         }
2196         if (type == REDIS_EOF) break;
2197         /* Handle SELECT DB opcode as a special case */
2198         if (type == REDIS_SELECTDB) {
2199             if ((dbid = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR)
2200                 goto eoferr;
2201             if (dbid >= (unsigned)server.dbnum) {
2202                 redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
2203                 exit(1);
2204             }
2205             db = server.db+dbid;
2206             d = db->dict;
2207             continue;
2208         }
2209         /* Read key */
2210         if ((keyobj = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
2211
2212         if (type == REDIS_STRING) {
2213             /* Read string value */
2214             if ((o = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
2215         } else if (type == REDIS_LIST || type == REDIS_SET) {
2216             /* Read list/set value */
2217             uint32_t listlen;
2218
2219             if ((listlen = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR)
2220                 goto eoferr;
2221             o = (type == REDIS_LIST) ? createListObject() : createSetObject();
2222             /* Load every single element of the list/set */
2223             while(listlen--) {
2224                 robj *ele;
2225
2226                 if ((ele = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
2227                 if (type == REDIS_LIST) {
2228                     if (!listAddNodeTail((list*)o->ptr,ele))
2229                         oom("listAddNodeTail");
2230                 } else {
2231                     if (dictAdd((dict*)o->ptr,ele,NULL) == DICT_ERR)
2232                         oom("dictAdd");
2233                 }
2234             }
2235         } else {
2236             assert(0 != 0);
2237         }
2238         /* Add the new object in the hash table */
2239         retval = dictAdd(d,keyobj,o);
2240         if (retval == DICT_ERR) {
2241             redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", keyobj->ptr);
2242             exit(1);
2243         }
2244         /* Set the expire time if needed */
2245         if (expiretime != -1) {
2246             setExpire(db,keyobj,expiretime);
2247             /* Delete this key if already expired */
2248             if (expiretime < now) deleteKey(db,keyobj);
2249             expiretime = -1;
2250         }
2251         keyobj = o = NULL;
2252     }
2253     fclose(fp);
2254     return REDIS_OK;
2255
2256 eoferr: /* unexpected end of file is handled here with a fatal exit */
2257     if (keyobj) decrRefCount(keyobj);
2258     redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, exiting now.");
2259     exit(1);
2260     return REDIS_ERR; /* Just to avoid warning */
2261 }
2262
2263 /*================================== Commands =============================== */
2264
2265 static void authCommand(redisClient *c) {
2266     if (!server.requirepass || !strcmp(c->argv[1]->ptr, server.requirepass)) {
2267       c->authenticated = 1;
2268       addReply(c,shared.ok);
2269     } else {
2270       c->authenticated = 0;
2271       addReply(c,shared.err);
2272     }
2273 }
2274
2275 static void pingCommand(redisClient *c) {
2276     addReply(c,shared.pong);
2277 }
2278
2279 static void echoCommand(redisClient *c) {
2280     addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
2281         (int)sdslen(c->argv[1]->ptr)));
2282     addReply(c,c->argv[1]);
2283     addReply(c,shared.crlf);
2284 }
2285
2286 /*=================================== Strings =============================== */
2287
2288 static void setGenericCommand(redisClient *c, int nx) {
2289     int retval;
2290
2291     retval = dictAdd(c->db->dict,c->argv[1],c->argv[2]);
2292     if (retval == DICT_ERR) {
2293         if (!nx) {
2294             dictReplace(c->db->dict,c->argv[1],c->argv[2]);
2295             incrRefCount(c->argv[2]);
2296         } else {
2297             addReply(c,shared.czero);
2298             return;
2299         }
2300     } else {
2301         incrRefCount(c->argv[1]);
2302         incrRefCount(c->argv[2]);
2303     }
2304     server.dirty++;
2305     removeExpire(c->db,c->argv[1]);
2306     addReply(c, nx ? shared.cone : shared.ok);
2307 }
2308
2309 static void setCommand(redisClient *c) {
2310     setGenericCommand(c,0);
2311 }
2312
2313 static void setnxCommand(redisClient *c) {
2314     setGenericCommand(c,1);
2315 }
2316
2317 static void getCommand(redisClient *c) {
2318     robj *o = lookupKeyRead(c->db,c->argv[1]);
2319
2320     if (o == NULL) {
2321         addReply(c,shared.nullbulk);
2322     } else {
2323         if (o->type != REDIS_STRING) {
2324             addReply(c,shared.wrongtypeerr);
2325         } else {
2326             addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(o->ptr)));
2327             addReply(c,o);
2328             addReply(c,shared.crlf);
2329         }
2330     }
2331 }
2332
2333 static void getSetCommand(redisClient *c) {
2334     getCommand(c);
2335     if (dictAdd(c->db->dict,c->argv[1],c->argv[2]) == DICT_ERR) {
2336         dictReplace(c->db->dict,c->argv[1],c->argv[2]);
2337     } else {
2338         incrRefCount(c->argv[1]);
2339     }
2340     incrRefCount(c->argv[2]);
2341     server.dirty++;
2342     removeExpire(c->db,c->argv[1]);
2343 }
2344
2345 static void mgetCommand(redisClient *c) {
2346     int j;
2347
2348     addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",c->argc-1));
2349     for (j = 1; j < c->argc; j++) {
2350         robj *o = lookupKeyRead(c->db,c->argv[j]);
2351         if (o == NULL) {
2352             addReply(c,shared.nullbulk);
2353         } else {
2354             if (o->type != REDIS_STRING) {
2355                 addReply(c,shared.nullbulk);
2356             } else {
2357                 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(o->ptr)));
2358                 addReply(c,o);
2359                 addReply(c,shared.crlf);
2360             }
2361         }
2362     }
2363 }
2364
2365 static void incrDecrCommand(redisClient *c, long long incr) {
2366     long long value;
2367     int retval;
2368     robj *o;
2369
2370     o = lookupKeyWrite(c->db,c->argv[1]);
2371     if (o == NULL) {
2372         value = 0;
2373     } else {
2374         if (o->type != REDIS_STRING) {
2375             value = 0;
2376         } else {
2377             char *eptr;
2378
2379             value = strtoll(o->ptr, &eptr, 10);
2380         }
2381     }
2382
2383     value += incr;
2384     o = createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",value));
2385     retval = dictAdd(c->db->dict,c->argv[1],o);
2386     if (retval == DICT_ERR) {
2387         dictReplace(c->db->dict,c->argv[1],o);
2388         removeExpire(c->db,c->argv[1]);
2389     } else {
2390         incrRefCount(c->argv[1]);
2391     }
2392     server.dirty++;
2393     addReply(c,shared.colon);
2394     addReply(c,o);
2395     addReply(c,shared.crlf);
2396 }
2397
2398 static void incrCommand(redisClient *c) {
2399     incrDecrCommand(c,1);
2400 }
2401
2402 static void decrCommand(redisClient *c) {
2403     incrDecrCommand(c,-1);
2404 }
2405
2406 static void incrbyCommand(redisClient *c) {
2407     long long incr = strtoll(c->argv[2]->ptr, NULL, 10);
2408     incrDecrCommand(c,incr);
2409 }
2410
2411 static void decrbyCommand(redisClient *c) {
2412     long long incr = strtoll(c->argv[2]->ptr, NULL, 10);
2413     incrDecrCommand(c,-incr);
2414 }
2415
2416 /* ========================= Type agnostic commands ========================= */
2417
2418 static void delCommand(redisClient *c) {
2419     int deleted = 0, j;
2420
2421     for (j = 1; j < c->argc; j++) {
2422         if (deleteKey(c->db,c->argv[j])) {
2423             server.dirty++;
2424             deleted++;
2425         }
2426     }
2427     switch(deleted) {
2428     case 0:
2429         addReply(c,shared.czero);
2430         break;
2431     case 1:
2432         addReply(c,shared.cone);
2433         break;
2434     default:
2435         addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",deleted));
2436         break;
2437     }
2438 }
2439
2440 static void existsCommand(redisClient *c) {
2441     addReply(c,lookupKeyRead(c->db,c->argv[1]) ? shared.cone : shared.czero);
2442 }
2443
2444 static void selectCommand(redisClient *c) {
2445     int id = atoi(c->argv[1]->ptr);
2446
2447     if (selectDb(c,id) == REDIS_ERR) {
2448         addReplySds(c,sdsnew("-ERR invalid DB index\r\n"));
2449     } else {
2450         addReply(c,shared.ok);
2451     }
2452 }
2453
2454 static void randomkeyCommand(redisClient *c) {
2455     dictEntry *de;
2456
2457     while(1) {
2458         de = dictGetRandomKey(c->db->dict);
2459         if (!de || expireIfNeeded(c->db,dictGetEntryKey(de)) == 0) break;
2460     }
2461     if (de == NULL) {
2462         addReply(c,shared.plus);
2463         addReply(c,shared.crlf);
2464     } else {
2465         addReply(c,shared.plus);
2466         addReply(c,dictGetEntryKey(de));
2467         addReply(c,shared.crlf);
2468     }
2469 }
2470
2471 static void keysCommand(redisClient *c) {
2472     dictIterator *di;
2473     dictEntry *de;
2474     sds pattern = c->argv[1]->ptr;
2475     int plen = sdslen(pattern);
2476     int numkeys = 0, keyslen = 0;
2477     robj *lenobj = createObject(REDIS_STRING,NULL);
2478
2479     di = dictGetIterator(c->db->dict);
2480     if (!di) oom("dictGetIterator");
2481     addReply(c,lenobj);
2482     decrRefCount(lenobj);
2483     while((de = dictNext(di)) != NULL) {
2484         robj *keyobj = dictGetEntryKey(de);
2485
2486         sds key = keyobj->ptr;
2487         if ((pattern[0] == '*' && pattern[1] == '\0') ||
2488             stringmatchlen(pattern,plen,key,sdslen(key),0)) {
2489             if (expireIfNeeded(c->db,keyobj) == 0) {
2490                 if (numkeys != 0)
2491                     addReply(c,shared.space);
2492                 addReply(c,keyobj);
2493                 numkeys++;
2494                 keyslen += sdslen(key);
2495             }
2496         }
2497     }
2498     dictReleaseIterator(di);
2499     lenobj->ptr = sdscatprintf(sdsempty(),"$%lu\r\n",keyslen+(numkeys ? (numkeys-1) : 0));
2500     addReply(c,shared.crlf);
2501 }
2502
2503 static void dbsizeCommand(redisClient *c) {
2504     addReplySds(c,
2505         sdscatprintf(sdsempty(),":%lu\r\n",dictSize(c->db->dict)));
2506 }
2507
2508 static void lastsaveCommand(redisClient *c) {
2509     addReplySds(c,
2510         sdscatprintf(sdsempty(),":%lu\r\n",server.lastsave));
2511 }
2512
2513 static void typeCommand(redisClient *c) {
2514     robj *o;
2515     char *type;
2516
2517     o = lookupKeyRead(c->db,c->argv[1]);
2518     if (o == NULL) {
2519         type = "+none";
2520     } else {
2521         switch(o->type) {
2522         case REDIS_STRING: type = "+string"; break;
2523         case REDIS_LIST: type = "+list"; break;
2524         case REDIS_SET: type = "+set"; break;
2525         default: type = "unknown"; break;
2526         }
2527     }
2528     addReplySds(c,sdsnew(type));
2529     addReply(c,shared.crlf);
2530 }
2531
2532 static void saveCommand(redisClient *c) {
2533     if (server.bgsaveinprogress) {
2534         addReplySds(c,sdsnew("-ERR background save in progress\r\n"));
2535         return;
2536     }
2537     if (rdbSave(server.dbfilename) == REDIS_OK) {
2538         addReply(c,shared.ok);
2539     } else {
2540         addReply(c,shared.err);
2541     }
2542 }
2543
2544 static void bgsaveCommand(redisClient *c) {
2545     if (server.bgsaveinprogress) {
2546         addReplySds(c,sdsnew("-ERR background save already in progress\r\n"));
2547         return;
2548     }
2549     if (rdbSaveBackground(server.dbfilename) == REDIS_OK) {
2550         addReply(c,shared.ok);
2551     } else {
2552         addReply(c,shared.err);
2553     }
2554 }
2555
2556 static void shutdownCommand(redisClient *c) {
2557     redisLog(REDIS_WARNING,"User requested shutdown, saving DB...");
2558     /* Kill the saving child if there is a background saving in progress.
2559        We want to avoid race conditions, for instance our saving child may
2560        overwrite the synchronous saving did by SHUTDOWN. */
2561     if (server.bgsaveinprogress) {
2562         redisLog(REDIS_WARNING,"There is a live saving child. Killing it!");
2563         kill(server.bgsavechildpid,SIGKILL);
2564         rdbRemoveTempFile(server.bgsavechildpid);
2565     }
2566     /* SYNC SAVE */
2567     if (rdbSave(server.dbfilename) == REDIS_OK) {
2568         if (server.daemonize)
2569             unlink(server.pidfile);
2570         redisLog(REDIS_WARNING,"%zu bytes used at exit",zmalloc_used_memory());
2571         redisLog(REDIS_WARNING,"Server exit now, bye bye...");
2572         exit(1);
2573     } else {
2574         /* Ooops.. error saving! The best we can do is to continue operating.
2575          * Note that if there was a background saving process, in the next
2576          * cron() Redis will be notified that the background saving aborted,
2577          * handling special stuff like slaves pending for synchronization... */
2578         redisLog(REDIS_WARNING,"Error trying to save the DB, can't exit");
2579         addReplySds(c,sdsnew("-ERR can't quit, problems saving the DB\r\n"));
2580     }
2581 }
2582
2583 static void renameGenericCommand(redisClient *c, int nx) {
2584     robj *o;
2585
2586     /* To use the same key as src and dst is probably an error */
2587     if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) {
2588         addReply(c,shared.sameobjecterr);
2589         return;
2590     }
2591
2592     o = lookupKeyWrite(c->db,c->argv[1]);
2593     if (o == NULL) {
2594         addReply(c,shared.nokeyerr);
2595         return;
2596     }
2597     incrRefCount(o);
2598     deleteIfVolatile(c->db,c->argv[2]);
2599     if (dictAdd(c->db->dict,c->argv[2],o) == DICT_ERR) {
2600         if (nx) {
2601             decrRefCount(o);
2602             addReply(c,shared.czero);
2603             return;
2604         }
2605         dictReplace(c->db->dict,c->argv[2],o);
2606     } else {
2607         incrRefCount(c->argv[2]);
2608     }
2609     deleteKey(c->db,c->argv[1]);
2610     server.dirty++;
2611     addReply(c,nx ? shared.cone : shared.ok);
2612 }
2613
2614 static void renameCommand(redisClient *c) {
2615     renameGenericCommand(c,0);
2616 }
2617
2618 static void renamenxCommand(redisClient *c) {
2619     renameGenericCommand(c,1);
2620 }
2621
2622 static void moveCommand(redisClient *c) {
2623     robj *o;
2624     redisDb *src, *dst;
2625     int srcid;
2626
2627     /* Obtain source and target DB pointers */
2628     src = c->db;
2629     srcid = c->db->id;
2630     if (selectDb(c,atoi(c->argv[2]->ptr)) == REDIS_ERR) {
2631         addReply(c,shared.outofrangeerr);
2632         return;
2633     }
2634     dst = c->db;
2635     selectDb(c,srcid); /* Back to the source DB */
2636
2637     /* If the user is moving using as target the same
2638      * DB as the source DB it is probably an error. */
2639     if (src == dst) {
2640         addReply(c,shared.sameobjecterr);
2641         return;
2642     }
2643
2644     /* Check if the element exists and get a reference */
2645     o = lookupKeyWrite(c->db,c->argv[1]);
2646     if (!o) {
2647         addReply(c,shared.czero);
2648         return;
2649     }
2650
2651     /* Try to add the element to the target DB */
2652     deleteIfVolatile(dst,c->argv[1]);
2653     if (dictAdd(dst->dict,c->argv[1],o) == DICT_ERR) {
2654         addReply(c,shared.czero);
2655         return;
2656     }
2657     incrRefCount(c->argv[1]);
2658     incrRefCount(o);
2659
2660     /* OK! key moved, free the entry in the source DB */
2661     deleteKey(src,c->argv[1]);
2662     server.dirty++;
2663     addReply(c,shared.cone);
2664 }
2665
2666 /* =================================== Lists ================================ */
2667 static void pushGenericCommand(redisClient *c, int where) {
2668     robj *lobj;
2669     list *list;
2670
2671     lobj = lookupKeyWrite(c->db,c->argv[1]);
2672     if (lobj == NULL) {
2673         lobj = createListObject();
2674         list = lobj->ptr;
2675         if (where == REDIS_HEAD) {
2676             if (!listAddNodeHead(list,c->argv[2])) oom("listAddNodeHead");
2677         } else {
2678             if (!listAddNodeTail(list,c->argv[2])) oom("listAddNodeTail");
2679         }
2680         dictAdd(c->db->dict,c->argv[1],lobj);
2681         incrRefCount(c->argv[1]);
2682         incrRefCount(c->argv[2]);
2683     } else {
2684         if (lobj->type != REDIS_LIST) {
2685             addReply(c,shared.wrongtypeerr);
2686             return;
2687         }
2688         list = lobj->ptr;
2689         if (where == REDIS_HEAD) {
2690             if (!listAddNodeHead(list,c->argv[2])) oom("listAddNodeHead");
2691         } else {
2692             if (!listAddNodeTail(list,c->argv[2])) oom("listAddNodeTail");
2693         }
2694         incrRefCount(c->argv[2]);
2695     }
2696     server.dirty++;
2697     addReply(c,shared.ok);
2698 }
2699
2700 static void lpushCommand(redisClient *c) {
2701     pushGenericCommand(c,REDIS_HEAD);
2702 }
2703
2704 static void rpushCommand(redisClient *c) {
2705     pushGenericCommand(c,REDIS_TAIL);
2706 }
2707
2708 static void llenCommand(redisClient *c) {
2709     robj *o;
2710     list *l;
2711
2712     o = lookupKeyRead(c->db,c->argv[1]);
2713     if (o == NULL) {
2714         addReply(c,shared.czero);
2715         return;
2716     } else {
2717         if (o->type != REDIS_LIST) {
2718             addReply(c,shared.wrongtypeerr);
2719         } else {
2720             l = o->ptr;
2721             addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",listLength(l)));
2722         }
2723     }
2724 }
2725
2726 static void lindexCommand(redisClient *c) {
2727     robj *o;
2728     int index = atoi(c->argv[2]->ptr);
2729
2730     o = lookupKeyRead(c->db,c->argv[1]);
2731     if (o == NULL) {
2732         addReply(c,shared.nullbulk);
2733     } else {
2734         if (o->type != REDIS_LIST) {
2735             addReply(c,shared.wrongtypeerr);
2736         } else {
2737             list *list = o->ptr;
2738             listNode *ln;
2739
2740             ln = listIndex(list, index);
2741             if (ln == NULL) {
2742                 addReply(c,shared.nullbulk);
2743             } else {
2744                 robj *ele = listNodeValue(ln);
2745                 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
2746                 addReply(c,ele);
2747                 addReply(c,shared.crlf);
2748             }
2749         }
2750     }
2751 }
2752
2753 static void lsetCommand(redisClient *c) {
2754     robj *o;
2755     int index = atoi(c->argv[2]->ptr);
2756
2757     o = lookupKeyWrite(c->db,c->argv[1]);
2758     if (o == NULL) {
2759         addReply(c,shared.nokeyerr);
2760     } else {
2761         if (o->type != REDIS_LIST) {
2762             addReply(c,shared.wrongtypeerr);
2763         } else {
2764             list *list = o->ptr;
2765             listNode *ln;
2766
2767             ln = listIndex(list, index);
2768             if (ln == NULL) {
2769                 addReply(c,shared.outofrangeerr);
2770             } else {
2771                 robj *ele = listNodeValue(ln);
2772
2773                 decrRefCount(ele);
2774                 listNodeValue(ln) = c->argv[3];
2775                 incrRefCount(c->argv[3]);
2776                 addReply(c,shared.ok);
2777                 server.dirty++;
2778             }
2779         }
2780     }
2781 }
2782
2783 static void popGenericCommand(redisClient *c, int where) {
2784     robj *o;
2785
2786     o = lookupKeyWrite(c->db,c->argv[1]);
2787     if (o == NULL) {
2788         addReply(c,shared.nullbulk);
2789     } else {
2790         if (o->type != REDIS_LIST) {
2791             addReply(c,shared.wrongtypeerr);
2792         } else {
2793             list *list = o->ptr;
2794             listNode *ln;
2795
2796             if (where == REDIS_HEAD)
2797                 ln = listFirst(list);
2798             else
2799                 ln = listLast(list);
2800
2801             if (ln == NULL) {
2802                 addReply(c,shared.nullbulk);
2803             } else {
2804                 robj *ele = listNodeValue(ln);
2805                 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
2806                 addReply(c,ele);
2807                 addReply(c,shared.crlf);
2808                 listDelNode(list,ln);
2809                 server.dirty++;
2810             }
2811         }
2812     }
2813 }
2814
2815 static void lpopCommand(redisClient *c) {
2816     popGenericCommand(c,REDIS_HEAD);
2817 }
2818
2819 static void rpopCommand(redisClient *c) {
2820     popGenericCommand(c,REDIS_TAIL);
2821 }
2822
2823 static void lrangeCommand(redisClient *c) {
2824     robj *o;
2825     int start = atoi(c->argv[2]->ptr);
2826     int end = atoi(c->argv[3]->ptr);
2827
2828     o = lookupKeyRead(c->db,c->argv[1]);
2829     if (o == NULL) {
2830         addReply(c,shared.nullmultibulk);
2831     } else {
2832         if (o->type != REDIS_LIST) {
2833             addReply(c,shared.wrongtypeerr);
2834         } else {
2835             list *list = o->ptr;
2836             listNode *ln;
2837             int llen = listLength(list);
2838             int rangelen, j;
2839             robj *ele;
2840
2841             /* convert negative indexes */
2842             if (start < 0) start = llen+start;
2843             if (end < 0) end = llen+end;
2844             if (start < 0) start = 0;
2845             if (end < 0) end = 0;
2846
2847             /* indexes sanity checks */
2848             if (start > end || start >= llen) {
2849                 /* Out of range start or start > end result in empty list */
2850                 addReply(c,shared.emptymultibulk);
2851                 return;
2852             }
2853             if (end >= llen) end = llen-1;
2854             rangelen = (end-start)+1;
2855
2856             /* Return the result in form of a multi-bulk reply */
2857             ln = listIndex(list, start);
2858             addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",rangelen));
2859             for (j = 0; j < rangelen; j++) {
2860                 ele = listNodeValue(ln);
2861                 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
2862                 addReply(c,ele);
2863                 addReply(c,shared.crlf);
2864                 ln = ln->next;
2865             }
2866         }
2867     }
2868 }
2869
2870 static void ltrimCommand(redisClient *c) {
2871     robj *o;
2872     int start = atoi(c->argv[2]->ptr);
2873     int end = atoi(c->argv[3]->ptr);
2874
2875     o = lookupKeyWrite(c->db,c->argv[1]);
2876     if (o == NULL) {
2877         addReply(c,shared.nokeyerr);
2878     } else {
2879         if (o->type != REDIS_LIST) {
2880             addReply(c,shared.wrongtypeerr);
2881         } else {
2882             list *list = o->ptr;
2883             listNode *ln;
2884             int llen = listLength(list);
2885             int j, ltrim, rtrim;
2886
2887             /* convert negative indexes */
2888             if (start < 0) start = llen+start;
2889             if (end < 0) end = llen+end;
2890             if (start < 0) start = 0;
2891             if (end < 0) end = 0;
2892
2893             /* indexes sanity checks */
2894             if (start > end || start >= llen) {
2895                 /* Out of range start or start > end result in empty list */
2896                 ltrim = llen;
2897                 rtrim = 0;
2898             } else {
2899                 if (end >= llen) end = llen-1;
2900                 ltrim = start;
2901                 rtrim = llen-end-1;
2902             }
2903
2904             /* Remove list elements to perform the trim */
2905             for (j = 0; j < ltrim; j++) {
2906                 ln = listFirst(list);
2907                 listDelNode(list,ln);
2908             }
2909             for (j = 0; j < rtrim; j++) {
2910                 ln = listLast(list);
2911                 listDelNode(list,ln);
2912             }
2913             server.dirty++;
2914             addReply(c,shared.ok);
2915         }
2916     }
2917 }
2918
2919 static void lremCommand(redisClient *c) {
2920     robj *o;
2921
2922     o = lookupKeyWrite(c->db,c->argv[1]);
2923     if (o == NULL) {
2924         addReply(c,shared.czero);
2925     } else {
2926         if (o->type != REDIS_LIST) {
2927             addReply(c,shared.wrongtypeerr);
2928         } else {
2929             list *list = o->ptr;
2930             listNode *ln, *next;
2931             int toremove = atoi(c->argv[2]->ptr);
2932             int removed = 0;
2933             int fromtail = 0;
2934
2935             if (toremove < 0) {
2936                 toremove = -toremove;
2937                 fromtail = 1;
2938             }
2939             ln = fromtail ? list->tail : list->head;
2940             while (ln) {
2941                 robj *ele = listNodeValue(ln);
2942
2943                 next = fromtail ? ln->prev : ln->next;
2944                 if (sdscmp(ele->ptr,c->argv[3]->ptr) == 0) {
2945                     listDelNode(list,ln);
2946                     server.dirty++;
2947                     removed++;
2948                     if (toremove && removed == toremove) break;
2949                 }
2950                 ln = next;
2951             }
2952             addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",removed));
2953         }
2954     }
2955 }
2956
2957 /* ==================================== Sets ================================ */
2958
2959 static void saddCommand(redisClient *c) {
2960     robj *set;
2961
2962     set = lookupKeyWrite(c->db,c->argv[1]);
2963     if (set == NULL) {
2964         set = createSetObject();
2965         dictAdd(c->db->dict,c->argv[1],set);
2966         incrRefCount(c->argv[1]);
2967     } else {
2968         if (set->type != REDIS_SET) {
2969             addReply(c,shared.wrongtypeerr);
2970             return;
2971         }
2972     }
2973     if (dictAdd(set->ptr,c->argv[2],NULL) == DICT_OK) {
2974         incrRefCount(c->argv[2]);
2975         server.dirty++;
2976         addReply(c,shared.cone);
2977     } else {
2978         addReply(c,shared.czero);
2979     }
2980 }
2981
2982 static void sremCommand(redisClient *c) {
2983     robj *set;
2984
2985     set = lookupKeyWrite(c->db,c->argv[1]);
2986     if (set == NULL) {
2987         addReply(c,shared.czero);
2988     } else {
2989         if (set->type != REDIS_SET) {
2990             addReply(c,shared.wrongtypeerr);
2991             return;
2992         }
2993         if (dictDelete(set->ptr,c->argv[2]) == DICT_OK) {
2994             server.dirty++;
2995             if (htNeedsResize(set->ptr)) dictResize(set->ptr);
2996             addReply(c,shared.cone);
2997         } else {
2998             addReply(c,shared.czero);
2999         }
3000     }
3001 }
3002
3003 static void smoveCommand(redisClient *c) {
3004     robj *srcset, *dstset;
3005
3006     srcset = lookupKeyWrite(c->db,c->argv[1]);
3007     dstset = lookupKeyWrite(c->db,c->argv[2]);
3008
3009     /* If the source key does not exist return 0, if it's of the wrong type
3010      * raise an error */
3011     if (srcset == NULL || srcset->type != REDIS_SET) {
3012         addReply(c, srcset ? shared.wrongtypeerr : shared.czero);
3013         return;
3014     }
3015     /* Error if the destination key is not a set as well */
3016     if (dstset && dstset->type != REDIS_SET) {
3017         addReply(c,shared.wrongtypeerr);
3018         return;
3019     }
3020     /* Remove the element from the source set */
3021     if (dictDelete(srcset->ptr,c->argv[3]) == DICT_ERR) {
3022         /* Key not found in the src set! return zero */
3023         addReply(c,shared.czero);
3024         return;
3025     }
3026     server.dirty++;
3027     /* Add the element to the destination set */
3028     if (!dstset) {
3029         dstset = createSetObject();
3030         dictAdd(c->db->dict,c->argv[2],dstset);
3031         incrRefCount(c->argv[2]);
3032     }
3033     if (dictAdd(dstset->ptr,c->argv[3],NULL) == DICT_OK)
3034         incrRefCount(c->argv[3]);
3035     addReply(c,shared.cone);
3036 }
3037
3038 static void sismemberCommand(redisClient *c) {
3039     robj *set;
3040
3041     set = lookupKeyRead(c->db,c->argv[1]);
3042     if (set == NULL) {
3043         addReply(c,shared.czero);
3044     } else {
3045         if (set->type != REDIS_SET) {
3046             addReply(c,shared.wrongtypeerr);
3047             return;
3048         }
3049         if (dictFind(set->ptr,c->argv[2]))
3050             addReply(c,shared.cone);
3051         else
3052             addReply(c,shared.czero);
3053     }
3054 }
3055
3056 static void scardCommand(redisClient *c) {
3057     robj *o;
3058     dict *s;
3059
3060     o = lookupKeyRead(c->db,c->argv[1]);
3061     if (o == NULL) {
3062         addReply(c,shared.czero);
3063         return;
3064     } else {
3065         if (o->type != REDIS_SET) {
3066             addReply(c,shared.wrongtypeerr);
3067         } else {
3068             s = o->ptr;
3069             addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",
3070                 dictSize(s)));
3071         }
3072     }
3073 }
3074
3075 static void spopCommand(redisClient *c) {
3076     robj *set;
3077     dictEntry *de;
3078
3079     set = lookupKeyWrite(c->db,c->argv[1]);
3080     if (set == NULL) {
3081         addReply(c,shared.nullbulk);
3082     } else {
3083         if (set->type != REDIS_SET) {
3084             addReply(c,shared.wrongtypeerr);
3085             return;
3086         }
3087         de = dictGetRandomKey(set->ptr);
3088         if (de == NULL) {
3089             addReply(c,shared.nullbulk);
3090         } else {
3091             robj *ele = dictGetEntryKey(de);
3092
3093             addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(ele->ptr)));
3094             addReply(c,ele);
3095             addReply(c,shared.crlf);
3096             dictDelete(set->ptr,ele);
3097             if (htNeedsResize(set->ptr)) dictResize(set->ptr);
3098             server.dirty++;
3099         }
3100     }
3101 }
3102
3103 static int qsortCompareSetsByCardinality(const void *s1, const void *s2) {
3104     dict **d1 = (void*) s1, **d2 = (void*) s2;
3105
3106     return dictSize(*d1)-dictSize(*d2);
3107 }
3108
3109 static void sinterGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey) {
3110     dict **dv = zmalloc(sizeof(dict*)*setsnum);
3111     dictIterator *di;
3112     dictEntry *de;
3113     robj *lenobj = NULL, *dstset = NULL;
3114     int j, cardinality = 0;
3115
3116     if (!dv) oom("sinterGenericCommand");
3117     for (j = 0; j < setsnum; j++) {
3118         robj *setobj;
3119
3120         setobj = dstkey ?
3121                     lookupKeyWrite(c->db,setskeys[j]) :
3122                     lookupKeyRead(c->db,setskeys[j]);
3123         if (!setobj) {
3124             zfree(dv);
3125             if (dstkey) {
3126                 deleteKey(c->db,dstkey);
3127                 addReply(c,shared.ok);
3128             } else {
3129                 addReply(c,shared.nullmultibulk);
3130             }
3131             return;
3132         }
3133         if (setobj->type != REDIS_SET) {
3134             zfree(dv);
3135             addReply(c,shared.wrongtypeerr);
3136             return;
3137         }
3138         dv[j] = setobj->ptr;
3139     }
3140     /* Sort sets from the smallest to largest, this will improve our
3141      * algorithm's performace */
3142     qsort(dv,setsnum,sizeof(dict*),qsortCompareSetsByCardinality);
3143
3144     /* The first thing we should output is the total number of elements...
3145      * since this is a multi-bulk write, but at this stage we don't know
3146      * the intersection set size, so we use a trick, append an empty object
3147      * to the output list and save the pointer to later modify it with the
3148      * right length */
3149     if (!dstkey) {
3150         lenobj = createObject(REDIS_STRING,NULL);
3151         addReply(c,lenobj);
3152         decrRefCount(lenobj);
3153     } else {
3154         /* If we have a target key where to store the resulting set
3155          * create this key with an empty set inside */
3156         dstset = createSetObject();
3157     }
3158
3159     /* Iterate all the elements of the first (smallest) set, and test
3160      * the element against all the other sets, if at least one set does
3161      * not include the element it is discarded */
3162     di = dictGetIterator(dv[0]);
3163     if (!di) oom("dictGetIterator");
3164
3165     while((de = dictNext(di)) != NULL) {
3166         robj *ele;
3167
3168         for (j = 1; j < setsnum; j++)
3169             if (dictFind(dv[j],dictGetEntryKey(de)) == NULL) break;
3170         if (j != setsnum)
3171             continue; /* at least one set does not contain the member */
3172         ele = dictGetEntryKey(de);
3173         if (!dstkey) {
3174             addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(ele->ptr)));
3175             addReply(c,ele);
3176             addReply(c,shared.crlf);
3177             cardinality++;
3178         } else {
3179             dictAdd(dstset->ptr,ele,NULL);
3180             incrRefCount(ele);
3181         }
3182     }
3183     dictReleaseIterator(di);
3184
3185     if (dstkey) {
3186         /* Store the resulting set into the target */
3187         deleteKey(c->db,dstkey);
3188         dictAdd(c->db->dict,dstkey,dstset);
3189         incrRefCount(dstkey);
3190     }
3191
3192     if (!dstkey) {
3193         lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality);
3194     } else {
3195         addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",
3196             dictSize((dict*)dstset->ptr)));
3197         server.dirty++;
3198     }
3199     zfree(dv);
3200 }
3201
3202 static void sinterCommand(redisClient *c) {
3203     sinterGenericCommand(c,c->argv+1,c->argc-1,NULL);
3204 }
3205
3206 static void sinterstoreCommand(redisClient *c) {
3207     sinterGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
3208 }
3209
3210 #define REDIS_OP_UNION 0
3211 #define REDIS_OP_DIFF 1
3212
3213 static void sunionDiffGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey, int op) {
3214     dict **dv = zmalloc(sizeof(dict*)*setsnum);
3215     dictIterator *di;
3216     dictEntry *de;
3217     robj *dstset = NULL;
3218     int j, cardinality = 0;
3219
3220     if (!dv) oom("sunionDiffGenericCommand");
3221     for (j = 0; j < setsnum; j++) {
3222         robj *setobj;
3223
3224         setobj = dstkey ?
3225                     lookupKeyWrite(c->db,setskeys[j]) :
3226                     lookupKeyRead(c->db,setskeys[j]);
3227         if (!setobj) {
3228             dv[j] = NULL;
3229             continue;
3230         }
3231         if (setobj->type != REDIS_SET) {
3232             zfree(dv);
3233             addReply(c,shared.wrongtypeerr);
3234             return;
3235         }
3236         dv[j] = setobj->ptr;
3237     }
3238
3239     /* We need a temp set object to store our union. If the dstkey
3240      * is not NULL (that is, we are inside an SUNIONSTORE operation) then
3241      * this set object will be the resulting object to set into the target key*/
3242     dstset = createSetObject();
3243
3244     /* Iterate all the elements of all the sets, add every element a single
3245      * time to the result set */
3246     for (j = 0; j < setsnum; j++) {
3247         if (op == REDIS_OP_DIFF && j == 0 && !dv[j]) break; /* result set is empty */
3248         if (!dv[j]) continue; /* non existing keys are like empty sets */
3249
3250         di = dictGetIterator(dv[j]);
3251         if (!di) oom("dictGetIterator");
3252
3253         while((de = dictNext(di)) != NULL) {
3254             robj *ele;
3255
3256             /* dictAdd will not add the same element multiple times */
3257             ele = dictGetEntryKey(de);
3258             if (op == REDIS_OP_UNION || j == 0) {
3259                 if (dictAdd(dstset->ptr,ele,NULL) == DICT_OK) {
3260                     incrRefCount(ele);
3261                     cardinality++;
3262                 }
3263             } else if (op == REDIS_OP_DIFF) {
3264                 if (dictDelete(dstset->ptr,ele) == DICT_OK) {
3265                     cardinality--;
3266                 }
3267             }
3268         }
3269         dictReleaseIterator(di);
3270
3271         if (op == REDIS_OP_DIFF && cardinality == 0) break; /* result set is empty */
3272     }
3273
3274     /* Output the content of the resulting set, if not in STORE mode */
3275     if (!dstkey) {
3276         addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",cardinality));
3277         di = dictGetIterator(dstset->ptr);
3278         if (!di) oom("dictGetIterator");
3279         while((de = dictNext(di)) != NULL) {
3280             robj *ele;
3281
3282             ele = dictGetEntryKey(de);
3283             addReplySds(c,sdscatprintf(sdsempty(),
3284                     "$%d\r\n",sdslen(ele->ptr)));
3285             addReply(c,ele);
3286             addReply(c,shared.crlf);
3287         }
3288         dictReleaseIterator(di);
3289     } else {
3290         /* If we have a target key where to store the resulting set
3291          * create this key with the result set inside */
3292         deleteKey(c->db,dstkey);
3293         dictAdd(c->db->dict,dstkey,dstset);
3294         incrRefCount(dstkey);
3295     }
3296
3297     /* Cleanup */
3298     if (!dstkey) {
3299         decrRefCount(dstset);
3300     } else {
3301         addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",
3302             dictSize((dict*)dstset->ptr)));
3303         server.dirty++;
3304     }
3305     zfree(dv);
3306 }
3307
3308 static void sunionCommand(redisClient *c) {
3309     sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,REDIS_OP_UNION);
3310 }
3311
3312 static void sunionstoreCommand(redisClient *c) {
3313     sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],REDIS_OP_UNION);
3314 }
3315
3316 static void sdiffCommand(redisClient *c) {
3317     sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,REDIS_OP_DIFF);
3318 }
3319
3320 static void sdiffstoreCommand(redisClient *c) {
3321     sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],REDIS_OP_DIFF);
3322 }
3323
3324 static void flushdbCommand(redisClient *c) {
3325     server.dirty += dictSize(c->db->dict);
3326     dictEmpty(c->db->dict);
3327     dictEmpty(c->db->expires);
3328     addReply(c,shared.ok);
3329 }
3330
3331 static void flushallCommand(redisClient *c) {
3332     server.dirty += emptyDb();
3333     addReply(c,shared.ok);
3334     rdbSave(server.dbfilename);
3335     server.dirty++;
3336 }
3337
3338 static redisSortOperation *createSortOperation(int type, robj *pattern) {
3339     redisSortOperation *so = zmalloc(sizeof(*so));
3340     if (!so) oom("createSortOperation");
3341     so->type = type;
3342     so->pattern = pattern;
3343     return so;
3344 }
3345
3346 /* Return the value associated to the key with a name obtained
3347  * substituting the first occurence of '*' in 'pattern' with 'subst' */
3348 static robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
3349     char *p;
3350     sds spat, ssub;
3351     robj keyobj;
3352     int prefixlen, sublen, postfixlen;
3353     /* Expoit the internal sds representation to create a sds string allocated on the stack in order to make this function faster */
3354     struct {
3355         long len;
3356         long free;
3357         char buf[REDIS_SORTKEY_MAX+1];
3358     } keyname;
3359
3360     spat = pattern->ptr;
3361     ssub = subst->ptr;
3362     if (sdslen(spat)+sdslen(ssub)-1 > REDIS_SORTKEY_MAX) return NULL;
3363     p = strchr(spat,'*');
3364     if (!p) return NULL;
3365
3366     prefixlen = p-spat;
3367     sublen = sdslen(ssub);
3368     postfixlen = sdslen(spat)-(prefixlen+1);
3369     memcpy(keyname.buf,spat,prefixlen);
3370     memcpy(keyname.buf+prefixlen,ssub,sublen);
3371     memcpy(keyname.buf+prefixlen+sublen,p+1,postfixlen);
3372     keyname.buf[prefixlen+sublen+postfixlen] = '\0';
3373     keyname.len = prefixlen+sublen+postfixlen;
3374
3375     keyobj.refcount = 1;
3376     keyobj.type = REDIS_STRING;
3377     keyobj.ptr = ((char*)&keyname)+(sizeof(long)*2);
3378
3379     /* printf("lookup '%s' => %p\n", keyname.buf,de); */
3380     return lookupKeyRead(db,&keyobj);
3381 }
3382
3383 /* sortCompare() is used by qsort in sortCommand(). Given that qsort_r with
3384  * the additional parameter is not standard but a BSD-specific we have to
3385  * pass sorting parameters via the global 'server' structure */
3386 static int sortCompare(const void *s1, const void *s2) {
3387     const redisSortObject *so1 = s1, *so2 = s2;
3388     int cmp;
3389
3390     if (!server.sort_alpha) {
3391         /* Numeric sorting. Here it's trivial as we precomputed scores */
3392         if (so1->u.score > so2->u.score) {
3393             cmp = 1;
3394         } else if (so1->u.score < so2->u.score) {
3395             cmp = -1;
3396         } else {
3397             cmp = 0;
3398         }
3399     } else {
3400         /* Alphanumeric sorting */
3401         if (server.sort_bypattern) {
3402             if (!so1->u.cmpobj || !so2->u.cmpobj) {
3403                 /* At least one compare object is NULL */
3404                 if (so1->u.cmpobj == so2->u.cmpobj)
3405                     cmp = 0;
3406                 else if (so1->u.cmpobj == NULL)
3407                     cmp = -1;
3408                 else
3409                     cmp = 1;
3410             } else {
3411                 /* We have both the objects, use strcoll */
3412                 cmp = strcoll(so1->u.cmpobj->ptr,so2->u.cmpobj->ptr);
3413             }
3414         } else {
3415             /* Compare elements directly */
3416             cmp = strcoll(so1->obj->ptr,so2->obj->ptr);
3417         }
3418     }
3419     return server.sort_desc ? -cmp : cmp;
3420 }
3421
3422 /* The SORT command is the most complex command in Redis. Warning: this code
3423  * is optimized for speed and a bit less for readability */
3424 static void sortCommand(redisClient *c) {
3425     list *operations;
3426     int outputlen = 0;
3427     int desc = 0, alpha = 0;
3428     int limit_start = 0, limit_count = -1, start, end;
3429     int j, dontsort = 0, vectorlen;
3430     int getop = 0; /* GET operation counter */
3431     robj *sortval, *sortby = NULL;
3432     redisSortObject *vector; /* Resulting vector to sort */
3433
3434     /* Lookup the key to sort. It must be of the right types */
3435     sortval = lookupKeyRead(c->db,c->argv[1]);
3436     if (sortval == NULL) {
3437         addReply(c,shared.nokeyerr);
3438         return;
3439     }
3440     if (sortval->type != REDIS_SET && sortval->type != REDIS_LIST) {
3441         addReply(c,shared.wrongtypeerr);
3442         return;
3443     }
3444
3445     /* Create a list of operations to perform for every sorted element.
3446      * Operations can be GET/DEL/INCR/DECR */
3447     operations = listCreate();
3448     listSetFreeMethod(operations,zfree);
3449     j = 2;
3450
3451     /* Now we need to protect sortval incrementing its count, in the future
3452      * SORT may have options able to overwrite/delete keys during the sorting
3453      * and the sorted key itself may get destroied */
3454     incrRefCount(sortval);
3455
3456     /* The SORT command has an SQL-alike syntax, parse it */
3457     while(j < c->argc) {
3458         int leftargs = c->argc-j-1;
3459         if (!strcasecmp(c->argv[j]->ptr,"asc")) {
3460             desc = 0;
3461         } else if (!strcasecmp(c->argv[j]->ptr,"desc")) {
3462             desc = 1;
3463         } else if (!strcasecmp(c->argv[j]->ptr,"alpha")) {
3464             alpha = 1;
3465         } else if (!strcasecmp(c->argv[j]->ptr,"limit") && leftargs >= 2) {
3466             limit_start = atoi(c->argv[j+1]->ptr);
3467             limit_count = atoi(c->argv[j+2]->ptr);
3468             j+=2;
3469         } else if (!strcasecmp(c->argv[j]->ptr,"by") && leftargs >= 1) {
3470             sortby = c->argv[j+1];
3471             /* If the BY pattern does not contain '*', i.e. it is constant,
3472              * we don't need to sort nor to lookup the weight keys. */
3473             if (strchr(c->argv[j+1]->ptr,'*') == NULL) dontsort = 1;
3474             j++;
3475         } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
3476             listAddNodeTail(operations,createSortOperation(
3477                 REDIS_SORT_GET,c->argv[j+1]));
3478             getop++;
3479             j++;
3480         } else if (!strcasecmp(c->argv[j]->ptr,"del") && leftargs >= 1) {
3481             listAddNodeTail(operations,createSortOperation(
3482                 REDIS_SORT_DEL,c->argv[j+1]));
3483             j++;
3484         } else if (!strcasecmp(c->argv[j]->ptr,"incr") && leftargs >= 1) {
3485             listAddNodeTail(operations,createSortOperation(
3486                 REDIS_SORT_INCR,c->argv[j+1]));
3487             j++;
3488         } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
3489             listAddNodeTail(operations,createSortOperation(
3490                 REDIS_SORT_DECR,c->argv[j+1]));
3491             j++;
3492         } else {
3493             decrRefCount(sortval);
3494             listRelease(operations);
3495             addReply(c,shared.syntaxerr);
3496             return;
3497         }
3498         j++;
3499     }
3500
3501     /* Load the sorting vector with all the objects to sort */
3502     vectorlen = (sortval->type == REDIS_LIST) ?
3503         listLength((list*)sortval->ptr) :
3504         dictSize((dict*)sortval->ptr);
3505     vector = zmalloc(sizeof(redisSortObject)*vectorlen);
3506     if (!vector) oom("allocating objects vector for SORT");
3507     j = 0;
3508     if (sortval->type == REDIS_LIST) {
3509         list *list = sortval->ptr;
3510         listNode *ln;
3511
3512         listRewind(list);
3513         while((ln = listYield(list))) {
3514             robj *ele = ln->value;
3515             vector[j].obj = ele;
3516             vector[j].u.score = 0;
3517             vector[j].u.cmpobj = NULL;
3518             j++;
3519         }
3520     } else {
3521         dict *set = sortval->ptr;
3522         dictIterator *di;
3523         dictEntry *setele;
3524
3525         di = dictGetIterator(set);
3526         if (!di) oom("dictGetIterator");
3527         while((setele = dictNext(di)) != NULL) {
3528             vector[j].obj = dictGetEntryKey(setele);
3529             vector[j].u.score = 0;
3530             vector[j].u.cmpobj = NULL;
3531             j++;
3532         }
3533         dictReleaseIterator(di);
3534     }
3535     assert(j == vectorlen);
3536
3537     /* Now it's time to load the right scores in the sorting vector */
3538     if (dontsort == 0) {
3539         for (j = 0; j < vectorlen; j++) {
3540             if (sortby) {
3541                 robj *byval;
3542
3543                 byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
3544                 if (!byval || byval->type != REDIS_STRING) continue;
3545                 if (alpha) {
3546                     vector[j].u.cmpobj = byval;
3547                     incrRefCount(byval);
3548                 } else {
3549                     vector[j].u.score = strtod(byval->ptr,NULL);
3550                 }
3551             } else {
3552                 if (!alpha) vector[j].u.score = strtod(vector[j].obj->ptr,NULL);
3553             }
3554         }
3555     }
3556
3557     /* We are ready to sort the vector... perform a bit of sanity check
3558      * on the LIMIT option too. We'll use a partial version of quicksort. */
3559     start = (limit_start < 0) ? 0 : limit_start;
3560     end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
3561     if (start >= vectorlen) {
3562         start = vectorlen-1;
3563         end = vectorlen-2;
3564     }
3565     if (end >= vectorlen) end = vectorlen-1;
3566
3567     if (dontsort == 0) {
3568         server.sort_desc = desc;
3569         server.sort_alpha = alpha;
3570         server.sort_bypattern = sortby ? 1 : 0;
3571         if (sortby && (start != 0 || end != vectorlen-1))
3572             pqsort(vector,vectorlen,sizeof(redisSortObject),sortCompare, start,end);
3573         else
3574             qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare);
3575     }
3576
3577     /* Send command output to the output buffer, performing the specified
3578      * GET/DEL/INCR/DECR operations if any. */
3579     outputlen = getop ? getop*(end-start+1) : end-start+1;
3580     addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",outputlen));
3581     for (j = start; j <= end; j++) {
3582         listNode *ln;
3583         if (!getop) {
3584             addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
3585                 sdslen(vector[j].obj->ptr)));
3586             addReply(c,vector[j].obj);
3587             addReply(c,shared.crlf);
3588         }
3589         listRewind(operations);
3590         while((ln = listYield(operations))) {
3591             redisSortOperation *sop = ln->value;
3592             robj *val = lookupKeyByPattern(c->db,sop->pattern,
3593                 vector[j].obj);
3594
3595             if (sop->type == REDIS_SORT_GET) {
3596                 if (!val || val->type != REDIS_STRING) {
3597                     addReply(c,shared.nullbulk);
3598                 } else {
3599                     addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
3600                         sdslen(val->ptr)));
3601                     addReply(c,val);
3602                     addReply(c,shared.crlf);
3603                 }
3604             } else if (sop->type == REDIS_SORT_DEL) {
3605                 /* TODO */
3606             }
3607         }
3608     }
3609
3610     /* Cleanup */
3611     decrRefCount(sortval);
3612     listRelease(operations);
3613     for (j = 0; j < vectorlen; j++) {
3614         if (sortby && alpha && vector[j].u.cmpobj)
3615             decrRefCount(vector[j].u.cmpobj);
3616     }
3617     zfree(vector);
3618 }
3619
3620 static void infoCommand(redisClient *c) {
3621     sds info;
3622     time_t uptime = time(NULL)-server.stat_starttime;
3623     int j;
3624
3625     info = sdscatprintf(sdsempty(),
3626         "redis_version:%s\r\n"
3627         "uptime_in_seconds:%d\r\n"
3628         "uptime_in_days:%d\r\n"
3629         "connected_clients:%d\r\n"
3630         "connected_slaves:%d\r\n"
3631         "used_memory:%zu\r\n"
3632         "changes_since_last_save:%lld\r\n"
3633         "bgsave_in_progress:%d\r\n"
3634         "last_save_time:%d\r\n"
3635         "total_connections_received:%lld\r\n"
3636         "total_commands_processed:%lld\r\n"
3637         "role:%s\r\n"
3638         ,REDIS_VERSION,
3639         uptime,
3640         uptime/(3600*24),
3641         listLength(server.clients)-listLength(server.slaves),
3642         listLength(server.slaves),
3643         server.usedmemory,
3644         server.dirty,
3645         server.bgsaveinprogress,
3646         server.lastsave,
3647         server.stat_numconnections,
3648         server.stat_numcommands,
3649         server.masterhost == NULL ? "master" : "slave"
3650     );
3651     if (server.masterhost) {
3652         info = sdscatprintf(info,
3653             "master_host:%s\r\n"
3654             "master_port:%d\r\n"
3655             "master_link_status:%s\r\n"
3656             "master_last_io_seconds_ago:%d\r\n"
3657             ,server.masterhost,
3658             server.masterport,
3659             (server.replstate == REDIS_REPL_CONNECTED) ?
3660                 "up" : "down",
3661             (int)(time(NULL)-server.master->lastinteraction)
3662         );
3663     }
3664     for (j = 0; j < server.dbnum; j++) {
3665         long long keys, vkeys;
3666
3667         keys = dictSize(server.db[j].dict);
3668         vkeys = dictSize(server.db[j].expires);
3669         if (keys || vkeys) {
3670             info = sdscatprintf(info, "db%d: keys=%lld,expires=%lld\r\n",
3671                 j, keys, vkeys);
3672         }
3673     }
3674     addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(info)));
3675     addReplySds(c,info);
3676     addReply(c,shared.crlf);
3677 }
3678
3679 static void monitorCommand(redisClient *c) {
3680     /* ignore MONITOR if aleady slave or in monitor mode */
3681     if (c->flags & REDIS_SLAVE) return;
3682
3683     c->flags |= (REDIS_SLAVE|REDIS_MONITOR);
3684     c->slaveseldb = 0;
3685     if (!listAddNodeTail(server.monitors,c)) oom("listAddNodeTail");
3686     addReply(c,shared.ok);
3687 }
3688
3689 /* ================================= Expire ================================= */
3690 static int removeExpire(redisDb *db, robj *key) {
3691     if (dictDelete(db->expires,key) == DICT_OK) {
3692         return 1;
3693     } else {
3694         return 0;
3695     }
3696 }
3697
3698 static int setExpire(redisDb *db, robj *key, time_t when) {
3699     if (dictAdd(db->expires,key,(void*)when) == DICT_ERR) {
3700         return 0;
3701     } else {
3702         incrRefCount(key);
3703         return 1;
3704     }
3705 }
3706
3707 /* Return the expire time of the specified key, or -1 if no expire
3708  * is associated with this key (i.e. the key is non volatile) */
3709 static time_t getExpire(redisDb *db, robj *key) {
3710     dictEntry *de;
3711
3712     /* No expire? return ASAP */
3713     if (dictSize(db->expires) == 0 ||
3714        (de = dictFind(db->expires,key)) == NULL) return -1;
3715
3716     return (time_t) dictGetEntryVal(de);
3717 }
3718
3719 static int expireIfNeeded(redisDb *db, robj *key) {
3720     time_t when;
3721     dictEntry *de;
3722
3723     /* No expire? return ASAP */
3724     if (dictSize(db->expires) == 0 ||
3725        (de = dictFind(db->expires,key)) == NULL) return 0;
3726
3727     /* Lookup the expire */
3728     when = (time_t) dictGetEntryVal(de);
3729     if (time(NULL) <= when) return 0;
3730
3731     /* Delete the key */
3732     dictDelete(db->expires,key);
3733     return dictDelete(db->dict,key) == DICT_OK;
3734 }
3735
3736 static int deleteIfVolatile(redisDb *db, robj *key) {
3737     dictEntry *de;
3738
3739     /* No expire? return ASAP */
3740     if (dictSize(db->expires) == 0 ||
3741        (de = dictFind(db->expires,key)) == NULL) return 0;
3742
3743     /* Delete the key */
3744     server.dirty++;
3745     dictDelete(db->expires,key);
3746     return dictDelete(db->dict,key) == DICT_OK;
3747 }
3748
3749 static void expireCommand(redisClient *c) {
3750     dictEntry *de;
3751     int seconds = atoi(c->argv[2]->ptr);
3752
3753     de = dictFind(c->db->dict,c->argv[1]);
3754     if (de == NULL) {
3755         addReply(c,shared.czero);
3756         return;
3757     }
3758     if (seconds <= 0) {
3759         addReply(c, shared.czero);
3760         return;
3761     } else {
3762         time_t when = time(NULL)+seconds;
3763         if (setExpire(c->db,c->argv[1],when)) {
3764             addReply(c,shared.cone);
3765             server.dirty++;
3766         } else {
3767             addReply(c,shared.czero);
3768         }
3769         return;
3770     }
3771 }
3772
3773 static void ttlCommand(redisClient *c) {
3774     time_t expire;
3775     int ttl = -1;
3776
3777     expire = getExpire(c->db,c->argv[1]);
3778     if (expire != -1) {
3779         ttl = (int) (expire-time(NULL));
3780         if (ttl < 0) ttl = -1;
3781     }
3782     addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",ttl));
3783 }
3784
3785 /* =============================== Replication  ============================= */
3786
3787 static int syncWrite(int fd, char *ptr, ssize_t size, int timeout) {
3788     ssize_t nwritten, ret = size;
3789     time_t start = time(NULL);
3790
3791     timeout++;
3792     while(size) {
3793         if (aeWait(fd,AE_WRITABLE,1000) & AE_WRITABLE) {
3794             nwritten = write(fd,ptr,size);
3795             if (nwritten == -1) return -1;
3796             ptr += nwritten;
3797             size -= nwritten;
3798         }
3799         if ((time(NULL)-start) > timeout) {
3800             errno = ETIMEDOUT;
3801             return -1;
3802         }
3803     }
3804     return ret;
3805 }
3806
3807 static int syncRead(int fd, char *ptr, ssize_t size, int timeout) {
3808     ssize_t nread, totread = 0;
3809     time_t start = time(NULL);
3810
3811     timeout++;
3812     while(size) {
3813         if (aeWait(fd,AE_READABLE,1000) & AE_READABLE) {
3814             nread = read(fd,ptr,size);
3815             if (nread == -1) return -1;
3816             ptr += nread;
3817             size -= nread;
3818             totread += nread;
3819         }
3820         if ((time(NULL)-start) > timeout) {
3821             errno = ETIMEDOUT;
3822             return -1;
3823         }
3824     }
3825     return totread;
3826 }
3827
3828 static int syncReadLine(int fd, char *ptr, ssize_t size, int timeout) {
3829     ssize_t nread = 0;
3830
3831     size--;
3832     while(size) {
3833         char c;
3834
3835         if (syncRead(fd,&c,1,timeout) == -1) return -1;
3836         if (c == '\n') {
3837             *ptr = '\0';
3838             if (nread && *(ptr-1) == '\r') *(ptr-1) = '\0';
3839             return nread;
3840         } else {
3841             *ptr++ = c;
3842             *ptr = '\0';
3843             nread++;
3844         }
3845     }
3846     return nread;
3847 }
3848
3849 static void syncCommand(redisClient *c) {
3850     /* ignore SYNC if aleady slave or in monitor mode */
3851     if (c->flags & REDIS_SLAVE) return;
3852
3853     /* SYNC can't be issued when the server has pending data to send to
3854      * the client about already issued commands. We need a fresh reply
3855      * buffer registering the differences between the BGSAVE and the current
3856      * dataset, so that we can copy to other slaves if needed. */
3857     if (listLength(c->reply) != 0) {
3858         addReplySds(c,sdsnew("-ERR SYNC is invalid with pending input\r\n"));
3859         return;
3860     }
3861
3862     redisLog(REDIS_NOTICE,"Slave ask for synchronization");
3863     /* Here we need to check if there is a background saving operation
3864      * in progress, or if it is required to start one */
3865     if (server.bgsaveinprogress) {
3866         /* Ok a background save is in progress. Let's check if it is a good
3867          * one for replication, i.e. if there is another slave that is
3868          * registering differences since the server forked to save */
3869         redisClient *slave;
3870         listNode *ln;
3871
3872         listRewind(server.slaves);
3873         while((ln = listYield(server.slaves))) {
3874             slave = ln->value;
3875             if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
3876         }
3877         if (ln) {
3878             /* Perfect, the server is already registering differences for
3879              * another slave. Set the right state, and copy the buffer. */
3880             listRelease(c->reply);
3881             c->reply = listDup(slave->reply);
3882             if (!c->reply) oom("listDup copying slave reply list");
3883             c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
3884             redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC");
3885         } else {
3886             /* No way, we need to wait for the next BGSAVE in order to
3887              * register differences */
3888             c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
3889             redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
3890         }
3891     } else {
3892         /* Ok we don't have a BGSAVE in progress, let's start one */
3893         redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC");
3894         if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
3895             redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE");
3896             addReplySds(c,sdsnew("-ERR Unalbe to perform background save\r\n"));
3897             return;
3898         }
3899         c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
3900     }
3901     c->repldbfd = -1;
3902     c->flags |= REDIS_SLAVE;
3903     c->slaveseldb = 0;
3904     if (!listAddNodeTail(server.slaves,c)) oom("listAddNodeTail");
3905     return;
3906 }
3907
3908 static void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
3909     redisClient *slave = privdata;
3910     REDIS_NOTUSED(el);
3911     REDIS_NOTUSED(mask);
3912     char buf[REDIS_IOBUF_LEN];
3913     ssize_t nwritten, buflen;
3914
3915     if (slave->repldboff == 0) {
3916         /* Write the bulk write count before to transfer the DB. In theory here
3917          * we don't know how much room there is in the output buffer of the
3918          * socket, but in pratice SO_SNDLOWAT (the minimum count for output
3919          * operations) will never be smaller than the few bytes we need. */
3920         sds bulkcount;
3921
3922         bulkcount = sdscatprintf(sdsempty(),"$%lld\r\n",(unsigned long long)
3923             slave->repldbsize);
3924         if (write(fd,bulkcount,sdslen(bulkcount)) != (signed)sdslen(bulkcount))
3925         {
3926             sdsfree(bulkcount);
3927             freeClient(slave);
3928             return;
3929         }
3930         sdsfree(bulkcount);
3931     }
3932     lseek(slave->repldbfd,slave->repldboff,SEEK_SET);
3933     buflen = read(slave->repldbfd,buf,REDIS_IOBUF_LEN);
3934     if (buflen <= 0) {
3935         redisLog(REDIS_WARNING,"Read error sending DB to slave: %s",
3936             (buflen == 0) ? "premature EOF" : strerror(errno));
3937         freeClient(slave);
3938         return;
3939     }
3940     if ((nwritten = write(fd,buf,buflen)) == -1) {
3941         redisLog(REDIS_DEBUG,"Write error sending DB to slave: %s",
3942             strerror(errno));
3943         freeClient(slave);
3944         return;
3945     }
3946     slave->repldboff += nwritten;
3947     if (slave->repldboff == slave->repldbsize) {
3948         close(slave->repldbfd);
3949         slave->repldbfd = -1;
3950         aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
3951         slave->replstate = REDIS_REPL_ONLINE;
3952         if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE,
3953             sendReplyToClient, slave, NULL) == AE_ERR) {
3954             freeClient(slave);
3955             return;
3956         }
3957         addReplySds(slave,sdsempty());
3958         redisLog(REDIS_NOTICE,"Synchronization with slave succeeded");
3959     }
3960 }
3961
3962 /* This function is called at the end of every backgrond saving.
3963  * The argument bgsaveerr is REDIS_OK if the background saving succeeded
3964  * otherwise REDIS_ERR is passed to the function.
3965  *
3966  * The goal of this function is to handle slaves waiting for a successful
3967  * background saving in order to perform non-blocking synchronization. */
3968 static void updateSlavesWaitingBgsave(int bgsaveerr) {
3969     listNode *ln;
3970     int startbgsave = 0;
3971
3972     listRewind(server.slaves);
3973     while((ln = listYield(server.slaves))) {
3974         redisClient *slave = ln->value;
3975
3976         if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
3977             startbgsave = 1;
3978             slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
3979         } else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
3980             struct redis_stat buf;
3981
3982             if (bgsaveerr != REDIS_OK) {
3983                 freeClient(slave);
3984                 redisLog(REDIS_WARNING,"SYNC failed. BGSAVE child returned an error");
3985                 continue;
3986             }
3987             if ((slave->repldbfd = open(server.dbfilename,O_RDONLY)) == -1 ||
3988                 redis_fstat(slave->repldbfd,&buf) == -1) {
3989                 freeClient(slave);
3990                 redisLog(REDIS_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno));
3991                 continue;
3992             }
3993             slave->repldboff = 0;
3994             slave->repldbsize = buf.st_size;
3995             slave->replstate = REDIS_REPL_SEND_BULK;
3996             aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
3997             if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave, NULL) == AE_ERR) {
3998                 freeClient(slave);
3999                 continue;
4000             }
4001         }
4002     }
4003     if (startbgsave) {
4004         if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
4005             listRewind(server.slaves);
4006             redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
4007             while((ln = listYield(server.slaves))) {
4008                 redisClient *slave = ln->value;
4009
4010                 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
4011                     freeClient(slave);
4012             }
4013         }
4014     }
4015 }
4016
4017 static int syncWithMaster(void) {
4018     char buf[1024], tmpfile[256];
4019     int dumpsize;
4020     int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);
4021     int dfd;
4022
4023     if (fd == -1) {
4024         redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
4025             strerror(errno));
4026         return REDIS_ERR;
4027     }
4028     /* Issue the SYNC command */
4029     if (syncWrite(fd,"SYNC \r\n",7,5) == -1) {
4030         close(fd);
4031         redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
4032             strerror(errno));
4033         return REDIS_ERR;
4034     }
4035     /* Read the bulk write count */
4036     if (syncReadLine(fd,buf,1024,3600) == -1) {
4037         close(fd);
4038         redisLog(REDIS_WARNING,"I/O error reading bulk count from MASTER: %s",
4039             strerror(errno));
4040         return REDIS_ERR;
4041     }
4042     dumpsize = atoi(buf+1);
4043     redisLog(REDIS_NOTICE,"Receiving %d bytes data dump from MASTER",dumpsize);
4044     /* Read the bulk write data on a temp file */
4045     snprintf(tmpfile,256,"temp-%d.%ld.rdb",(int)time(NULL),(long int)random());
4046     dfd = open(tmpfile,O_CREAT|O_WRONLY,0644);
4047     if (dfd == -1) {
4048         close(fd);
4049         redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
4050         return REDIS_ERR;
4051     }
4052     while(dumpsize) {
4053         int nread, nwritten;
4054
4055         nread = read(fd,buf,(dumpsize < 1024)?dumpsize:1024);
4056         if (nread == -1) {
4057             redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
4058                 strerror(errno));
4059             close(fd);
4060             close(dfd);
4061             return REDIS_ERR;
4062         }
4063         nwritten = write(dfd,buf,nread);
4064         if (nwritten == -1) {
4065             redisLog(REDIS_WARNING,"Write error writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno));
4066             close(fd);
4067             close(dfd);
4068             return REDIS_ERR;
4069         }
4070         dumpsize -= nread;
4071     }
4072     close(dfd);
4073     if (rename(tmpfile,server.dbfilename) == -1) {
4074         redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
4075         unlink(tmpfile);
4076         close(fd);
4077         return REDIS_ERR;
4078     }
4079     emptyDb();
4080     if (rdbLoad(server.dbfilename) != REDIS_OK) {
4081         redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
4082         close(fd);
4083         return REDIS_ERR;
4084     }
4085     server.master = createClient(fd);
4086     server.master->flags |= REDIS_MASTER;
4087     server.replstate = REDIS_REPL_CONNECTED;
4088     return REDIS_OK;
4089 }
4090
4091 static void slaveofCommand(redisClient *c) {
4092     if (!strcasecmp(c->argv[1]->ptr,"no") &&
4093         !strcasecmp(c->argv[2]->ptr,"one")) {
4094         if (server.masterhost) {
4095             sdsfree(server.masterhost);
4096             server.masterhost = NULL;
4097             if (server.master) freeClient(server.master);
4098             server.replstate = REDIS_REPL_NONE;
4099             redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)");
4100         }
4101     } else {
4102         sdsfree(server.masterhost);
4103         server.masterhost = sdsdup(c->argv[1]->ptr);
4104         server.masterport = atoi(c->argv[2]->ptr);
4105         if (server.master) freeClient(server.master);
4106         server.replstate = REDIS_REPL_CONNECT;
4107         redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)",
4108             server.masterhost, server.masterport);
4109     }
4110     addReply(c,shared.ok);
4111 }
4112
4113 /* ============================ Maxmemory directive  ======================== */
4114
4115 /* This function gets called when 'maxmemory' is set on the config file to limit
4116  * the max memory used by the server, and we are out of memory.
4117  * This function will try to, in order:
4118  *
4119  * - Free objects from the free list
4120  * - Try to remove keys with an EXPIRE set
4121  *
4122  * It is not possible to free enough memory to reach used-memory < maxmemory
4123  * the server will start refusing commands that will enlarge even more the
4124  * memory usage.
4125  */
4126 static void freeMemoryIfNeeded(void) {
4127     while (server.maxmemory && zmalloc_used_memory() > server.maxmemory) {
4128         if (listLength(server.objfreelist)) {
4129             robj *o;
4130
4131             listNode *head = listFirst(server.objfreelist);
4132             o = listNodeValue(head);
4133             listDelNode(server.objfreelist,head);
4134             zfree(o);
4135         } else {
4136             int j, k, freed = 0;
4137
4138             for (j = 0; j < server.dbnum; j++) {
4139                 int minttl = -1;
4140                 robj *minkey = NULL;
4141                 struct dictEntry *de;
4142
4143                 if (dictSize(server.db[j].expires)) {
4144                     freed = 1;
4145                     /* From a sample of three keys drop the one nearest to
4146                      * the natural expire */
4147                     for (k = 0; k < 3; k++) {
4148                         time_t t;
4149
4150                         de = dictGetRandomKey(server.db[j].expires);
4151                         t = (time_t) dictGetEntryVal(de);
4152                         if (minttl == -1 || t < minttl) {
4153                             minkey = dictGetEntryKey(de);
4154                             minttl = t;
4155                         }
4156                     }
4157                     deleteKey(server.db+j,minkey);
4158                 }
4159             }
4160             if (!freed) return; /* nothing to free... */
4161         }
4162     }
4163 }
4164
4165 /* ================================= Debugging ============================== */
4166
4167 static void debugCommand(redisClient *c) {
4168     if (!strcasecmp(c->argv[1]->ptr,"segfault")) {
4169         *((char*)-1) = 'x';
4170     } else if (!strcasecmp(c->argv[1]->ptr,"object") && c->argc == 3) {
4171         dictEntry *de = dictFind(c->db->dict,c->argv[2]);
4172         robj *key, *val;
4173
4174         if (!de) {
4175             addReply(c,shared.nokeyerr);
4176             return;
4177         }
4178         key = dictGetEntryKey(de);
4179         val = dictGetEntryVal(de);
4180         addReplySds(c,sdscatprintf(sdsempty(),
4181             "+Key at:%p refcount:%d, value at:%p refcount:%d\r\n",
4182                 key, key->refcount, val, val->refcount));
4183     } else {
4184         addReplySds(c,sdsnew(
4185             "-ERR Syntax error, try DEBUG [SEGFAULT|OBJECT <key>]\r\n"));
4186     }
4187 }
4188
4189 #ifdef HAVE_BACKTRACE
4190 static struct redisFunctionSym symsTable[] = {
4191 {"freeStringObject", (unsigned long)freeStringObject},
4192 {"freeListObject", (unsigned long)freeListObject},
4193 {"freeSetObject", (unsigned long)freeSetObject},
4194 {"decrRefCount", (unsigned long)decrRefCount},
4195 {"createObject", (unsigned long)createObject},
4196 {"freeClient", (unsigned long)freeClient},
4197 {"rdbLoad", (unsigned long)rdbLoad},
4198 {"addReply", (unsigned long)addReply},
4199 {"addReplySds", (unsigned long)addReplySds},
4200 {"incrRefCount", (unsigned long)incrRefCount},
4201 {"rdbSaveBackground", (unsigned long)rdbSaveBackground},
4202 {"createStringObject", (unsigned long)createStringObject},
4203 {"replicationFeedSlaves", (unsigned long)replicationFeedSlaves},
4204 {"syncWithMaster", (unsigned long)syncWithMaster},
4205 {"tryObjectSharing", (unsigned long)tryObjectSharing},
4206 {"removeExpire", (unsigned long)removeExpire},
4207 {"expireIfNeeded", (unsigned long)expireIfNeeded},
4208 {"deleteIfVolatile", (unsigned long)deleteIfVolatile},
4209 {"deleteKey", (unsigned long)deleteKey},
4210 {"getExpire", (unsigned long)getExpire},
4211 {"setExpire", (unsigned long)setExpire},
4212 {"updateSlavesWaitingBgsave", (unsigned long)updateSlavesWaitingBgsave},
4213 {"freeMemoryIfNeeded", (unsigned long)freeMemoryIfNeeded},
4214 {"authCommand", (unsigned long)authCommand},
4215 {"pingCommand", (unsigned long)pingCommand},
4216 {"echoCommand", (unsigned long)echoCommand},
4217 {"setCommand", (unsigned long)setCommand},
4218 {"setnxCommand", (unsigned long)setnxCommand},
4219 {"getCommand", (unsigned long)getCommand},
4220 {"delCommand", (unsigned long)delCommand},
4221 {"existsCommand", (unsigned long)existsCommand},
4222 {"incrCommand", (unsigned long)incrCommand},
4223 {"decrCommand", (unsigned long)decrCommand},
4224 {"incrbyCommand", (unsigned long)incrbyCommand},
4225 {"decrbyCommand", (unsigned long)decrbyCommand},
4226 {"selectCommand", (unsigned long)selectCommand},
4227 {"randomkeyCommand", (unsigned long)randomkeyCommand},
4228 {"keysCommand", (unsigned long)keysCommand},
4229 {"dbsizeCommand", (unsigned long)dbsizeCommand},
4230 {"lastsaveCommand", (unsigned long)lastsaveCommand},
4231 {"saveCommand", (unsigned long)saveCommand},
4232 {"bgsaveCommand", (unsigned long)bgsaveCommand},
4233 {"shutdownCommand", (unsigned long)shutdownCommand},
4234 {"moveCommand", (unsigned long)moveCommand},
4235 {"renameCommand", (unsigned long)renameCommand},
4236 {"renamenxCommand", (unsigned long)renamenxCommand},
4237 {"lpushCommand", (unsigned long)lpushCommand},
4238 {"rpushCommand", (unsigned long)rpushCommand},
4239 {"lpopCommand", (unsigned long)lpopCommand},
4240 {"rpopCommand", (unsigned long)rpopCommand},
4241 {"llenCommand", (unsigned long)llenCommand},
4242 {"lindexCommand", (unsigned long)lindexCommand},
4243 {"lrangeCommand", (unsigned long)lrangeCommand},
4244 {"ltrimCommand", (unsigned long)ltrimCommand},
4245 {"typeCommand", (unsigned long)typeCommand},
4246 {"lsetCommand", (unsigned long)lsetCommand},
4247 {"saddCommand", (unsigned long)saddCommand},
4248 {"sremCommand", (unsigned long)sremCommand},
4249 {"smoveCommand", (unsigned long)smoveCommand},
4250 {"sismemberCommand", (unsigned long)sismemberCommand},
4251 {"scardCommand", (unsigned long)scardCommand},
4252 {"spopCommand", (unsigned long)spopCommand},
4253 {"sinterCommand", (unsigned long)sinterCommand},
4254 {"sinterstoreCommand", (unsigned long)sinterstoreCommand},
4255 {"sunionCommand", (unsigned long)sunionCommand},
4256 {"sunionstoreCommand", (unsigned long)sunionstoreCommand},
4257 {"sdiffCommand", (unsigned long)sdiffCommand},
4258 {"sdiffstoreCommand", (unsigned long)sdiffstoreCommand},
4259 {"syncCommand", (unsigned long)syncCommand},
4260 {"flushdbCommand", (unsigned long)flushdbCommand},
4261 {"flushallCommand", (unsigned long)flushallCommand},
4262 {"sortCommand", (unsigned long)sortCommand},
4263 {"lremCommand", (unsigned long)lremCommand},
4264 {"infoCommand", (unsigned long)infoCommand},
4265 {"mgetCommand", (unsigned long)mgetCommand},
4266 {"monitorCommand", (unsigned long)monitorCommand},
4267 {"expireCommand", (unsigned long)expireCommand},
4268 {"getSetCommand", (unsigned long)getSetCommand},
4269 {"ttlCommand", (unsigned long)ttlCommand},
4270 {"slaveofCommand", (unsigned long)slaveofCommand},
4271 {"debugCommand", (unsigned long)debugCommand},
4272 {"processCommand", (unsigned long)processCommand},
4273 {"setupSigSegvAction", (unsigned long)setupSigSegvAction},
4274 {"readQueryFromClient", (unsigned long)readQueryFromClient},
4275 {"rdbRemoveTempFile", (unsigned long)rdbRemoveTempFile},
4276 {NULL,0}
4277 };
4278
4279 /* This function try to convert a pointer into a function name. It's used in
4280  * oreder to provide a backtrace under segmentation fault that's able to
4281  * display functions declared as static (otherwise the backtrace is useless). */
4282 static char *findFuncName(void *pointer, unsigned long *offset){
4283     int i, ret = -1;
4284     unsigned long off, minoff = 0;
4285
4286     /* Try to match against the Symbol with the smallest offset */
4287     for (i=0; symsTable[i].pointer; i++) {
4288         unsigned long lp = (unsigned long) pointer;
4289
4290         if (lp != (unsigned long)-1 && lp >= symsTable[i].pointer) {
4291             off=lp-symsTable[i].pointer;
4292             if (ret < 0 || off < minoff) {
4293                 minoff=off;
4294                 ret=i;
4295             }
4296         }
4297     }
4298     if (ret == -1) return NULL;
4299     *offset = minoff;
4300     return symsTable[ret].name;
4301 }
4302
4303 static void *getMcontextEip(ucontext_t *uc) {
4304 #if defined(__FreeBSD__)
4305     return (void*) uc->uc_mcontext.mc_eip;
4306 #elif defined(__dietlibc__)
4307     return (void*) uc->uc_mcontext.eip;
4308 #elif defined(__APPLE__) && !defined(MAC_OS_X_VERSION_10_6)
4309     return (void*) uc->uc_mcontext->__ss.__eip;
4310 #elif defined(__APPLE__) && defined(MAC_OS_X_VERSION_10_6)
4311   #ifdef _STRUCT_X86_THREAD_STATE64
4312     return (void*) uc->uc_mcontext->__ss.__rip;
4313   #else
4314     return (void*) uc->uc_mcontext->__ss.__eip;
4315   #endif
4316 #elif defined(__i386__) || defined(__X86_64__) /* Linux x86 */
4317     return (void*) uc->uc_mcontext.gregs[REG_EIP];
4318 #elif defined(__ia64__) /* Linux IA64 */
4319     return (void*) uc->uc_mcontext.sc_ip;
4320 #else
4321     return NULL;
4322 #endif
4323 }
4324
4325 static void segvHandler(int sig, siginfo_t *info, void *secret) {
4326     void *trace[100];
4327     char **messages = NULL;
4328     int i, trace_size = 0;
4329     unsigned long offset=0;
4330     time_t uptime = time(NULL)-server.stat_starttime;
4331     ucontext_t *uc = (ucontext_t*) secret;
4332     REDIS_NOTUSED(info);
4333
4334     redisLog(REDIS_WARNING,
4335         "======= Ooops! Redis %s got signal: -%d- =======", REDIS_VERSION, sig);
4336     redisLog(REDIS_WARNING, "%s", sdscatprintf(sdsempty(),
4337         "redis_version:%s; "
4338         "uptime_in_seconds:%d; "
4339         "connected_clients:%d; "
4340         "connected_slaves:%d; "
4341         "used_memory:%zu; "
4342         "changes_since_last_save:%lld; "
4343         "bgsave_in_progress:%d; "
4344         "last_save_time:%d; "
4345         "total_connections_received:%lld; "
4346         "total_commands_processed:%lld; "
4347         "role:%s;"
4348         ,REDIS_VERSION,
4349         uptime,
4350         listLength(server.clients)-listLength(server.slaves),
4351         listLength(server.slaves),
4352         server.usedmemory,
4353         server.dirty,
4354         server.bgsaveinprogress,
4355         server.lastsave,
4356         server.stat_numconnections,
4357         server.stat_numcommands,
4358         server.masterhost == NULL ? "master" : "slave"
4359     ));
4360
4361     trace_size = backtrace(trace, 100);
4362     /* overwrite sigaction with caller's address */
4363     if (getMcontextEip(uc) != NULL) {
4364         trace[1] = getMcontextEip(uc);
4365     }
4366     messages = backtrace_symbols(trace, trace_size);
4367
4368     for (i=1; i<trace_size; ++i) {
4369         char *fn = findFuncName(trace[i], &offset), *p;
4370
4371         p = strchr(messages[i],'+');
4372         if (!fn || (p && ((unsigned long)strtol(p+1,NULL,10)) < offset)) {
4373             redisLog(REDIS_WARNING,"%s", messages[i]);
4374         } else {
4375             redisLog(REDIS_WARNING,"%d redis-server %p %s + %d", i, trace[i], fn, (unsigned int)offset);
4376         }
4377     }
4378     free(messages);
4379     exit(0);
4380 }
4381
4382 static void setupSigSegvAction(void) {
4383     struct sigaction act;
4384
4385     sigemptyset (&act.sa_mask);
4386     /* When the SA_SIGINFO flag is set in sa_flags then sa_sigaction
4387      * is used. Otherwise, sa_handler is used */
4388     act.sa_flags = SA_NODEFER | SA_ONSTACK | SA_RESETHAND | SA_SIGINFO;
4389     act.sa_sigaction = segvHandler;
4390     sigaction (SIGSEGV, &act, NULL);
4391     sigaction (SIGBUS, &act, NULL);
4392     sigaction (SIGFPE, &act, NULL);
4393     sigaction (SIGILL, &act, NULL);
4394     sigaction (SIGBUS, &act, NULL);
4395     return;
4396 }
4397 #else /* HAVE_BACKTRACE */
4398 static void setupSigSegvAction(void) {
4399 }
4400 #endif /* HAVE_BACKTRACE */
4401
4402 /* =================================== Main! ================================ */
4403
4404 #ifdef __linux__
4405 int linuxOvercommitMemoryValue(void) {
4406     FILE *fp = fopen("/proc/sys/vm/overcommit_memory","r");
4407     char buf[64];
4408
4409     if (!fp) return -1;
4410     if (fgets(buf,64,fp) == NULL) {
4411         fclose(fp);
4412         return -1;
4413     }
4414     fclose(fp);
4415
4416     return atoi(buf);
4417 }
4418
4419 void linuxOvercommitMemoryWarning(void) {
4420     if (linuxOvercommitMemoryValue() == 0) {
4421         redisLog(REDIS_WARNING,"WARNING overcommit_memory is set to 0! Background save may fail under low condition memory. 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.");
4422     }
4423 }
4424 #endif /* __linux__ */
4425
4426 static void daemonize(void) {
4427     int fd;
4428     FILE *fp;
4429
4430     if (fork() != 0) exit(0); /* parent exits */
4431     setsid(); /* create a new session */
4432
4433     /* Every output goes to /dev/null. If Redis is daemonized but
4434      * the 'logfile' is set to 'stdout' in the configuration file
4435      * it will not log at all. */
4436     if ((fd = open("/dev/null", O_RDWR, 0)) != -1) {
4437         dup2(fd, STDIN_FILENO);
4438         dup2(fd, STDOUT_FILENO);
4439         dup2(fd, STDERR_FILENO);
4440         if (fd > STDERR_FILENO) close(fd);
4441     }
4442     /* Try to write the pid file */
4443     fp = fopen(server.pidfile,"w");
4444     if (fp) {
4445         fprintf(fp,"%d\n",getpid());
4446         fclose(fp);
4447     }
4448 }
4449
4450 int main(int argc, char **argv) {
4451     initServerConfig();
4452     if (argc == 2) {
4453         ResetServerSaveParams();
4454         loadServerConfig(argv[1]);
4455     } else if (argc > 2) {
4456         fprintf(stderr,"Usage: ./redis-server [/path/to/redis.conf]\n");
4457         exit(1);
4458     } else {
4459         redisLog(REDIS_WARNING,"Warning: no config file specified, using the default config. In order to specify a config file use 'redis-server /path/to/redis.conf'");
4460     }
4461     initServer();
4462     if (server.daemonize) daemonize();
4463     redisLog(REDIS_NOTICE,"Server started, Redis version " REDIS_VERSION);
4464 #ifdef __linux__
4465     linuxOvercommitMemoryWarning();
4466 #endif
4467     if (rdbLoad(server.dbfilename) == REDIS_OK)
4468         redisLog(REDIS_NOTICE,"DB loaded from disk");
4469     if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
4470         acceptHandler, NULL, NULL) == AE_ERR) oom("creating file event");
4471     redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
4472     aeMain(server.el);
4473     aeDeleteEventLoop(server.el);
4474     return 0;
4475 }