redis.c

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