Merge branch 'vm-speedup'

[redis.git] / redis.c

diff --git a/redis.c b/redis.c
index 05f1c64a4f2bab232315fc81ca25c5ce26c000f0..90fe6f95b82f45f0104eec866474b77db3d7d5f9 100644 (file)
--- a/redis.c
+++ b/redis.c
@@ -370,6 +370,7 @@ struct redisServer {
     pid_t bgsavechildpid;
     pid_t bgrewritechildpid;
     sds bgrewritebuf; /* buffer taken by parent during oppend only rewrite */
     pid_t bgsavechildpid;
     pid_t bgrewritechildpid;
     sds bgrewritebuf; /* buffer taken by parent during oppend only rewrite */

+    sds aofbuf;       /* AOF buffer, written before entering the event loop */

     struct saveparam *saveparams;
     int saveparamslen;
     char *logfile;
     struct saveparam *saveparams;
     int saveparamslen;
     char *logfile;


@@ -557,6 +558,7 @@ static robj *createStringObject(char *ptr, size_t len);
 static robj *dupStringObject(robj *o);
 static void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc);
 static void replicationFeedMonitors(list *monitors, int dictid, robj **argv, int argc);
 static robj *dupStringObject(robj *o);
 static void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc);
 static void replicationFeedMonitors(list *monitors, int dictid, robj **argv, int argc);

+static void flushAppendOnlyFile(void);

 static void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc);
 static int syncWithMaster(void);
 static robj *tryObjectEncoding(robj *o);
 static void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc);
 static int syncWithMaster(void);
 static robj *tryObjectEncoding(robj *o);


@@ -620,8 +622,10 @@ static int pubsubUnsubscribeAllPatterns(redisClient *c, int notify);
 static void freePubsubPattern(void *p);
 static int listMatchPubsubPattern(void *a, void *b);
 static int compareStringObjects(robj *a, robj *b);
 static void freePubsubPattern(void *p);
 static int listMatchPubsubPattern(void *a, void *b);
 static int compareStringObjects(robj *a, robj *b);

+static int equalStringObjects(robj *a, robj *b);

 static void usage();
 static int rewriteAppendOnlyFileBackground(void);
 static void usage();
 static int rewriteAppendOnlyFileBackground(void);

+static int vmSwapObjectBlocking(robj *key, robj *val);

 
 static void authCommand(redisClient *c);
 static void pingCommand(redisClient *c);
 
 static void authCommand(redisClient *c);
 static void pingCommand(redisClient *c);


@@ -1019,6 +1023,30 @@ static long long memtoll(const char *p, int *err) {
     return val*mul;
 }
 
     return val*mul;
 }
 

+/* Convert a long long into a string. Returns the number of
+ * characters needed to represent the number, that can be shorter if passed
+ * buffer length is not enough to store the whole number. */
+static int ll2string(char *s, size_t len, long long value) {
+    char buf[32], *p;
+    unsigned long long v;
+    size_t l;
+
+    if (len == 0) return 0;
+    v = (value < 0) ? -value : value;
+    p = buf+31; /* point to the last character */
+    do {
+        *p-- = '0'+(v%10);
+        v /= 10;
+    } while(v);
+    if (value < 0) *p-- = '-';
+    p++;
+    l = 32-(p-buf);
+    if (l+1 > len) l = len-1; /* Make sure it fits, including the nul term */
+    memcpy(s,p,l);
+    s[l] = '\0';
+    return l;
+}
+

 static void redisLog(int level, const char *fmt, ...) {
     va_list ap;
     FILE *fp;
 static void redisLog(int level, const char *fmt, ...) {
     va_list ap;
     FILE *fp;


@@ -1101,8 +1129,8 @@ static int dictEncObjKeyCompare(void *privdata, const void *key1,
     int cmp;
 
     if (o1->encoding == REDIS_ENCODING_INT &&
     int cmp;
 
     if (o1->encoding == REDIS_ENCODING_INT &&

-        o2->encoding == REDIS_ENCODING_INT &&
-        o1->ptr == o2->ptr) return 1;
+        o2->encoding == REDIS_ENCODING_INT)
+            return o1->ptr == o2->ptr;

 
     o1 = getDecodedObject(o1);
     o2 = getDecodedObject(o2);
 
     o1 = getDecodedObject(o1);
     o2 = getDecodedObject(o2);


@@ -1122,7 +1150,7 @@ static unsigned int dictEncObjHash(const void *key) {
             char buf[32];
             int len;
 
             char buf[32];
             int len;
 

-            len = snprintf(buf,32,"%ld",(long)o->ptr);
+            len = ll2string(buf,32,(long)o->ptr);

             return dictGenHashFunction((unsigned char*)buf, len);
         } else {
             unsigned int hash;
             return dictGenHashFunction((unsigned char*)buf, len);
         } else {
             unsigned int hash;


@@ -1528,6 +1556,7 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
 static void beforeSleep(struct aeEventLoop *eventLoop) {
     REDIS_NOTUSED(eventLoop);
 
 static void beforeSleep(struct aeEventLoop *eventLoop) {
     REDIS_NOTUSED(eventLoop);
 

+    /* Awake clients that got all the swapped keys they requested */

     if (server.vm_enabled && listLength(server.io_ready_clients)) {
         listIter li;
         listNode *ln;
     if (server.vm_enabled && listLength(server.io_ready_clients)) {
         listIter li;
         listNode *ln;


@@ -1552,6 +1581,8 @@ static void beforeSleep(struct aeEventLoop *eventLoop) {
                 processInputBuffer(c);
         }
     }
                 processInputBuffer(c);
         }
     }

+    /* Write the AOF buffer on disk */
+    flushAppendOnlyFile();

 }
 
 static void createSharedObjects(void) {
 }
 
 static void createSharedObjects(void) {


@@ -1629,7 +1660,7 @@ static void initServerConfig() {
     server.glueoutputbuf = 1;
     server.daemonize = 0;
     server.appendonly = 0;
     server.glueoutputbuf = 1;
     server.daemonize = 0;
     server.appendonly = 0;

-    server.appendfsync = APPENDFSYNC_ALWAYS;
+    server.appendfsync = APPENDFSYNC_EVERYSEC;

     server.lastfsync = time(NULL);
     server.appendfd = -1;
     server.appendseldb = -1; /* Make sure the first time will not match */
     server.lastfsync = time(NULL);
     server.appendfd = -1;
     server.appendseldb = -1; /* Make sure the first time will not match */


@@ -1712,6 +1743,7 @@ static void initServer() {
     server.bgsavechildpid = -1;
     server.bgrewritechildpid = -1;
     server.bgrewritebuf = sdsempty();
     server.bgsavechildpid = -1;
     server.bgrewritechildpid = -1;
     server.bgrewritebuf = sdsempty();

+    server.aofbuf = sdsempty();

     server.lastsave = time(NULL);
     server.dirty = 0;
     server.stat_numcommands = 0;
     server.lastsave = time(NULL);
     server.dirty = 0;
     server.stat_numcommands = 0;


@@ -2649,7 +2681,7 @@ static void *dupClientReplyValue(void *o) {
 }
 
 static int listMatchObjects(void *a, void *b) {
 }
 
 static int listMatchObjects(void *a, void *b) {

-    return compareStringObjects(a,b) == 0;
+    return equalStringObjects(a,b);

 }
 
 static redisClient *createClient(int fd) {
 }
 
 static redisClient *createClient(int fd) {


@@ -2889,7 +2921,7 @@ static robj *createStringObjectFromLongLong(long long value) {
             o->encoding = REDIS_ENCODING_INT;
             o->ptr = (void*)((long)value);
         } else {
             o->encoding = REDIS_ENCODING_INT;
             o->ptr = (void*)((long)value);
         } else {

-            o->ptr = sdscatprintf(sdsempty(),"%lld",value);
+            o = createObject(REDIS_STRING,sdsfromlonglong(value));

         }
     }
     return o;
         }
     }
     return o;


@@ -3104,7 +3136,7 @@ static int isStringRepresentableAsLong(sds s, long *longval) {
 
     value = strtol(s, &endptr, 10);
     if (endptr[0] != '\0') return REDIS_ERR;
 
     value = strtol(s, &endptr, 10);
     if (endptr[0] != '\0') return REDIS_ERR;

-    slen = snprintf(buf,32,"%ld",value);
+    slen = ll2string(buf,32,value);

 
     /* If the number converted back into a string is not identical
      * then it's not possible to encode the string as integer */
 
     /* If the number converted back into a string is not identical
      * then it's not possible to encode the string as integer */


@@ -3157,7 +3189,7 @@ static robj *getDecodedObject(robj *o) {
     if (o->type == REDIS_STRING && o->encoding == REDIS_ENCODING_INT) {
         char buf[32];
 
     if (o->type == REDIS_STRING && o->encoding == REDIS_ENCODING_INT) {
         char buf[32];
 

-        snprintf(buf,32,"%ld",(long)o->ptr);
+        ll2string(buf,32,(long)o->ptr);

         dec = createStringObject(buf,strlen(buf));
         return dec;
     } else {
         dec = createStringObject(buf,strlen(buf));
         return dec;
     } else {


@@ -3167,7 +3199,7 @@ static robj *getDecodedObject(robj *o) {
 
 /* Compare two string objects via strcmp() or alike.
  * Note that the objects may be integer-encoded. In such a case we
 
 /* Compare two string objects via strcmp() or alike.
  * Note that the objects may be integer-encoded. In such a case we

- * use snprintf() to get a string representation of the numbers on the stack
+ * use ll2string() to get a string representation of the numbers on the stack

  * and compare the strings, it's much faster than calling getDecodedObject().
  *
  * Important note: if objects are not integer encoded, but binary-safe strings,
  * and compare the strings, it's much faster than calling getDecodedObject().
  *
  * Important note: if objects are not integer encoded, but binary-safe strings,


@@ -3180,14 +3212,14 @@ static int compareStringObjects(robj *a, robj *b) {
 
     if (a == b) return 0;
     if (a->encoding != REDIS_ENCODING_RAW) {
 
     if (a == b) return 0;
     if (a->encoding != REDIS_ENCODING_RAW) {

-        snprintf(bufa,sizeof(bufa),"%ld",(long) a->ptr);
+        ll2string(bufa,sizeof(bufa),(long) a->ptr);

         astr = bufa;
         bothsds = 0;
     } else {
         astr = a->ptr;
     }
     if (b->encoding != REDIS_ENCODING_RAW) {
         astr = bufa;
         bothsds = 0;
     } else {
         astr = a->ptr;
     }
     if (b->encoding != REDIS_ENCODING_RAW) {

-        snprintf(bufb,sizeof(bufb),"%ld",(long) b->ptr);
+        ll2string(bufb,sizeof(bufb),(long) b->ptr);

         bstr = bufb;
         bothsds = 0;
     } else {
         bstr = bufb;
         bothsds = 0;
     } else {


@@ -3196,6 +3228,18 @@ static int compareStringObjects(robj *a, robj *b) {
     return bothsds ? sdscmp(astr,bstr) : strcmp(astr,bstr);
 }
 
     return bothsds ? sdscmp(astr,bstr) : strcmp(astr,bstr);
 }
 

+/* Equal string objects return 1 if the two objects are the same from the
+ * point of view of a string comparison, otherwise 0 is returned. Note that
+ * this function is faster then checking for (compareStringObject(a,b) == 0)
+ * because it can perform some more optimization. */
+static int equalStringObjects(robj *a, robj *b) {
+    if (a->encoding != REDIS_ENCODING_RAW && b->encoding != REDIS_ENCODING_RAW){
+        return a->ptr == b->ptr;
+    } else {
+        return compareStringObjects(a,b) == 0;
+    }
+}
+

 static size_t stringObjectLen(robj *o) {
     redisAssert(o->type == REDIS_STRING);
     if (o->encoding == REDIS_ENCODING_RAW) {
 static size_t stringObjectLen(robj *o) {
     redisAssert(o->type == REDIS_STRING);
     if (o->encoding == REDIS_ENCODING_RAW) {


@@ -3203,7 +3247,7 @@ static size_t stringObjectLen(robj *o) {
     } else {
         char buf[32];
 
     } else {
         char buf[32];
 

-        return snprintf(buf,32,"%ld",(long)o->ptr);
+        return ll2string(buf,32,(long)o->ptr);

     }
 }
 
     }
 }
 


@@ -3221,7 +3265,7 @@ static int getDoubleFromObject(robj *o, double *target) {
         } else if (o->encoding == REDIS_ENCODING_INT) {
             value = (long)o->ptr;
         } else {
         } else if (o->encoding == REDIS_ENCODING_INT) {
             value = (long)o->ptr;
         } else {

-            redisAssert(1 != 1);
+            redisPanic("Unknown string encoding");

         }
     }
 
         }
     }
 


@@ -3258,7 +3302,7 @@ static int getLongLongFromObject(robj *o, long long *target) {
         } else if (o->encoding == REDIS_ENCODING_INT) {
             value = (long)o->ptr;
         } else {
         } else if (o->encoding == REDIS_ENCODING_INT) {
             value = (long)o->ptr;
         } else {

-            redisAssert(1 != 1);
+            redisPanic("Unknown string encoding");

         }
     }
 
         }
     }
 


@@ -3344,7 +3388,7 @@ static int rdbTryIntegerEncoding(char *s, size_t len, unsigned char *enc) {
     /* Check if it's possible to encode this value as a number */
     value = strtoll(s, &endptr, 10);
     if (endptr[0] != '\0') return 0;
     /* Check if it's possible to encode this value as a number */
     value = strtoll(s, &endptr, 10);
     if (endptr[0] != '\0') return 0;

-    snprintf(buf,32,"%lld",value);
+    ll2string(buf,32,value);

 
     /* If the number converted back into a string is not identical
      * then it's not possible to encode the string as integer */
 
     /* If the number converted back into a string is not identical
      * then it's not possible to encode the string as integer */


@@ -3753,7 +3797,11 @@ static uint32_t rdbLoadLen(FILE *fp, int *isencoded) {
     }
 }
 
     }
 }
 

-static robj *rdbLoadIntegerObject(FILE *fp, int enctype) {
+/* Load an integer-encoded object from file 'fp', with the specified
+ * encoding type 'enctype'. If encode is true the function may return
+ * an integer-encoded object as reply, otherwise the returned object
+ * will always be encoded as a raw string. */
+static robj *rdbLoadIntegerObject(FILE *fp, int enctype, int encode) {

     unsigned char enc[4];
     long long val;
 
     unsigned char enc[4];
     long long val;
 


@@ -3774,7 +3822,10 @@ static robj *rdbLoadIntegerObject(FILE *fp, int enctype) {
         val = 0; /* anti-warning */
         redisPanic("Unknown RDB integer encoding type");
     }
         val = 0; /* anti-warning */
         redisPanic("Unknown RDB integer encoding type");
     }

-    return createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",val));
+    if (encode)
+        return createStringObjectFromLongLong(val);
+    else
+        return createObject(REDIS_STRING,sdsfromlonglong(val));

 }
 
 static robj *rdbLoadLzfStringObject(FILE*fp) {
 }
 
 static robj *rdbLoadLzfStringObject(FILE*fp) {


@@ -3796,7 +3847,7 @@ err:
     return NULL;
 }
 
     return NULL;
 }
 

-static robj *rdbLoadStringObject(FILE*fp) {
+static robj *rdbGenericLoadStringObject(FILE*fp, int encode) {

     int isencoded;
     uint32_t len;
     sds val;
     int isencoded;
     uint32_t len;
     sds val;


@@ -3807,7 +3858,7 @@ static robj *rdbLoadStringObject(FILE*fp) {
         case REDIS_RDB_ENC_INT8:
         case REDIS_RDB_ENC_INT16:
         case REDIS_RDB_ENC_INT32:
         case REDIS_RDB_ENC_INT8:
         case REDIS_RDB_ENC_INT16:
         case REDIS_RDB_ENC_INT32:

-            return rdbLoadIntegerObject(fp,len);
+            return rdbLoadIntegerObject(fp,len,encode);

         case REDIS_RDB_ENC_LZF:
             return rdbLoadLzfStringObject(fp);
         default:
         case REDIS_RDB_ENC_LZF:
             return rdbLoadLzfStringObject(fp);
         default:


@@ -3824,6 +3875,14 @@ static robj *rdbLoadStringObject(FILE*fp) {
     return createObject(REDIS_STRING,val);
 }
 
     return createObject(REDIS_STRING,val);
 }
 

+static robj *rdbLoadStringObject(FILE *fp) {
+    return rdbGenericLoadStringObject(fp,0);
+}
+
+static robj *rdbLoadEncodedStringObject(FILE *fp) {
+    return rdbGenericLoadStringObject(fp,1);
+}
+

 /* For information about double serialization check rdbSaveDoubleValue() */
 static int rdbLoadDoubleValue(FILE *fp, double *val) {
     char buf[128];
 /* For information about double serialization check rdbSaveDoubleValue() */
 static int rdbLoadDoubleValue(FILE *fp, double *val) {
     char buf[128];


@@ -3850,7 +3909,7 @@ static robj *rdbLoadObject(int type, FILE *fp) {
     redisLog(REDIS_DEBUG,"LOADING OBJECT %d (at %d)\n",type,ftell(fp));
     if (type == REDIS_STRING) {
         /* Read string value */
     redisLog(REDIS_DEBUG,"LOADING OBJECT %d (at %d)\n",type,ftell(fp));
     if (type == REDIS_STRING) {
         /* Read string value */

-        if ((o = rdbLoadStringObject(fp)) == NULL) return NULL;
+        if ((o = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;

         o = tryObjectEncoding(o);
     } else if (type == REDIS_LIST || type == REDIS_SET) {
         /* Read list/set value */
         o = tryObjectEncoding(o);
     } else if (type == REDIS_LIST || type == REDIS_SET) {
         /* Read list/set value */


@@ -3866,7 +3925,7 @@ static robj *rdbLoadObject(int type, FILE *fp) {
         while(listlen--) {
             robj *ele;
 
         while(listlen--) {
             robj *ele;
 

-            if ((ele = rdbLoadStringObject(fp)) == NULL) return NULL;
+            if ((ele = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;

             ele = tryObjectEncoding(ele);
             if (type == REDIS_LIST) {
                 listAddNodeTail((list*)o->ptr,ele);
             ele = tryObjectEncoding(ele);
             if (type == REDIS_LIST) {
                 listAddNodeTail((list*)o->ptr,ele);


@@ -3887,7 +3946,7 @@ static robj *rdbLoadObject(int type, FILE *fp) {
             robj *ele;
             double *score = zmalloc(sizeof(double));
 
             robj *ele;
             double *score = zmalloc(sizeof(double));
 

-            if ((ele = rdbLoadStringObject(fp)) == NULL) return NULL;
+            if ((ele = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;

             ele = tryObjectEncoding(ele);
             if (rdbLoadDoubleValue(fp,score) == -1) return NULL;
             dictAdd(zs->dict,ele,score);
             ele = tryObjectEncoding(ele);
             if (rdbLoadDoubleValue(fp,score) == -1) return NULL;
             dictAdd(zs->dict,ele,score);


@@ -3940,13 +3999,13 @@ static robj *rdbLoadObject(int type, FILE *fp) {
 
 static int rdbLoad(char *filename) {
     FILE *fp;
 
 static int rdbLoad(char *filename) {
     FILE *fp;

-    robj *keyobj = NULL;

     uint32_t dbid;
     int type, retval, rdbver;
     uint32_t dbid;
     int type, retval, rdbver;

+    int swap_all_values = 0;

     dict *d = server.db[0].dict;
     redisDb *db = server.db+0;
     char buf[1024];
     dict *d = server.db[0].dict;
     redisDb *db = server.db+0;
     char buf[1024];

-    time_t expiretime = -1, now = time(NULL);
+    time_t expiretime, now = time(NULL);

     long long loadedkeys = 0;
 
     fp = fopen(filename,"r");
     long long loadedkeys = 0;
 
     fp = fopen(filename,"r");


@@ -3965,8 +4024,9 @@ static int rdbLoad(char *filename) {
         return REDIS_ERR;
     }
     while(1) {
         return REDIS_ERR;
     }
     while(1) {

-        robj *o;
+        robj *key, *val;

 
 

+        expiretime = -1;

         /* Read type. */
         if ((type = rdbLoadType(fp)) == -1) goto eoferr;
         if (type == REDIS_EXPIRETIME) {
         /* Read type. */
         if ((type = rdbLoadType(fp)) == -1) goto eoferr;
         if (type == REDIS_EXPIRETIME) {


@@ -3988,36 +4048,61 @@ static int rdbLoad(char *filename) {
             continue;
         }
         /* Read key */
             continue;
         }
         /* Read key */

-        if ((keyobj = rdbLoadStringObject(fp)) == NULL) goto eoferr;
+        if ((key = rdbLoadStringObject(fp)) == NULL) goto eoferr;

         /* Read value */
         /* Read value */

-        if ((o = rdbLoadObject(type,fp)) == NULL) goto eoferr;
+        if ((val = rdbLoadObject(type,fp)) == NULL) goto eoferr;
+        /* Check if the key already expired */
+        if (expiretime != -1 && expiretime < now) {
+            decrRefCount(key);
+            decrRefCount(val);
+            continue;
+        }

         /* Add the new object in the hash table */
         /* Add the new object in the hash table */

-        retval = dictAdd(d,keyobj,o);
+        retval = dictAdd(d,key,val);

         if (retval == DICT_ERR) {
         if (retval == DICT_ERR) {

-            redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", keyobj->ptr);
+            redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", key->ptr);

             exit(1);
         }
             exit(1);
         }

+        loadedkeys++;

         /* Set the expire time if needed */
         /* Set the expire time if needed */

-        if (expiretime != -1) {
-            setExpire(db,keyobj,expiretime);
-            /* Delete this key if already expired */
-            if (expiretime < now) deleteKey(db,keyobj);
-            expiretime = -1;
-        }
-        keyobj = o = NULL;
+        if (expiretime != -1) setExpire(db,key,expiretime);
+

         /* Handle swapping while loading big datasets when VM is on */
         /* Handle swapping while loading big datasets when VM is on */

-        loadedkeys++;
-        if (server.vm_enabled && (loadedkeys % 5000) == 0) {
+
+        /* If we detecter we are hopeless about fitting something in memory
+         * we just swap every new key on disk. Directly...
+         * Note that's important to check for this condition before resorting
+         * to random sampling, otherwise we may try to swap already
+         * swapped keys. */
+        if (swap_all_values) {
+            dictEntry *de = dictFind(d,key);
+
+            /* de may be NULL since the key already expired */
+            if (de) {
+                key = dictGetEntryKey(de);
+                val = dictGetEntryVal(de);
+
+                if (vmSwapObjectBlocking(key,val) == REDIS_OK) {
+                    dictGetEntryVal(de) = NULL;
+                }
+            }
+            continue;
+        }
+
+        /* If we have still some hope of having some value fitting memory
+         * then we try random sampling. */
+        if (!swap_all_values && server.vm_enabled && (loadedkeys % 5000) == 0) {

             while (zmalloc_used_memory() > server.vm_max_memory) {
                 if (vmSwapOneObjectBlocking() == REDIS_ERR) break;
             }
             while (zmalloc_used_memory() > server.vm_max_memory) {
                 if (vmSwapOneObjectBlocking() == REDIS_ERR) break;
             }

+            if (zmalloc_used_memory() > server.vm_max_memory)
+                swap_all_values = 1; /* We are already using too much mem */

         }
     }
     fclose(fp);
     return REDIS_OK;
 
 eoferr: /* unexpected end of file is handled here with a fatal exit */
         }
     }
     fclose(fp);
     return REDIS_OK;
 
 eoferr: /* unexpected end of file is handled here with a fatal exit */

-    if (keyobj) decrRefCount(keyobj);

     redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now.");
     exit(1);
     return REDIS_ERR; /* Just to avoid warning */
     redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now.");
     exit(1);
     return REDIS_ERR; /* Just to avoid warning */


@@ -4198,8 +4283,8 @@ static void incrDecrCommand(redisClient *c, long long incr) {
     robj *o;
 
     o = lookupKeyWrite(c->db,c->argv[1]);
     robj *o;
 
     o = lookupKeyWrite(c->db,c->argv[1]);

-
-    if (getLongLongFromObjectOrReply(c, o, &value, NULL) != REDIS_OK) return;
+    if (o != NULL && checkType(c,o,REDIS_STRING)) return;
+    if (getLongLongFromObjectOrReply(c,o,&value,NULL) != REDIS_OK) return;

 
     value += incr;
     o = createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",value));
 
     value += incr;
     o = createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",value));


@@ -4479,7 +4564,6 @@ static void shutdownCommand(redisClient *c) {
                 unlink(server.pidfile);
             redisLog(REDIS_WARNING,"%zu bytes used at exit",zmalloc_used_memory());
             redisLog(REDIS_WARNING,"Server exit now, bye bye...");
                 unlink(server.pidfile);
             redisLog(REDIS_WARNING,"%zu bytes used at exit",zmalloc_used_memory());
             redisLog(REDIS_WARNING,"Server exit now, bye bye...");

-            if (server.vm_enabled) unlink(server.vm_swap_file);

             exit(0);
         } else {
             /* Ooops.. error saving! The best we can do is to continue
             exit(0);
         } else {
             /* Ooops.. error saving! The best we can do is to continue


@@ -4818,7 +4902,7 @@ static void lremCommand(redisClient *c) {
         robj *ele = listNodeValue(ln);
 
         next = fromtail ? ln->prev : ln->next;
         robj *ele = listNodeValue(ln);
 
         next = fromtail ? ln->prev : ln->next;

-        if (compareStringObjects(ele,c->argv[3]) == 0) {
+        if (equalStringObjects(ele,c->argv[3])) {

             listDelNode(list,ln);
             server.dirty++;
             removed++;
             listDelNode(list,ln);
             server.dirty++;
             removed++;


@@ -5422,7 +5506,7 @@ static int zslDelete(zskiplist *zsl, double score, robj *obj) {
     /* We may have multiple elements with the same score, what we need
      * is to find the element with both the right score and object. */
     x = x->forward[0];
     /* We may have multiple elements with the same score, what we need
      * is to find the element with both the right score and object. */
     x = x->forward[0];

-    if (x && score == x->score && compareStringObjects(x->obj,obj) == 0) {
+    if (x && score == x->score && equalStringObjects(x->obj,obj)) {

         zslDeleteNode(zsl, x, update);
         zslFreeNode(x);
         return 1;
         zslDeleteNode(zsl, x, update);
         zslFreeNode(x);
         return 1;


@@ -5527,7 +5611,7 @@ static unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) {
         }
 
         /* x might be equal to zsl->header, so test if obj is non-NULL */
         }
 
         /* x might be equal to zsl->header, so test if obj is non-NULL */

-        if (x->obj && compareStringObjects(x->obj,o) == 0) {
+        if (x->obj && equalStringObjects(x->obj,o)) {

             return rank;
         }
     }
             return rank;
         }
     }


@@ -6462,12 +6546,11 @@ static void hincrbyCommand(redisClient *c) {
     if (getLongLongFromObjectOrReply(c,c->argv[3],&incr,NULL) != REDIS_OK) return;
     if ((o = hashLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
     if ((current = hashGet(o,c->argv[2])) != NULL) {
     if (getLongLongFromObjectOrReply(c,c->argv[3],&incr,NULL) != REDIS_OK) return;
     if ((o = hashLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
     if ((current = hashGet(o,c->argv[2])) != NULL) {

-        if (current->encoding == REDIS_ENCODING_RAW)
-            value = strtoll(current->ptr,NULL,10);
-        else if (current->encoding == REDIS_ENCODING_INT)
-            value = (long)current->ptr;
-        else
-            redisAssert(1 != 1);
+        if (getLongLongFromObjectOrReply(c,current,&value,
+            "hash value is not an integer") != REDIS_OK) {
+            decrRefCount(current);
+            return;
+        }

         decrRefCount(current);
     } else {
         value = 0;
         decrRefCount(current);
     } else {
         value = 0;


@@ -8008,11 +8091,55 @@ static void freeMemoryIfNeeded(void) {
 
 /* ============================== Append Only file ========================== */
 
 
 /* ============================== Append Only file ========================== */
 

+/* Write the append only file buffer on disk.
+ *
+ * Since we are required to write the AOF before replying to the client,
+ * and the only way the client socket can get a write is entering when the
+ * the event loop, we accumulate all the AOF writes in a memory
+ * buffer and write it on disk using this function just before entering
+ * the event loop again. */
+static void flushAppendOnlyFile(void) {
+    time_t now;
+    ssize_t nwritten;
+
+    if (sdslen(server.aofbuf) == 0) return;
+
+    /* We want to perform a single write. This should be guaranteed atomic
+     * at least if the filesystem we are writing is a real physical one.
+     * While this will save us against the server being killed I don't think
+     * there is much to do about the whole server stopping for power problems
+     * or alike */
+     nwritten = write(server.appendfd,server.aofbuf,sdslen(server.aofbuf));
+     if (nwritten != (signed)sdslen(server.aofbuf)) {
+        /* Ooops, we are in troubles. The best thing to do for now is
+         * aborting instead of giving the illusion that everything is
+         * working as expected. */
+         if (nwritten == -1) {
+            redisLog(REDIS_WARNING,"Exiting on error writing to the append-only file: %s",strerror(errno));
+         } else {
+            redisLog(REDIS_WARNING,"Exiting on short write while writing to the append-only file: %s",strerror(errno));
+         }
+         exit(1);
+    }
+    sdsfree(server.aofbuf);
+    server.aofbuf = sdsempty();
+
+    /* Fsync if needed */
+    now = time(NULL);
+    if (server.appendfsync == APPENDFSYNC_ALWAYS ||
+        (server.appendfsync == APPENDFSYNC_EVERYSEC &&
+         now-server.lastfsync > 1))
+    {
+        /* aof_fsync is defined as fdatasync() for Linux in order to avoid
+         * flushing metadata. */
+        aof_fsync(server.appendfd); /* Let's try to get this data on the disk */
+        server.lastfsync = now;
+    }
+}
+

 static void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) {
     sds buf = sdsempty();
     int j;
 static void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) {
     sds buf = sdsempty();
     int j;

-    ssize_t nwritten;
-    time_t now;

     robj *tmpargv[3];
 
     /* The DB this command was targetting is not the same as the last command
     robj *tmpargv[3];
 
     /* The DB this command was targetting is not the same as the last command


@@ -8058,23 +8185,11 @@ static void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv
             decrRefCount(argv[j]);
     }
 
             decrRefCount(argv[j]);
     }
 

-    /* We want to perform a single write. This should be guaranteed atomic
-     * at least if the filesystem we are writing is a real physical one.
-     * While this will save us against the server being killed I don't think
-     * there is much to do about the whole server stopping for power problems
-     * or alike */
-     nwritten = write(server.appendfd,buf,sdslen(buf));
-     if (nwritten != (signed)sdslen(buf)) {
-        /* Ooops, we are in troubles. The best thing to do for now is
-         * to simply exit instead to give the illusion that everything is
-         * working as expected. */
-         if (nwritten == -1) {
-            redisLog(REDIS_WARNING,"Exiting on error writing to the append-only file: %s",strerror(errno));
-         } else {
-            redisLog(REDIS_WARNING,"Exiting on short write while writing to the append-only file: %s",strerror(errno));
-         }
-         exit(1);
-    }
+    /* Append to the AOF buffer. This will be flushed on disk just before
+     * of re-entering the event loop, so before the client will get a
+     * positive reply about the operation performed. */
+    server.aofbuf = sdscatlen(server.aofbuf,buf,sdslen(buf));
+

     /* If a background append only file rewriting is in progress we want to
      * accumulate the differences between the child DB and the current one
      * in a buffer, so that when the child process will do its work we
     /* If a background append only file rewriting is in progress we want to
      * accumulate the differences between the child DB and the current one
      * in a buffer, so that when the child process will do its work we


@@ -8083,16 +8198,6 @@ static void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv
         server.bgrewritebuf = sdscatlen(server.bgrewritebuf,buf,sdslen(buf));
 
     sdsfree(buf);
         server.bgrewritebuf = sdscatlen(server.bgrewritebuf,buf,sdslen(buf));
 
     sdsfree(buf);

-    now = time(NULL);
-    if (server.appendfsync == APPENDFSYNC_ALWAYS ||
-        (server.appendfsync == APPENDFSYNC_EVERYSEC &&
-         now-server.lastfsync > 1))
-    {
-        /* aof_fsync is defined as fdatasync() for Linux in order to avoid
-         * flushing metadata. */
-        aof_fsync(server.appendfd); /* Let's try to get this data on the disk */
-        server.lastfsync = now;
-    }

 }
 
 /* In Redis commands are always executed in the context of a client, so in
 }
 
 /* In Redis commands are always executed in the context of a client, so in


@@ -8112,12 +8217,14 @@ static struct redisClient *createFakeClient(void) {
     c->reply = listCreate();
     listSetFreeMethod(c->reply,decrRefCount);
     listSetDupMethod(c->reply,dupClientReplyValue);
     c->reply = listCreate();
     listSetFreeMethod(c->reply,decrRefCount);
     listSetDupMethod(c->reply,dupClientReplyValue);

+    initClientMultiState(c);

     return c;
 }
 
 static void freeFakeClient(struct redisClient *c) {
     sdsfree(c->querybuf);
     listRelease(c->reply);
     return c;
 }
 
 static void freeFakeClient(struct redisClient *c) {
     sdsfree(c->querybuf);
     listRelease(c->reply);

+    freeClientMultiState(c);

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


@@ -8129,6 +8236,7 @@ int loadAppendOnlyFile(char *filename) {
     FILE *fp = fopen(filename,"r");
     struct redis_stat sb;
     unsigned long long loadedkeys = 0;
     FILE *fp = fopen(filename,"r");
     struct redis_stat sb;
     unsigned long long loadedkeys = 0;

+    int appendonly = server.appendonly;

 
     if (redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0)
         return REDIS_ERR;
 
     if (redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0)
         return REDIS_ERR;


@@ -8138,6 +8246,10 @@ int loadAppendOnlyFile(char *filename) {
         exit(1);
     }
 
         exit(1);
     }
 

+    /* Temporarily disable AOF, to prevent EXEC from feeding a MULTI
+     * to the same file we're about to read. */
+    server.appendonly = 0;
+

     fakeClient = createFakeClient();
     while(1) {
         int argc, j;
     fakeClient = createFakeClient();
     while(1) {
         int argc, j;


@@ -8193,8 +8305,14 @@ int loadAppendOnlyFile(char *filename) {
             }
         }
     }
             }
         }
     }

+
+    /* This point can only be reached when EOF is reached without errors.
+     * If the client is in the middle of a MULTI/EXEC, log error and quit. */
+    if (fakeClient->flags & REDIS_MULTI) goto readerr;
+

     fclose(fp);
     freeFakeClient(fakeClient);
     fclose(fp);
     freeFakeClient(fakeClient);

+    server.appendonly = appendonly;

     return REDIS_OK;
 
 readerr:
     return REDIS_OK;
 
 readerr:


@@ -8544,42 +8662,38 @@ static void aofRemoveTempFile(pid_t childpid) {
 
 /* =================== Virtual Memory - Blocking Side  ====================== */
 
 
 /* =================== Virtual Memory - Blocking Side  ====================== */
 

-/* substitute the first occurrence of '%p' with the process pid in the
- * swap file name. */
-static void expandVmSwapFilename(void) {
-    char *p = strstr(server.vm_swap_file,"%p");
-    sds new;
-
-    if (!p) return;
-    new = sdsempty();
-    *p = '\0';
-    new = sdscat(new,server.vm_swap_file);
-    new = sdscatprintf(new,"%ld",(long) getpid());
-    new = sdscat(new,p+2);
-    zfree(server.vm_swap_file);
-    server.vm_swap_file = new;
-}
-

 static void vmInit(void) {
     off_t totsize;
     int pipefds[2];
     size_t stacksize;
 static void vmInit(void) {
     off_t totsize;
     int pipefds[2];
     size_t stacksize;

+    struct flock fl;

 
     if (server.vm_max_threads != 0)
         zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
 
 
     if (server.vm_max_threads != 0)
         zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
 

-    expandVmSwapFilename();

     redisLog(REDIS_NOTICE,"Using '%s' as swap file",server.vm_swap_file);
     redisLog(REDIS_NOTICE,"Using '%s' as swap file",server.vm_swap_file);

+    /* Try to open the old swap file, otherwise create it */

     if ((server.vm_fp = fopen(server.vm_swap_file,"r+b")) == NULL) {
         server.vm_fp = fopen(server.vm_swap_file,"w+b");
     }
     if (server.vm_fp == NULL) {
         redisLog(REDIS_WARNING,
     if ((server.vm_fp = fopen(server.vm_swap_file,"r+b")) == NULL) {
         server.vm_fp = fopen(server.vm_swap_file,"w+b");
     }
     if (server.vm_fp == NULL) {
         redisLog(REDIS_WARNING,

-            "Impossible to open the swap file: %s. Exiting.",
+            "Can't open the swap file: %s. Exiting.",

             strerror(errno));
         exit(1);
     }
     server.vm_fd = fileno(server.vm_fp);
             strerror(errno));
         exit(1);
     }
     server.vm_fd = fileno(server.vm_fp);

+    /* Lock the swap file for writing, this is useful in order to avoid
+     * another instance to use the same swap file for a config error. */
+    fl.l_type = F_WRLCK;
+    fl.l_whence = SEEK_SET;
+    fl.l_start = fl.l_len = 0;
+    if (fcntl(server.vm_fd,F_SETLK,&fl) == -1) {
+        redisLog(REDIS_WARNING,
+            "Can't lock the swap file at '%s': %s. Make sure it is not used by another Redis instance.", server.vm_swap_file, strerror(errno));
+        exit(1);
+    }
+    /* Initialize */

     server.vm_next_page = 0;
     server.vm_near_pages = 0;
     server.vm_stats_used_pages = 0;
     server.vm_next_page = 0;
     server.vm_near_pages = 0;
     server.vm_stats_used_pages = 0;


@@ -9561,7 +9675,7 @@ static int dontWaitForSwappedKey(redisClient *c, robj *key) {
     /* Remove the key from the list of keys this client is waiting for. */
     listRewind(c->io_keys,&li);
     while ((ln = listNext(&li)) != NULL) {
     /* Remove the key from the list of keys this client is waiting for. */
     listRewind(c->io_keys,&li);
     while ((ln = listNext(&li)) != NULL) {

-        if (compareStringObjects(ln->value,key) == 0) {
+        if (equalStringObjects(ln->value,key)) {

             listDelNode(c->io_keys,ln);
             break;
         }
             listDelNode(c->io_keys,ln);
             break;
         }


@@ -9621,6 +9735,50 @@ static void configSetCommand(redisClient *c) {
         server.masterauth = zstrdup(o->ptr);
     } else if (!strcasecmp(c->argv[2]->ptr,"maxmemory")) {
         server.maxmemory = strtoll(o->ptr, NULL, 10);
         server.masterauth = zstrdup(o->ptr);
     } else if (!strcasecmp(c->argv[2]->ptr,"maxmemory")) {
         server.maxmemory = strtoll(o->ptr, NULL, 10);

+    } else if (!strcasecmp(c->argv[2]->ptr,"appendfsync")) {
+        if (!strcasecmp(o->ptr,"no")) {
+            server.appendfsync = APPENDFSYNC_NO;
+        } else if (!strcasecmp(o->ptr,"everysec")) {
+            server.appendfsync = APPENDFSYNC_EVERYSEC;
+        } else if (!strcasecmp(o->ptr,"always")) {
+            server.appendfsync = APPENDFSYNC_ALWAYS;
+        } else {
+            goto badfmt;
+        }
+    } else if (!strcasecmp(c->argv[2]->ptr,"save")) {
+        int vlen, j;
+        sds *v = sdssplitlen(o->ptr,sdslen(o->ptr)," ",1,&vlen);
+
+        /* Perform sanity check before setting the new config:
+         * - Even number of args
+         * - Seconds >= 1, changes >= 0 */
+        if (vlen & 1) {
+            sdsfreesplitres(v,vlen);
+            goto badfmt;
+        }
+        for (j = 0; j < vlen; j++) {
+            char *eptr;
+            long val;
+
+            val = strtoll(v[j], &eptr, 10);
+            if (eptr[0] != '\0' ||
+                ((j & 1) == 0 && val < 1) ||
+                ((j & 1) == 1 && val < 0)) {
+                sdsfreesplitres(v,vlen);
+                goto badfmt;
+            }
+        }
+        /* Finally set the new config */
+        resetServerSaveParams();
+        for (j = 0; j < vlen; j += 2) {
+            time_t seconds;
+            int changes;
+
+            seconds = strtoll(v[j],NULL,10);
+            changes = strtoll(v[j+1],NULL,10);
+            appendServerSaveParams(seconds, changes);
+        }
+        sdsfreesplitres(v,vlen);

     } else {
         addReplySds(c,sdscatprintf(sdsempty(),
             "-ERR not supported CONFIG parameter %s\r\n",
     } else {
         addReplySds(c,sdscatprintf(sdsempty(),
             "-ERR not supported CONFIG parameter %s\r\n",


@@ -9630,6 +9788,14 @@ static void configSetCommand(redisClient *c) {
     }
     decrRefCount(o);
     addReply(c,shared.ok);
     }
     decrRefCount(o);
     addReply(c,shared.ok);

+    return;
+
+badfmt: /* Bad format errors */
+    addReplySds(c,sdscatprintf(sdsempty(),
+        "-ERR invalid argument '%s' for CONFIG SET '%s'\r\n",
+            (char*)o->ptr,
+            (char*)c->argv[2]->ptr));
+    decrRefCount(o);

 }
 
 static void configGetCommand(redisClient *c) {
 }
 
 static void configGetCommand(redisClient *c) {


@@ -9664,6 +9830,35 @@ static void configGetCommand(redisClient *c) {
         addReplyBulkCString(c,buf);
         matches++;
     }
         addReplyBulkCString(c,buf);
         matches++;
     }

+    if (stringmatch(pattern,"appendfsync",0)) {
+        char *policy;
+
+        switch(server.appendfsync) {
+        case APPENDFSYNC_NO: policy = "no"; break;
+        case APPENDFSYNC_EVERYSEC: policy = "everysec"; break;
+        case APPENDFSYNC_ALWAYS: policy = "always"; break;
+        default: policy = "unknown"; break; /* too harmless to panic */
+        }
+        addReplyBulkCString(c,"appendfsync");
+        addReplyBulkCString(c,policy);
+        matches++;
+    }
+    if (stringmatch(pattern,"save",0)) {
+        sds buf = sdsempty();
+        int j;
+
+        for (j = 0; j < server.saveparamslen; j++) {
+            buf = sdscatprintf(buf,"%ld %d",
+                    server.saveparams[j].seconds,
+                    server.saveparams[j].changes);
+            if (j != server.saveparamslen-1)
+                buf = sdscatlen(buf," ",1);
+        }
+        addReplyBulkCString(c,"save");
+        addReplyBulkCString(c,buf);
+        sdsfree(buf);
+        matches++;
+    }

     decrRefCount(o);
     lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",matches*2);
 }
     decrRefCount(o);
     lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",matches*2);
 }


@@ -9707,7 +9902,7 @@ static int listMatchPubsubPattern(void *a, void *b) {
     pubsubPattern *pa = a, *pb = b;
 
     return (pa->client == pb->client) &&
     pubsubPattern *pa = a, *pb = b;
 
     return (pa->client == pb->client) &&

-           (compareStringObjects(pa->pattern,pb->pattern) == 0);
+           (equalStringObjects(pa->pattern,pb->pattern));

 }
 
 /* Subscribe a client to a channel. Returns 1 if the operation succeeded, or
 }
 
 /* Subscribe a client to a channel. Returns 1 if the operation succeeded, or


@@ -10044,6 +10239,25 @@ static void debugCommand(redisClient *c) {
         } else {
             addReply(c,shared.err);
         }
         } else {
             addReply(c,shared.err);
         }

+    } else if (!strcasecmp(c->argv[1]->ptr,"populate") && c->argc == 3) {
+        long keys, j;
+        robj *key, *val;
+        char buf[128];
+
+        if (getLongFromObjectOrReply(c, c->argv[2], &keys, NULL) != REDIS_OK)
+            return;
+        for (j = 0; j < keys; j++) {
+            snprintf(buf,sizeof(buf),"key:%lu",j);
+            key = createStringObject(buf,strlen(buf));
+            if (lookupKeyRead(c->db,key) != NULL) {
+                decrRefCount(key);
+                continue;
+            }
+            snprintf(buf,sizeof(buf),"value:%lu",j);
+            val = createStringObject(buf,strlen(buf));
+            dictAdd(c->db->dict,key,val);
+        }
+        addReply(c,shared.ok);

     } else {
         addReplySds(c,sdsnew(
             "-ERR Syntax error, try DEBUG [SEGFAULT|OBJECT <key>|SWAPIN <key>|SWAPOUT <key>|RELOAD]\r\n"));
     } else {
         addReplySds(c,sdsnew(
             "-ERR Syntax error, try DEBUG [SEGFAULT|OBJECT <key>|SWAPIN <key>|SWAPOUT <key>|RELOAD]\r\n"));