X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/93ea375904fa2a5d1cab110698d15f43e63d84f1..4d30232b0408f78d904ee484456e53b2b8fa6b0f:/redis.c diff --git a/redis.c b/redis.c index 0d45040a..a7c3f211 100644 --- a/redis.c +++ b/redis.c @@ -56,7 +56,8 @@ #include "dict.h" /* Hash tables */ #include "adlist.h" /* Linked lists */ #include "zmalloc.h" /* total memory usage aware version of malloc/free */ -#include "lzf.h" +#include "lzf.h" /* LZF compression library */ +#include "pqsort.h" /* Partial qsort for SORT+LIMIT */ /* Error codes */ #define REDIS_OK 0 @@ -348,6 +349,8 @@ static void sinterCommand(redisClient *c); static void sinterstoreCommand(redisClient *c); static void sunionCommand(redisClient *c); static void sunionstoreCommand(redisClient *c); +static void sdiffCommand(redisClient *c); +static void sdiffstoreCommand(redisClient *c); static void syncCommand(redisClient *c); static void flushdbCommand(redisClient *c); static void flushallCommand(redisClient *c); @@ -358,6 +361,8 @@ static void mgetCommand(redisClient *c); static void monitorCommand(redisClient *c); static void expireCommand(redisClient *c); static void getSetCommand(redisClient *c); +static void ttlCommand(redisClient *c); +static void slaveofCommand(redisClient *c); /*================================= Globals ================================= */ @@ -367,7 +372,7 @@ static struct redisCommand cmdTable[] = { {"get",getCommand,2,REDIS_CMD_INLINE}, {"set",setCommand,3,REDIS_CMD_BULK}, {"setnx",setnxCommand,3,REDIS_CMD_BULK}, - {"del",delCommand,2,REDIS_CMD_INLINE}, + {"del",delCommand,-2,REDIS_CMD_INLINE}, {"exists",existsCommand,2,REDIS_CMD_INLINE}, {"incr",incrCommand,2,REDIS_CMD_INLINE}, {"decr",decrCommand,2,REDIS_CMD_INLINE}, @@ -391,6 +396,8 @@ static struct redisCommand cmdTable[] = { {"sinterstore",sinterstoreCommand,-3,REDIS_CMD_INLINE}, {"sunion",sunionCommand,-2,REDIS_CMD_INLINE}, {"sunionstore",sunionstoreCommand,-3,REDIS_CMD_INLINE}, + {"sdiff",sdiffCommand,-2,REDIS_CMD_INLINE}, + {"sdiffstore",sdiffstoreCommand,-3,REDIS_CMD_INLINE}, {"smembers",sinterCommand,2,REDIS_CMD_INLINE}, {"incrby",incrbyCommand,3,REDIS_CMD_INLINE}, {"decrby",decrbyCommand,3,REDIS_CMD_INLINE}, @@ -400,6 +407,7 @@ static struct redisCommand cmdTable[] = { {"move",moveCommand,3,REDIS_CMD_INLINE}, {"rename",renameCommand,3,REDIS_CMD_INLINE}, {"renamenx",renamenxCommand,3,REDIS_CMD_INLINE}, + {"expire",expireCommand,3,REDIS_CMD_INLINE}, {"keys",keysCommand,2,REDIS_CMD_INLINE}, {"dbsize",dbsizeCommand,1,REDIS_CMD_INLINE}, {"auth",authCommand,2,REDIS_CMD_INLINE}, @@ -416,7 +424,8 @@ static struct redisCommand cmdTable[] = { {"sort",sortCommand,-2,REDIS_CMD_INLINE}, {"info",infoCommand,1,REDIS_CMD_INLINE}, {"monitor",monitorCommand,1,REDIS_CMD_INLINE}, - {"expire",expireCommand,3,REDIS_CMD_INLINE}, + {"ttl",ttlCommand,2,REDIS_CMD_INLINE}, + {"slaveof",slaveofCommand,3,REDIS_CMD_INLINE}, {NULL,NULL,0,0} }; @@ -556,7 +565,12 @@ void redisLog(int level, const char *fmt, ...) va_start(ap, fmt); if (level >= server.verbosity) { char *c = ".-*"; - fprintf(fp,"%c ",c[level]); + char buf[64]; + time_t now; + + now = time(NULL); + strftime(buf,64,"%d %b %H:%M:%S",gmtime(&now)); + fprintf(fp,"%s %c ",buf,c[level]); vfprintf(fp, fmt, ap); fprintf(fp,"\n"); fflush(fp); @@ -645,6 +659,7 @@ void closeTimedoutClients(void) { while ((ln = listYield(server.clients)) != NULL) { c = listNodeValue(ln); if (!(c->flags & REDIS_SLAVE) && /* no timeout for slaves */ + !(c->flags & REDIS_MASTER) && /* no timeout for masters */ (now - c->lastinteraction > server.maxidletime)) { redisLog(REDIS_DEBUG,"Closing idle client"); freeClient(c); @@ -711,7 +726,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { } /* Close connections of timedout clients */ - if (!(loops % 10)) + if (server.maxidletime && !(loops % 10)) closeTimedoutClients(); /* Check if a background saving in progress terminated */ @@ -950,7 +965,7 @@ static void loadServerConfig(char *filename) { /* Execute config directives */ if (!strcasecmp(argv[0],"timeout") && argc == 2) { server.maxidletime = atoi(argv[1]); - if (server.maxidletime < 1) { + if (server.maxidletime < 0) { err = "Invalid timeout value"; goto loaderr; } } else if (!strcasecmp(argv[0],"port") && argc == 2) { @@ -1107,7 +1122,8 @@ static void glueReplyBuffersIfNeeded(redisClient *c) { listDelNode(c->reply,ln); } /* Now the output buffer is empty, add the new single element */ - addReplySds(c,sdsnewlen(buf,totlen)); + o = createObject(REDIS_STRING,sdsnewlen(buf,totlen)); + if (!listAddNodeTail(c->reply,o)) oom("listAddNodeTail"); } } @@ -2012,6 +2028,7 @@ static robj *rdbLoadLzfStringObject(FILE*fp, int rdbver) { if ((val = sdsnewlen(NULL,len)) == NULL) goto err; if (fread(c,clen,1,fp) == 0) goto err; if (lzf_decompress(c,clen,val,len) == 0) goto err; + zfree(c); return createObject(REDIS_STRING,val); err: zfree(c); @@ -2305,11 +2322,24 @@ static void decrbyCommand(redisClient *c) { /* ========================= Type agnostic commands ========================= */ static void delCommand(redisClient *c) { - if (deleteKey(c->db,c->argv[1])) { - server.dirty++; - addReply(c,shared.cone); - } else { + int deleted = 0, j; + + for (j = 1; j < c->argc; j++) { + if (deleteKey(c->db,c->argv[j])) { + server.dirty++; + deleted++; + } + } + switch(deleted) { + case 0: addReply(c,shared.czero); + break; + case 1: + addReply(c,shared.cone); + break; + default: + addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",deleted)); + break; } } @@ -2949,7 +2979,7 @@ static void sinterGenericCommand(redisClient *c, robj **setskeys, int setsnum, r robj *lenobj = NULL, *dstset = NULL; int j, cardinality = 0; - if (!dv) oom("sinterCommand"); + if (!dv) oom("sinterGenericCommand"); for (j = 0; j < setsnum; j++) { robj *setobj; @@ -2990,9 +3020,6 @@ static void sinterGenericCommand(redisClient *c, robj **setskeys, int setsnum, r /* If we have a target key where to store the resulting set * create this key with an empty set inside */ dstset = createSetObject(); - deleteKey(c->db,dstkey); - dictAdd(c->db->dict,dstkey,dstset); - incrRefCount(dstkey); } /* Iterate all the elements of the first (smallest) set, and test @@ -3021,10 +3048,18 @@ static void sinterGenericCommand(redisClient *c, robj **setskeys, int setsnum, r } dictReleaseIterator(di); + if (dstkey) { + /* Store the resulting set into the target */ + deleteKey(c->db,dstkey); + dictAdd(c->db->dict,dstkey,dstset); + incrRefCount(dstkey); + } + if (!dstkey) { lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality); } else { - addReply(c,shared.ok); + addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n", + dictSize((dict*)dstset->ptr))); server.dirty++; } zfree(dv); @@ -3038,14 +3073,17 @@ static void sinterstoreCommand(redisClient *c) { sinterGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]); } -static void sunionGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey) { +#define REDIS_OP_UNION 0 +#define REDIS_OP_DIFF 1 + +static void sunionDiffGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey, int op) { dict **dv = zmalloc(sizeof(dict*)*setsnum); dictIterator *di; dictEntry *de; - robj *lenobj = NULL, *dstset = NULL; + robj *dstset = NULL; int j, cardinality = 0; - if (!dv) oom("sunionCommand"); + if (!dv) oom("sunionDiffGenericCommand"); for (j = 0; j < setsnum; j++) { robj *setobj; @@ -3069,27 +3107,10 @@ static void sunionGenericCommand(redisClient *c, robj **setskeys, int setsnum, r * this set object will be the resulting object to set into the target key*/ dstset = createSetObject(); - /* The first thing we should output is the total number of elements... - * since this is a multi-bulk write, but at this stage we don't know - * the intersection set size, so we use a trick, append an empty object - * to the output list and save the pointer to later modify it with the - * right length */ - if (!dstkey) { - lenobj = createObject(REDIS_STRING,NULL); - addReply(c,lenobj); - decrRefCount(lenobj); - } else { - /* If we have a target key where to store the resulting set - * create this key with an empty set inside */ - deleteKey(c->db,dstkey); - dictAdd(c->db->dict,dstkey,dstset); - incrRefCount(dstkey); - server.dirty++; - } - /* Iterate all the elements of all the sets, add every element a single * time to the result set */ for (j = 0; j < setsnum; j++) { + if (op == REDIS_OP_DIFF && j == 0 && !dv[j]) break; /* result set is empty */ if (!dv[j]) continue; /* non existing keys are like empty sets */ di = dictGetIterator(dv[j]); @@ -3100,36 +3121,70 @@ static void sunionGenericCommand(redisClient *c, robj **setskeys, int setsnum, r /* dictAdd will not add the same element multiple times */ ele = dictGetEntryKey(de); - if (dictAdd(dstset->ptr,ele,NULL) == DICT_OK) { - incrRefCount(ele); - if (!dstkey) { - addReplySds(c,sdscatprintf(sdsempty(), - "$%d\r\n",sdslen(ele->ptr))); - addReply(c,ele); - addReply(c,shared.crlf); + if (op == REDIS_OP_UNION || j == 0) { + if (dictAdd(dstset->ptr,ele,NULL) == DICT_OK) { + incrRefCount(ele); cardinality++; } + } else if (op == REDIS_OP_DIFF) { + if (dictDelete(dstset->ptr,ele) == DICT_OK) { + cardinality--; + } } } dictReleaseIterator(di); + + if (op == REDIS_OP_DIFF && cardinality == 0) break; /* result set is empty */ } + /* Output the content of the resulting set, if not in STORE mode */ + if (!dstkey) { + addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",cardinality)); + di = dictGetIterator(dstset->ptr); + if (!di) oom("dictGetIterator"); + while((de = dictNext(di)) != NULL) { + robj *ele; + + ele = dictGetEntryKey(de); + addReplySds(c,sdscatprintf(sdsempty(), + "$%d\r\n",sdslen(ele->ptr))); + addReply(c,ele); + addReply(c,shared.crlf); + } + dictReleaseIterator(di); + } else { + /* If we have a target key where to store the resulting set + * create this key with the result set inside */ + deleteKey(c->db,dstkey); + dictAdd(c->db->dict,dstkey,dstset); + incrRefCount(dstkey); + } + + /* Cleanup */ if (!dstkey) { - lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality); decrRefCount(dstset); } else { - addReply(c,shared.ok); + addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n", + dictSize((dict*)dstset->ptr))); server.dirty++; } zfree(dv); } static void sunionCommand(redisClient *c) { - sunionGenericCommand(c,c->argv+1,c->argc-1,NULL); + sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,REDIS_OP_UNION); } static void sunionstoreCommand(redisClient *c) { - sunionGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]); + sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],REDIS_OP_UNION); +} + +static void sdiffCommand(redisClient *c) { + sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,REDIS_OP_DIFF); +} + +static void sdiffstoreCommand(redisClient *c) { + sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],REDIS_OP_DIFF); } static void flushdbCommand(redisClient *c) { @@ -3379,7 +3434,10 @@ static void sortCommand(redisClient *c) { server.sort_desc = desc; server.sort_alpha = alpha; server.sort_bypattern = sortby ? 1 : 0; - qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare); + if (sortby && (start != 0 || end != vectorlen-1)) + pqsort(vector,vectorlen,sizeof(redisSortObject),sortCompare, start,end); + else + qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare); } /* Send command output to the output buffer, performing the specified @@ -3550,6 +3608,18 @@ static void expireCommand(redisClient *c) { } } +static void ttlCommand(redisClient *c) { + time_t expire; + int ttl = -1; + + expire = getExpire(c->db,c->argv[1]); + if (expire != -1) { + ttl = (int) (expire-time(NULL)); + if (ttl < 0) ttl = -1; + } + addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",ttl)); +} + /* =============================== Replication ============================= */ static int syncWrite(int fd, char *ptr, ssize_t size, int timeout) { @@ -3795,7 +3865,7 @@ static int syncWithMaster(void) { return REDIS_ERR; } /* Read the bulk write count */ - if (syncReadLine(fd,buf,1024,5) == -1) { + if (syncReadLine(fd,buf,1024,3600) == -1) { close(fd); redisLog(REDIS_WARNING,"I/O error reading bulk count from MASTER: %s", strerror(errno)); @@ -3850,6 +3920,28 @@ static int syncWithMaster(void) { return REDIS_OK; } +static void slaveofCommand(redisClient *c) { + if (!strcasecmp(c->argv[1]->ptr,"no") && + !strcasecmp(c->argv[2]->ptr,"one")) { + if (server.masterhost) { + sdsfree(server.masterhost); + server.masterhost = NULL; + if (server.master) freeClient(server.master); + server.replstate = REDIS_REPL_NONE; + redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)"); + } + } else { + sdsfree(server.masterhost); + server.masterhost = sdsdup(c->argv[1]->ptr); + server.masterport = atoi(c->argv[2]->ptr); + if (server.master) freeClient(server.master); + server.replstate = REDIS_REPL_CONNECT; + redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)", + server.masterhost, server.masterport); + } + addReply(c,shared.ok); +} + /* =================================== Main! ================================ */ #ifdef __linux__