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