fixed a few problems with redis-trib resharding.

[redis.git] / src / t_set.c

diff --git a/src/t_set.c b/src/t_set.c
index 234efc7de6291fa90b9e913f9c10961502715864..bffea3d4d6d256dcf25b7c794b10a651389a5606 100644 (file)
--- a/src/t_set.c
+++ b/src/t_set.c
@@ -47,17 +47,17 @@ int setTypeAdd(robj *subject, robj *value) {
     return 0;
 }
 
     return 0;
 }
 

-int setTypeRemove(robj *subject, robj *value) {
+int setTypeRemove(robj *setobj, robj *value) {

     long long llval;
     long long llval;

-    if (subject->encoding == REDIS_ENCODING_HT) {
-        if (dictDelete(subject->ptr,value) == DICT_OK) {
-            if (htNeedsResize(subject->ptr)) dictResize(subject->ptr);
+    if (setobj->encoding == REDIS_ENCODING_HT) {
+        if (dictDelete(setobj->ptr,value) == DICT_OK) {
+            if (htNeedsResize(setobj->ptr)) dictResize(setobj->ptr);

             return 1;
         }
             return 1;
         }

-    } else if (subject->encoding == REDIS_ENCODING_INTSET) {
+    } else if (setobj->encoding == REDIS_ENCODING_INTSET) {

         if (isObjectRepresentableAsLongLong(value,&llval) == REDIS_OK) {
         if (isObjectRepresentableAsLongLong(value,&llval) == REDIS_OK) {

-            uint8_t success;
-            subject->ptr = intsetRemove(subject->ptr,llval,&success);
+            int success;
+            setobj->ptr = intsetRemove(setobj->ptr,llval,&success);

             if (success) return 1;
         }
     } else {
             if (success) return 1;
         }
     } else {


@@ -101,40 +101,77 @@ void setTypeReleaseIterator(setTypeIterator *si) {
 }
 
 /* Move to the next entry in the set. Returns the object at the current
 }
 
 /* Move to the next entry in the set. Returns the object at the current

- * position, or NULL when the end is reached. This object will have its
- * refcount incremented, so the caller needs to take care of this. */
-robj *setTypeNext(setTypeIterator *si) {
-    robj *ret = NULL;
+ * position.
+ *
+ * Since set elements can be internally be stored as redis objects or
+ * simple arrays of integers, setTypeNext returns the encoding of the
+ * set object you are iterating, and will populate the appropriate pointer
+ * (eobj) or (llobj) accordingly.
+ *
+ * When there are no longer elements -1 is returned.
+ * Returned objects ref count is not incremented, so this function is
+ * copy on write friendly. */
+int setTypeNext(setTypeIterator *si, robj **objele, int64_t *llele) {

     if (si->encoding == REDIS_ENCODING_HT) {
         dictEntry *de = dictNext(si->di);
     if (si->encoding == REDIS_ENCODING_HT) {
         dictEntry *de = dictNext(si->di);

-        if (de != NULL) {
-            ret = dictGetEntryKey(de);
-            incrRefCount(ret);
-        }
+        if (de == NULL) return -1;
+        *objele = dictGetEntryKey(de);

     } else if (si->encoding == REDIS_ENCODING_INTSET) {
     } else if (si->encoding == REDIS_ENCODING_INTSET) {

-        int64_t llval;
-        if (intsetGet(si->subject->ptr,si->ii++,&llval))
-            ret = createStringObjectFromLongLong(llval);
+        if (!intsetGet(si->subject->ptr,si->ii++,llele))
+            return -1;

     }
     }

-    return ret;
+    return si->encoding;

 }
 
 }
 

+/* The not copy on write friendly version but easy to use version
+ * of setTypeNext() is setTypeNextObject(), returning new objects
+ * or incrementing the ref count of returned objects. So if you don't
+ * retain a pointer to this object you should call decrRefCount() against it.
+ *
+ * This function is the way to go for write operations where COW is not
+ * an issue as the result will be anyway of incrementing the ref count. */
+robj *setTypeNextObject(setTypeIterator *si) {
+    int64_t intele;
+    robj *objele;
+    int encoding;
+
+    encoding = setTypeNext(si,&objele,&intele);
+    switch(encoding) {
+        case -1:    return NULL;
+        case REDIS_ENCODING_INTSET:
+            return createStringObjectFromLongLong(intele);
+        case REDIS_ENCODING_HT:
+            incrRefCount(objele);
+            return objele;
+        default:
+            redisPanic("Unsupported encoding");
+    }
+    return NULL; /* just to suppress warnings */
+}

 
 

-/* Return random element from set. The returned object will always have
- * an incremented refcount. */
-robj *setTypeRandomElement(robj *subject) {
-    robj *ret = NULL;
-    if (subject->encoding == REDIS_ENCODING_HT) {
-        dictEntry *de = dictGetRandomKey(subject->ptr);
-        ret = dictGetEntryKey(de);
-        incrRefCount(ret);
-    } else if (subject->encoding == REDIS_ENCODING_INTSET) {
-        long long llval = intsetRandom(subject->ptr);
-        ret = createStringObjectFromLongLong(llval);
+/* Return random element from a non empty set.
+ * The returned element can be a int64_t value if the set is encoded
+ * as an "intset" blob of integers, or a redis object if the set
+ * is a regular set.
+ *
+ * The caller provides both pointers to be populated with the right
+ * object. The return value of the function is the object->encoding
+ * field of the object and is used by the caller to check if the
+ * int64_t pointer or the redis object pointere was populated.
+ *
+ * When an object is returned (the set was a real set) the ref count
+ * of the object is not incremented so this function can be considered
+ * copy on write friendly. */
+int setTypeRandomElement(robj *setobj, robj **objele, int64_t *llele) {
+    if (setobj->encoding == REDIS_ENCODING_HT) {
+        dictEntry *de = dictGetRandomKey(setobj->ptr);
+        *objele = dictGetEntryKey(de);
+    } else if (setobj->encoding == REDIS_ENCODING_INTSET) {
+        *llele = intsetRandom(setobj->ptr);

     } else {
         redisPanic("Unknown set encoding");
     }
     } else {
         redisPanic("Unknown set encoding");
     }

-    return ret;
+    return setobj->encoding;

 }
 
 unsigned long setTypeSize(robj *subject) {
 }
 
 unsigned long setTypeSize(robj *subject) {


@@ -150,25 +187,30 @@ unsigned long setTypeSize(robj *subject) {
 /* Convert the set to specified encoding. The resulting dict (when converting
  * to a hashtable) is presized to hold the number of elements in the original
  * set. */
 /* Convert the set to specified encoding. The resulting dict (when converting
  * to a hashtable) is presized to hold the number of elements in the original
  * set. */

-void setTypeConvert(robj *subject, int enc) {
+void setTypeConvert(robj *setobj, int enc) {

     setTypeIterator *si;
     setTypeIterator *si;

-    robj *element;
-    redisAssert(subject->type == REDIS_SET);
+    redisAssert(setobj->type == REDIS_SET &&
+                setobj->encoding == REDIS_ENCODING_INTSET);

 
     if (enc == REDIS_ENCODING_HT) {
 
     if (enc == REDIS_ENCODING_HT) {

+        int64_t intele;

         dict *d = dictCreate(&setDictType,NULL);
         dict *d = dictCreate(&setDictType,NULL);

+        robj *element;
+

         /* Presize the dict to avoid rehashing */
         /* Presize the dict to avoid rehashing */

-        dictExpand(d,intsetLen(subject->ptr));
+        dictExpand(d,intsetLen(setobj->ptr));

 
 

-        /* setTypeGet returns a robj with incremented refcount */
-        si = setTypeInitIterator(subject);
-        while ((element = setTypeNext(si)) != NULL)
+        /* To add the elements we extract integers and create redis objects */
+        si = setTypeInitIterator(setobj);
+        while (setTypeNext(si,NULL,&intele) != -1) {
+            element = createStringObjectFromLongLong(intele);

             redisAssert(dictAdd(d,element,NULL) == DICT_OK);
             redisAssert(dictAdd(d,element,NULL) == DICT_OK);

+        }

         setTypeReleaseIterator(si);
 
         setTypeReleaseIterator(si);
 

-        subject->encoding = REDIS_ENCODING_HT;
-        zfree(subject->ptr);
-        subject->ptr = d;
+        setobj->encoding = REDIS_ENCODING_HT;
+        zfree(setobj->ptr);
+        setobj->ptr = d;

     } else {
         redisPanic("Unsupported set conversion");
     }
     } else {
         redisPanic("Unsupported set conversion");
     }


@@ -176,9 +218,9 @@ void setTypeConvert(robj *subject, int enc) {
 
 void saddCommand(redisClient *c) {
     robj *set;
 
 void saddCommand(redisClient *c) {
     robj *set;

+    int j, added = 0;

 
     set = lookupKeyWrite(c->db,c->argv[1]);
 
     set = lookupKeyWrite(c->db,c->argv[1]);

-    c->argv[2] = tryObjectEncoding(c->argv[2]);

     if (set == NULL) {
         set = setTypeCreate(c->argv[2]);
         dbAdd(c->db,c->argv[1],set);
     if (set == NULL) {
         set = setTypeCreate(c->argv[2]);
         dbAdd(c->db,c->argv[1],set);


@@ -188,30 +230,37 @@ void saddCommand(redisClient *c) {
             return;
         }
     }
             return;
         }
     }

-    if (setTypeAdd(set,c->argv[2])) {
-        touchWatchedKey(c->db,c->argv[1]);
-        server.dirty++;
-        addReply(c,shared.cone);
-    } else {
-        addReply(c,shared.czero);
+
+    for (j = 2; j < c->argc; j++) {
+        c->argv[j] = tryObjectEncoding(c->argv[j]);
+        if (setTypeAdd(set,c->argv[j])) added++;

     }
     }

+    if (added) signalModifiedKey(c->db,c->argv[1]);
+    server.dirty += added;
+    addReplyLongLong(c,added);

 }
 
 void sremCommand(redisClient *c) {
     robj *set;
 }
 
 void sremCommand(redisClient *c) {
     robj *set;

+    int j, deleted = 0;

 
     if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
         checkType(c,set,REDIS_SET)) return;
 
 
     if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
         checkType(c,set,REDIS_SET)) return;
 

-    c->argv[2] = tryObjectEncoding(c->argv[2]);
-    if (setTypeRemove(set,c->argv[2])) {
-        if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
-        touchWatchedKey(c->db,c->argv[1]);
-        server.dirty++;
-        addReply(c,shared.cone);
-    } else {
-        addReply(c,shared.czero);
+    for (j = 2; j < c->argc; j++) {
+        if (setTypeRemove(set,c->argv[j])) {
+            deleted++;
+            if (setTypeSize(set) == 0) {
+                dbDelete(c->db,c->argv[1]);
+                break;
+            }
+        }
+    }
+    if (deleted) {
+        signalModifiedKey(c->db,c->argv[1]);
+        server.dirty += deleted;

     }
     }

+    addReplyLongLong(c,deleted);

 }
 
 void smoveCommand(redisClient *c) {
 }
 
 void smoveCommand(redisClient *c) {


@@ -245,8 +294,8 @@ void smoveCommand(redisClient *c) {
 
     /* Remove the src set from the database when empty */
     if (setTypeSize(srcset) == 0) dbDelete(c->db,c->argv[1]);
 
     /* Remove the src set from the database when empty */
     if (setTypeSize(srcset) == 0) dbDelete(c->db,c->argv[1]);

-    touchWatchedKey(c->db,c->argv[1]);
-    touchWatchedKey(c->db,c->argv[2]);
+    signalModifiedKey(c->db,c->argv[1]);
+    signalModifiedKey(c->db,c->argv[2]);

     server.dirty++;
 
     /* Create the destination set when it doesn't exist */
     server.dirty++;
 
     /* Create the destination set when it doesn't exist */


@@ -283,36 +332,47 @@ void scardCommand(redisClient *c) {
 }
 
 void spopCommand(redisClient *c) {
 }
 
 void spopCommand(redisClient *c) {

-    robj *set, *ele;
+    robj *set, *ele, *aux;
+    int64_t llele;
+    int encoding;

 
     if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
         checkType(c,set,REDIS_SET)) return;
 
 
     if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
         checkType(c,set,REDIS_SET)) return;
 

-    ele = setTypeRandomElement(set);
-    if (ele == NULL) {
-        addReply(c,shared.nullbulk);
+    encoding = setTypeRandomElement(set,&ele,&llele);
+    if (encoding == REDIS_ENCODING_INTSET) {
+        ele = createStringObjectFromLongLong(llele);
+        set->ptr = intsetRemove(set->ptr,llele,NULL);

     } else {
     } else {

+        incrRefCount(ele);

         setTypeRemove(set,ele);
         setTypeRemove(set,ele);

-        addReplyBulk(c,ele);
-        decrRefCount(ele);
-        if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
-        touchWatchedKey(c->db,c->argv[1]);
-        server.dirty++;

     }
     }

+
+    /* Replicate/AOF this command as an SREM operation */
+    aux = createStringObject("SREM",4);
+    rewriteClientCommandVector(c,3,aux,c->argv[1],ele);
+    decrRefCount(ele);
+    decrRefCount(aux);
+
+    addReplyBulk(c,ele);
+    if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
+    signalModifiedKey(c->db,c->argv[1]);
+    server.dirty++;

 }
 
 void srandmemberCommand(redisClient *c) {
     robj *set, *ele;
 }
 
 void srandmemberCommand(redisClient *c) {
     robj *set, *ele;

+    int64_t llele;
+    int encoding;

 
     if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
         checkType(c,set,REDIS_SET)) return;
 
 
     if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
         checkType(c,set,REDIS_SET)) return;
 

-    ele = setTypeRandomElement(set);
-    if (ele == NULL) {
-        addReply(c,shared.nullbulk);
+    encoding = setTypeRandomElement(set,&ele,&llele);
+    if (encoding == REDIS_ENCODING_INTSET) {
+        addReplyBulkLongLong(c,llele);

     } else {
         addReplyBulk(c,ele);
     } else {
         addReplyBulk(c,ele);

-        decrRefCount(ele);

     }
 }
 
     }
 }
 


@@ -323,9 +383,11 @@ int qsortCompareSetsByCardinality(const void *s1, const void *s2) {
 void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum, robj *dstkey) {
     robj **sets = zmalloc(sizeof(robj*)*setnum);
     setTypeIterator *si;
 void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum, robj *dstkey) {
     robj **sets = zmalloc(sizeof(robj*)*setnum);
     setTypeIterator *si;

-    robj *ele, *dstset = NULL;
+    robj *eleobj, *dstset = NULL;
+    int64_t intobj;

     void *replylen = NULL;
     unsigned long j, cardinality = 0;
     void *replylen = NULL;
     unsigned long j, cardinality = 0;

+    int encoding;

 
     for (j = 0; j < setnum; j++) {
         robj *setobj = dstkey ?
 
     for (j = 0; j < setnum; j++) {
         robj *setobj = dstkey ?


@@ -335,7 +397,7 @@ void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum,
             zfree(sets);
             if (dstkey) {
                 if (dbDelete(c->db,dstkey)) {
             zfree(sets);
             if (dstkey) {
                 if (dbDelete(c->db,dstkey)) {

-                    touchWatchedKey(c->db,dstkey);
+                    signalModifiedKey(c->db,dstkey);

                     server.dirty++;
                 }
                 addReply(c,shared.czero);
                     server.dirty++;
                 }
                 addReply(c,shared.czero);


@@ -371,20 +433,61 @@ void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum,
      * the element against all the other sets, if at least one set does
      * not include the element it is discarded */
     si = setTypeInitIterator(sets[0]);
      * the element against all the other sets, if at least one set does
      * not include the element it is discarded */
     si = setTypeInitIterator(sets[0]);

-    while((ele = setTypeNext(si)) != NULL) {
-        for (j = 1; j < setnum; j++)
-            if (!setTypeIsMember(sets[j],ele)) break;
+    while((encoding = setTypeNext(si,&eleobj,&intobj)) != -1) {
+        for (j = 1; j < setnum; j++) {
+            if (sets[j] == sets[0]) continue;
+            if (encoding == REDIS_ENCODING_INTSET) {
+                /* intset with intset is simple... and fast */
+                if (sets[j]->encoding == REDIS_ENCODING_INTSET &&
+                    !intsetFind((intset*)sets[j]->ptr,intobj))
+                {
+                    break;
+                /* in order to compare an integer with an object we
+                 * have to use the generic function, creating an object
+                 * for this */
+                } else if (sets[j]->encoding == REDIS_ENCODING_HT) {
+                    eleobj = createStringObjectFromLongLong(intobj);
+                    if (!setTypeIsMember(sets[j],eleobj)) {
+                        decrRefCount(eleobj);
+                        break;
+                    }
+                    decrRefCount(eleobj);
+                }
+            } else if (encoding == REDIS_ENCODING_HT) {
+                /* Optimization... if the source object is integer
+                 * encoded AND the target set is an intset, we can get
+                 * a much faster path. */
+                if (eleobj->encoding == REDIS_ENCODING_INT &&
+                    sets[j]->encoding == REDIS_ENCODING_INTSET &&
+                    !intsetFind((intset*)sets[j]->ptr,(long)eleobj->ptr))
+                {
+                    break;
+                /* else... object to object check is easy as we use the
+                 * type agnostic API here. */
+                } else if (!setTypeIsMember(sets[j],eleobj)) {
+                    break;
+                }
+            }
+        }

 
         /* Only take action when all sets contain the member */
         if (j == setnum) {
             if (!dstkey) {
 
         /* Only take action when all sets contain the member */
         if (j == setnum) {
             if (!dstkey) {

-                addReplyBulk(c,ele);
+                if (encoding == REDIS_ENCODING_HT)
+                    addReplyBulk(c,eleobj);
+                else
+                    addReplyBulkLongLong(c,intobj);

                 cardinality++;
             } else {
                 cardinality++;
             } else {

-                setTypeAdd(dstset,ele);
+                if (encoding == REDIS_ENCODING_INTSET) {
+                    eleobj = createStringObjectFromLongLong(intobj);
+                    setTypeAdd(dstset,eleobj);
+                    decrRefCount(eleobj);
+                } else {
+                    setTypeAdd(dstset,eleobj);
+                }

             }
         }
             }
         }

-        decrRefCount(ele);

     }
     setTypeReleaseIterator(si);
 
     }
     setTypeReleaseIterator(si);
 


@@ -399,7 +502,7 @@ void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum,
             decrRefCount(dstset);
             addReply(c,shared.czero);
         }
             decrRefCount(dstset);
             addReply(c,shared.czero);
         }

-        touchWatchedKey(c->db,dstkey);
+        signalModifiedKey(c->db,dstkey);

         server.dirty++;
     } else {
         setDeferredMultiBulkLength(c,replylen,cardinality);
         server.dirty++;
     } else {
         setDeferredMultiBulkLength(c,replylen,cardinality);


@@ -452,7 +555,7 @@ void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *
         if (!sets[j]) continue; /* non existing keys are like empty sets */
 
         si = setTypeInitIterator(sets[j]);
         if (!sets[j]) continue; /* non existing keys are like empty sets */
 
         si = setTypeInitIterator(sets[j]);

-        while((ele = setTypeNext(si)) != NULL) {
+        while((ele = setTypeNextObject(si)) != NULL) {

             if (op == REDIS_OP_UNION || j == 0) {
                 if (setTypeAdd(dstset,ele)) {
                     cardinality++;
             if (op == REDIS_OP_UNION || j == 0) {
                 if (setTypeAdd(dstset,ele)) {
                     cardinality++;


@@ -474,7 +577,7 @@ void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *
     if (!dstkey) {
         addReplyMultiBulkLen(c,cardinality);
         si = setTypeInitIterator(dstset);
     if (!dstkey) {
         addReplyMultiBulkLen(c,cardinality);
         si = setTypeInitIterator(dstset);

-        while((ele = setTypeNext(si)) != NULL) {
+        while((ele = setTypeNextObject(si)) != NULL) {

             addReplyBulk(c,ele);
             decrRefCount(ele);
         }
             addReplyBulk(c,ele);
             decrRefCount(ele);
         }


@@ -491,7 +594,7 @@ void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *
             decrRefCount(dstset);
             addReply(c,shared.czero);
         }
             decrRefCount(dstset);
             addReply(c,shared.czero);
         }

-        touchWatchedKey(c->db,dstkey);
+        signalModifiedKey(c->db,dstkey);

         server.dirty++;
     }
     zfree(sets);
         server.dirty++;
     }
     zfree(sets);