git.saurik.com Git - redis.git/blame_incremental

... / ...

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