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