]> git.saurik.com Git - redis.git/blob - redis.c
CPP client added thanks to Brian Hammond
[redis.git] / redis.c
1 /*
2 * Copyright (c) 2006-2009, Salvatore Sanfilippo <antirez at gmail dot com>
3 * All rights reserved.
4 *
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions are met:
7 *
8 * * Redistributions of source code must retain the above copyright notice,
9 * this list of conditions and the following disclaimer.
10 * * Redistributions in binary form must reproduce the above copyright
11 * notice, this list of conditions and the following disclaimer in the
12 * documentation and/or other materials provided with the distribution.
13 * * Neither the name of Redis nor the names of its contributors may be used
14 * to endorse or promote products derived from this software without
15 * specific prior written permission.
16 *
17 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
18 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
19 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
20 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
21 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
22 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
23 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
24 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
25 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
26 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27 * POSSIBILITY OF SUCH DAMAGE.
28 */
29
30 #define REDIS_VERSION "0.100"
31
32 #include "fmacros.h"
33
34 #include <stdio.h>
35 #include <stdlib.h>
36 #include <string.h>
37 #include <time.h>
38 #include <unistd.h>
39 #include <signal.h>
40 #include <sys/wait.h>
41 #include <errno.h>
42 #include <assert.h>
43 #include <ctype.h>
44 #include <stdarg.h>
45 #include <inttypes.h>
46 #include <arpa/inet.h>
47 #include <sys/stat.h>
48 #include <fcntl.h>
49 #include <sys/time.h>
50 #include <sys/resource.h>
51 #include <limits.h>
52
53 #include "ae.h" /* Event driven programming library */
54 #include "sds.h" /* Dynamic safe strings */
55 #include "anet.h" /* Networking the easy way */
56 #include "dict.h" /* Hash tables */
57 #include "adlist.h" /* Linked lists */
58 #include "zmalloc.h" /* total memory usage aware version of malloc/free */
59 #include "lzf.h"
60
61 /* Error codes */
62 #define REDIS_OK 0
63 #define REDIS_ERR -1
64
65 /* Static server configuration */
66 #define REDIS_SERVERPORT 6379 /* TCP port */
67 #define REDIS_MAXIDLETIME (60*5) /* default client timeout */
68 #define REDIS_IOBUF_LEN 1024
69 #define REDIS_LOADBUF_LEN 1024
70 #define REDIS_STATIC_ARGS 4
71 #define REDIS_DEFAULT_DBNUM 16
72 #define REDIS_CONFIGLINE_MAX 1024
73 #define REDIS_OBJFREELIST_MAX 1000000 /* Max number of objects to cache */
74 #define REDIS_MAX_SYNC_TIME 60 /* Slave can't take more to sync */
75 #define REDIS_EXPIRELOOKUPS_PER_CRON 100 /* try to expire 100 keys/second */
76
77 /* Hash table parameters */
78 #define REDIS_HT_MINFILL 10 /* Minimal hash table fill 10% */
79 #define REDIS_HT_MINSLOTS 16384 /* Never resize the HT under this */
80
81 /* Command flags */
82 #define REDIS_CMD_BULK 1
83 #define REDIS_CMD_INLINE 2
84
85 /* Object types */
86 #define REDIS_STRING 0
87 #define REDIS_LIST 1
88 #define REDIS_SET 2
89 #define REDIS_HASH 3
90
91 /* Object types only used for dumping to disk */
92 #define REDIS_EXPIRETIME 253
93 #define REDIS_SELECTDB 254
94 #define REDIS_EOF 255
95
96 /* Defines related to the dump file format. To store 32 bits lengths for short
97 * keys requires a lot of space, so we check the most significant 2 bits of
98 * the first byte to interpreter the length:
99 *
100 * 00|000000 => if the two MSB are 00 the len is the 6 bits of this byte
101 * 01|000000 00000000 => 01, the len is 14 byes, 6 bits + 8 bits of next byte
102 * 10|000000 [32 bit integer] => if it's 01, a full 32 bit len will follow
103 * 11|000000 this means: specially encoded object will follow. The six bits
104 * number specify the kind of object that follows.
105 * See the REDIS_RDB_ENC_* defines.
106 *
107 * Lenghts up to 63 are stored using a single byte, most DB keys, and may
108 * values, will fit inside. */
109 #define REDIS_RDB_6BITLEN 0
110 #define REDIS_RDB_14BITLEN 1
111 #define REDIS_RDB_32BITLEN 2
112 #define REDIS_RDB_ENCVAL 3
113 #define REDIS_RDB_LENERR UINT_MAX
114
115 /* When a length of a string object stored on disk has the first two bits
116 * set, the remaining two bits specify a special encoding for the object
117 * accordingly to the following defines: */
118 #define REDIS_RDB_ENC_INT8 0 /* 8 bit signed integer */
119 #define REDIS_RDB_ENC_INT16 1 /* 16 bit signed integer */
120 #define REDIS_RDB_ENC_INT32 2 /* 32 bit signed integer */
121 #define REDIS_RDB_ENC_LZF 3 /* string compressed with FASTLZ */
122
123 /* Client flags */
124 #define REDIS_CLOSE 1 /* This client connection should be closed ASAP */
125 #define REDIS_SLAVE 2 /* This client is a slave server */
126 #define REDIS_MASTER 4 /* This client is a master server */
127 #define REDIS_MONITOR 8 /* This client is a slave monitor, see MONITOR */
128
129 /* Slave replication state - slave side */
130 #define REDIS_REPL_NONE 0 /* No active replication */
131 #define REDIS_REPL_CONNECT 1 /* Must connect to master */
132 #define REDIS_REPL_CONNECTED 2 /* Connected to master */
133
134 /* Slave replication state - from the point of view of master
135 * Note that in SEND_BULK and ONLINE state the slave receives new updates
136 * in its output queue. In the WAIT_BGSAVE state instead the server is waiting
137 * to start the next background saving in order to send updates to it. */
138 #define REDIS_REPL_WAIT_BGSAVE_START 3 /* master waits bgsave to start feeding it */
139 #define REDIS_REPL_WAIT_BGSAVE_END 4 /* master waits bgsave to start bulk DB transmission */
140 #define REDIS_REPL_SEND_BULK 5 /* master is sending the bulk DB */
141 #define REDIS_REPL_ONLINE 6 /* bulk DB already transmitted, receive updates */
142
143 /* List related stuff */
144 #define REDIS_HEAD 0
145 #define REDIS_TAIL 1
146
147 /* Sort operations */
148 #define REDIS_SORT_GET 0
149 #define REDIS_SORT_DEL 1
150 #define REDIS_SORT_INCR 2
151 #define REDIS_SORT_DECR 3
152 #define REDIS_SORT_ASC 4
153 #define REDIS_SORT_DESC 5
154 #define REDIS_SORTKEY_MAX 1024
155
156 /* Log levels */
157 #define REDIS_DEBUG 0
158 #define REDIS_NOTICE 1
159 #define REDIS_WARNING 2
160
161 /* Anti-warning macro... */
162 #define REDIS_NOTUSED(V) ((void) V)
163
164 /*================================= Data types ============================== */
165
166 /* A redis object, that is a type able to hold a string / list / set */
167 typedef struct redisObject {
168 void *ptr;
169 int type;
170 int refcount;
171 } robj;
172
173 typedef struct redisDb {
174 dict *dict;
175 dict *expires;
176 int id;
177 } redisDb;
178
179 /* With multiplexing we need to take per-clinet state.
180 * Clients are taken in a liked list. */
181 typedef struct redisClient {
182 int fd;
183 redisDb *db;
184 int dictid;
185 sds querybuf;
186 robj **argv;
187 int argc;
188 int bulklen; /* bulk read len. -1 if not in bulk read mode */
189 list *reply;
190 int sentlen;
191 time_t lastinteraction; /* time of the last interaction, used for timeout */
192 int flags; /* REDIS_CLOSE | REDIS_SLAVE | REDIS_MONITOR */
193 int slaveseldb; /* slave selected db, if this client is a slave */
194 int authenticated; /* when requirepass is non-NULL */
195 int replstate; /* replication state if this is a slave */
196 int repldbfd; /* replication DB file descriptor */
197 long repldboff; /* replication DB file offset */
198 off_t repldbsize; /* replication DB file size */
199 } redisClient;
200
201 struct saveparam {
202 time_t seconds;
203 int changes;
204 };
205
206 /* Global server state structure */
207 struct redisServer {
208 int port;
209 int fd;
210 redisDb *db;
211 dict *sharingpool;
212 unsigned int sharingpoolsize;
213 long long dirty; /* changes to DB from the last save */
214 list *clients;
215 list *slaves, *monitors;
216 char neterr[ANET_ERR_LEN];
217 aeEventLoop *el;
218 int cronloops; /* number of times the cron function run */
219 list *objfreelist; /* A list of freed objects to avoid malloc() */
220 time_t lastsave; /* Unix time of last save succeeede */
221 size_t usedmemory; /* Used memory in megabytes */
222 /* Fields used only for stats */
223 time_t stat_starttime; /* server start time */
224 long long stat_numcommands; /* number of processed commands */
225 long long stat_numconnections; /* number of connections received */
226 /* Configuration */
227 int verbosity;
228 int glueoutputbuf;
229 int maxidletime;
230 int dbnum;
231 int daemonize;
232 char *pidfile;
233 int bgsaveinprogress;
234 struct saveparam *saveparams;
235 int saveparamslen;
236 char *logfile;
237 char *bindaddr;
238 char *dbfilename;
239 char *requirepass;
240 int shareobjects;
241 /* Replication related */
242 int isslave;
243 char *masterhost;
244 int masterport;
245 redisClient *master; /* client that is master for this slave */
246 int replstate;
247 /* Sort parameters - qsort_r() is only available under BSD so we
248 * have to take this state global, in order to pass it to sortCompare() */
249 int sort_desc;
250 int sort_alpha;
251 int sort_bypattern;
252 };
253
254 typedef void redisCommandProc(redisClient *c);
255 struct redisCommand {
256 char *name;
257 redisCommandProc *proc;
258 int arity;
259 int flags;
260 };
261
262 typedef struct _redisSortObject {
263 robj *obj;
264 union {
265 double score;
266 robj *cmpobj;
267 } u;
268 } redisSortObject;
269
270 typedef struct _redisSortOperation {
271 int type;
272 robj *pattern;
273 } redisSortOperation;
274
275 struct sharedObjectsStruct {
276 robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *pong, *space,
277 *colon, *nullbulk, *nullmultibulk,
278 *emptymultibulk, *wrongtypeerr, *nokeyerr, *syntaxerr, *sameobjecterr,
279 *outofrangeerr, *plus,
280 *select0, *select1, *select2, *select3, *select4,
281 *select5, *select6, *select7, *select8, *select9;
282 } shared;
283
284 /*================================ Prototypes =============================== */
285
286 static void freeStringObject(robj *o);
287 static void freeListObject(robj *o);
288 static void freeSetObject(robj *o);
289 static void decrRefCount(void *o);
290 static robj *createObject(int type, void *ptr);
291 static void freeClient(redisClient *c);
292 static int rdbLoad(char *filename);
293 static void addReply(redisClient *c, robj *obj);
294 static void addReplySds(redisClient *c, sds s);
295 static void incrRefCount(robj *o);
296 static int rdbSaveBackground(char *filename);
297 static robj *createStringObject(char *ptr, size_t len);
298 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc);
299 static int syncWithMaster(void);
300 static robj *tryObjectSharing(robj *o);
301 static int removeExpire(redisDb *db, robj *key);
302 static int expireIfNeeded(redisDb *db, robj *key);
303 static int deleteIfVolatile(redisDb *db, robj *key);
304 static int deleteKey(redisDb *db, robj *key);
305 static time_t getExpire(redisDb *db, robj *key);
306 static int setExpire(redisDb *db, robj *key, time_t when);
307 static void updateSalvesWaitingBgsave(int bgsaveerr);
308
309 static void authCommand(redisClient *c);
310 static void pingCommand(redisClient *c);
311 static void echoCommand(redisClient *c);
312 static void setCommand(redisClient *c);
313 static void setnxCommand(redisClient *c);
314 static void getCommand(redisClient *c);
315 static void delCommand(redisClient *c);
316 static void existsCommand(redisClient *c);
317 static void incrCommand(redisClient *c);
318 static void decrCommand(redisClient *c);
319 static void incrbyCommand(redisClient *c);
320 static void decrbyCommand(redisClient *c);
321 static void selectCommand(redisClient *c);
322 static void randomkeyCommand(redisClient *c);
323 static void keysCommand(redisClient *c);
324 static void dbsizeCommand(redisClient *c);
325 static void lastsaveCommand(redisClient *c);
326 static void saveCommand(redisClient *c);
327 static void bgsaveCommand(redisClient *c);
328 static void shutdownCommand(redisClient *c);
329 static void moveCommand(redisClient *c);
330 static void renameCommand(redisClient *c);
331 static void renamenxCommand(redisClient *c);
332 static void lpushCommand(redisClient *c);
333 static void rpushCommand(redisClient *c);
334 static void lpopCommand(redisClient *c);
335 static void rpopCommand(redisClient *c);
336 static void llenCommand(redisClient *c);
337 static void lindexCommand(redisClient *c);
338 static void lrangeCommand(redisClient *c);
339 static void ltrimCommand(redisClient *c);
340 static void typeCommand(redisClient *c);
341 static void lsetCommand(redisClient *c);
342 static void saddCommand(redisClient *c);
343 static void sremCommand(redisClient *c);
344 static void smoveCommand(redisClient *c);
345 static void sismemberCommand(redisClient *c);
346 static void scardCommand(redisClient *c);
347 static void sinterCommand(redisClient *c);
348 static void sinterstoreCommand(redisClient *c);
349 static void sunionCommand(redisClient *c);
350 static void sunionstoreCommand(redisClient *c);
351 static void syncCommand(redisClient *c);
352 static void flushdbCommand(redisClient *c);
353 static void flushallCommand(redisClient *c);
354 static void sortCommand(redisClient *c);
355 static void lremCommand(redisClient *c);
356 static void infoCommand(redisClient *c);
357 static void mgetCommand(redisClient *c);
358 static void monitorCommand(redisClient *c);
359 static void expireCommand(redisClient *c);
360 static void getSetCommand(redisClient *c);
361
362 /*================================= Globals ================================= */
363
364 /* Global vars */
365 static struct redisServer server; /* server global state */
366 static struct redisCommand cmdTable[] = {
367 {"get",getCommand,2,REDIS_CMD_INLINE},
368 {"set",setCommand,3,REDIS_CMD_BULK},
369 {"setnx",setnxCommand,3,REDIS_CMD_BULK},
370 {"del",delCommand,2,REDIS_CMD_INLINE},
371 {"exists",existsCommand,2,REDIS_CMD_INLINE},
372 {"incr",incrCommand,2,REDIS_CMD_INLINE},
373 {"decr",decrCommand,2,REDIS_CMD_INLINE},
374 {"mget",mgetCommand,-2,REDIS_CMD_INLINE},
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},
387 {"smove",smoveCommand,4,REDIS_CMD_BULK},
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},
392 {"sunion",sunionCommand,-2,REDIS_CMD_INLINE},
393 {"sunionstore",sunionstoreCommand,-3,REDIS_CMD_INLINE},
394 {"smembers",sinterCommand,2,REDIS_CMD_INLINE},
395 {"incrby",incrbyCommand,3,REDIS_CMD_INLINE},
396 {"decrby",decrbyCommand,3,REDIS_CMD_INLINE},
397 {"getset",getSetCommand,3,REDIS_CMD_BULK},
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},
405 {"auth",authCommand,2,REDIS_CMD_INLINE},
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},
418 {"monitor",monitorCommand,1,REDIS_CMD_INLINE},
419 {"expire",expireCommand,3,REDIS_CMD_INLINE},
420 {NULL,NULL,0,0}
421 };
422
423 /*============================ Utility functions ============================ */
424
425 /* Glob-style pattern matching. */
426 int 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
548 void 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
575 static 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
587 static void dictRedisObjectDestructor(void *privdata, void *val)
588 {
589 DICT_NOTUSED(privdata);
590
591 decrRefCount(val);
592 }
593
594 static 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
601 static unsigned int dictSdsHash(const void *key) {
602 const robj *o = key;
603 return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
604 }
605
606 static 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
615 static 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... */
631 static 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 ===================== */
639 void closeTimedoutClients(void) {
640 redisClient *c;
641 listNode *ln;
642 time_t now = time(NULL);
643
644 listRewind(server.clients);
645 while ((ln = listYield(server.clients)) != NULL) {
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 }
653 }
654
655 /* If the percentage of used slots in the HT reaches REDIS_HT_MINFILL
656 * we resize the hash table to save memory */
657 void 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
674 int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
675 int j, loops = server.cronloops++;
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
683 /* Show some info about non-empty databases */
684 for (j = 0; j < server.dbnum; j++) {
685 long long size, used, vkeys;
686
687 size = dictSlots(server.db[j].dict);
688 used = dictSize(server.db[j].dict);
689 vkeys = dictSize(server.db[j].expires);
690 if (!(loops % 5) && used > 0) {
691 redisLog(REDIS_DEBUG,"DB %d: %d keys (%d volatile) in %d slots HT.",j,used,vkeys,size);
692 /* dictPrintStats(server.dict); */
693 }
694 }
695
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
704 /* Show information about connected clients */
705 if (!(loops % 5)) {
706 redisLog(REDIS_DEBUG,"%d clients connected (%d slaves), %zu bytes in use",
707 listLength(server.clients)-listLength(server.slaves),
708 listLength(server.slaves),
709 server.usedmemory,
710 dictSize(server.sharingpool));
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;
720 /* XXX: TODO handle the case of the saving child killed */
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;
733 updateSalvesWaitingBgsave(exitcode == 0 ? REDIS_OK : REDIS_ERR);
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);
746 rdbSaveBackground(server.dbfilename);
747 break;
748 }
749 }
750 }
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
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
785 static 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"));
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"));
795 /* no such key */
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"));
799 shared.nokeyerr = createObject(REDIS_STRING,sdsnew(
800 "-ERR no such key\r\n"));
801 shared.syntaxerr = createObject(REDIS_STRING,sdsnew(
802 "-ERR syntax error\r\n"));
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"));
807 shared.space = createObject(REDIS_STRING,sdsnew(" "));
808 shared.colon = createObject(REDIS_STRING,sdsnew(":"));
809 shared.plus = createObject(REDIS_STRING,sdsnew("+"));
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
822 static 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
830 static void ResetServerSaveParams() {
831 zfree(server.saveparams);
832 server.saveparams = NULL;
833 server.saveparamslen = 0;
834 }
835
836 static 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;
846 server.pidfile = "/var/run/redis.pid";
847 server.dbfilename = "dump.rdb";
848 server.requirepass = NULL;
849 server.shareobjects = 0;
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
863 static 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();
871 server.monitors = listCreate();
872 server.objfreelist = listCreate();
873 createSharedObjects();
874 server.el = aeCreateEventLoop();
875 server.db = zmalloc(sizeof(redisDb)*server.dbnum);
876 server.sharingpool = dictCreate(&setDictType,NULL);
877 server.sharingpoolsize = 1024;
878 if (!server.db || !server.clients || !server.slaves || !server.monitors || !server.el || !server.objfreelist)
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 }
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 }
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 */
902 static long long emptyDb() {
903 int j;
904 long long removed = 0;
905
906 for (j = 0; j < server.dbnum; j++) {
907 removed += dictSize(server.db[j].dict);
908 dictEmpty(server.db[j].dict);
909 dictEmpty(server.db[j].expires);
910 }
911 return removed;
912 }
913
914 static 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
920 /* I agree, this is a very rudimental way to load a configuration...
921 will improve later if the config gets more complex */
922 static 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 */
951 if (!strcasecmp(argv[0],"timeout") && argc == 2) {
952 server.maxidletime = atoi(argv[1]);
953 if (server.maxidletime < 1) {
954 err = "Invalid timeout value"; goto loaderr;
955 }
956 } else if (!strcasecmp(argv[0],"port") && argc == 2) {
957 server.port = atoi(argv[1]);
958 if (server.port < 1 || server.port > 65535) {
959 err = "Invalid port"; goto loaderr;
960 }
961 } else if (!strcasecmp(argv[0],"bind") && argc == 2) {
962 server.bindaddr = zstrdup(argv[1]);
963 } else if (!strcasecmp(argv[0],"save") && argc == 3) {
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);
970 } else if (!strcasecmp(argv[0],"dir") && argc == 2) {
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 }
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;
980 else {
981 err = "Invalid log level. Must be one of debug, notice, warning";
982 goto loaderr;
983 }
984 } else if (!strcasecmp(argv[0],"logfile") && argc == 2) {
985 FILE *fp;
986
987 server.logfile = zstrdup(argv[1]);
988 if (!strcasecmp(server.logfile,"stdout")) {
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 }
1003 } else if (!strcasecmp(argv[0],"databases") && argc == 2) {
1004 server.dbnum = atoi(argv[1]);
1005 if (server.dbnum < 1) {
1006 err = "Invalid number of databases"; goto loaderr;
1007 }
1008 } else if (!strcasecmp(argv[0],"slaveof") && argc == 3) {
1009 server.masterhost = sdsnew(argv[1]);
1010 server.masterport = atoi(argv[2]);
1011 server.replstate = REDIS_REPL_CONNECT;
1012 } else if (!strcasecmp(argv[0],"glueoutputbuf") && argc == 2) {
1013 if ((server.glueoutputbuf = yesnotoi(argv[1])) == -1) {
1014 err = "argument must be 'yes' or 'no'"; goto loaderr;
1015 }
1016 } else if (!strcasecmp(argv[0],"shareobjects") && argc == 2) {
1017 if ((server.shareobjects = yesnotoi(argv[1])) == -1) {
1018 err = "argument must be 'yes' or 'no'"; goto loaderr;
1019 }
1020 } else if (!strcasecmp(argv[0],"daemonize") && argc == 2) {
1021 if ((server.daemonize = yesnotoi(argv[1])) == -1) {
1022 err = "argument must be 'yes' or 'no'"; goto loaderr;
1023 }
1024 } else if (!strcasecmp(argv[0],"requirepass") && argc == 2) {
1025 server.requirepass = zstrdup(argv[1]);
1026 } else if (!strcasecmp(argv[0],"pidfile") && argc == 2) {
1027 server.pidfile = zstrdup(argv[1]);
1028 } else if (!strcasecmp(argv[0],"dbfilename") && argc == 2) {
1029 server.dbfilename = zstrdup(argv[1]);
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
1041 loaderr:
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
1049 static 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
1057 static 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) {
1070 if (c->replstate == REDIS_REPL_SEND_BULK && c->repldbfd != -1)
1071 close(c->repldbfd);
1072 list *l = (c->flags & REDIS_MONITOR) ? server.monitors : server.slaves;
1073 ln = listSearchKey(l,c);
1074 assert(ln != NULL);
1075 listDelNode(l,ln);
1076 }
1077 if (c->flags & REDIS_MASTER) {
1078 server.master = NULL;
1079 server.replstate = REDIS_REPL_CONNECT;
1080 }
1081 zfree(c->argv);
1082 zfree(c);
1083 }
1084
1085 static void glueReplyBuffersIfNeeded(redisClient *c) {
1086 int totlen = 0;
1087 listNode *ln;
1088 robj *o;
1089
1090 listRewind(c->reply);
1091 while((ln = listYield(c->reply))) {
1092 o = ln->value;
1093 totlen += sdslen(o->ptr);
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
1102 listRewind(c->reply);
1103 while((ln = listYield(c->reply))) {
1104 o = ln->value;
1105 memcpy(buf+copylen,o->ptr,sdslen(o->ptr));
1106 copylen += sdslen(o->ptr);
1107 listDelNode(c->reply,ln);
1108 }
1109 /* Now the output buffer is empty, add the new single element */
1110 addReplySds(c,sdsnewlen(buf,totlen));
1111 }
1112 }
1113
1114 static 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 {
1135 nwritten = write(fd, ((char*)o->ptr)+c->sentlen, objlen - c->sentlen);
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
1163 static struct redisCommand *lookupCommand(char *name) {
1164 int j = 0;
1165 while(cmdTable[j].name != NULL) {
1166 if (!strcasecmp(name,cmdTable[j].name)) return &cmdTable[j];
1167 j++;
1168 }
1169 return NULL;
1170 }
1171
1172 /* resetClient prepare the client to process the next command */
1173 static 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). */
1186 static int processCommand(redisClient *c) {
1187 struct redisCommand *cmd;
1188 long long dirty;
1189
1190 /* The QUIT command is handled as a special case. Normal command
1191 * procs are unable to close the client connection safely */
1192 if (!strcasecmp(c->argv[0]->ptr,"quit")) {
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 }
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 }
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
1241 /* Exec the command */
1242 dirty = server.dirty;
1243 cmd->proc(c);
1244 if (server.dirty-dirty != 0 && listLength(server.slaves))
1245 replicationFeedSlaves(server.slaves,cmd,c->db->id,c->argv,c->argc);
1246 if (listLength(server.monitors))
1247 replicationFeedSlaves(server.monitors,cmd,c->db->id,c->argv,c->argc);
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
1259 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc) {
1260 listNode *ln;
1261 int outc = 0, j;
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 }
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
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]);
1291 listRewind(slaves);
1292 while((ln = listYield(slaves))) {
1293 redisClient *slave = ln->value;
1294
1295 /* Don't feed slaves that are still waiting for BGSAVE to start */
1296 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) continue;
1297
1298 /* Feed all the other slaves, MONITORs and so on */
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]);
1323 }
1324 for (j = 0; j < outc; j++) decrRefCount(outv[j]);
1325 if (outv != static_outv) zfree(outv);
1326 }
1327
1328 static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
1329 redisClient *c = (redisClient*) privdata;
1330 char buf[REDIS_IOBUF_LEN];
1331 int nread;
1332 REDIS_NOTUSED(el);
1333 REDIS_NOTUSED(mask);
1334
1335 nread = read(fd, buf, REDIS_IOBUF_LEN);
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
1356 again:
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);
1383 if (argv == NULL) oom("sdssplitlen");
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++) {
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
1427 static int selectDb(redisClient *c, int id) {
1428 if (id < 0 || id >= server.dbnum)
1429 return REDIS_ERR;
1430 c->db = &server.db[id];
1431 return REDIS_OK;
1432 }
1433
1434 static void *dupClientReplyValue(void *o) {
1435 incrRefCount((robj*)o);
1436 return 0;
1437 }
1438
1439 static 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;
1449 c->argv = NULL;
1450 c->bulklen = -1;
1451 c->sentlen = 0;
1452 c->flags = 0;
1453 c->lastinteraction = time(NULL);
1454 c->authenticated = 0;
1455 c->replstate = REDIS_REPL_NONE;
1456 if ((c->reply = listCreate()) == NULL) oom("listCreate");
1457 listSetFreeMethod(c->reply,decrRefCount);
1458 listSetDupMethod(c->reply,dupClientReplyValue);
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
1468 static void addReply(redisClient *c, robj *obj) {
1469 if (listLength(c->reply) == 0 &&
1470 (c->replstate == REDIS_REPL_NONE ||
1471 c->replstate == REDIS_REPL_ONLINE) &&
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
1478 static void addReplySds(redisClient *c, sds s) {
1479 robj *o = createObject(REDIS_STRING,s);
1480 addReply(c,o);
1481 decrRefCount(o);
1482 }
1483
1484 static 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
1507 static 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
1524 static robj *createStringObject(char *ptr, size_t len) {
1525 return createObject(REDIS_STRING,sdsnewlen(ptr,len));
1526 }
1527
1528 static 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
1536 static robj *createSetObject(void) {
1537 dict *d = dictCreate(&setDictType,NULL);
1538 if (!d) oom("dictCreate");
1539 return createObject(REDIS_SET,d);
1540 }
1541
1542 static void freeStringObject(robj *o) {
1543 sdsfree(o->ptr);
1544 }
1545
1546 static void freeListObject(robj *o) {
1547 listRelease((list*) o->ptr);
1548 }
1549
1550 static void freeSetObject(robj *o) {
1551 dictRelease((dict*) o->ptr);
1552 }
1553
1554 static void freeHashObject(robj *o) {
1555 dictRelease((dict*) o->ptr);
1556 }
1557
1558 static void incrRefCount(robj *o) {
1559 o->refcount++;
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
1564 }
1565
1566 static void decrRefCount(void *obj) {
1567 robj *o = obj;
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
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
1587 /* Try to share an object against the shared objects pool */
1588 static robj *tryObjectSharing(robj *o) {
1589 struct dictEntry *de;
1590 unsigned long c;
1591
1592 if (o == NULL || server.shareobjects == 0) return o;
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. */
1609 if (dictSize(server.sharingpool) >=
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
1632 static robj *lookupKey(redisDb *db, robj *key) {
1633 dictEntry *de = dictFind(db->dict,key);
1634 return de ? dictGetEntryVal(de) : NULL;
1635 }
1636
1637 static robj *lookupKeyRead(redisDb *db, robj *key) {
1638 expireIfNeeded(db,key);
1639 return lookupKey(db,key);
1640 }
1641
1642 static robj *lookupKeyWrite(redisDb *db, robj *key) {
1643 deleteIfVolatile(db,key);
1644 return lookupKey(db,key);
1645 }
1646
1647 static int deleteKey(redisDb *db, robj *key) {
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);
1655 if (dictSize(db->expires)) dictDelete(db->expires,key);
1656 retval = dictDelete(db->dict,key);
1657 decrRefCount(key);
1658
1659 return retval == DICT_OK;
1660 }
1661
1662 /*============================ DB saving/loading ============================ */
1663
1664 static int rdbSaveType(FILE *fp, unsigned char type) {
1665 if (fwrite(&type,1,1,fp) == 0) return -1;
1666 return 0;
1667 }
1668
1669 static 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
1675 /* check rdbLoadLen() comments for more info */
1676 static int rdbSaveLen(FILE *fp, uint32_t len) {
1677 unsigned char buf[2];
1678
1679 if (len < (1<<6)) {
1680 /* Save a 6 bit len */
1681 buf[0] = (len&0xFF)|(REDIS_RDB_6BITLEN<<6);
1682 if (fwrite(buf,1,1,fp) == 0) return -1;
1683 } else if (len < (1<<14)) {
1684 /* Save a 14 bit len */
1685 buf[0] = ((len>>8)&0xFF)|(REDIS_RDB_14BITLEN<<6);
1686 buf[1] = len&0xFF;
1687 if (fwrite(buf,2,1,fp) == 0) return -1;
1688 } else {
1689 /* Save a 32 bit len */
1690 buf[0] = (REDIS_RDB_32BITLEN<<6);
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
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 */
1701 int 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
1736 static 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;
1744 if ((out = zmalloc(outlen+1)) == NULL) return 0;
1745 comprlen = lzf_compress(obj->ptr, sdslen(obj->ptr), out, outlen);
1746 if (comprlen == 0) {
1747 zfree(out);
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;
1756 zfree(out);
1757 return comprlen;
1758
1759 writeerr:
1760 zfree(out);
1761 return -1;
1762 }
1763
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 */
1766 static int rdbSaveStringObject(FILE *fp, robj *obj) {
1767 size_t len = sdslen(obj->ptr);
1768 int enclen;
1769
1770 /* Try integer encoding */
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 }
1778
1779 /* Try LZF compression - under 20 bytes it's unable to compress even
1780 * aaaaaaaaaaaaaaaaaa so skip it */
1781 if (1 && len > 20) {
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 */
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
1796 /* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success */
1797 static int rdbSave(char *filename) {
1798 dictIterator *di = NULL;
1799 dictEntry *de;
1800 FILE *fp;
1801 char tmpfile[256];
1802 int j;
1803 time_t now = time(NULL);
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 }
1811 if (fwrite("REDIS0001",9,1,fp) == 0) goto werr;
1812 for (j = 0; j < server.dbnum; j++) {
1813 redisDb *db = server.db+j;
1814 dict *d = db->dict;
1815 if (dictSize(d) == 0) continue;
1816 di = dictGetIterator(d);
1817 if (!di) {
1818 fclose(fp);
1819 return REDIS_ERR;
1820 }
1821
1822 /* Write the SELECT DB opcode */
1823 if (rdbSaveType(fp,REDIS_SELECTDB) == -1) goto werr;
1824 if (rdbSaveLen(fp,j) == -1) goto werr;
1825
1826 /* Iterate this DB writing every entry */
1827 while((de = dictNext(di)) != NULL) {
1828 robj *key = dictGetEntryKey(de);
1829 robj *o = dictGetEntryVal(de);
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 */
1840 if (rdbSaveType(fp,o->type) == -1) goto werr;
1841 if (rdbSaveStringObject(fp,key) == -1) goto werr;
1842 if (o->type == REDIS_STRING) {
1843 /* Save a string value */
1844 if (rdbSaveStringObject(fp,o) == -1) goto werr;
1845 } else if (o->type == REDIS_LIST) {
1846 /* Save a list value */
1847 list *list = o->ptr;
1848 listNode *ln;
1849
1850 listRewind(list);
1851 if (rdbSaveLen(fp,listLength(list)) == -1) goto werr;
1852 while((ln = listYield(list))) {
1853 robj *eleobj = listNodeValue(ln);
1854
1855 if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
1856 }
1857 } else if (o->type == REDIS_SET) {
1858 /* Save a set value */
1859 dict *set = o->ptr;
1860 dictIterator *di = dictGetIterator(set);
1861 dictEntry *de;
1862
1863 if (!set) oom("dictGetIteraotr");
1864 if (rdbSaveLen(fp,dictSize(set)) == -1) goto werr;
1865 while((de = dictNext(di)) != NULL) {
1866 robj *eleobj = dictGetEntryKey(de);
1867
1868 if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
1869 }
1870 dictReleaseIterator(di);
1871 } else {
1872 assert(0 != 0);
1873 }
1874 }
1875 dictReleaseIterator(di);
1876 }
1877 /* EOF opcode */
1878 if (rdbSaveType(fp,REDIS_EOF) == -1) goto werr;
1879
1880 /* Make sure data will not remain on the OS's output buffers */
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
1897 werr:
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
1905 static int rdbSaveBackground(char *filename) {
1906 pid_t childpid;
1907
1908 if (server.bgsaveinprogress) return REDIS_ERR;
1909 if ((childpid = fork()) == 0) {
1910 /* Child */
1911 close(server.fd);
1912 if (rdbSave(filename) == REDIS_OK) {
1913 exit(0);
1914 } else {
1915 exit(1);
1916 }
1917 } else {
1918 /* Parent */
1919 if (childpid == -1) {
1920 redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
1921 strerror(errno));
1922 return REDIS_ERR;
1923 }
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
1931 static int rdbLoadType(FILE *fp) {
1932 unsigned char type;
1933 if (fread(&type,1,1,fp) == 0) return -1;
1934 return type;
1935 }
1936
1937 static 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
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 */
1948 static uint32_t rdbLoadLen(FILE *fp, int rdbver, int *isencoded) {
1949 unsigned char buf[2];
1950 uint32_t len;
1951
1952 if (isencoded) *isencoded = 0;
1953 if (rdbver == 0) {
1954 if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
1955 return ntohl(len);
1956 } else {
1957 int type;
1958
1959 if (fread(buf,1,1,fp) == 0) return REDIS_RDB_LENERR;
1960 type = (buf[0]&0xC0)>>6;
1961 if (type == REDIS_RDB_6BITLEN) {
1962 /* Read a 6 bit len */
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;
1968 } else if (type == REDIS_RDB_14BITLEN) {
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 }
1978 }
1979
1980 static 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
2004 static 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;
2015 return createObject(REDIS_STRING,val);
2016 err:
2017 zfree(c);
2018 sdsfree(val);
2019 return NULL;
2020 }
2021
2022 static robj *rdbLoadStringObject(FILE*fp, int rdbver) {
2023 int isencoded;
2024 uint32_t len;
2025 sds val;
2026
2027 len = rdbLoadLen(fp,rdbver,&isencoded);
2028 if (isencoded) {
2029 switch(len) {
2030 case REDIS_RDB_ENC_INT8:
2031 case REDIS_RDB_ENC_INT16:
2032 case REDIS_RDB_ENC_INT32:
2033 return tryObjectSharing(rdbLoadIntegerObject(fp,len));
2034 case REDIS_RDB_ENC_LZF:
2035 return tryObjectSharing(rdbLoadLzfStringObject(fp,rdbver));
2036 default:
2037 assert(0!=0);
2038 }
2039 }
2040
2041 if (len == REDIS_RDB_LENERR) return NULL;
2042 val = sdsnewlen(NULL,len);
2043 if (len && fread(val,len,1,fp) == 0) {
2044 sdsfree(val);
2045 return NULL;
2046 }
2047 return tryObjectSharing(createObject(REDIS_STRING,val));
2048 }
2049
2050 static int rdbLoad(char *filename) {
2051 FILE *fp;
2052 robj *keyobj = NULL;
2053 uint32_t dbid;
2054 int type, retval, rdbver;
2055 dict *d = server.db[0].dict;
2056 redisDb *db = server.db+0;
2057 char buf[1024];
2058 time_t expiretime = -1, now = time(NULL);
2059
2060 fp = fopen(filename,"r");
2061 if (!fp) return REDIS_ERR;
2062 if (fread(buf,9,1,fp) == 0) goto eoferr;
2063 buf[9] = '\0';
2064 if (memcmp(buf,"REDIS",5) != 0) {
2065 fclose(fp);
2066 redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
2067 return REDIS_ERR;
2068 }
2069 rdbver = atoi(buf+5);
2070 if (rdbver > 1) {
2071 fclose(fp);
2072 redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
2073 return REDIS_ERR;
2074 }
2075 while(1) {
2076 robj *o;
2077
2078 /* Read type. */
2079 if ((type = rdbLoadType(fp)) == -1) goto eoferr;
2080 if (type == REDIS_EXPIRETIME) {
2081 if ((expiretime = rdbLoadTime(fp)) == -1) goto eoferr;
2082 /* We read the time so we need to read the object type again */
2083 if ((type = rdbLoadType(fp)) == -1) goto eoferr;
2084 }
2085 if (type == REDIS_EOF) break;
2086 /* Handle SELECT DB opcode as a special case */
2087 if (type == REDIS_SELECTDB) {
2088 if ((dbid = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR)
2089 goto eoferr;
2090 if (dbid >= (unsigned)server.dbnum) {
2091 redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
2092 exit(1);
2093 }
2094 db = server.db+dbid;
2095 d = db->dict;
2096 continue;
2097 }
2098 /* Read key */
2099 if ((keyobj = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
2100
2101 if (type == REDIS_STRING) {
2102 /* Read string value */
2103 if ((o = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
2104 } else if (type == REDIS_LIST || type == REDIS_SET) {
2105 /* Read list/set value */
2106 uint32_t listlen;
2107
2108 if ((listlen = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR)
2109 goto eoferr;
2110 o = (type == REDIS_LIST) ? createListObject() : createSetObject();
2111 /* Load every single element of the list/set */
2112 while(listlen--) {
2113 robj *ele;
2114
2115 if ((ele = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
2116 if (type == REDIS_LIST) {
2117 if (!listAddNodeTail((list*)o->ptr,ele))
2118 oom("listAddNodeTail");
2119 } else {
2120 if (dictAdd((dict*)o->ptr,ele,NULL) == DICT_ERR)
2121 oom("dictAdd");
2122 }
2123 }
2124 } else {
2125 assert(0 != 0);
2126 }
2127 /* Add the new object in the hash table */
2128 retval = dictAdd(d,keyobj,o);
2129 if (retval == DICT_ERR) {
2130 redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", keyobj->ptr);
2131 exit(1);
2132 }
2133 /* Set the expire time if needed */
2134 if (expiretime != -1) {
2135 setExpire(db,keyobj,expiretime);
2136 /* Delete this key if already expired */
2137 if (expiretime < now) deleteKey(db,keyobj);
2138 expiretime = -1;
2139 }
2140 keyobj = o = NULL;
2141 }
2142 fclose(fp);
2143 return REDIS_OK;
2144
2145 eoferr: /* unexpected end of file is handled here with a fatal exit */
2146 if (keyobj) decrRefCount(keyobj);
2147 redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, exiting now.");
2148 exit(1);
2149 return REDIS_ERR; /* Just to avoid warning */
2150 }
2151
2152 /*================================== Commands =============================== */
2153
2154 static void authCommand(redisClient *c) {
2155 if (!server.requirepass || !strcmp(c->argv[1]->ptr, server.requirepass)) {
2156 c->authenticated = 1;
2157 addReply(c,shared.ok);
2158 } else {
2159 c->authenticated = 0;
2160 addReply(c,shared.err);
2161 }
2162 }
2163
2164 static void pingCommand(redisClient *c) {
2165 addReply(c,shared.pong);
2166 }
2167
2168 static void echoCommand(redisClient *c) {
2169 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
2170 (int)sdslen(c->argv[1]->ptr)));
2171 addReply(c,c->argv[1]);
2172 addReply(c,shared.crlf);
2173 }
2174
2175 /*=================================== Strings =============================== */
2176
2177 static void setGenericCommand(redisClient *c, int nx) {
2178 int retval;
2179
2180 retval = dictAdd(c->db->dict,c->argv[1],c->argv[2]);
2181 if (retval == DICT_ERR) {
2182 if (!nx) {
2183 dictReplace(c->db->dict,c->argv[1],c->argv[2]);
2184 incrRefCount(c->argv[2]);
2185 } else {
2186 addReply(c,shared.czero);
2187 return;
2188 }
2189 } else {
2190 incrRefCount(c->argv[1]);
2191 incrRefCount(c->argv[2]);
2192 }
2193 server.dirty++;
2194 removeExpire(c->db,c->argv[1]);
2195 addReply(c, nx ? shared.cone : shared.ok);
2196 }
2197
2198 static void setCommand(redisClient *c) {
2199 setGenericCommand(c,0);
2200 }
2201
2202 static void setnxCommand(redisClient *c) {
2203 setGenericCommand(c,1);
2204 }
2205
2206 static void getCommand(redisClient *c) {
2207 robj *o = lookupKeyRead(c->db,c->argv[1]);
2208
2209 if (o == NULL) {
2210 addReply(c,shared.nullbulk);
2211 } else {
2212 if (o->type != REDIS_STRING) {
2213 addReply(c,shared.wrongtypeerr);
2214 } else {
2215 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(o->ptr)));
2216 addReply(c,o);
2217 addReply(c,shared.crlf);
2218 }
2219 }
2220 }
2221
2222 static void getSetCommand(redisClient *c) {
2223 getCommand(c);
2224 if (dictAdd(c->db->dict,c->argv[1],c->argv[2]) == DICT_ERR) {
2225 dictReplace(c->db->dict,c->argv[1],c->argv[2]);
2226 } else {
2227 incrRefCount(c->argv[1]);
2228 }
2229 incrRefCount(c->argv[2]);
2230 server.dirty++;
2231 removeExpire(c->db,c->argv[1]);
2232 }
2233
2234 static void mgetCommand(redisClient *c) {
2235 int j;
2236
2237 addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",c->argc-1));
2238 for (j = 1; j < c->argc; j++) {
2239 robj *o = lookupKeyRead(c->db,c->argv[j]);
2240 if (o == NULL) {
2241 addReply(c,shared.nullbulk);
2242 } else {
2243 if (o->type != REDIS_STRING) {
2244 addReply(c,shared.nullbulk);
2245 } else {
2246 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(o->ptr)));
2247 addReply(c,o);
2248 addReply(c,shared.crlf);
2249 }
2250 }
2251 }
2252 }
2253
2254 static void incrDecrCommand(redisClient *c, long long incr) {
2255 long long value;
2256 int retval;
2257 robj *o;
2258
2259 o = lookupKeyWrite(c->db,c->argv[1]);
2260 if (o == NULL) {
2261 value = 0;
2262 } else {
2263 if (o->type != REDIS_STRING) {
2264 value = 0;
2265 } else {
2266 char *eptr;
2267
2268 value = strtoll(o->ptr, &eptr, 10);
2269 }
2270 }
2271
2272 value += incr;
2273 o = createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",value));
2274 retval = dictAdd(c->db->dict,c->argv[1],o);
2275 if (retval == DICT_ERR) {
2276 dictReplace(c->db->dict,c->argv[1],o);
2277 removeExpire(c->db,c->argv[1]);
2278 } else {
2279 incrRefCount(c->argv[1]);
2280 }
2281 server.dirty++;
2282 addReply(c,shared.colon);
2283 addReply(c,o);
2284 addReply(c,shared.crlf);
2285 }
2286
2287 static void incrCommand(redisClient *c) {
2288 incrDecrCommand(c,1);
2289 }
2290
2291 static void decrCommand(redisClient *c) {
2292 incrDecrCommand(c,-1);
2293 }
2294
2295 static void incrbyCommand(redisClient *c) {
2296 long long incr = strtoll(c->argv[2]->ptr, NULL, 10);
2297 incrDecrCommand(c,incr);
2298 }
2299
2300 static void decrbyCommand(redisClient *c) {
2301 long long incr = strtoll(c->argv[2]->ptr, NULL, 10);
2302 incrDecrCommand(c,-incr);
2303 }
2304
2305 /* ========================= Type agnostic commands ========================= */
2306
2307 static void delCommand(redisClient *c) {
2308 if (deleteKey(c->db,c->argv[1])) {
2309 server.dirty++;
2310 addReply(c,shared.cone);
2311 } else {
2312 addReply(c,shared.czero);
2313 }
2314 }
2315
2316 static void existsCommand(redisClient *c) {
2317 addReply(c,lookupKeyRead(c->db,c->argv[1]) ? shared.cone : shared.czero);
2318 }
2319
2320 static void selectCommand(redisClient *c) {
2321 int id = atoi(c->argv[1]->ptr);
2322
2323 if (selectDb(c,id) == REDIS_ERR) {
2324 addReplySds(c,sdsnew("-ERR invalid DB index\r\n"));
2325 } else {
2326 addReply(c,shared.ok);
2327 }
2328 }
2329
2330 static void randomkeyCommand(redisClient *c) {
2331 dictEntry *de;
2332
2333 while(1) {
2334 de = dictGetRandomKey(c->db->dict);
2335 if (!de || expireIfNeeded(c->db,dictGetEntryKey(de)) == 0) break;
2336 }
2337 if (de == NULL) {
2338 addReply(c,shared.plus);
2339 addReply(c,shared.crlf);
2340 } else {
2341 addReply(c,shared.plus);
2342 addReply(c,dictGetEntryKey(de));
2343 addReply(c,shared.crlf);
2344 }
2345 }
2346
2347 static void keysCommand(redisClient *c) {
2348 dictIterator *di;
2349 dictEntry *de;
2350 sds pattern = c->argv[1]->ptr;
2351 int plen = sdslen(pattern);
2352 int numkeys = 0, keyslen = 0;
2353 robj *lenobj = createObject(REDIS_STRING,NULL);
2354
2355 di = dictGetIterator(c->db->dict);
2356 if (!di) oom("dictGetIterator");
2357 addReply(c,lenobj);
2358 decrRefCount(lenobj);
2359 while((de = dictNext(di)) != NULL) {
2360 robj *keyobj = dictGetEntryKey(de);
2361
2362 sds key = keyobj->ptr;
2363 if ((pattern[0] == '*' && pattern[1] == '\0') ||
2364 stringmatchlen(pattern,plen,key,sdslen(key),0)) {
2365 if (expireIfNeeded(c->db,keyobj) == 0) {
2366 if (numkeys != 0)
2367 addReply(c,shared.space);
2368 addReply(c,keyobj);
2369 numkeys++;
2370 keyslen += sdslen(key);
2371 }
2372 }
2373 }
2374 dictReleaseIterator(di);
2375 lenobj->ptr = sdscatprintf(sdsempty(),"$%lu\r\n",keyslen+(numkeys ? (numkeys-1) : 0));
2376 addReply(c,shared.crlf);
2377 }
2378
2379 static void dbsizeCommand(redisClient *c) {
2380 addReplySds(c,
2381 sdscatprintf(sdsempty(),":%lu\r\n",dictSize(c->db->dict)));
2382 }
2383
2384 static void lastsaveCommand(redisClient *c) {
2385 addReplySds(c,
2386 sdscatprintf(sdsempty(),":%lu\r\n",server.lastsave));
2387 }
2388
2389 static void typeCommand(redisClient *c) {
2390 robj *o;
2391 char *type;
2392
2393 o = lookupKeyRead(c->db,c->argv[1]);
2394 if (o == NULL) {
2395 type = "+none";
2396 } else {
2397 switch(o->type) {
2398 case REDIS_STRING: type = "+string"; break;
2399 case REDIS_LIST: type = "+list"; break;
2400 case REDIS_SET: type = "+set"; break;
2401 default: type = "unknown"; break;
2402 }
2403 }
2404 addReplySds(c,sdsnew(type));
2405 addReply(c,shared.crlf);
2406 }
2407
2408 static void saveCommand(redisClient *c) {
2409 if (server.bgsaveinprogress) {
2410 addReplySds(c,sdsnew("-ERR background save in progress\r\n"));
2411 return;
2412 }
2413 if (rdbSave(server.dbfilename) == REDIS_OK) {
2414 addReply(c,shared.ok);
2415 } else {
2416 addReply(c,shared.err);
2417 }
2418 }
2419
2420 static void bgsaveCommand(redisClient *c) {
2421 if (server.bgsaveinprogress) {
2422 addReplySds(c,sdsnew("-ERR background save already in progress\r\n"));
2423 return;
2424 }
2425 if (rdbSaveBackground(server.dbfilename) == REDIS_OK) {
2426 addReply(c,shared.ok);
2427 } else {
2428 addReply(c,shared.err);
2429 }
2430 }
2431
2432 static void shutdownCommand(redisClient *c) {
2433 redisLog(REDIS_WARNING,"User requested shutdown, saving DB...");
2434 /* XXX: TODO kill the child if there is a bgsave in progress */
2435 if (rdbSave(server.dbfilename) == REDIS_OK) {
2436 if (server.daemonize) {
2437 unlink(server.pidfile);
2438 }
2439 redisLog(REDIS_WARNING,"%zu bytes used at exit",zmalloc_used_memory());
2440 redisLog(REDIS_WARNING,"Server exit now, bye bye...");
2441 exit(1);
2442 } else {
2443 redisLog(REDIS_WARNING,"Error trying to save the DB, can't exit");
2444 addReplySds(c,sdsnew("-ERR can't quit, problems saving the DB\r\n"));
2445 }
2446 }
2447
2448 static void renameGenericCommand(redisClient *c, int nx) {
2449 robj *o;
2450
2451 /* To use the same key as src and dst is probably an error */
2452 if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) {
2453 addReply(c,shared.sameobjecterr);
2454 return;
2455 }
2456
2457 o = lookupKeyWrite(c->db,c->argv[1]);
2458 if (o == NULL) {
2459 addReply(c,shared.nokeyerr);
2460 return;
2461 }
2462 incrRefCount(o);
2463 deleteIfVolatile(c->db,c->argv[2]);
2464 if (dictAdd(c->db->dict,c->argv[2],o) == DICT_ERR) {
2465 if (nx) {
2466 decrRefCount(o);
2467 addReply(c,shared.czero);
2468 return;
2469 }
2470 dictReplace(c->db->dict,c->argv[2],o);
2471 } else {
2472 incrRefCount(c->argv[2]);
2473 }
2474 deleteKey(c->db,c->argv[1]);
2475 server.dirty++;
2476 addReply(c,nx ? shared.cone : shared.ok);
2477 }
2478
2479 static void renameCommand(redisClient *c) {
2480 renameGenericCommand(c,0);
2481 }
2482
2483 static void renamenxCommand(redisClient *c) {
2484 renameGenericCommand(c,1);
2485 }
2486
2487 static void moveCommand(redisClient *c) {
2488 robj *o;
2489 redisDb *src, *dst;
2490 int srcid;
2491
2492 /* Obtain source and target DB pointers */
2493 src = c->db;
2494 srcid = c->db->id;
2495 if (selectDb(c,atoi(c->argv[2]->ptr)) == REDIS_ERR) {
2496 addReply(c,shared.outofrangeerr);
2497 return;
2498 }
2499 dst = c->db;
2500 selectDb(c,srcid); /* Back to the source DB */
2501
2502 /* If the user is moving using as target the same
2503 * DB as the source DB it is probably an error. */
2504 if (src == dst) {
2505 addReply(c,shared.sameobjecterr);
2506 return;
2507 }
2508
2509 /* Check if the element exists and get a reference */
2510 o = lookupKeyWrite(c->db,c->argv[1]);
2511 if (!o) {
2512 addReply(c,shared.czero);
2513 return;
2514 }
2515
2516 /* Try to add the element to the target DB */
2517 deleteIfVolatile(dst,c->argv[1]);
2518 if (dictAdd(dst->dict,c->argv[1],o) == DICT_ERR) {
2519 addReply(c,shared.czero);
2520 return;
2521 }
2522 incrRefCount(c->argv[1]);
2523 incrRefCount(o);
2524
2525 /* OK! key moved, free the entry in the source DB */
2526 deleteKey(src,c->argv[1]);
2527 server.dirty++;
2528 addReply(c,shared.cone);
2529 }
2530
2531 /* =================================== Lists ================================ */
2532 static void pushGenericCommand(redisClient *c, int where) {
2533 robj *lobj;
2534 list *list;
2535
2536 lobj = lookupKeyWrite(c->db,c->argv[1]);
2537 if (lobj == NULL) {
2538 lobj = createListObject();
2539 list = lobj->ptr;
2540 if (where == REDIS_HEAD) {
2541 if (!listAddNodeHead(list,c->argv[2])) oom("listAddNodeHead");
2542 } else {
2543 if (!listAddNodeTail(list,c->argv[2])) oom("listAddNodeTail");
2544 }
2545 dictAdd(c->db->dict,c->argv[1],lobj);
2546 incrRefCount(c->argv[1]);
2547 incrRefCount(c->argv[2]);
2548 } else {
2549 if (lobj->type != REDIS_LIST) {
2550 addReply(c,shared.wrongtypeerr);
2551 return;
2552 }
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 }
2559 incrRefCount(c->argv[2]);
2560 }
2561 server.dirty++;
2562 addReply(c,shared.ok);
2563 }
2564
2565 static void lpushCommand(redisClient *c) {
2566 pushGenericCommand(c,REDIS_HEAD);
2567 }
2568
2569 static void rpushCommand(redisClient *c) {
2570 pushGenericCommand(c,REDIS_TAIL);
2571 }
2572
2573 static void llenCommand(redisClient *c) {
2574 robj *o;
2575 list *l;
2576
2577 o = lookupKeyRead(c->db,c->argv[1]);
2578 if (o == NULL) {
2579 addReply(c,shared.czero);
2580 return;
2581 } else {
2582 if (o->type != REDIS_LIST) {
2583 addReply(c,shared.wrongtypeerr);
2584 } else {
2585 l = o->ptr;
2586 addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",listLength(l)));
2587 }
2588 }
2589 }
2590
2591 static void lindexCommand(redisClient *c) {
2592 robj *o;
2593 int index = atoi(c->argv[2]->ptr);
2594
2595 o = lookupKeyRead(c->db,c->argv[1]);
2596 if (o == NULL) {
2597 addReply(c,shared.nullbulk);
2598 } else {
2599 if (o->type != REDIS_LIST) {
2600 addReply(c,shared.wrongtypeerr);
2601 } else {
2602 list *list = o->ptr;
2603 listNode *ln;
2604
2605 ln = listIndex(list, index);
2606 if (ln == NULL) {
2607 addReply(c,shared.nullbulk);
2608 } else {
2609 robj *ele = listNodeValue(ln);
2610 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
2611 addReply(c,ele);
2612 addReply(c,shared.crlf);
2613 }
2614 }
2615 }
2616 }
2617
2618 static void lsetCommand(redisClient *c) {
2619 robj *o;
2620 int index = atoi(c->argv[2]->ptr);
2621
2622 o = lookupKeyWrite(c->db,c->argv[1]);
2623 if (o == NULL) {
2624 addReply(c,shared.nokeyerr);
2625 } else {
2626 if (o->type != REDIS_LIST) {
2627 addReply(c,shared.wrongtypeerr);
2628 } else {
2629 list *list = o->ptr;
2630 listNode *ln;
2631
2632 ln = listIndex(list, index);
2633 if (ln == NULL) {
2634 addReply(c,shared.outofrangeerr);
2635 } else {
2636 robj *ele = listNodeValue(ln);
2637
2638 decrRefCount(ele);
2639 listNodeValue(ln) = c->argv[3];
2640 incrRefCount(c->argv[3]);
2641 addReply(c,shared.ok);
2642 server.dirty++;
2643 }
2644 }
2645 }
2646 }
2647
2648 static void popGenericCommand(redisClient *c, int where) {
2649 robj *o;
2650
2651 o = lookupKeyWrite(c->db,c->argv[1]);
2652 if (o == NULL) {
2653 addReply(c,shared.nullbulk);
2654 } else {
2655 if (o->type != REDIS_LIST) {
2656 addReply(c,shared.wrongtypeerr);
2657 } else {
2658 list *list = o->ptr;
2659 listNode *ln;
2660
2661 if (where == REDIS_HEAD)
2662 ln = listFirst(list);
2663 else
2664 ln = listLast(list);
2665
2666 if (ln == NULL) {
2667 addReply(c,shared.nullbulk);
2668 } else {
2669 robj *ele = listNodeValue(ln);
2670 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
2671 addReply(c,ele);
2672 addReply(c,shared.crlf);
2673 listDelNode(list,ln);
2674 server.dirty++;
2675 }
2676 }
2677 }
2678 }
2679
2680 static void lpopCommand(redisClient *c) {
2681 popGenericCommand(c,REDIS_HEAD);
2682 }
2683
2684 static void rpopCommand(redisClient *c) {
2685 popGenericCommand(c,REDIS_TAIL);
2686 }
2687
2688 static void lrangeCommand(redisClient *c) {
2689 robj *o;
2690 int start = atoi(c->argv[2]->ptr);
2691 int end = atoi(c->argv[3]->ptr);
2692
2693 o = lookupKeyRead(c->db,c->argv[1]);
2694 if (o == NULL) {
2695 addReply(c,shared.nullmultibulk);
2696 } else {
2697 if (o->type != REDIS_LIST) {
2698 addReply(c,shared.wrongtypeerr);
2699 } else {
2700 list *list = o->ptr;
2701 listNode *ln;
2702 int llen = listLength(list);
2703 int rangelen, j;
2704 robj *ele;
2705
2706 /* convert negative indexes */
2707 if (start < 0) start = llen+start;
2708 if (end < 0) end = llen+end;
2709 if (start < 0) start = 0;
2710 if (end < 0) end = 0;
2711
2712 /* indexes sanity checks */
2713 if (start > end || start >= llen) {
2714 /* Out of range start or start > end result in empty list */
2715 addReply(c,shared.emptymultibulk);
2716 return;
2717 }
2718 if (end >= llen) end = llen-1;
2719 rangelen = (end-start)+1;
2720
2721 /* Return the result in form of a multi-bulk reply */
2722 ln = listIndex(list, start);
2723 addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",rangelen));
2724 for (j = 0; j < rangelen; j++) {
2725 ele = listNodeValue(ln);
2726 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
2727 addReply(c,ele);
2728 addReply(c,shared.crlf);
2729 ln = ln->next;
2730 }
2731 }
2732 }
2733 }
2734
2735 static void ltrimCommand(redisClient *c) {
2736 robj *o;
2737 int start = atoi(c->argv[2]->ptr);
2738 int end = atoi(c->argv[3]->ptr);
2739
2740 o = lookupKeyWrite(c->db,c->argv[1]);
2741 if (o == NULL) {
2742 addReply(c,shared.nokeyerr);
2743 } else {
2744 if (o->type != REDIS_LIST) {
2745 addReply(c,shared.wrongtypeerr);
2746 } else {
2747 list *list = o->ptr;
2748 listNode *ln;
2749 int llen = listLength(list);
2750 int j, ltrim, rtrim;
2751
2752 /* convert negative indexes */
2753 if (start < 0) start = llen+start;
2754 if (end < 0) end = llen+end;
2755 if (start < 0) start = 0;
2756 if (end < 0) end = 0;
2757
2758 /* indexes sanity checks */
2759 if (start > end || start >= llen) {
2760 /* Out of range start or start > end result in empty list */
2761 ltrim = llen;
2762 rtrim = 0;
2763 } else {
2764 if (end >= llen) end = llen-1;
2765 ltrim = start;
2766 rtrim = llen-end-1;
2767 }
2768
2769 /* Remove list elements to perform the trim */
2770 for (j = 0; j < ltrim; j++) {
2771 ln = listFirst(list);
2772 listDelNode(list,ln);
2773 }
2774 for (j = 0; j < rtrim; j++) {
2775 ln = listLast(list);
2776 listDelNode(list,ln);
2777 }
2778 addReply(c,shared.ok);
2779 server.dirty++;
2780 }
2781 }
2782 }
2783
2784 static void lremCommand(redisClient *c) {
2785 robj *o;
2786
2787 o = lookupKeyWrite(c->db,c->argv[1]);
2788 if (o == NULL) {
2789 addReply(c,shared.nokeyerr);
2790 } else {
2791 if (o->type != REDIS_LIST) {
2792 addReply(c,shared.wrongtypeerr);
2793 } else {
2794 list *list = o->ptr;
2795 listNode *ln, *next;
2796 int toremove = atoi(c->argv[2]->ptr);
2797 int removed = 0;
2798 int fromtail = 0;
2799
2800 if (toremove < 0) {
2801 toremove = -toremove;
2802 fromtail = 1;
2803 }
2804 ln = fromtail ? list->tail : list->head;
2805 while (ln) {
2806 robj *ele = listNodeValue(ln);
2807
2808 next = fromtail ? ln->prev : ln->next;
2809 if (sdscmp(ele->ptr,c->argv[3]->ptr) == 0) {
2810 listDelNode(list,ln);
2811 server.dirty++;
2812 removed++;
2813 if (toremove && removed == toremove) break;
2814 }
2815 ln = next;
2816 }
2817 addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",removed));
2818 }
2819 }
2820 }
2821
2822 /* ==================================== Sets ================================ */
2823
2824 static void saddCommand(redisClient *c) {
2825 robj *set;
2826
2827 set = lookupKeyWrite(c->db,c->argv[1]);
2828 if (set == NULL) {
2829 set = createSetObject();
2830 dictAdd(c->db->dict,c->argv[1],set);
2831 incrRefCount(c->argv[1]);
2832 } else {
2833 if (set->type != REDIS_SET) {
2834 addReply(c,shared.wrongtypeerr);
2835 return;
2836 }
2837 }
2838 if (dictAdd(set->ptr,c->argv[2],NULL) == DICT_OK) {
2839 incrRefCount(c->argv[2]);
2840 server.dirty++;
2841 addReply(c,shared.cone);
2842 } else {
2843 addReply(c,shared.czero);
2844 }
2845 }
2846
2847 static void sremCommand(redisClient *c) {
2848 robj *set;
2849
2850 set = lookupKeyWrite(c->db,c->argv[1]);
2851 if (set == NULL) {
2852 addReply(c,shared.czero);
2853 } else {
2854 if (set->type != REDIS_SET) {
2855 addReply(c,shared.wrongtypeerr);
2856 return;
2857 }
2858 if (dictDelete(set->ptr,c->argv[2]) == DICT_OK) {
2859 server.dirty++;
2860 addReply(c,shared.cone);
2861 } else {
2862 addReply(c,shared.czero);
2863 }
2864 }
2865 }
2866
2867 static void smoveCommand(redisClient *c) {
2868 robj *srcset, *dstset;
2869
2870 srcset = lookupKeyWrite(c->db,c->argv[1]);
2871 dstset = lookupKeyWrite(c->db,c->argv[2]);
2872
2873 /* If the source key does not exist return 0, if it's of the wrong type
2874 * raise an error */
2875 if (srcset == NULL || srcset->type != REDIS_SET) {
2876 addReply(c, srcset ? shared.wrongtypeerr : shared.czero);
2877 return;
2878 }
2879 /* Error if the destination key is not a set as well */
2880 if (dstset && dstset->type != REDIS_SET) {
2881 addReply(c,shared.wrongtypeerr);
2882 return;
2883 }
2884 /* Remove the element from the source set */
2885 if (dictDelete(srcset->ptr,c->argv[3]) == DICT_ERR) {
2886 /* Key not found in the src set! return zero */
2887 addReply(c,shared.czero);
2888 return;
2889 }
2890 server.dirty++;
2891 /* Add the element to the destination set */
2892 if (!dstset) {
2893 dstset = createSetObject();
2894 dictAdd(c->db->dict,c->argv[2],dstset);
2895 incrRefCount(c->argv[2]);
2896 }
2897 if (dictAdd(dstset->ptr,c->argv[3],NULL) == DICT_OK)
2898 incrRefCount(c->argv[3]);
2899 addReply(c,shared.cone);
2900 }
2901
2902 static void sismemberCommand(redisClient *c) {
2903 robj *set;
2904
2905 set = lookupKeyRead(c->db,c->argv[1]);
2906 if (set == NULL) {
2907 addReply(c,shared.czero);
2908 } else {
2909 if (set->type != REDIS_SET) {
2910 addReply(c,shared.wrongtypeerr);
2911 return;
2912 }
2913 if (dictFind(set->ptr,c->argv[2]))
2914 addReply(c,shared.cone);
2915 else
2916 addReply(c,shared.czero);
2917 }
2918 }
2919
2920 static void scardCommand(redisClient *c) {
2921 robj *o;
2922 dict *s;
2923
2924 o = lookupKeyRead(c->db,c->argv[1]);
2925 if (o == NULL) {
2926 addReply(c,shared.czero);
2927 return;
2928 } else {
2929 if (o->type != REDIS_SET) {
2930 addReply(c,shared.wrongtypeerr);
2931 } else {
2932 s = o->ptr;
2933 addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",
2934 dictSize(s)));
2935 }
2936 }
2937 }
2938
2939 static int qsortCompareSetsByCardinality(const void *s1, const void *s2) {
2940 dict **d1 = (void*) s1, **d2 = (void*) s2;
2941
2942 return dictSize(*d1)-dictSize(*d2);
2943 }
2944
2945 static void sinterGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey) {
2946 dict **dv = zmalloc(sizeof(dict*)*setsnum);
2947 dictIterator *di;
2948 dictEntry *de;
2949 robj *lenobj = NULL, *dstset = NULL;
2950 int j, cardinality = 0;
2951
2952 if (!dv) oom("sinterCommand");
2953 for (j = 0; j < setsnum; j++) {
2954 robj *setobj;
2955
2956 setobj = dstkey ?
2957 lookupKeyWrite(c->db,setskeys[j]) :
2958 lookupKeyRead(c->db,setskeys[j]);
2959 if (!setobj) {
2960 zfree(dv);
2961 if (dstkey) {
2962 deleteKey(c->db,dstkey);
2963 addReply(c,shared.ok);
2964 } else {
2965 addReply(c,shared.nullmultibulk);
2966 }
2967 return;
2968 }
2969 if (setobj->type != REDIS_SET) {
2970 zfree(dv);
2971 addReply(c,shared.wrongtypeerr);
2972 return;
2973 }
2974 dv[j] = setobj->ptr;
2975 }
2976 /* Sort sets from the smallest to largest, this will improve our
2977 * algorithm's performace */
2978 qsort(dv,setsnum,sizeof(dict*),qsortCompareSetsByCardinality);
2979
2980 /* The first thing we should output is the total number of elements...
2981 * since this is a multi-bulk write, but at this stage we don't know
2982 * the intersection set size, so we use a trick, append an empty object
2983 * to the output list and save the pointer to later modify it with the
2984 * right length */
2985 if (!dstkey) {
2986 lenobj = createObject(REDIS_STRING,NULL);
2987 addReply(c,lenobj);
2988 decrRefCount(lenobj);
2989 } else {
2990 /* If we have a target key where to store the resulting set
2991 * create this key with an empty set inside */
2992 dstset = createSetObject();
2993 deleteKey(c->db,dstkey);
2994 dictAdd(c->db->dict,dstkey,dstset);
2995 incrRefCount(dstkey);
2996 }
2997
2998 /* Iterate all the elements of the first (smallest) set, and test
2999 * the element against all the other sets, if at least one set does
3000 * not include the element it is discarded */
3001 di = dictGetIterator(dv[0]);
3002 if (!di) oom("dictGetIterator");
3003
3004 while((de = dictNext(di)) != NULL) {
3005 robj *ele;
3006
3007 for (j = 1; j < setsnum; j++)
3008 if (dictFind(dv[j],dictGetEntryKey(de)) == NULL) break;
3009 if (j != setsnum)
3010 continue; /* at least one set does not contain the member */
3011 ele = dictGetEntryKey(de);
3012 if (!dstkey) {
3013 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(ele->ptr)));
3014 addReply(c,ele);
3015 addReply(c,shared.crlf);
3016 cardinality++;
3017 } else {
3018 dictAdd(dstset->ptr,ele,NULL);
3019 incrRefCount(ele);
3020 }
3021 }
3022 dictReleaseIterator(di);
3023
3024 if (!dstkey) {
3025 lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality);
3026 } else {
3027 addReply(c,shared.ok);
3028 server.dirty++;
3029 }
3030 zfree(dv);
3031 }
3032
3033 static void sinterCommand(redisClient *c) {
3034 sinterGenericCommand(c,c->argv+1,c->argc-1,NULL);
3035 }
3036
3037 static void sinterstoreCommand(redisClient *c) {
3038 sinterGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
3039 }
3040
3041 static void sunionGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey) {
3042 dict **dv = zmalloc(sizeof(dict*)*setsnum);
3043 dictIterator *di;
3044 dictEntry *de;
3045 robj *lenobj = NULL, *dstset = NULL;
3046 int j, cardinality = 0;
3047
3048 if (!dv) oom("sunionCommand");
3049 for (j = 0; j < setsnum; j++) {
3050 robj *setobj;
3051
3052 setobj = dstkey ?
3053 lookupKeyWrite(c->db,setskeys[j]) :
3054 lookupKeyRead(c->db,setskeys[j]);
3055 if (!setobj) {
3056 dv[j] = NULL;
3057 continue;
3058 }
3059 if (setobj->type != REDIS_SET) {
3060 zfree(dv);
3061 addReply(c,shared.wrongtypeerr);
3062 return;
3063 }
3064 dv[j] = setobj->ptr;
3065 }
3066
3067 /* We need a temp set object to store our union. If the dstkey
3068 * is not NULL (that is, we are inside an SUNIONSTORE operation) then
3069 * this set object will be the resulting object to set into the target key*/
3070 dstset = createSetObject();
3071
3072 /* The first thing we should output is the total number of elements...
3073 * since this is a multi-bulk write, but at this stage we don't know
3074 * the intersection set size, so we use a trick, append an empty object
3075 * to the output list and save the pointer to later modify it with the
3076 * right length */
3077 if (!dstkey) {
3078 lenobj = createObject(REDIS_STRING,NULL);
3079 addReply(c,lenobj);
3080 decrRefCount(lenobj);
3081 } else {
3082 /* If we have a target key where to store the resulting set
3083 * create this key with an empty set inside */
3084 deleteKey(c->db,dstkey);
3085 dictAdd(c->db->dict,dstkey,dstset);
3086 incrRefCount(dstkey);
3087 server.dirty++;
3088 }
3089
3090 /* Iterate all the elements of all the sets, add every element a single
3091 * time to the result set */
3092 for (j = 0; j < setsnum; j++) {
3093 if (!dv[j]) continue; /* non existing keys are like empty sets */
3094
3095 di = dictGetIterator(dv[j]);
3096 if (!di) oom("dictGetIterator");
3097
3098 while((de = dictNext(di)) != NULL) {
3099 robj *ele;
3100
3101 /* dictAdd will not add the same element multiple times */
3102 ele = dictGetEntryKey(de);
3103 if (dictAdd(dstset->ptr,ele,NULL) == DICT_OK) {
3104 incrRefCount(ele);
3105 if (!dstkey) {
3106 addReplySds(c,sdscatprintf(sdsempty(),
3107 "$%d\r\n",sdslen(ele->ptr)));
3108 addReply(c,ele);
3109 addReply(c,shared.crlf);
3110 cardinality++;
3111 }
3112 }
3113 }
3114 dictReleaseIterator(di);
3115 }
3116
3117 if (!dstkey) {
3118 lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality);
3119 decrRefCount(dstset);
3120 } else {
3121 addReply(c,shared.ok);
3122 server.dirty++;
3123 }
3124 zfree(dv);
3125 }
3126
3127 static void sunionCommand(redisClient *c) {
3128 sunionGenericCommand(c,c->argv+1,c->argc-1,NULL);
3129 }
3130
3131 static void sunionstoreCommand(redisClient *c) {
3132 sunionGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
3133 }
3134
3135 static void flushdbCommand(redisClient *c) {
3136 server.dirty += dictSize(c->db->dict);
3137 dictEmpty(c->db->dict);
3138 dictEmpty(c->db->expires);
3139 addReply(c,shared.ok);
3140 }
3141
3142 static void flushallCommand(redisClient *c) {
3143 server.dirty += emptyDb();
3144 addReply(c,shared.ok);
3145 rdbSave(server.dbfilename);
3146 server.dirty++;
3147 }
3148
3149 redisSortOperation *createSortOperation(int type, robj *pattern) {
3150 redisSortOperation *so = zmalloc(sizeof(*so));
3151 if (!so) oom("createSortOperation");
3152 so->type = type;
3153 so->pattern = pattern;
3154 return so;
3155 }
3156
3157 /* Return the value associated to the key with a name obtained
3158 * substituting the first occurence of '*' in 'pattern' with 'subst' */
3159 robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
3160 char *p;
3161 sds spat, ssub;
3162 robj keyobj;
3163 int prefixlen, sublen, postfixlen;
3164 /* Expoit the internal sds representation to create a sds string allocated on the stack in order to make this function faster */
3165 struct {
3166 long len;
3167 long free;
3168 char buf[REDIS_SORTKEY_MAX+1];
3169 } keyname;
3170
3171 spat = pattern->ptr;
3172 ssub = subst->ptr;
3173 if (sdslen(spat)+sdslen(ssub)-1 > REDIS_SORTKEY_MAX) return NULL;
3174 p = strchr(spat,'*');
3175 if (!p) return NULL;
3176
3177 prefixlen = p-spat;
3178 sublen = sdslen(ssub);
3179 postfixlen = sdslen(spat)-(prefixlen+1);
3180 memcpy(keyname.buf,spat,prefixlen);
3181 memcpy(keyname.buf+prefixlen,ssub,sublen);
3182 memcpy(keyname.buf+prefixlen+sublen,p+1,postfixlen);
3183 keyname.buf[prefixlen+sublen+postfixlen] = '\0';
3184 keyname.len = prefixlen+sublen+postfixlen;
3185
3186 keyobj.refcount = 1;
3187 keyobj.type = REDIS_STRING;
3188 keyobj.ptr = ((char*)&keyname)+(sizeof(long)*2);
3189
3190 /* printf("lookup '%s' => %p\n", keyname.buf,de); */
3191 return lookupKeyRead(db,&keyobj);
3192 }
3193
3194 /* sortCompare() is used by qsort in sortCommand(). Given that qsort_r with
3195 * the additional parameter is not standard but a BSD-specific we have to
3196 * pass sorting parameters via the global 'server' structure */
3197 static int sortCompare(const void *s1, const void *s2) {
3198 const redisSortObject *so1 = s1, *so2 = s2;
3199 int cmp;
3200
3201 if (!server.sort_alpha) {
3202 /* Numeric sorting. Here it's trivial as we precomputed scores */
3203 if (so1->u.score > so2->u.score) {
3204 cmp = 1;
3205 } else if (so1->u.score < so2->u.score) {
3206 cmp = -1;
3207 } else {
3208 cmp = 0;
3209 }
3210 } else {
3211 /* Alphanumeric sorting */
3212 if (server.sort_bypattern) {
3213 if (!so1->u.cmpobj || !so2->u.cmpobj) {
3214 /* At least one compare object is NULL */
3215 if (so1->u.cmpobj == so2->u.cmpobj)
3216 cmp = 0;
3217 else if (so1->u.cmpobj == NULL)
3218 cmp = -1;
3219 else
3220 cmp = 1;
3221 } else {
3222 /* We have both the objects, use strcoll */
3223 cmp = strcoll(so1->u.cmpobj->ptr,so2->u.cmpobj->ptr);
3224 }
3225 } else {
3226 /* Compare elements directly */
3227 cmp = strcoll(so1->obj->ptr,so2->obj->ptr);
3228 }
3229 }
3230 return server.sort_desc ? -cmp : cmp;
3231 }
3232
3233 /* The SORT command is the most complex command in Redis. Warning: this code
3234 * is optimized for speed and a bit less for readability */
3235 static void sortCommand(redisClient *c) {
3236 list *operations;
3237 int outputlen = 0;
3238 int desc = 0, alpha = 0;
3239 int limit_start = 0, limit_count = -1, start, end;
3240 int j, dontsort = 0, vectorlen;
3241 int getop = 0; /* GET operation counter */
3242 robj *sortval, *sortby = NULL;
3243 redisSortObject *vector; /* Resulting vector to sort */
3244
3245 /* Lookup the key to sort. It must be of the right types */
3246 sortval = lookupKeyRead(c->db,c->argv[1]);
3247 if (sortval == NULL) {
3248 addReply(c,shared.nokeyerr);
3249 return;
3250 }
3251 if (sortval->type != REDIS_SET && sortval->type != REDIS_LIST) {
3252 addReply(c,shared.wrongtypeerr);
3253 return;
3254 }
3255
3256 /* Create a list of operations to perform for every sorted element.
3257 * Operations can be GET/DEL/INCR/DECR */
3258 operations = listCreate();
3259 listSetFreeMethod(operations,zfree);
3260 j = 2;
3261
3262 /* Now we need to protect sortval incrementing its count, in the future
3263 * SORT may have options able to overwrite/delete keys during the sorting
3264 * and the sorted key itself may get destroied */
3265 incrRefCount(sortval);
3266
3267 /* The SORT command has an SQL-alike syntax, parse it */
3268 while(j < c->argc) {
3269 int leftargs = c->argc-j-1;
3270 if (!strcasecmp(c->argv[j]->ptr,"asc")) {
3271 desc = 0;
3272 } else if (!strcasecmp(c->argv[j]->ptr,"desc")) {
3273 desc = 1;
3274 } else if (!strcasecmp(c->argv[j]->ptr,"alpha")) {
3275 alpha = 1;
3276 } else if (!strcasecmp(c->argv[j]->ptr,"limit") && leftargs >= 2) {
3277 limit_start = atoi(c->argv[j+1]->ptr);
3278 limit_count = atoi(c->argv[j+2]->ptr);
3279 j+=2;
3280 } else if (!strcasecmp(c->argv[j]->ptr,"by") && leftargs >= 1) {
3281 sortby = c->argv[j+1];
3282 /* If the BY pattern does not contain '*', i.e. it is constant,
3283 * we don't need to sort nor to lookup the weight keys. */
3284 if (strchr(c->argv[j+1]->ptr,'*') == NULL) dontsort = 1;
3285 j++;
3286 } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
3287 listAddNodeTail(operations,createSortOperation(
3288 REDIS_SORT_GET,c->argv[j+1]));
3289 getop++;
3290 j++;
3291 } else if (!strcasecmp(c->argv[j]->ptr,"del") && leftargs >= 1) {
3292 listAddNodeTail(operations,createSortOperation(
3293 REDIS_SORT_DEL,c->argv[j+1]));
3294 j++;
3295 } else if (!strcasecmp(c->argv[j]->ptr,"incr") && leftargs >= 1) {
3296 listAddNodeTail(operations,createSortOperation(
3297 REDIS_SORT_INCR,c->argv[j+1]));
3298 j++;
3299 } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
3300 listAddNodeTail(operations,createSortOperation(
3301 REDIS_SORT_DECR,c->argv[j+1]));
3302 j++;
3303 } else {
3304 decrRefCount(sortval);
3305 listRelease(operations);
3306 addReply(c,shared.syntaxerr);
3307 return;
3308 }
3309 j++;
3310 }
3311
3312 /* Load the sorting vector with all the objects to sort */
3313 vectorlen = (sortval->type == REDIS_LIST) ?
3314 listLength((list*)sortval->ptr) :
3315 dictSize((dict*)sortval->ptr);
3316 vector = zmalloc(sizeof(redisSortObject)*vectorlen);
3317 if (!vector) oom("allocating objects vector for SORT");
3318 j = 0;
3319 if (sortval->type == REDIS_LIST) {
3320 list *list = sortval->ptr;
3321 listNode *ln;
3322
3323 listRewind(list);
3324 while((ln = listYield(list))) {
3325 robj *ele = ln->value;
3326 vector[j].obj = ele;
3327 vector[j].u.score = 0;
3328 vector[j].u.cmpobj = NULL;
3329 j++;
3330 }
3331 } else {
3332 dict *set = sortval->ptr;
3333 dictIterator *di;
3334 dictEntry *setele;
3335
3336 di = dictGetIterator(set);
3337 if (!di) oom("dictGetIterator");
3338 while((setele = dictNext(di)) != NULL) {
3339 vector[j].obj = dictGetEntryKey(setele);
3340 vector[j].u.score = 0;
3341 vector[j].u.cmpobj = NULL;
3342 j++;
3343 }
3344 dictReleaseIterator(di);
3345 }
3346 assert(j == vectorlen);
3347
3348 /* Now it's time to load the right scores in the sorting vector */
3349 if (dontsort == 0) {
3350 for (j = 0; j < vectorlen; j++) {
3351 if (sortby) {
3352 robj *byval;
3353
3354 byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
3355 if (!byval || byval->type != REDIS_STRING) continue;
3356 if (alpha) {
3357 vector[j].u.cmpobj = byval;
3358 incrRefCount(byval);
3359 } else {
3360 vector[j].u.score = strtod(byval->ptr,NULL);
3361 }
3362 } else {
3363 if (!alpha) vector[j].u.score = strtod(vector[j].obj->ptr,NULL);
3364 }
3365 }
3366 }
3367
3368 /* We are ready to sort the vector... perform a bit of sanity check
3369 * on the LIMIT option too. We'll use a partial version of quicksort. */
3370 start = (limit_start < 0) ? 0 : limit_start;
3371 end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
3372 if (start >= vectorlen) {
3373 start = vectorlen-1;
3374 end = vectorlen-2;
3375 }
3376 if (end >= vectorlen) end = vectorlen-1;
3377
3378 if (dontsort == 0) {
3379 server.sort_desc = desc;
3380 server.sort_alpha = alpha;
3381 server.sort_bypattern = sortby ? 1 : 0;
3382 qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare);
3383 }
3384
3385 /* Send command output to the output buffer, performing the specified
3386 * GET/DEL/INCR/DECR operations if any. */
3387 outputlen = getop ? getop*(end-start+1) : end-start+1;
3388 addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",outputlen));
3389 for (j = start; j <= end; j++) {
3390 listNode *ln;
3391 if (!getop) {
3392 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
3393 sdslen(vector[j].obj->ptr)));
3394 addReply(c,vector[j].obj);
3395 addReply(c,shared.crlf);
3396 }
3397 listRewind(operations);
3398 while((ln = listYield(operations))) {
3399 redisSortOperation *sop = ln->value;
3400 robj *val = lookupKeyByPattern(c->db,sop->pattern,
3401 vector[j].obj);
3402
3403 if (sop->type == REDIS_SORT_GET) {
3404 if (!val || val->type != REDIS_STRING) {
3405 addReply(c,shared.nullbulk);
3406 } else {
3407 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
3408 sdslen(val->ptr)));
3409 addReply(c,val);
3410 addReply(c,shared.crlf);
3411 }
3412 } else if (sop->type == REDIS_SORT_DEL) {
3413 /* TODO */
3414 }
3415 }
3416 }
3417
3418 /* Cleanup */
3419 decrRefCount(sortval);
3420 listRelease(operations);
3421 for (j = 0; j < vectorlen; j++) {
3422 if (sortby && alpha && vector[j].u.cmpobj)
3423 decrRefCount(vector[j].u.cmpobj);
3424 }
3425 zfree(vector);
3426 }
3427
3428 static void infoCommand(redisClient *c) {
3429 sds info;
3430 time_t uptime = time(NULL)-server.stat_starttime;
3431
3432 info = sdscatprintf(sdsempty(),
3433 "redis_version:%s\r\n"
3434 "connected_clients:%d\r\n"
3435 "connected_slaves:%d\r\n"
3436 "used_memory:%zu\r\n"
3437 "changes_since_last_save:%lld\r\n"
3438 "bgsave_in_progress:%d\r\n"
3439 "last_save_time:%d\r\n"
3440 "total_connections_received:%lld\r\n"
3441 "total_commands_processed:%lld\r\n"
3442 "uptime_in_seconds:%d\r\n"
3443 "uptime_in_days:%d\r\n"
3444 ,REDIS_VERSION,
3445 listLength(server.clients)-listLength(server.slaves),
3446 listLength(server.slaves),
3447 server.usedmemory,
3448 server.dirty,
3449 server.bgsaveinprogress,
3450 server.lastsave,
3451 server.stat_numconnections,
3452 server.stat_numcommands,
3453 uptime,
3454 uptime/(3600*24)
3455 );
3456 addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(info)));
3457 addReplySds(c,info);
3458 addReply(c,shared.crlf);
3459 }
3460
3461 static void monitorCommand(redisClient *c) {
3462 /* ignore MONITOR if aleady slave or in monitor mode */
3463 if (c->flags & REDIS_SLAVE) return;
3464
3465 c->flags |= (REDIS_SLAVE|REDIS_MONITOR);
3466 c->slaveseldb = 0;
3467 if (!listAddNodeTail(server.monitors,c)) oom("listAddNodeTail");
3468 addReply(c,shared.ok);
3469 }
3470
3471 /* ================================= Expire ================================= */
3472 static int removeExpire(redisDb *db, robj *key) {
3473 if (dictDelete(db->expires,key) == DICT_OK) {
3474 return 1;
3475 } else {
3476 return 0;
3477 }
3478 }
3479
3480 static int setExpire(redisDb *db, robj *key, time_t when) {
3481 if (dictAdd(db->expires,key,(void*)when) == DICT_ERR) {
3482 return 0;
3483 } else {
3484 incrRefCount(key);
3485 return 1;
3486 }
3487 }
3488
3489 /* Return the expire time of the specified key, or -1 if no expire
3490 * is associated with this key (i.e. the key is non volatile) */
3491 static time_t getExpire(redisDb *db, robj *key) {
3492 dictEntry *de;
3493
3494 /* No expire? return ASAP */
3495 if (dictSize(db->expires) == 0 ||
3496 (de = dictFind(db->expires,key)) == NULL) return -1;
3497
3498 return (time_t) dictGetEntryVal(de);
3499 }
3500
3501 static int expireIfNeeded(redisDb *db, robj *key) {
3502 time_t when;
3503 dictEntry *de;
3504
3505 /* No expire? return ASAP */
3506 if (dictSize(db->expires) == 0 ||
3507 (de = dictFind(db->expires,key)) == NULL) return 0;
3508
3509 /* Lookup the expire */
3510 when = (time_t) dictGetEntryVal(de);
3511 if (time(NULL) <= when) return 0;
3512
3513 /* Delete the key */
3514 dictDelete(db->expires,key);
3515 return dictDelete(db->dict,key) == DICT_OK;
3516 }
3517
3518 static int deleteIfVolatile(redisDb *db, robj *key) {
3519 dictEntry *de;
3520
3521 /* No expire? return ASAP */
3522 if (dictSize(db->expires) == 0 ||
3523 (de = dictFind(db->expires,key)) == NULL) return 0;
3524
3525 /* Delete the key */
3526 server.dirty++;
3527 dictDelete(db->expires,key);
3528 return dictDelete(db->dict,key) == DICT_OK;
3529 }
3530
3531 static void expireCommand(redisClient *c) {
3532 dictEntry *de;
3533 int seconds = atoi(c->argv[2]->ptr);
3534
3535 de = dictFind(c->db->dict,c->argv[1]);
3536 if (de == NULL) {
3537 addReply(c,shared.czero);
3538 return;
3539 }
3540 if (seconds <= 0) {
3541 addReply(c, shared.czero);
3542 return;
3543 } else {
3544 time_t when = time(NULL)+seconds;
3545 if (setExpire(c->db,c->argv[1],when))
3546 addReply(c,shared.cone);
3547 else
3548 addReply(c,shared.czero);
3549 return;
3550 }
3551 }
3552
3553 /* =============================== Replication ============================= */
3554
3555 static int syncWrite(int fd, char *ptr, ssize_t size, int timeout) {
3556 ssize_t nwritten, ret = size;
3557 time_t start = time(NULL);
3558
3559 timeout++;
3560 while(size) {
3561 if (aeWait(fd,AE_WRITABLE,1000) & AE_WRITABLE) {
3562 nwritten = write(fd,ptr,size);
3563 if (nwritten == -1) return -1;
3564 ptr += nwritten;
3565 size -= nwritten;
3566 }
3567 if ((time(NULL)-start) > timeout) {
3568 errno = ETIMEDOUT;
3569 return -1;
3570 }
3571 }
3572 return ret;
3573 }
3574
3575 static int syncRead(int fd, char *ptr, ssize_t size, int timeout) {
3576 ssize_t nread, totread = 0;
3577 time_t start = time(NULL);
3578
3579 timeout++;
3580 while(size) {
3581 if (aeWait(fd,AE_READABLE,1000) & AE_READABLE) {
3582 nread = read(fd,ptr,size);
3583 if (nread == -1) return -1;
3584 ptr += nread;
3585 size -= nread;
3586 totread += nread;
3587 }
3588 if ((time(NULL)-start) > timeout) {
3589 errno = ETIMEDOUT;
3590 return -1;
3591 }
3592 }
3593 return totread;
3594 }
3595
3596 static int syncReadLine(int fd, char *ptr, ssize_t size, int timeout) {
3597 ssize_t nread = 0;
3598
3599 size--;
3600 while(size) {
3601 char c;
3602
3603 if (syncRead(fd,&c,1,timeout) == -1) return -1;
3604 if (c == '\n') {
3605 *ptr = '\0';
3606 if (nread && *(ptr-1) == '\r') *(ptr-1) = '\0';
3607 return nread;
3608 } else {
3609 *ptr++ = c;
3610 *ptr = '\0';
3611 nread++;
3612 }
3613 }
3614 return nread;
3615 }
3616
3617 static void syncCommand(redisClient *c) {
3618 /* ignore SYNC if aleady slave or in monitor mode */
3619 if (c->flags & REDIS_SLAVE) return;
3620
3621 /* SYNC can't be issued when the server has pending data to send to
3622 * the client about already issued commands. We need a fresh reply
3623 * buffer registering the differences between the BGSAVE and the current
3624 * dataset, so that we can copy to other slaves if needed. */
3625 if (listLength(c->reply) != 0) {
3626 addReplySds(c,sdsnew("-ERR SYNC is invalid with pending input\r\n"));
3627 return;
3628 }
3629
3630 redisLog(REDIS_NOTICE,"Slave ask for synchronization");
3631 /* Here we need to check if there is a background saving operation
3632 * in progress, or if it is required to start one */
3633 if (server.bgsaveinprogress) {
3634 /* Ok a background save is in progress. Let's check if it is a good
3635 * one for replication, i.e. if there is another slave that is
3636 * registering differences since the server forked to save */
3637 redisClient *slave;
3638 listNode *ln;
3639
3640 listRewind(server.slaves);
3641 while((ln = listYield(server.slaves))) {
3642 slave = ln->value;
3643 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
3644 }
3645 if (ln) {
3646 /* Perfect, the server is already registering differences for
3647 * another slave. Set the right state, and copy the buffer. */
3648 listRelease(c->reply);
3649 c->reply = listDup(slave->reply);
3650 if (!c->reply) oom("listDup copying slave reply list");
3651 c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
3652 redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC");
3653 } else {
3654 /* No way, we need to wait for the next BGSAVE in order to
3655 * register differences */
3656 c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
3657 redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
3658 }
3659 } else {
3660 /* Ok we don't have a BGSAVE in progress, let's start one */
3661 redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC");
3662 if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
3663 redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE");
3664 addReplySds(c,sdsnew("-ERR Unalbe to perform background save\r\n"));
3665 return;
3666 }
3667 c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
3668 }
3669 c->repldbfd = -1;
3670 c->flags |= REDIS_SLAVE;
3671 c->slaveseldb = 0;
3672 if (!listAddNodeTail(server.slaves,c)) oom("listAddNodeTail");
3673 return;
3674 }
3675
3676 static void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
3677 redisClient *slave = privdata;
3678 REDIS_NOTUSED(el);
3679 REDIS_NOTUSED(mask);
3680 char buf[REDIS_IOBUF_LEN];
3681 ssize_t nwritten, buflen;
3682
3683 if (slave->repldboff == 0) {
3684 /* Write the bulk write count before to transfer the DB. In theory here
3685 * we don't know how much room there is in the output buffer of the
3686 * socket, but in pratice SO_SNDLOWAT (the minimum count for output
3687 * operations) will never be smaller than the few bytes we need. */
3688 sds bulkcount;
3689
3690 bulkcount = sdscatprintf(sdsempty(),"$%lld\r\n",(unsigned long long)
3691 slave->repldbsize);
3692 if (write(fd,bulkcount,sdslen(bulkcount)) != (signed)sdslen(bulkcount))
3693 {
3694 sdsfree(bulkcount);
3695 freeClient(slave);
3696 return;
3697 }
3698 sdsfree(bulkcount);
3699 }
3700 lseek(slave->repldbfd,slave->repldboff,SEEK_SET);
3701 buflen = read(slave->repldbfd,buf,REDIS_IOBUF_LEN);
3702 if (buflen <= 0) {
3703 redisLog(REDIS_WARNING,"Read error sending DB to slave: %s",
3704 (buflen == 0) ? "premature EOF" : strerror(errno));
3705 freeClient(slave);
3706 return;
3707 }
3708 if ((nwritten = write(fd,buf,buflen)) == -1) {
3709 redisLog(REDIS_DEBUG,"Write error sending DB to slave: %s",
3710 strerror(errno));
3711 freeClient(slave);
3712 return;
3713 }
3714 slave->repldboff += nwritten;
3715 if (slave->repldboff == slave->repldbsize) {
3716 close(slave->repldbfd);
3717 slave->repldbfd = -1;
3718 aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
3719 slave->replstate = REDIS_REPL_ONLINE;
3720 if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE,
3721 sendReplyToClient, slave, NULL) == AE_ERR) {
3722 freeClient(slave);
3723 return;
3724 }
3725 addReplySds(slave,sdsempty());
3726 redisLog(REDIS_NOTICE,"Synchronization with slave succeeded");
3727 }
3728 }
3729
3730 static void updateSalvesWaitingBgsave(int bgsaveerr) {
3731 listNode *ln;
3732 int startbgsave = 0;
3733
3734 listRewind(server.slaves);
3735 while((ln = listYield(server.slaves))) {
3736 redisClient *slave = ln->value;
3737
3738 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
3739 startbgsave = 1;
3740 slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
3741 } else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
3742 struct stat buf;
3743
3744 if (bgsaveerr != REDIS_OK) {
3745 freeClient(slave);
3746 redisLog(REDIS_WARNING,"SYNC failed. BGSAVE child returned an error");
3747 continue;
3748 }
3749 if ((slave->repldbfd = open(server.dbfilename,O_RDONLY)) == -1 ||
3750 fstat(slave->repldbfd,&buf) == -1) {
3751 freeClient(slave);
3752 redisLog(REDIS_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno));
3753 continue;
3754 }
3755 slave->repldboff = 0;
3756 slave->repldbsize = buf.st_size;
3757 slave->replstate = REDIS_REPL_SEND_BULK;
3758 aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
3759 if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave, NULL) == AE_ERR) {
3760 freeClient(slave);
3761 continue;
3762 }
3763 }
3764 }
3765 if (startbgsave) {
3766 if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
3767 listRewind(server.slaves);
3768 redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
3769 while((ln = listYield(server.slaves))) {
3770 redisClient *slave = ln->value;
3771
3772 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
3773 freeClient(slave);
3774 }
3775 }
3776 }
3777 }
3778
3779 static int syncWithMaster(void) {
3780 char buf[1024], tmpfile[256];
3781 int dumpsize;
3782 int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);
3783 int dfd;
3784
3785 if (fd == -1) {
3786 redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
3787 strerror(errno));
3788 return REDIS_ERR;
3789 }
3790 /* Issue the SYNC command */
3791 if (syncWrite(fd,"SYNC \r\n",7,5) == -1) {
3792 close(fd);
3793 redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
3794 strerror(errno));
3795 return REDIS_ERR;
3796 }
3797 /* Read the bulk write count */
3798 if (syncReadLine(fd,buf,1024,5) == -1) {
3799 close(fd);
3800 redisLog(REDIS_WARNING,"I/O error reading bulk count from MASTER: %s",
3801 strerror(errno));
3802 return REDIS_ERR;
3803 }
3804 dumpsize = atoi(buf+1);
3805 redisLog(REDIS_NOTICE,"Receiving %d bytes data dump from MASTER",dumpsize);
3806 /* Read the bulk write data on a temp file */
3807 snprintf(tmpfile,256,"temp-%d.%ld.rdb",(int)time(NULL),(long int)random());
3808 dfd = open(tmpfile,O_CREAT|O_WRONLY,0644);
3809 if (dfd == -1) {
3810 close(fd);
3811 redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
3812 return REDIS_ERR;
3813 }
3814 while(dumpsize) {
3815 int nread, nwritten;
3816
3817 nread = read(fd,buf,(dumpsize < 1024)?dumpsize:1024);
3818 if (nread == -1) {
3819 redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
3820 strerror(errno));
3821 close(fd);
3822 close(dfd);
3823 return REDIS_ERR;
3824 }
3825 nwritten = write(dfd,buf,nread);
3826 if (nwritten == -1) {
3827 redisLog(REDIS_WARNING,"Write error writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno));
3828 close(fd);
3829 close(dfd);
3830 return REDIS_ERR;
3831 }
3832 dumpsize -= nread;
3833 }
3834 close(dfd);
3835 if (rename(tmpfile,server.dbfilename) == -1) {
3836 redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
3837 unlink(tmpfile);
3838 close(fd);
3839 return REDIS_ERR;
3840 }
3841 emptyDb();
3842 if (rdbLoad(server.dbfilename) != REDIS_OK) {
3843 redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
3844 close(fd);
3845 return REDIS_ERR;
3846 }
3847 server.master = createClient(fd);
3848 server.master->flags |= REDIS_MASTER;
3849 server.replstate = REDIS_REPL_CONNECTED;
3850 return REDIS_OK;
3851 }
3852
3853 /* =================================== Main! ================================ */
3854
3855 #ifdef __linux__
3856 int linuxOvercommitMemoryValue(void) {
3857 FILE *fp = fopen("/proc/sys/vm/overcommit_memory","r");
3858 char buf[64];
3859
3860 if (!fp) return -1;
3861 if (fgets(buf,64,fp) == NULL) {
3862 fclose(fp);
3863 return -1;
3864 }
3865 fclose(fp);
3866
3867 return atoi(buf);
3868 }
3869
3870 void linuxOvercommitMemoryWarning(void) {
3871 if (linuxOvercommitMemoryValue() == 0) {
3872 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.");
3873 }
3874 }
3875 #endif /* __linux__ */
3876
3877 static void daemonize(void) {
3878 int fd;
3879 FILE *fp;
3880
3881 if (fork() != 0) exit(0); /* parent exits */
3882 setsid(); /* create a new session */
3883
3884 /* Every output goes to /dev/null. If Redis is daemonized but
3885 * the 'logfile' is set to 'stdout' in the configuration file
3886 * it will not log at all. */
3887 if ((fd = open("/dev/null", O_RDWR, 0)) != -1) {
3888 dup2(fd, STDIN_FILENO);
3889 dup2(fd, STDOUT_FILENO);
3890 dup2(fd, STDERR_FILENO);
3891 if (fd > STDERR_FILENO) close(fd);
3892 }
3893 /* Try to write the pid file */
3894 fp = fopen(server.pidfile,"w");
3895 if (fp) {
3896 fprintf(fp,"%d\n",getpid());
3897 fclose(fp);
3898 }
3899 }
3900
3901 int main(int argc, char **argv) {
3902 #ifdef __linux__
3903 linuxOvercommitMemoryWarning();
3904 #endif
3905
3906 initServerConfig();
3907 if (argc == 2) {
3908 ResetServerSaveParams();
3909 loadServerConfig(argv[1]);
3910 } else if (argc > 2) {
3911 fprintf(stderr,"Usage: ./redis-server [/path/to/redis.conf]\n");
3912 exit(1);
3913 }
3914 initServer();
3915 if (server.daemonize) daemonize();
3916 redisLog(REDIS_NOTICE,"Server started, Redis version " REDIS_VERSION);
3917 if (rdbLoad(server.dbfilename) == REDIS_OK)
3918 redisLog(REDIS_NOTICE,"DB loaded from disk");
3919 if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
3920 acceptHandler, NULL, NULL) == AE_ERR) oom("creating file event");
3921 redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
3922 aeMain(server.el);
3923 aeDeleteEventLoop(server.el);
3924 return 0;
3925 }