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