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


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