redis.c

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