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