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 "0.091"
	31
	32	#include "fmacros.h"
	33
	34	#include <stdio.h>
	35	#include <stdlib.h>
	36	#include <string.h>
	37	#include <time.h>
	38	#include <unistd.h>
	39	#include <signal.h>
	40	#include <sys/wait.h>
	41	#include <errno.h>
	42	#include <assert.h>
	43	#include <ctype.h>
	44	#include <stdarg.h>
	45	#include <inttypes.h>
	46	#include <arpa/inet.h>
	47	#include <sys/stat.h>
	48	#include <fcntl.h>
	49	#include <sys/time.h>
	50	#include <sys/resource.h>
	51	#include <limits.h>
	52
	53	#include "ae.h" /* Event driven programming library */
	54	#include "sds.h" /* Dynamic safe strings */
	55	#include "anet.h" /* Networking the easy way */
	56	#include "dict.h" /* Hash tables */
	57	#include "adlist.h" /* Linked lists */
	58	#include "zmalloc.h" /* total memory usage aware version of malloc/free */
	59	#include "lzf.h"
	60
	61	/* Error codes */
	62	#define REDIS_OK 0
	63	#define REDIS_ERR -1
	64
	65	/* Static server configuration */
	66	#define REDIS_SERVERPORT 6379 /* TCP port */
	67	#define REDIS_MAXIDLETIME (605) / default client timeout */
	68	#define REDIS_QUERYBUF_LEN 1024
	69	#define REDIS_LOADBUF_LEN 1024
	70	#define REDIS_MAX_ARGS 16
	71	#define REDIS_DEFAULT_DBNUM 16
	72	#define REDIS_CONFIGLINE_MAX 1024
	73	#define REDIS_OBJFREELIST_MAX 1000000 /* Max number of objects to cache */
	74	#define REDIS_MAX_SYNC_TIME 60 /* Slave can't take more to sync */
	75	#define REDIS_EXPIRELOOKUPS_PER_CRON 100 /* try to expire 100 keys/second */
	76
	77	/* Hash table parameters */
	78	#define REDIS_HT_MINFILL 10 /* Minimal hash table fill 10% */
	79	#define REDIS_HT_MINSLOTS 16384 /* Never resize the HT under this */
	80
	81	/* Command flags */
	82	#define REDIS_CMD_BULK 1
	83	#define REDIS_CMD_INLINE 2
	84
	85	/* Object types */
	86	#define REDIS_STRING 0
	87	#define REDIS_LIST 1
	88	#define REDIS_SET 2
	89	#define REDIS_HASH 3
	90
	91	/* Object types only used for dumping to disk */
	92	#define REDIS_EXPIRETIME 253
	93	#define REDIS_SELECTDB 254
	94	#define REDIS_EOF 255
	95
	96	/* Defines related to the dump file format. To store 32 bits lengths for short
	97	* keys requires a lot of space, so we check the most significant 2 bits of
	98	* the first byte to interpreter the length:
	99	*
	100	* 00\|000000 => if the two MSB are 00 the len is the 6 bits of this byte
	101	* 01\|000000 00000000 => 01, the len is 14 byes, 6 bits + 8 bits of next byte
	102	* 10\|000000 [32 bit integer] => if it's 01, a full 32 bit len will follow
	103	* 11\|000000 this means: specially encoded object will follow. The six bits
	104	* number specify the kind of object that follows.
	105	* See the REDIS_RDB_ENC_* defines.
	106	*
	107	* Lenghts up to 63 are stored using a single byte, most DB keys, and may
	108	* values, will fit inside. */
	109	#define REDIS_RDB_6BITLEN 0
	110	#define REDIS_RDB_14BITLEN 1
	111	#define REDIS_RDB_32BITLEN 2
	112	#define REDIS_RDB_ENCVAL 3
	113	#define REDIS_RDB_LENERR UINT_MAX
	114
	115	/* When a length of a string object stored on disk has the first two bits
	116	* set, the remaining two bits specify a special encoding for the object
	117	* accordingly to the following defines: */
	118	#define REDIS_RDB_ENC_INT8 0 /* 8 bit signed integer */
	119	#define REDIS_RDB_ENC_INT16 1 /* 16 bit signed integer */
	120	#define REDIS_RDB_ENC_INT32 2 /* 32 bit signed integer */
	121	#define REDIS_RDB_ENC_LZF 3 /* string compressed with FASTLZ */
	122
	123	/* Client flags */
	124	#define REDIS_CLOSE 1 /* This client connection should be closed ASAP */
	125	#define REDIS_SLAVE 2 /* This client is a slave server */
	126	#define REDIS_MASTER 4 /* This client is a master server */
	127	#define REDIS_MONITOR 8 /* This client is a slave monitor, see MONITOR */
	128
	129	/* Slave replication state - slave side */
	130	#define REDIS_REPL_NONE 0 /* No active replication */
	131	#define REDIS_REPL_CONNECT 1 /* Must connect to master */
	132	#define REDIS_REPL_CONNECTED 2 /* Connected to master */
	133
	134	/* Slave replication state - from the point of view of master
	135	* Note that in SEND_BULK and ONLINE state the slave receives new updates
	136	* in its output queue. In the WAIT_BGSAVE state instead the server is waiting
	137	* to start the next background saving in order to send updates to it. */
	138	#define REDIS_REPL_WAIT_BGSAVE_START 3 /* master waits bgsave to start feeding it */
	139	#define REDIS_REPL_WAIT_BGSAVE_END 4 /* master waits bgsave to start bulk DB transmission */
	140	#define REDIS_REPL_SEND_BULK 5 /* master is sending the bulk DB */
	141	#define REDIS_REPL_ONLINE 6 /* bulk DB already transmitted, receive updates */
	142
	143	/* List related stuff */
	144	#define REDIS_HEAD 0
	145	#define REDIS_TAIL 1
	146
	147	/* Sort operations */
	148	#define REDIS_SORT_GET 0
	149	#define REDIS_SORT_DEL 1
	150	#define REDIS_SORT_INCR 2
	151	#define REDIS_SORT_DECR 3
	152	#define REDIS_SORT_ASC 4
	153	#define REDIS_SORT_DESC 5
	154	#define REDIS_SORTKEY_MAX 1024
	155
	156	/* Log levels */
	157	#define REDIS_DEBUG 0
	158	#define REDIS_NOTICE 1
	159	#define REDIS_WARNING 2
	160
	161	/* Anti-warning macro... */
	162	#define REDIS_NOTUSED(V) ((void) V)
	163
	164	/================================= Data types ============================== /
	165
	166	/* A redis object, that is a type able to hold a string / list / set */
	167	typedef struct redisObject {
	168	void *ptr;
	169	int type;
	170	int refcount;
	171	} robj;
	172
	173	typedef struct redisDb {
	174	dict *dict;
	175	dict *expires;
	176	int id;
	177	} redisDb;
	178
	179	/* With multiplexing we need to take per-clinet state.
	180	* Clients are taken in a liked list. */
	181	typedef struct redisClient {
	182	int fd;
	183	redisDb *db;
	184	int dictid;
	185	sds querybuf;
	186	robj *argv[REDIS_MAX_ARGS];
	187	int argc;
	188	int bulklen; /* bulk read len. -1 if not in bulk read mode */
	189	list *reply;
	190	int sentlen;
	191	time_t lastinteraction; /* time of the last interaction, used for timeout */
	192	int flags; /* REDIS_CLOSE \| REDIS_SLAVE \| REDIS_MONITOR */
	193	int slaveseldb; /* slave selected db, if this client is a slave */
	194	int authenticated; /* when requirepass is non-NULL */
	195	int replstate; /* replication state if this is a slave */
	196	int repldbfd; /* replication DB file descriptor */
	197	int repldboff; /* replication DB file offset */
	198	off_t repldbsize; /* replication DB file size */
	199	} redisClient;
	200
	201	struct saveparam {
	202	time_t seconds;
	203	int changes;
	204	};
	205
	206	/* Global server state structure */
	207	struct redisServer {
	208	int port;
	209	int fd;
	210	redisDb *db;
	211	dict *sharingpool;
	212	unsigned int sharingpoolsize;
	213	long long dirty; /* changes to DB from the last save */
	214	list *clients;
	215	list slaves, monitors;
	216	char neterr[ANET_ERR_LEN];
	217	aeEventLoop *el;
	218	int cronloops; /* number of times the cron function run */
	219	list objfreelist; / A list of freed objects to avoid malloc() */
	220	time_t lastsave; /* Unix time of last save succeeede */
	221	size_t usedmemory; /* Used memory in megabytes */
	222	/* Fields used only for stats */
	223	time_t stat_starttime; /* server start time */
	224	long long stat_numcommands; /* number of processed commands */
	225	long long stat_numconnections; /* number of connections received */
	226	/* Configuration */
	227	int verbosity;
	228	int glueoutputbuf;
	229	int maxidletime;
	230	int dbnum;
	231	int daemonize;
	232	char *pidfile;
	233	int bgsaveinprogress;
	234	struct saveparam *saveparams;
	235	int saveparamslen;
	236	char *logfile;
	237	char *bindaddr;
	238	char *dbfilename;
	239	char *requirepass;
	240	int shareobjects;
	241	/* Replication related */
	242	int isslave;
	243	char *masterhost;
	244	int masterport;
	245	redisClient master; / client that is master for this slave */
	246	int replstate;
	247	/* Sort parameters - qsort_r() is only available under BSD so we
	248	* have to take this state global, in order to pass it to sortCompare() */
	249	int sort_desc;
	250	int sort_alpha;
	251	int sort_bypattern;
	252	};
	253
	254	typedef void redisCommandProc(redisClient *c);
	255	struct redisCommand {
	256	char *name;
	257	redisCommandProc *proc;
	258	int arity;
	259	int flags;
	260	};
	261
	262	typedef struct _redisSortObject {
	263	robj *obj;
	264	union {
	265	double score;
	266	robj *cmpobj;
	267	} u;
	268	} redisSortObject;
	269
	270	typedef struct _redisSortOperation {
	271	int type;
	272	robj *pattern;
	273	} redisSortOperation;
	274
	275	struct sharedObjectsStruct {
	276	robj crlf, ok, err, emptybulk, czero, cone, pong, space,
	277	colon, nullbulk, *nullmultibulk,
	278	emptymultibulk, wrongtypeerr, nokeyerr, syntaxerr, *sameobjecterr,
	279	outofrangeerr, plus,
	280	select0, select1, select2, select3, *select4,
	281	select5, select6, select7, select8, *select9;
	282	} shared;
	283
	284	/================================ Prototypes =============================== /
	285
	286	static void freeStringObject(robj *o);
	287	static void freeListObject(robj *o);
	288	static void freeSetObject(robj *o);
	289	static void decrRefCount(void *o);
	290	static robj createObject(int type, void ptr);
	291	static void freeClient(redisClient *c);
	292	static int rdbLoad(char *filename);
	293	static void addReply(redisClient c, robj obj);
	294	static void addReplySds(redisClient *c, sds s);
	295	static void incrRefCount(robj *o);
	296	static int rdbSaveBackground(char *filename);
	297	static robj createStringObject(char ptr, size_t len);
	298	static void replicationFeedSlaves(list slaves, struct redisCommand cmd, int dictid, robj **argv, int argc);
	299	static int syncWithMaster(void);
	300	static robj tryObjectSharing(robj o);
	301	static int removeExpire(redisDb db, robj key);
	302	static int expireIfNeeded(redisDb db, robj key);
	303	static int deleteIfVolatile(redisDb db, robj key);
	304	static int deleteKey(redisDb db, robj key);
	305	static time_t getExpire(redisDb db, robj key);
	306	static int setExpire(redisDb db, robj key, time_t when);
	307
	308	static void authCommand(redisClient *c);
	309	static void pingCommand(redisClient *c);
	310	static void echoCommand(redisClient *c);
	311	static void setCommand(redisClient *c);
	312	static void setnxCommand(redisClient *c);
	313	static void getCommand(redisClient *c);
	314	static void delCommand(redisClient *c);
	315	static void existsCommand(redisClient *c);
	316	static void incrCommand(redisClient *c);
	317	static void decrCommand(redisClient *c);
	318	static void incrbyCommand(redisClient *c);
	319	static void decrbyCommand(redisClient *c);
	320	static void selectCommand(redisClient *c);
	321	static void randomkeyCommand(redisClient *c);
	322	static void keysCommand(redisClient *c);
	323	static void dbsizeCommand(redisClient *c);
	324	static void lastsaveCommand(redisClient *c);
	325	static void saveCommand(redisClient *c);
	326	static void bgsaveCommand(redisClient *c);
	327	static void shutdownCommand(redisClient *c);
	328	static void moveCommand(redisClient *c);
	329	static void renameCommand(redisClient *c);
	330	static void renamenxCommand(redisClient *c);
	331	static void lpushCommand(redisClient *c);
	332	static void rpushCommand(redisClient *c);
	333	static void lpopCommand(redisClient *c);
	334	static void rpopCommand(redisClient *c);
	335	static void llenCommand(redisClient *c);
	336	static void lindexCommand(redisClient *c);
	337	static void lrangeCommand(redisClient *c);
	338	static void ltrimCommand(redisClient *c);
	339	static void typeCommand(redisClient *c);
	340	static void lsetCommand(redisClient *c);
	341	static void saddCommand(redisClient *c);
	342	static void sremCommand(redisClient *c);
	343	static void sismemberCommand(redisClient *c);
	344	static void scardCommand(redisClient *c);
	345	static void sinterCommand(redisClient *c);
	346	static void sinterstoreCommand(redisClient *c);
	347	static void sunionCommand(redisClient *c);
	348	static void sunionstoreCommand(redisClient *c);
	349	static void syncCommand(redisClient *c);
	350	static void flushdbCommand(redisClient *c);
	351	static void flushallCommand(redisClient *c);
	352	static void sortCommand(redisClient *c);
	353	static void lremCommand(redisClient *c);
	354	static void infoCommand(redisClient *c);
	355	static void mgetCommand(redisClient *c);
	356	static void monitorCommand(redisClient *c);
	357	static void expireCommand(redisClient *c);
	358
	359	/================================= Globals ================================= /
	360
	361	/* Global vars */
	362	static struct redisServer server; /* server global state */
	363	static struct redisCommand cmdTable[] = {
	364	{"get",getCommand,2,REDIS_CMD_INLINE},
	365	{"set",setCommand,3,REDIS_CMD_BULK},
	366	{"setnx",setnxCommand,3,REDIS_CMD_BULK},
	367	{"del",delCommand,2,REDIS_CMD_INLINE},
	368	{"exists",existsCommand,2,REDIS_CMD_INLINE},
	369	{"incr",incrCommand,2,REDIS_CMD_INLINE},
	370	{"decr",decrCommand,2,REDIS_CMD_INLINE},
	371	{"mget",mgetCommand,-2,REDIS_CMD_INLINE},
	372	{"rpush",rpushCommand,3,REDIS_CMD_BULK},
	373	{"lpush",lpushCommand,3,REDIS_CMD_BULK},
	374	{"rpop",rpopCommand,2,REDIS_CMD_INLINE},
	375	{"lpop",lpopCommand,2,REDIS_CMD_INLINE},
	376	{"llen",llenCommand,2,REDIS_CMD_INLINE},
	377	{"lindex",lindexCommand,3,REDIS_CMD_INLINE},
	378	{"lset",lsetCommand,4,REDIS_CMD_BULK},
	379	{"lrange",lrangeCommand,4,REDIS_CMD_INLINE},
	380	{"ltrim",ltrimCommand,4,REDIS_CMD_INLINE},
	381	{"lrem",lremCommand,4,REDIS_CMD_BULK},
	382	{"sadd",saddCommand,3,REDIS_CMD_BULK},
	383	{"srem",sremCommand,3,REDIS_CMD_BULK},
	384	{"sismember",sismemberCommand,3,REDIS_CMD_BULK},
	385	{"scard",scardCommand,2,REDIS_CMD_INLINE},
	386	{"sinter",sinterCommand,-2,REDIS_CMD_INLINE},
	387	{"sinterstore",sinterstoreCommand,-3,REDIS_CMD_INLINE},
	388	{"sunion",sunionCommand,-2,REDIS_CMD_INLINE},
	389	{"sunionstore",sunionstoreCommand,-3,REDIS_CMD_INLINE},
	390	{"smembers",sinterCommand,2,REDIS_CMD_INLINE},
	391	{"incrby",incrbyCommand,3,REDIS_CMD_INLINE},
	392	{"decrby",decrbyCommand,3,REDIS_CMD_INLINE},
	393	{"randomkey",randomkeyCommand,1,REDIS_CMD_INLINE},
	394	{"select",selectCommand,2,REDIS_CMD_INLINE},
	395	{"move",moveCommand,3,REDIS_CMD_INLINE},
	396	{"rename",renameCommand,3,REDIS_CMD_INLINE},
	397	{"renamenx",renamenxCommand,3,REDIS_CMD_INLINE},
	398	{"keys",keysCommand,2,REDIS_CMD_INLINE},
	399	{"dbsize",dbsizeCommand,1,REDIS_CMD_INLINE},
	400	{"auth",authCommand,2,REDIS_CMD_INLINE},
	401	{"ping",pingCommand,1,REDIS_CMD_INLINE},
	402	{"echo",echoCommand,2,REDIS_CMD_BULK},
	403	{"save",saveCommand,1,REDIS_CMD_INLINE},
	404	{"bgsave",bgsaveCommand,1,REDIS_CMD_INLINE},
	405	{"shutdown",shutdownCommand,1,REDIS_CMD_INLINE},
	406	{"lastsave",lastsaveCommand,1,REDIS_CMD_INLINE},
	407	{"type",typeCommand,2,REDIS_CMD_INLINE},
	408	{"sync",syncCommand,1,REDIS_CMD_INLINE},
	409	{"flushdb",flushdbCommand,1,REDIS_CMD_INLINE},
	410	{"flushall",flushallCommand,1,REDIS_CMD_INLINE},
	411	{"sort",sortCommand,-2,REDIS_CMD_INLINE},
	412	{"info",infoCommand,1,REDIS_CMD_INLINE},
	413	{"monitor",monitorCommand,1,REDIS_CMD_INLINE},
	414	{"expire",expireCommand,3,REDIS_CMD_INLINE},
	415	{NULL,NULL,0,0}
	416	};
	417
	418	/============================ Utility functions ============================ /
	419
	420	/* Glob-style pattern matching. */
	421	int stringmatchlen(const char *pattern, int patternLen,
	422	const char *string, int stringLen, int nocase)
	423	{
	424	while(patternLen) {
	425	switch(pattern[0]) {
	426	case '*':
	427	while (pattern[1] == '*') {
	428	pattern++;
	429	patternLen--;
	430	}
	431	if (patternLen == 1)
	432	return 1; /* match */
	433	while(stringLen) {
	434	if (stringmatchlen(pattern+1, patternLen-1,
	435	string, stringLen, nocase))
	436	return 1; /* match */
	437	string++;
	438	stringLen--;
	439	}
	440	return 0; /* no match */
	441	break;
	442	case '?':
	443	if (stringLen == 0)
	444	return 0; /* no match */
	445	string++;
	446	stringLen--;
	447	break;
	448	case '[':
	449	{
	450	int not, match;
	451
	452	pattern++;
	453	patternLen--;
	454	not = pattern[0] == '^';
	455	if (not) {
	456	pattern++;
	457	patternLen--;
	458	}
	459	match = 0;
	460	while(1) {
	461	if (pattern[0] == '\\') {
	462	pattern++;
	463	patternLen--;
	464	if (pattern[0] == string[0])
	465	match = 1;
	466	} else if (pattern[0] == ']') {
	467	break;
	468	} else if (patternLen == 0) {
	469	pattern--;
	470	patternLen++;
	471	break;
	472	} else if (pattern[1] == '-' && patternLen >= 3) {
	473	int start = pattern[0];
	474	int end = pattern[2];
	475	int c = string[0];
	476	if (start > end) {
	477	int t = start;
	478	start = end;
	479	end = t;
	480	}
	481	if (nocase) {
	482	start = tolower(start);
	483	end = tolower(end);
	484	c = tolower(c);
	485	}
	486	pattern += 2;
	487	patternLen -= 2;
	488	if (c >= start && c <= end)
	489	match = 1;
	490	} else {
	491	if (!nocase) {
	492	if (pattern[0] == string[0])
	493	match = 1;
	494	} else {
	495	if (tolower((int)pattern[0]) == tolower((int)string[0]))
	496	match = 1;
	497	}
	498	}
	499	pattern++;
	500	patternLen--;
	501	}
	502	if (not)
	503	match = !match;
	504	if (!match)
	505	return 0; /* no match */
	506	string++;
	507	stringLen--;
	508	break;
	509	}
	510	case '\\':
	511	if (patternLen >= 2) {
	512	pattern++;
	513	patternLen--;
	514	}
	515	/* fall through */
	516	default:
	517	if (!nocase) {
	518	if (pattern[0] != string[0])
	519	return 0; /* no match */
	520	} else {
	521	if (tolower((int)pattern[0]) != tolower((int)string[0]))
	522	return 0; /* no match */
	523	}
	524	string++;
	525	stringLen--;
	526	break;
	527	}
	528	pattern++;
	529	patternLen--;
	530	if (stringLen == 0) {
	531	while(pattern == '') {
	532	pattern++;
	533	patternLen--;
	534	}
	535	break;
	536	}
	537	}
	538	if (patternLen == 0 && stringLen == 0)
	539	return 1;
	540	return 0;
	541	}
	542
	543	void redisLog(int level, const char *fmt, ...)
	544	{
	545	va_list ap;
	546	FILE *fp;
	547
	548	fp = (server.logfile == NULL) ? stdout : fopen(server.logfile,"a");
	549	if (!fp) return;
	550
	551	va_start(ap, fmt);
	552	if (level >= server.verbosity) {
	553	char c = ".-";
	554	fprintf(fp,"%c ",c[level]);
	555	vfprintf(fp, fmt, ap);
	556	fprintf(fp,"\n");
	557	fflush(fp);
	558	}
	559	va_end(ap);
	560
	561	if (server.logfile) fclose(fp);
	562	}
	563
	564	/====================== Hash table type implementation ==================== /
	565
	566	/* This is an hash table type that uses the SDS dynamic strings libary as
	567	* keys and radis objects as values (objects can hold SDS strings,
	568	* lists, sets). */
	569
	570	static int sdsDictKeyCompare(void privdata, const void key1,
	571	const void *key2)
	572	{
	573	int l1,l2;
	574	DICT_NOTUSED(privdata);
	575
	576	l1 = sdslen((sds)key1);
	577	l2 = sdslen((sds)key2);
	578	if (l1 != l2) return 0;
	579	return memcmp(key1, key2, l1) == 0;
	580	}
	581
	582	static void dictRedisObjectDestructor(void privdata, void val)
	583	{
	584	DICT_NOTUSED(privdata);
	585
	586	decrRefCount(val);
	587	}
	588
	589	static int dictSdsKeyCompare(void privdata, const void key1,
	590	const void *key2)
	591	{
	592	const robj o1 = key1, o2 = key2;
	593	return sdsDictKeyCompare(privdata,o1->ptr,o2->ptr);
	594	}
	595
	596	static unsigned int dictSdsHash(const void *key) {
	597	const robj *o = key;
	598	return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
	599	}
	600
	601	static dictType setDictType = {
	602	dictSdsHash, /* hash function */
	603	NULL, /* key dup */
	604	NULL, /* val dup */
	605	dictSdsKeyCompare, /* key compare */
	606	dictRedisObjectDestructor, /* key destructor */
	607	NULL /* val destructor */
	608	};
	609
	610	static dictType hashDictType = {
	611	dictSdsHash, /* hash function */
	612	NULL, /* key dup */
	613	NULL, /* val dup */
	614	dictSdsKeyCompare, /* key compare */
	615	dictRedisObjectDestructor, /* key destructor */
	616	dictRedisObjectDestructor /* val destructor */
	617	};
	618
	619	/* ========================= Random utility functions ======================= */
	620
	621	/* Redis generally does not try to recover from out of memory conditions
	622	* when allocating objects or strings, it is not clear if it will be possible
	623	* to report this condition to the client since the networking layer itself
	624	* is based on heap allocation for send buffers, so we simply abort.
	625	* At least the code will be simpler to read... */
	626	static void oom(const char *msg) {
	627	fprintf(stderr, "%s: Out of memory\n",msg);
	628	fflush(stderr);
	629	sleep(1);
	630	abort();
	631	}
	632
	633	/* ====================== Redis server networking stuff ===================== */
	634	void closeTimedoutClients(void) {
	635	redisClient *c;
	636	listIter *li;
	637	listNode *ln;
	638	time_t now = time(NULL);
	639
	640	li = listGetIterator(server.clients,AL_START_HEAD);
	641	if (!li) return;
	642	while ((ln = listNextElement(li)) != NULL) {
	643	c = listNodeValue(ln);
	644	if (!(c->flags & REDIS_SLAVE) && /* no timeout for slaves */
	645	(now - c->lastinteraction > server.maxidletime)) {
	646	redisLog(REDIS_DEBUG,"Closing idle client");
	647	freeClient(c);
	648	}
	649	}
	650	listReleaseIterator(li);
	651	}
	652
	653	int serverCron(struct aeEventLoop eventLoop, long long id, void clientData) {
	654	int j, loops = server.cronloops++;
	655	REDIS_NOTUSED(eventLoop);
	656	REDIS_NOTUSED(id);
	657	REDIS_NOTUSED(clientData);
	658
	659	/* Update the global state with the amount of used memory */
	660	server.usedmemory = zmalloc_used_memory();
	661
	662	/* If the percentage of used slots in the HT reaches REDIS_HT_MINFILL
	663	* we resize the hash table to save memory */
	664	for (j = 0; j < server.dbnum; j++) {
	665	int size, used, vkeys;
	666
	667	size = dictSlots(server.db[j].dict);
	668	used = dictSize(server.db[j].dict);
	669	vkeys = dictSize(server.db[j].expires);
	670	if (!(loops % 5) && used > 0) {
	671	redisLog(REDIS_DEBUG,"DB %d: %d keys (%d volatile) in %d slots HT.",j,used,vkeys,size);
	672	/* dictPrintStats(server.dict); */
	673	}
	674	if (size && used && size > REDIS_HT_MINSLOTS &&
	675	(used*100/size < REDIS_HT_MINFILL)) {
	676	redisLog(REDIS_NOTICE,"The hash table %d is too sparse, resize it...",j);
	677	dictResize(server.db[j].dict);
	678	redisLog(REDIS_NOTICE,"Hash table %d resized.",j);
	679	}
	680	}
	681
	682	/* Show information about connected clients */
	683	if (!(loops % 5)) {
	684	redisLog(REDIS_DEBUG,"%d clients connected (%d slaves), %zu bytes in use",
	685	listLength(server.clients)-listLength(server.slaves),
	686	listLength(server.slaves),
	687	server.usedmemory,
	688	dictSize(server.sharingpool));
	689	}
	690
	691	/* Close connections of timedout clients */
	692	if (!(loops % 10))
	693	closeTimedoutClients();
	694
	695	/* Check if a background saving in progress terminated */
	696	if (server.bgsaveinprogress) {
	697	int statloc;
	698	if (wait4(-1,&statloc,WNOHANG,NULL)) {
	699	int exitcode = WEXITSTATUS(statloc);
	700	if (exitcode == 0) {
	701	redisLog(REDIS_NOTICE,
	702	"Background saving terminated with success");
	703	server.dirty = 0;
	704	server.lastsave = time(NULL);
	705	} else {
	706	redisLog(REDIS_WARNING,
	707	"Background saving error");
	708	}
	709	server.bgsaveinprogress = 0;
	710	}
	711	} else {
	712	/* If there is not a background saving in progress check if
	713	* we have to save now */
	714	time_t now = time(NULL);
	715	for (j = 0; j < server.saveparamslen; j++) {
	716	struct saveparam *sp = server.saveparams+j;
	717
	718	if (server.dirty >= sp->changes &&
	719	now-server.lastsave > sp->seconds) {
	720	redisLog(REDIS_NOTICE,"%d changes in %d seconds. Saving...",
	721	sp->changes, sp->seconds);
	722	rdbSaveBackground(server.dbfilename);
	723	break;
	724	}
	725	}
	726	}
	727
	728	/* Try to expire a few timed out keys */
	729	for (j = 0; j < server.dbnum; j++) {
	730	redisDb *db = server.db+j;
	731	int num = dictSize(db->expires);
	732
	733	if (num) {
	734	time_t now = time(NULL);
	735
	736	if (num > REDIS_EXPIRELOOKUPS_PER_CRON)
	737	num = REDIS_EXPIRELOOKUPS_PER_CRON;
	738	while (num--) {
	739	dictEntry *de;
	740	time_t t;
	741
	742	if ((de = dictGetRandomKey(db->expires)) == NULL) break;
	743	t = (time_t) dictGetEntryVal(de);
	744	if (now > t) {
	745	deleteKey(db,dictGetEntryKey(de));
	746	}
	747	}
	748	}
	749	}
	750
	751	/* Check if we should connect to a MASTER */
	752	if (server.replstate == REDIS_REPL_CONNECT) {
	753	redisLog(REDIS_NOTICE,"Connecting to MASTER...");
	754	if (syncWithMaster() == REDIS_OK) {
	755	redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync succeeded");
	756	}
	757	}
	758	return 1000;
	759	}
	760
	761	static void createSharedObjects(void) {
	762	shared.crlf = createObject(REDIS_STRING,sdsnew("\r\n"));
	763	shared.ok = createObject(REDIS_STRING,sdsnew("+OK\r\n"));
	764	shared.err = createObject(REDIS_STRING,sdsnew("-ERR\r\n"));
	765	shared.emptybulk = createObject(REDIS_STRING,sdsnew("$0\r\n\r\n"));
	766	shared.czero = createObject(REDIS_STRING,sdsnew(":0\r\n"));
	767	shared.cone = createObject(REDIS_STRING,sdsnew(":1\r\n"));
	768	shared.nullbulk = createObject(REDIS_STRING,sdsnew("$-1\r\n"));
	769	shared.nullmultibulk = createObject(REDIS_STRING,sdsnew("*-1\r\n"));
	770	shared.emptymultibulk = createObject(REDIS_STRING,sdsnew("*0\r\n"));
	771	/* no such key */
	772	shared.pong = createObject(REDIS_STRING,sdsnew("+PONG\r\n"));
	773	shared.wrongtypeerr = createObject(REDIS_STRING,sdsnew(
	774	"-ERR Operation against a key holding the wrong kind of value\r\n"));
	775	shared.nokeyerr = createObject(REDIS_STRING,sdsnew(
	776	"-ERR no such key\r\n"));
	777	shared.syntaxerr = createObject(REDIS_STRING,sdsnew(
	778	"-ERR syntax error\r\n"));
	779	shared.sameobjecterr = createObject(REDIS_STRING,sdsnew(
	780	"-ERR source and destination objects are the same\r\n"));
	781	shared.outofrangeerr = createObject(REDIS_STRING,sdsnew(
	782	"-ERR index out of range\r\n"));
	783	shared.space = createObject(REDIS_STRING,sdsnew(" "));
	784	shared.colon = createObject(REDIS_STRING,sdsnew(":"));
	785	shared.plus = createObject(REDIS_STRING,sdsnew("+"));
	786	shared.select0 = createStringObject("select 0\r\n",10);
	787	shared.select1 = createStringObject("select 1\r\n",10);
	788	shared.select2 = createStringObject("select 2\r\n",10);
	789	shared.select3 = createStringObject("select 3\r\n",10);
	790	shared.select4 = createStringObject("select 4\r\n",10);
	791	shared.select5 = createStringObject("select 5\r\n",10);
	792	shared.select6 = createStringObject("select 6\r\n",10);
	793	shared.select7 = createStringObject("select 7\r\n",10);
	794	shared.select8 = createStringObject("select 8\r\n",10);
	795	shared.select9 = createStringObject("select 9\r\n",10);
	796	}
	797
	798	static void appendServerSaveParams(time_t seconds, int changes) {
	799	server.saveparams = zrealloc(server.saveparams,sizeof(struct saveparam)*(server.saveparamslen+1));
	800	if (server.saveparams == NULL) oom("appendServerSaveParams");
	801	server.saveparams[server.saveparamslen].seconds = seconds;
	802	server.saveparams[server.saveparamslen].changes = changes;
	803	server.saveparamslen++;
	804	}
	805
	806	static void ResetServerSaveParams() {
	807	zfree(server.saveparams);
	808	server.saveparams = NULL;
	809	server.saveparamslen = 0;
	810	}
	811
	812	static void initServerConfig() {
	813	server.dbnum = REDIS_DEFAULT_DBNUM;
	814	server.port = REDIS_SERVERPORT;
	815	server.verbosity = REDIS_DEBUG;
	816	server.maxidletime = REDIS_MAXIDLETIME;
	817	server.saveparams = NULL;
	818	server.logfile = NULL; /* NULL = log on standard output */
	819	server.bindaddr = NULL;
	820	server.glueoutputbuf = 1;
	821	server.daemonize = 0;
	822	server.pidfile = "/var/run/redis.pid";
	823	server.dbfilename = "dump.rdb";
	824	server.requirepass = NULL;
	825	server.shareobjects = 0;
	826	ResetServerSaveParams();
	827
	828	appendServerSaveParams(6060,1); / save after 1 hour and 1 change */
	829	appendServerSaveParams(300,100); /* save after 5 minutes and 100 changes */
	830	appendServerSaveParams(60,10000); /* save after 1 minute and 10000 changes */
	831	/* Replication related */
	832	server.isslave = 0;
	833	server.masterhost = NULL;
	834	server.masterport = 6379;
	835	server.master = NULL;
	836	server.replstate = REDIS_REPL_NONE;
	837	}
	838
	839	static void initServer() {
	840	int j;
	841
	842	signal(SIGHUP, SIG_IGN);
	843	signal(SIGPIPE, SIG_IGN);
	844
	845	server.clients = listCreate();
	846	server.slaves = listCreate();
	847	server.monitors = listCreate();
	848	server.objfreelist = listCreate();
	849	createSharedObjects();
	850	server.el = aeCreateEventLoop();
	851	server.db = zmalloc(sizeof(redisDb)*server.dbnum);
	852	server.sharingpool = dictCreate(&setDictType,NULL);
	853	server.sharingpoolsize = 1024;
	854	if (!server.db \|\| !server.clients \|\| !server.slaves \|\| !server.monitors \|\| !server.el \|\| !server.objfreelist)
	855	oom("server initialization"); /* Fatal OOM */
	856	server.fd = anetTcpServer(server.neterr, server.port, server.bindaddr);
	857	if (server.fd == -1) {
	858	redisLog(REDIS_WARNING, "Opening TCP port: %s", server.neterr);
	859	exit(1);
	860	}
	861	for (j = 0; j < server.dbnum; j++) {
	862	server.db[j].dict = dictCreate(&hashDictType,NULL);
	863	server.db[j].expires = dictCreate(&setDictType,NULL);
	864	server.db[j].id = j;
	865	}
	866	server.cronloops = 0;
	867	server.bgsaveinprogress = 0;
	868	server.lastsave = time(NULL);
	869	server.dirty = 0;
	870	server.usedmemory = 0;
	871	server.stat_numcommands = 0;
	872	server.stat_numconnections = 0;
	873	server.stat_starttime = time(NULL);
	874	aeCreateTimeEvent(server.el, 1000, serverCron, NULL, NULL);
	875	}
	876
	877	/* Empty the whole database */
	878	static void emptyDb() {
	879	int j;
	880
	881	for (j = 0; j < server.dbnum; j++) {
	882	dictEmpty(server.db[j].dict);
	883	dictEmpty(server.db[j].expires);
	884	}
	885	}
	886
	887	/* I agree, this is a very rudimental way to load a configuration...
	888	will improve later if the config gets more complex */
	889	static void loadServerConfig(char *filename) {
	890	FILE *fp = fopen(filename,"r");
	891	char buf[REDIS_CONFIGLINE_MAX+1], *err = NULL;
	892	int linenum = 0;
	893	sds line = NULL;
	894
	895	if (!fp) {
	896	redisLog(REDIS_WARNING,"Fatal error, can't open config file");
	897	exit(1);
	898	}
	899	while(fgets(buf,REDIS_CONFIGLINE_MAX+1,fp) != NULL) {
	900	sds *argv;
	901	int argc, j;
	902
	903	linenum++;
	904	line = sdsnew(buf);
	905	line = sdstrim(line," \t\r\n");
	906
	907	/* Skip comments and blank lines*/
	908	if (line[0] == '#' \|\| line[0] == '\0') {
	909	sdsfree(line);
	910	continue;
	911	}
	912
	913	/* Split into arguments */
	914	argv = sdssplitlen(line,sdslen(line)," ",1,&argc);
	915	sdstolower(argv[0]);
	916
	917	/* Execute config directives */
	918	if (!strcmp(argv[0],"timeout") && argc == 2) {
	919	server.maxidletime = atoi(argv[1]);
	920	if (server.maxidletime < 1) {
	921	err = "Invalid timeout value"; goto loaderr;
	922	}
	923	} else if (!strcmp(argv[0],"port") && argc == 2) {
	924	server.port = atoi(argv[1]);
	925	if (server.port < 1 \|\| server.port > 65535) {
	926	err = "Invalid port"; goto loaderr;
	927	}
	928	} else if (!strcmp(argv[0],"bind") && argc == 2) {
	929	server.bindaddr = zstrdup(argv[1]);
	930	} else if (!strcmp(argv[0],"save") && argc == 3) {
	931	int seconds = atoi(argv[1]);
	932	int changes = atoi(argv[2]);
	933	if (seconds < 1 \|\| changes < 0) {
	934	err = "Invalid save parameters"; goto loaderr;
	935	}
	936	appendServerSaveParams(seconds,changes);
	937	} else if (!strcmp(argv[0],"dir") && argc == 2) {
	938	if (chdir(argv[1]) == -1) {
	939	redisLog(REDIS_WARNING,"Can't chdir to '%s': %s",
	940	argv[1], strerror(errno));
	941	exit(1);
	942	}
	943	} else if (!strcmp(argv[0],"loglevel") && argc == 2) {
	944	if (!strcmp(argv[1],"debug")) server.verbosity = REDIS_DEBUG;
	945	else if (!strcmp(argv[1],"notice")) server.verbosity = REDIS_NOTICE;
	946	else if (!strcmp(argv[1],"warning")) server.verbosity = REDIS_WARNING;
	947	else {
	948	err = "Invalid log level. Must be one of debug, notice, warning";
	949	goto loaderr;
	950	}
	951	} else if (!strcmp(argv[0],"logfile") && argc == 2) {
	952	FILE *fp;
	953
	954	server.logfile = zstrdup(argv[1]);
	955	if (!strcmp(server.logfile,"stdout")) {
	956	zfree(server.logfile);
	957	server.logfile = NULL;
	958	}
	959	if (server.logfile) {
	960	/* Test if we are able to open the file. The server will not
	961	* be able to abort just for this problem later... */
	962	fp = fopen(server.logfile,"a");
	963	if (fp == NULL) {
	964	err = sdscatprintf(sdsempty(),
	965	"Can't open the log file: %s", strerror(errno));
	966	goto loaderr;
	967	}
	968	fclose(fp);
	969	}
	970	} else if (!strcmp(argv[0],"databases") && argc == 2) {
	971	server.dbnum = atoi(argv[1]);
	972	if (server.dbnum < 1) {
	973	err = "Invalid number of databases"; goto loaderr;
	974	}
	975	} else if (!strcmp(argv[0],"slaveof") && argc == 3) {
	976	server.masterhost = sdsnew(argv[1]);
	977	server.masterport = atoi(argv[2]);
	978	server.replstate = REDIS_REPL_CONNECT;
	979	} else if (!strcmp(argv[0],"glueoutputbuf") && argc == 2) {
	980	sdstolower(argv[1]);
	981	if (!strcmp(argv[1],"yes")) server.glueoutputbuf = 1;
	982	else if (!strcmp(argv[1],"no")) server.glueoutputbuf = 0;
	983	else {
	984	err = "argument must be 'yes' or 'no'"; goto loaderr;
	985	}
	986	} else if (!strcmp(argv[0],"shareobjects") && argc == 2) {
	987	sdstolower(argv[1]);
	988	if (!strcmp(argv[1],"yes")) server.shareobjects = 1;
	989	else if (!strcmp(argv[1],"no")) server.shareobjects = 0;
	990	else {
	991	err = "argument must be 'yes' or 'no'"; goto loaderr;
	992	}
	993	} else if (!strcmp(argv[0],"daemonize") && argc == 2) {
	994	sdstolower(argv[1]);
	995	if (!strcmp(argv[1],"yes")) server.daemonize = 1;
	996	else if (!strcmp(argv[1],"no")) server.daemonize = 0;
	997	else {
	998	err = "argument must be 'yes' or 'no'"; goto loaderr;
	999	}
	1000	} else if (!strcmp(argv[0],"requirepass") && argc == 2) {
	1001	server.requirepass = zstrdup(argv[1]);
	1002	} else if (!strcmp(argv[0],"pidfile") && argc == 2) {
	1003	server.pidfile = zstrdup(argv[1]);
	1004	} else {
	1005	err = "Bad directive or wrong number of arguments"; goto loaderr;
	1006	}
	1007	for (j = 0; j < argc; j++)
	1008	sdsfree(argv[j]);
	1009	zfree(argv);
	1010	sdsfree(line);
	1011	}
	1012	fclose(fp);
	1013	return;
	1014
	1015	loaderr:
	1016	fprintf(stderr, "\n* FATAL CONFIG FILE ERROR *\n");
	1017	fprintf(stderr, "Reading the configuration file, at line %d\n", linenum);
	1018	fprintf(stderr, ">>> '%s'\n", line);
	1019	fprintf(stderr, "%s\n", err);
	1020	exit(1);
	1021	}
	1022
	1023	static void freeClientArgv(redisClient *c) {
	1024	int j;
	1025
	1026	for (j = 0; j < c->argc; j++)
	1027	decrRefCount(c->argv[j]);
	1028	c->argc = 0;
	1029	}
	1030
	1031	static void freeClient(redisClient *c) {
	1032	listNode *ln;
	1033
	1034	aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
	1035	aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
	1036	sdsfree(c->querybuf);
	1037	listRelease(c->reply);
	1038	freeClientArgv(c);
	1039	close(c->fd);
	1040	ln = listSearchKey(server.clients,c);
	1041	assert(ln != NULL);
	1042	listDelNode(server.clients,ln);
	1043	if (c->flags & REDIS_SLAVE) {
	1044	list *l = (c->flags & REDIS_MONITOR) ? server.monitors : server.slaves;
	1045	ln = listSearchKey(l,c);
	1046	assert(ln != NULL);
	1047	listDelNode(l,ln);
	1048	}
	1049	if (c->flags & REDIS_MASTER) {
	1050	server.master = NULL;
	1051	server.replstate = REDIS_REPL_CONNECT;
	1052	}
	1053	zfree(c);
	1054	}
	1055
	1056	static void glueReplyBuffersIfNeeded(redisClient *c) {
	1057	int totlen = 0;
	1058	listNode ln = c->reply->head, next;
	1059	robj *o;
	1060
	1061	while(ln) {
	1062	o = ln->value;
	1063	totlen += sdslen(o->ptr);
	1064	ln = ln->next;
	1065	/* This optimization makes more sense if we don't have to copy
	1066	* too much data */
	1067	if (totlen > 1024) return;
	1068	}
	1069	if (totlen > 0) {
	1070	char buf[1024];
	1071	int copylen = 0;
	1072
	1073	ln = c->reply->head;
	1074	while(ln) {
	1075	next = ln->next;
	1076	o = ln->value;
	1077	memcpy(buf+copylen,o->ptr,sdslen(o->ptr));
	1078	copylen += sdslen(o->ptr);
	1079	listDelNode(c->reply,ln);
	1080	ln = next;
	1081	}
	1082	/* Now the output buffer is empty, add the new single element */
	1083	addReplySds(c,sdsnewlen(buf,totlen));
	1084	}
	1085	}
	1086
	1087	static void sendReplyToClient(aeEventLoop el, int fd, void privdata, int mask) {
	1088	redisClient *c = privdata;
	1089	int nwritten = 0, totwritten = 0, objlen;
	1090	robj *o;
	1091	REDIS_NOTUSED(el);
	1092	REDIS_NOTUSED(mask);
	1093
	1094	if (server.glueoutputbuf && listLength(c->reply) > 1)
	1095	glueReplyBuffersIfNeeded(c);
	1096	while(listLength(c->reply)) {
	1097	o = listNodeValue(listFirst(c->reply));
	1098	objlen = sdslen(o->ptr);
	1099
	1100	if (objlen == 0) {
	1101	listDelNode(c->reply,listFirst(c->reply));
	1102	continue;
	1103	}
	1104
	1105	if (c->flags & REDIS_MASTER) {
	1106	nwritten = objlen - c->sentlen;
	1107	} else {
	1108	nwritten = write(fd, ((char*)o->ptr)+c->sentlen, objlen - c->sentlen);
	1109	if (nwritten <= 0) break;
	1110	}
	1111	c->sentlen += nwritten;
	1112	totwritten += nwritten;
	1113	/* If we fully sent the object on head go to the next one */
	1114	if (c->sentlen == objlen) {
	1115	listDelNode(c->reply,listFirst(c->reply));
	1116	c->sentlen = 0;
	1117	}
	1118	}
	1119	if (nwritten == -1) {
	1120	if (errno == EAGAIN) {
	1121	nwritten = 0;
	1122	} else {
	1123	redisLog(REDIS_DEBUG,
	1124	"Error writing to client: %s", strerror(errno));
	1125	freeClient(c);
	1126	return;
	1127	}
	1128	}
	1129	if (totwritten > 0) c->lastinteraction = time(NULL);
	1130	if (listLength(c->reply) == 0) {
	1131	c->sentlen = 0;
	1132	aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
	1133	}
	1134	}
	1135
	1136	static struct redisCommand lookupCommand(char name) {
	1137	int j = 0;
	1138	while(cmdTable[j].name != NULL) {
	1139	if (!strcmp(name,cmdTable[j].name)) return &cmdTable[j];
	1140	j++;
	1141	}
	1142	return NULL;
	1143	}
	1144
	1145	/* resetClient prepare the client to process the next command */
	1146	static void resetClient(redisClient *c) {
	1147	freeClientArgv(c);
	1148	c->bulklen = -1;
	1149	}
	1150
	1151	/* If this function gets called we already read a whole
	1152	* command, argments are in the client argv/argc fields.
	1153	* processCommand() execute the command or prepare the
	1154	* server for a bulk read from the client.
	1155	*
	1156	* If 1 is returned the client is still alive and valid and
	1157	* and other operations can be performed by the caller. Otherwise
	1158	* if 0 is returned the client was destroied (i.e. after QUIT). */
	1159	static int processCommand(redisClient *c) {
	1160	struct redisCommand *cmd;
	1161	long long dirty;
	1162
	1163	sdstolower(c->argv[0]->ptr);
	1164	/* The QUIT command is handled as a special case. Normal command
	1165	* procs are unable to close the client connection safely */
	1166	if (!strcmp(c->argv[0]->ptr,"quit")) {
	1167	freeClient(c);
	1168	return 0;
	1169	}
	1170	cmd = lookupCommand(c->argv[0]->ptr);
	1171	if (!cmd) {
	1172	addReplySds(c,sdsnew("-ERR unknown command\r\n"));
	1173	resetClient(c);
	1174	return 1;
	1175	} else if ((cmd->arity > 0 && cmd->arity != c->argc) \|\|
	1176	(c->argc < -cmd->arity)) {
	1177	addReplySds(c,sdsnew("-ERR wrong number of arguments\r\n"));
	1178	resetClient(c);
	1179	return 1;
	1180	} else if (cmd->flags & REDIS_CMD_BULK && c->bulklen == -1) {
	1181	int bulklen = atoi(c->argv[c->argc-1]->ptr);
	1182
	1183	decrRefCount(c->argv[c->argc-1]);
	1184	if (bulklen < 0 \|\| bulklen > 102410241024) {
	1185	c->argc--;
	1186	addReplySds(c,sdsnew("-ERR invalid bulk write count\r\n"));
	1187	resetClient(c);
	1188	return 1;
	1189	}
	1190	c->argc--;
	1191	c->bulklen = bulklen+2; /* add two bytes for CR+LF */
	1192	/* It is possible that the bulk read is already in the
	1193	* buffer. Check this condition and handle it accordingly */
	1194	if ((signed)sdslen(c->querybuf) >= c->bulklen) {
	1195	c->argv[c->argc] = createStringObject(c->querybuf,c->bulklen-2);
	1196	c->argc++;
	1197	c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
	1198	} else {
	1199	return 1;
	1200	}
	1201	}
	1202	/* Let's try to share objects on the command arguments vector */
	1203	if (server.shareobjects) {
	1204	int j;
	1205	for(j = 1; j < c->argc; j++)
	1206	c->argv[j] = tryObjectSharing(c->argv[j]);
	1207	}
	1208	/* Check if the user is authenticated */
	1209	if (server.requirepass && !c->authenticated && cmd->proc != authCommand) {
	1210	addReplySds(c,sdsnew("-ERR operation not permitted\r\n"));
	1211	resetClient(c);
	1212	return 1;
	1213	}
	1214
	1215	/* Exec the command */
	1216	dirty = server.dirty;
	1217	cmd->proc(c);
	1218	if (server.dirty-dirty != 0 && listLength(server.slaves))
	1219	replicationFeedSlaves(server.slaves,cmd,c->db->id,c->argv,c->argc);
	1220	if (listLength(server.monitors))
	1221	replicationFeedSlaves(server.monitors,cmd,c->db->id,c->argv,c->argc);
	1222	server.stat_numcommands++;
	1223
	1224	/* Prepare the client for the next command */
	1225	if (c->flags & REDIS_CLOSE) {
	1226	freeClient(c);
	1227	return 0;
	1228	}
	1229	resetClient(c);
	1230	return 1;
	1231	}
	1232
	1233	static void replicationFeedSlaves(list slaves, struct redisCommand cmd, int dictid, robj **argv, int argc) {
	1234	listNode *ln = slaves->head;
	1235	robj outv[REDIS_MAX_ARGS4]; /* enough room for args, spaces, newlines */
	1236	int outc = 0, j;
	1237
	1238	for (j = 0; j < argc; j++) {
	1239	if (j != 0) outv[outc++] = shared.space;
	1240	if ((cmd->flags & REDIS_CMD_BULK) && j == argc-1) {
	1241	robj *lenobj;
	1242
	1243	lenobj = createObject(REDIS_STRING,
	1244	sdscatprintf(sdsempty(),"%d\r\n",sdslen(argv[j]->ptr)));
	1245	lenobj->refcount = 0;
	1246	outv[outc++] = lenobj;
	1247	}
	1248	outv[outc++] = argv[j];
	1249	}
	1250	outv[outc++] = shared.crlf;
	1251
	1252	/* Increment all the refcounts at start and decrement at end in order to
	1253	* be sure to free objects if there is no slave in a replication state
	1254	* able to be feed with commands */
	1255	for (j = 0; j < outc; j++) incrRefCount(outv[j]);
	1256	while(ln) {
	1257	redisClient *slave = ln->value;
	1258
	1259	/* Don't feed slaves that are still waiting for BGSAVE to start */
	1260	if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
	1261	ln = ln->next;
	1262	continue;
	1263	}
	1264
	1265	/* Feed all the other slaves, MONITORs and so on */
	1266	if (slave->slaveseldb != dictid) {
	1267	robj *selectcmd;
	1268
	1269	switch(dictid) {
	1270	case 0: selectcmd = shared.select0; break;
	1271	case 1: selectcmd = shared.select1; break;
	1272	case 2: selectcmd = shared.select2; break;
	1273	case 3: selectcmd = shared.select3; break;
	1274	case 4: selectcmd = shared.select4; break;
	1275	case 5: selectcmd = shared.select5; break;
	1276	case 6: selectcmd = shared.select6; break;
	1277	case 7: selectcmd = shared.select7; break;
	1278	case 8: selectcmd = shared.select8; break;
	1279	case 9: selectcmd = shared.select9; break;
	1280	default:
	1281	selectcmd = createObject(REDIS_STRING,
	1282	sdscatprintf(sdsempty(),"select %d\r\n",dictid));
	1283	selectcmd->refcount = 0;
	1284	break;
	1285	}
	1286	addReply(slave,selectcmd);
	1287	slave->slaveseldb = dictid;
	1288	}
	1289	for (j = 0; j < outc; j++) addReply(slave,outv[j]);
	1290	ln = ln->next;
	1291	}
	1292	for (j = 0; j < outc; j++) decrRefCount(outv[j]);
	1293	}
	1294
	1295	static void readQueryFromClient(aeEventLoop el, int fd, void privdata, int mask) {
	1296	redisClient c = (redisClient) privdata;
	1297	char buf[REDIS_QUERYBUF_LEN];
	1298	int nread;
	1299	REDIS_NOTUSED(el);
	1300	REDIS_NOTUSED(mask);
	1301
	1302	nread = read(fd, buf, REDIS_QUERYBUF_LEN);
	1303	if (nread == -1) {
	1304	if (errno == EAGAIN) {
	1305	nread = 0;
	1306	} else {
	1307	redisLog(REDIS_DEBUG, "Reading from client: %s",strerror(errno));
	1308	freeClient(c);
	1309	return;
	1310	}
	1311	} else if (nread == 0) {
	1312	redisLog(REDIS_DEBUG, "Client closed connection");
	1313	freeClient(c);
	1314	return;
	1315	}
	1316	if (nread) {
	1317	c->querybuf = sdscatlen(c->querybuf, buf, nread);
	1318	c->lastinteraction = time(NULL);
	1319	} else {
	1320	return;
	1321	}
	1322
	1323	again:
	1324	if (c->bulklen == -1) {
	1325	/* Read the first line of the query */
	1326	char *p = strchr(c->querybuf,'\n');
	1327	size_t querylen;
	1328	if (p) {
	1329	sds query, *argv;
	1330	int argc, j;
	1331
	1332	query = c->querybuf;
	1333	c->querybuf = sdsempty();
	1334	querylen = 1+(p-(query));
	1335	if (sdslen(query) > querylen) {
	1336	/* leave data after the first line of the query in the buffer */
	1337	c->querybuf = sdscatlen(c->querybuf,query+querylen,sdslen(query)-querylen);
	1338	}
	1339	p = '\0'; / remove "\n" */
	1340	if ((p-1) == '\r') (p-1) = '\0'; /* and "\r" if any */
	1341	sdsupdatelen(query);
	1342
	1343	/* Now we can split the query in arguments */
	1344	if (sdslen(query) == 0) {
	1345	/* Ignore empty query */
	1346	sdsfree(query);
	1347	return;
	1348	}
	1349	argv = sdssplitlen(query,sdslen(query)," ",1,&argc);
	1350	sdsfree(query);
	1351	if (argv == NULL) oom("sdssplitlen");
	1352	for (j = 0; j < argc && j < REDIS_MAX_ARGS; j++) {
	1353	if (sdslen(argv[j])) {
	1354	c->argv[c->argc] = createObject(REDIS_STRING,argv[j]);
	1355	c->argc++;
	1356	} else {
	1357	sdsfree(argv[j]);
	1358	}
	1359	}
	1360	zfree(argv);
	1361	/* Execute the command. If the client is still valid
	1362	* after processCommand() return and there is something
	1363	* on the query buffer try to process the next command. */
	1364	if (processCommand(c) && sdslen(c->querybuf)) goto again;
	1365	return;
	1366	} else if (sdslen(c->querybuf) >= 1024) {
	1367	redisLog(REDIS_DEBUG, "Client protocol error");
	1368	freeClient(c);
	1369	return;
	1370	}
	1371	} else {
	1372	/* Bulk read handling. Note that if we are at this point
	1373	the client already sent a command terminated with a newline,
	1374	we are reading the bulk data that is actually the last
	1375	argument of the command. */
	1376	int qbl = sdslen(c->querybuf);
	1377
	1378	if (c->bulklen <= qbl) {
	1379	/* Copy everything but the final CRLF as final argument */
	1380	c->argv[c->argc] = createStringObject(c->querybuf,c->bulklen-2);
	1381	c->argc++;
	1382	c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
	1383	processCommand(c);
	1384	return;
	1385	}
	1386	}
	1387	}
	1388
	1389	static int selectDb(redisClient *c, int id) {
	1390	if (id < 0 \|\| id >= server.dbnum)
	1391	return REDIS_ERR;
	1392	c->db = &server.db[id];
	1393	return REDIS_OK;
	1394	}
	1395
	1396	static void dupClientReplyValue(void o) {
	1397	incrRefCount((robj*)o);
	1398	return 0;
	1399	}
	1400
	1401	static redisClient *createClient(int fd) {
	1402	redisClient c = zmalloc(sizeof(c));
	1403
	1404	anetNonBlock(NULL,fd);
	1405	anetTcpNoDelay(NULL,fd);
	1406	if (!c) return NULL;
	1407	selectDb(c,0);
	1408	c->fd = fd;
	1409	c->querybuf = sdsempty();
	1410	c->argc = 0;
	1411	c->bulklen = -1;
	1412	c->sentlen = 0;
	1413	c->flags = 0;
	1414	c->lastinteraction = time(NULL);
	1415	c->authenticated = 0;
	1416	c->replstate = REDIS_REPL_NONE;
	1417	if ((c->reply = listCreate()) == NULL) oom("listCreate");
	1418	listSetFreeMethod(c->reply,decrRefCount);
	1419	listSetDupMethod(c->reply,dupClientReplyValue);
	1420	if (aeCreateFileEvent(server.el, c->fd, AE_READABLE,
	1421	readQueryFromClient, c, NULL) == AE_ERR) {
	1422	freeClient(c);
	1423	return NULL;
	1424	}
	1425	if (!listAddNodeTail(server.clients,c)) oom("listAddNodeTail");
	1426	return c;
	1427	}
	1428
	1429	static void addReply(redisClient c, robj obj) {
	1430	if (listLength(c->reply) == 0 &&
	1431	aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
	1432	sendReplyToClient, c, NULL) == AE_ERR) return;
	1433	if (!listAddNodeTail(c->reply,obj)) oom("listAddNodeTail");
	1434	incrRefCount(obj);
	1435	}
	1436
	1437	static void addReplySds(redisClient *c, sds s) {
	1438	robj *o = createObject(REDIS_STRING,s);
	1439	addReply(c,o);
	1440	decrRefCount(o);
	1441	}
	1442
	1443	static void acceptHandler(aeEventLoop el, int fd, void privdata, int mask) {
	1444	int cport, cfd;
	1445	char cip[128];
	1446	REDIS_NOTUSED(el);
	1447	REDIS_NOTUSED(mask);
	1448	REDIS_NOTUSED(privdata);
	1449
	1450	cfd = anetAccept(server.neterr, fd, cip, &cport);
	1451	if (cfd == AE_ERR) {
	1452	redisLog(REDIS_DEBUG,"Accepting client connection: %s", server.neterr);
	1453	return;
	1454	}
	1455	redisLog(REDIS_DEBUG,"Accepted %s:%d", cip, cport);
	1456	if (createClient(cfd) == NULL) {
	1457	redisLog(REDIS_WARNING,"Error allocating resoures for the client");
	1458	close(cfd); /* May be already closed, just ingore errors */
	1459	return;
	1460	}
	1461	server.stat_numconnections++;
	1462	}
	1463
	1464	/* ======================= Redis objects implementation ===================== */
	1465
	1466	static robj createObject(int type, void ptr) {
	1467	robj *o;
	1468
	1469	if (listLength(server.objfreelist)) {
	1470	listNode *head = listFirst(server.objfreelist);
	1471	o = listNodeValue(head);
	1472	listDelNode(server.objfreelist,head);
	1473	} else {
	1474	o = zmalloc(sizeof(*o));
	1475	}
	1476	if (!o) oom("createObject");
	1477	o->type = type;
	1478	o->ptr = ptr;
	1479	o->refcount = 1;
	1480	return o;
	1481	}
	1482
	1483	static robj createStringObject(char ptr, size_t len) {
	1484	return createObject(REDIS_STRING,sdsnewlen(ptr,len));
	1485	}
	1486
	1487	static robj *createListObject(void) {
	1488	list *l = listCreate();
	1489
	1490	if (!l) oom("listCreate");
	1491	listSetFreeMethod(l,decrRefCount);
	1492	return createObject(REDIS_LIST,l);
	1493	}
	1494
	1495	static robj *createSetObject(void) {
	1496	dict *d = dictCreate(&setDictType,NULL);
	1497	if (!d) oom("dictCreate");
	1498	return createObject(REDIS_SET,d);
	1499	}
	1500
	1501	static void freeStringObject(robj *o) {
	1502	sdsfree(o->ptr);
	1503	}
	1504
	1505	static void freeListObject(robj *o) {
	1506	listRelease((list*) o->ptr);
	1507	}
	1508
	1509	static void freeSetObject(robj *o) {
	1510	dictRelease((dict*) o->ptr);
	1511	}
	1512
	1513	static void freeHashObject(robj *o) {
	1514	dictRelease((dict*) o->ptr);
	1515	}
	1516
	1517	static void incrRefCount(robj *o) {
	1518	o->refcount++;
	1519	#ifdef DEBUG_REFCOUNT
	1520	if (o->type == REDIS_STRING)
	1521	printf("Increment '%s'(%p), now is: %d\n",o->ptr,o,o->refcount);
	1522	#endif
	1523	}
	1524
	1525	static void decrRefCount(void *obj) {
	1526	robj *o = obj;
	1527
	1528	#ifdef DEBUG_REFCOUNT
	1529	if (o->type == REDIS_STRING)
	1530	printf("Decrement '%s'(%p), now is: %d\n",o->ptr,o,o->refcount-1);
	1531	#endif
	1532	if (--(o->refcount) == 0) {
	1533	switch(o->type) {
	1534	case REDIS_STRING: freeStringObject(o); break;
	1535	case REDIS_LIST: freeListObject(o); break;
	1536	case REDIS_SET: freeSetObject(o); break;
	1537	case REDIS_HASH: freeHashObject(o); break;
	1538	default: assert(0 != 0); break;
	1539	}
	1540	if (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX \|\|
	1541	!listAddNodeHead(server.objfreelist,o))
	1542	zfree(o);
	1543	}
	1544	}
	1545
	1546	/* Try to share an object against the shared objects pool */
	1547	static robj tryObjectSharing(robj o) {
	1548	struct dictEntry *de;
	1549	unsigned long c;
	1550
	1551	if (o == NULL \|\| server.shareobjects == 0) return o;
	1552
	1553	assert(o->type == REDIS_STRING);
	1554	de = dictFind(server.sharingpool,o);
	1555	if (de) {
	1556	robj *shared = dictGetEntryKey(de);
	1557
	1558	c = ((unsigned long) dictGetEntryVal(de))+1;
	1559	dictGetEntryVal(de) = (void*) c;
	1560	incrRefCount(shared);
	1561	decrRefCount(o);
	1562	return shared;
	1563	} else {
	1564	/* Here we are using a stream algorihtm: Every time an object is
	1565	* shared we increment its count, everytime there is a miss we
	1566	* recrement the counter of a random object. If this object reaches
	1567	* zero we remove the object and put the current object instead. */
	1568	if (dictSize(server.sharingpool) >=
	1569	server.sharingpoolsize) {
	1570	de = dictGetRandomKey(server.sharingpool);
	1571	assert(de != NULL);
	1572	c = ((unsigned long) dictGetEntryVal(de))-1;
	1573	dictGetEntryVal(de) = (void*) c;
	1574	if (c == 0) {
	1575	dictDelete(server.sharingpool,de->key);
	1576	}
	1577	} else {
	1578	c = 0; /* If the pool is empty we want to add this object */
	1579	}
	1580	if (c == 0) {
	1581	int retval;
	1582
	1583	retval = dictAdd(server.sharingpool,o,(void*)1);
	1584	assert(retval == DICT_OK);
	1585	incrRefCount(o);
	1586	}
	1587	return o;
	1588	}
	1589	}
	1590
	1591	static robj lookupKey(redisDb db, robj *key) {
	1592	dictEntry *de = dictFind(db->dict,key);
	1593	return de ? dictGetEntryVal(de) : NULL;
	1594	}
	1595
	1596	static robj lookupKeyRead(redisDb db, robj *key) {
	1597	expireIfNeeded(db,key);
	1598	return lookupKey(db,key);
	1599	}
	1600
	1601	static robj lookupKeyWrite(redisDb db, robj *key) {
	1602	deleteIfVolatile(db,key);
	1603	return lookupKey(db,key);
	1604	}
	1605
	1606	static int deleteKey(redisDb db, robj key) {
	1607	int retval;
	1608
	1609	/* We need to protect key from destruction: after the first dictDelete()
	1610	* it may happen that 'key' is no longer valid if we don't increment
	1611	* it's count. This may happen when we get the object reference directly
	1612	* from the hash table with dictRandomKey() or dict iterators */
	1613	incrRefCount(key);
	1614	if (dictSize(db->expires)) dictDelete(db->expires,key);
	1615	retval = dictDelete(db->dict,key);
	1616	decrRefCount(key);
	1617
	1618	return retval == DICT_OK;
	1619	}
	1620
	1621	/============================ DB saving/loading ============================ /
	1622
	1623	static int rdbSaveType(FILE *fp, unsigned char type) {
	1624	if (fwrite(&type,1,1,fp) == 0) return -1;
	1625	return 0;
	1626	}
	1627
	1628	static int rdbSaveTime(FILE *fp, time_t t) {
	1629	int32_t t32 = (int32_t) t;
	1630	if (fwrite(&t32,4,1,fp) == 0) return -1;
	1631	return 0;
	1632	}
	1633
	1634	/* check rdbLoadLen() comments for more info */
	1635	static int rdbSaveLen(FILE *fp, uint32_t len) {
	1636	unsigned char buf[2];
	1637
	1638	if (len < (1<<6)) {
	1639	/* Save a 6 bit len */
	1640	buf[0] = (len&0xFF)\|(REDIS_RDB_6BITLEN<<6);
	1641	if (fwrite(buf,1,1,fp) == 0) return -1;
	1642	} else if (len < (1<<14)) {
	1643	/* Save a 14 bit len */
	1644	buf[0] = ((len>>8)&0xFF)\|(REDIS_RDB_14BITLEN<<6);
	1645	buf[1] = len&0xFF;
	1646	if (fwrite(buf,2,1,fp) == 0) return -1;
	1647	} else {
	1648	/* Save a 32 bit len */
	1649	buf[0] = (REDIS_RDB_32BITLEN<<6);
	1650	if (fwrite(buf,1,1,fp) == 0) return -1;
	1651	len = htonl(len);
	1652	if (fwrite(&len,4,1,fp) == 0) return -1;
	1653	}
	1654	return 0;
	1655	}
	1656
	1657	/* String objects in the form "2391" "-100" without any space and with a
	1658	* range of values that can fit in an 8, 16 or 32 bit signed value can be
	1659	* encoded as integers to save space */
	1660	int rdbTryIntegerEncoding(sds s, unsigned char *enc) {
	1661	long long value;
	1662	char *endptr, buf[32];
	1663
	1664	/* Check if it's possible to encode this value as a number */
	1665	value = strtoll(s, &endptr, 10);
	1666	if (endptr[0] != '\0') return 0;
	1667	snprintf(buf,32,"%lld",value);
	1668
	1669	/* If the number converted back into a string is not identical
	1670	* then it's not possible to encode the string as integer */
	1671	if (strlen(buf) != sdslen(s) \|\| memcmp(buf,s,sdslen(s))) return 0;
	1672
	1673	/* Finally check if it fits in our ranges */
	1674	if (value >= -(1<<7) && value <= (1<<7)-1) {
	1675	enc[0] = (REDIS_RDB_ENCVAL<<6)\|REDIS_RDB_ENC_INT8;
	1676	enc[1] = value&0xFF;
	1677	return 2;
	1678	} else if (value >= -(1<<15) && value <= (1<<15)-1) {
	1679	enc[0] = (REDIS_RDB_ENCVAL<<6)\|REDIS_RDB_ENC_INT16;
	1680	enc[1] = value&0xFF;
	1681	enc[2] = (value>>8)&0xFF;
	1682	return 3;
	1683	} else if (value >= -((long long)1<<31) && value <= ((long long)1<<31)-1) {
	1684	enc[0] = (REDIS_RDB_ENCVAL<<6)\|REDIS_RDB_ENC_INT32;
	1685	enc[1] = value&0xFF;
	1686	enc[2] = (value>>8)&0xFF;
	1687	enc[3] = (value>>16)&0xFF;
	1688	enc[4] = (value>>24)&0xFF;
	1689	return 5;
	1690	} else {
	1691	return 0;
	1692	}
	1693	}
	1694
	1695	static int rdbSaveLzfStringObject(FILE fp, robj obj) {
	1696	unsigned int comprlen, outlen;
	1697	unsigned char byte;
	1698	void *out;
	1699
	1700	/* We require at least four bytes compression for this to be worth it */
	1701	outlen = sdslen(obj->ptr)-4;
	1702	if (outlen <= 0) return 0;
	1703	if ((out = zmalloc(outlen)) == NULL) return 0;
	1704	comprlen = lzf_compress(obj->ptr, sdslen(obj->ptr), out, outlen);
	1705	if (comprlen == 0) {
	1706	zfree(out);
	1707	return 0;
	1708	}
	1709	/* Data compressed! Let's save it on disk */
	1710	byte = (REDIS_RDB_ENCVAL<<6)\|REDIS_RDB_ENC_LZF;
	1711	if (fwrite(&byte,1,1,fp) == 0) goto writeerr;
	1712	if (rdbSaveLen(fp,comprlen) == -1) goto writeerr;
	1713	if (rdbSaveLen(fp,sdslen(obj->ptr)) == -1) goto writeerr;
	1714	if (fwrite(out,comprlen,1,fp) == 0) goto writeerr;
	1715	zfree(out);
	1716	return comprlen;
	1717
	1718	writeerr:
	1719	zfree(out);
	1720	return -1;
	1721	}
	1722
	1723	/* Save a string objet as [len][data] on disk. If the object is a string
	1724	* representation of an integer value we try to safe it in a special form */
	1725	static int rdbSaveStringObject(FILE fp, robj obj) {
	1726	size_t len = sdslen(obj->ptr);
	1727	int enclen;
	1728
	1729	/* Try integer encoding */
	1730	if (len <= 11) {
	1731	unsigned char buf[5];
	1732	if ((enclen = rdbTryIntegerEncoding(obj->ptr,buf)) > 0) {
	1733	if (fwrite(buf,enclen,1,fp) == 0) return -1;
	1734	return 0;
	1735	}
	1736	}
	1737
	1738	/* Try LZF compression - under 20 bytes it's unable to compress even
	1739	* aaaaaaaaaaaaaaaaaa so skip it */
	1740	if (len > 20) {
	1741	int retval;
	1742
	1743	retval = rdbSaveLzfStringObject(fp,obj);
	1744	if (retval == -1) return -1;
	1745	if (retval > 0) return 0;
	1746	/* retval == 0 means data can't be compressed, save the old way */
	1747	}
	1748
	1749	/* Store verbatim */
	1750	if (rdbSaveLen(fp,len) == -1) return -1;
	1751	if (len && fwrite(obj->ptr,len,1,fp) == 0) return -1;
	1752	return 0;
	1753	}
	1754
	1755	/* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success */
	1756	static int rdbSave(char *filename) {
	1757	dictIterator *di = NULL;
	1758	dictEntry *de;
	1759	FILE *fp;
	1760	char tmpfile[256];
	1761	int j;
	1762	time_t now = time(NULL);
	1763
	1764	snprintf(tmpfile,256,"temp-%d.%ld.rdb",(int)time(NULL),(long int)random());
	1765	fp = fopen(tmpfile,"w");
	1766	if (!fp) {
	1767	redisLog(REDIS_WARNING, "Failed saving the DB: %s", strerror(errno));
	1768	return REDIS_ERR;
	1769	}
	1770	if (fwrite("REDIS0001",9,1,fp) == 0) goto werr;
	1771	for (j = 0; j < server.dbnum; j++) {
	1772	redisDb *db = server.db+j;
	1773	dict *d = db->dict;
	1774	if (dictSize(d) == 0) continue;
	1775	di = dictGetIterator(d);
	1776	if (!di) {
	1777	fclose(fp);
	1778	return REDIS_ERR;
	1779	}
	1780
	1781	/* Write the SELECT DB opcode */
	1782	if (rdbSaveType(fp,REDIS_SELECTDB) == -1) goto werr;
	1783	if (rdbSaveLen(fp,j) == -1) goto werr;
	1784
	1785	/* Iterate this DB writing every entry */
	1786	while((de = dictNext(di)) != NULL) {
	1787	robj *key = dictGetEntryKey(de);
	1788	robj *o = dictGetEntryVal(de);
	1789	time_t expiretime = getExpire(db,key);
	1790
	1791	/* Save the expire time */
	1792	if (expiretime != -1) {
	1793	/* If this key is already expired skip it */
	1794	if (expiretime < now) continue;
	1795	if (rdbSaveType(fp,REDIS_EXPIRETIME) == -1) goto werr;
	1796	if (rdbSaveTime(fp,expiretime) == -1) goto werr;
	1797	}
	1798	/* Save the key and associated value */
	1799	if (rdbSaveType(fp,o->type) == -1) goto werr;
	1800	if (rdbSaveStringObject(fp,key) == -1) goto werr;
	1801	if (o->type == REDIS_STRING) {
	1802	/* Save a string value */
	1803	if (rdbSaveStringObject(fp,o) == -1) goto werr;
	1804	} else if (o->type == REDIS_LIST) {
	1805	/* Save a list value */
	1806	list *list = o->ptr;
	1807	listNode *ln = list->head;
	1808
	1809	if (rdbSaveLen(fp,listLength(list)) == -1) goto werr;
	1810	while(ln) {
	1811	robj *eleobj = listNodeValue(ln);
	1812
	1813	if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
	1814	ln = ln->next;
	1815	}
	1816	} else if (o->type == REDIS_SET) {
	1817	/* Save a set value */
	1818	dict *set = o->ptr;
	1819	dictIterator *di = dictGetIterator(set);
	1820	dictEntry *de;
	1821
	1822	if (!set) oom("dictGetIteraotr");
	1823	if (rdbSaveLen(fp,dictSize(set)) == -1) goto werr;
	1824	while((de = dictNext(di)) != NULL) {
	1825	robj *eleobj = dictGetEntryKey(de);
	1826
	1827	if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
	1828	}
	1829	dictReleaseIterator(di);
	1830	} else {
	1831	assert(0 != 0);
	1832	}
	1833	}
	1834	dictReleaseIterator(di);
	1835	}
	1836	/* EOF opcode */
	1837	if (rdbSaveType(fp,REDIS_EOF) == -1) goto werr;
	1838
	1839	/* Make sure data will not remain on the OS's output buffers */
	1840	fflush(fp);
	1841	fsync(fileno(fp));
	1842	fclose(fp);
	1843
	1844	/* Use RENAME to make sure the DB file is changed atomically only
	1845	* if the generate DB file is ok. */
	1846	if (rename(tmpfile,filename) == -1) {
	1847	redisLog(REDIS_WARNING,"Error moving temp DB file on the final destionation: %s", strerror(errno));
	1848	unlink(tmpfile);
	1849	return REDIS_ERR;
	1850	}
	1851	redisLog(REDIS_NOTICE,"DB saved on disk");
	1852	server.dirty = 0;
	1853	server.lastsave = time(NULL);
	1854	return REDIS_OK;
	1855
	1856	werr:
	1857	fclose(fp);
	1858	unlink(tmpfile);
	1859	redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
	1860	if (di) dictReleaseIterator(di);
	1861	return REDIS_ERR;
	1862	}
	1863
	1864	static int rdbSaveBackground(char *filename) {
	1865	pid_t childpid;
	1866
	1867	if (server.bgsaveinprogress) return REDIS_ERR;
	1868	if ((childpid = fork()) == 0) {
	1869	/* Child */
	1870	close(server.fd);
	1871	if (rdbSave(filename) == REDIS_OK) {
	1872	exit(0);
	1873	} else {
	1874	exit(1);
	1875	}
	1876	} else {
	1877	/* Parent */
	1878	redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
	1879	server.bgsaveinprogress = 1;
	1880	return REDIS_OK;
	1881	}
	1882	return REDIS_OK; /* unreached */
	1883	}
	1884
	1885	static int rdbLoadType(FILE *fp) {
	1886	unsigned char type;
	1887	if (fread(&type,1,1,fp) == 0) return -1;
	1888	return type;
	1889	}
	1890
	1891	static time_t rdbLoadTime(FILE *fp) {
	1892	int32_t t32;
	1893	if (fread(&t32,4,1,fp) == 0) return -1;
	1894	return (time_t) t32;
	1895	}
	1896
	1897	/* Load an encoded length from the DB, see the REDIS_RDB_* defines on the top
	1898	* of this file for a description of how this are stored on disk.
	1899	*
	1900	* isencoded is set to 1 if the readed length is not actually a length but
	1901	* an "encoding type", check the above comments for more info */
	1902	static uint32_t rdbLoadLen(FILE fp, int rdbver, int isencoded) {
	1903	unsigned char buf[2];
	1904	uint32_t len;
	1905
	1906	if (isencoded) *isencoded = 0;
	1907	if (rdbver == 0) {
	1908	if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
	1909	return ntohl(len);
	1910	} else {
	1911	int type;
	1912
	1913	if (fread(buf,1,1,fp) == 0) return REDIS_RDB_LENERR;
	1914	type = (buf[0]&0xC0)>>6;
	1915	if (type == REDIS_RDB_6BITLEN) {
	1916	/* Read a 6 bit len */
	1917	return buf[0]&0x3F;
	1918	} else if (type == REDIS_RDB_ENCVAL) {
	1919	/* Read a 6 bit len encoding type */
	1920	if (isencoded) *isencoded = 1;
	1921	return buf[0]&0x3F;
	1922	} else if (type == REDIS_RDB_14BITLEN) {
	1923	/* Read a 14 bit len */
	1924	if (fread(buf+1,1,1,fp) == 0) return REDIS_RDB_LENERR;
	1925	return ((buf[0]&0x3F)<<8)\|buf[1];
	1926	} else {
	1927	/* Read a 32 bit len */
	1928	if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
	1929	return ntohl(len);
	1930	}
	1931	}
	1932	}
	1933
	1934	static robj rdbLoadIntegerObject(FILE fp, int enctype) {
	1935	unsigned char enc[4];
	1936	long long val;
	1937
	1938	if (enctype == REDIS_RDB_ENC_INT8) {
	1939	if (fread(enc,1,1,fp) == 0) return NULL;
	1940	val = (signed char)enc[0];
	1941	} else if (enctype == REDIS_RDB_ENC_INT16) {
	1942	uint16_t v;
	1943	if (fread(enc,2,1,fp) == 0) return NULL;
	1944	v = enc[0]\|(enc[1]<<8);
	1945	val = (int16_t)v;
	1946	} else if (enctype == REDIS_RDB_ENC_INT32) {
	1947	uint32_t v;
	1948	if (fread(enc,4,1,fp) == 0) return NULL;
	1949	v = enc[0]\|(enc[1]<<8)\|(enc[2]<<16)\|(enc[3]<<24);
	1950	val = (int32_t)v;
	1951	} else {
	1952	val = 0; /* anti-warning */
	1953	assert(0!=0);
	1954	}
	1955	return createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",val));
	1956	}
	1957
	1958	static robj rdbLoadLzfStringObject(FILEfp, int rdbver) {
	1959	unsigned int len, clen;
	1960	unsigned char *c = NULL;
	1961	sds val = NULL;
	1962
	1963	if ((clen = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR) return NULL;
	1964	if ((len = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR) return NULL;
	1965	if ((c = zmalloc(clen)) == NULL) goto err;
	1966	if ((val = sdsnewlen(NULL,len)) == NULL) goto err;
	1967	if (fread(c,clen,1,fp) == 0) goto err;
	1968	if (lzf_decompress(c,clen,val,len) == 0) goto err;
	1969	return createObject(REDIS_STRING,val);
	1970	err:
	1971	zfree(c);
	1972	sdsfree(val);
	1973	return NULL;
	1974	}
	1975
	1976	static robj rdbLoadStringObject(FILEfp, int rdbver) {
	1977	int isencoded;
	1978	uint32_t len;
	1979	sds val;
	1980
	1981	len = rdbLoadLen(fp,rdbver,&isencoded);
	1982	if (isencoded) {
	1983	switch(len) {
	1984	case REDIS_RDB_ENC_INT8:
	1985	case REDIS_RDB_ENC_INT16:
	1986	case REDIS_RDB_ENC_INT32:
	1987	return tryObjectSharing(rdbLoadIntegerObject(fp,len));
	1988	case REDIS_RDB_ENC_LZF:
	1989	return tryObjectSharing(rdbLoadLzfStringObject(fp,rdbver));
	1990	default:
	1991	assert(0!=0);
	1992	}
	1993	}
	1994
	1995	if (len == REDIS_RDB_LENERR) return NULL;
	1996	val = sdsnewlen(NULL,len);
	1997	if (len && fread(val,len,1,fp) == 0) {
	1998	sdsfree(val);
	1999	return NULL;
	2000	}
	2001	return tryObjectSharing(createObject(REDIS_STRING,val));
	2002	}
	2003
	2004	static int rdbLoad(char *filename) {
	2005	FILE *fp;
	2006	robj *keyobj = NULL;
	2007	uint32_t dbid;
	2008	int type, retval, rdbver;
	2009	dict *d = server.db[0].dict;
	2010	redisDb *db = server.db+0;
	2011	char buf[1024];
	2012	time_t expiretime = -1, now = time(NULL);
	2013
	2014	fp = fopen(filename,"r");
	2015	if (!fp) return REDIS_ERR;
	2016	if (fread(buf,9,1,fp) == 0) goto eoferr;
	2017	buf[9] = '\0';
	2018	if (memcmp(buf,"REDIS",5) != 0) {
	2019	fclose(fp);
	2020	redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
	2021	return REDIS_ERR;
	2022	}
	2023	rdbver = atoi(buf+5);
	2024	if (rdbver > 1) {
	2025	fclose(fp);
	2026	redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
	2027	return REDIS_ERR;
	2028	}
	2029	while(1) {
	2030	robj *o;
	2031
	2032	/* Read type. */
	2033	if ((type = rdbLoadType(fp)) == -1) goto eoferr;
	2034	if (type == REDIS_EXPIRETIME) {
	2035	if ((expiretime = rdbLoadTime(fp)) == -1) goto eoferr;
	2036	/* We read the time so we need to read the object type again */
	2037	if ((type = rdbLoadType(fp)) == -1) goto eoferr;
	2038	}
	2039	if (type == REDIS_EOF) break;
	2040	/* Handle SELECT DB opcode as a special case */
	2041	if (type == REDIS_SELECTDB) {
	2042	if ((dbid = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR)
	2043	goto eoferr;
	2044	if (dbid >= (unsigned)server.dbnum) {
	2045	redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
	2046	exit(1);
	2047	}
	2048	db = server.db+dbid;
	2049	d = db->dict;
	2050	continue;
	2051	}
	2052	/* Read key */
	2053	if ((keyobj = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
	2054
	2055	if (type == REDIS_STRING) {
	2056	/* Read string value */
	2057	if ((o = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
	2058	} else if (type == REDIS_LIST \|\| type == REDIS_SET) {
	2059	/* Read list/set value */
	2060	uint32_t listlen;
	2061
	2062	if ((listlen = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR)
	2063	goto eoferr;
	2064	o = (type == REDIS_LIST) ? createListObject() : createSetObject();
	2065	/* Load every single element of the list/set */
	2066	while(listlen--) {
	2067	robj *ele;
	2068
	2069	if ((ele = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
	2070	if (type == REDIS_LIST) {
	2071	if (!listAddNodeTail((list*)o->ptr,ele))
	2072	oom("listAddNodeTail");
	2073	} else {
	2074	if (dictAdd((dict*)o->ptr,ele,NULL) == DICT_ERR)
	2075	oom("dictAdd");
	2076	}
	2077	}
	2078	} else {
	2079	assert(0 != 0);
	2080	}
	2081	/* Add the new object in the hash table */
	2082	retval = dictAdd(d,keyobj,o);
	2083	if (retval == DICT_ERR) {
	2084	redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", keyobj->ptr);
	2085	exit(1);
	2086	}
	2087	/* Set the expire time if needed */
	2088	if (expiretime != -1) {
	2089	setExpire(db,keyobj,expiretime);
	2090	/* Delete this key if already expired */
	2091	if (expiretime < now) deleteKey(db,keyobj);
	2092	expiretime = -1;
	2093	}
	2094	keyobj = o = NULL;
	2095	}
	2096	fclose(fp);
	2097	return REDIS_OK;
	2098
	2099	eoferr: /* unexpected end of file is handled here with a fatal exit */
	2100	if (keyobj) decrRefCount(keyobj);
	2101	redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, exiting now.");
	2102	exit(1);
	2103	return REDIS_ERR; /* Just to avoid warning */
	2104	}
	2105
	2106	/================================== Commands =============================== /
	2107
	2108	static void authCommand(redisClient *c) {
	2109	if (!server.requirepass \|\| !strcmp(c->argv[1]->ptr, server.requirepass)) {
	2110	c->authenticated = 1;
	2111	addReply(c,shared.ok);
	2112	} else {
	2113	c->authenticated = 0;
	2114	addReply(c,shared.err);
	2115	}
	2116	}
	2117
	2118	static void pingCommand(redisClient *c) {
	2119	addReply(c,shared.pong);
	2120	}
	2121
	2122	static void echoCommand(redisClient *c) {
	2123	addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
	2124	(int)sdslen(c->argv[1]->ptr)));
	2125	addReply(c,c->argv[1]);
	2126	addReply(c,shared.crlf);
	2127	}
	2128
	2129	/=================================== Strings =============================== /
	2130
	2131	static void setGenericCommand(redisClient *c, int nx) {
	2132	int retval;
	2133
	2134	retval = dictAdd(c->db->dict,c->argv[1],c->argv[2]);
	2135	if (retval == DICT_ERR) {
	2136	if (!nx) {
	2137	dictReplace(c->db->dict,c->argv[1],c->argv[2]);
	2138	incrRefCount(c->argv[2]);
	2139	} else {
	2140	addReply(c,shared.czero);
	2141	return;
	2142	}
	2143	} else {
	2144	incrRefCount(c->argv[1]);
	2145	incrRefCount(c->argv[2]);
	2146	}
	2147	server.dirty++;
	2148	removeExpire(c->db,c->argv[1]);
	2149	addReply(c, nx ? shared.cone : shared.ok);
	2150	}
	2151
	2152	static void setCommand(redisClient *c) {
	2153	setGenericCommand(c,0);
	2154	}
	2155
	2156	static void setnxCommand(redisClient *c) {
	2157	setGenericCommand(c,1);
	2158	}
	2159
	2160	static void getCommand(redisClient *c) {
	2161	robj *o = lookupKeyRead(c->db,c->argv[1]);
	2162
	2163	if (o == NULL) {
	2164	addReply(c,shared.nullbulk);
	2165	} else {
	2166	if (o->type != REDIS_STRING) {
	2167	addReply(c,shared.wrongtypeerr);
	2168	} else {
	2169	addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(o->ptr)));
	2170	addReply(c,o);
	2171	addReply(c,shared.crlf);
	2172	}
	2173	}
	2174	}
	2175
	2176	static void mgetCommand(redisClient *c) {
	2177	int j;
	2178
	2179	addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",c->argc-1));
	2180	for (j = 1; j < c->argc; j++) {
	2181	robj *o = lookupKeyRead(c->db,c->argv[j]);
	2182	if (o == NULL) {
	2183	addReply(c,shared.nullbulk);
	2184	} else {
	2185	if (o->type != REDIS_STRING) {
	2186	addReply(c,shared.nullbulk);
	2187	} else {
	2188	addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(o->ptr)));
	2189	addReply(c,o);
	2190	addReply(c,shared.crlf);
	2191	}
	2192	}
	2193	}
	2194	}
	2195
	2196	static void incrDecrCommand(redisClient *c, int incr) {
	2197	long long value;
	2198	int retval;
	2199	robj *o;
	2200
	2201	o = lookupKeyWrite(c->db,c->argv[1]);
	2202	if (o == NULL) {
	2203	value = 0;
	2204	} else {
	2205	if (o->type != REDIS_STRING) {
	2206	value = 0;
	2207	} else {
	2208	char *eptr;
	2209
	2210	value = strtoll(o->ptr, &eptr, 10);
	2211	}
	2212	}
	2213
	2214	value += incr;
	2215	o = createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",value));
	2216	retval = dictAdd(c->db->dict,c->argv[1],o);
	2217	if (retval == DICT_ERR) {
	2218	dictReplace(c->db->dict,c->argv[1],o);
	2219	removeExpire(c->db,c->argv[1]);
	2220	} else {
	2221	incrRefCount(c->argv[1]);
	2222	}
	2223	server.dirty++;
	2224	addReply(c,shared.colon);
	2225	addReply(c,o);
	2226	addReply(c,shared.crlf);
	2227	}
	2228
	2229	static void incrCommand(redisClient *c) {
	2230	incrDecrCommand(c,1);
	2231	}
	2232
	2233	static void decrCommand(redisClient *c) {
	2234	incrDecrCommand(c,-1);
	2235	}
	2236
	2237	static void incrbyCommand(redisClient *c) {
	2238	int incr = atoi(c->argv[2]->ptr);
	2239	incrDecrCommand(c,incr);
	2240	}
	2241
	2242	static void decrbyCommand(redisClient *c) {
	2243	int incr = atoi(c->argv[2]->ptr);
	2244	incrDecrCommand(c,-incr);
	2245	}
	2246
	2247	/* ========================= Type agnostic commands ========================= */
	2248
	2249	static void delCommand(redisClient *c) {
	2250	if (deleteKey(c->db,c->argv[1])) {
	2251	server.dirty++;
	2252	addReply(c,shared.cone);
	2253	} else {
	2254	addReply(c,shared.czero);
	2255	}
	2256	}
	2257
	2258	static void existsCommand(redisClient *c) {
	2259	addReply(c,lookupKeyRead(c->db,c->argv[1]) ? shared.cone : shared.czero);
	2260	}
	2261
	2262	static void selectCommand(redisClient *c) {
	2263	int id = atoi(c->argv[1]->ptr);
	2264
	2265	if (selectDb(c,id) == REDIS_ERR) {
	2266	addReplySds(c,sdsnew("-ERR invalid DB index\r\n"));
	2267	} else {
	2268	addReply(c,shared.ok);
	2269	}
	2270	}
	2271
	2272	static void randomkeyCommand(redisClient *c) {
	2273	dictEntry *de;
	2274
	2275	while(1) {
	2276	de = dictGetRandomKey(c->db->dict);
	2277	if (!de \|\| expireIfNeeded(c->db,dictGetEntryKey(de)) == 0) break;
	2278	}
	2279	if (de == NULL) {
	2280	addReply(c,shared.plus);
	2281	addReply(c,shared.crlf);
	2282	} else {
	2283	addReply(c,shared.plus);
	2284	addReply(c,dictGetEntryKey(de));
	2285	addReply(c,shared.crlf);
	2286	}
	2287	}
	2288
	2289	static void keysCommand(redisClient *c) {
	2290	dictIterator *di;
	2291	dictEntry *de;
	2292	sds pattern = c->argv[1]->ptr;
	2293	int plen = sdslen(pattern);
	2294	int numkeys = 0, keyslen = 0;
	2295	robj *lenobj = createObject(REDIS_STRING,NULL);
	2296
	2297	di = dictGetIterator(c->db->dict);
	2298	if (!di) oom("dictGetIterator");
	2299	addReply(c,lenobj);
	2300	decrRefCount(lenobj);
	2301	while((de = dictNext(di)) != NULL) {
	2302	robj *keyobj = dictGetEntryKey(de);
	2303
	2304	sds key = keyobj->ptr;
	2305	if ((pattern[0] == '*' && pattern[1] == '\0') \|\|
	2306	stringmatchlen(pattern,plen,key,sdslen(key),0)) {
	2307	if (expireIfNeeded(c->db,keyobj) == 0) {
	2308	if (numkeys != 0)
	2309	addReply(c,shared.space);
	2310	addReply(c,keyobj);
	2311	numkeys++;
	2312	keyslen += sdslen(key);
	2313	}
	2314	}
	2315	}
	2316	dictReleaseIterator(di);
	2317	lenobj->ptr = sdscatprintf(sdsempty(),"$%lu\r\n",keyslen+(numkeys ? (numkeys-1) : 0));
	2318	addReply(c,shared.crlf);
	2319	}
	2320
	2321	static void dbsizeCommand(redisClient *c) {
	2322	addReplySds(c,
	2323	sdscatprintf(sdsempty(),":%lu\r\n",dictSize(c->db->dict)));
	2324	}
	2325
	2326	static void lastsaveCommand(redisClient *c) {
	2327	addReplySds(c,
	2328	sdscatprintf(sdsempty(),":%lu\r\n",server.lastsave));
	2329	}
	2330
	2331	static void typeCommand(redisClient *c) {
	2332	robj *o;
	2333	char *type;
	2334
	2335	o = lookupKeyRead(c->db,c->argv[1]);
	2336	if (o == NULL) {
	2337	type = "+none";
	2338	} else {
	2339	switch(o->type) {
	2340	case REDIS_STRING: type = "+string"; break;
	2341	case REDIS_LIST: type = "+list"; break;
	2342	case REDIS_SET: type = "+set"; break;
	2343	default: type = "unknown"; break;
	2344	}
	2345	}
	2346	addReplySds(c,sdsnew(type));
	2347	addReply(c,shared.crlf);
	2348	}
	2349
	2350	static void saveCommand(redisClient *c) {
	2351	if (server.bgsaveinprogress) {
	2352	addReplySds(c,sdsnew("-ERR background save in progress\r\n"));
	2353	return;
	2354	}
	2355	if (rdbSave(server.dbfilename) == REDIS_OK) {
	2356	addReply(c,shared.ok);
	2357	} else {
	2358	addReply(c,shared.err);
	2359	}
	2360	}
	2361
	2362	static void bgsaveCommand(redisClient *c) {
	2363	if (server.bgsaveinprogress) {
	2364	addReplySds(c,sdsnew("-ERR background save already in progress\r\n"));
	2365	return;
	2366	}
	2367	if (rdbSaveBackground(server.dbfilename) == REDIS_OK) {
	2368	addReply(c,shared.ok);
	2369	} else {
	2370	addReply(c,shared.err);
	2371	}
	2372	}
	2373
	2374	static void shutdownCommand(redisClient *c) {
	2375	redisLog(REDIS_WARNING,"User requested shutdown, saving DB...");
	2376	if (rdbSave(server.dbfilename) == REDIS_OK) {
	2377	if (server.daemonize) {
	2378	unlink(server.pidfile);
	2379	}
	2380	redisLog(REDIS_WARNING,"Server exit now, bye bye...");
	2381	exit(1);
	2382	} else {
	2383	redisLog(REDIS_WARNING,"Error trying to save the DB, can't exit");
	2384	addReplySds(c,sdsnew("-ERR can't quit, problems saving the DB\r\n"));
	2385	}
	2386	}
	2387
	2388	static void renameGenericCommand(redisClient *c, int nx) {
	2389	robj *o;
	2390
	2391	/* To use the same key as src and dst is probably an error */
	2392	if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) {
	2393	addReply(c,shared.sameobjecterr);
	2394	return;
	2395	}
	2396
	2397	o = lookupKeyWrite(c->db,c->argv[1]);
	2398	if (o == NULL) {
	2399	addReply(c,shared.nokeyerr);
	2400	return;
	2401	}
	2402	incrRefCount(o);
	2403	deleteIfVolatile(c->db,c->argv[2]);
	2404	if (dictAdd(c->db->dict,c->argv[2],o) == DICT_ERR) {
	2405	if (nx) {
	2406	decrRefCount(o);
	2407	addReply(c,shared.czero);
	2408	return;
	2409	}
	2410	dictReplace(c->db->dict,c->argv[2],o);
	2411	} else {
	2412	incrRefCount(c->argv[2]);
	2413	}
	2414	deleteKey(c->db,c->argv[1]);
	2415	server.dirty++;
	2416	addReply(c,nx ? shared.cone : shared.ok);
	2417	}
	2418
	2419	static void renameCommand(redisClient *c) {
	2420	renameGenericCommand(c,0);
	2421	}
	2422
	2423	static void renamenxCommand(redisClient *c) {
	2424	renameGenericCommand(c,1);
	2425	}
	2426
	2427	static void moveCommand(redisClient *c) {
	2428	robj *o;
	2429	redisDb src, dst;
	2430	int srcid;
	2431
	2432	/* Obtain source and target DB pointers */
	2433	src = c->db;
	2434	srcid = c->db->id;
	2435	if (selectDb(c,atoi(c->argv[2]->ptr)) == REDIS_ERR) {
	2436	addReply(c,shared.outofrangeerr);
	2437	return;
	2438	}
	2439	dst = c->db;
	2440	selectDb(c,srcid); /* Back to the source DB */
	2441
	2442	/* If the user is moving using as target the same
	2443	* DB as the source DB it is probably an error. */
	2444	if (src == dst) {
	2445	addReply(c,shared.sameobjecterr);
	2446	return;
	2447	}
	2448
	2449	/* Check if the element exists and get a reference */
	2450	o = lookupKeyWrite(c->db,c->argv[1]);
	2451	if (!o) {
	2452	addReply(c,shared.czero);
	2453	return;
	2454	}
	2455
	2456	/* Try to add the element to the target DB */
	2457	deleteIfVolatile(dst,c->argv[1]);
	2458	if (dictAdd(dst->dict,c->argv[1],o) == DICT_ERR) {
	2459	addReply(c,shared.czero);
	2460	return;
	2461	}
	2462	incrRefCount(c->argv[1]);
	2463	incrRefCount(o);
	2464
	2465	/* OK! key moved, free the entry in the source DB */
	2466	deleteKey(src,c->argv[1]);
	2467	server.dirty++;
	2468	addReply(c,shared.cone);
	2469	}
	2470
	2471	/* =================================== Lists ================================ */
	2472	static void pushGenericCommand(redisClient *c, int where) {
	2473	robj *lobj;
	2474	list *list;
	2475
	2476	lobj = lookupKeyWrite(c->db,c->argv[1]);
	2477	if (lobj == NULL) {
	2478	lobj = createListObject();
	2479	list = lobj->ptr;
	2480	if (where == REDIS_HEAD) {
	2481	if (!listAddNodeHead(list,c->argv[2])) oom("listAddNodeHead");
	2482	} else {
	2483	if (!listAddNodeTail(list,c->argv[2])) oom("listAddNodeTail");
	2484	}
	2485	dictAdd(c->db->dict,c->argv[1],lobj);
	2486	incrRefCount(c->argv[1]);
	2487	incrRefCount(c->argv[2]);
	2488	} else {
	2489	if (lobj->type != REDIS_LIST) {
	2490	addReply(c,shared.wrongtypeerr);
	2491	return;
	2492	}
	2493	list = lobj->ptr;
	2494	if (where == REDIS_HEAD) {
	2495	if (!listAddNodeHead(list,c->argv[2])) oom("listAddNodeHead");
	2496	} else {
	2497	if (!listAddNodeTail(list,c->argv[2])) oom("listAddNodeTail");
	2498	}
	2499	incrRefCount(c->argv[2]);
	2500	}
	2501	server.dirty++;
	2502	addReply(c,shared.ok);
	2503	}
	2504
	2505	static void lpushCommand(redisClient *c) {
	2506	pushGenericCommand(c,REDIS_HEAD);
	2507	}
	2508
	2509	static void rpushCommand(redisClient *c) {
	2510	pushGenericCommand(c,REDIS_TAIL);
	2511	}
	2512
	2513	static void llenCommand(redisClient *c) {
	2514	robj *o;
	2515	list *l;
	2516
	2517	o = lookupKeyRead(c->db,c->argv[1]);
	2518	if (o == NULL) {
	2519	addReply(c,shared.czero);
	2520	return;
	2521	} else {
	2522	if (o->type != REDIS_LIST) {
	2523	addReply(c,shared.wrongtypeerr);
	2524	} else {
	2525	l = o->ptr;
	2526	addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",listLength(l)));
	2527	}
	2528	}
	2529	}
	2530
	2531	static void lindexCommand(redisClient *c) {
	2532	robj *o;
	2533	int index = atoi(c->argv[2]->ptr);
	2534
	2535	o = lookupKeyRead(c->db,c->argv[1]);
	2536	if (o == NULL) {
	2537	addReply(c,shared.nullbulk);
	2538	} else {
	2539	if (o->type != REDIS_LIST) {
	2540	addReply(c,shared.wrongtypeerr);
	2541	} else {
	2542	list *list = o->ptr;
	2543	listNode *ln;
	2544
	2545	ln = listIndex(list, index);
	2546	if (ln == NULL) {
	2547	addReply(c,shared.nullbulk);
	2548	} else {
	2549	robj *ele = listNodeValue(ln);
	2550	addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
	2551	addReply(c,ele);
	2552	addReply(c,shared.crlf);
	2553	}
	2554	}
	2555	}
	2556	}
	2557
	2558	static void lsetCommand(redisClient *c) {
	2559	robj *o;
	2560	int index = atoi(c->argv[2]->ptr);
	2561
	2562	o = lookupKeyWrite(c->db,c->argv[1]);
	2563	if (o == NULL) {
	2564	addReply(c,shared.nokeyerr);
	2565	} else {
	2566	if (o->type != REDIS_LIST) {
	2567	addReply(c,shared.wrongtypeerr);
	2568	} else {
	2569	list *list = o->ptr;
	2570	listNode *ln;
	2571
	2572	ln = listIndex(list, index);
	2573	if (ln == NULL) {
	2574	addReply(c,shared.outofrangeerr);
	2575	} else {
	2576	robj *ele = listNodeValue(ln);
	2577
	2578	decrRefCount(ele);
	2579	listNodeValue(ln) = c->argv[3];
	2580	incrRefCount(c->argv[3]);
	2581	addReply(c,shared.ok);
	2582	server.dirty++;
	2583	}
	2584	}
	2585	}
	2586	}
	2587
	2588	static void popGenericCommand(redisClient *c, int where) {
	2589	robj *o;
	2590
	2591	o = lookupKeyWrite(c->db,c->argv[1]);
	2592	if (o == NULL) {
	2593	addReply(c,shared.nullbulk);
	2594	} else {
	2595	if (o->type != REDIS_LIST) {
	2596	addReply(c,shared.wrongtypeerr);
	2597	} else {
	2598	list *list = o->ptr;
	2599	listNode *ln;
	2600
	2601	if (where == REDIS_HEAD)
	2602	ln = listFirst(list);
	2603	else
	2604	ln = listLast(list);
	2605
	2606	if (ln == NULL) {
	2607	addReply(c,shared.nullbulk);
	2608	} else {
	2609	robj *ele = listNodeValue(ln);
	2610	addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
	2611	addReply(c,ele);
	2612	addReply(c,shared.crlf);
	2613	listDelNode(list,ln);
	2614	server.dirty++;
	2615	}
	2616	}
	2617	}
	2618	}
	2619
	2620	static void lpopCommand(redisClient *c) {
	2621	popGenericCommand(c,REDIS_HEAD);
	2622	}
	2623
	2624	static void rpopCommand(redisClient *c) {
	2625	popGenericCommand(c,REDIS_TAIL);
	2626	}
	2627
	2628	static void lrangeCommand(redisClient *c) {
	2629	robj *o;
	2630	int start = atoi(c->argv[2]->ptr);
	2631	int end = atoi(c->argv[3]->ptr);
	2632
	2633	o = lookupKeyRead(c->db,c->argv[1]);
	2634	if (o == NULL) {
	2635	addReply(c,shared.nullmultibulk);
	2636	} else {
	2637	if (o->type != REDIS_LIST) {
	2638	addReply(c,shared.wrongtypeerr);
	2639	} else {
	2640	list *list = o->ptr;
	2641	listNode *ln;
	2642	int llen = listLength(list);
	2643	int rangelen, j;
	2644	robj *ele;
	2645
	2646	/* convert negative indexes */
	2647	if (start < 0) start = llen+start;
	2648	if (end < 0) end = llen+end;
	2649	if (start < 0) start = 0;
	2650	if (end < 0) end = 0;
	2651
	2652	/* indexes sanity checks */
	2653	if (start > end \|\| start >= llen) {
	2654	/* Out of range start or start > end result in empty list */
	2655	addReply(c,shared.emptymultibulk);
	2656	return;
	2657	}
	2658	if (end >= llen) end = llen-1;
	2659	rangelen = (end-start)+1;
	2660
	2661	/* Return the result in form of a multi-bulk reply */
	2662	ln = listIndex(list, start);
	2663	addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",rangelen));
	2664	for (j = 0; j < rangelen; j++) {
	2665	ele = listNodeValue(ln);
	2666	addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",(int)sdslen(ele->ptr)));
	2667	addReply(c,ele);
	2668	addReply(c,shared.crlf);
	2669	ln = ln->next;
	2670	}
	2671	}
	2672	}
	2673	}
	2674
	2675	static void ltrimCommand(redisClient *c) {
	2676	robj *o;
	2677	int start = atoi(c->argv[2]->ptr);
	2678	int end = atoi(c->argv[3]->ptr);
	2679
	2680	o = lookupKeyWrite(c->db,c->argv[1]);
	2681	if (o == NULL) {
	2682	addReply(c,shared.nokeyerr);
	2683	} else {
	2684	if (o->type != REDIS_LIST) {
	2685	addReply(c,shared.wrongtypeerr);
	2686	} else {
	2687	list *list = o->ptr;
	2688	listNode *ln;
	2689	int llen = listLength(list);
	2690	int j, ltrim, rtrim;
	2691
	2692	/* convert negative indexes */
	2693	if (start < 0) start = llen+start;
	2694	if (end < 0) end = llen+end;
	2695	if (start < 0) start = 0;
	2696	if (end < 0) end = 0;
	2697
	2698	/* indexes sanity checks */
	2699	if (start > end \|\| start >= llen) {
	2700	/* Out of range start or start > end result in empty list */
	2701	ltrim = llen;
	2702	rtrim = 0;
	2703	} else {
	2704	if (end >= llen) end = llen-1;
	2705	ltrim = start;
	2706	rtrim = llen-end-1;
	2707	}
	2708
	2709	/* Remove list elements to perform the trim */
	2710	for (j = 0; j < ltrim; j++) {
	2711	ln = listFirst(list);
	2712	listDelNode(list,ln);
	2713	}
	2714	for (j = 0; j < rtrim; j++) {
	2715	ln = listLast(list);
	2716	listDelNode(list,ln);
	2717	}
	2718	addReply(c,shared.ok);
	2719	server.dirty++;
	2720	}
	2721	}
	2722	}
	2723
	2724	static void lremCommand(redisClient *c) {
	2725	robj *o;
	2726
	2727	o = lookupKeyWrite(c->db,c->argv[1]);
	2728	if (o == NULL) {
	2729	addReply(c,shared.nokeyerr);
	2730	} else {
	2731	if (o->type != REDIS_LIST) {
	2732	addReply(c,shared.wrongtypeerr);
	2733	} else {
	2734	list *list = o->ptr;
	2735	listNode ln, next;
	2736	int toremove = atoi(c->argv[2]->ptr);
	2737	int removed = 0;
	2738	int fromtail = 0;
	2739
	2740	if (toremove < 0) {
	2741	toremove = -toremove;
	2742	fromtail = 1;
	2743	}
	2744	ln = fromtail ? list->tail : list->head;
	2745	while (ln) {
	2746	robj *ele = listNodeValue(ln);
	2747
	2748	next = fromtail ? ln->prev : ln->next;
	2749	if (sdscmp(ele->ptr,c->argv[3]->ptr) == 0) {
	2750	listDelNode(list,ln);
	2751	server.dirty++;
	2752	removed++;
	2753	if (toremove && removed == toremove) break;
	2754	}
	2755	ln = next;
	2756	}
	2757	addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",removed));
	2758	}
	2759	}
	2760	}
	2761
	2762	/* ==================================== Sets ================================ */
	2763
	2764	static void saddCommand(redisClient *c) {
	2765	robj *set;
	2766
	2767	set = lookupKeyWrite(c->db,c->argv[1]);
	2768	if (set == NULL) {
	2769	set = createSetObject();
	2770	dictAdd(c->db->dict,c->argv[1],set);
	2771	incrRefCount(c->argv[1]);
	2772	} else {
	2773	if (set->type != REDIS_SET) {
	2774	addReply(c,shared.wrongtypeerr);
	2775	return;
	2776	}
	2777	}
	2778	if (dictAdd(set->ptr,c->argv[2],NULL) == DICT_OK) {
	2779	incrRefCount(c->argv[2]);
	2780	server.dirty++;
	2781	addReply(c,shared.cone);
	2782	} else {
	2783	addReply(c,shared.czero);
	2784	}
	2785	}
	2786
	2787	static void sremCommand(redisClient *c) {
	2788	robj *set;
	2789
	2790	set = lookupKeyWrite(c->db,c->argv[1]);
	2791	if (set == NULL) {
	2792	addReply(c,shared.czero);
	2793	} else {
	2794	if (set->type != REDIS_SET) {
	2795	addReply(c,shared.wrongtypeerr);
	2796	return;
	2797	}
	2798	if (dictDelete(set->ptr,c->argv[2]) == DICT_OK) {
	2799	server.dirty++;
	2800	addReply(c,shared.cone);
	2801	} else {
	2802	addReply(c,shared.czero);
	2803	}
	2804	}
	2805	}
	2806
	2807	static void sismemberCommand(redisClient *c) {
	2808	robj *set;
	2809
	2810	set = lookupKeyRead(c->db,c->argv[1]);
	2811	if (set == NULL) {
	2812	addReply(c,shared.czero);
	2813	} else {
	2814	if (set->type != REDIS_SET) {
	2815	addReply(c,shared.wrongtypeerr);
	2816	return;
	2817	}
	2818	if (dictFind(set->ptr,c->argv[2]))
	2819	addReply(c,shared.cone);
	2820	else
	2821	addReply(c,shared.czero);
	2822	}
	2823	}
	2824
	2825	static void scardCommand(redisClient *c) {
	2826	robj *o;
	2827	dict *s;
	2828
	2829	o = lookupKeyRead(c->db,c->argv[1]);
	2830	if (o == NULL) {
	2831	addReply(c,shared.czero);
	2832	return;
	2833	} else {
	2834	if (o->type != REDIS_SET) {
	2835	addReply(c,shared.wrongtypeerr);
	2836	} else {
	2837	s = o->ptr;
	2838	addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",
	2839	dictSize(s)));
	2840	}
	2841	}
	2842	}
	2843
	2844	static int qsortCompareSetsByCardinality(const void s1, const void s2) {
	2845	dict *d1 = (void) s1, *d2 = (void) s2;
	2846
	2847	return dictSize(d1)-dictSize(d2);
	2848	}
	2849
	2850	static void sinterGenericCommand(redisClient c, robj setskeys, int setsnum, robj dstkey) {
	2851	dict *dv = zmalloc(sizeof(dict)*setsnum);
	2852	dictIterator *di;
	2853	dictEntry *de;
	2854	robj lenobj = NULL, dstset = NULL;
	2855	int j, cardinality = 0;
	2856
	2857	if (!dv) oom("sinterCommand");
	2858	for (j = 0; j < setsnum; j++) {
	2859	robj *setobj;
	2860
	2861	setobj = dstkey ?
	2862	lookupKeyWrite(c->db,setskeys[j]) :
	2863	lookupKeyRead(c->db,setskeys[j]);
	2864	if (!setobj) {
	2865	zfree(dv);
	2866	if (dstkey) {
	2867	deleteKey(c->db,dstkey);
	2868	addReply(c,shared.ok);
	2869	} else {
	2870	addReply(c,shared.nullmultibulk);
	2871	}
	2872	return;
	2873	}
	2874	if (setobj->type != REDIS_SET) {
	2875	zfree(dv);
	2876	addReply(c,shared.wrongtypeerr);
	2877	return;
	2878	}
	2879	dv[j] = setobj->ptr;
	2880	}
	2881	/* Sort sets from the smallest to largest, this will improve our
	2882	* algorithm's performace */
	2883	qsort(dv,setsnum,sizeof(dict*),qsortCompareSetsByCardinality);
	2884
	2885	/* The first thing we should output is the total number of elements...
	2886	* since this is a multi-bulk write, but at this stage we don't know
	2887	* the intersection set size, so we use a trick, append an empty object
	2888	* to the output list and save the pointer to later modify it with the
	2889	* right length */
	2890	if (!dstkey) {
	2891	lenobj = createObject(REDIS_STRING,NULL);
	2892	addReply(c,lenobj);
	2893	decrRefCount(lenobj);
	2894	} else {
	2895	/* If we have a target key where to store the resulting set
	2896	* create this key with an empty set inside */
	2897	dstset = createSetObject();
	2898	deleteKey(c->db,dstkey);
	2899	dictAdd(c->db->dict,dstkey,dstset);
	2900	incrRefCount(dstkey);
	2901	}
	2902
	2903	/* Iterate all the elements of the first (smallest) set, and test
	2904	* the element against all the other sets, if at least one set does
	2905	* not include the element it is discarded */
	2906	di = dictGetIterator(dv[0]);
	2907	if (!di) oom("dictGetIterator");
	2908
	2909	while((de = dictNext(di)) != NULL) {
	2910	robj *ele;
	2911
	2912	for (j = 1; j < setsnum; j++)
	2913	if (dictFind(dv[j],dictGetEntryKey(de)) == NULL) break;
	2914	if (j != setsnum)
	2915	continue; /* at least one set does not contain the member */
	2916	ele = dictGetEntryKey(de);
	2917	if (!dstkey) {
	2918	addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(ele->ptr)));
	2919	addReply(c,ele);
	2920	addReply(c,shared.crlf);
	2921	cardinality++;
	2922	} else {
	2923	dictAdd(dstset->ptr,ele,NULL);
	2924	incrRefCount(ele);
	2925	}
	2926	}
	2927	dictReleaseIterator(di);
	2928
	2929	if (!dstkey) {
	2930	lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality);
	2931	} else {
	2932	addReply(c,shared.ok);
	2933	server.dirty++;
	2934	}
	2935	zfree(dv);
	2936	}
	2937
	2938	static void sinterCommand(redisClient *c) {
	2939	sinterGenericCommand(c,c->argv+1,c->argc-1,NULL);
	2940	}
	2941
	2942	static void sinterstoreCommand(redisClient *c) {
	2943	sinterGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
	2944	}
	2945
	2946	static void sunionGenericCommand(redisClient c, robj setskeys, int setsnum, robj dstkey) {
	2947	dict *dv = zmalloc(sizeof(dict)*setsnum);
	2948	dictIterator *di;
	2949	dictEntry *de;
	2950	robj lenobj = NULL, dstset = NULL;
	2951	int j, cardinality = 0;
	2952
	2953	if (!dv) oom("sunionCommand");
	2954	for (j = 0; j < setsnum; j++) {
	2955	robj *setobj;
	2956
	2957	setobj = dstkey ?
	2958	lookupKeyWrite(c->db,setskeys[j]) :
	2959	lookupKeyRead(c->db,setskeys[j]);
	2960	if (!setobj) {
	2961	dv[j] = NULL;
	2962	continue;
	2963	}
	2964	if (setobj->type != REDIS_SET) {
	2965	zfree(dv);
	2966	addReply(c,shared.wrongtypeerr);
	2967	return;
	2968	}
	2969	dv[j] = setobj->ptr;
	2970	}
	2971
	2972	/* We need a temp set object to store our union. If the dstkey
	2973	* is not NULL (that is, we are inside an SUNIONSTORE operation) then
	2974	* this set object will be the resulting object to set into the target key*/
	2975	dstset = createSetObject();
	2976
	2977	/* The first thing we should output is the total number of elements...
	2978	* since this is a multi-bulk write, but at this stage we don't know
	2979	* the intersection set size, so we use a trick, append an empty object
	2980	* to the output list and save the pointer to later modify it with the
	2981	* right length */
	2982	if (!dstkey) {
	2983	lenobj = createObject(REDIS_STRING,NULL);
	2984	addReply(c,lenobj);
	2985	decrRefCount(lenobj);
	2986	} else {
	2987	/* If we have a target key where to store the resulting set
	2988	* create this key with an empty set inside */
	2989	deleteKey(c->db,dstkey);
	2990	dictAdd(c->db->dict,dstkey,dstset);
	2991	incrRefCount(dstkey);
	2992	server.dirty++;
	2993	}
	2994
	2995	/* Iterate all the elements of all the sets, add every element a single
	2996	* time to the result set */
	2997	for (j = 0; j < setsnum; j++) {
	2998	if (!dv[j]) continue; /* non existing keys are like empty sets */
	2999
	3000	di = dictGetIterator(dv[j]);
	3001	if (!di) oom("dictGetIterator");
	3002
	3003	while((de = dictNext(di)) != NULL) {
	3004	robj *ele;
	3005
	3006	/* dictAdd will not add the same element multiple times */
	3007	ele = dictGetEntryKey(de);
	3008	if (dictAdd(dstset->ptr,ele,NULL) == DICT_OK) {
	3009	incrRefCount(ele);
	3010	if (!dstkey) {
	3011	addReplySds(c,sdscatprintf(sdsempty(),
	3012	"$%d\r\n",sdslen(ele->ptr)));
	3013	addReply(c,ele);
	3014	addReply(c,shared.crlf);
	3015	cardinality++;
	3016	}
	3017	}
	3018	}
	3019	dictReleaseIterator(di);
	3020	}
	3021
	3022	if (!dstkey) {
	3023	lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality);
	3024	decrRefCount(dstset);
	3025	} else {
	3026	addReply(c,shared.ok);
	3027	server.dirty++;
	3028	}
	3029	zfree(dv);
	3030	}
	3031
	3032	static void sunionCommand(redisClient *c) {
	3033	sunionGenericCommand(c,c->argv+1,c->argc-1,NULL);
	3034	}
	3035
	3036	static void sunionstoreCommand(redisClient *c) {
	3037	sunionGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
	3038	}
	3039
	3040	static void flushdbCommand(redisClient *c) {
	3041	dictEmpty(c->db->dict);
	3042	dictEmpty(c->db->expires);
	3043	server.dirty++;
	3044	addReply(c,shared.ok);
	3045	rdbSave(server.dbfilename);
	3046	}
	3047
	3048	static void flushallCommand(redisClient *c) {
	3049	emptyDb();
	3050	server.dirty++;
	3051	addReply(c,shared.ok);
	3052	rdbSave(server.dbfilename);
	3053	}
	3054
	3055	redisSortOperation createSortOperation(int type, robj pattern) {
	3056	redisSortOperation so = zmalloc(sizeof(so));
	3057	if (!so) oom("createSortOperation");
	3058	so->type = type;
	3059	so->pattern = pattern;
	3060	return so;
	3061	}
	3062
	3063	/* Return the value associated to the key with a name obtained
	3064	* substituting the first occurence of '' in 'pattern' with 'subst' /
	3065	robj lookupKeyByPattern(redisDb db, robj pattern, robj subst) {
	3066	char *p;
	3067	sds spat, ssub;
	3068	robj keyobj;
	3069	int prefixlen, sublen, postfixlen;
	3070	/* Expoit the internal sds representation to create a sds string allocated on the stack in order to make this function faster */
	3071	struct {
	3072	long len;
	3073	long free;
	3074	char buf[REDIS_SORTKEY_MAX+1];
	3075	} keyname;
	3076
	3077	spat = pattern->ptr;
	3078	ssub = subst->ptr;
	3079	if (sdslen(spat)+sdslen(ssub)-1 > REDIS_SORTKEY_MAX) return NULL;
	3080	p = strchr(spat,'*');
	3081	if (!p) return NULL;
	3082
	3083	prefixlen = p-spat;
	3084	sublen = sdslen(ssub);
	3085	postfixlen = sdslen(spat)-(prefixlen+1);
	3086	memcpy(keyname.buf,spat,prefixlen);
	3087	memcpy(keyname.buf+prefixlen,ssub,sublen);
	3088	memcpy(keyname.buf+prefixlen+sublen,p+1,postfixlen);
	3089	keyname.buf[prefixlen+sublen+postfixlen] = '\0';
	3090	keyname.len = prefixlen+sublen+postfixlen;
	3091
	3092	keyobj.refcount = 1;
	3093	keyobj.type = REDIS_STRING;
	3094	keyobj.ptr = ((char)&keyname)+(sizeof(long)2);
	3095
	3096	/* printf("lookup '%s' => %p\n", keyname.buf,de); */
	3097	return lookupKeyRead(db,&keyobj);
	3098	}
	3099
	3100	/* sortCompare() is used by qsort in sortCommand(). Given that qsort_r with
	3101	* the additional parameter is not standard but a BSD-specific we have to
	3102	* pass sorting parameters via the global 'server' structure */
	3103	static int sortCompare(const void s1, const void s2) {
	3104	const redisSortObject so1 = s1, so2 = s2;
	3105	int cmp;
	3106
	3107	if (!server.sort_alpha) {
	3108	/* Numeric sorting. Here it's trivial as we precomputed scores */
	3109	if (so1->u.score > so2->u.score) {
	3110	cmp = 1;
	3111	} else if (so1->u.score < so2->u.score) {
	3112	cmp = -1;
	3113	} else {
	3114	cmp = 0;
	3115	}
	3116	} else {
	3117	/* Alphanumeric sorting */
	3118	if (server.sort_bypattern) {
	3119	if (!so1->u.cmpobj \|\| !so2->u.cmpobj) {
	3120	/* At least one compare object is NULL */
	3121	if (so1->u.cmpobj == so2->u.cmpobj)
	3122	cmp = 0;
	3123	else if (so1->u.cmpobj == NULL)
	3124	cmp = -1;
	3125	else
	3126	cmp = 1;
	3127	} else {
	3128	/* We have both the objects, use strcoll */
	3129	cmp = strcoll(so1->u.cmpobj->ptr,so2->u.cmpobj->ptr);
	3130	}
	3131	} else {
	3132	/* Compare elements directly */
	3133	cmp = strcoll(so1->obj->ptr,so2->obj->ptr);
	3134	}
	3135	}
	3136	return server.sort_desc ? -cmp : cmp;
	3137	}
	3138
	3139	/* The SORT command is the most complex command in Redis. Warning: this code
	3140	* is optimized for speed and a bit less for readability */
	3141	static void sortCommand(redisClient *c) {
	3142	list *operations;
	3143	int outputlen = 0;
	3144	int desc = 0, alpha = 0;
	3145	int limit_start = 0, limit_count = -1, start, end;
	3146	int j, dontsort = 0, vectorlen;
	3147	int getop = 0; /* GET operation counter */
	3148	robj sortval, sortby = NULL;
	3149	redisSortObject vector; / Resulting vector to sort */
	3150
	3151	/* Lookup the key to sort. It must be of the right types */
	3152	sortval = lookupKeyRead(c->db,c->argv[1]);
	3153	if (sortval == NULL) {
	3154	addReply(c,shared.nokeyerr);
	3155	return;
	3156	}
	3157	if (sortval->type != REDIS_SET && sortval->type != REDIS_LIST) {
	3158	addReply(c,shared.wrongtypeerr);
	3159	return;
	3160	}
	3161
	3162	/* Create a list of operations to perform for every sorted element.
	3163	* Operations can be GET/DEL/INCR/DECR */
	3164	operations = listCreate();
	3165	listSetFreeMethod(operations,zfree);
	3166	j = 2;
	3167
	3168	/* Now we need to protect sortval incrementing its count, in the future
	3169	* SORT may have options able to overwrite/delete keys during the sorting
	3170	* and the sorted key itself may get destroied */
	3171	incrRefCount(sortval);
	3172
	3173	/* The SORT command has an SQL-alike syntax, parse it */
	3174	while(j < c->argc) {
	3175	int leftargs = c->argc-j-1;
	3176	if (!strcasecmp(c->argv[j]->ptr,"asc")) {
	3177	desc = 0;
	3178	} else if (!strcasecmp(c->argv[j]->ptr,"desc")) {
	3179	desc = 1;
	3180	} else if (!strcasecmp(c->argv[j]->ptr,"alpha")) {
	3181	alpha = 1;
	3182	} else if (!strcasecmp(c->argv[j]->ptr,"limit") && leftargs >= 2) {
	3183	limit_start = atoi(c->argv[j+1]->ptr);
	3184	limit_count = atoi(c->argv[j+2]->ptr);
	3185	j+=2;
	3186	} else if (!strcasecmp(c->argv[j]->ptr,"by") && leftargs >= 1) {
	3187	sortby = c->argv[j+1];
	3188	/* If the BY pattern does not contain '*', i.e. it is constant,
	3189	* we don't need to sort nor to lookup the weight keys. */
	3190	if (strchr(c->argv[j+1]->ptr,'*') == NULL) dontsort = 1;
	3191	j++;
	3192	} else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
	3193	listAddNodeTail(operations,createSortOperation(
	3194	REDIS_SORT_GET,c->argv[j+1]));
	3195	getop++;
	3196	j++;
	3197	} else if (!strcasecmp(c->argv[j]->ptr,"del") && leftargs >= 1) {
	3198	listAddNodeTail(operations,createSortOperation(
	3199	REDIS_SORT_DEL,c->argv[j+1]));
	3200	j++;
	3201	} else if (!strcasecmp(c->argv[j]->ptr,"incr") && leftargs >= 1) {
	3202	listAddNodeTail(operations,createSortOperation(
	3203	REDIS_SORT_INCR,c->argv[j+1]));
	3204	j++;
	3205	} else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
	3206	listAddNodeTail(operations,createSortOperation(
	3207	REDIS_SORT_DECR,c->argv[j+1]));
	3208	j++;
	3209	} else {
	3210	decrRefCount(sortval);
	3211	listRelease(operations);
	3212	addReply(c,shared.syntaxerr);
	3213	return;
	3214	}
	3215	j++;
	3216	}
	3217
	3218	/* Load the sorting vector with all the objects to sort */
	3219	vectorlen = (sortval->type == REDIS_LIST) ?
	3220	listLength((list*)sortval->ptr) :
	3221	dictSize((dict*)sortval->ptr);
	3222	vector = zmalloc(sizeof(redisSortObject)*vectorlen);
	3223	if (!vector) oom("allocating objects vector for SORT");
	3224	j = 0;
	3225	if (sortval->type == REDIS_LIST) {
	3226	list *list = sortval->ptr;
	3227	listNode *ln = list->head;
	3228	while(ln) {
	3229	robj *ele = ln->value;
	3230	vector[j].obj = ele;
	3231	vector[j].u.score = 0;
	3232	vector[j].u.cmpobj = NULL;
	3233	ln = ln->next;
	3234	j++;
	3235	}
	3236	} else {
	3237	dict *set = sortval->ptr;
	3238	dictIterator *di;
	3239	dictEntry *setele;
	3240
	3241	di = dictGetIterator(set);
	3242	if (!di) oom("dictGetIterator");
	3243	while((setele = dictNext(di)) != NULL) {
	3244	vector[j].obj = dictGetEntryKey(setele);
	3245	vector[j].u.score = 0;
	3246	vector[j].u.cmpobj = NULL;
	3247	j++;
	3248	}
	3249	dictReleaseIterator(di);
	3250	}
	3251	assert(j == vectorlen);
	3252
	3253	/* Now it's time to load the right scores in the sorting vector */
	3254	if (dontsort == 0) {
	3255	for (j = 0; j < vectorlen; j++) {
	3256	if (sortby) {
	3257	robj *byval;
	3258
	3259	byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
	3260	if (!byval \|\| byval->type != REDIS_STRING) continue;
	3261	if (alpha) {
	3262	vector[j].u.cmpobj = byval;
	3263	incrRefCount(byval);
	3264	} else {
	3265	vector[j].u.score = strtod(byval->ptr,NULL);
	3266	}
	3267	} else {
	3268	if (!alpha) vector[j].u.score = strtod(vector[j].obj->ptr,NULL);
	3269	}
	3270	}
	3271	}
	3272
	3273	/* We are ready to sort the vector... perform a bit of sanity check
	3274	* on the LIMIT option too. We'll use a partial version of quicksort. */
	3275	start = (limit_start < 0) ? 0 : limit_start;
	3276	end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
	3277	if (start >= vectorlen) {
	3278	start = vectorlen-1;
	3279	end = vectorlen-2;
	3280	}
	3281	if (end >= vectorlen) end = vectorlen-1;
	3282
	3283	if (dontsort == 0) {
	3284	server.sort_desc = desc;
	3285	server.sort_alpha = alpha;
	3286	server.sort_bypattern = sortby ? 1 : 0;
	3287	qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare);
	3288	}
	3289
	3290	/* Send command output to the output buffer, performing the specified
	3291	* GET/DEL/INCR/DECR operations if any. */
	3292	outputlen = getop ? getop*(end-start+1) : end-start+1;
	3293	addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",outputlen));
	3294	for (j = start; j <= end; j++) {
	3295	listNode *ln = operations->head;
	3296	if (!getop) {
	3297	addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
	3298	sdslen(vector[j].obj->ptr)));
	3299	addReply(c,vector[j].obj);
	3300	addReply(c,shared.crlf);
	3301	}
	3302	while(ln) {
	3303	redisSortOperation *sop = ln->value;
	3304	robj *val = lookupKeyByPattern(c->db,sop->pattern,
	3305	vector[j].obj);
	3306
	3307	if (sop->type == REDIS_SORT_GET) {
	3308	if (!val \|\| val->type != REDIS_STRING) {
	3309	addReply(c,shared.nullbulk);
	3310	} else {
	3311	addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
	3312	sdslen(val->ptr)));
	3313	addReply(c,val);
	3314	addReply(c,shared.crlf);
	3315	}
	3316	} else if (sop->type == REDIS_SORT_DEL) {
	3317	/* TODO */
	3318	}
	3319	ln = ln->next;
	3320	}
	3321	}
	3322
	3323	/* Cleanup */
	3324	decrRefCount(sortval);
	3325	listRelease(operations);
	3326	for (j = 0; j < vectorlen; j++) {
	3327	if (sortby && alpha && vector[j].u.cmpobj)
	3328	decrRefCount(vector[j].u.cmpobj);
	3329	}
	3330	zfree(vector);
	3331	}
	3332
	3333	static void infoCommand(redisClient *c) {
	3334	sds info;
	3335	time_t uptime = time(NULL)-server.stat_starttime;
	3336
	3337	info = sdscatprintf(sdsempty(),
	3338	"redis_version:%s\r\n"
	3339	"connected_clients:%d\r\n"
	3340	"connected_slaves:%d\r\n"
	3341	"used_memory:%zu\r\n"
	3342	"changes_since_last_save:%lld\r\n"
	3343	"last_save_time:%d\r\n"
	3344	"total_connections_received:%lld\r\n"
	3345	"total_commands_processed:%lld\r\n"
	3346	"uptime_in_seconds:%d\r\n"
	3347	"uptime_in_days:%d\r\n"
	3348	,REDIS_VERSION,
	3349	listLength(server.clients)-listLength(server.slaves),
	3350	listLength(server.slaves),
	3351	server.usedmemory,
	3352	server.dirty,
	3353	server.lastsave,
	3354	server.stat_numconnections,
	3355	server.stat_numcommands,
	3356	uptime,
	3357	uptime/(3600*24)
	3358	);
	3359	addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(info)));
	3360	addReplySds(c,info);
	3361	addReply(c,shared.crlf);
	3362	}
	3363
	3364	static void monitorCommand(redisClient *c) {
	3365	/* ignore MONITOR if aleady slave or in monitor mode */
	3366	if (c->flags & REDIS_SLAVE) return;
	3367
	3368	c->flags \|= (REDIS_SLAVE\|REDIS_MONITOR);
	3369	c->slaveseldb = 0;
	3370	if (!listAddNodeTail(server.monitors,c)) oom("listAddNodeTail");
	3371	addReply(c,shared.ok);
	3372	}
	3373
	3374	/* ================================= Expire ================================= */
	3375	static int removeExpire(redisDb db, robj key) {
	3376	if (dictDelete(db->expires,key) == DICT_OK) {
	3377	return 1;
	3378	} else {
	3379	return 0;
	3380	}
	3381	}
	3382
	3383	static int setExpire(redisDb db, robj key, time_t when) {
	3384	if (dictAdd(db->expires,key,(void*)when) == DICT_ERR) {
	3385	return 0;
	3386	} else {
	3387	incrRefCount(key);
	3388	return 1;
	3389	}
	3390	}
	3391
	3392	/* Return the expire time of the specified key, or -1 if no expire
	3393	* is associated with this key (i.e. the key is non volatile) */
	3394	static time_t getExpire(redisDb db, robj key) {
	3395	dictEntry *de;
	3396
	3397	/* No expire? return ASAP */
	3398	if (dictSize(db->expires) == 0 \|\|
	3399	(de = dictFind(db->expires,key)) == NULL) return -1;
	3400
	3401	return (time_t) dictGetEntryVal(de);
	3402	}
	3403
	3404	static int expireIfNeeded(redisDb db, robj key) {
	3405	time_t when;
	3406	dictEntry *de;
	3407
	3408	/* No expire? return ASAP */
	3409	if (dictSize(db->expires) == 0 \|\|
	3410	(de = dictFind(db->expires,key)) == NULL) return 0;
	3411
	3412	/* Lookup the expire */
	3413	when = (time_t) dictGetEntryVal(de);
	3414	if (time(NULL) <= when) return 0;
	3415
	3416	/* Delete the key */
	3417	dictDelete(db->expires,key);
	3418	return dictDelete(db->dict,key) == DICT_OK;
	3419	}
	3420
	3421	static int deleteIfVolatile(redisDb db, robj key) {
	3422	dictEntry *de;
	3423
	3424	/* No expire? return ASAP */
	3425	if (dictSize(db->expires) == 0 \|\|
	3426	(de = dictFind(db->expires,key)) == NULL) return 0;
	3427
	3428	/* Delete the key */
	3429	server.dirty++;
	3430	dictDelete(db->expires,key);
	3431	return dictDelete(db->dict,key) == DICT_OK;
	3432	}
	3433
	3434	static void expireCommand(redisClient *c) {
	3435	dictEntry *de;
	3436	int seconds = atoi(c->argv[2]->ptr);
	3437
	3438	de = dictFind(c->db->dict,c->argv[1]);
	3439	if (de == NULL) {
	3440	addReply(c,shared.czero);
	3441	return;
	3442	}
	3443	if (seconds <= 0) {
	3444	addReply(c, shared.czero);
	3445	return;
	3446	} else {
	3447	time_t when = time(NULL)+seconds;
	3448	if (setExpire(c->db,c->argv[1],when))
	3449	addReply(c,shared.cone);
	3450	else
	3451	addReply(c,shared.czero);
	3452	return;
	3453	}
	3454	}
	3455
	3456	/* =============================== Replication ============================= */
	3457
	3458	/* Send the whole output buffer syncronously to the slave. This a general operation in theory, but it is actually useful only for replication. */
	3459	static int flushClientOutput(redisClient *c) {
	3460	int retval;
	3461	time_t start = time(NULL);
	3462
	3463	while(listLength(c->reply)) {
	3464	if (time(NULL)-start > 5) return REDIS_ERR; /* 5 seconds timeout */
	3465	retval = aeWait(c->fd,AE_WRITABLE,1000);
	3466	if (retval == -1) {
	3467	return REDIS_ERR;
	3468	} else if (retval & AE_WRITABLE) {
	3469	sendReplyToClient(NULL, c->fd, c, AE_WRITABLE);
	3470	}
	3471	}
	3472	return REDIS_OK;
	3473	}
	3474
	3475	static int syncWrite(int fd, char *ptr, ssize_t size, int timeout) {
	3476	ssize_t nwritten, ret = size;
	3477	time_t start = time(NULL);
	3478
	3479	timeout++;
	3480	while(size) {
	3481	if (aeWait(fd,AE_WRITABLE,1000) & AE_WRITABLE) {
	3482	nwritten = write(fd,ptr,size);
	3483	if (nwritten == -1) return -1;
	3484	ptr += nwritten;
	3485	size -= nwritten;
	3486	}
	3487	if ((time(NULL)-start) > timeout) {
	3488	errno = ETIMEDOUT;
	3489	return -1;
	3490	}
	3491	}
	3492	return ret;
	3493	}
	3494
	3495	static int syncRead(int fd, char *ptr, ssize_t size, int timeout) {
	3496	ssize_t nread, totread = 0;
	3497	time_t start = time(NULL);
	3498
	3499	timeout++;
	3500	while(size) {
	3501	if (aeWait(fd,AE_READABLE,1000) & AE_READABLE) {
	3502	nread = read(fd,ptr,size);
	3503	if (nread == -1) return -1;
	3504	ptr += nread;
	3505	size -= nread;
	3506	totread += nread;
	3507	}
	3508	if ((time(NULL)-start) > timeout) {
	3509	errno = ETIMEDOUT;
	3510	return -1;
	3511	}
	3512	}
	3513	return totread;
	3514	}
	3515
	3516	static int syncReadLine(int fd, char *ptr, ssize_t size, int timeout) {
	3517	ssize_t nread = 0;
	3518
	3519	size--;
	3520	while(size) {
	3521	char c;
	3522
	3523	if (syncRead(fd,&c,1,timeout) == -1) return -1;
	3524	if (c == '\n') {
	3525	*ptr = '\0';
	3526	if (nread && (ptr-1) == '\r') (ptr-1) = '\0';
	3527	return nread;
	3528	} else {
	3529	*ptr++ = c;
	3530	*ptr = '\0';
	3531	nread++;
	3532	}
	3533	}
	3534	return nread;
	3535	}
	3536
	3537	static void syncCommand(redisClient *c) {
	3538	/* ignore SYNC if aleady slave or in monitor mode */
	3539	if (c->flags & REDIS_SLAVE) return;
	3540
	3541	/* SYNC can't be issued when the server has pending data to send to
	3542	* the client about already issued commands. We need a fresh reply
	3543	* buffer registering the differences between the BGSAVE and the current
	3544	* dataset, so that we can copy to other slaves if needed. */
	3545	if (listLength(c->reply) != 0) {
	3546	addReplySds(c,sdsnew("-ERR SYNC is invalid with pending input\r\n"));
	3547	return;
	3548	}
	3549
	3550	redisLog(REDIS_NOTICE,"Slave ask for synchronization");
	3551	/* Here we need to check if there is a background saving operation
	3552	* in progress, or if it is required to start one */
	3553	if (server.bgsaveinprogress) {
	3554	/* Ok a background save is in progress. Let's check if it is a good
	3555	* one for replication, i.e. if there is another slave that is
	3556	* registering differences since the server forked to save */
	3557	redisClient *slave;
	3558	listNode *ln;
	3559
	3560	ln = server.slaves->head;
	3561	while(ln) {
	3562	slave = ln->value;
	3563	if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
	3564	ln = ln->next;
	3565	}
	3566	if (ln) {
	3567	/* Perfect, the server is already registering differences for
	3568	* another slave. Set the right state, and copy the buffer. */
	3569	listRelease(c->reply);
	3570	c->reply = listDup(slave->reply);
	3571	if (!c->reply) oom("listDup copying slave reply list");
	3572	c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
	3573	redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC");
	3574	} else {
	3575	/* No way, we need to wait for the next BGSAVE in order to
	3576	* register differences */
	3577	c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
	3578	redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
	3579	}
	3580	} else {
	3581	/* Ok we don't have a BGSAVE in progress, let's start one */
	3582	redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC");
	3583	if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
	3584	redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE");
	3585	addReplySds(c,sdsnew("-ERR Unalbe to perform background save\r\n"));
	3586	return;
	3587	}
	3588	c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
	3589	}
	3590	c->flags \|= REDIS_SLAVE;
	3591	c->slaveseldb = 0;
	3592	if (!listAddNodeTail(server.slaves,c)) oom("listAddNodeTail");
	3593	redisLog(REDIS_NOTICE,"Synchronization with slave succeeded");
	3594	return;
	3595	}
	3596
	3597	#if 0
	3598	static void _syncCommand(redisClient *c) {
	3599	struct stat sb;
	3600	int fd = -1, len;
	3601	time_t start = time(NULL);
	3602	char sizebuf[32];
	3603
	3604	/* ignore SYNC if aleady slave or in monitor mode */
	3605	if (c->flags & REDIS_SLAVE) return;
	3606
	3607	redisLog(REDIS_NOTICE,"Slave ask for synchronization");
	3608	if (flushClientOutput(c) == REDIS_ERR \|\|
	3609	rdbSave(server.dbfilename) != REDIS_OK)
	3610	goto closeconn;
	3611
	3612	fd = open(server.dbfilename, O_RDONLY);
	3613	if (fd == -1 \|\| fstat(fd,&sb) == -1) goto closeconn;
	3614	len = sb.st_size;
	3615
	3616	snprintf(sizebuf,32,"$%d\r\n",len);
	3617	if (syncWrite(c->fd,sizebuf,strlen(sizebuf),5) == -1) goto closeconn;
	3618	while(len) {
	3619	char buf[1024];
	3620	int nread;
	3621
	3622	if (time(NULL)-start > REDIS_MAX_SYNC_TIME) goto closeconn;
	3623	nread = read(fd,buf,1024);
	3624	if (nread == -1) goto closeconn;
	3625	len -= nread;
	3626	if (syncWrite(c->fd,buf,nread,5) == -1) goto closeconn;
	3627	}
	3628	if (syncWrite(c->fd,"\r\n",2,5) == -1) goto closeconn;
	3629	close(fd);
	3630	c->flags \|= REDIS_SLAVE;
	3631	c->slaveseldb = 0;
	3632	if (!listAddNodeTail(server.slaves,c)) oom("listAddNodeTail");
	3633	redisLog(REDIS_NOTICE,"Synchronization with slave succeeded");
	3634	return;
	3635
	3636	closeconn:
	3637	if (fd != -1) close(fd);
	3638	c->flags \|= REDIS_CLOSE;
	3639	redisLog(REDIS_WARNING,"Synchronization with slave failed");
	3640	return;
	3641	}
	3642	#endif
	3643
	3644	static int syncWithMaster(void) {
	3645	char buf[1024], tmpfile[256];
	3646	int dumpsize;
	3647	int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);
	3648	int dfd;
	3649
	3650	if (fd == -1) {
	3651	redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
	3652	strerror(errno));
	3653	return REDIS_ERR;
	3654	}
	3655	/* Issue the SYNC command */
	3656	if (syncWrite(fd,"SYNC \r\n",7,5) == -1) {
	3657	close(fd);
	3658	redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
	3659	strerror(errno));
	3660	return REDIS_ERR;
	3661	}
	3662	/* Read the bulk write count */
	3663	if (syncReadLine(fd,buf,1024,5) == -1) {
	3664	close(fd);
	3665	redisLog(REDIS_WARNING,"I/O error reading bulk count from MASTER: %s",
	3666	strerror(errno));
	3667	return REDIS_ERR;
	3668	}
	3669	dumpsize = atoi(buf+1);
	3670	redisLog(REDIS_NOTICE,"Receiving %d bytes data dump from MASTER",dumpsize);
	3671	/* Read the bulk write data on a temp file */
	3672	snprintf(tmpfile,256,"temp-%d.%ld.rdb",(int)time(NULL),(long int)random());
	3673	dfd = open(tmpfile,O_CREAT\|O_WRONLY,0644);
	3674	if (dfd == -1) {
	3675	close(fd);
	3676	redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
	3677	return REDIS_ERR;
	3678	}
	3679	while(dumpsize) {
	3680	int nread, nwritten;
	3681
	3682	nread = read(fd,buf,(dumpsize < 1024)?dumpsize:1024);
	3683	if (nread == -1) {
	3684	redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
	3685	strerror(errno));
	3686	close(fd);
	3687	close(dfd);
	3688	return REDIS_ERR;
	3689	}
	3690	nwritten = write(dfd,buf,nread);
	3691	if (nwritten == -1) {
	3692	redisLog(REDIS_WARNING,"Write error writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno));
	3693	close(fd);
	3694	close(dfd);
	3695	return REDIS_ERR;
	3696	}
	3697	dumpsize -= nread;
	3698	}
	3699	close(dfd);
	3700	if (rename(tmpfile,server.dbfilename) == -1) {
	3701	redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
	3702	unlink(tmpfile);
	3703	close(fd);
	3704	return REDIS_ERR;
	3705	}
	3706	emptyDb();
	3707	if (rdbLoad(server.dbfilename) != REDIS_OK) {
	3708	redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
	3709	close(fd);
	3710	return REDIS_ERR;
	3711	}
	3712	server.master = createClient(fd);
	3713	server.master->flags \|= REDIS_MASTER;
	3714	server.replstate = REDIS_REPL_CONNECTED;
	3715	return REDIS_OK;
	3716	}
	3717
	3718	/* =================================== Main! ================================ */
	3719
	3720	static void daemonize(void) {
	3721	int fd;
	3722	FILE *fp;
	3723
	3724	if (fork() != 0) exit(0); /* parent exits */
	3725	setsid(); /* create a new session */
	3726
	3727	/* Every output goes to /dev/null. If Redis is daemonized but
	3728	* the 'logfile' is set to 'stdout' in the configuration file
	3729	* it will not log at all. */
	3730	if ((fd = open("/dev/null", O_RDWR, 0)) != -1) {
	3731	dup2(fd, STDIN_FILENO);
	3732	dup2(fd, STDOUT_FILENO);
	3733	dup2(fd, STDERR_FILENO);
	3734	if (fd > STDERR_FILENO) close(fd);
	3735	}
	3736	/* Try to write the pid file */
	3737	fp = fopen(server.pidfile,"w");
	3738	if (fp) {
	3739	fprintf(fp,"%d\n",getpid());
	3740	fclose(fp);
	3741	}
	3742	}
	3743
	3744	int main(int argc, char **argv) {
	3745	initServerConfig();
	3746	if (argc == 2) {
	3747	ResetServerSaveParams();
	3748	loadServerConfig(argv[1]);
	3749	} else if (argc > 2) {
	3750	fprintf(stderr,"Usage: ./redis-server [/path/to/redis.conf]\n");
	3751	exit(1);
	3752	}
	3753	initServer();
	3754	if (server.daemonize) daemonize();
	3755	redisLog(REDIS_NOTICE,"Server started, Redis version " REDIS_VERSION);
	3756	if (rdbLoad(server.dbfilename) == REDIS_OK)
	3757	redisLog(REDIS_NOTICE,"DB loaded from disk");
	3758	if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
	3759	acceptHandler, NULL, NULL) == AE_ERR) oom("creating file event");
	3760	redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
	3761	aeMain(server.el);
	3762	aeDeleteEventLoop(server.el);
	3763	return 0;
	3764	}