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