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