client libraries synched in git

[redis.git] / redis.c

diff --git a/redis.c b/redis.c
index a3341f6d84161b916777f121d84de156affe7f91..98371ed408ac650815a0c5c23d9997a2fa008950 100644 (file)
--- 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 "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
 
 /* 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;
     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;
     /* 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 monitorCommand(redisClient *c);
 static void expireCommand(redisClient *c);
 static void getSetCommand(redisClient *c);

+static void ttlCommand(redisClient *c);
+static void slaveofCommand(redisClient *c);

 
 /*================================= Globals ================================= */
 
 
 /*================================= 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},
     {"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},
     {"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},
     {"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}
 };
 
     {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 */
     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);
              (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 */
     }
 
     /* Close connections of timedout clients */

-    if (!(loops % 10))
+    if (server.maxidletime && !(loops % 10))

         closeTimedoutClients();
 
     /* Check if a background saving in progress terminated */
         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.dbfilename = "dump.rdb";
     server.requirepass = NULL;
     server.shareobjects = 0;

+    server.maxclients = 0;

     ResetServerSaveParams();
 
     appendServerSaveParams(60*60,1);  /* save after 1 hour and 1 change */
     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]);
         /* 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) {
                 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;
             }
             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]);
         } 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 */
             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;
              * 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;
             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];
 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);
     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);
         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;
     }
         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++;
 }
 
     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;
 
     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;
 
     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();
         /* 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
     }
 
     /* 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);
 
     }
     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 {
     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);
         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();
 
      * 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++) {
     /* 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);
             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 {
     }
 
     /* 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);
         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;
         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
     }
 
     /* 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"
     
     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"
         "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"
         "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,
         ,REDIS_VERSION,

+        uptime,
+        uptime/(3600*24),

         listLength(server.clients)-listLength(server.slaves),
         listLength(server.slaves),
         server.usedmemory,
         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,
         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);
     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) {
 /* =============================== 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 */
         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));
         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;
 }
 
     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__
 /* =================================== 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 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();
     }
     initServer();
     if (server.daemonize) daemonize();