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.08"
  31
  32 #include <stdio.h>
  33 #include <stdlib.h>
  34 #include <string.h>
  35 #include <time.h>
  36 #include <unistd.h>
  37 #include <signal.h>
  38 #include <sys/wait.h>
  39 #include <errno.h>
  40 #include <assert.h>
  41 #include <ctype.h>
  42 #include <stdarg.h>
  43 #include <inttypes.h>
  44 #include <arpa/inet.h>
  45 #include <sys/stat.h>
  46 #include <fcntl.h>
  47 #include <sys/time.h>
  48 #include <sys/resource.h>
  49 #include <limits.h>
  50
  51 #include "ae.h"     /* Event driven programming library */
  52 #include "sds.h"    /* Dynamic safe strings */
  53 #include "anet.h"   /* Networking the easy way */
  54 #include "dict.h"   /* Hash tables */
  55 #include "adlist.h" /* Linked lists */
  56 #include "zmalloc.h" /* total memory usage aware version of malloc/free */
  57
  58 /* Error codes */
  59 #define REDIS_OK                0
  60 #define REDIS_ERR               -1
  61
  62 /* Static server configuration */
  63 #define REDIS_SERVERPORT        6379    /* TCP port */
  64 #define REDIS_MAXIDLETIME       (60*5)  /* default client timeout */
  65 #define REDIS_QUERYBUF_LEN      1024
  66 #define REDIS_LOADBUF_LEN       1024
  67 #define REDIS_MAX_ARGS          16
  68 #define REDIS_DEFAULT_DBNUM     16
  69 #define REDIS_CONFIGLINE_MAX    1024
  70 #define REDIS_OBJFREELIST_MAX   1000000 /* Max number of objects to cache */
  71 #define REDIS_MAX_SYNC_TIME     60      /* Slave can't take more to sync */
  72
  73 /* Hash table parameters */
  74 #define REDIS_HT_MINFILL        10      /* Minimal hash table fill 10% */
  75 #define REDIS_HT_MINSLOTS       16384   /* Never resize the HT under this */
  76
  77 /* Command flags */
  78 #define REDIS_CMD_BULK          1
  79 #define REDIS_CMD_INLINE        2
  80
  81 /* Object types */
  82 #define REDIS_STRING 0
  83 #define REDIS_LIST 1
  84 #define REDIS_SET 2
  85 #define REDIS_HASH 3
  86
  87 /* Object types only used for dumping to disk */
  88 #define REDIS_SELECTDB 254
  89 #define REDIS_EOF 255
  90
  91 /* Defines related to the dump file format. To store 32 bits lengths for short
  92  * keys requires a lot of space, so we check the most significant 2 bits of
  93  * the first byte to interpreter the length:
  94  *
  95  * 00|000000 => if the two MSB are 00 the len is the 6 bits of this byte
  96  * 01|000000 00000000 =>  01, the len is 14 byes, 6 bits + 8 bits of next byte
  97  * 10|000000 [32 bit integer] => if it's 01, a full 32 bit len will follow
  98  * 11|000000 this means: specially encoded object will follow. The six bits
  99  *           number specify the kind of object that follows.
 100  *           See the REDIS_RDB_ENC_* defines.
 101  *
 102  * Lenghts up to 63 are stored using a single byte, most DB keys, and may
 103  * values, will fit inside. */
 104 #define REDIS_RDB_6BITLEN 0
 105 #define REDIS_RDB_14BITLEN 1
 106 #define REDIS_RDB_32BITLEN 2
 107 #define REDIS_RDB_64BITLEN 3
 108 #define REDIS_RDB_LENERR UINT_MAX
 109
 110 /* When a length of a string object stored on disk has the first two bits
 111  * set, the remaining two bits specify a special encoding for the object
 112  * accordingly to the following defines: */
 113 #define REDIS_RDB_ENC_INT8 0        /* 8 bit signed integer */
 114 #define REDIS_RDB_ENC_INT16 1       /* 16 bit signed integer */
 115 #define REDIS_RDB_ENC_INT32 2       /* 32 bit signed integer */
 116 #define REDIS_RDB_ENC_FLZ 3         /* string compressed with FASTLZ */
 117
 118 /* Client flags */
 119 #define REDIS_CLOSE 1       /* This client connection should be closed ASAP */
 120 #define REDIS_SLAVE 2       /* This client is a slave server */
 121 #define REDIS_MASTER 4      /* This client is a master server */
 122 #define REDIS_MONITOR 8      /* This client is a slave monitor, see MONITOR */
 123
 124 /* Server replication state */
 125 #define REDIS_REPL_NONE 0   /* No active replication */
 126 #define REDIS_REPL_CONNECT 1    /* Must connect to master */
 127 #define REDIS_REPL_CONNECTED 2  /* Connected to master */
 128
 129 /* List related stuff */
 130 #define REDIS_HEAD 0
 131 #define REDIS_TAIL 1
 132
 133 /* Sort operations */
 134 #define REDIS_SORT_GET 0
 135 #define REDIS_SORT_DEL 1
 136 #define REDIS_SORT_INCR 2
 137 #define REDIS_SORT_DECR 3
 138 #define REDIS_SORT_ASC 4
 139 #define REDIS_SORT_DESC 5
 140 #define REDIS_SORTKEY_MAX 1024
 141
 142 /* Log levels */
 143 #define REDIS_DEBUG 0
 144 #define REDIS_NOTICE 1
 145 #define REDIS_WARNING 2
 146
 147 /* Anti-warning macro... */
 148 #define REDIS_NOTUSED(V) ((void) V)
 149
 150 /*================================= Data types ============================== */
 151
 152 /* A redis object, that is a type able to hold a string / list / set */
 153 typedef struct redisObject {
 154     int type;
 155     void *ptr;
 156     int refcount;
 157 } robj;
 158
 159 /* With multiplexing we need to take per-clinet state.
 160  * Clients are taken in a liked list. */
 161 typedef struct redisClient {
 162     int fd;
 163     dict *dict;
 164     int dictid;
 165     sds querybuf;
 166     robj *argv[REDIS_MAX_ARGS];
 167     int argc;
 168     int bulklen;    /* bulk read len. -1 if not in bulk read mode */
 169     list *reply;
 170     int sentlen;
 171     time_t lastinteraction; /* time of the last interaction, used for timeout */
 172     int flags; /* REDIS_CLOSE | REDIS_SLAVE | REDIS_MONITOR */
 173     int slaveseldb; /* slave selected db, if this client is a slave */
 174     int authenticated;    /* when requirepass is non-NULL */
 175 } redisClient;
 176
 177 struct saveparam {
 178     time_t seconds;
 179     int changes;
 180 };
 181
 182 /* Global server state structure */
 183 struct redisServer {
 184     int port;
 185     int fd;
 186     dict **dict;
 187     dict *sharingpool;
 188     unsigned int sharingpoolsize;
 189     long long dirty;            /* changes to DB from the last save */
 190     list *clients;
 191     list *slaves, *monitors;
 192     char neterr[ANET_ERR_LEN];
 193     aeEventLoop *el;
 194     int cronloops;              /* number of times the cron function run */
 195     list *objfreelist;          /* A list of freed objects to avoid malloc() */
 196     time_t lastsave;            /* Unix time of last save succeeede */
 197     int usedmemory;             /* Used memory in megabytes */
 198     /* Fields used only for stats */
 199     time_t stat_starttime;         /* server start time */
 200     long long stat_numcommands;    /* number of processed commands */
 201     long long stat_numconnections; /* number of connections received */
 202     /* Configuration */
 203     int verbosity;
 204     int glueoutputbuf;
 205     int maxidletime;
 206     int dbnum;
 207     int daemonize;
 208     char *pidfile;
 209     int bgsaveinprogress;
 210     struct saveparam *saveparams;
 211     int saveparamslen;
 212     char *logfile;
 213     char *bindaddr;
 214     char *dbfilename;
 215     char *requirepass;
 216     int shareobjects;
 217     /* Replication related */
 218     int isslave;
 219     char *masterhost;
 220     int masterport;
 221     redisClient *master;
 222     int replstate;
 223     /* Sort parameters - qsort_r() is only available under BSD so we
 224      * have to take this state global, in order to pass it to sortCompare() */
 225     int sort_desc;
 226     int sort_alpha;
 227     int sort_bypattern;
 228 };
 229
 230 typedef void redisCommandProc(redisClient *c);
 231 struct redisCommand {
 232     char *name;
 233     redisCommandProc *proc;
 234     int arity;
 235     int flags;
 236 };
 237
 238 typedef struct _redisSortObject {
 239     robj *obj;
 240     union {
 241         double score;
 242         robj *cmpobj;
 243     } u;
 244 } redisSortObject;
 245
 246 typedef struct _redisSortOperation {
 247     int type;
 248     robj *pattern;
 249 } redisSortOperation;
 250
 251 struct sharedObjectsStruct {
 252     robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *pong, *space,
 253     *colon, *nullbulk, *nullmultibulk,
 254     *emptymultibulk, *wrongtypeerr, *nokeyerr, *syntaxerr, *sameobjecterr,
 255     *outofrangeerr, *plus,
 256     *select0, *select1, *select2, *select3, *select4,
 257     *select5, *select6, *select7, *select8, *select9;
 258 } shared;
 259
 260 /*================================ Prototypes =============================== */
 261
 262 static void freeStringObject(robj *o);
 263 static void freeListObject(robj *o);
 264 static void freeSetObject(robj *o);
 265 static void decrRefCount(void *o);
 266 static robj *createObject(int type, void *ptr);
 267 static void freeClient(redisClient *c);
 268 static int rdbLoad(char *filename);
 269 static void addReply(redisClient *c, robj *obj);
 270 static void addReplySds(redisClient *c, sds s);
 271 static void incrRefCount(robj *o);
 272 static int rdbSaveBackground(char *filename);
 273 static robj *createStringObject(char *ptr, size_t len);
 274 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc);
 275 static int syncWithMaster(void);
 276 static robj *tryObjectSharing(robj *o);
 277
 278 static void authCommand(redisClient *c);
 279 static void pingCommand(redisClient *c);
 280 static void echoCommand(redisClient *c);
 281 static void setCommand(redisClient *c);
 282 static void setnxCommand(redisClient *c);
 283 static void getCommand(redisClient *c);
 284 static void delCommand(redisClient *c);
 285 static void existsCommand(redisClient *c);
 286 static void incrCommand(redisClient *c);
 287 static void decrCommand(redisClient *c);
 288 static void incrbyCommand(redisClient *c);
 289 static void decrbyCommand(redisClient *c);
 290 static void selectCommand(redisClient *c);
 291 static void randomkeyCommand(redisClient *c);
 292 static void keysCommand(redisClient *c);
 293 static void dbsizeCommand(redisClient *c);
 294 static void lastsaveCommand(redisClient *c);
 295 static void saveCommand(redisClient *c);
 296 static void bgsaveCommand(redisClient *c);
 297 static void shutdownCommand(redisClient *c);
 298 static void moveCommand(redisClient *c);
 299 static void renameCommand(redisClient *c);
 300 static void renamenxCommand(redisClient *c);
 301 static void lpushCommand(redisClient *c);
 302 static void rpushCommand(redisClient *c);
 303 static void lpopCommand(redisClient *c);
 304 static void rpopCommand(redisClient *c);
 305 static void llenCommand(redisClient *c);
 306 static void lindexCommand(redisClient *c);
 307 static void lrangeCommand(redisClient *c);
 308 static void ltrimCommand(redisClient *c);
 309 static void typeCommand(redisClient *c);
 310 static void lsetCommand(redisClient *c);
 311 static void saddCommand(redisClient *c);
 312 static void sremCommand(redisClient *c);
 313 static void sismemberCommand(redisClient *c);
 314 static void scardCommand(redisClient *c);
 315 static void sinterCommand(redisClient *c);
 316 static void sinterstoreCommand(redisClient *c);
 317 static void syncCommand(redisClient *c);
 318 static void flushdbCommand(redisClient *c);
 319 static void flushallCommand(redisClient *c);
 320 static void sortCommand(redisClient *c);
 321 static void lremCommand(redisClient *c);
 322 static void infoCommand(redisClient *c);
 323 static void mgetCommand(redisClient *c);
 324 static void monitorCommand(redisClient *c);
 325
 326 /*================================= Globals ================================= */
 327
 328 /* Global vars */
 329 static struct redisServer server; /* server global state */
 330 static struct redisCommand cmdTable[] = {
 331     {"get",getCommand,2,REDIS_CMD_INLINE},
 332     {"set",setCommand,3,REDIS_CMD_BULK},
 333     {"setnx",setnxCommand,3,REDIS_CMD_BULK},
 334     {"del",delCommand,2,REDIS_CMD_INLINE},
 335     {"exists",existsCommand,2,REDIS_CMD_INLINE},
 336     {"incr",incrCommand,2,REDIS_CMD_INLINE},
 337     {"decr",decrCommand,2,REDIS_CMD_INLINE},
 338     {"mget",mgetCommand,-2,REDIS_CMD_INLINE},
 339     {"rpush",rpushCommand,3,REDIS_CMD_BULK},
 340     {"lpush",lpushCommand,3,REDIS_CMD_BULK},
 341     {"rpop",rpopCommand,2,REDIS_CMD_INLINE},
 342     {"lpop",lpopCommand,2,REDIS_CMD_INLINE},
 343     {"llen",llenCommand,2,REDIS_CMD_INLINE},
 344     {"lindex",lindexCommand,3,REDIS_CMD_INLINE},
 345     {"lset",lsetCommand,4,REDIS_CMD_BULK},
 346     {"lrange",lrangeCommand,4,REDIS_CMD_INLINE},
 347     {"ltrim",ltrimCommand,4,REDIS_CMD_INLINE},
 348     {"lrem",lremCommand,4,REDIS_CMD_BULK},
 349     {"sadd",saddCommand,3,REDIS_CMD_BULK},
 350     {"srem",sremCommand,3,REDIS_CMD_BULK},
 351     {"sismember",sismemberCommand,3,REDIS_CMD_BULK},
 352     {"scard",scardCommand,2,REDIS_CMD_INLINE},
 353     {"sinter",sinterCommand,-2,REDIS_CMD_INLINE},
 354     {"sinterstore",sinterstoreCommand,-3,REDIS_CMD_INLINE},
 355     {"smembers",sinterCommand,2,REDIS_CMD_INLINE},
 356     {"incrby",incrbyCommand,3,REDIS_CMD_INLINE},
 357     {"decrby",decrbyCommand,3,REDIS_CMD_INLINE},
 358     {"randomkey",randomkeyCommand,1,REDIS_CMD_INLINE},
 359     {"select",selectCommand,2,REDIS_CMD_INLINE},
 360     {"move",moveCommand,3,REDIS_CMD_INLINE},
 361     {"rename",renameCommand,3,REDIS_CMD_INLINE},
 362     {"renamenx",renamenxCommand,3,REDIS_CMD_INLINE},
 363     {"keys",keysCommand,2,REDIS_CMD_INLINE},
 364     {"dbsize",dbsizeCommand,1,REDIS_CMD_INLINE},
 365     {"auth",authCommand,2,REDIS_CMD_INLINE},
 366     {"ping",pingCommand,1,REDIS_CMD_INLINE},
 367     {"echo",echoCommand,2,REDIS_CMD_BULK},
 368     {"save",saveCommand,1,REDIS_CMD_INLINE},
 369     {"bgsave",bgsaveCommand,1,REDIS_CMD_INLINE},
 370     {"shutdown",shutdownCommand,1,REDIS_CMD_INLINE},
 371     {"lastsave",lastsaveCommand,1,REDIS_CMD_INLINE},
 372     {"type",typeCommand,2,REDIS_CMD_INLINE},
 373     {"sync",syncCommand,1,REDIS_CMD_INLINE},
 374     {"flushdb",flushdbCommand,1,REDIS_CMD_INLINE},
 375     {"flushall",flushallCommand,1,REDIS_CMD_INLINE},
 376     {"sort",sortCommand,-2,REDIS_CMD_INLINE},
 377     {"info",infoCommand,1,REDIS_CMD_INLINE},
 378     {"monitor",monitorCommand,1,REDIS_CMD_INLINE},
 379     {NULL,NULL,0,0}
 380 };
 381
 382 /*============================ Utility functions ============================ */
 383
 384 /* Glob-style pattern matching. */
 385 int stringmatchlen(const char *pattern, int patternLen,
 386         const char *string, int stringLen, int nocase)
 387 {
 388     while(patternLen) {
 389         switch(pattern[0]) {
 390         case '*':
 391             while (pattern[1] == '*') {
 392                 pattern++;
 393                 patternLen--;
 394             }
 395             if (patternLen == 1)
 396                 return 1; /* match */
 397             while(stringLen) {
 398                 if (stringmatchlen(pattern+1, patternLen-1,
 399                             string, stringLen, nocase))
 400                     return 1; /* match */
 401                 string++;
 402                 stringLen--;
 403             }
 404             return 0; /* no match */
 405             break;
 406         case '?':
 407             if (stringLen == 0)
 408                 return 0; /* no match */
 409             string++;
 410             stringLen--;
 411             break;
 412         case '[':
 413         {
 414             int not, match;
 415
 416             pattern++;
 417             patternLen--;
 418             not = pattern[0] == '^';
 419             if (not) {
 420                 pattern++;
 421                 patternLen--;
 422             }
 423             match = 0;
 424             while(1) {
 425                 if (pattern[0] == '\\') {
 426                     pattern++;
 427                     patternLen--;
 428                     if (pattern[0] == string[0])
 429                         match = 1;
 430                 } else if (pattern[0] == ']') {
 431                     break;
 432                 } else if (patternLen == 0) {
 433                     pattern--;
 434                     patternLen++;
 435                     break;
 436                 } else if (pattern[1] == '-' && patternLen >= 3) {
 437                     int start = pattern[0];
 438                     int end = pattern[2];
 439                     int c = string[0];
 440                     if (start > end) {
 441                         int t = start;
 442                         start = end;
 443                         end = t;
 444                     }
 445                     if (nocase) {
 446                         start = tolower(start);
 447                         end = tolower(end);
 448                         c = tolower(c);
 449                     }
 450                     pattern += 2;
 451                     patternLen -= 2;
 452                     if (c >= start && c <= end)
 453                         match = 1;
 454                 } else {
 455                     if (!nocase) {
 456                         if (pattern[0] == string[0])
 457                             match = 1;
 458                     } else {
 459                         if (tolower((int)pattern[0]) == tolower((int)string[0]))
 460                             match = 1;
 461                     }
 462                 }
 463                 pattern++;
 464                 patternLen--;
 465             }
 466             if (not)
 467                 match = !match;
 468             if (!match)
 469                 return 0; /* no match */
 470             string++;
 471             stringLen--;
 472             break;
 473         }
 474         case '\\':
 475             if (patternLen >= 2) {
 476                 pattern++;
 477                 patternLen--;
 478             }
 479             /* fall through */
 480         default:
 481             if (!nocase) {
 482                 if (pattern[0] != string[0])
 483                     return 0; /* no match */
 484             } else {
 485                 if (tolower((int)pattern[0]) != tolower((int)string[0]))
 486                     return 0; /* no match */
 487             }
 488             string++;
 489             stringLen--;
 490             break;
 491         }
 492         pattern++;
 493         patternLen--;
 494         if (stringLen == 0) {
 495             while(*pattern == '*') {
 496                 pattern++;
 497                 patternLen--;
 498             }
 499             break;
 500         }
 501     }
 502     if (patternLen == 0 && stringLen == 0)
 503         return 1;
 504     return 0;
 505 }
 506
 507 void redisLog(int level, const char *fmt, ...)
 508 {
 509     va_list ap;
 510     FILE *fp;
 511
 512     fp = (server.logfile == NULL) ? stdout : fopen(server.logfile,"a");
 513     if (!fp) return;
 514
 515     va_start(ap, fmt);
 516     if (level >= server.verbosity) {
 517         char *c = ".-*";
 518         fprintf(fp,"%c ",c[level]);
 519         vfprintf(fp, fmt, ap);
 520         fprintf(fp,"\n");
 521         fflush(fp);
 522     }
 523     va_end(ap);
 524
 525     if (server.logfile) fclose(fp);
 526 }
 527
 528 /*====================== Hash table type implementation  ==================== */
 529
 530 /* This is an hash table type that uses the SDS dynamic strings libary as
 531  * keys and radis objects as values (objects can hold SDS strings,
 532  * lists, sets). */
 533
 534 static int sdsDictKeyCompare(void *privdata, const void *key1,
 535         const void *key2)
 536 {
 537     int l1,l2;
 538     DICT_NOTUSED(privdata);
 539
 540     l1 = sdslen((sds)key1);
 541     l2 = sdslen((sds)key2);
 542     if (l1 != l2) return 0;
 543     return memcmp(key1, key2, l1) == 0;
 544 }
 545
 546 static void dictRedisObjectDestructor(void *privdata, void *val)
 547 {
 548     DICT_NOTUSED(privdata);
 549
 550     decrRefCount(val);
 551 }
 552
 553 static int dictSdsKeyCompare(void *privdata, const void *key1,
 554         const void *key2)
 555 {
 556     const robj *o1 = key1, *o2 = key2;
 557     return sdsDictKeyCompare(privdata,o1->ptr,o2->ptr);
 558 }
 559
 560 static unsigned int dictSdsHash(const void *key) {
 561     const robj *o = key;
 562     return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
 563 }
 564
 565 static dictType setDictType = {
 566     dictSdsHash,               /* hash function */
 567     NULL,                      /* key dup */
 568     NULL,                      /* val dup */
 569     dictSdsKeyCompare,         /* key compare */
 570     dictRedisObjectDestructor, /* key destructor */
 571     NULL                       /* val destructor */
 572 };
 573
 574 static dictType hashDictType = {
 575     dictSdsHash,                /* hash function */
 576     NULL,                       /* key dup */
 577     NULL,                       /* val dup */
 578     dictSdsKeyCompare,          /* key compare */
 579     dictRedisObjectDestructor,  /* key destructor */
 580     dictRedisObjectDestructor   /* val destructor */
 581 };
 582
 583 /* ========================= Random utility functions ======================= */
 584
 585 /* Redis generally does not try to recover from out of memory conditions
 586  * when allocating objects or strings, it is not clear if it will be possible
 587  * to report this condition to the client since the networking layer itself
 588  * is based on heap allocation for send buffers, so we simply abort.
 589  * At least the code will be simpler to read... */
 590 static void oom(const char *msg) {
 591     fprintf(stderr, "%s: Out of memory\n",msg);
 592     fflush(stderr);
 593     sleep(1);
 594     abort();
 595 }
 596
 597 /* ====================== Redis server networking stuff ===================== */
 598 void closeTimedoutClients(void) {
 599     redisClient *c;
 600     listIter *li;
 601     listNode *ln;
 602     time_t now = time(NULL);
 603
 604     li = listGetIterator(server.clients,AL_START_HEAD);
 605     if (!li) return;
 606     while ((ln = listNextElement(li)) != NULL) {
 607         c = listNodeValue(ln);
 608         if (!(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
 609              (now - c->lastinteraction > server.maxidletime)) {
 610             redisLog(REDIS_DEBUG,"Closing idle client");
 611             freeClient(c);
 612         }
 613     }
 614     listReleaseIterator(li);
 615 }
 616
 617 int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
 618     int j, size, used, loops = server.cronloops++;
 619     REDIS_NOTUSED(eventLoop);
 620     REDIS_NOTUSED(id);
 621     REDIS_NOTUSED(clientData);
 622
 623     /* Update the global state with the amount of used memory */
 624     server.usedmemory = zmalloc_used_memory();
 625
 626     /* If the percentage of used slots in the HT reaches REDIS_HT_MINFILL
 627      * we resize the hash table to save memory */
 628     for (j = 0; j < server.dbnum; j++) {
 629         size = dictGetHashTableSize(server.dict[j]);
 630         used = dictGetHashTableUsed(server.dict[j]);
 631         if (!(loops % 5) && used > 0) {
 632             redisLog(REDIS_DEBUG,"DB %d: %d keys in %d slots HT.",j,used,size);
 633             /* dictPrintStats(server.dict); */
 634         }
 635         if (size && used && size > REDIS_HT_MINSLOTS &&
 636             (used*100/size < REDIS_HT_MINFILL)) {
 637             redisLog(REDIS_NOTICE,"The hash table %d is too sparse, resize it...",j);
 638             dictResize(server.dict[j]);
 639             redisLog(REDIS_NOTICE,"Hash table %d resized.",j);
 640         }
 641     }
 642
 643     /* Show information about connected clients */
 644     if (!(loops % 5)) {
 645         redisLog(REDIS_DEBUG,"%d clients connected (%d slaves), %d bytes in use",
 646             listLength(server.clients)-listLength(server.slaves),
 647             listLength(server.slaves),
 648             server.usedmemory,
 649             dictGetHashTableUsed(server.sharingpool));
 650     }
 651
 652     /* Close connections of timedout clients */
 653     if (!(loops % 10))
 654         closeTimedoutClients();
 655
 656     /* Check if a background saving in progress terminated */
 657     if (server.bgsaveinprogress) {
 658         int statloc;
 659         if (wait4(-1,&statloc,WNOHANG,NULL)) {
 660             int exitcode = WEXITSTATUS(statloc);
 661             if (exitcode == 0) {
 662                 redisLog(REDIS_NOTICE,
 663                     "Background saving terminated with success");
 664                 server.dirty = 0;
 665                 server.lastsave = time(NULL);
 666             } else {
 667                 redisLog(REDIS_WARNING,
 668                     "Background saving error");
 669             }
 670             server.bgsaveinprogress = 0;
 671         }
 672     } else {
 673         /* If there is not a background saving in progress check if
 674          * we have to save now */
 675          time_t now = time(NULL);
 676          for (j = 0; j < server.saveparamslen; j++) {
 677             struct saveparam *sp = server.saveparams+j;
 678
 679             if (server.dirty >= sp->changes &&
 680                 now-server.lastsave > sp->seconds) {
 681                 redisLog(REDIS_NOTICE,"%d changes in %d seconds. Saving...",
 682                     sp->changes, sp->seconds);
 683                 rdbSaveBackground(server.dbfilename);
 684                 break;
 685             }
 686          }
 687     }
 688     /* Check if we should connect to a MASTER */
 689     if (server.replstate == REDIS_REPL_CONNECT) {
 690         redisLog(REDIS_NOTICE,"Connecting to MASTER...");
 691         if (syncWithMaster() == REDIS_OK) {
 692             redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync succeeded");
 693         }
 694     }
 695     return 1000;
 696 }
 697
 698 static void createSharedObjects(void) {
 699     shared.crlf = createObject(REDIS_STRING,sdsnew("\r\n"));
 700     shared.ok = createObject(REDIS_STRING,sdsnew("+OK\r\n"));
 701     shared.err = createObject(REDIS_STRING,sdsnew("-ERR\r\n"));
 702     shared.emptybulk = createObject(REDIS_STRING,sdsnew("$0\r\n\r\n"));
 703     shared.czero = createObject(REDIS_STRING,sdsnew(":0\r\n"));
 704     shared.cone = createObject(REDIS_STRING,sdsnew(":1\r\n"));
 705     shared.nullbulk = createObject(REDIS_STRING,sdsnew("$-1\r\n"));
 706     shared.nullmultibulk = createObject(REDIS_STRING,sdsnew("*-1\r\n"));
 707     shared.emptymultibulk = createObject(REDIS_STRING,sdsnew("*0\r\n"));
 708     /* no such key */
 709     shared.pong = createObject(REDIS_STRING,sdsnew("+PONG\r\n"));
 710     shared.wrongtypeerr = createObject(REDIS_STRING,sdsnew(
 711         "-ERR Operation against a key holding the wrong kind of value\r\n"));
 712     shared.nokeyerr = createObject(REDIS_STRING,sdsnew(
 713         "-ERR no such key\r\n"));
 714     shared.syntaxerr = createObject(REDIS_STRING,sdsnew(
 715         "-ERR syntax error\r\n"));
 716     shared.sameobjecterr = createObject(REDIS_STRING,sdsnew(
 717         "-ERR source and destination objects are the same\r\n"));
 718     shared.outofrangeerr = createObject(REDIS_STRING,sdsnew(
 719         "-ERR index out of range\r\n"));
 720     shared.space = createObject(REDIS_STRING,sdsnew(" "));
 721     shared.colon = createObject(REDIS_STRING,sdsnew(":"));
 722     shared.plus = createObject(REDIS_STRING,sdsnew("+"));
 723     shared.select0 = createStringObject("select 0\r\n",10);
 724     shared.select1 = createStringObject("select 1\r\n",10);
 725     shared.select2 = createStringObject("select 2\r\n",10);
 726     shared.select3 = createStringObject("select 3\r\n",10);
 727     shared.select4 = createStringObject("select 4\r\n",10);
 728     shared.select5 = createStringObject("select 5\r\n",10);
 729     shared.select6 = createStringObject("select 6\r\n",10);
 730     shared.select7 = createStringObject("select 7\r\n",10);
 731     shared.select8 = createStringObject("select 8\r\n",10);
 732     shared.select9 = createStringObject("select 9\r\n",10);
 733 }
 734
 735 static void appendServerSaveParams(time_t seconds, int changes) {
 736     server.saveparams = zrealloc(server.saveparams,sizeof(struct saveparam)*(server.saveparamslen+1));
 737     if (server.saveparams == NULL) oom("appendServerSaveParams");
 738     server.saveparams[server.saveparamslen].seconds = seconds;
 739     server.saveparams[server.saveparamslen].changes = changes;
 740     server.saveparamslen++;
 741 }
 742
 743 static void ResetServerSaveParams() {
 744     zfree(server.saveparams);
 745     server.saveparams = NULL;
 746     server.saveparamslen = 0;
 747 }
 748
 749 static void initServerConfig() {
 750     server.dbnum = REDIS_DEFAULT_DBNUM;
 751     server.port = REDIS_SERVERPORT;
 752     server.verbosity = REDIS_DEBUG;
 753     server.maxidletime = REDIS_MAXIDLETIME;
 754     server.saveparams = NULL;
 755     server.logfile = NULL; /* NULL = log on standard output */
 756     server.bindaddr = NULL;
 757     server.glueoutputbuf = 1;
 758     server.daemonize = 0;
 759     server.pidfile = "/var/run/redis.pid";
 760     server.dbfilename = "dump.rdb";
 761     server.requirepass = NULL;
 762     server.shareobjects = 0;
 763     ResetServerSaveParams();
 764
 765     appendServerSaveParams(60*60,1);  /* save after 1 hour and 1 change */
 766     appendServerSaveParams(300,100);  /* save after 5 minutes and 100 changes */
 767     appendServerSaveParams(60,10000); /* save after 1 minute and 10000 changes */
 768     /* Replication related */
 769     server.isslave = 0;
 770     server.masterhost = NULL;
 771     server.masterport = 6379;
 772     server.master = NULL;
 773     server.replstate = REDIS_REPL_NONE;
 774 }
 775
 776 static void initServer() {
 777     int j;
 778
 779     signal(SIGHUP, SIG_IGN);
 780     signal(SIGPIPE, SIG_IGN);
 781
 782     server.clients = listCreate();
 783     server.slaves = listCreate();
 784     server.monitors = listCreate();
 785     server.objfreelist = listCreate();
 786     createSharedObjects();
 787     server.el = aeCreateEventLoop();
 788     server.dict = zmalloc(sizeof(dict*)*server.dbnum);
 789     server.sharingpool = dictCreate(&setDictType,NULL);
 790     server.sharingpoolsize = 1024;
 791     if (!server.dict || !server.clients || !server.slaves || !server.monitors || !server.el || !server.objfreelist)
 792         oom("server initialization"); /* Fatal OOM */
 793     server.fd = anetTcpServer(server.neterr, server.port, server.bindaddr);
 794     if (server.fd == -1) {
 795         redisLog(REDIS_WARNING, "Opening TCP port: %s", server.neterr);
 796         exit(1);
 797     }
 798     for (j = 0; j < server.dbnum; j++)
 799         server.dict[j] = dictCreate(&hashDictType,NULL);
 800     server.cronloops = 0;
 801     server.bgsaveinprogress = 0;
 802     server.lastsave = time(NULL);
 803     server.dirty = 0;
 804     server.usedmemory = 0;
 805     server.stat_numcommands = 0;
 806     server.stat_numconnections = 0;
 807     server.stat_starttime = time(NULL);
 808     aeCreateTimeEvent(server.el, 1000, serverCron, NULL, NULL);
 809 }
 810
 811 /* Empty the whole database */
 812 static void emptyDb() {
 813     int j;
 814
 815     for (j = 0; j < server.dbnum; j++)
 816         dictEmpty(server.dict[j]);
 817 }
 818
 819 /* I agree, this is a very rudimental way to load a configuration...
 820    will improve later if the config gets more complex */
 821 static void loadServerConfig(char *filename) {
 822     FILE *fp = fopen(filename,"r");
 823     char buf[REDIS_CONFIGLINE_MAX+1], *err = NULL;
 824     int linenum = 0;
 825     sds line = NULL;
 826
 827     if (!fp) {
 828         redisLog(REDIS_WARNING,"Fatal error, can't open config file");
 829         exit(1);
 830     }
 831     while(fgets(buf,REDIS_CONFIGLINE_MAX+1,fp) != NULL) {
 832         sds *argv;
 833         int argc, j;
 834
 835         linenum++;
 836         line = sdsnew(buf);
 837         line = sdstrim(line," \t\r\n");
 838
 839         /* Skip comments and blank lines*/
 840         if (line[0] == '#' || line[0] == '\0') {
 841             sdsfree(line);
 842             continue;
 843         }
 844
 845         /* Split into arguments */
 846         argv = sdssplitlen(line,sdslen(line)," ",1,&argc);
 847         sdstolower(argv[0]);
 848
 849         /* Execute config directives */
 850         if (!strcmp(argv[0],"timeout") && argc == 2) {
 851             server.maxidletime = atoi(argv[1]);
 852             if (server.maxidletime < 1) {
 853                 err = "Invalid timeout value"; goto loaderr;
 854             }
 855         } else if (!strcmp(argv[0],"port") && argc == 2) {
 856             server.port = atoi(argv[1]);
 857             if (server.port < 1 || server.port > 65535) {
 858                 err = "Invalid port"; goto loaderr;
 859             }
 860         } else if (!strcmp(argv[0],"bind") && argc == 2) {
 861             server.bindaddr = zstrdup(argv[1]);
 862         } else if (!strcmp(argv[0],"save") && argc == 3) {
 863             int seconds = atoi(argv[1]);
 864             int changes = atoi(argv[2]);
 865             if (seconds < 1 || changes < 0) {
 866                 err = "Invalid save parameters"; goto loaderr;
 867             }
 868             appendServerSaveParams(seconds,changes);
 869         } else if (!strcmp(argv[0],"dir") && argc == 2) {
 870             if (chdir(argv[1]) == -1) {
 871                 redisLog(REDIS_WARNING,"Can't chdir to '%s': %s",
 872                     argv[1], strerror(errno));
 873                 exit(1);
 874             }
 875         } else if (!strcmp(argv[0],"loglevel") && argc == 2) {
 876             if (!strcmp(argv[1],"debug")) server.verbosity = REDIS_DEBUG;
 877             else if (!strcmp(argv[1],"notice")) server.verbosity = REDIS_NOTICE;
 878             else if (!strcmp(argv[1],"warning")) server.verbosity = REDIS_WARNING;
 879             else {
 880                 err = "Invalid log level. Must be one of debug, notice, warning";
 881                 goto loaderr;
 882             }
 883         } else if (!strcmp(argv[0],"logfile") && argc == 2) {
 884             FILE *fp;
 885
 886             server.logfile = zstrdup(argv[1]);
 887             if (!strcmp(server.logfile,"stdout")) {
 888                 zfree(server.logfile);
 889                 server.logfile = NULL;
 890             }
 891             if (server.logfile) {
 892                 /* Test if we are able to open the file. The server will not
 893                  * be able to abort just for this problem later... */
 894                 fp = fopen(server.logfile,"a");
 895                 if (fp == NULL) {
 896                     err = sdscatprintf(sdsempty(),
 897                         "Can't open the log file: %s", strerror(errno));
 898                     goto loaderr;
 899                 }
 900                 fclose(fp);
 901             }
 902         } else if (!strcmp(argv[0],"databases") && argc == 2) {
 903             server.dbnum = atoi(argv[1]);
 904             if (server.dbnum < 1) {
 905                 err = "Invalid number of databases"; goto loaderr;
 906             }
 907         } else if (!strcmp(argv[0],"slaveof") && argc == 3) {
 908             server.masterhost = sdsnew(argv[1]);
 909             server.masterport = atoi(argv[2]);
 910             server.replstate = REDIS_REPL_CONNECT;
 911         } else if (!strcmp(argv[0],"glueoutputbuf") && argc == 2) {
 912             sdstolower(argv[1]);
 913             if (!strcmp(argv[1],"yes")) server.glueoutputbuf = 1;
 914             else if (!strcmp(argv[1],"no")) server.glueoutputbuf = 0;
 915             else {
 916                 err = "argument must be 'yes' or 'no'"; goto loaderr;
 917             }
 918         } else if (!strcmp(argv[0],"shareobjects") && argc == 2) {
 919             sdstolower(argv[1]);
 920             if (!strcmp(argv[1],"yes")) server.shareobjects = 1;
 921             else if (!strcmp(argv[1],"no")) server.shareobjects = 0;
 922             else {
 923                 err = "argument must be 'yes' or 'no'"; goto loaderr;
 924             }
 925         } else if (!strcmp(argv[0],"daemonize") && argc == 2) {
 926             sdstolower(argv[1]);
 927             if (!strcmp(argv[1],"yes")) server.daemonize = 1;
 928             else if (!strcmp(argv[1],"no")) server.daemonize = 0;
 929             else {
 930                 err = "argument must be 'yes' or 'no'"; goto loaderr;
 931             }
 932         } else if (!strcmp(argv[0],"requirepass") && argc == 2) {
 933           server.requirepass = zstrdup(argv[1]);
 934         } else if (!strcmp(argv[0],"pidfile") && argc == 2) {
 935           server.pidfile = zstrdup(argv[1]);
 936         } else {
 937             err = "Bad directive or wrong number of arguments"; goto loaderr;
 938         }
 939         for (j = 0; j < argc; j++)
 940             sdsfree(argv[j]);
 941         zfree(argv);
 942         sdsfree(line);
 943     }
 944     fclose(fp);
 945     return;
 946
 947 loaderr:
 948     fprintf(stderr, "\n*** FATAL CONFIG FILE ERROR ***\n");
 949     fprintf(stderr, "Reading the configuration file, at line %d\n", linenum);
 950     fprintf(stderr, ">>> '%s'\n", line);
 951     fprintf(stderr, "%s\n", err);
 952     exit(1);
 953 }
 954
 955 static void freeClientArgv(redisClient *c) {
 956     int j;
 957
 958     for (j = 0; j < c->argc; j++)
 959         decrRefCount(c->argv[j]);
 960     c->argc = 0;
 961 }
 962
 963 static void freeClient(redisClient *c) {
 964     listNode *ln;
 965
 966     aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
 967     aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
 968     sdsfree(c->querybuf);
 969     listRelease(c->reply);
 970     freeClientArgv(c);
 971     close(c->fd);
 972     ln = listSearchKey(server.clients,c);
 973     assert(ln != NULL);
 974     listDelNode(server.clients,ln);
 975     if (c->flags & REDIS_SLAVE) {
 976         list *l = (c->flags & REDIS_MONITOR) ? server.monitors : server.slaves;
 977         ln = listSearchKey(l,c);
 978         assert(ln != NULL);
 979         listDelNode(l,ln);
 980     }
 981     if (c->flags & REDIS_MASTER) {
 982         server.master = NULL;
 983         server.replstate = REDIS_REPL_CONNECT;
 984     }
 985     zfree(c);
 986 }
 987
 988 static void glueReplyBuffersIfNeeded(redisClient *c) {
 989     int totlen = 0;
 990     listNode *ln = c->reply->head, *next;
 991     robj *o;
 992
 993     while(ln) {
 994         o = ln->value;
 995         totlen += sdslen(o->ptr);
 996         ln = ln->next;
 997         /* This optimization makes more sense if we don't have to copy
 998          * too much data */
 999         if (totlen > 1024) return;
1000     }
1001     if (totlen > 0) {
1002         char buf[1024];
1003         int copylen = 0;
1004
1005         ln = c->reply->head;
1006         while(ln) {
1007             next = ln->next;
1008             o = ln->value;
1009             memcpy(buf+copylen,o->ptr,sdslen(o->ptr));
1010             copylen += sdslen(o->ptr);
1011             listDelNode(c->reply,ln);
1012             ln = next;
1013         }
1014         /* Now the output buffer is empty, add the new single element */
1015         addReplySds(c,sdsnewlen(buf,totlen));
1016     }
1017 }
1018
1019 static void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
1020     redisClient *c = privdata;
1021     int nwritten = 0, totwritten = 0, objlen;
1022     robj *o;
1023     REDIS_NOTUSED(el);
1024     REDIS_NOTUSED(mask);
1025
1026     if (server.glueoutputbuf && listLength(c->reply) > 1)
1027         glueReplyBuffersIfNeeded(c);
1028     while(listLength(c->reply)) {
1029         o = listNodeValue(listFirst(c->reply));
1030         objlen = sdslen(o->ptr);
1031
1032         if (objlen == 0) {
1033             listDelNode(c->reply,listFirst(c->reply));
1034             continue;
1035         }
1036
1037         if (c->flags & REDIS_MASTER) {
1038             nwritten = objlen - c->sentlen;
1039         } else {
1040             nwritten = write(fd, ((char*)o->ptr)+c->sentlen, objlen - c->sentlen);
1041             if (nwritten <= 0) break;
1042         }
1043         c->sentlen += nwritten;
1044         totwritten += nwritten;
1045         /* If we fully sent the object on head go to the next one */
1046         if (c->sentlen == objlen) {
1047             listDelNode(c->reply,listFirst(c->reply));
1048             c->sentlen = 0;
1049         }
1050     }
1051     if (nwritten == -1) {
1052         if (errno == EAGAIN) {
1053             nwritten = 0;
1054         } else {
1055             redisLog(REDIS_DEBUG,
1056                 "Error writing to client: %s", strerror(errno));
1057             freeClient(c);
1058             return;
1059         }
1060     }
1061     if (totwritten > 0) c->lastinteraction = time(NULL);
1062     if (listLength(c->reply) == 0) {
1063         c->sentlen = 0;
1064         aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
1065     }
1066 }
1067
1068 static struct redisCommand *lookupCommand(char *name) {
1069     int j = 0;
1070     while(cmdTable[j].name != NULL) {
1071         if (!strcmp(name,cmdTable[j].name)) return &cmdTable[j];
1072         j++;
1073     }
1074     return NULL;
1075 }
1076
1077 /* resetClient prepare the client to process the next command */
1078 static void resetClient(redisClient *c) {
1079     freeClientArgv(c);
1080     c->bulklen = -1;
1081 }
1082
1083 /* If this function gets called we already read a whole
1084  * command, argments are in the client argv/argc fields.
1085  * processCommand() execute the command or prepare the
1086  * server for a bulk read from the client.
1087  *
1088  * If 1 is returned the client is still alive and valid and
1089  * and other operations can be performed by the caller. Otherwise
1090  * if 0 is returned the client was destroied (i.e. after QUIT). */
1091 static int processCommand(redisClient *c) {
1092     struct redisCommand *cmd;
1093     long long dirty;
1094
1095     sdstolower(c->argv[0]->ptr);
1096     /* The QUIT command is handled as a special case. Normal command
1097      * procs are unable to close the client connection safely */
1098     if (!strcmp(c->argv[0]->ptr,"quit")) {
1099         freeClient(c);
1100         return 0;
1101     }
1102     cmd = lookupCommand(c->argv[0]->ptr);
1103     if (!cmd) {
1104         addReplySds(c,sdsnew("-ERR unknown command\r\n"));
1105         resetClient(c);
1106         return 1;
1107     } else if ((cmd->arity > 0 && cmd->arity != c->argc) ||
1108                (c->argc < -cmd->arity)) {
1109         addReplySds(c,sdsnew("-ERR wrong number of arguments\r\n"));
1110         resetClient(c);
1111         return 1;
1112     } else if (cmd->flags & REDIS_CMD_BULK && c->bulklen == -1) {
1113         int bulklen = atoi(c->argv[c->argc-1]->ptr);
1114
1115         decrRefCount(c->argv[c->argc-1]);
1116         if (bulklen < 0 || bulklen > 1024*1024*1024) {
1117             c->argc--;
1118             addReplySds(c,sdsnew("-ERR invalid bulk write count\r\n"));
1119             resetClient(c);
1120             return 1;
1121         }
1122         c->argc--;
1123         c->bulklen = bulklen+2; /* add two bytes for CR+LF */
1124         /* It is possible that the bulk read is already in the
1125          * buffer. Check this condition and handle it accordingly */
1126         if ((signed)sdslen(c->querybuf) >= c->bulklen) {
1127             c->argv[c->argc] = createStringObject(c->querybuf,c->bulklen-2);
1128             c->argc++;
1129             c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
1130         } else {
1131             return 1;
1132         }
1133     }
1134     /* Let's try to share objects on the command arguments vector */
1135     if (server.shareobjects) {
1136         int j;
1137         for(j = 1; j < c->argc; j++)
1138             c->argv[j] = tryObjectSharing(c->argv[j]);
1139     }
1140     /* Check if the user is authenticated */
1141     if (server.requirepass && !c->authenticated && cmd->proc != authCommand) {
1142         addReplySds(c,sdsnew("-ERR operation not permitted\r\n"));
1143         resetClient(c);
1144         return 1;
1145     }
1146
1147     /* Exec the command */
1148     dirty = server.dirty;
1149     cmd->proc(c);
1150     if (server.dirty-dirty != 0 && listLength(server.slaves))
1151         replicationFeedSlaves(server.slaves,cmd,c->dictid,c->argv,c->argc);
1152     if (listLength(server.monitors))
1153         replicationFeedSlaves(server.monitors,cmd,c->dictid,c->argv,c->argc);
1154     server.stat_numcommands++;
1155
1156     /* Prepare the client for the next command */
1157     if (c->flags & REDIS_CLOSE) {
1158         freeClient(c);
1159         return 0;
1160     }
1161     resetClient(c);
1162     return 1;
1163 }
1164
1165 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc) {
1166     listNode *ln = slaves->head;
1167     robj *outv[REDIS_MAX_ARGS*4]; /* enough room for args, spaces, newlines */
1168     int outc = 0, j;
1169
1170     for (j = 0; j < argc; j++) {
1171         if (j != 0) outv[outc++] = shared.space;
1172         if ((cmd->flags & REDIS_CMD_BULK) && j == argc-1) {
1173             robj *lenobj;
1174
1175             lenobj = createObject(REDIS_STRING,
1176                 sdscatprintf(sdsempty(),"%d\r\n",sdslen(argv[j]->ptr)));
1177             lenobj->refcount = 0;
1178             outv[outc++] = lenobj;
1179         }
1180         outv[outc++] = argv[j];
1181     }
1182     outv[outc++] = shared.crlf;
1183
1184     while(ln) {
1185         redisClient *slave = ln->value;
1186         if (slave->slaveseldb != dictid) {
1187             robj *selectcmd;
1188
1189             switch(dictid) {
1190             case 0: selectcmd = shared.select0; break;
1191             case 1: selectcmd = shared.select1; break;
1192             case 2: selectcmd = shared.select2; break;
1193             case 3: selectcmd = shared.select3; break;
1194             case 4: selectcmd = shared.select4; break;
1195             case 5: selectcmd = shared.select5; break;
1196             case 6: selectcmd = shared.select6; break;
1197             case 7: selectcmd = shared.select7; break;
1198             case 8: selectcmd = shared.select8; break;
1199             case 9: selectcmd = shared.select9; break;
1200             default:
1201                 selectcmd = createObject(REDIS_STRING,
1202                     sdscatprintf(sdsempty(),"select %d\r\n",dictid));
1203                 selectcmd->refcount = 0;
1204                 break;
1205             }
1206             addReply(slave,selectcmd);
1207             slave->slaveseldb = dictid;
1208         }
1209         for (j = 0; j < outc; j++) addReply(slave,outv[j]);
1210         ln = ln->next;
1211     }
1212 }
1213
1214 static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
1215     redisClient *c = (redisClient*) privdata;
1216     char buf[REDIS_QUERYBUF_LEN];
1217     int nread;
1218     REDIS_NOTUSED(el);
1219     REDIS_NOTUSED(mask);
1220
1221     nread = read(fd, buf, REDIS_QUERYBUF_LEN);
1222     if (nread == -1) {
1223         if (errno == EAGAIN) {
1224             nread = 0;
1225         } else {
1226             redisLog(REDIS_DEBUG, "Reading from client: %s",strerror(errno));
1227             freeClient(c);
1228             return;
1229         }
1230     } else if (nread == 0) {
1231         redisLog(REDIS_DEBUG, "Client closed connection");
1232         freeClient(c);
1233         return;
1234     }
1235     if (nread) {
1236         c->querybuf = sdscatlen(c->querybuf, buf, nread);
1237         c->lastinteraction = time(NULL);
1238     } else {
1239         return;
1240     }
1241
1242 again:
1243     if (c->bulklen == -1) {
1244         /* Read the first line of the query */
1245         char *p = strchr(c->querybuf,'\n');
1246         size_t querylen;
1247         if (p) {
1248             sds query, *argv;
1249             int argc, j;
1250
1251             query = c->querybuf;
1252             c->querybuf = sdsempty();
1253             querylen = 1+(p-(query));
1254             if (sdslen(query) > querylen) {
1255                 /* leave data after the first line of the query in the buffer */
1256                 c->querybuf = sdscatlen(c->querybuf,query+querylen,sdslen(query)-querylen);
1257             }
1258             *p = '\0'; /* remove "\n" */
1259             if (*(p-1) == '\r') *(p-1) = '\0'; /* and "\r" if any */
1260             sdsupdatelen(query);
1261
1262             /* Now we can split the query in arguments */
1263             if (sdslen(query) == 0) {
1264                 /* Ignore empty query */
1265                 sdsfree(query);
1266                 return;
1267             }
1268             argv = sdssplitlen(query,sdslen(query)," ",1,&argc);
1269             sdsfree(query);
1270             if (argv == NULL) oom("sdssplitlen");
1271             for (j = 0; j < argc && j < REDIS_MAX_ARGS; j++) {
1272                 if (sdslen(argv[j])) {
1273                     c->argv[c->argc] = createObject(REDIS_STRING,argv[j]);
1274                     c->argc++;
1275                 } else {
1276                     sdsfree(argv[j]);
1277                 }
1278             }
1279             zfree(argv);
1280             /* Execute the command. If the client is still valid
1281              * after processCommand() return and there is something
1282              * on the query buffer try to process the next command. */
1283             if (processCommand(c) && sdslen(c->querybuf)) goto again;
1284             return;
1285         } else if (sdslen(c->querybuf) >= 1024) {
1286             redisLog(REDIS_DEBUG, "Client protocol error");
1287             freeClient(c);
1288             return;
1289         }
1290     } else {
1291         /* Bulk read handling. Note that if we are at this point
1292            the client already sent a command terminated with a newline,
1293            we are reading the bulk data that is actually the last
1294            argument of the command. */
1295         int qbl = sdslen(c->querybuf);
1296
1297         if (c->bulklen <= qbl) {
1298             /* Copy everything but the final CRLF as final argument */
1299             c->argv[c->argc] = createStringObject(c->querybuf,c->bulklen-2);
1300             c->argc++;
1301             c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
1302             processCommand(c);
1303             return;
1304         }
1305     }
1306 }
1307
1308 static int selectDb(redisClient *c, int id) {
1309     if (id < 0 || id >= server.dbnum)
1310         return REDIS_ERR;
1311     c->dict = server.dict[id];
1312     c->dictid = id;
1313     return REDIS_OK;
1314 }
1315
1316 static redisClient *createClient(int fd) {
1317     redisClient *c = zmalloc(sizeof(*c));
1318
1319     anetNonBlock(NULL,fd);
1320     anetTcpNoDelay(NULL,fd);
1321     if (!c) return NULL;
1322     selectDb(c,0);
1323     c->fd = fd;
1324     c->querybuf = sdsempty();
1325     c->argc = 0;
1326     c->bulklen = -1;
1327     c->sentlen = 0;
1328     c->flags = 0;
1329     c->lastinteraction = time(NULL);
1330     c->authenticated = 0;
1331     if ((c->reply = listCreate()) == NULL) oom("listCreate");
1332     listSetFreeMethod(c->reply,decrRefCount);
1333     if (aeCreateFileEvent(server.el, c->fd, AE_READABLE,
1334         readQueryFromClient, c, NULL) == AE_ERR) {
1335         freeClient(c);
1336         return NULL;
1337     }
1338     if (!listAddNodeTail(server.clients,c)) oom("listAddNodeTail");
1339     return c;
1340 }
1341
1342 static void addReply(redisClient *c, robj *obj) {
1343     if (listLength(c->reply) == 0 &&
1344         aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
1345         sendReplyToClient, c, NULL) == AE_ERR) return;
1346     if (!listAddNodeTail(c->reply,obj)) oom("listAddNodeTail");
1347     incrRefCount(obj);
1348 }
1349
1350 static void addReplySds(redisClient *c, sds s) {
1351     robj *o = createObject(REDIS_STRING,s);
1352     addReply(c,o);
1353     decrRefCount(o);
1354 }
1355
1356 static void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
1357     int cport, cfd;
1358     char cip[128];
1359     REDIS_NOTUSED(el);
1360     REDIS_NOTUSED(mask);
1361     REDIS_NOTUSED(privdata);
1362
1363     cfd = anetAccept(server.neterr, fd, cip, &cport);
1364     if (cfd == AE_ERR) {
1365         redisLog(REDIS_DEBUG,"Accepting client connection: %s", server.neterr);
1366         return;
1367     }
1368     redisLog(REDIS_DEBUG,"Accepted %s:%d", cip, cport);
1369     if (createClient(cfd) == NULL) {
1370         redisLog(REDIS_WARNING,"Error allocating resoures for the client");
1371         close(cfd); /* May be already closed, just ingore errors */
1372         return;
1373     }
1374     server.stat_numconnections++;
1375 }
1376
1377 /* ======================= Redis objects implementation ===================== */
1378
1379 static robj *createObject(int type, void *ptr) {
1380     robj *o;
1381
1382     if (listLength(server.objfreelist)) {
1383         listNode *head = listFirst(server.objfreelist);
1384         o = listNodeValue(head);
1385         listDelNode(server.objfreelist,head);
1386     } else {
1387         o = zmalloc(sizeof(*o));
1388     }
1389     if (!o) oom("createObject");
1390     o->type = type;
1391     o->ptr = ptr;
1392     o->refcount = 1;
1393     return o;
1394 }
1395
1396 static robj *createStringObject(char *ptr, size_t len) {
1397     return createObject(REDIS_STRING,sdsnewlen(ptr,len));
1398 }
1399
1400 static robj *createListObject(void) {
1401     list *l = listCreate();
1402
1403     if (!l) oom("listCreate");
1404     listSetFreeMethod(l,decrRefCount);
1405     return createObject(REDIS_LIST,l);
1406 }
1407
1408 static robj *createSetObject(void) {
1409     dict *d = dictCreate(&setDictType,NULL);
1410     if (!d) oom("dictCreate");
1411     return createObject(REDIS_SET,d);
1412 }
1413
1414 #if 0
1415 static robj *createHashObject(void) {
1416     dict *d = dictCreate(&hashDictType,NULL);
1417     if (!d) oom("dictCreate");
1418     return createObject(REDIS_SET,d);
1419 }
1420 #endif
1421
1422 static void freeStringObject(robj *o) {
1423     sdsfree(o->ptr);
1424 }
1425
1426 static void freeListObject(robj *o) {
1427     listRelease((list*) o->ptr);
1428 }
1429
1430 static void freeSetObject(robj *o) {
1431     dictRelease((dict*) o->ptr);
1432 }
1433
1434 static void freeHashObject(robj *o) {
1435     dictRelease((dict*) o->ptr);
1436 }
1437
1438 static void incrRefCount(robj *o) {
1439     o->refcount++;
1440 }
1441
1442 static void decrRefCount(void *obj) {
1443     robj *o = obj;
1444     if (--(o->refcount) == 0) {
1445         switch(o->type) {
1446         case REDIS_STRING: freeStringObject(o); break;
1447         case REDIS_LIST: freeListObject(o); break;
1448         case REDIS_SET: freeSetObject(o); break;
1449         case REDIS_HASH: freeHashObject(o); break;
1450         default: assert(0 != 0); break;
1451         }
1452         if (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX ||
1453             !listAddNodeHead(server.objfreelist,o))
1454             zfree(o);
1455     }
1456 }
1457
1458 /* Try to share an object against the shared objects pool */
1459 static robj *tryObjectSharing(robj *o) {
1460     struct dictEntry *de;
1461     unsigned long c;
1462
1463     if (server.shareobjects == 0) return o;
1464
1465     assert(o->type == REDIS_STRING);
1466     de = dictFind(server.sharingpool,o);
1467     if (de) {
1468         robj *shared = dictGetEntryKey(de);
1469
1470         c = ((unsigned long) dictGetEntryVal(de))+1;
1471         dictGetEntryVal(de) = (void*) c;
1472         incrRefCount(shared);
1473         decrRefCount(o);
1474         return shared;
1475     } else {
1476         /* Here we are using a stream algorihtm: Every time an object is
1477          * shared we increment its count, everytime there is a miss we
1478          * recrement the counter of a random object. If this object reaches
1479          * zero we remove the object and put the current object instead. */
1480         if (dictGetHashTableUsed(server.sharingpool) >=
1481                 server.sharingpoolsize) {
1482             de = dictGetRandomKey(server.sharingpool);
1483             assert(de != NULL);
1484             c = ((unsigned long) dictGetEntryVal(de))-1;
1485             dictGetEntryVal(de) = (void*) c;
1486             if (c == 0) {
1487                 dictDelete(server.sharingpool,de->key);
1488             }
1489         } else {
1490             c = 0; /* If the pool is empty we want to add this object */
1491         }
1492         if (c == 0) {
1493             int retval;
1494
1495             retval = dictAdd(server.sharingpool,o,(void*)1);
1496             assert(retval == DICT_OK);
1497             incrRefCount(o);
1498         }
1499         return o;
1500     }
1501 }
1502
1503 /*============================ DB saving/loading ============================ */
1504
1505 static int rdbSaveType(FILE *fp, unsigned char type) {
1506     if (fwrite(&type,1,1,fp) == 0) return -1;
1507     return 0;
1508 }
1509
1510 static int rdbSaveLen(FILE *fp, uint32_t len) {
1511     unsigned char buf[2];
1512
1513     if (len < (1<<6)) {
1514         /* Save a 6 bit len */
1515         buf[0] = (len&0xFF)|(REDIS_RDB_6BITLEN<<6);
1516         if (fwrite(buf,1,1,fp) == 0) return -1;
1517     } else if (len < (1<<14)) {
1518         /* Save a 14 bit len */
1519         buf[0] = ((len>>8)&0xFF)|(REDIS_RDB_14BITLEN<<6);
1520         buf[1] = len&0xFF;
1521         if (fwrite(buf,4,1,fp) == 0) return -1;
1522     } else {
1523         /* Save a 32 bit len */
1524         buf[0] = (REDIS_RDB_32BITLEN<<6);
1525         if (fwrite(buf,1,1,fp) == 0) return -1;
1526         len = htonl(len);
1527         if (fwrite(&len,4,1,fp) == 0) return -1;
1528     }
1529     return 0;
1530 }
1531
1532 static int rdbSaveStringObject(FILE *fp, robj *obj) {
1533     size_t len = sdslen(obj->ptr);
1534
1535     if (rdbSaveLen(fp,len) == -1) return -1;
1536     if (len && fwrite(obj->ptr,len,1,fp) == 0) return -1;
1537     return 0;
1538 }
1539
1540 /* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success */
1541 static int rdbSave(char *filename) {
1542     dictIterator *di = NULL;
1543     dictEntry *de;
1544     FILE *fp;
1545     char tmpfile[256];
1546     int j;
1547
1548     snprintf(tmpfile,256,"temp-%d.%ld.rdb",(int)time(NULL),(long int)random());
1549     fp = fopen(tmpfile,"w");
1550     if (!fp) {
1551         redisLog(REDIS_WARNING, "Failed saving the DB: %s", strerror(errno));
1552         return REDIS_ERR;
1553     }
1554     if (fwrite("REDIS0001",9,1,fp) == 0) goto werr;
1555     for (j = 0; j < server.dbnum; j++) {
1556         dict *d = server.dict[j];
1557         if (dictGetHashTableUsed(d) == 0) continue;
1558         di = dictGetIterator(d);
1559         if (!di) {
1560             fclose(fp);
1561             return REDIS_ERR;
1562         }
1563
1564         /* Write the SELECT DB opcode */
1565         if (rdbSaveType(fp,REDIS_SELECTDB) == -1) goto werr;
1566         if (rdbSaveLen(fp,j) == -1) goto werr;
1567
1568         /* Iterate this DB writing every entry */
1569         while((de = dictNext(di)) != NULL) {
1570             robj *key = dictGetEntryKey(de);
1571             robj *o = dictGetEntryVal(de);
1572
1573             if (rdbSaveType(fp,o->type) == -1) goto werr;
1574             if (rdbSaveStringObject(fp,key) == -1) goto werr;
1575             if (o->type == REDIS_STRING) {
1576                 /* Save a string value */
1577                 if (rdbSaveStringObject(fp,o) == -1) goto werr;
1578             } else if (o->type == REDIS_LIST) {
1579                 /* Save a list value */
1580                 list *list = o->ptr;
1581                 listNode *ln = list->head;
1582
1583                 if (rdbSaveLen(fp,listLength(list)) == -1) goto werr;
1584                 while(ln) {
1585                     robj *eleobj = listNodeValue(ln);
1586
1587                     if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
1588                     ln = ln->next;
1589                 }
1590             } else if (o->type == REDIS_SET) {
1591                 /* Save a set value */
1592                 dict *set = o->ptr;
1593                 dictIterator *di = dictGetIterator(set);
1594                 dictEntry *de;
1595
1596                 if (!set) oom("dictGetIteraotr");
1597                 if (rdbSaveLen(fp,dictGetHashTableUsed(set)) == -1) goto werr;
1598                 while((de = dictNext(di)) != NULL) {
1599                     robj *eleobj = dictGetEntryKey(de);
1600
1601                     if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
1602                 }
1603                 dictReleaseIterator(di);
1604             } else {
1605                 assert(0 != 0);
1606             }
1607         }
1608         dictReleaseIterator(di);
1609     }
1610     /* EOF opcode */
1611     if (rdbSaveType(fp,REDIS_EOF) == -1) goto werr;
1612
1613     /* Make sure data will not remain on the OS's output buffers */
1614     fflush(fp);
1615     fsync(fileno(fp));
1616     fclose(fp);
1617
1618     /* Use RENAME to make sure the DB file is changed atomically only
1619      * if the generate DB file is ok. */
1620     if (rename(tmpfile,filename) == -1) {
1621         redisLog(REDIS_WARNING,"Error moving temp DB file on the final destionation: %s", strerror(errno));
1622         unlink(tmpfile);
1623         return REDIS_ERR;
1624     }
1625     redisLog(REDIS_NOTICE,"DB saved on disk");
1626     server.dirty = 0;
1627     server.lastsave = time(NULL);
1628     return REDIS_OK;
1629
1630 werr:
1631     fclose(fp);
1632     unlink(tmpfile);
1633     redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
1634     if (di) dictReleaseIterator(di);
1635     return REDIS_ERR;
1636 }
1637
1638 static int rdbSaveBackground(char *filename) {
1639     pid_t childpid;
1640
1641     if (server.bgsaveinprogress) return REDIS_ERR;
1642     if ((childpid = fork()) == 0) {
1643         /* Child */
1644         close(server.fd);
1645         if (rdbSave(filename) == REDIS_OK) {
1646             exit(0);
1647         } else {
1648             exit(1);
1649         }
1650     } else {
1651         /* Parent */
1652         redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
1653         server.bgsaveinprogress = 1;
1654         return REDIS_OK;
1655     }
1656     return REDIS_OK; /* unreached */
1657 }
1658
1659 static int rdbLoadType(FILE *fp) {
1660     unsigned char type;
1661     if (fread(&type,1,1,fp) == 0) return -1;
1662     return type;
1663 }
1664
1665 static uint32_t rdbLoadLen(FILE *fp, int rdbver) {
1666     unsigned char buf[2];
1667     uint32_t len;
1668
1669     if (rdbver == 0) {
1670         if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
1671         return ntohl(len);
1672     } else {
1673         if (fread(buf,1,1,fp) == 0) return REDIS_RDB_LENERR;
1674         if ((buf[0]&0xC0) == REDIS_RDB_6BITLEN) {
1675             /* Read a 6 bit len */
1676             return buf[0];
1677         } else if ((buf[0]&0xC0) == REDIS_RDB_14BITLEN) {
1678             /* Read a 14 bit len */
1679             if (fread(buf+1,1,1,fp) == 0) return REDIS_RDB_LENERR;
1680             return ((buf[0]&0x3F)<<8)|buf[1];
1681         } else {
1682             /* Read a 32 bit len */
1683             if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
1684             return ntohl(len);
1685         }
1686     }
1687 }
1688
1689 static robj *rdbLoadStringObject(FILE*fp,int rdbver) {
1690     uint32_t len = rdbLoadLen(fp,rdbver);
1691     sds val;
1692
1693     if (len == REDIS_RDB_LENERR) return NULL;
1694     val = sdsnewlen(NULL,len);
1695     if (len && fread(val,len,1,fp) == 0) {
1696         sdsfree(val);
1697         return NULL;
1698     }
1699     return tryObjectSharing(createObject(REDIS_STRING,val));
1700 }
1701
1702 static int rdbLoad(char *filename) {
1703     FILE *fp;
1704     robj *keyobj = NULL;
1705     uint32_t dbid;
1706     int type;
1707     int retval;
1708     dict *d = server.dict[0];
1709     char buf[1024];
1710     int rdbver;
1711     fp = fopen(filename,"r");
1712     if (!fp) return REDIS_ERR;
1713     if (fread(buf,9,1,fp) == 0) goto eoferr;
1714     buf[9] = '\0';
1715     if (memcmp(buf,"REDIS",5) != 0) {
1716         fclose(fp);
1717         redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
1718         return REDIS_ERR;
1719     }
1720     rdbver = atoi(buf+5);
1721     if (rdbver > 1) {
1722         fclose(fp);
1723         redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
1724         return REDIS_ERR;
1725     }
1726     while(1) {
1727         robj *o;
1728
1729         /* Read type. */
1730         if ((type = rdbLoadType(fp)) == -1) goto eoferr;
1731         if (type == REDIS_EOF) break;
1732         /* Handle SELECT DB opcode as a special case */
1733         if (type == REDIS_SELECTDB) {
1734             if ((dbid = rdbLoadLen(fp,rdbver)) == REDIS_RDB_LENERR) goto eoferr;
1735             if (dbid >= (unsigned)server.dbnum) {
1736                 redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
1737                 exit(1);
1738             }
1739             d = server.dict[dbid];
1740             continue;
1741         }
1742         /* Read key */
1743         if ((keyobj = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
1744
1745         if (type == REDIS_STRING) {
1746             /* Read string value */
1747             if ((o = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
1748         } else if (type == REDIS_LIST || type == REDIS_SET) {
1749             /* Read list/set value */
1750             uint32_t listlen;
1751
1752             if ((listlen = rdbLoadLen(fp,rdbver)) == REDIS_RDB_LENERR)
1753                 goto eoferr;
1754             o = (type == REDIS_LIST) ? createListObject() : createSetObject();
1755             /* Load every single element of the list/set */
1756             while(listlen--) {
1757                 robj *ele;
1758
1759                 if ((ele = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
1760                 if (type == REDIS_LIST) {
1761                     if (!listAddNodeTail((list*)o->ptr,ele))
1762                         oom("listAddNodeTail");
1763                 } else {
1764                     if (dictAdd((dict*)o->ptr,ele,NULL) == DICT_ERR)
1765                         oom("dictAdd");
1766                 }
1767             }
1768         } else {
1769             assert(0 != 0);
1770         }
1771         /* Add the new object in the hash table */
1772         retval = dictAdd(d,keyobj,o);
1773         if (retval == DICT_ERR) {
1774             redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", keyobj->ptr);
1775             exit(1);
1776         }
1777         keyobj = o = NULL;
1778     }
1779     fclose(fp);
1780     return REDIS_OK;
1781
1782 eoferr: /* unexpected end of file is handled here with a fatal exit */
1783     decrRefCount(keyobj);
1784     redisLog(REDIS_WARNING,"Short read loading DB. Unrecoverable error, exiting now.");
1785     exit(1);
1786     return REDIS_ERR; /* Just to avoid warning */
1787 }
1788
1789 /*================================== Commands =============================== */
1790
1791 static void authCommand(redisClient *c) {
1792     if (!strcmp(c->argv[1]->ptr, server.requirepass)) {
1793       c->authenticated = 1;
1794       addReply(c,shared.ok);
1795     } else {
1796       c->authenticated = 0;
1797       addReply(c,shared.err);
1798     }
1799 }
1800
1801 static void pingCommand(redisClient *c) {
1802     addReply(c,shared.pong);
1803 }
1804
1805 static void echoCommand(redisClient *c) {
1806     addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
1807         (int)sdslen(c->argv[1]->ptr)));
1808     addReply(c,c->argv[1]);
1809     addReply(c,shared.crlf);
1810 }
1811
1812 /*=================================== Strings =============================== */
1813
1814 static void setGenericCommand(redisClient *c, int nx) {
1815     int retval;
1816
1817     retval = dictAdd(c->dict,c->argv[1],c->argv[2]);
1818     if (retval == DICT_ERR) {
1819         if (!nx) {
1820             dictReplace(c->dict,c->argv[1],c->argv[2]);
1821             incrRefCount(c->argv[2]);
1822         } else {
1823             addReply(c,shared.czero);
1824             return;
1825         }
1826     } else {
1827         incrRefCount(c->argv[1]);
1828         incrRefCount(c->argv[2]);
1829     }
1830     server.dirty++;
1831     addReply(c, nx ? shared.cone : shared.ok);
1832 }
1833
1834 static void setCommand(redisClient *c) {
1835     setGenericCommand(c,0);
1836 }
1837
1838 static void setnxCommand(redisClient *c) {
1839     setGenericCommand(c,1);
1840 }
1841
1842 static void getCommand(redisClient *c) {
1843     dictEntry *de;
1844
1845     de = dictFind(c->dict,c->argv[1]);
1846     if (de == NULL) {
1847         addReply(c,shared.nullbulk);
1848     } else {
1849         robj *o = dictGetEntryVal(de);
1850
1851         if (o->type != REDIS_STRING) {
1852             addReply(c,shared.wrongtypeerr);
1853         } else {
1854             addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(o->ptr)));
1855             addReply(c,o);
1856             addReply(c,shared.crlf);
1857         }
1858     }
1859 }
1860
1861 static void mgetCommand(redisClient *c) {
1862     dictEntry *de;
1863     int j;
1864
1865     addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",c->argc-1));
1866     for (j = 1; j < c->argc; j++) {
1867         de = dictFind(c->dict,c->argv[j]);
1868         if (de == NULL) {
1869             addReply(c,shared.nullbulk);
1870         } else {
1871             robj *o = dictGetEntryVal(de);
1872
1873             if (o->type != REDIS_STRING) {
1874                 addReply(c,shared.nullbulk);
1875             } else {
1876                 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(o->ptr)));
1877                 addReply(c,o);
1878                 addReply(c,shared.crlf);
1879             }
1880         }
1881     }
1882 }
1883
1884 static void incrDecrCommand(redisClient *c, int incr) {
1885     dictEntry *de;
1886     long long value;
1887     int retval;
1888     robj *o;
1889
1890     de = dictFind(c->dict,c->argv[1]);
1891     if (de == NULL) {
1892         value = 0;
1893     } else {
1894         robj *o = dictGetEntryVal(de);
1895
1896         if (o->type != REDIS_STRING) {
1897             value = 0;
1898         } else {
1899             char *eptr;
1900
1901             value = strtoll(o->ptr, &eptr, 10);
1902         }
1903     }
1904
1905     value += incr;
1906     o = createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",value));
1907     retval = dictAdd(c->dict,c->argv[1],o);
1908     if (retval == DICT_ERR) {
1909         dictReplace(c->dict,c->argv[1],o);
1910     } else {
1911         incrRefCount(c->argv[1]);
1912     }
1913     server.dirty++;
1914     addReply(c,shared.colon);
1915     addReply(c,o);
1916     addReply(c,shared.crlf);
1917 }
1918
1919 static void incrCommand(redisClient *c) {
1920     incrDecrCommand(c,1);
1921 }
1922
1923 static void decrCommand(redisClient *c) {
1924     incrDecrCommand(c,-1);
1925 }
1926
1927 static void incrbyCommand(redisClient *c) {
1928     int incr = atoi(c->argv[2]->ptr);
1929     incrDecrCommand(c,incr);
1930 }
1931
1932 static void decrbyCommand(redisClient *c) {
1933     int incr = atoi(c->argv[2]->ptr);
1934     incrDecrCommand(c,-incr);
1935 }
1936
1937 /* ========================= Type agnostic commands ========================= */
1938
1939 static void delCommand(redisClient *c) {
1940     if (dictDelete(c->dict,c->argv[1]) == DICT_OK) {
1941         server.dirty++;
1942         addReply(c,shared.cone);
1943     } else {
1944         addReply(c,shared.czero);
1945     }
1946 }
1947
1948 static void existsCommand(redisClient *c) {
1949     dictEntry *de;
1950
1951     de = dictFind(c->dict,c->argv[1]);
1952     if (de == NULL)
1953         addReply(c,shared.czero);
1954     else
1955         addReply(c,shared.cone);
1956 }
1957
1958 static void selectCommand(redisClient *c) {
1959     int id = atoi(c->argv[1]->ptr);
1960
1961     if (selectDb(c,id) == REDIS_ERR) {
1962         addReplySds(c,"-ERR invalid DB index\r\n");
1963     } else {
1964         addReply(c,shared.ok);
1965     }
1966 }
1967
1968 static void randomkeyCommand(redisClient *c) {
1969     dictEntry *de;
1970
1971     de = dictGetRandomKey(c->dict);
1972     if (de == NULL) {
1973         addReply(c,shared.crlf);
1974     } else {
1975         addReply(c,shared.plus);
1976         addReply(c,dictGetEntryKey(de));
1977         addReply(c,shared.crlf);
1978     }
1979 }
1980
1981 static void keysCommand(redisClient *c) {
1982     dictIterator *di;
1983     dictEntry *de;
1984     sds pattern = c->argv[1]->ptr;
1985     int plen = sdslen(pattern);
1986     int numkeys = 0, keyslen = 0;
1987     robj *lenobj = createObject(REDIS_STRING,NULL);
1988
1989     di = dictGetIterator(c->dict);
1990     if (!di) oom("dictGetIterator");
1991     addReply(c,lenobj);
1992     decrRefCount(lenobj);
1993     while((de = dictNext(di)) != NULL) {
1994         robj *keyobj = dictGetEntryKey(de);
1995         sds key = keyobj->ptr;
1996         if ((pattern[0] == '*' && pattern[1] == '\0') ||
1997             stringmatchlen(pattern,plen,key,sdslen(key),0)) {
1998             if (numkeys != 0)
1999                 addReply(c,shared.space);
2000             addReply(c,keyobj);
2001             numkeys++;
2002             keyslen += sdslen(key);
2003         }
2004     }
2005     dictReleaseIterator(di);
2006     lenobj->ptr = sdscatprintf(sdsempty(),"$%lu\r\n",keyslen+(numkeys ? (numkeys-1) : 0));
2007     addReply(c,shared.crlf);
2008 }
2009
2010 static void dbsizeCommand(redisClient *c) {
2011     addReplySds(c,
2012         sdscatprintf(sdsempty(),":%lu\r\n",dictGetHashTableUsed(c->dict)));
2013 }
2014
2015 static void lastsaveCommand(redisClient *c) {
2016     addReplySds(c,
2017         sdscatprintf(sdsempty(),":%lu\r\n",server.lastsave));
2018 }
2019
2020 static void typeCommand(redisClient *c) {
2021     dictEntry *de;
2022     char *type;
2023
2024     de = dictFind(c->dict,c->argv[1]);
2025     if (de == NULL) {
2026         type = "+none";
2027     } else {
2028         robj *o = dictGetEntryVal(de);
2029
2030         switch(o->type) {
2031         case REDIS_STRING: type = "+string"; break;
2032         case REDIS_LIST: type = "+list"; break;
2033         case REDIS_SET: type = "+set"; break;
2034         default: type = "unknown"; break;
2035         }
2036     }
2037     addReplySds(c,sdsnew(type));
2038     addReply(c,shared.crlf);
2039 }
2040
2041 static void saveCommand(redisClient *c) {
2042     if (rdbSave(server.dbfilename) == REDIS_OK) {
2043         addReply(c,shared.ok);
2044     } else {
2045         addReply(c,shared.err);
2046     }
2047 }
2048
2049 static void bgsaveCommand(redisClient *c) {
2050     if (server.bgsaveinprogress) {
2051         addReplySds(c,sdsnew("-ERR background save already in progress\r\n"));
2052         return;
2053     }
2054     if (rdbSaveBackground(server.dbfilename) == REDIS_OK) {
2055         addReply(c,shared.ok);
2056     } else {
2057         addReply(c,shared.err);
2058     }
2059 }
2060
2061 static void shutdownCommand(redisClient *c) {
2062     redisLog(REDIS_WARNING,"User requested shutdown, saving DB...");
2063     if (rdbSave(server.dbfilename) == REDIS_OK) {
2064         if (server.daemonize) {
2065           unlink(server.pidfile);
2066         }
2067         redisLog(REDIS_WARNING,"Server exit now, bye bye...");
2068         exit(1);
2069     } else {
2070         redisLog(REDIS_WARNING,"Error trying to save the DB, can't exit");
2071         addReplySds(c,sdsnew("-ERR can't quit, problems saving the DB\r\n"));
2072     }
2073 }
2074
2075 static void renameGenericCommand(redisClient *c, int nx) {
2076     dictEntry *de;
2077     robj *o;
2078
2079     /* To use the same key as src and dst is probably an error */
2080     if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) {
2081         addReply(c,shared.sameobjecterr);
2082         return;
2083     }
2084
2085     de = dictFind(c->dict,c->argv[1]);
2086     if (de == NULL) {
2087         addReply(c,shared.nokeyerr);
2088         return;
2089     }
2090     o = dictGetEntryVal(de);
2091     incrRefCount(o);
2092     if (dictAdd(c->dict,c->argv[2],o) == DICT_ERR) {
2093         if (nx) {
2094             decrRefCount(o);
2095             addReply(c,shared.czero);
2096             return;
2097         }
2098         dictReplace(c->dict,c->argv[2],o);
2099     } else {
2100         incrRefCount(c->argv[2]);
2101     }
2102     dictDelete(c->dict,c->argv[1]);
2103     server.dirty++;
2104     addReply(c,nx ? shared.cone : shared.ok);
2105 }
2106
2107 static void renameCommand(redisClient *c) {
2108     renameGenericCommand(c,0);
2109 }
2110
2111 static void renamenxCommand(redisClient *c) {
2112     renameGenericCommand(c,1);
2113 }
2114
2115 static void moveCommand(redisClient *c) {
2116     dictEntry *de;
2117     robj *o, *key;
2118     dict *src, *dst;
2119     int srcid;
2120
2121     /* Obtain source and target DB pointers */
2122     src = c->dict;
2123     srcid = c->dictid;
2124     if (selectDb(c,atoi(c->argv[2]->ptr)) == REDIS_ERR) {
2125         addReply(c,shared.outofrangeerr);
2126         return;
2127     }
2128     dst = c->dict;
2129     c->dict = src;
2130     c->dictid = srcid;
2131
2132     /* If the user is moving using as target the same
2133      * DB as the source DB it is probably an error. */
2134     if (src == dst) {
2135         addReply(c,shared.sameobjecterr);
2136         return;
2137     }
2138
2139     /* Check if the element exists and get a reference */
2140     de = dictFind(c->dict,c->argv[1]);
2141     if (!de) {
2142         addReply(c,shared.czero);
2143         return;
2144     }
2145
2146     /* Try to add the element to the target DB */
2147     key = dictGetEntryKey(de);
2148     o = dictGetEntryVal(de);
2149     if (dictAdd(dst,key,o) == DICT_ERR) {
2150         addReply(c,shared.czero);
2151         return;
2152     }
2153     incrRefCount(key);
2154     incrRefCount(o);
2155
2156     /* OK! key moved, free the entry in the source DB */
2157     dictDelete(src,c->argv[1]);
2158     server.dirty++;
2159     addReply(c,shared.cone);
2160 }
2161
2162 /* =================================== Lists ================================ */
2163 static void pushGenericCommand(redisClient *c, int where) {
2164     robj *lobj;
2165     dictEntry *de;
2166     list *list;
2167
2168     de = dictFind(c->dict,c->argv[1]);
2169     if (de == NULL) {
2170         lobj = createListObject();
2171         list = lobj->ptr;
2172         if (where == REDIS_HEAD) {
2173             if (!listAddNodeHead(list,c->argv[2])) oom("listAddNodeHead");
2174         } else {
2175             if (!listAddNodeTail(list,c->argv[2])) oom("listAddNodeTail");
2176         }
2177         dictAdd(c->dict,c->argv[1],lobj);
2178         incrRefCount(c->argv[1]);
2179         incrRefCount(c->argv[2]);
2180     } else {
2181         lobj = dictGetEntryVal(de);
2182         if (lobj->type != REDIS_LIST) {
2183             addReply(c,shared.wrongtypeerr);
2184             return;
2185         }
2186         list = lobj->ptr;
2187         if (where == REDIS_HEAD) {
2188             if (!listAddNodeHead(list,c->argv[2])) oom("listAddNodeHead");
2189         } else {
2190             if (!listAddNodeTail(list,c->argv[2])) oom("listAddNodeTail");
2191         }
2192         incrRefCount(c->argv[2]);
2193     }
2194     server.dirty++;
2195     addReply(c,shared.ok);
2196 }
2197
2198 static void lpushCommand(redisClient *c) {
2199     pushGenericCommand(c,REDIS_HEAD);
2200 }
2201
2202 static void rpushCommand(redisClient *c) {
2203     pushGenericCommand(c,REDIS_TAIL);
2204 }
2205
2206 static void llenCommand(redisClient *c) {
2207     dictEntry *de;
2208     list *l;
2209
2210     de = dictFind(c->dict,c->argv[1]);
2211     if (de == NULL) {
2212         addReply(c,shared.czero);
2213         return;
2214     } else {
2215         robj *o = dictGetEntryVal(de);
2216         if (o->type != REDIS_LIST) {
2217             addReply(c,shared.wrongtypeerr);
2218         } else {
2219             l = o->ptr;
2220             addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",listLength(l)));
2221         }
2222     }
2223 }
2224
2225 static void lindexCommand(redisClient *c) {
2226     dictEntry *de;
2227     int index = atoi(c->argv[2]->ptr);
2228
2229     de = dictFind(c->dict,c->argv[1]);
2230     if (de == NULL) {
2231         addReply(c,shared.nullbulk);
2232     } else {
2233         robj *o = dictGetEntryVal(de);
2234
2235         if (o->type != REDIS_LIST) {
2236             addReply(c,shared.wrongtypeerr);
2237         } else {
2238             list *list = o->ptr;
2239             listNode *ln;
2240
2241             ln = listIndex(list, index);
2242             if (ln == NULL) {
2243                 addReply(c,shared.nullbulk);
2244             } else {
2245                 robj *ele = listNodeValue(ln);
2246                 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
2247                 addReply(c,ele);
2248                 addReply(c,shared.crlf);
2249             }
2250         }
2251     }
2252 }
2253
2254 static void lsetCommand(redisClient *c) {
2255     dictEntry *de;
2256     int index = atoi(c->argv[2]->ptr);
2257
2258     de = dictFind(c->dict,c->argv[1]);
2259     if (de == NULL) {
2260         addReply(c,shared.nokeyerr);
2261     } else {
2262         robj *o = dictGetEntryVal(de);
2263
2264         if (o->type != REDIS_LIST) {
2265             addReply(c,shared.wrongtypeerr);
2266         } else {
2267             list *list = o->ptr;
2268             listNode *ln;
2269
2270             ln = listIndex(list, index);
2271             if (ln == NULL) {
2272                 addReply(c,shared.outofrangeerr);
2273             } else {
2274                 robj *ele = listNodeValue(ln);
2275
2276                 decrRefCount(ele);
2277                 listNodeValue(ln) = c->argv[3];
2278                 incrRefCount(c->argv[3]);
2279                 addReply(c,shared.ok);
2280                 server.dirty++;
2281             }
2282         }
2283     }
2284 }
2285
2286 static void popGenericCommand(redisClient *c, int where) {
2287     dictEntry *de;
2288
2289     de = dictFind(c->dict,c->argv[1]);
2290     if (de == NULL) {
2291         addReply(c,shared.nullbulk);
2292     } else {
2293         robj *o = dictGetEntryVal(de);
2294
2295         if (o->type != REDIS_LIST) {
2296             addReply(c,shared.wrongtypeerr);
2297         } else {
2298             list *list = o->ptr;
2299             listNode *ln;
2300
2301             if (where == REDIS_HEAD)
2302                 ln = listFirst(list);
2303             else
2304                 ln = listLast(list);
2305
2306             if (ln == NULL) {
2307                 addReply(c,shared.nullbulk);
2308             } else {
2309                 robj *ele = listNodeValue(ln);
2310                 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
2311                 addReply(c,ele);
2312                 addReply(c,shared.crlf);
2313                 listDelNode(list,ln);
2314                 server.dirty++;
2315             }
2316         }
2317     }
2318 }
2319
2320 static void lpopCommand(redisClient *c) {
2321     popGenericCommand(c,REDIS_HEAD);
2322 }
2323
2324 static void rpopCommand(redisClient *c) {
2325     popGenericCommand(c,REDIS_TAIL);
2326 }
2327
2328 static void lrangeCommand(redisClient *c) {
2329     dictEntry *de;
2330     int start = atoi(c->argv[2]->ptr);
2331     int end = atoi(c->argv[3]->ptr);
2332
2333     de = dictFind(c->dict,c->argv[1]);
2334     if (de == NULL) {
2335         addReply(c,shared.nullmultibulk);
2336     } else {
2337         robj *o = dictGetEntryVal(de);
2338
2339         if (o->type != REDIS_LIST) {
2340             addReply(c,shared.wrongtypeerr);
2341         } else {
2342             list *list = o->ptr;
2343             listNode *ln;
2344             int llen = listLength(list);
2345             int rangelen, j;
2346             robj *ele;
2347
2348             /* convert negative indexes */
2349             if (start < 0) start = llen+start;
2350             if (end < 0) end = llen+end;
2351             if (start < 0) start = 0;
2352             if (end < 0) end = 0;
2353
2354             /* indexes sanity checks */
2355             if (start > end || start >= llen) {
2356                 /* Out of range start or start > end result in empty list */
2357                 addReply(c,shared.emptymultibulk);
2358                 return;
2359             }
2360             if (end >= llen) end = llen-1;
2361             rangelen = (end-start)+1;
2362
2363             /* Return the result in form of a multi-bulk reply */
2364             ln = listIndex(list, start);
2365             addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",rangelen));
2366             for (j = 0; j < rangelen; j++) {
2367                 ele = listNodeValue(ln);
2368                 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
2369                 addReply(c,ele);
2370                 addReply(c,shared.crlf);
2371                 ln = ln->next;
2372             }
2373         }
2374     }
2375 }
2376
2377 static void ltrimCommand(redisClient *c) {
2378     dictEntry *de;
2379     int start = atoi(c->argv[2]->ptr);
2380     int end = atoi(c->argv[3]->ptr);
2381
2382     de = dictFind(c->dict,c->argv[1]);
2383     if (de == NULL) {
2384         addReply(c,shared.nokeyerr);
2385     } else {
2386         robj *o = dictGetEntryVal(de);
2387
2388         if (o->type != REDIS_LIST) {
2389             addReply(c,shared.wrongtypeerr);
2390         } else {
2391             list *list = o->ptr;
2392             listNode *ln;
2393             int llen = listLength(list);
2394             int j, ltrim, rtrim;
2395
2396             /* convert negative indexes */
2397             if (start < 0) start = llen+start;
2398             if (end < 0) end = llen+end;
2399             if (start < 0) start = 0;
2400             if (end < 0) end = 0;
2401
2402             /* indexes sanity checks */
2403             if (start > end || start >= llen) {
2404                 /* Out of range start or start > end result in empty list */
2405                 ltrim = llen;
2406                 rtrim = 0;
2407             } else {
2408                 if (end >= llen) end = llen-1;
2409                 ltrim = start;
2410                 rtrim = llen-end-1;
2411             }
2412
2413             /* Remove list elements to perform the trim */
2414             for (j = 0; j < ltrim; j++) {
2415                 ln = listFirst(list);
2416                 listDelNode(list,ln);
2417             }
2418             for (j = 0; j < rtrim; j++) {
2419                 ln = listLast(list);
2420                 listDelNode(list,ln);
2421             }
2422             addReply(c,shared.ok);
2423             server.dirty++;
2424         }
2425     }
2426 }
2427
2428 static void lremCommand(redisClient *c) {
2429     dictEntry *de;
2430
2431     de = dictFind(c->dict,c->argv[1]);
2432     if (de == NULL) {
2433         addReply(c,shared.nokeyerr);
2434     } else {
2435         robj *o = dictGetEntryVal(de);
2436
2437         if (o->type != REDIS_LIST) {
2438             addReply(c,shared.wrongtypeerr);
2439         } else {
2440             list *list = o->ptr;
2441             listNode *ln, *next;
2442             int toremove = atoi(c->argv[2]->ptr);
2443             int removed = 0;
2444             int fromtail = 0;
2445
2446             if (toremove < 0) {
2447                 toremove = -toremove;
2448                 fromtail = 1;
2449             }
2450             ln = fromtail ? list->tail : list->head;
2451             while (ln) {
2452                 robj *ele = listNodeValue(ln);
2453
2454                 next = fromtail ? ln->prev : ln->next;
2455                 if (sdscmp(ele->ptr,c->argv[3]->ptr) == 0) {
2456                     listDelNode(list,ln);
2457                     server.dirty++;
2458                     removed++;
2459                     if (toremove && removed == toremove) break;
2460                 }
2461                 ln = next;
2462             }
2463             addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",removed));
2464         }
2465     }
2466 }
2467
2468 /* ==================================== Sets ================================ */
2469
2470 static void saddCommand(redisClient *c) {
2471     dictEntry *de;
2472     robj *set;
2473
2474     de = dictFind(c->dict,c->argv[1]);
2475     if (de == NULL) {
2476         set = createSetObject();
2477         dictAdd(c->dict,c->argv[1],set);
2478         incrRefCount(c->argv[1]);
2479     } else {
2480         set = dictGetEntryVal(de);
2481         if (set->type != REDIS_SET) {
2482             addReply(c,shared.wrongtypeerr);
2483             return;
2484         }
2485     }
2486     if (dictAdd(set->ptr,c->argv[2],NULL) == DICT_OK) {
2487         incrRefCount(c->argv[2]);
2488         server.dirty++;
2489         addReply(c,shared.cone);
2490     } else {
2491         addReply(c,shared.czero);
2492     }
2493 }
2494
2495 static void sremCommand(redisClient *c) {
2496     dictEntry *de;
2497
2498     de = dictFind(c->dict,c->argv[1]);
2499     if (de == NULL) {
2500         addReply(c,shared.czero);
2501     } else {
2502         robj *set;
2503
2504         set = dictGetEntryVal(de);
2505         if (set->type != REDIS_SET) {
2506             addReply(c,shared.wrongtypeerr);
2507             return;
2508         }
2509         if (dictDelete(set->ptr,c->argv[2]) == DICT_OK) {
2510             server.dirty++;
2511             addReply(c,shared.cone);
2512         } else {
2513             addReply(c,shared.czero);
2514         }
2515     }
2516 }
2517
2518 static void sismemberCommand(redisClient *c) {
2519     dictEntry *de;
2520
2521     de = dictFind(c->dict,c->argv[1]);
2522     if (de == NULL) {
2523         addReply(c,shared.czero);
2524     } else {
2525         robj *set;
2526
2527         set = dictGetEntryVal(de);
2528         if (set->type != REDIS_SET) {
2529             addReply(c,shared.wrongtypeerr);
2530             return;
2531         }
2532         if (dictFind(set->ptr,c->argv[2]))
2533             addReply(c,shared.cone);
2534         else
2535             addReply(c,shared.czero);
2536     }
2537 }
2538
2539 static void scardCommand(redisClient *c) {
2540     dictEntry *de;
2541     dict *s;
2542
2543     de = dictFind(c->dict,c->argv[1]);
2544     if (de == NULL) {
2545         addReply(c,shared.czero);
2546         return;
2547     } else {
2548         robj *o = dictGetEntryVal(de);
2549         if (o->type != REDIS_SET) {
2550             addReply(c,shared.wrongtypeerr);
2551         } else {
2552             s = o->ptr;
2553             addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",
2554                 dictGetHashTableUsed(s)));
2555         }
2556     }
2557 }
2558
2559 static int qsortCompareSetsByCardinality(const void *s1, const void *s2) {
2560     dict **d1 = (void*) s1, **d2 = (void*) s2;
2561
2562     return dictGetHashTableUsed(*d1)-dictGetHashTableUsed(*d2);
2563 }
2564
2565 static void sinterGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey) {
2566     dict **dv = zmalloc(sizeof(dict*)*setsnum);
2567     dictIterator *di;
2568     dictEntry *de;
2569     robj *lenobj = NULL, *dstset = NULL;
2570     int j, cardinality = 0;
2571
2572     if (!dv) oom("sinterCommand");
2573     for (j = 0; j < setsnum; j++) {
2574         robj *setobj;
2575         dictEntry *de;
2576
2577         de = dictFind(c->dict,setskeys[j]);
2578         if (!de) {
2579             zfree(dv);
2580             addReply(c,shared.nokeyerr);
2581             return;
2582         }
2583         setobj = dictGetEntryVal(de);
2584         if (setobj->type != REDIS_SET) {
2585             zfree(dv);
2586             addReply(c,shared.wrongtypeerr);
2587             return;
2588         }
2589         dv[j] = setobj->ptr;
2590     }
2591     /* Sort sets from the smallest to largest, this will improve our
2592      * algorithm's performace */
2593     qsort(dv,setsnum,sizeof(dict*),qsortCompareSetsByCardinality);
2594
2595     /* The first thing we should output is the total number of elements...
2596      * since this is a multi-bulk write, but at this stage we don't know
2597      * the intersection set size, so we use a trick, append an empty object
2598      * to the output list and save the pointer to later modify it with the
2599      * right length */
2600     if (!dstkey) {
2601         lenobj = createObject(REDIS_STRING,NULL);
2602         addReply(c,lenobj);
2603         decrRefCount(lenobj);
2604     } else {
2605         /* If we have a target key where to store the resulting set
2606          * create this key with an empty set inside */
2607         dstset = createSetObject();
2608         dictDelete(c->dict,dstkey);
2609         dictAdd(c->dict,dstkey,dstset);
2610         incrRefCount(dstkey);
2611     }
2612
2613     /* Iterate all the elements of the first (smallest) set, and test
2614      * the element against all the other sets, if at least one set does
2615      * not include the element it is discarded */
2616     di = dictGetIterator(dv[0]);
2617     if (!di) oom("dictGetIterator");
2618
2619     while((de = dictNext(di)) != NULL) {
2620         robj *ele;
2621
2622         for (j = 1; j < setsnum; j++)
2623             if (dictFind(dv[j],dictGetEntryKey(de)) == NULL) break;
2624         if (j != setsnum)
2625             continue; /* at least one set does not contain the member */
2626         ele = dictGetEntryKey(de);
2627         if (!dstkey) {
2628             addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(ele->ptr)));
2629             addReply(c,ele);
2630             addReply(c,shared.crlf);
2631             cardinality++;
2632         } else {
2633             dictAdd(dstset->ptr,ele,NULL);
2634             incrRefCount(ele);
2635         }
2636     }
2637     dictReleaseIterator(di);
2638
2639     if (!dstkey)
2640         lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality);
2641     else
2642         addReply(c,shared.ok);
2643     zfree(dv);
2644 }
2645
2646 static void sinterCommand(redisClient *c) {
2647     sinterGenericCommand(c,c->argv+1,c->argc-1,NULL);
2648 }
2649
2650 static void sinterstoreCommand(redisClient *c) {
2651     sinterGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
2652 }
2653
2654 static void flushdbCommand(redisClient *c) {
2655     dictEmpty(c->dict);
2656     addReply(c,shared.ok);
2657     rdbSave(server.dbfilename);
2658 }
2659
2660 static void flushallCommand(redisClient *c) {
2661     emptyDb();
2662     addReply(c,shared.ok);
2663     rdbSave(server.dbfilename);
2664 }
2665
2666 redisSortOperation *createSortOperation(int type, robj *pattern) {
2667     redisSortOperation *so = zmalloc(sizeof(*so));
2668     if (!so) oom("createSortOperation");
2669     so->type = type;
2670     so->pattern = pattern;
2671     return so;
2672 }
2673
2674 /* Return the value associated to the key with a name obtained
2675  * substituting the first occurence of '*' in 'pattern' with 'subst' */
2676 robj *lookupKeyByPattern(dict *dict, robj *pattern, robj *subst) {
2677     char *p;
2678     sds spat, ssub;
2679     robj keyobj;
2680     int prefixlen, sublen, postfixlen;
2681     dictEntry *de;
2682     /* Expoit the internal sds representation to create a sds string allocated on the stack in order to make this function faster */
2683     struct {
2684         long len;
2685         long free;
2686         char buf[REDIS_SORTKEY_MAX+1];
2687     } keyname;
2688
2689
2690     spat = pattern->ptr;
2691     ssub = subst->ptr;
2692     if (sdslen(spat)+sdslen(ssub)-1 > REDIS_SORTKEY_MAX) return NULL;
2693     p = strchr(spat,'*');
2694     if (!p) return NULL;
2695
2696     prefixlen = p-spat;
2697     sublen = sdslen(ssub);
2698     postfixlen = sdslen(spat)-(prefixlen+1);
2699     memcpy(keyname.buf,spat,prefixlen);
2700     memcpy(keyname.buf+prefixlen,ssub,sublen);
2701     memcpy(keyname.buf+prefixlen+sublen,p+1,postfixlen);
2702     keyname.buf[prefixlen+sublen+postfixlen] = '\0';
2703     keyname.len = prefixlen+sublen+postfixlen;
2704
2705     keyobj.refcount = 1;
2706     keyobj.type = REDIS_STRING;
2707     keyobj.ptr = ((char*)&keyname)+(sizeof(long)*2);
2708
2709     de = dictFind(dict,&keyobj);
2710     /* printf("lookup '%s' => %p\n", keyname.buf,de); */
2711     if (!de) return NULL;
2712     return dictGetEntryVal(de);
2713 }
2714
2715 /* sortCompare() is used by qsort in sortCommand(). Given that qsort_r with
2716  * the additional parameter is not standard but a BSD-specific we have to
2717  * pass sorting parameters via the global 'server' structure */
2718 static int sortCompare(const void *s1, const void *s2) {
2719     const redisSortObject *so1 = s1, *so2 = s2;
2720     int cmp;
2721
2722     if (!server.sort_alpha) {
2723         /* Numeric sorting. Here it's trivial as we precomputed scores */
2724         if (so1->u.score > so2->u.score) {
2725             cmp = 1;
2726         } else if (so1->u.score < so2->u.score) {
2727             cmp = -1;
2728         } else {
2729             cmp = 0;
2730         }
2731     } else {
2732         /* Alphanumeric sorting */
2733         if (server.sort_bypattern) {
2734             if (!so1->u.cmpobj || !so2->u.cmpobj) {
2735                 /* At least one compare object is NULL */
2736                 if (so1->u.cmpobj == so2->u.cmpobj)
2737                     cmp = 0;
2738                 else if (so1->u.cmpobj == NULL)
2739                     cmp = -1;
2740                 else
2741                     cmp = 1;
2742             } else {
2743                 /* We have both the objects, use strcoll */
2744                 cmp = strcoll(so1->u.cmpobj->ptr,so2->u.cmpobj->ptr);
2745             }
2746         } else {
2747             /* Compare elements directly */
2748             cmp = strcoll(so1->obj->ptr,so2->obj->ptr);
2749         }
2750     }
2751     return server.sort_desc ? -cmp : cmp;
2752 }
2753
2754 /* The SORT command is the most complex command in Redis. Warning: this code
2755  * is optimized for speed and a bit less for readability */
2756 static void sortCommand(redisClient *c) {
2757     dictEntry *de;
2758     list *operations;
2759     int outputlen = 0;
2760     int desc = 0, alpha = 0;
2761     int limit_start = 0, limit_count = -1, start, end;
2762     int j, dontsort = 0, vectorlen;
2763     int getop = 0; /* GET operation counter */
2764     robj *sortval, *sortby = NULL;
2765     redisSortObject *vector; /* Resulting vector to sort */
2766
2767     /* Lookup the key to sort. It must be of the right types */
2768     de = dictFind(c->dict,c->argv[1]);
2769     if (de == NULL) {
2770         addReply(c,shared.nokeyerr);
2771         return;
2772     }
2773     sortval = dictGetEntryVal(de);
2774     if (sortval->type != REDIS_SET && sortval->type != REDIS_LIST) {
2775         addReply(c,shared.wrongtypeerr);
2776         return;
2777     }
2778
2779     /* Create a list of operations to perform for every sorted element.
2780      * Operations can be GET/DEL/INCR/DECR */
2781     operations = listCreate();
2782     listSetFreeMethod(operations,zfree);
2783     j = 2;
2784
2785     /* Now we need to protect sortval incrementing its count, in the future
2786      * SORT may have options able to overwrite/delete keys during the sorting
2787      * and the sorted key itself may get destroied */
2788     incrRefCount(sortval);
2789
2790     /* The SORT command has an SQL-alike syntax, parse it */
2791     while(j < c->argc) {
2792         int leftargs = c->argc-j-1;
2793         if (!strcasecmp(c->argv[j]->ptr,"asc")) {
2794             desc = 0;
2795         } else if (!strcasecmp(c->argv[j]->ptr,"desc")) {
2796             desc = 1;
2797         } else if (!strcasecmp(c->argv[j]->ptr,"alpha")) {
2798             alpha = 1;
2799         } else if (!strcasecmp(c->argv[j]->ptr,"limit") && leftargs >= 2) {
2800             limit_start = atoi(c->argv[j+1]->ptr);
2801             limit_count = atoi(c->argv[j+2]->ptr);
2802             j+=2;
2803         } else if (!strcasecmp(c->argv[j]->ptr,"by") && leftargs >= 1) {
2804             sortby = c->argv[j+1];
2805             /* If the BY pattern does not contain '*', i.e. it is constant,
2806              * we don't need to sort nor to lookup the weight keys. */
2807             if (strchr(c->argv[j+1]->ptr,'*') == NULL) dontsort = 1;
2808             j++;
2809         } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
2810             listAddNodeTail(operations,createSortOperation(
2811                 REDIS_SORT_GET,c->argv[j+1]));
2812             getop++;
2813             j++;
2814         } else if (!strcasecmp(c->argv[j]->ptr,"del") && leftargs >= 1) {
2815             listAddNodeTail(operations,createSortOperation(
2816                 REDIS_SORT_DEL,c->argv[j+1]));
2817             j++;
2818         } else if (!strcasecmp(c->argv[j]->ptr,"incr") && leftargs >= 1) {
2819             listAddNodeTail(operations,createSortOperation(
2820                 REDIS_SORT_INCR,c->argv[j+1]));
2821             j++;
2822         } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
2823             listAddNodeTail(operations,createSortOperation(
2824                 REDIS_SORT_DECR,c->argv[j+1]));
2825             j++;
2826         } else {
2827             decrRefCount(sortval);
2828             listRelease(operations);
2829             addReply(c,shared.syntaxerr);
2830             return;
2831         }
2832         j++;
2833     }
2834
2835     /* Load the sorting vector with all the objects to sort */
2836     vectorlen = (sortval->type == REDIS_LIST) ?
2837         listLength((list*)sortval->ptr) :
2838         dictGetHashTableUsed((dict*)sortval->ptr);
2839     vector = zmalloc(sizeof(redisSortObject)*vectorlen);
2840     if (!vector) oom("allocating objects vector for SORT");
2841     j = 0;
2842     if (sortval->type == REDIS_LIST) {
2843         list *list = sortval->ptr;
2844         listNode *ln = list->head;
2845         while(ln) {
2846             robj *ele = ln->value;
2847             vector[j].obj = ele;
2848             vector[j].u.score = 0;
2849             vector[j].u.cmpobj = NULL;
2850             ln = ln->next;
2851             j++;
2852         }
2853     } else {
2854         dict *set = sortval->ptr;
2855         dictIterator *di;
2856         dictEntry *setele;
2857
2858         di = dictGetIterator(set);
2859         if (!di) oom("dictGetIterator");
2860         while((setele = dictNext(di)) != NULL) {
2861             vector[j].obj = dictGetEntryKey(setele);
2862             vector[j].u.score = 0;
2863             vector[j].u.cmpobj = NULL;
2864             j++;
2865         }
2866         dictReleaseIterator(di);
2867     }
2868     assert(j == vectorlen);
2869
2870     /* Now it's time to load the right scores in the sorting vector */
2871     if (dontsort == 0) {
2872         for (j = 0; j < vectorlen; j++) {
2873             if (sortby) {
2874                 robj *byval;
2875
2876                 byval = lookupKeyByPattern(c->dict,sortby,vector[j].obj);
2877                 if (!byval || byval->type != REDIS_STRING) continue;
2878                 if (alpha) {
2879                     vector[j].u.cmpobj = byval;
2880                     incrRefCount(byval);
2881                 } else {
2882                     vector[j].u.score = strtod(byval->ptr,NULL);
2883                 }
2884             } else {
2885                 if (!alpha) vector[j].u.score = strtod(vector[j].obj->ptr,NULL);
2886             }
2887         }
2888     }
2889
2890     /* We are ready to sort the vector... perform a bit of sanity check
2891      * on the LIMIT option too. We'll use a partial version of quicksort. */
2892     start = (limit_start < 0) ? 0 : limit_start;
2893     end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
2894     if (start >= vectorlen) {
2895         start = vectorlen-1;
2896         end = vectorlen-2;
2897     }
2898     if (end >= vectorlen) end = vectorlen-1;
2899
2900     if (dontsort == 0) {
2901         server.sort_desc = desc;
2902         server.sort_alpha = alpha;
2903         server.sort_bypattern = sortby ? 1 : 0;
2904         qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare);
2905     }
2906
2907     /* Send command output to the output buffer, performing the specified
2908      * GET/DEL/INCR/DECR operations if any. */
2909     outputlen = getop ? getop*(end-start+1) : end-start+1;
2910     addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",outputlen));
2911     for (j = start; j <= end; j++) {
2912         listNode *ln = operations->head;
2913         if (!getop) {
2914             addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
2915                 sdslen(vector[j].obj->ptr)));
2916             addReply(c,vector[j].obj);
2917             addReply(c,shared.crlf);
2918         }
2919         while(ln) {
2920             redisSortOperation *sop = ln->value;
2921             robj *val = lookupKeyByPattern(c->dict,sop->pattern,
2922                 vector[j].obj);
2923
2924             if (sop->type == REDIS_SORT_GET) {
2925                 if (!val || val->type != REDIS_STRING) {
2926                     addReply(c,shared.nullbulk);
2927                 } else {
2928                     addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
2929                         sdslen(val->ptr)));
2930                     addReply(c,val);
2931                     addReply(c,shared.crlf);
2932                 }
2933             } else if (sop->type == REDIS_SORT_DEL) {
2934                 /* TODO */
2935             }
2936             ln = ln->next;
2937         }
2938     }
2939
2940     /* Cleanup */
2941     decrRefCount(sortval);
2942     listRelease(operations);
2943     for (j = 0; j < vectorlen; j++) {
2944         if (sortby && alpha && vector[j].u.cmpobj)
2945             decrRefCount(vector[j].u.cmpobj);
2946     }
2947     zfree(vector);
2948 }
2949
2950 static void infoCommand(redisClient *c) {
2951     sds info;
2952     time_t uptime = time(NULL)-server.stat_starttime;
2953
2954     info = sdscatprintf(sdsempty(),
2955         "redis_version:%s\r\n"
2956         "connected_clients:%d\r\n"
2957         "connected_slaves:%d\r\n"
2958         "used_memory:%d\r\n"
2959         "changes_since_last_save:%lld\r\n"
2960         "last_save_time:%d\r\n"
2961         "total_connections_received:%lld\r\n"
2962         "total_commands_processed:%lld\r\n"
2963         "uptime_in_seconds:%d\r\n"
2964         "uptime_in_days:%d\r\n"
2965         ,REDIS_VERSION,
2966         listLength(server.clients)-listLength(server.slaves),
2967         listLength(server.slaves),
2968         server.usedmemory,
2969         server.dirty,
2970         server.lastsave,
2971         server.stat_numconnections,
2972         server.stat_numcommands,
2973         uptime,
2974         uptime/(3600*24)
2975     );
2976     addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(info)));
2977     addReplySds(c,info);
2978     addReply(c,shared.crlf);
2979 }
2980
2981 /* =============================== Replication  ============================= */
2982
2983 /* Send the whole output buffer syncronously to the slave. This a general operation in theory, but it is actually useful only for replication. */
2984 static int flushClientOutput(redisClient *c) {
2985     int retval;
2986     time_t start = time(NULL);
2987
2988     while(listLength(c->reply)) {
2989         if (time(NULL)-start > 5) return REDIS_ERR; /* 5 seconds timeout */
2990         retval = aeWait(c->fd,AE_WRITABLE,1000);
2991         if (retval == -1) {
2992             return REDIS_ERR;
2993         } else if (retval & AE_WRITABLE) {
2994             sendReplyToClient(NULL, c->fd, c, AE_WRITABLE);
2995         }
2996     }
2997     return REDIS_OK;
2998 }
2999
3000 static int syncWrite(int fd, char *ptr, ssize_t size, int timeout) {
3001     ssize_t nwritten, ret = size;
3002     time_t start = time(NULL);
3003
3004     timeout++;
3005     while(size) {
3006         if (aeWait(fd,AE_WRITABLE,1000) & AE_WRITABLE) {
3007             nwritten = write(fd,ptr,size);
3008             if (nwritten == -1) return -1;
3009             ptr += nwritten;
3010             size -= nwritten;
3011         }
3012         if ((time(NULL)-start) > timeout) {
3013             errno = ETIMEDOUT;
3014             return -1;
3015         }
3016     }
3017     return ret;
3018 }
3019
3020 static int syncRead(int fd, char *ptr, ssize_t size, int timeout) {
3021     ssize_t nread, totread = 0;
3022     time_t start = time(NULL);
3023
3024     timeout++;
3025     while(size) {
3026         if (aeWait(fd,AE_READABLE,1000) & AE_READABLE) {
3027             nread = read(fd,ptr,size);
3028             if (nread == -1) return -1;
3029             ptr += nread;
3030             size -= nread;
3031             totread += nread;
3032         }
3033         if ((time(NULL)-start) > timeout) {
3034             errno = ETIMEDOUT;
3035             return -1;
3036         }
3037     }
3038     return totread;
3039 }
3040
3041 static int syncReadLine(int fd, char *ptr, ssize_t size, int timeout) {
3042     ssize_t nread = 0;
3043
3044     size--;
3045     while(size) {
3046         char c;
3047
3048         if (syncRead(fd,&c,1,timeout) == -1) return -1;
3049         if (c == '\n') {
3050             *ptr = '\0';
3051             if (nread && *(ptr-1) == '\r') *(ptr-1) = '\0';
3052             return nread;
3053         } else {
3054             *ptr++ = c;
3055             *ptr = '\0';
3056             nread++;
3057         }
3058     }
3059     return nread;
3060 }
3061
3062 static void syncCommand(redisClient *c) {
3063     struct stat sb;
3064     int fd = -1, len;
3065     time_t start = time(NULL);
3066     char sizebuf[32];
3067
3068     /* ignore SYNC if aleady slave or in monitor mode */
3069     if (c->flags & REDIS_SLAVE) return;
3070
3071     redisLog(REDIS_NOTICE,"Slave ask for syncronization");
3072     if (flushClientOutput(c) == REDIS_ERR ||
3073         rdbSave(server.dbfilename) != REDIS_OK)
3074         goto closeconn;
3075
3076     fd = open(server.dbfilename, O_RDONLY);
3077     if (fd == -1 || fstat(fd,&sb) == -1) goto closeconn;
3078     len = sb.st_size;
3079
3080     snprintf(sizebuf,32,"$%d\r\n",len);
3081     if (syncWrite(c->fd,sizebuf,strlen(sizebuf),5) == -1) goto closeconn;
3082     while(len) {
3083         char buf[1024];
3084         int nread;
3085
3086         if (time(NULL)-start > REDIS_MAX_SYNC_TIME) goto closeconn;
3087         nread = read(fd,buf,1024);
3088         if (nread == -1) goto closeconn;
3089         len -= nread;
3090         if (syncWrite(c->fd,buf,nread,5) == -1) goto closeconn;
3091     }
3092     if (syncWrite(c->fd,"\r\n",2,5) == -1) goto closeconn;
3093     close(fd);
3094     c->flags |= REDIS_SLAVE;
3095     c->slaveseldb = 0;
3096     if (!listAddNodeTail(server.slaves,c)) oom("listAddNodeTail");
3097     redisLog(REDIS_NOTICE,"Syncronization with slave succeeded");
3098     return;
3099
3100 closeconn:
3101     if (fd != -1) close(fd);
3102     c->flags |= REDIS_CLOSE;
3103     redisLog(REDIS_WARNING,"Syncronization with slave failed");
3104     return;
3105 }
3106
3107 static int syncWithMaster(void) {
3108     char buf[1024], tmpfile[256];
3109     int dumpsize;
3110     int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);
3111     int dfd;
3112
3113     if (fd == -1) {
3114         redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
3115             strerror(errno));
3116         return REDIS_ERR;
3117     }
3118     /* Issue the SYNC command */
3119     if (syncWrite(fd,"SYNC \r\n",7,5) == -1) {
3120         close(fd);
3121         redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
3122             strerror(errno));
3123         return REDIS_ERR;
3124     }
3125     /* Read the bulk write count */
3126     if (syncReadLine(fd,buf,1024,5) == -1) {
3127         close(fd);
3128         redisLog(REDIS_WARNING,"I/O error reading bulk count from MASTER: %s",
3129             strerror(errno));
3130         return REDIS_ERR;
3131     }
3132     dumpsize = atoi(buf+1);
3133     redisLog(REDIS_NOTICE,"Receiving %d bytes data dump from MASTER",dumpsize);
3134     /* Read the bulk write data on a temp file */
3135     snprintf(tmpfile,256,"temp-%d.%ld.rdb",(int)time(NULL),(long int)random());
3136     dfd = open(tmpfile,O_CREAT|O_WRONLY,0644);
3137     if (dfd == -1) {
3138         close(fd);
3139         redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
3140         return REDIS_ERR;
3141     }
3142     while(dumpsize) {
3143         int nread, nwritten;
3144
3145         nread = read(fd,buf,(dumpsize < 1024)?dumpsize:1024);
3146         if (nread == -1) {
3147             redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
3148                 strerror(errno));
3149             close(fd);
3150             close(dfd);
3151             return REDIS_ERR;
3152         }
3153         nwritten = write(dfd,buf,nread);
3154         if (nwritten == -1) {
3155             redisLog(REDIS_WARNING,"Write error writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno));
3156             close(fd);
3157             close(dfd);
3158             return REDIS_ERR;
3159         }
3160         dumpsize -= nread;
3161     }
3162     close(dfd);
3163     if (rename(tmpfile,server.dbfilename) == -1) {
3164         redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
3165         unlink(tmpfile);
3166         close(fd);
3167         return REDIS_ERR;
3168     }
3169     emptyDb();
3170     if (rdbLoad(server.dbfilename) != REDIS_OK) {
3171         redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
3172         close(fd);
3173         return REDIS_ERR;
3174     }
3175     server.master = createClient(fd);
3176     server.master->flags |= REDIS_MASTER;
3177     server.replstate = REDIS_REPL_CONNECTED;
3178     return REDIS_OK;
3179 }
3180
3181 static void monitorCommand(redisClient *c) {
3182     /* ignore MONITOR if aleady slave or in monitor mode */
3183     if (c->flags & REDIS_SLAVE) return;
3184
3185     c->flags |= (REDIS_SLAVE|REDIS_MONITOR);
3186     c->slaveseldb = 0;
3187     if (!listAddNodeTail(server.monitors,c)) oom("listAddNodeTail");
3188     addReply(c,shared.ok);
3189 }
3190
3191 /* =================================== Main! ================================ */
3192
3193 static void daemonize(void) {
3194     int fd;
3195     FILE *fp;
3196
3197     if (fork() != 0) exit(0); /* parent exits */
3198     setsid(); /* create a new session */
3199
3200     /* Every output goes to /dev/null. If Redis is daemonized but
3201      * the 'logfile' is set to 'stdout' in the configuration file
3202      * it will not log at all. */
3203     if ((fd = open("/dev/null", O_RDWR, 0)) != -1) {
3204         dup2(fd, STDIN_FILENO);
3205         dup2(fd, STDOUT_FILENO);
3206         dup2(fd, STDERR_FILENO);
3207         if (fd > STDERR_FILENO) close(fd);
3208     }
3209     /* Try to write the pid file */
3210     fp = fopen(server.pidfile,"w");
3211     if (fp) {
3212         fprintf(fp,"%d\n",getpid());
3213         fclose(fp);
3214     }
3215 }
3216
3217 int main(int argc, char **argv) {
3218     initServerConfig();
3219     if (argc == 2) {
3220         ResetServerSaveParams();
3221         loadServerConfig(argv[1]);
3222     } else if (argc > 2) {
3223         fprintf(stderr,"Usage: ./redis-server [/path/to/redis.conf]\n");
3224         exit(1);
3225     }
3226     initServer();
3227     if (server.daemonize) daemonize();
3228     redisLog(REDIS_NOTICE,"Server started, Redis version " REDIS_VERSION);
3229     if (rdbLoad(server.dbfilename) == REDIS_OK)
3230         redisLog(REDIS_NOTICE,"DB loaded from disk");
3231     if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
3232         acceptHandler, NULL, NULL) == AE_ERR) oom("creating file event");
3233     redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
3234     aeMain(server.el);
3235     aeDeleteEventLoop(server.el);
3236     return 0;
3237 }