]> git.saurik.com Git - redis.git/blob - redis.c
projects / redis.git / blob
? search:


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