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