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