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 (!strcmp(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 (!strcmp(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 (!strcmp(argv[0],"bind") && argc == 2) {
 940             server.bindaddr = zstrdup(argv[1]);
 941         } else if (!strcmp(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 (!strcmp(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 (!strcmp(argv[0],"loglevel") && argc == 2) {
 955             if (!strcmp(argv[1],"debug")) server.verbosity = REDIS_DEBUG;
 956             else if (!strcmp(argv[1],"notice")) server.verbosity = REDIS_NOTICE;
 957             else if (!strcmp(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 (!strcmp(argv[0],"logfile") && argc == 2) {
 963             FILE *fp;
 964
 965             server.logfile = zstrdup(argv[1]);
 966             if (!strcmp(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 (!strcmp(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 (!strcmp(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 (!strcmp(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 (!strcmp(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 (!strcmp(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 (!strcmp(argv[0],"requirepass") && argc == 2) {
1003           server.requirepass = zstrdup(argv[1]);
1004         } else if (!strcmp(argv[0],"pidfile") && argc == 2) {
1005           server.pidfile = zstrdup(argv[1]);
1006         } else if (!strcmp(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 (!strcmp(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     sdstolower(c->argv[0]->ptr);
1168     /* The QUIT command is handled as a special case. Normal command
1169      * procs are unable to close the client connection safely */
1170     if (!strcmp(c->argv[0]->ptr,"quit")) {
1171         freeClient(c);
1172         return 0;
1173     }
1174     cmd = lookupCommand(c->argv[0]->ptr);
1175     if (!cmd) {
1176         addReplySds(c,sdsnew("-ERR unknown command\r\n"));
1177         resetClient(c);
1178         return 1;
1179     } else if ((cmd->arity > 0 && cmd->arity != c->argc) ||
1180                (c->argc < -cmd->arity)) {
1181         addReplySds(c,sdsnew("-ERR wrong number of arguments\r\n"));
1182         resetClient(c);
1183         return 1;
1184     } else if (cmd->flags & REDIS_CMD_BULK && c->bulklen == -1) {
1185         int bulklen = atoi(c->argv[c->argc-1]->ptr);
1186
1187         decrRefCount(c->argv[c->argc-1]);
1188         if (bulklen < 0 || bulklen > 1024*1024*1024) {
1189             c->argc--;
1190             addReplySds(c,sdsnew("-ERR invalid bulk write count\r\n"));
1191             resetClient(c);
1192             return 1;
1193         }
1194         c->argc--;
1195         c->bulklen = bulklen+2; /* add two bytes for CR+LF */
1196         /* It is possible that the bulk read is already in the
1197          * buffer. Check this condition and handle it accordingly */
1198         if ((signed)sdslen(c->querybuf) >= c->bulklen) {
1199             c->argv[c->argc] = createStringObject(c->querybuf,c->bulklen-2);
1200             c->argc++;
1201             c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
1202         } else {
1203             return 1;
1204         }
1205     }
1206     /* Let's try to share objects on the command arguments vector */
1207     if (server.shareobjects) {
1208         int j;
1209         for(j = 1; j < c->argc; j++)
1210             c->argv[j] = tryObjectSharing(c->argv[j]);
1211     }
1212     /* Check if the user is authenticated */
1213     if (server.requirepass && !c->authenticated && cmd->proc != authCommand) {
1214         addReplySds(c,sdsnew("-ERR operation not permitted\r\n"));
1215         resetClient(c);
1216         return 1;
1217     }
1218
1219     /* Exec the command */
1220     dirty = server.dirty;
1221     cmd->proc(c);
1222     if (server.dirty-dirty != 0 && listLength(server.slaves))
1223         replicationFeedSlaves(server.slaves,cmd,c->db->id,c->argv,c->argc);
1224     if (listLength(server.monitors))
1225         replicationFeedSlaves(server.monitors,cmd,c->db->id,c->argv,c->argc);
1226     server.stat_numcommands++;
1227
1228     /* Prepare the client for the next command */
1229     if (c->flags & REDIS_CLOSE) {
1230         freeClient(c);
1231         return 0;
1232     }
1233     resetClient(c);
1234     return 1;
1235 }
1236
1237 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc) {
1238     listNode *ln;
1239     robj *outv[REDIS_MAX_ARGS*4]; /* enough room for args, spaces, newlines */
1240     int outc = 0, j;
1241
1242     for (j = 0; j < argc; j++) {
1243         if (j != 0) outv[outc++] = shared.space;
1244         if ((cmd->flags & REDIS_CMD_BULK) && j == argc-1) {
1245             robj *lenobj;
1246
1247             lenobj = createObject(REDIS_STRING,
1248                 sdscatprintf(sdsempty(),"%d\r\n",sdslen(argv[j]->ptr)));
1249             lenobj->refcount = 0;
1250             outv[outc++] = lenobj;
1251         }
1252         outv[outc++] = argv[j];
1253     }
1254     outv[outc++] = shared.crlf;
1255
1256     /* Increment all the refcounts at start and decrement at end in order to
1257      * be sure to free objects if there is no slave in a replication state
1258      * able to be feed with commands */
1259     for (j = 0; j < outc; j++) incrRefCount(outv[j]);
1260     listRewind(slaves);
1261     while((ln = listYield(slaves))) {
1262         redisClient *slave = ln->value;
1263
1264         /* Don't feed slaves that are still waiting for BGSAVE to start */
1265         if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) continue;
1266
1267         /* Feed all the other slaves, MONITORs and so on */
1268         if (slave->slaveseldb != dictid) {
1269             robj *selectcmd;
1270
1271             switch(dictid) {
1272             case 0: selectcmd = shared.select0; break;
1273             case 1: selectcmd = shared.select1; break;
1274             case 2: selectcmd = shared.select2; break;
1275             case 3: selectcmd = shared.select3; break;
1276             case 4: selectcmd = shared.select4; break;
1277             case 5: selectcmd = shared.select5; break;
1278             case 6: selectcmd = shared.select6; break;
1279             case 7: selectcmd = shared.select7; break;
1280             case 8: selectcmd = shared.select8; break;
1281             case 9: selectcmd = shared.select9; break;
1282             default:
1283                 selectcmd = createObject(REDIS_STRING,
1284                     sdscatprintf(sdsempty(),"select %d\r\n",dictid));
1285                 selectcmd->refcount = 0;
1286                 break;
1287             }
1288             addReply(slave,selectcmd);
1289             slave->slaveseldb = dictid;
1290         }
1291         for (j = 0; j < outc; j++) addReply(slave,outv[j]);
1292     }
1293     for (j = 0; j < outc; j++) decrRefCount(outv[j]);
1294 }
1295
1296 static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
1297     redisClient *c = (redisClient*) privdata;
1298     char buf[REDIS_IOBUF_LEN];
1299     int nread;
1300     REDIS_NOTUSED(el);
1301     REDIS_NOTUSED(mask);
1302
1303     nread = read(fd, buf, REDIS_IOBUF_LEN);
1304     if (nread == -1) {
1305         if (errno == EAGAIN) {
1306             nread = 0;
1307         } else {
1308             redisLog(REDIS_DEBUG, "Reading from client: %s",strerror(errno));
1309             freeClient(c);
1310             return;
1311         }
1312     } else if (nread == 0) {
1313         redisLog(REDIS_DEBUG, "Client closed connection");
1314         freeClient(c);
1315         return;
1316     }
1317     if (nread) {
1318         c->querybuf = sdscatlen(c->querybuf, buf, nread);
1319         c->lastinteraction = time(NULL);
1320     } else {
1321         return;
1322     }
1323
1324 again:
1325     if (c->bulklen == -1) {
1326         /* Read the first line of the query */
1327         char *p = strchr(c->querybuf,'\n');
1328         size_t querylen;
1329         if (p) {
1330             sds query, *argv;
1331             int argc, j;
1332
1333             query = c->querybuf;
1334             c->querybuf = sdsempty();
1335             querylen = 1+(p-(query));
1336             if (sdslen(query) > querylen) {
1337                 /* leave data after the first line of the query in the buffer */
1338                 c->querybuf = sdscatlen(c->querybuf,query+querylen,sdslen(query)-querylen);
1339             }
1340             *p = '\0'; /* remove "\n" */
1341             if (*(p-1) == '\r') *(p-1) = '\0'; /* and "\r" if any */
1342             sdsupdatelen(query);
1343
1344             /* Now we can split the query in arguments */
1345             if (sdslen(query) == 0) {
1346                 /* Ignore empty query */
1347                 sdsfree(query);
1348                 return;
1349             }
1350             argv = sdssplitlen(query,sdslen(query)," ",1,&argc);
1351             sdsfree(query);
1352             if (argv == NULL) oom("sdssplitlen");
1353             for (j = 0; j < argc && j < REDIS_MAX_ARGS; j++) {
1354                 if (sdslen(argv[j])) {
1355                     c->argv[c->argc] = createObject(REDIS_STRING,argv[j]);
1356                     c->argc++;
1357                 } else {
1358                     sdsfree(argv[j]);
1359                 }
1360             }
1361             zfree(argv);
1362             /* Execute the command. If the client is still valid
1363              * after processCommand() return and there is something
1364              * on the query buffer try to process the next command. */
1365             if (processCommand(c) && sdslen(c->querybuf)) goto again;
1366             return;
1367         } else if (sdslen(c->querybuf) >= 1024) {
1368             redisLog(REDIS_DEBUG, "Client protocol error");
1369             freeClient(c);
1370             return;
1371         }
1372     } else {
1373         /* Bulk read handling. Note that if we are at this point
1374            the client already sent a command terminated with a newline,
1375            we are reading the bulk data that is actually the last
1376            argument of the command. */
1377         int qbl = sdslen(c->querybuf);
1378
1379         if (c->bulklen <= qbl) {
1380             /* Copy everything but the final CRLF as final argument */
1381             c->argv[c->argc] = createStringObject(c->querybuf,c->bulklen-2);
1382             c->argc++;
1383             c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
1384             processCommand(c);
1385             return;
1386         }
1387     }
1388 }
1389
1390 static int selectDb(redisClient *c, int id) {
1391     if (id < 0 || id >= server.dbnum)
1392         return REDIS_ERR;
1393     c->db = &server.db[id];
1394     return REDIS_OK;
1395 }
1396
1397 static void *dupClientReplyValue(void *o) {
1398     incrRefCount((robj*)o);
1399     return 0;
1400 }
1401
1402 static redisClient *createClient(int fd) {
1403     redisClient *c = zmalloc(sizeof(*c));
1404
1405     anetNonBlock(NULL,fd);
1406     anetTcpNoDelay(NULL,fd);
1407     if (!c) return NULL;
1408     selectDb(c,0);
1409     c->fd = fd;
1410     c->querybuf = sdsempty();
1411     c->argc = 0;
1412     c->bulklen = -1;
1413     c->sentlen = 0;
1414     c->flags = 0;
1415     c->lastinteraction = time(NULL);
1416     c->authenticated = 0;
1417     c->replstate = REDIS_REPL_NONE;
1418     if ((c->reply = listCreate()) == NULL) oom("listCreate");
1419     listSetFreeMethod(c->reply,decrRefCount);
1420     listSetDupMethod(c->reply,dupClientReplyValue);
1421     if (aeCreateFileEvent(server.el, c->fd, AE_READABLE,
1422         readQueryFromClient, c, NULL) == AE_ERR) {
1423         freeClient(c);
1424         return NULL;
1425     }
1426     if (!listAddNodeTail(server.clients,c)) oom("listAddNodeTail");
1427     return c;
1428 }
1429
1430 static void addReply(redisClient *c, robj *obj) {
1431     if (listLength(c->reply) == 0 &&
1432         (c->replstate == REDIS_REPL_NONE ||
1433          c->replstate == REDIS_REPL_ONLINE) &&
1434         aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
1435         sendReplyToClient, c, NULL) == AE_ERR) return;
1436     if (!listAddNodeTail(c->reply,obj)) oom("listAddNodeTail");
1437     incrRefCount(obj);
1438 }
1439
1440 static void addReplySds(redisClient *c, sds s) {
1441     robj *o = createObject(REDIS_STRING,s);
1442     addReply(c,o);
1443     decrRefCount(o);
1444 }
1445
1446 static void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
1447     int cport, cfd;
1448     char cip[128];
1449     REDIS_NOTUSED(el);
1450     REDIS_NOTUSED(mask);
1451     REDIS_NOTUSED(privdata);
1452
1453     cfd = anetAccept(server.neterr, fd, cip, &cport);
1454     if (cfd == AE_ERR) {
1455         redisLog(REDIS_DEBUG,"Accepting client connection: %s", server.neterr);
1456         return;
1457     }
1458     redisLog(REDIS_DEBUG,"Accepted %s:%d", cip, cport);
1459     if (createClient(cfd) == NULL) {
1460         redisLog(REDIS_WARNING,"Error allocating resoures for the client");
1461         close(cfd); /* May be already closed, just ingore errors */
1462         return;
1463     }
1464     server.stat_numconnections++;
1465 }
1466
1467 /* ======================= Redis objects implementation ===================== */
1468
1469 static robj *createObject(int type, void *ptr) {
1470     robj *o;
1471
1472     if (listLength(server.objfreelist)) {
1473         listNode *head = listFirst(server.objfreelist);
1474         o = listNodeValue(head);
1475         listDelNode(server.objfreelist,head);
1476     } else {
1477         o = zmalloc(sizeof(*o));
1478     }
1479     if (!o) oom("createObject");
1480     o->type = type;
1481     o->ptr = ptr;
1482     o->refcount = 1;
1483     return o;
1484 }
1485
1486 static robj *createStringObject(char *ptr, size_t len) {
1487     return createObject(REDIS_STRING,sdsnewlen(ptr,len));
1488 }
1489
1490 static robj *createListObject(void) {
1491     list *l = listCreate();
1492
1493     if (!l) oom("listCreate");
1494     listSetFreeMethod(l,decrRefCount);
1495     return createObject(REDIS_LIST,l);
1496 }
1497
1498 static robj *createSetObject(void) {
1499     dict *d = dictCreate(&setDictType,NULL);
1500     if (!d) oom("dictCreate");
1501     return createObject(REDIS_SET,d);
1502 }
1503
1504 static void freeStringObject(robj *o) {
1505     sdsfree(o->ptr);
1506 }
1507
1508 static void freeListObject(robj *o) {
1509     listRelease((list*) o->ptr);
1510 }
1511
1512 static void freeSetObject(robj *o) {
1513     dictRelease((dict*) o->ptr);
1514 }
1515
1516 static void freeHashObject(robj *o) {
1517     dictRelease((dict*) o->ptr);
1518 }
1519
1520 static void incrRefCount(robj *o) {
1521     o->refcount++;
1522 #ifdef DEBUG_REFCOUNT
1523     if (o->type == REDIS_STRING)
1524         printf("Increment '%s'(%p), now is: %d\n",o->ptr,o,o->refcount);
1525 #endif
1526 }
1527
1528 static void decrRefCount(void *obj) {
1529     robj *o = obj;
1530
1531 #ifdef DEBUG_REFCOUNT
1532     if (o->type == REDIS_STRING)
1533         printf("Decrement '%s'(%p), now is: %d\n",o->ptr,o,o->refcount-1);
1534 #endif
1535     if (--(o->refcount) == 0) {
1536         switch(o->type) {
1537         case REDIS_STRING: freeStringObject(o); break;
1538         case REDIS_LIST: freeListObject(o); break;
1539         case REDIS_SET: freeSetObject(o); break;
1540         case REDIS_HASH: freeHashObject(o); break;
1541         default: assert(0 != 0); break;
1542         }
1543         if (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX ||
1544             !listAddNodeHead(server.objfreelist,o))
1545             zfree(o);
1546     }
1547 }
1548
1549 /* Try to share an object against the shared objects pool */
1550 static robj *tryObjectSharing(robj *o) {
1551     struct dictEntry *de;
1552     unsigned long c;
1553
1554     if (o == NULL || server.shareobjects == 0) return o;
1555
1556     assert(o->type == REDIS_STRING);
1557     de = dictFind(server.sharingpool,o);
1558     if (de) {
1559         robj *shared = dictGetEntryKey(de);
1560
1561         c = ((unsigned long) dictGetEntryVal(de))+1;
1562         dictGetEntryVal(de) = (void*) c;
1563         incrRefCount(shared);
1564         decrRefCount(o);
1565         return shared;
1566     } else {
1567         /* Here we are using a stream algorihtm: Every time an object is
1568          * shared we increment its count, everytime there is a miss we
1569          * recrement the counter of a random object. If this object reaches
1570          * zero we remove the object and put the current object instead. */
1571         if (dictSize(server.sharingpool) >=
1572                 server.sharingpoolsize) {
1573             de = dictGetRandomKey(server.sharingpool);
1574             assert(de != NULL);
1575             c = ((unsigned long) dictGetEntryVal(de))-1;
1576             dictGetEntryVal(de) = (void*) c;
1577             if (c == 0) {
1578                 dictDelete(server.sharingpool,de->key);
1579             }
1580         } else {
1581             c = 0; /* If the pool is empty we want to add this object */
1582         }
1583         if (c == 0) {
1584             int retval;
1585
1586             retval = dictAdd(server.sharingpool,o,(void*)1);
1587             assert(retval == DICT_OK);
1588             incrRefCount(o);
1589         }
1590         return o;
1591     }
1592 }
1593
1594 static robj *lookupKey(redisDb *db, robj *key) {
1595     dictEntry *de = dictFind(db->dict,key);
1596     return de ? dictGetEntryVal(de) : NULL;
1597 }
1598
1599 static robj *lookupKeyRead(redisDb *db, robj *key) {
1600     expireIfNeeded(db,key);
1601     return lookupKey(db,key);
1602 }
1603
1604 static robj *lookupKeyWrite(redisDb *db, robj *key) {
1605     deleteIfVolatile(db,key);
1606     return lookupKey(db,key);
1607 }
1608
1609 static int deleteKey(redisDb *db, robj *key) {
1610     int retval;
1611
1612     /* We need to protect key from destruction: after the first dictDelete()
1613      * it may happen that 'key' is no longer valid if we don't increment
1614      * it's count. This may happen when we get the object reference directly
1615      * from the hash table with dictRandomKey() or dict iterators */
1616     incrRefCount(key);
1617     if (dictSize(db->expires)) dictDelete(db->expires,key);
1618     retval = dictDelete(db->dict,key);
1619     decrRefCount(key);
1620
1621     return retval == DICT_OK;
1622 }
1623
1624 /*============================ DB saving/loading ============================ */
1625
1626 static int rdbSaveType(FILE *fp, unsigned char type) {
1627     if (fwrite(&type,1,1,fp) == 0) return -1;
1628     return 0;
1629 }
1630
1631 static int rdbSaveTime(FILE *fp, time_t t) {
1632     int32_t t32 = (int32_t) t;
1633     if (fwrite(&t32,4,1,fp) == 0) return -1;
1634     return 0;
1635 }
1636
1637 /* check rdbLoadLen() comments for more info */
1638 static int rdbSaveLen(FILE *fp, uint32_t len) {
1639     unsigned char buf[2];
1640
1641     if (len < (1<<6)) {
1642         /* Save a 6 bit len */
1643         buf[0] = (len&0xFF)|(REDIS_RDB_6BITLEN<<6);
1644         if (fwrite(buf,1,1,fp) == 0) return -1;
1645     } else if (len < (1<<14)) {
1646         /* Save a 14 bit len */
1647         buf[0] = ((len>>8)&0xFF)|(REDIS_RDB_14BITLEN<<6);
1648         buf[1] = len&0xFF;
1649         if (fwrite(buf,2,1,fp) == 0) return -1;
1650     } else {
1651         /* Save a 32 bit len */
1652         buf[0] = (REDIS_RDB_32BITLEN<<6);
1653         if (fwrite(buf,1,1,fp) == 0) return -1;
1654         len = htonl(len);
1655         if (fwrite(&len,4,1,fp) == 0) return -1;
1656     }
1657     return 0;
1658 }
1659
1660 /* String objects in the form "2391" "-100" without any space and with a
1661  * range of values that can fit in an 8, 16 or 32 bit signed value can be
1662  * encoded as integers to save space */
1663 int rdbTryIntegerEncoding(sds s, unsigned char *enc) {
1664     long long value;
1665     char *endptr, buf[32];
1666
1667     /* Check if it's possible to encode this value as a number */
1668     value = strtoll(s, &endptr, 10);
1669     if (endptr[0] != '\0') return 0;
1670     snprintf(buf,32,"%lld",value);
1671
1672     /* If the number converted back into a string is not identical
1673      * then it's not possible to encode the string as integer */
1674     if (strlen(buf) != sdslen(s) || memcmp(buf,s,sdslen(s))) return 0;
1675
1676     /* Finally check if it fits in our ranges */
1677     if (value >= -(1<<7) && value <= (1<<7)-1) {
1678         enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT8;
1679         enc[1] = value&0xFF;
1680         return 2;
1681     } else if (value >= -(1<<15) && value <= (1<<15)-1) {
1682         enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT16;
1683         enc[1] = value&0xFF;
1684         enc[2] = (value>>8)&0xFF;
1685         return 3;
1686     } else if (value >= -((long long)1<<31) && value <= ((long long)1<<31)-1) {
1687         enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT32;
1688         enc[1] = value&0xFF;
1689         enc[2] = (value>>8)&0xFF;
1690         enc[3] = (value>>16)&0xFF;
1691         enc[4] = (value>>24)&0xFF;
1692         return 5;
1693     } else {
1694         return 0;
1695     }
1696 }
1697
1698 static int rdbSaveLzfStringObject(FILE *fp, robj *obj) {
1699     unsigned int comprlen, outlen;
1700     unsigned char byte;
1701     void *out;
1702
1703     /* We require at least four bytes compression for this to be worth it */
1704     outlen = sdslen(obj->ptr)-4;
1705     if (outlen <= 0) return 0;
1706     if ((out = zmalloc(outlen)) == NULL) return 0;
1707     comprlen = lzf_compress(obj->ptr, sdslen(obj->ptr), out, outlen);
1708     if (comprlen == 0) {
1709         zfree(out);
1710         return 0;
1711     }
1712     /* Data compressed! Let's save it on disk */
1713     byte = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_LZF;
1714     if (fwrite(&byte,1,1,fp) == 0) goto writeerr;
1715     if (rdbSaveLen(fp,comprlen) == -1) goto writeerr;
1716     if (rdbSaveLen(fp,sdslen(obj->ptr)) == -1) goto writeerr;
1717     if (fwrite(out,comprlen,1,fp) == 0) goto writeerr;
1718     zfree(out);
1719     return comprlen;
1720
1721 writeerr:
1722     zfree(out);
1723     return -1;
1724 }
1725
1726 /* Save a string objet as [len][data] on disk. If the object is a string
1727  * representation of an integer value we try to safe it in a special form */
1728 static int rdbSaveStringObject(FILE *fp, robj *obj) {
1729     size_t len = sdslen(obj->ptr);
1730     int enclen;
1731
1732     /* Try integer encoding */
1733     if (len <= 11) {
1734         unsigned char buf[5];
1735         if ((enclen = rdbTryIntegerEncoding(obj->ptr,buf)) > 0) {
1736             if (fwrite(buf,enclen,1,fp) == 0) return -1;
1737             return 0;
1738         }
1739     }
1740
1741     /* Try LZF compression - under 20 bytes it's unable to compress even
1742      * aaaaaaaaaaaaaaaaaa so skip it */
1743     if (len > 20) {
1744         int retval;
1745
1746         retval = rdbSaveLzfStringObject(fp,obj);
1747         if (retval == -1) return -1;
1748         if (retval > 0) return 0;
1749         /* retval == 0 means data can't be compressed, save the old way */
1750     }
1751
1752     /* Store verbatim */
1753     if (rdbSaveLen(fp,len) == -1) return -1;
1754     if (len && fwrite(obj->ptr,len,1,fp) == 0) return -1;
1755     return 0;
1756 }
1757
1758 /* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success */
1759 static int rdbSave(char *filename) {
1760     dictIterator *di = NULL;
1761     dictEntry *de;
1762     FILE *fp;
1763     char tmpfile[256];
1764     int j;
1765     time_t now = time(NULL);
1766
1767     snprintf(tmpfile,256,"temp-%d.%ld.rdb",(int)time(NULL),(long int)random());
1768     fp = fopen(tmpfile,"w");
1769     if (!fp) {
1770         redisLog(REDIS_WARNING, "Failed saving the DB: %s", strerror(errno));
1771         return REDIS_ERR;
1772     }
1773     if (fwrite("REDIS0001",9,1,fp) == 0) goto werr;
1774     for (j = 0; j < server.dbnum; j++) {
1775         redisDb *db = server.db+j;
1776         dict *d = db->dict;
1777         if (dictSize(d) == 0) continue;
1778         di = dictGetIterator(d);
1779         if (!di) {
1780             fclose(fp);
1781             return REDIS_ERR;
1782         }
1783
1784         /* Write the SELECT DB opcode */
1785         if (rdbSaveType(fp,REDIS_SELECTDB) == -1) goto werr;
1786         if (rdbSaveLen(fp,j) == -1) goto werr;
1787
1788         /* Iterate this DB writing every entry */
1789         while((de = dictNext(di)) != NULL) {
1790             robj *key = dictGetEntryKey(de);
1791             robj *o = dictGetEntryVal(de);
1792             time_t expiretime = getExpire(db,key);
1793
1794             /* Save the expire time */
1795             if (expiretime != -1) {
1796                 /* If this key is already expired skip it */
1797                 if (expiretime < now) continue;
1798                 if (rdbSaveType(fp,REDIS_EXPIRETIME) == -1) goto werr;
1799                 if (rdbSaveTime(fp,expiretime) == -1) goto werr;
1800             }
1801             /* Save the key and associated value */
1802             if (rdbSaveType(fp,o->type) == -1) goto werr;
1803             if (rdbSaveStringObject(fp,key) == -1) goto werr;
1804             if (o->type == REDIS_STRING) {
1805                 /* Save a string value */
1806                 if (rdbSaveStringObject(fp,o) == -1) goto werr;
1807             } else if (o->type == REDIS_LIST) {
1808                 /* Save a list value */
1809                 list *list = o->ptr;
1810                 listNode *ln;
1811
1812                 listRewind(list);
1813                 if (rdbSaveLen(fp,listLength(list)) == -1) goto werr;
1814                 while((ln = listYield(list))) {
1815                     robj *eleobj = listNodeValue(ln);
1816
1817                     if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
1818                 }
1819             } else if (o->type == REDIS_SET) {
1820                 /* Save a set value */
1821                 dict *set = o->ptr;
1822                 dictIterator *di = dictGetIterator(set);
1823                 dictEntry *de;
1824
1825                 if (!set) oom("dictGetIteraotr");
1826                 if (rdbSaveLen(fp,dictSize(set)) == -1) goto werr;
1827                 while((de = dictNext(di)) != NULL) {
1828                     robj *eleobj = dictGetEntryKey(de);
1829
1830                     if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
1831                 }
1832                 dictReleaseIterator(di);
1833             } else {
1834                 assert(0 != 0);
1835             }
1836         }
1837         dictReleaseIterator(di);
1838     }
1839     /* EOF opcode */
1840     if (rdbSaveType(fp,REDIS_EOF) == -1) goto werr;
1841
1842     /* Make sure data will not remain on the OS's output buffers */
1843     fflush(fp);
1844     fsync(fileno(fp));
1845     fclose(fp);
1846
1847     /* Use RENAME to make sure the DB file is changed atomically only
1848      * if the generate DB file is ok. */
1849     if (rename(tmpfile,filename) == -1) {
1850         redisLog(REDIS_WARNING,"Error moving temp DB file on the final destionation: %s", strerror(errno));
1851         unlink(tmpfile);
1852         return REDIS_ERR;
1853     }
1854     redisLog(REDIS_NOTICE,"DB saved on disk");
1855     server.dirty = 0;
1856     server.lastsave = time(NULL);
1857     return REDIS_OK;
1858
1859 werr:
1860     fclose(fp);
1861     unlink(tmpfile);
1862     redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
1863     if (di) dictReleaseIterator(di);
1864     return REDIS_ERR;
1865 }
1866
1867 static int rdbSaveBackground(char *filename) {
1868     pid_t childpid;
1869
1870     if (server.bgsaveinprogress) return REDIS_ERR;
1871     if ((childpid = fork()) == 0) {
1872         /* Child */
1873         close(server.fd);
1874         if (rdbSave(filename) == REDIS_OK) {
1875             exit(0);
1876         } else {
1877             exit(1);
1878         }
1879     } else {
1880         /* Parent */
1881         redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
1882         server.bgsaveinprogress = 1;
1883         return REDIS_OK;
1884     }
1885     return REDIS_OK; /* unreached */
1886 }
1887
1888 static int rdbLoadType(FILE *fp) {
1889     unsigned char type;
1890     if (fread(&type,1,1,fp) == 0) return -1;
1891     return type;
1892 }
1893
1894 static time_t rdbLoadTime(FILE *fp) {
1895     int32_t t32;
1896     if (fread(&t32,4,1,fp) == 0) return -1;
1897     return (time_t) t32;
1898 }
1899
1900 /* Load an encoded length from the DB, see the REDIS_RDB_* defines on the top
1901  * of this file for a description of how this are stored on disk.
1902  *
1903  * isencoded is set to 1 if the readed length is not actually a length but
1904  * an "encoding type", check the above comments for more info */
1905 static uint32_t rdbLoadLen(FILE *fp, int rdbver, int *isencoded) {
1906     unsigned char buf[2];
1907     uint32_t len;
1908
1909     if (isencoded) *isencoded = 0;
1910     if (rdbver == 0) {
1911         if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
1912         return ntohl(len);
1913     } else {
1914         int type;
1915
1916         if (fread(buf,1,1,fp) == 0) return REDIS_RDB_LENERR;
1917         type = (buf[0]&0xC0)>>6;
1918         if (type == REDIS_RDB_6BITLEN) {
1919             /* Read a 6 bit len */
1920             return buf[0]&0x3F;
1921         } else if (type == REDIS_RDB_ENCVAL) {
1922             /* Read a 6 bit len encoding type */
1923             if (isencoded) *isencoded = 1;
1924             return buf[0]&0x3F;
1925         } else if (type == REDIS_RDB_14BITLEN) {
1926             /* Read a 14 bit len */
1927             if (fread(buf+1,1,1,fp) == 0) return REDIS_RDB_LENERR;
1928             return ((buf[0]&0x3F)<<8)|buf[1];
1929         } else {
1930             /* Read a 32 bit len */
1931             if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
1932             return ntohl(len);
1933         }
1934     }
1935 }
1936
1937 static robj *rdbLoadIntegerObject(FILE *fp, int enctype) {
1938     unsigned char enc[4];
1939     long long val;
1940
1941     if (enctype == REDIS_RDB_ENC_INT8) {
1942         if (fread(enc,1,1,fp) == 0) return NULL;
1943         val = (signed char)enc[0];
1944     } else if (enctype == REDIS_RDB_ENC_INT16) {
1945         uint16_t v;
1946         if (fread(enc,2,1,fp) == 0) return NULL;
1947         v = enc[0]|(enc[1]<<8);
1948         val = (int16_t)v;
1949     } else if (enctype == REDIS_RDB_ENC_INT32) {
1950         uint32_t v;
1951         if (fread(enc,4,1,fp) == 0) return NULL;
1952         v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24);
1953         val = (int32_t)v;
1954     } else {
1955         val = 0; /* anti-warning */
1956         assert(0!=0);
1957     }
1958     return createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",val));
1959 }
1960
1961 static robj *rdbLoadLzfStringObject(FILE*fp, int rdbver) {
1962     unsigned int len, clen;
1963     unsigned char *c = NULL;
1964     sds val = NULL;
1965
1966     if ((clen = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR) return NULL;
1967     if ((len = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR) return NULL;
1968     if ((c = zmalloc(clen)) == NULL) goto err;
1969     if ((val = sdsnewlen(NULL,len)) == NULL) goto err;
1970     if (fread(c,clen,1,fp) == 0) goto err;
1971     if (lzf_decompress(c,clen,val,len) == 0) goto err;
1972     return createObject(REDIS_STRING,val);
1973 err:
1974     zfree(c);
1975     sdsfree(val);
1976     return NULL;
1977 }
1978
1979 static robj *rdbLoadStringObject(FILE*fp, int rdbver) {
1980     int isencoded;
1981     uint32_t len;
1982     sds val;
1983
1984     len = rdbLoadLen(fp,rdbver,&isencoded);
1985     if (isencoded) {
1986         switch(len) {
1987         case REDIS_RDB_ENC_INT8:
1988         case REDIS_RDB_ENC_INT16:
1989         case REDIS_RDB_ENC_INT32:
1990             return tryObjectSharing(rdbLoadIntegerObject(fp,len));
1991         case REDIS_RDB_ENC_LZF:
1992             return tryObjectSharing(rdbLoadLzfStringObject(fp,rdbver));
1993         default:
1994             assert(0!=0);
1995         }
1996     }
1997
1998     if (len == REDIS_RDB_LENERR) return NULL;
1999     val = sdsnewlen(NULL,len);
2000     if (len && fread(val,len,1,fp) == 0) {
2001         sdsfree(val);
2002         return NULL;
2003     }
2004     return tryObjectSharing(createObject(REDIS_STRING,val));
2005 }
2006
2007 static int rdbLoad(char *filename) {
2008     FILE *fp;
2009     robj *keyobj = NULL;
2010     uint32_t dbid;
2011     int type, retval, rdbver;
2012     dict *d = server.db[0].dict;
2013     redisDb *db = server.db+0;
2014     char buf[1024];
2015     time_t expiretime = -1, now = time(NULL);
2016
2017     fp = fopen(filename,"r");
2018     if (!fp) return REDIS_ERR;
2019     if (fread(buf,9,1,fp) == 0) goto eoferr;
2020     buf[9] = '\0';
2021     if (memcmp(buf,"REDIS",5) != 0) {
2022         fclose(fp);
2023         redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
2024         return REDIS_ERR;
2025     }
2026     rdbver = atoi(buf+5);
2027     if (rdbver > 1) {
2028         fclose(fp);
2029         redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
2030         return REDIS_ERR;
2031     }
2032     while(1) {
2033         robj *o;
2034
2035         /* Read type. */
2036         if ((type = rdbLoadType(fp)) == -1) goto eoferr;
2037         if (type == REDIS_EXPIRETIME) {
2038             if ((expiretime = rdbLoadTime(fp)) == -1) goto eoferr;
2039             /* We read the time so we need to read the object type again */
2040             if ((type = rdbLoadType(fp)) == -1) goto eoferr;
2041         }
2042         if (type == REDIS_EOF) break;
2043         /* Handle SELECT DB opcode as a special case */
2044         if (type == REDIS_SELECTDB) {
2045             if ((dbid = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR)
2046                 goto eoferr;
2047             if (dbid >= (unsigned)server.dbnum) {
2048                 redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
2049                 exit(1);
2050             }
2051             db = server.db+dbid;
2052             d = db->dict;
2053             continue;
2054         }
2055         /* Read key */
2056         if ((keyobj = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
2057
2058         if (type == REDIS_STRING) {
2059             /* Read string value */
2060             if ((o = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
2061         } else if (type == REDIS_LIST || type == REDIS_SET) {
2062             /* Read list/set value */
2063             uint32_t listlen;
2064
2065             if ((listlen = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR)
2066                 goto eoferr;
2067             o = (type == REDIS_LIST) ? createListObject() : createSetObject();
2068             /* Load every single element of the list/set */
2069             while(listlen--) {
2070                 robj *ele;
2071
2072                 if ((ele = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
2073                 if (type == REDIS_LIST) {
2074                     if (!listAddNodeTail((list*)o->ptr,ele))
2075                         oom("listAddNodeTail");
2076                 } else {
2077                     if (dictAdd((dict*)o->ptr,ele,NULL) == DICT_ERR)
2078                         oom("dictAdd");
2079                 }
2080             }
2081         } else {
2082             assert(0 != 0);
2083         }
2084         /* Add the new object in the hash table */
2085         retval = dictAdd(d,keyobj,o);
2086         if (retval == DICT_ERR) {
2087             redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", keyobj->ptr);
2088             exit(1);
2089         }
2090         /* Set the expire time if needed */
2091         if (expiretime != -1) {
2092             setExpire(db,keyobj,expiretime);
2093             /* Delete this key if already expired */
2094             if (expiretime < now) deleteKey(db,keyobj);
2095             expiretime = -1;
2096         }
2097         keyobj = o = NULL;
2098     }
2099     fclose(fp);
2100     return REDIS_OK;
2101
2102 eoferr: /* unexpected end of file is handled here with a fatal exit */
2103     if (keyobj) decrRefCount(keyobj);
2104     redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, exiting now.");
2105     exit(1);
2106     return REDIS_ERR; /* Just to avoid warning */
2107 }
2108
2109 /*================================== Commands =============================== */
2110
2111 static void authCommand(redisClient *c) {
2112     if (!server.requirepass || !strcmp(c->argv[1]->ptr, server.requirepass)) {
2113       c->authenticated = 1;
2114       addReply(c,shared.ok);
2115     } else {
2116       c->authenticated = 0;
2117       addReply(c,shared.err);
2118     }
2119 }
2120
2121 static void pingCommand(redisClient *c) {
2122     addReply(c,shared.pong);
2123 }
2124
2125 static void echoCommand(redisClient *c) {
2126     addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
2127         (int)sdslen(c->argv[1]->ptr)));
2128     addReply(c,c->argv[1]);
2129     addReply(c,shared.crlf);
2130 }
2131
2132 /*=================================== Strings =============================== */
2133
2134 static void setGenericCommand(redisClient *c, int nx) {
2135     int retval;
2136
2137     retval = dictAdd(c->db->dict,c->argv[1],c->argv[2]);
2138     if (retval == DICT_ERR) {
2139         if (!nx) {
2140             dictReplace(c->db->dict,c->argv[1],c->argv[2]);
2141             incrRefCount(c->argv[2]);
2142         } else {
2143             addReply(c,shared.czero);
2144             return;
2145         }
2146     } else {
2147         incrRefCount(c->argv[1]);
2148         incrRefCount(c->argv[2]);
2149     }
2150     server.dirty++;
2151     removeExpire(c->db,c->argv[1]);
2152     addReply(c, nx ? shared.cone : shared.ok);
2153 }
2154
2155 static void setCommand(redisClient *c) {
2156     setGenericCommand(c,0);
2157 }
2158
2159 static void setnxCommand(redisClient *c) {
2160     setGenericCommand(c,1);
2161 }
2162
2163 static void getCommand(redisClient *c) {
2164     robj *o = lookupKeyRead(c->db,c->argv[1]);
2165
2166     if (o == NULL) {
2167         addReply(c,shared.nullbulk);
2168     } else {
2169         if (o->type != REDIS_STRING) {
2170             addReply(c,shared.wrongtypeerr);
2171         } else {
2172             addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(o->ptr)));
2173             addReply(c,o);
2174             addReply(c,shared.crlf);
2175         }
2176     }
2177 }
2178
2179 static void getSetCommand(redisClient *c) {
2180     getCommand(c);
2181     if (dictAdd(c->db->dict,c->argv[1],c->argv[2]) == DICT_ERR) {
2182         dictReplace(c->db->dict,c->argv[1],c->argv[2]);
2183     } else {
2184         incrRefCount(c->argv[1]);
2185     }
2186     incrRefCount(c->argv[2]);
2187     server.dirty++;
2188     removeExpire(c->db,c->argv[1]);
2189 }
2190
2191 static void mgetCommand(redisClient *c) {
2192     int j;
2193
2194     addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",c->argc-1));
2195     for (j = 1; j < c->argc; j++) {
2196         robj *o = lookupKeyRead(c->db,c->argv[j]);
2197         if (o == NULL) {
2198             addReply(c,shared.nullbulk);
2199         } else {
2200             if (o->type != REDIS_STRING) {
2201                 addReply(c,shared.nullbulk);
2202             } else {
2203                 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(o->ptr)));
2204                 addReply(c,o);
2205                 addReply(c,shared.crlf);
2206             }
2207         }
2208     }
2209 }
2210
2211 static void incrDecrCommand(redisClient *c, long long incr) {
2212     long long value;
2213     int retval;
2214     robj *o;
2215
2216     o = lookupKeyWrite(c->db,c->argv[1]);
2217     if (o == NULL) {
2218         value = 0;
2219     } else {
2220         if (o->type != REDIS_STRING) {
2221             value = 0;
2222         } else {
2223             char *eptr;
2224
2225             value = strtoll(o->ptr, &eptr, 10);
2226         }
2227     }
2228
2229     value += incr;
2230     o = createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",value));
2231     retval = dictAdd(c->db->dict,c->argv[1],o);
2232     if (retval == DICT_ERR) {
2233         dictReplace(c->db->dict,c->argv[1],o);
2234         removeExpire(c->db,c->argv[1]);
2235     } else {
2236         incrRefCount(c->argv[1]);
2237     }
2238     server.dirty++;
2239     addReply(c,shared.colon);
2240     addReply(c,o);
2241     addReply(c,shared.crlf);
2242 }
2243
2244 static void incrCommand(redisClient *c) {
2245     incrDecrCommand(c,1);
2246 }
2247
2248 static void decrCommand(redisClient *c) {
2249     incrDecrCommand(c,-1);
2250 }
2251
2252 static void incrbyCommand(redisClient *c) {
2253     long long incr = strtoll(c->argv[2]->ptr, NULL, 10);
2254     incrDecrCommand(c,incr);
2255 }
2256
2257 static void decrbyCommand(redisClient *c) {
2258     long long incr = strtoll(c->argv[2]->ptr, NULL, 10);
2259     incrDecrCommand(c,-incr);
2260 }
2261
2262 /* ========================= Type agnostic commands ========================= */
2263
2264 static void delCommand(redisClient *c) {
2265     if (deleteKey(c->db,c->argv[1])) {
2266         server.dirty++;
2267         addReply(c,shared.cone);
2268     } else {
2269         addReply(c,shared.czero);
2270     }
2271 }
2272
2273 static void existsCommand(redisClient *c) {
2274     addReply(c,lookupKeyRead(c->db,c->argv[1]) ? shared.cone : shared.czero);
2275 }
2276
2277 static void selectCommand(redisClient *c) {
2278     int id = atoi(c->argv[1]->ptr);
2279
2280     if (selectDb(c,id) == REDIS_ERR) {
2281         addReplySds(c,sdsnew("-ERR invalid DB index\r\n"));
2282     } else {
2283         addReply(c,shared.ok);
2284     }
2285 }
2286
2287 static void randomkeyCommand(redisClient *c) {
2288     dictEntry *de;
2289
2290     while(1) {
2291         de = dictGetRandomKey(c->db->dict);
2292         if (!de || expireIfNeeded(c->db,dictGetEntryKey(de)) == 0) break;
2293     }
2294     if (de == NULL) {
2295         addReply(c,shared.plus);
2296         addReply(c,shared.crlf);
2297     } else {
2298         addReply(c,shared.plus);
2299         addReply(c,dictGetEntryKey(de));
2300         addReply(c,shared.crlf);
2301     }
2302 }
2303
2304 static void keysCommand(redisClient *c) {
2305     dictIterator *di;
2306     dictEntry *de;
2307     sds pattern = c->argv[1]->ptr;
2308     int plen = sdslen(pattern);
2309     int numkeys = 0, keyslen = 0;
2310     robj *lenobj = createObject(REDIS_STRING,NULL);
2311
2312     di = dictGetIterator(c->db->dict);
2313     if (!di) oom("dictGetIterator");
2314     addReply(c,lenobj);
2315     decrRefCount(lenobj);
2316     while((de = dictNext(di)) != NULL) {
2317         robj *keyobj = dictGetEntryKey(de);
2318
2319         sds key = keyobj->ptr;
2320         if ((pattern[0] == '*' && pattern[1] == '\0') ||
2321             stringmatchlen(pattern,plen,key,sdslen(key),0)) {
2322             if (expireIfNeeded(c->db,keyobj) == 0) {
2323                 if (numkeys != 0)
2324                     addReply(c,shared.space);
2325                 addReply(c,keyobj);
2326                 numkeys++;
2327                 keyslen += sdslen(key);
2328             }
2329         }
2330     }
2331     dictReleaseIterator(di);
2332     lenobj->ptr = sdscatprintf(sdsempty(),"$%lu\r\n",keyslen+(numkeys ? (numkeys-1) : 0));
2333     addReply(c,shared.crlf);
2334 }
2335
2336 static void dbsizeCommand(redisClient *c) {
2337     addReplySds(c,
2338         sdscatprintf(sdsempty(),":%lu\r\n",dictSize(c->db->dict)));
2339 }
2340
2341 static void lastsaveCommand(redisClient *c) {
2342     addReplySds(c,
2343         sdscatprintf(sdsempty(),":%lu\r\n",server.lastsave));
2344 }
2345
2346 static void typeCommand(redisClient *c) {
2347     robj *o;
2348     char *type;
2349
2350     o = lookupKeyRead(c->db,c->argv[1]);
2351     if (o == NULL) {
2352         type = "+none";
2353     } else {
2354         switch(o->type) {
2355         case REDIS_STRING: type = "+string"; break;
2356         case REDIS_LIST: type = "+list"; break;
2357         case REDIS_SET: type = "+set"; break;
2358         default: type = "unknown"; break;
2359         }
2360     }
2361     addReplySds(c,sdsnew(type));
2362     addReply(c,shared.crlf);
2363 }
2364
2365 static void saveCommand(redisClient *c) {
2366     if (server.bgsaveinprogress) {
2367         addReplySds(c,sdsnew("-ERR background save in progress\r\n"));
2368         return;
2369     }
2370     if (rdbSave(server.dbfilename) == REDIS_OK) {
2371         addReply(c,shared.ok);
2372     } else {
2373         addReply(c,shared.err);
2374     }
2375 }
2376
2377 static void bgsaveCommand(redisClient *c) {
2378     if (server.bgsaveinprogress) {
2379         addReplySds(c,sdsnew("-ERR background save already in progress\r\n"));
2380         return;
2381     }
2382     if (rdbSaveBackground(server.dbfilename) == REDIS_OK) {
2383         addReply(c,shared.ok);
2384     } else {
2385         addReply(c,shared.err);
2386     }
2387 }
2388
2389 static void shutdownCommand(redisClient *c) {
2390     redisLog(REDIS_WARNING,"User requested shutdown, saving DB...");
2391     /* XXX: TODO kill the child if there is a bgsave in progress */
2392     if (rdbSave(server.dbfilename) == REDIS_OK) {
2393         if (server.daemonize) {
2394           unlink(server.pidfile);
2395         }
2396         redisLog(REDIS_WARNING,"Server exit now, bye bye...");
2397         exit(1);
2398     } else {
2399         redisLog(REDIS_WARNING,"Error trying to save the DB, can't exit");
2400         addReplySds(c,sdsnew("-ERR can't quit, problems saving the DB\r\n"));
2401     }
2402 }
2403
2404 static void renameGenericCommand(redisClient *c, int nx) {
2405     robj *o;
2406
2407     /* To use the same key as src and dst is probably an error */
2408     if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) {
2409         addReply(c,shared.sameobjecterr);
2410         return;
2411     }
2412
2413     o = lookupKeyWrite(c->db,c->argv[1]);
2414     if (o == NULL) {
2415         addReply(c,shared.nokeyerr);
2416         return;
2417     }
2418     incrRefCount(o);
2419     deleteIfVolatile(c->db,c->argv[2]);
2420     if (dictAdd(c->db->dict,c->argv[2],o) == DICT_ERR) {
2421         if (nx) {
2422             decrRefCount(o);
2423             addReply(c,shared.czero);
2424             return;
2425         }
2426         dictReplace(c->db->dict,c->argv[2],o);
2427     } else {
2428         incrRefCount(c->argv[2]);
2429     }
2430     deleteKey(c->db,c->argv[1]);
2431     server.dirty++;
2432     addReply(c,nx ? shared.cone : shared.ok);
2433 }
2434
2435 static void renameCommand(redisClient *c) {
2436     renameGenericCommand(c,0);
2437 }
2438
2439 static void renamenxCommand(redisClient *c) {
2440     renameGenericCommand(c,1);
2441 }
2442
2443 static void moveCommand(redisClient *c) {
2444     robj *o;
2445     redisDb *src, *dst;
2446     int srcid;
2447
2448     /* Obtain source and target DB pointers */
2449     src = c->db;
2450     srcid = c->db->id;
2451     if (selectDb(c,atoi(c->argv[2]->ptr)) == REDIS_ERR) {
2452         addReply(c,shared.outofrangeerr);
2453         return;
2454     }
2455     dst = c->db;
2456     selectDb(c,srcid); /* Back to the source DB */
2457
2458     /* If the user is moving using as target the same
2459      * DB as the source DB it is probably an error. */
2460     if (src == dst) {
2461         addReply(c,shared.sameobjecterr);
2462         return;
2463     }
2464
2465     /* Check if the element exists and get a reference */
2466     o = lookupKeyWrite(c->db,c->argv[1]);
2467     if (!o) {
2468         addReply(c,shared.czero);
2469         return;
2470     }
2471
2472     /* Try to add the element to the target DB */
2473     deleteIfVolatile(dst,c->argv[1]);
2474     if (dictAdd(dst->dict,c->argv[1],o) == DICT_ERR) {
2475         addReply(c,shared.czero);
2476         return;
2477     }
2478     incrRefCount(c->argv[1]);
2479     incrRefCount(o);
2480
2481     /* OK! key moved, free the entry in the source DB */
2482     deleteKey(src,c->argv[1]);
2483     server.dirty++;
2484     addReply(c,shared.cone);
2485 }
2486
2487 /* =================================== Lists ================================ */
2488 static void pushGenericCommand(redisClient *c, int where) {
2489     robj *lobj;
2490     list *list;
2491
2492     lobj = lookupKeyWrite(c->db,c->argv[1]);
2493     if (lobj == NULL) {
2494         lobj = createListObject();
2495         list = lobj->ptr;
2496         if (where == REDIS_HEAD) {
2497             if (!listAddNodeHead(list,c->argv[2])) oom("listAddNodeHead");
2498         } else {
2499             if (!listAddNodeTail(list,c->argv[2])) oom("listAddNodeTail");
2500         }
2501         dictAdd(c->db->dict,c->argv[1],lobj);
2502         incrRefCount(c->argv[1]);
2503         incrRefCount(c->argv[2]);
2504     } else {
2505         if (lobj->type != REDIS_LIST) {
2506             addReply(c,shared.wrongtypeerr);
2507             return;
2508         }
2509         list = lobj->ptr;
2510         if (where == REDIS_HEAD) {
2511             if (!listAddNodeHead(list,c->argv[2])) oom("listAddNodeHead");
2512         } else {
2513             if (!listAddNodeTail(list,c->argv[2])) oom("listAddNodeTail");
2514         }
2515         incrRefCount(c->argv[2]);
2516     }
2517     server.dirty++;
2518     addReply(c,shared.ok);
2519 }
2520
2521 static void lpushCommand(redisClient *c) {
2522     pushGenericCommand(c,REDIS_HEAD);
2523 }
2524
2525 static void rpushCommand(redisClient *c) {
2526     pushGenericCommand(c,REDIS_TAIL);
2527 }
2528
2529 static void llenCommand(redisClient *c) {
2530     robj *o;
2531     list *l;
2532
2533     o = lookupKeyRead(c->db,c->argv[1]);
2534     if (o == NULL) {
2535         addReply(c,shared.czero);
2536         return;
2537     } else {
2538         if (o->type != REDIS_LIST) {
2539             addReply(c,shared.wrongtypeerr);
2540         } else {
2541             l = o->ptr;
2542             addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",listLength(l)));
2543         }
2544     }
2545 }
2546
2547 static void lindexCommand(redisClient *c) {
2548     robj *o;
2549     int index = atoi(c->argv[2]->ptr);
2550
2551     o = lookupKeyRead(c->db,c->argv[1]);
2552     if (o == NULL) {
2553         addReply(c,shared.nullbulk);
2554     } else {
2555         if (o->type != REDIS_LIST) {
2556             addReply(c,shared.wrongtypeerr);
2557         } else {
2558             list *list = o->ptr;
2559             listNode *ln;
2560
2561             ln = listIndex(list, index);
2562             if (ln == NULL) {
2563                 addReply(c,shared.nullbulk);
2564             } else {
2565                 robj *ele = listNodeValue(ln);
2566                 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
2567                 addReply(c,ele);
2568                 addReply(c,shared.crlf);
2569             }
2570         }
2571     }
2572 }
2573
2574 static void lsetCommand(redisClient *c) {
2575     robj *o;
2576     int index = atoi(c->argv[2]->ptr);
2577
2578     o = lookupKeyWrite(c->db,c->argv[1]);
2579     if (o == NULL) {
2580         addReply(c,shared.nokeyerr);
2581     } else {
2582         if (o->type != REDIS_LIST) {
2583             addReply(c,shared.wrongtypeerr);
2584         } else {
2585             list *list = o->ptr;
2586             listNode *ln;
2587
2588             ln = listIndex(list, index);
2589             if (ln == NULL) {
2590                 addReply(c,shared.outofrangeerr);
2591             } else {
2592                 robj *ele = listNodeValue(ln);
2593
2594                 decrRefCount(ele);
2595                 listNodeValue(ln) = c->argv[3];
2596                 incrRefCount(c->argv[3]);
2597                 addReply(c,shared.ok);
2598                 server.dirty++;
2599             }
2600         }
2601     }
2602 }
2603
2604 static void popGenericCommand(redisClient *c, int where) {
2605     robj *o;
2606
2607     o = lookupKeyWrite(c->db,c->argv[1]);
2608     if (o == NULL) {
2609         addReply(c,shared.nullbulk);
2610     } else {
2611         if (o->type != REDIS_LIST) {
2612             addReply(c,shared.wrongtypeerr);
2613         } else {
2614             list *list = o->ptr;
2615             listNode *ln;
2616
2617             if (where == REDIS_HEAD)
2618                 ln = listFirst(list);
2619             else
2620                 ln = listLast(list);
2621
2622             if (ln == NULL) {
2623                 addReply(c,shared.nullbulk);
2624             } else {
2625                 robj *ele = listNodeValue(ln);
2626                 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
2627                 addReply(c,ele);
2628                 addReply(c,shared.crlf);
2629                 listDelNode(list,ln);
2630                 server.dirty++;
2631             }
2632         }
2633     }
2634 }
2635
2636 static void lpopCommand(redisClient *c) {
2637     popGenericCommand(c,REDIS_HEAD);
2638 }
2639
2640 static void rpopCommand(redisClient *c) {
2641     popGenericCommand(c,REDIS_TAIL);
2642 }
2643
2644 static void lrangeCommand(redisClient *c) {
2645     robj *o;
2646     int start = atoi(c->argv[2]->ptr);
2647     int end = atoi(c->argv[3]->ptr);
2648
2649     o = lookupKeyRead(c->db,c->argv[1]);
2650     if (o == NULL) {
2651         addReply(c,shared.nullmultibulk);
2652     } else {
2653         if (o->type != REDIS_LIST) {
2654             addReply(c,shared.wrongtypeerr);
2655         } else {
2656             list *list = o->ptr;
2657             listNode *ln;
2658             int llen = listLength(list);
2659             int rangelen, j;
2660             robj *ele;
2661
2662             /* convert negative indexes */
2663             if (start < 0) start = llen+start;
2664             if (end < 0) end = llen+end;
2665             if (start < 0) start = 0;
2666             if (end < 0) end = 0;
2667
2668             /* indexes sanity checks */
2669             if (start > end || start >= llen) {
2670                 /* Out of range start or start > end result in empty list */
2671                 addReply(c,shared.emptymultibulk);
2672                 return;
2673             }
2674             if (end >= llen) end = llen-1;
2675             rangelen = (end-start)+1;
2676
2677             /* Return the result in form of a multi-bulk reply */
2678             ln = listIndex(list, start);
2679             addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",rangelen));
2680             for (j = 0; j < rangelen; j++) {
2681                 ele = listNodeValue(ln);
2682                 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
2683                 addReply(c,ele);
2684                 addReply(c,shared.crlf);
2685                 ln = ln->next;
2686             }
2687         }
2688     }
2689 }
2690
2691 static void ltrimCommand(redisClient *c) {
2692     robj *o;
2693     int start = atoi(c->argv[2]->ptr);
2694     int end = atoi(c->argv[3]->ptr);
2695
2696     o = lookupKeyWrite(c->db,c->argv[1]);
2697     if (o == NULL) {
2698         addReply(c,shared.nokeyerr);
2699     } else {
2700         if (o->type != REDIS_LIST) {
2701             addReply(c,shared.wrongtypeerr);
2702         } else {
2703             list *list = o->ptr;
2704             listNode *ln;
2705             int llen = listLength(list);
2706             int j, ltrim, rtrim;
2707
2708             /* convert negative indexes */
2709             if (start < 0) start = llen+start;
2710             if (end < 0) end = llen+end;
2711             if (start < 0) start = 0;
2712             if (end < 0) end = 0;
2713
2714             /* indexes sanity checks */
2715             if (start > end || start >= llen) {
2716                 /* Out of range start or start > end result in empty list */
2717                 ltrim = llen;
2718                 rtrim = 0;
2719             } else {
2720                 if (end >= llen) end = llen-1;
2721                 ltrim = start;
2722                 rtrim = llen-end-1;
2723             }
2724
2725             /* Remove list elements to perform the trim */
2726             for (j = 0; j < ltrim; j++) {
2727                 ln = listFirst(list);
2728                 listDelNode(list,ln);
2729             }
2730             for (j = 0; j < rtrim; j++) {
2731                 ln = listLast(list);
2732                 listDelNode(list,ln);
2733             }
2734             addReply(c,shared.ok);
2735             server.dirty++;
2736         }
2737     }
2738 }
2739
2740 static void lremCommand(redisClient *c) {
2741     robj *o;
2742
2743     o = lookupKeyWrite(c->db,c->argv[1]);
2744     if (o == NULL) {
2745         addReply(c,shared.nokeyerr);
2746     } else {
2747         if (o->type != REDIS_LIST) {
2748             addReply(c,shared.wrongtypeerr);
2749         } else {
2750             list *list = o->ptr;
2751             listNode *ln, *next;
2752             int toremove = atoi(c->argv[2]->ptr);
2753             int removed = 0;
2754             int fromtail = 0;
2755
2756             if (toremove < 0) {
2757                 toremove = -toremove;
2758                 fromtail = 1;
2759             }
2760             ln = fromtail ? list->tail : list->head;
2761             while (ln) {
2762                 robj *ele = listNodeValue(ln);
2763
2764                 next = fromtail ? ln->prev : ln->next;
2765                 if (sdscmp(ele->ptr,c->argv[3]->ptr) == 0) {
2766                     listDelNode(list,ln);
2767                     server.dirty++;
2768                     removed++;
2769                     if (toremove && removed == toremove) break;
2770                 }
2771                 ln = next;
2772             }
2773             addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",removed));
2774         }
2775     }
2776 }
2777
2778 /* ==================================== Sets ================================ */
2779
2780 static void saddCommand(redisClient *c) {
2781     robj *set;
2782
2783     set = lookupKeyWrite(c->db,c->argv[1]);
2784     if (set == NULL) {
2785         set = createSetObject();
2786         dictAdd(c->db->dict,c->argv[1],set);
2787         incrRefCount(c->argv[1]);
2788     } else {
2789         if (set->type != REDIS_SET) {
2790             addReply(c,shared.wrongtypeerr);
2791             return;
2792         }
2793     }
2794     if (dictAdd(set->ptr,c->argv[2],NULL) == DICT_OK) {
2795         incrRefCount(c->argv[2]);
2796         server.dirty++;
2797         addReply(c,shared.cone);
2798     } else {
2799         addReply(c,shared.czero);
2800     }
2801 }
2802
2803 static void sremCommand(redisClient *c) {
2804     robj *set;
2805
2806     set = lookupKeyWrite(c->db,c->argv[1]);
2807     if (set == NULL) {
2808         addReply(c,shared.czero);
2809     } else {
2810         if (set->type != REDIS_SET) {
2811             addReply(c,shared.wrongtypeerr);
2812             return;
2813         }
2814         if (dictDelete(set->ptr,c->argv[2]) == DICT_OK) {
2815             server.dirty++;
2816             addReply(c,shared.cone);
2817         } else {
2818             addReply(c,shared.czero);
2819         }
2820     }
2821 }
2822
2823 static void sismemberCommand(redisClient *c) {
2824     robj *set;
2825
2826     set = lookupKeyRead(c->db,c->argv[1]);
2827     if (set == NULL) {
2828         addReply(c,shared.czero);
2829     } else {
2830         if (set->type != REDIS_SET) {
2831             addReply(c,shared.wrongtypeerr);
2832             return;
2833         }
2834         if (dictFind(set->ptr,c->argv[2]))
2835             addReply(c,shared.cone);
2836         else
2837             addReply(c,shared.czero);
2838     }
2839 }
2840
2841 static void scardCommand(redisClient *c) {
2842     robj *o;
2843     dict *s;
2844
2845     o = lookupKeyRead(c->db,c->argv[1]);
2846     if (o == NULL) {
2847         addReply(c,shared.czero);
2848         return;
2849     } else {
2850         if (o->type != REDIS_SET) {
2851             addReply(c,shared.wrongtypeerr);
2852         } else {
2853             s = o->ptr;
2854             addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",
2855                 dictSize(s)));
2856         }
2857     }
2858 }
2859
2860 static int qsortCompareSetsByCardinality(const void *s1, const void *s2) {
2861     dict **d1 = (void*) s1, **d2 = (void*) s2;
2862
2863     return dictSize(*d1)-dictSize(*d2);
2864 }
2865
2866 static void sinterGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey) {
2867     dict **dv = zmalloc(sizeof(dict*)*setsnum);
2868     dictIterator *di;
2869     dictEntry *de;
2870     robj *lenobj = NULL, *dstset = NULL;
2871     int j, cardinality = 0;
2872
2873     if (!dv) oom("sinterCommand");
2874     for (j = 0; j < setsnum; j++) {
2875         robj *setobj;
2876
2877         setobj = dstkey ?
2878                     lookupKeyWrite(c->db,setskeys[j]) :
2879                     lookupKeyRead(c->db,setskeys[j]);
2880         if (!setobj) {
2881             zfree(dv);
2882             if (dstkey) {
2883                 deleteKey(c->db,dstkey);
2884                 addReply(c,shared.ok);
2885             } else {
2886                 addReply(c,shared.nullmultibulk);
2887             }
2888             return;
2889         }
2890         if (setobj->type != REDIS_SET) {
2891             zfree(dv);
2892             addReply(c,shared.wrongtypeerr);
2893             return;
2894         }
2895         dv[j] = setobj->ptr;
2896     }
2897     /* Sort sets from the smallest to largest, this will improve our
2898      * algorithm's performace */
2899     qsort(dv,setsnum,sizeof(dict*),qsortCompareSetsByCardinality);
2900
2901     /* The first thing we should output is the total number of elements...
2902      * since this is a multi-bulk write, but at this stage we don't know
2903      * the intersection set size, so we use a trick, append an empty object
2904      * to the output list and save the pointer to later modify it with the
2905      * right length */
2906     if (!dstkey) {
2907         lenobj = createObject(REDIS_STRING,NULL);
2908         addReply(c,lenobj);
2909         decrRefCount(lenobj);
2910     } else {
2911         /* If we have a target key where to store the resulting set
2912          * create this key with an empty set inside */
2913         dstset = createSetObject();
2914         deleteKey(c->db,dstkey);
2915         dictAdd(c->db->dict,dstkey,dstset);
2916         incrRefCount(dstkey);
2917     }
2918
2919     /* Iterate all the elements of the first (smallest) set, and test
2920      * the element against all the other sets, if at least one set does
2921      * not include the element it is discarded */
2922     di = dictGetIterator(dv[0]);
2923     if (!di) oom("dictGetIterator");
2924
2925     while((de = dictNext(di)) != NULL) {
2926         robj *ele;
2927
2928         for (j = 1; j < setsnum; j++)
2929             if (dictFind(dv[j],dictGetEntryKey(de)) == NULL) break;
2930         if (j != setsnum)
2931             continue; /* at least one set does not contain the member */
2932         ele = dictGetEntryKey(de);
2933         if (!dstkey) {
2934             addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(ele->ptr)));
2935             addReply(c,ele);
2936             addReply(c,shared.crlf);
2937             cardinality++;
2938         } else {
2939             dictAdd(dstset->ptr,ele,NULL);
2940             incrRefCount(ele);
2941         }
2942     }
2943     dictReleaseIterator(di);
2944
2945     if (!dstkey) {
2946         lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality);
2947     } else {
2948         addReply(c,shared.ok);
2949         server.dirty++;
2950     }
2951     zfree(dv);
2952 }
2953
2954 static void sinterCommand(redisClient *c) {
2955     sinterGenericCommand(c,c->argv+1,c->argc-1,NULL);
2956 }
2957
2958 static void sinterstoreCommand(redisClient *c) {
2959     sinterGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
2960 }
2961
2962 static void sunionGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey) {
2963     dict **dv = zmalloc(sizeof(dict*)*setsnum);
2964     dictIterator *di;
2965     dictEntry *de;
2966     robj *lenobj = NULL, *dstset = NULL;
2967     int j, cardinality = 0;
2968
2969     if (!dv) oom("sunionCommand");
2970     for (j = 0; j < setsnum; j++) {
2971         robj *setobj;
2972
2973         setobj = dstkey ?
2974                     lookupKeyWrite(c->db,setskeys[j]) :
2975                     lookupKeyRead(c->db,setskeys[j]);
2976         if (!setobj) {
2977             dv[j] = NULL;
2978             continue;
2979         }
2980         if (setobj->type != REDIS_SET) {
2981             zfree(dv);
2982             addReply(c,shared.wrongtypeerr);
2983             return;
2984         }
2985         dv[j] = setobj->ptr;
2986     }
2987
2988     /* We need a temp set object to store our union. If the dstkey
2989      * is not NULL (that is, we are inside an SUNIONSTORE operation) then
2990      * this set object will be the resulting object to set into the target key*/
2991     dstset = createSetObject();
2992
2993     /* The first thing we should output is the total number of elements...
2994      * since this is a multi-bulk write, but at this stage we don't know
2995      * the intersection set size, so we use a trick, append an empty object
2996      * to the output list and save the pointer to later modify it with the
2997      * right length */
2998     if (!dstkey) {
2999         lenobj = createObject(REDIS_STRING,NULL);
3000         addReply(c,lenobj);
3001         decrRefCount(lenobj);
3002     } else {
3003         /* If we have a target key where to store the resulting set
3004          * create this key with an empty set inside */
3005         deleteKey(c->db,dstkey);
3006         dictAdd(c->db->dict,dstkey,dstset);
3007         incrRefCount(dstkey);
3008         server.dirty++;
3009     }
3010
3011     /* Iterate all the elements of all the sets, add every element a single
3012      * time to the result set */
3013     for (j = 0; j < setsnum; j++) {
3014         if (!dv[j]) continue; /* non existing keys are like empty sets */
3015
3016         di = dictGetIterator(dv[j]);
3017         if (!di) oom("dictGetIterator");
3018
3019         while((de = dictNext(di)) != NULL) {
3020             robj *ele;
3021
3022             /* dictAdd will not add the same element multiple times */
3023             ele = dictGetEntryKey(de);
3024             if (dictAdd(dstset->ptr,ele,NULL) == DICT_OK) {
3025                 incrRefCount(ele);
3026                 if (!dstkey) {
3027                     addReplySds(c,sdscatprintf(sdsempty(),
3028                             "$%d\r\n",sdslen(ele->ptr)));
3029                     addReply(c,ele);
3030                     addReply(c,shared.crlf);
3031                     cardinality++;
3032                 }
3033             }
3034         }
3035         dictReleaseIterator(di);
3036     }
3037
3038     if (!dstkey) {
3039         lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality);
3040         decrRefCount(dstset);
3041     } else {
3042         addReply(c,shared.ok);
3043         server.dirty++;
3044     }
3045     zfree(dv);
3046 }
3047
3048 static void sunionCommand(redisClient *c) {
3049     sunionGenericCommand(c,c->argv+1,c->argc-1,NULL);
3050 }
3051
3052 static void sunionstoreCommand(redisClient *c) {
3053     sunionGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
3054 }
3055
3056 static void flushdbCommand(redisClient *c) {
3057     server.dirty += dictSize(c->db->dict);
3058     dictEmpty(c->db->dict);
3059     dictEmpty(c->db->expires);
3060     addReply(c,shared.ok);
3061 }
3062
3063 static void flushallCommand(redisClient *c) {
3064     server.dirty += emptyDb();
3065     addReply(c,shared.ok);
3066     rdbSave(server.dbfilename);
3067     server.dirty++;
3068 }
3069
3070 redisSortOperation *createSortOperation(int type, robj *pattern) {
3071     redisSortOperation *so = zmalloc(sizeof(*so));
3072     if (!so) oom("createSortOperation");
3073     so->type = type;
3074     so->pattern = pattern;
3075     return so;
3076 }
3077
3078 /* Return the value associated to the key with a name obtained
3079  * substituting the first occurence of '*' in 'pattern' with 'subst' */
3080 robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
3081     char *p;
3082     sds spat, ssub;
3083     robj keyobj;
3084     int prefixlen, sublen, postfixlen;
3085     /* Expoit the internal sds representation to create a sds string allocated on the stack in order to make this function faster */
3086     struct {
3087         long len;
3088         long free;
3089         char buf[REDIS_SORTKEY_MAX+1];
3090     } keyname;
3091
3092     spat = pattern->ptr;
3093     ssub = subst->ptr;
3094     if (sdslen(spat)+sdslen(ssub)-1 > REDIS_SORTKEY_MAX) return NULL;
3095     p = strchr(spat,'*');
3096     if (!p) return NULL;
3097
3098     prefixlen = p-spat;
3099     sublen = sdslen(ssub);
3100     postfixlen = sdslen(spat)-(prefixlen+1);
3101     memcpy(keyname.buf,spat,prefixlen);
3102     memcpy(keyname.buf+prefixlen,ssub,sublen);
3103     memcpy(keyname.buf+prefixlen+sublen,p+1,postfixlen);
3104     keyname.buf[prefixlen+sublen+postfixlen] = '\0';
3105     keyname.len = prefixlen+sublen+postfixlen;
3106
3107     keyobj.refcount = 1;
3108     keyobj.type = REDIS_STRING;
3109     keyobj.ptr = ((char*)&keyname)+(sizeof(long)*2);
3110
3111     /* printf("lookup '%s' => %p\n", keyname.buf,de); */
3112     return lookupKeyRead(db,&keyobj);
3113 }
3114
3115 /* sortCompare() is used by qsort in sortCommand(). Given that qsort_r with
3116  * the additional parameter is not standard but a BSD-specific we have to
3117  * pass sorting parameters via the global 'server' structure */
3118 static int sortCompare(const void *s1, const void *s2) {
3119     const redisSortObject *so1 = s1, *so2 = s2;
3120     int cmp;
3121
3122     if (!server.sort_alpha) {
3123         /* Numeric sorting. Here it's trivial as we precomputed scores */
3124         if (so1->u.score > so2->u.score) {
3125             cmp = 1;
3126         } else if (so1->u.score < so2->u.score) {
3127             cmp = -1;
3128         } else {
3129             cmp = 0;
3130         }
3131     } else {
3132         /* Alphanumeric sorting */
3133         if (server.sort_bypattern) {
3134             if (!so1->u.cmpobj || !so2->u.cmpobj) {
3135                 /* At least one compare object is NULL */
3136                 if (so1->u.cmpobj == so2->u.cmpobj)
3137                     cmp = 0;
3138                 else if (so1->u.cmpobj == NULL)
3139                     cmp = -1;
3140                 else
3141                     cmp = 1;
3142             } else {
3143                 /* We have both the objects, use strcoll */
3144                 cmp = strcoll(so1->u.cmpobj->ptr,so2->u.cmpobj->ptr);
3145             }
3146         } else {
3147             /* Compare elements directly */
3148             cmp = strcoll(so1->obj->ptr,so2->obj->ptr);
3149         }
3150     }
3151     return server.sort_desc ? -cmp : cmp;
3152 }
3153
3154 /* The SORT command is the most complex command in Redis. Warning: this code
3155  * is optimized for speed and a bit less for readability */
3156 static void sortCommand(redisClient *c) {
3157     list *operations;
3158     int outputlen = 0;
3159     int desc = 0, alpha = 0;
3160     int limit_start = 0, limit_count = -1, start, end;
3161     int j, dontsort = 0, vectorlen;
3162     int getop = 0; /* GET operation counter */
3163     robj *sortval, *sortby = NULL;
3164     redisSortObject *vector; /* Resulting vector to sort */
3165
3166     /* Lookup the key to sort. It must be of the right types */
3167     sortval = lookupKeyRead(c->db,c->argv[1]);
3168     if (sortval == NULL) {
3169         addReply(c,shared.nokeyerr);
3170         return;
3171     }
3172     if (sortval->type != REDIS_SET && sortval->type != REDIS_LIST) {
3173         addReply(c,shared.wrongtypeerr);
3174         return;
3175     }
3176
3177     /* Create a list of operations to perform for every sorted element.
3178      * Operations can be GET/DEL/INCR/DECR */
3179     operations = listCreate();
3180     listSetFreeMethod(operations,zfree);
3181     j = 2;
3182
3183     /* Now we need to protect sortval incrementing its count, in the future
3184      * SORT may have options able to overwrite/delete keys during the sorting
3185      * and the sorted key itself may get destroied */
3186     incrRefCount(sortval);
3187
3188     /* The SORT command has an SQL-alike syntax, parse it */
3189     while(j < c->argc) {
3190         int leftargs = c->argc-j-1;
3191         if (!strcasecmp(c->argv[j]->ptr,"asc")) {
3192             desc = 0;
3193         } else if (!strcasecmp(c->argv[j]->ptr,"desc")) {
3194             desc = 1;
3195         } else if (!strcasecmp(c->argv[j]->ptr,"alpha")) {
3196             alpha = 1;
3197         } else if (!strcasecmp(c->argv[j]->ptr,"limit") && leftargs >= 2) {
3198             limit_start = atoi(c->argv[j+1]->ptr);
3199             limit_count = atoi(c->argv[j+2]->ptr);
3200             j+=2;
3201         } else if (!strcasecmp(c->argv[j]->ptr,"by") && leftargs >= 1) {
3202             sortby = c->argv[j+1];
3203             /* If the BY pattern does not contain '*', i.e. it is constant,
3204              * we don't need to sort nor to lookup the weight keys. */
3205             if (strchr(c->argv[j+1]->ptr,'*') == NULL) dontsort = 1;
3206             j++;
3207         } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
3208             listAddNodeTail(operations,createSortOperation(
3209                 REDIS_SORT_GET,c->argv[j+1]));
3210             getop++;
3211             j++;
3212         } else if (!strcasecmp(c->argv[j]->ptr,"del") && leftargs >= 1) {
3213             listAddNodeTail(operations,createSortOperation(
3214                 REDIS_SORT_DEL,c->argv[j+1]));
3215             j++;
3216         } else if (!strcasecmp(c->argv[j]->ptr,"incr") && leftargs >= 1) {
3217             listAddNodeTail(operations,createSortOperation(
3218                 REDIS_SORT_INCR,c->argv[j+1]));
3219             j++;
3220         } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
3221             listAddNodeTail(operations,createSortOperation(
3222                 REDIS_SORT_DECR,c->argv[j+1]));
3223             j++;
3224         } else {
3225             decrRefCount(sortval);
3226             listRelease(operations);
3227             addReply(c,shared.syntaxerr);
3228             return;
3229         }
3230         j++;
3231     }
3232
3233     /* Load the sorting vector with all the objects to sort */
3234     vectorlen = (sortval->type == REDIS_LIST) ?
3235         listLength((list*)sortval->ptr) :
3236         dictSize((dict*)sortval->ptr);
3237     vector = zmalloc(sizeof(redisSortObject)*vectorlen);
3238     if (!vector) oom("allocating objects vector for SORT");
3239     j = 0;
3240     if (sortval->type == REDIS_LIST) {
3241         list *list = sortval->ptr;
3242         listNode *ln;
3243
3244         listRewind(list);
3245         while((ln = listYield(list))) {
3246             robj *ele = ln->value;
3247             vector[j].obj = ele;
3248             vector[j].u.score = 0;
3249             vector[j].u.cmpobj = NULL;
3250             j++;
3251         }
3252     } else {
3253         dict *set = sortval->ptr;
3254         dictIterator *di;
3255         dictEntry *setele;
3256
3257         di = dictGetIterator(set);
3258         if (!di) oom("dictGetIterator");
3259         while((setele = dictNext(di)) != NULL) {
3260             vector[j].obj = dictGetEntryKey(setele);
3261             vector[j].u.score = 0;
3262             vector[j].u.cmpobj = NULL;
3263             j++;
3264         }
3265         dictReleaseIterator(di);
3266     }
3267     assert(j == vectorlen);
3268
3269     /* Now it's time to load the right scores in the sorting vector */
3270     if (dontsort == 0) {
3271         for (j = 0; j < vectorlen; j++) {
3272             if (sortby) {
3273                 robj *byval;
3274
3275                 byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
3276                 if (!byval || byval->type != REDIS_STRING) continue;
3277                 if (alpha) {
3278                     vector[j].u.cmpobj = byval;
3279                     incrRefCount(byval);
3280                 } else {
3281                     vector[j].u.score = strtod(byval->ptr,NULL);
3282                 }
3283             } else {
3284                 if (!alpha) vector[j].u.score = strtod(vector[j].obj->ptr,NULL);
3285             }
3286         }
3287     }
3288
3289     /* We are ready to sort the vector... perform a bit of sanity check
3290      * on the LIMIT option too. We'll use a partial version of quicksort. */
3291     start = (limit_start < 0) ? 0 : limit_start;
3292     end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
3293     if (start >= vectorlen) {
3294         start = vectorlen-1;
3295         end = vectorlen-2;
3296     }
3297     if (end >= vectorlen) end = vectorlen-1;
3298
3299     if (dontsort == 0) {
3300         server.sort_desc = desc;
3301         server.sort_alpha = alpha;
3302         server.sort_bypattern = sortby ? 1 : 0;
3303         qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare);
3304     }
3305
3306     /* Send command output to the output buffer, performing the specified
3307      * GET/DEL/INCR/DECR operations if any. */
3308     outputlen = getop ? getop*(end-start+1) : end-start+1;
3309     addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",outputlen));
3310     for (j = start; j <= end; j++) {
3311         listNode *ln;
3312         if (!getop) {
3313             addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
3314                 sdslen(vector[j].obj->ptr)));
3315             addReply(c,vector[j].obj);
3316             addReply(c,shared.crlf);
3317         }
3318         listRewind(operations);
3319         while((ln = listYield(operations))) {
3320             redisSortOperation *sop = ln->value;
3321             robj *val = lookupKeyByPattern(c->db,sop->pattern,
3322                 vector[j].obj);
3323
3324             if (sop->type == REDIS_SORT_GET) {
3325                 if (!val || val->type != REDIS_STRING) {
3326                     addReply(c,shared.nullbulk);
3327                 } else {
3328                     addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
3329                         sdslen(val->ptr)));
3330                     addReply(c,val);
3331                     addReply(c,shared.crlf);
3332                 }
3333             } else if (sop->type == REDIS_SORT_DEL) {
3334                 /* TODO */
3335             }
3336         }
3337     }
3338
3339     /* Cleanup */
3340     decrRefCount(sortval);
3341     listRelease(operations);
3342     for (j = 0; j < vectorlen; j++) {
3343         if (sortby && alpha && vector[j].u.cmpobj)
3344             decrRefCount(vector[j].u.cmpobj);
3345     }
3346     zfree(vector);
3347 }
3348
3349 static void infoCommand(redisClient *c) {
3350     sds info;
3351     time_t uptime = time(NULL)-server.stat_starttime;
3352
3353     info = sdscatprintf(sdsempty(),
3354         "redis_version:%s\r\n"
3355         "connected_clients:%d\r\n"
3356         "connected_slaves:%d\r\n"
3357         "used_memory:%zu\r\n"
3358         "changes_since_last_save:%lld\r\n"
3359         "bgsave_in_progress:%d\r\n"
3360         "last_save_time:%d\r\n"
3361         "total_connections_received:%lld\r\n"
3362         "total_commands_processed:%lld\r\n"
3363         "uptime_in_seconds:%d\r\n"
3364         "uptime_in_days:%d\r\n"
3365         ,REDIS_VERSION,
3366         listLength(server.clients)-listLength(server.slaves),
3367         listLength(server.slaves),
3368         server.usedmemory,
3369         server.dirty,
3370         server.bgsaveinprogress,
3371         server.lastsave,
3372         server.stat_numconnections,
3373         server.stat_numcommands,
3374         uptime,
3375         uptime/(3600*24)
3376     );
3377     addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(info)));
3378     addReplySds(c,info);
3379     addReply(c,shared.crlf);
3380 }
3381
3382 static void monitorCommand(redisClient *c) {
3383     /* ignore MONITOR if aleady slave or in monitor mode */
3384     if (c->flags & REDIS_SLAVE) return;
3385
3386     c->flags |= (REDIS_SLAVE|REDIS_MONITOR);
3387     c->slaveseldb = 0;
3388     if (!listAddNodeTail(server.monitors,c)) oom("listAddNodeTail");
3389     addReply(c,shared.ok);
3390 }
3391
3392 /* ================================= Expire ================================= */
3393 static int removeExpire(redisDb *db, robj *key) {
3394     if (dictDelete(db->expires,key) == DICT_OK) {
3395         return 1;
3396     } else {
3397         return 0;
3398     }
3399 }
3400
3401 static int setExpire(redisDb *db, robj *key, time_t when) {
3402     if (dictAdd(db->expires,key,(void*)when) == DICT_ERR) {
3403         return 0;
3404     } else {
3405         incrRefCount(key);
3406         return 1;
3407     }
3408 }
3409
3410 /* Return the expire time of the specified key, or -1 if no expire
3411  * is associated with this key (i.e. the key is non volatile) */
3412 static time_t getExpire(redisDb *db, robj *key) {
3413     dictEntry *de;
3414
3415     /* No expire? return ASAP */
3416     if (dictSize(db->expires) == 0 ||
3417        (de = dictFind(db->expires,key)) == NULL) return -1;
3418
3419     return (time_t) dictGetEntryVal(de);
3420 }
3421
3422 static int expireIfNeeded(redisDb *db, robj *key) {
3423     time_t when;
3424     dictEntry *de;
3425
3426     /* No expire? return ASAP */
3427     if (dictSize(db->expires) == 0 ||
3428        (de = dictFind(db->expires,key)) == NULL) return 0;
3429
3430     /* Lookup the expire */
3431     when = (time_t) dictGetEntryVal(de);
3432     if (time(NULL) <= when) return 0;
3433
3434     /* Delete the key */
3435     dictDelete(db->expires,key);
3436     return dictDelete(db->dict,key) == DICT_OK;
3437 }
3438
3439 static int deleteIfVolatile(redisDb *db, robj *key) {
3440     dictEntry *de;
3441
3442     /* No expire? return ASAP */
3443     if (dictSize(db->expires) == 0 ||
3444        (de = dictFind(db->expires,key)) == NULL) return 0;
3445
3446     /* Delete the key */
3447     server.dirty++;
3448     dictDelete(db->expires,key);
3449     return dictDelete(db->dict,key) == DICT_OK;
3450 }
3451
3452 static void expireCommand(redisClient *c) {
3453     dictEntry *de;
3454     int seconds = atoi(c->argv[2]->ptr);
3455
3456     de = dictFind(c->db->dict,c->argv[1]);
3457     if (de == NULL) {
3458         addReply(c,shared.czero);
3459         return;
3460     }
3461     if (seconds <= 0) {
3462         addReply(c, shared.czero);
3463         return;
3464     } else {
3465         time_t when = time(NULL)+seconds;
3466         if (setExpire(c->db,c->argv[1],when))
3467             addReply(c,shared.cone);
3468         else
3469             addReply(c,shared.czero);
3470         return;
3471     }
3472 }
3473
3474 /* =============================== Replication  ============================= */
3475
3476 static int syncWrite(int fd, char *ptr, ssize_t size, int timeout) {
3477     ssize_t nwritten, ret = size;
3478     time_t start = time(NULL);
3479
3480     timeout++;
3481     while(size) {
3482         if (aeWait(fd,AE_WRITABLE,1000) & AE_WRITABLE) {
3483             nwritten = write(fd,ptr,size);
3484             if (nwritten == -1) return -1;
3485             ptr += nwritten;
3486             size -= nwritten;
3487         }
3488         if ((time(NULL)-start) > timeout) {
3489             errno = ETIMEDOUT;
3490             return -1;
3491         }
3492     }
3493     return ret;
3494 }
3495
3496 static int syncRead(int fd, char *ptr, ssize_t size, int timeout) {
3497     ssize_t nread, totread = 0;
3498     time_t start = time(NULL);
3499
3500     timeout++;
3501     while(size) {
3502         if (aeWait(fd,AE_READABLE,1000) & AE_READABLE) {
3503             nread = read(fd,ptr,size);
3504             if (nread == -1) return -1;
3505             ptr += nread;
3506             size -= nread;
3507             totread += nread;
3508         }
3509         if ((time(NULL)-start) > timeout) {
3510             errno = ETIMEDOUT;
3511             return -1;
3512         }
3513     }
3514     return totread;
3515 }
3516
3517 static int syncReadLine(int fd, char *ptr, ssize_t size, int timeout) {
3518     ssize_t nread = 0;
3519
3520     size--;
3521     while(size) {
3522         char c;
3523
3524         if (syncRead(fd,&c,1,timeout) == -1) return -1;
3525         if (c == '\n') {
3526             *ptr = '\0';
3527             if (nread && *(ptr-1) == '\r') *(ptr-1) = '\0';
3528             return nread;
3529         } else {
3530             *ptr++ = c;
3531             *ptr = '\0';
3532             nread++;
3533         }
3534     }
3535     return nread;
3536 }
3537
3538 static void syncCommand(redisClient *c) {
3539     /* ignore SYNC if aleady slave or in monitor mode */
3540     if (c->flags & REDIS_SLAVE) return;
3541
3542     /* SYNC can't be issued when the server has pending data to send to
3543      * the client about already issued commands. We need a fresh reply
3544      * buffer registering the differences between the BGSAVE and the current
3545      * dataset, so that we can copy to other slaves if needed. */
3546     if (listLength(c->reply) != 0) {
3547         addReplySds(c,sdsnew("-ERR SYNC is invalid with pending input\r\n"));
3548         return;
3549     }
3550
3551     redisLog(REDIS_NOTICE,"Slave ask for synchronization");
3552     /* Here we need to check if there is a background saving operation
3553      * in progress, or if it is required to start one */
3554     if (server.bgsaveinprogress) {
3555         /* Ok a background save is in progress. Let's check if it is a good
3556          * one for replication, i.e. if there is another slave that is
3557          * registering differences since the server forked to save */
3558         redisClient *slave;
3559         listNode *ln;
3560
3561         listRewind(server.slaves);
3562         while((ln = listYield(server.slaves))) {
3563             slave = ln->value;
3564             if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
3565         }
3566         if (ln) {
3567             /* Perfect, the server is already registering differences for
3568              * another slave. Set the right state, and copy the buffer. */
3569             listRelease(c->reply);
3570             c->reply = listDup(slave->reply);
3571             if (!c->reply) oom("listDup copying slave reply list");
3572             c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
3573             redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC");
3574         } else {
3575             /* No way, we need to wait for the next BGSAVE in order to
3576              * register differences */
3577             c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
3578             redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
3579         }
3580     } else {
3581         /* Ok we don't have a BGSAVE in progress, let's start one */
3582         redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC");
3583         if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
3584             redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE");
3585             addReplySds(c,sdsnew("-ERR Unalbe to perform background save\r\n"));
3586             return;
3587         }
3588         c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
3589     }
3590     c->repldbfd = -1;
3591     c->flags |= REDIS_SLAVE;
3592     c->slaveseldb = 0;
3593     if (!listAddNodeTail(server.slaves,c)) oom("listAddNodeTail");
3594     return;
3595 }
3596
3597 static void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
3598     redisClient *slave = privdata;
3599     REDIS_NOTUSED(el);
3600     REDIS_NOTUSED(mask);
3601     char buf[REDIS_IOBUF_LEN];
3602     ssize_t nwritten, buflen;
3603
3604     if (slave->repldboff == 0) {
3605         /* Write the bulk write count before to transfer the DB. In theory here
3606          * we don't know how much room there is in the output buffer of the
3607          * socket, but in pratice SO_SNDLOWAT (the minimum count for output
3608          * operations) will never be smaller than the few bytes we need. */
3609         sds bulkcount;
3610
3611         bulkcount = sdscatprintf(sdsempty(),"$%lld\r\n",(unsigned long long)
3612             slave->repldbsize);
3613         if (write(fd,bulkcount,sdslen(bulkcount)) != (signed)sdslen(bulkcount))
3614         {
3615             sdsfree(bulkcount);
3616             freeClient(slave);
3617             return;
3618         }
3619         sdsfree(bulkcount);
3620     }
3621     lseek(slave->repldbfd,slave->repldboff,SEEK_SET);
3622     buflen = read(slave->repldbfd,buf,REDIS_IOBUF_LEN);
3623     if (buflen <= 0) {
3624         redisLog(REDIS_WARNING,"Read error sending DB to slave: %s",
3625             (buflen == 0) ? "premature EOF" : strerror(errno));
3626         freeClient(slave);
3627         return;
3628     }
3629     if ((nwritten = write(fd,buf,buflen)) == -1) {
3630         redisLog(REDIS_DEBUG,"Write error sending DB to slave: %s",
3631             strerror(errno));
3632         freeClient(slave);
3633         return;
3634     }
3635     slave->repldboff += nwritten;
3636     if (slave->repldboff == slave->repldbsize) {
3637         close(slave->repldbfd);
3638         slave->repldbfd = -1;
3639         aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
3640         slave->replstate = REDIS_REPL_ONLINE;
3641         if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE,
3642             sendReplyToClient, slave, NULL) == AE_ERR) {
3643             freeClient(slave);
3644             return;
3645         }
3646         addReplySds(slave,sdsempty());
3647         redisLog(REDIS_NOTICE,"Synchronization with slave succeeded");
3648     }
3649 }
3650
3651 static void updateSalvesWaitingBgsave(int bgsaveerr) {
3652     listNode *ln;
3653     int startbgsave = 0;
3654
3655     listRewind(server.slaves);
3656     while((ln = listYield(server.slaves))) {
3657         redisClient *slave = ln->value;
3658
3659         if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
3660             startbgsave = 1;
3661             slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
3662         } else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
3663             struct stat buf;
3664
3665             if (bgsaveerr != REDIS_OK) {
3666                 freeClient(slave);
3667                 redisLog(REDIS_WARNING,"SYNC failed. BGSAVE child returned an error");
3668                 continue;
3669             }
3670             if ((slave->repldbfd = open(server.dbfilename,O_RDONLY)) == -1 ||
3671                 fstat(slave->repldbfd,&buf) == -1) {
3672                 freeClient(slave);
3673                 redisLog(REDIS_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno));
3674                 continue;
3675             }
3676             slave->repldboff = 0;
3677             slave->repldbsize = buf.st_size;
3678             slave->replstate = REDIS_REPL_SEND_BULK;
3679             aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
3680             if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave, NULL) == AE_ERR) {
3681                 freeClient(slave);
3682                 continue;
3683             }
3684         }
3685     }
3686     if (startbgsave) {
3687         if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
3688             listRewind(server.slaves);
3689             redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
3690             while((ln = listYield(server.slaves))) {
3691                 redisClient *slave = ln->value;
3692
3693                 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
3694                     freeClient(slave);
3695             }
3696         }
3697     }
3698 }
3699
3700 static int syncWithMaster(void) {
3701     char buf[1024], tmpfile[256];
3702     int dumpsize;
3703     int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);
3704     int dfd;
3705
3706     if (fd == -1) {
3707         redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
3708             strerror(errno));
3709         return REDIS_ERR;
3710     }
3711     /* Issue the SYNC command */
3712     if (syncWrite(fd,"SYNC \r\n",7,5) == -1) {
3713         close(fd);
3714         redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
3715             strerror(errno));
3716         return REDIS_ERR;
3717     }
3718     /* Read the bulk write count */
3719     if (syncReadLine(fd,buf,1024,5) == -1) {
3720         close(fd);
3721         redisLog(REDIS_WARNING,"I/O error reading bulk count from MASTER: %s",
3722             strerror(errno));
3723         return REDIS_ERR;
3724     }
3725     dumpsize = atoi(buf+1);
3726     redisLog(REDIS_NOTICE,"Receiving %d bytes data dump from MASTER",dumpsize);
3727     /* Read the bulk write data on a temp file */
3728     snprintf(tmpfile,256,"temp-%d.%ld.rdb",(int)time(NULL),(long int)random());
3729     dfd = open(tmpfile,O_CREAT|O_WRONLY,0644);
3730     if (dfd == -1) {
3731         close(fd);
3732         redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
3733         return REDIS_ERR;
3734     }
3735     while(dumpsize) {
3736         int nread, nwritten;
3737
3738         nread = read(fd,buf,(dumpsize < 1024)?dumpsize:1024);
3739         if (nread == -1) {
3740             redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
3741                 strerror(errno));
3742             close(fd);
3743             close(dfd);
3744             return REDIS_ERR;
3745         }
3746         nwritten = write(dfd,buf,nread);
3747         if (nwritten == -1) {
3748             redisLog(REDIS_WARNING,"Write error writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno));
3749             close(fd);
3750             close(dfd);
3751             return REDIS_ERR;
3752         }
3753         dumpsize -= nread;
3754     }
3755     close(dfd);
3756     if (rename(tmpfile,server.dbfilename) == -1) {
3757         redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
3758         unlink(tmpfile);
3759         close(fd);
3760         return REDIS_ERR;
3761     }
3762     emptyDb();
3763     if (rdbLoad(server.dbfilename) != REDIS_OK) {
3764         redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
3765         close(fd);
3766         return REDIS_ERR;
3767     }
3768     server.master = createClient(fd);
3769     server.master->flags |= REDIS_MASTER;
3770     server.replstate = REDIS_REPL_CONNECTED;
3771     return REDIS_OK;
3772 }
3773
3774 /* =================================== Main! ================================ */
3775
3776 static void daemonize(void) {
3777     int fd;
3778     FILE *fp;
3779
3780     if (fork() != 0) exit(0); /* parent exits */
3781     setsid(); /* create a new session */
3782
3783     /* Every output goes to /dev/null. If Redis is daemonized but
3784      * the 'logfile' is set to 'stdout' in the configuration file
3785      * it will not log at all. */
3786     if ((fd = open("/dev/null", O_RDWR, 0)) != -1) {
3787         dup2(fd, STDIN_FILENO);
3788         dup2(fd, STDOUT_FILENO);
3789         dup2(fd, STDERR_FILENO);
3790         if (fd > STDERR_FILENO) close(fd);
3791     }
3792     /* Try to write the pid file */
3793     fp = fopen(server.pidfile,"w");
3794     if (fp) {
3795         fprintf(fp,"%d\n",getpid());
3796         fclose(fp);
3797     }
3798 }
3799
3800 int main(int argc, char **argv) {
3801     initServerConfig();
3802     if (argc == 2) {
3803         ResetServerSaveParams();
3804         loadServerConfig(argv[1]);
3805     } else if (argc > 2) {
3806         fprintf(stderr,"Usage: ./redis-server [/path/to/redis.conf]\n");
3807         exit(1);
3808     }
3809     initServer();
3810     if (server.daemonize) daemonize();
3811     redisLog(REDIS_NOTICE,"Server started, Redis version " REDIS_VERSION);
3812     if (rdbLoad(server.dbfilename) == REDIS_OK)
3813         redisLog(REDIS_NOTICE,"DB loaded from disk");
3814     if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
3815         acceptHandler, NULL, NULL) == AE_ERR) oom("creating file event");
3816     redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
3817     aeMain(server.el);
3818     aeDeleteEventLoop(server.el);
3819     return 0;
3820 }