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