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