X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/51829ed3f09bb4381f075900521bcaafee8819bb..d7fc9edb182d361ac5c31b21fcdd25345c972898:/redis.c diff --git a/redis.c b/redis.c index a3341f6d..98371ed4 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 @@ -244,6 +245,7 @@ struct redisServer { int masterport; redisClient *master; /* client that is master for this slave */ int replstate; + unsigned int maxclients; /* Sort parameters - qsort_r() is only available under BSD so we * have to take this state global, in order to pass it to sortCompare() */ int sort_desc; @@ -360,6 +362,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 ================================= */ @@ -404,6 +408,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}, @@ -420,7 +425,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} }; @@ -654,6 +660,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); @@ -720,7 +727,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 */ @@ -856,6 +863,7 @@ static void initServerConfig() { server.dbfilename = "dump.rdb"; server.requirepass = NULL; server.shareobjects = 0; + server.maxclients = 0; ResetServerSaveParams(); appendServerSaveParams(60*60,1); /* save after 1 hour and 1 change */ @@ -959,7 +967,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) { @@ -1014,6 +1022,8 @@ static void loadServerConfig(char *filename) { if (server.dbnum < 1) { err = "Invalid number of databases"; goto loaderr; } + } else if (!strcasecmp(argv[0],"maxclients") && argc == 2) { + server.maxclients = atoi(argv[1]); } else if (!strcasecmp(argv[0],"slaveof") && argc == 3) { server.masterhost = sdsnew(argv[1]); server.masterport = atoi(argv[2]); @@ -1116,7 +1126,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"); } } @@ -1410,7 +1421,7 @@ again: * on the query buffer try to process the next command. */ if (processCommand(c) && sdslen(c->querybuf)) goto again; return; - } else if (sdslen(c->querybuf) >= 1024) { + } else if (sdslen(c->querybuf) >= 1024*32) { redisLog(REDIS_DEBUG, "Client protocol error"); freeClient(c); return; @@ -1493,6 +1504,7 @@ static void addReplySds(redisClient *c, sds s) { static void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) { int cport, cfd; char cip[128]; + redisClient *c; REDIS_NOTUSED(el); REDIS_NOTUSED(mask); REDIS_NOTUSED(privdata); @@ -1503,11 +1515,23 @@ static void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) { return; } redisLog(REDIS_DEBUG,"Accepted %s:%d", cip, cport); - if (createClient(cfd) == NULL) { + if ((c = createClient(cfd)) == NULL) { redisLog(REDIS_WARNING,"Error allocating resoures for the client"); close(cfd); /* May be already closed, just ingore errors */ return; } + /* If maxclient directive is set and this is one client more... close the + * connection. Note that we create the client instead to check before + * for this condition, since now the socket is already set in nonblocking + * mode and we can send an error for free using the Kernel I/O */ + if (server.maxclients && listLength(server.clients) > server.maxclients) { + char *err = "-ERR max number of clients reached\r\n"; + + /* That's a best effort error message, don't check write errors */ + (void) write(c->fd,err,strlen(err)); + freeClient(c); + return; + } server.stat_numconnections++; } @@ -2972,7 +2996,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; @@ -3013,9 +3037,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 @@ -3044,10 +3065,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); @@ -3095,20 +3124,6 @@ static void sunionDiffGenericCommand(redisClient *c, robj **setskeys, int setsnu * 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) { - /* 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++) { @@ -3154,13 +3169,20 @@ static void sunionDiffGenericCommand(redisClient *c, robj **setskeys, int setsnu 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) { decrRefCount(dstset); } else { - addReply(c,shared.ok); + addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n", + dictSize((dict*)dstset->ptr))); server.dirty++; } zfree(dv); @@ -3429,7 +3451,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 @@ -3481,6 +3506,8 @@ static void infoCommand(redisClient *c) { info = sdscatprintf(sdsempty(), "redis_version:%s\r\n" + "uptime_in_seconds:%d\r\n" + "uptime_in_days:%d\r\n" "connected_clients:%d\r\n" "connected_slaves:%d\r\n" "used_memory:%zu\r\n" @@ -3489,9 +3516,10 @@ static void infoCommand(redisClient *c) { "last_save_time:%d\r\n" "total_connections_received:%lld\r\n" "total_commands_processed:%lld\r\n" - "uptime_in_seconds:%d\r\n" - "uptime_in_days:%d\r\n" + "role:%s\r\n" ,REDIS_VERSION, + uptime, + uptime/(3600*24), listLength(server.clients)-listLength(server.slaves), listLength(server.slaves), server.usedmemory, @@ -3500,9 +3528,21 @@ static void infoCommand(redisClient *c) { server.lastsave, server.stat_numconnections, server.stat_numcommands, - uptime, - uptime/(3600*24) + server.masterhost == NULL ? "master" : "slave" ); + if (server.masterhost) { + info = sdscatprintf(info, + "master_host:%s\r\n" + "master_port:%d\r\n" + "master_link_status:%s\r\n" + "master_last_io_seconds_ago:%d\r\n" + ,server.masterhost, + server.masterport, + (server.replstate == REDIS_REPL_CONNECTED) ? + "up" : "down", + (int)(time(NULL)-server.master->lastinteraction) + ); + } addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(info))); addReplySds(c,info); addReply(c,shared.crlf); @@ -3600,6 +3640,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) { @@ -3845,7 +3897,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)); @@ -3900,6 +3952,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__ @@ -3960,6 +4034,8 @@ int main(int argc, char **argv) { } else if (argc > 2) { fprintf(stderr,"Usage: ./redis-server [/path/to/redis.conf]\n"); exit(1); + } else { + redisLog(REDIS_WARNING,"Warning: no config file specified, using the default config. In order to specify a config file use 'redis-server /path/to/redis.conf'"); } initServer(); if (server.daemonize) daemonize();