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