redis.c

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