Automagically reconnection of RubyRedis

[redis.git] / redis.c

diff --git a/redis.c b/redis.c
index ae0e869e71547e5b1868b39a7ada58a3ecade08e..fbd1bc3df1807678cc297dcdc8ec3958f4061c84 100644 (file)
--- a/redis.c
+++ b/redis.c
@@ -27,7 +27,7 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
  * POSSIBILITY OF SUCH DAMAGE.
  */
 

-#define REDIS_VERSION "0.091"
+#define REDIS_VERSION "0.100"

 
 #include "fmacros.h"
 
 
 #include "fmacros.h"
 


@@ -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


@@ -67,7 +68,7 @@
 #define REDIS_MAXIDLETIME       (60*5)  /* default client timeout */
 #define REDIS_IOBUF_LEN         1024
 #define REDIS_LOADBUF_LEN       1024
 #define REDIS_MAXIDLETIME       (60*5)  /* default client timeout */
 #define REDIS_IOBUF_LEN         1024
 #define REDIS_LOADBUF_LEN       1024

-#define REDIS_MAX_ARGS          16
+#define REDIS_STATIC_ARGS       4

 #define REDIS_DEFAULT_DBNUM     16
 #define REDIS_CONFIGLINE_MAX    1024
 #define REDIS_OBJFREELIST_MAX   1000000 /* Max number of objects to cache */
 #define REDIS_DEFAULT_DBNUM     16
 #define REDIS_CONFIGLINE_MAX    1024
 #define REDIS_OBJFREELIST_MAX   1000000 /* Max number of objects to cache */


@@ -183,7 +184,7 @@ typedef struct redisClient {
     redisDb *db;
     int dictid;
     sds querybuf;
     redisDb *db;
     int dictid;
     sds querybuf;

-    robj *argv[REDIS_MAX_ARGS];
+    robj **argv;

     int argc;
     int bulklen;            /* bulk read len. -1 if not in bulk read mode */
     list *reply;
     int argc;
     int bulklen;            /* bulk read len. -1 if not in bulk read mode */
     list *reply;


@@ -341,12 +342,15 @@ static void typeCommand(redisClient *c);
 static void lsetCommand(redisClient *c);
 static void saddCommand(redisClient *c);
 static void sremCommand(redisClient *c);
 static void lsetCommand(redisClient *c);
 static void saddCommand(redisClient *c);
 static void sremCommand(redisClient *c);

+static void smoveCommand(redisClient *c);

 static void sismemberCommand(redisClient *c);
 static void scardCommand(redisClient *c);
 static void sinterCommand(redisClient *c);
 static void sinterstoreCommand(redisClient *c);
 static void sunionCommand(redisClient *c);
 static void sunionstoreCommand(redisClient *c);
 static void sismemberCommand(redisClient *c);
 static void scardCommand(redisClient *c);
 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);
 static void syncCommand(redisClient *c);
 static void flushdbCommand(redisClient *c);
 static void flushallCommand(redisClient *c);


@@ -356,6 +360,7 @@ static void infoCommand(redisClient *c);
 static void mgetCommand(redisClient *c);
 static void monitorCommand(redisClient *c);
 static void expireCommand(redisClient *c);
 static void mgetCommand(redisClient *c);
 static void monitorCommand(redisClient *c);
 static void expireCommand(redisClient *c);

+static void getSetCommand(redisClient *c);

 
 /*================================= Globals ================================= */
 
 
 /*================================= Globals ================================= */
 


@@ -365,7 +370,7 @@ static struct redisCommand cmdTable[] = {
     {"get",getCommand,2,REDIS_CMD_INLINE},
     {"set",setCommand,3,REDIS_CMD_BULK},
     {"setnx",setnxCommand,3,REDIS_CMD_BULK},
     {"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},
     {"exists",existsCommand,2,REDIS_CMD_INLINE},
     {"incr",incrCommand,2,REDIS_CMD_INLINE},
     {"decr",decrCommand,2,REDIS_CMD_INLINE},


@@ -382,15 +387,19 @@ static struct redisCommand cmdTable[] = {
     {"lrem",lremCommand,4,REDIS_CMD_BULK},
     {"sadd",saddCommand,3,REDIS_CMD_BULK},
     {"srem",sremCommand,3,REDIS_CMD_BULK},
     {"lrem",lremCommand,4,REDIS_CMD_BULK},
     {"sadd",saddCommand,3,REDIS_CMD_BULK},
     {"srem",sremCommand,3,REDIS_CMD_BULK},

+    {"smove",smoveCommand,4,REDIS_CMD_BULK},

     {"sismember",sismemberCommand,3,REDIS_CMD_BULK},
     {"scard",scardCommand,2,REDIS_CMD_INLINE},
     {"sinter",sinterCommand,-2,REDIS_CMD_INLINE},
     {"sinterstore",sinterstoreCommand,-3,REDIS_CMD_INLINE},
     {"sunion",sunionCommand,-2,REDIS_CMD_INLINE},
     {"sunionstore",sunionstoreCommand,-3,REDIS_CMD_INLINE},
     {"sismember",sismemberCommand,3,REDIS_CMD_BULK},
     {"scard",scardCommand,2,REDIS_CMD_INLINE},
     {"sinter",sinterCommand,-2,REDIS_CMD_INLINE},
     {"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},
     {"smembers",sinterCommand,2,REDIS_CMD_INLINE},
     {"incrby",incrbyCommand,3,REDIS_CMD_INLINE},
     {"decrby",decrbyCommand,3,REDIS_CMD_INLINE},

+    {"getset",getSetCommand,3,REDIS_CMD_BULK},

     {"randomkey",randomkeyCommand,1,REDIS_CMD_INLINE},
     {"select",selectCommand,2,REDIS_CMD_INLINE},
     {"move",moveCommand,3,REDIS_CMD_INLINE},
     {"randomkey",randomkeyCommand,1,REDIS_CMD_INLINE},
     {"select",selectCommand,2,REDIS_CMD_INLINE},
     {"move",moveCommand,3,REDIS_CMD_INLINE},


@@ -552,7 +561,12 @@ void redisLog(int level, const char *fmt, ...)
     va_start(ap, fmt);
     if (level >= server.verbosity) {
         char *c = ".-*";
     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);
         vfprintf(fp, fmt, ap);
         fprintf(fp,"\n");
         fflush(fp);


@@ -648,6 +662,25 @@ void closeTimedoutClients(void) {
     }
 }
 
     }
 }
 

+/* If the percentage of used slots in the HT reaches REDIS_HT_MINFILL
+ * we resize the hash table to save memory */
+void tryResizeHashTables(void) {
+    int j;
+
+    for (j = 0; j < server.dbnum; j++) {
+        long long size, used;
+
+        size = dictSlots(server.db[j].dict);
+        used = dictSize(server.db[j].dict);
+        if (size && used && size > REDIS_HT_MINSLOTS &&
+            (used*100/size < REDIS_HT_MINFILL)) {
+            redisLog(REDIS_NOTICE,"The hash table %d is too sparse, resize it...",j);
+            dictResize(server.db[j].dict);
+            redisLog(REDIS_NOTICE,"Hash table %d resized.",j);
+        }
+    }
+}
+

 int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     int j, loops = server.cronloops++;
     REDIS_NOTUSED(eventLoop);
 int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     int j, loops = server.cronloops++;
     REDIS_NOTUSED(eventLoop);


@@ -657,10 +690,9 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     /* Update the global state with the amount of used memory */
     server.usedmemory = zmalloc_used_memory();
 
     /* Update the global state with the amount of used memory */
     server.usedmemory = zmalloc_used_memory();
 

-    /* If the percentage of used slots in the HT reaches REDIS_HT_MINFILL
-     * we resize the hash table to save memory */
+    /* Show some info about non-empty databases */

     for (j = 0; j < server.dbnum; j++) {
     for (j = 0; j < server.dbnum; j++) {

-        int size, used, vkeys;
+        long long size, used, vkeys;

 
         size = dictSlots(server.db[j].dict);
         used = dictSize(server.db[j].dict);
 
         size = dictSlots(server.db[j].dict);
         used = dictSize(server.db[j].dict);


@@ -669,14 +701,16 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
             redisLog(REDIS_DEBUG,"DB %d: %d keys (%d volatile) in %d slots HT.",j,used,vkeys,size);
             /* dictPrintStats(server.dict); */
         }
             redisLog(REDIS_DEBUG,"DB %d: %d keys (%d volatile) in %d slots HT.",j,used,vkeys,size);
             /* dictPrintStats(server.dict); */
         }

-        if (size && used && size > REDIS_HT_MINSLOTS &&
-            (used*100/size < REDIS_HT_MINFILL)) {
-            redisLog(REDIS_NOTICE,"The hash table %d is too sparse, resize it...",j);
-            dictResize(server.db[j].dict);
-            redisLog(REDIS_NOTICE,"Hash table %d resized.",j);
-        }

     }
 
     }
 

+    /* We don't want to resize the hash tables while a bacground saving
+     * is in progress: the saving child is created using fork() that is
+     * implemented with a copy-on-write semantic in most modern systems, so
+     * if we resize the HT while there is the saving child at work actually
+     * a lot of memory movements in the parent will cause a lot of pages
+     * copied. */
+    if (!server.bgsaveinprogress) tryResizeHashTables();
+

     /* Show information about connected clients */
     if (!(loops % 5)) {
         redisLog(REDIS_DEBUG,"%d clients connected (%d slaves), %zu bytes in use",
     /* Show information about connected clients */
     if (!(loops % 5)) {
         redisLog(REDIS_DEBUG,"%d clients connected (%d slaves), %zu bytes in use",


@@ -687,7 +721,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 */


@@ -924,44 +958,44 @@ static void loadServerConfig(char *filename) {
         sdstolower(argv[0]);
 
         /* Execute config directives */
         sdstolower(argv[0]);
 
         /* Execute config directives */

-        if (!strcmp(argv[0],"timeout") && argc == 2) {
+        if (!strcasecmp(argv[0],"timeout") && argc == 2) {

             server.maxidletime = atoi(argv[1]);
             server.maxidletime = atoi(argv[1]);

-            if (server.maxidletime < 1) {
+            if (server.maxidletime < 0) {

                 err = "Invalid timeout value"; goto loaderr;
             }
                 err = "Invalid timeout value"; goto loaderr;
             }

-        } else if (!strcmp(argv[0],"port") && argc == 2) {
+        } else if (!strcasecmp(argv[0],"port") && argc == 2) {

             server.port = atoi(argv[1]);
             if (server.port < 1 || server.port > 65535) {
                 err = "Invalid port"; goto loaderr;
             }
             server.port = atoi(argv[1]);
             if (server.port < 1 || server.port > 65535) {
                 err = "Invalid port"; goto loaderr;
             }

-        } else if (!strcmp(argv[0],"bind") && argc == 2) {
+        } else if (!strcasecmp(argv[0],"bind") && argc == 2) {

             server.bindaddr = zstrdup(argv[1]);
             server.bindaddr = zstrdup(argv[1]);

-        } else if (!strcmp(argv[0],"save") && argc == 3) {
+        } else if (!strcasecmp(argv[0],"save") && argc == 3) {

             int seconds = atoi(argv[1]);
             int changes = atoi(argv[2]);
             if (seconds < 1 || changes < 0) {
                 err = "Invalid save parameters"; goto loaderr;
             }
             appendServerSaveParams(seconds,changes);
             int seconds = atoi(argv[1]);
             int changes = atoi(argv[2]);
             if (seconds < 1 || changes < 0) {
                 err = "Invalid save parameters"; goto loaderr;
             }
             appendServerSaveParams(seconds,changes);

-        } else if (!strcmp(argv[0],"dir") && argc == 2) {
+        } else if (!strcasecmp(argv[0],"dir") && argc == 2) {

             if (chdir(argv[1]) == -1) {
                 redisLog(REDIS_WARNING,"Can't chdir to '%s': %s",
                     argv[1], strerror(errno));
                 exit(1);
             }
             if (chdir(argv[1]) == -1) {
                 redisLog(REDIS_WARNING,"Can't chdir to '%s': %s",
                     argv[1], strerror(errno));
                 exit(1);
             }

-        } else if (!strcmp(argv[0],"loglevel") && argc == 2) {
-            if (!strcmp(argv[1],"debug")) server.verbosity = REDIS_DEBUG;
-            else if (!strcmp(argv[1],"notice")) server.verbosity = REDIS_NOTICE;
-            else if (!strcmp(argv[1],"warning")) server.verbosity = REDIS_WARNING;
+        } else if (!strcasecmp(argv[0],"loglevel") && argc == 2) {
+            if (!strcasecmp(argv[1],"debug")) server.verbosity = REDIS_DEBUG;
+            else if (!strcasecmp(argv[1],"notice")) server.verbosity = REDIS_NOTICE;
+            else if (!strcasecmp(argv[1],"warning")) server.verbosity = REDIS_WARNING;

             else {
                 err = "Invalid log level. Must be one of debug, notice, warning";
                 goto loaderr;
             }
             else {
                 err = "Invalid log level. Must be one of debug, notice, warning";
                 goto loaderr;
             }

-        } else if (!strcmp(argv[0],"logfile") && argc == 2) {
+        } else if (!strcasecmp(argv[0],"logfile") && argc == 2) {

             FILE *fp;
 
             server.logfile = zstrdup(argv[1]);
             FILE *fp;
 
             server.logfile = zstrdup(argv[1]);

-            if (!strcmp(server.logfile,"stdout")) {
+            if (!strcasecmp(server.logfile,"stdout")) {

                 zfree(server.logfile);
                 server.logfile = NULL;
             }
                 zfree(server.logfile);
                 server.logfile = NULL;
             }


@@ -976,31 +1010,33 @@ static void loadServerConfig(char *filename) {
                 }
                 fclose(fp);
             }
                 }
                 fclose(fp);
             }

-        } else if (!strcmp(argv[0],"databases") && argc == 2) {
+        } else if (!strcasecmp(argv[0],"databases") && argc == 2) {

             server.dbnum = atoi(argv[1]);
             if (server.dbnum < 1) {
                 err = "Invalid number of databases"; goto loaderr;
             }
             server.dbnum = atoi(argv[1]);
             if (server.dbnum < 1) {
                 err = "Invalid number of databases"; goto loaderr;
             }

-        } else if (!strcmp(argv[0],"slaveof") && argc == 3) {
+        } else if (!strcasecmp(argv[0],"slaveof") && argc == 3) {

             server.masterhost = sdsnew(argv[1]);
             server.masterport = atoi(argv[2]);
             server.replstate = REDIS_REPL_CONNECT;
             server.masterhost = sdsnew(argv[1]);
             server.masterport = atoi(argv[2]);
             server.replstate = REDIS_REPL_CONNECT;

-        } else if (!strcmp(argv[0],"glueoutputbuf") && argc == 2) {
+        } else if (!strcasecmp(argv[0],"glueoutputbuf") && argc == 2) {

             if ((server.glueoutputbuf = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }
             if ((server.glueoutputbuf = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }

-        } else if (!strcmp(argv[0],"shareobjects") && argc == 2) {
+        } else if (!strcasecmp(argv[0],"shareobjects") && argc == 2) {

             if ((server.shareobjects = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }
             if ((server.shareobjects = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }

-        } else if (!strcmp(argv[0],"daemonize") && argc == 2) {
+        } else if (!strcasecmp(argv[0],"daemonize") && argc == 2) {

             if ((server.daemonize = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }
             if ((server.daemonize = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }

-        } else if (!strcmp(argv[0],"requirepass") && argc == 2) {
+        } else if (!strcasecmp(argv[0],"requirepass") && argc == 2) {

           server.requirepass = zstrdup(argv[1]);
           server.requirepass = zstrdup(argv[1]);

-        } else if (!strcmp(argv[0],"pidfile") && argc == 2) {
+        } else if (!strcasecmp(argv[0],"pidfile") && argc == 2) {

           server.pidfile = zstrdup(argv[1]);
           server.pidfile = zstrdup(argv[1]);

+        } else if (!strcasecmp(argv[0],"dbfilename") && argc == 2) {
+          server.dbfilename = zstrdup(argv[1]);

         } else {
             err = "Bad directive or wrong number of arguments"; goto loaderr;
         }
         } else {
             err = "Bad directive or wrong number of arguments"; goto loaderr;
         }


@@ -1052,6 +1088,7 @@ static void freeClient(redisClient *c) {
         server.master = NULL;
         server.replstate = REDIS_REPL_CONNECT;
     }
         server.master = NULL;
         server.replstate = REDIS_REPL_CONNECT;
     }

+    zfree(c->argv);

     zfree(c);
 }
 
     zfree(c);
 }
 


@@ -1136,7 +1173,7 @@ static void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask)
 static struct redisCommand *lookupCommand(char *name) {
     int j = 0;
     while(cmdTable[j].name != NULL) {
 static struct redisCommand *lookupCommand(char *name) {
     int j = 0;
     while(cmdTable[j].name != NULL) {

-        if (!strcmp(name,cmdTable[j].name)) return &cmdTable[j];
+        if (!strcasecmp(name,cmdTable[j].name)) return &cmdTable[j];

         j++;
     }
     return NULL;
         j++;
     }
     return NULL;


@@ -1160,10 +1197,9 @@ static int processCommand(redisClient *c) {
     struct redisCommand *cmd;
     long long dirty;
 
     struct redisCommand *cmd;
     long long dirty;
 

-    sdstolower(c->argv[0]->ptr);

     /* The QUIT command is handled as a special case. Normal command
      * procs are unable to close the client connection safely */
     /* The QUIT command is handled as a special case. Normal command
      * procs are unable to close the client connection safely */

-    if (!strcmp(c->argv[0]->ptr,"quit")) {
+    if (!strcasecmp(c->argv[0]->ptr,"quit")) {

         freeClient(c);
         return 0;
     }
         freeClient(c);
         return 0;
     }


@@ -1232,8 +1268,17 @@ static int processCommand(redisClient *c) {
 
 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc) {
     listNode *ln;
 
 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc) {
     listNode *ln;

-    robj *outv[REDIS_MAX_ARGS*4]; /* enough room for args, spaces, newlines */

     int outc = 0, j;
     int outc = 0, j;

+    robj **outv;
+    /* (args*2)+1 is enough room for args, spaces, newlines */
+    robj *static_outv[REDIS_STATIC_ARGS*2+1];
+
+    if (argc <= REDIS_STATIC_ARGS) {
+        outv = static_outv;
+    } else {
+        outv = zmalloc(sizeof(robj*)*(argc*2+1));
+        if (!outv) oom("replicationFeedSlaves");
+    }

     
     for (j = 0; j < argc; j++) {
         if (j != 0) outv[outc++] = shared.space;
     
     for (j = 0; j < argc; j++) {
         if (j != 0) outv[outc++] = shared.space;


@@ -1287,6 +1332,7 @@ static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int di
         for (j = 0; j < outc; j++) addReply(slave,outv[j]);
     }
     for (j = 0; j < outc; j++) decrRefCount(outv[j]);
         for (j = 0; j < outc; j++) addReply(slave,outv[j]);
     }
     for (j = 0; j < outc; j++) decrRefCount(outv[j]);

+    if (outv != static_outv) zfree(outv);

 }
 
 static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
 }
 
 static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {


@@ -1344,9 +1390,14 @@ again:
                 return;
             }
             argv = sdssplitlen(query,sdslen(query)," ",1,&argc);
                 return;
             }
             argv = sdssplitlen(query,sdslen(query)," ",1,&argc);

-            sdsfree(query);

             if (argv == NULL) oom("sdssplitlen");
             if (argv == NULL) oom("sdssplitlen");

-            for (j = 0; j < argc && j < REDIS_MAX_ARGS; j++) {
+            sdsfree(query);
+
+            if (c->argv) zfree(c->argv);
+            c->argv = zmalloc(sizeof(robj*)*argc);
+            if (c->argv == NULL) oom("allocating arguments list for client");
+
+            for (j = 0; j < argc; j++) {

                 if (sdslen(argv[j])) {
                     c->argv[c->argc] = createObject(REDIS_STRING,argv[j]);
                     c->argc++;
                 if (sdslen(argv[j])) {
                     c->argv[c->argc] = createObject(REDIS_STRING,argv[j]);
                     c->argc++;


@@ -1405,6 +1456,7 @@ static redisClient *createClient(int fd) {
     c->fd = fd;
     c->querybuf = sdsempty();
     c->argc = 0;
     c->fd = fd;
     c->querybuf = sdsempty();
     c->argc = 0;

+    c->argv = NULL;

     c->bulklen = -1;
     c->sentlen = 0;
     c->flags = 0;
     c->bulklen = -1;
     c->sentlen = 0;
     c->flags = 0;


@@ -1699,7 +1751,7 @@ static int rdbSaveLzfStringObject(FILE *fp, robj *obj) {
     /* We require at least four bytes compression for this to be worth it */
     outlen = sdslen(obj->ptr)-4;
     if (outlen <= 0) return 0;
     /* We require at least four bytes compression for this to be worth it */
     outlen = sdslen(obj->ptr)-4;
     if (outlen <= 0) return 0;

-    if ((out = zmalloc(outlen)) == NULL) return 0;
+    if ((out = zmalloc(outlen+1)) == NULL) return 0;

     comprlen = lzf_compress(obj->ptr, sdslen(obj->ptr), out, outlen);
     if (comprlen == 0) {
         zfree(out);
     comprlen = lzf_compress(obj->ptr, sdslen(obj->ptr), out, outlen);
     if (comprlen == 0) {
         zfree(out);


@@ -1736,7 +1788,7 @@ static int rdbSaveStringObject(FILE *fp, robj *obj) {
 
     /* Try LZF compression - under 20 bytes it's unable to compress even
      * aaaaaaaaaaaaaaaaaa so skip it */
 
     /* Try LZF compression - under 20 bytes it's unable to compress even
      * aaaaaaaaaaaaaaaaaa so skip it */

-    if (len > 20) {
+    if (1 && len > 20) {

         int retval;
 
         retval = rdbSaveLzfStringObject(fp,obj);
         int retval;
 
         retval = rdbSaveLzfStringObject(fp,obj);


@@ -1874,6 +1926,11 @@ static int rdbSaveBackground(char *filename) {
         }
     } else {
         /* Parent */
         }
     } else {
         /* Parent */

+        if (childpid == -1) {
+            redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
+                strerror(errno));
+            return REDIS_ERR;
+        }

         redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
         server.bgsaveinprogress = 1;
         return REDIS_OK;
         redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
         server.bgsaveinprogress = 1;
         return REDIS_OK;


@@ -1965,6 +2022,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;
     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);
     return createObject(REDIS_STRING,val);
 err:
     zfree(c);


@@ -2172,6 +2230,18 @@ static void getCommand(redisClient *c) {
     }
 }
 
     }
 }
 

+static void getSetCommand(redisClient *c) {
+    getCommand(c);
+    if (dictAdd(c->db->dict,c->argv[1],c->argv[2]) == DICT_ERR) {
+        dictReplace(c->db->dict,c->argv[1],c->argv[2]);
+    } else {
+        incrRefCount(c->argv[1]);
+    }
+    incrRefCount(c->argv[2]);
+    server.dirty++;
+    removeExpire(c->db,c->argv[1]);
+}
+

 static void mgetCommand(redisClient *c) {
     int j;
   
 static void mgetCommand(redisClient *c) {
     int j;
   


@@ -2246,11 +2316,24 @@ static void decrbyCommand(redisClient *c) {
 /* ========================= Type agnostic commands ========================= */
 
 static void delCommand(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);
         addReply(c,shared.czero);

+        break;
+    case 1:
+        addReply(c,shared.cone);
+        break;
+    default:
+        addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",deleted));
+        break;

     }
 }
 
     }
 }
 


@@ -2375,8 +2458,9 @@ static void shutdownCommand(redisClient *c) {
     /* XXX: TODO kill the child if there is a bgsave in progress */
     if (rdbSave(server.dbfilename) == REDIS_OK) {
         if (server.daemonize) {
     /* XXX: TODO kill the child if there is a bgsave in progress */
     if (rdbSave(server.dbfilename) == REDIS_OK) {
         if (server.daemonize) {

-          unlink(server.pidfile);
+            unlink(server.pidfile);

         }
         }

+        redisLog(REDIS_WARNING,"%zu bytes used at exit",zmalloc_used_memory());

         redisLog(REDIS_WARNING,"Server exit now, bye bye...");
         exit(1);
     } else {
         redisLog(REDIS_WARNING,"Server exit now, bye bye...");
         exit(1);
     } else {


@@ -2804,6 +2888,41 @@ static void sremCommand(redisClient *c) {
     }
 }
 
     }
 }
 

+static void smoveCommand(redisClient *c) {
+    robj *srcset, *dstset;
+
+    srcset = lookupKeyWrite(c->db,c->argv[1]);
+    dstset = lookupKeyWrite(c->db,c->argv[2]);
+
+    /* If the source key does not exist return 0, if it's of the wrong type
+     * raise an error */
+    if (srcset == NULL || srcset->type != REDIS_SET) {
+        addReply(c, srcset ? shared.wrongtypeerr : shared.czero);
+        return;
+    }
+    /* Error if the destination key is not a set as well */
+    if (dstset && dstset->type != REDIS_SET) {
+        addReply(c,shared.wrongtypeerr);
+        return;
+    }
+    /* Remove the element from the source set */
+    if (dictDelete(srcset->ptr,c->argv[3]) == DICT_ERR) {
+        /* Key not found in the src set! return zero */
+        addReply(c,shared.czero);
+        return;
+    }
+    server.dirty++;
+    /* Add the element to the destination set */
+    if (!dstset) {
+        dstset = createSetObject();
+        dictAdd(c->db->dict,c->argv[2],dstset);
+        incrRefCount(c->argv[2]);
+    }
+    if (dictAdd(dstset->ptr,c->argv[3],NULL) == DICT_OK)
+        incrRefCount(c->argv[3]);
+    addReply(c,shared.cone);
+}
+

 static void sismemberCommand(redisClient *c) {
     robj *set;
 
 static void sismemberCommand(redisClient *c) {
     robj *set;
 


@@ -2895,9 +3014,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


@@ -2926,6 +3042,13 @@ 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 {


@@ -2943,14 +3066,17 @@ static void sinterstoreCommand(redisClient *c) {
     sinterGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
 }
 
     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;
     dict **dv = zmalloc(sizeof(dict*)*setsnum);
     dictIterator *di;
     dictEntry *de;

-    robj *lenobj = NULL, *dstset = NULL;
+    robj *dstset = NULL;

     int j, cardinality = 0;
 
     int j, cardinality = 0;
 

-    if (!dv) oom("sunionCommand");
+    if (!dv) oom("sunionDiffGenericCommand");

     for (j = 0; j < setsnum; j++) {
         robj *setobj;
 
     for (j = 0; j < setsnum; j++) {
         robj *setobj;
 


@@ -2974,27 +3100,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();
 
      * 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++) {
     /* 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]);
         if (!dv[j]) continue; /* non existing keys are like empty sets */
 
         di = dictGetIterator(dv[j]);


@@ -3005,22 +3114,48 @@ static void sunionGenericCommand(redisClient *c, robj **setskeys, int setsnum, r
 
             /* dictAdd will not add the same element multiple times */
             ele = dictGetEntryKey(de);
 
             /* 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++;
                 }
                     cardinality++;
                 }

+            } else if (op == REDIS_OP_DIFF) {
+                if (dictDelete(dstset->ptr,ele) == DICT_OK) {
+                    cardinality--;
+                }

             }
         }
         dictReleaseIterator(di);
             }
         }
         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);
+        server.dirty++;
+    }
+
+    /* Cleanup */

     if (!dstkey) {
     if (!dstkey) {

-        lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality);

         decrRefCount(dstset);
     } else {
         addReply(c,shared.ok);
         decrRefCount(dstset);
     } else {
         addReply(c,shared.ok);


@@ -3030,11 +3165,19 @@ static void sunionGenericCommand(redisClient *c, robj **setskeys, int setsnum, r
 }
 
 static void sunionCommand(redisClient *c) {
 }
 
 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) {
 }
 
 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) {
 }
 
 static void flushdbCommand(redisClient *c) {


@@ -3284,7 +3427,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


@@ -3757,6 +3903,28 @@ static int syncWithMaster(void) {
 
 /* =================================== Main! ================================ */
 
 
 /* =================================== Main! ================================ */
 

+#ifdef __linux__
+int linuxOvercommitMemoryValue(void) {
+    FILE *fp = fopen("/proc/sys/vm/overcommit_memory","r");
+    char buf[64];
+
+    if (!fp) return -1;
+    if (fgets(buf,64,fp) == NULL) {
+        fclose(fp);
+        return -1;
+    }
+    fclose(fp);
+
+    return atoi(buf);
+}
+
+void linuxOvercommitMemoryWarning(void) {
+    if (linuxOvercommitMemoryValue() == 0) {
+        redisLog(REDIS_WARNING,"WARNING overcommit_memory is set to 0! Background save may fail under low condition memory. To fix this issue add 'echo 1 > /proc/sys/vm/overcommit_memory' in your init scripts.");
+    }
+}
+#endif /* __linux__ */
+

 static void daemonize(void) {
     int fd;
     FILE *fp;
 static void daemonize(void) {
     int fd;
     FILE *fp;


@@ -3782,6 +3950,10 @@ static void daemonize(void) {
 }
 
 int main(int argc, char **argv) {
 }
 
 int main(int argc, char **argv) {

+#ifdef __linux__
+    linuxOvercommitMemoryWarning();
+#endif
+

     initServerConfig();
     if (argc == 2) {
         ResetServerSaveParams();
     initServerConfig();
     if (argc == 2) {
         ResetServerSaveParams();