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