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


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