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