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


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