redis.c

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