X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/96ffb2fe97c3e77879e7a4f6f7457397a18bf233..9f9e1ceaa0d4451a21aa56739e63cca012399317:/src/t_set.c?ds=sidebyside

diff --git a/src/t_set.c b/src/t_set.c
index 3fbf13a3..e15952c0 100644
--- a/src/t_set.c
+++ b/src/t_set.c
@@ -8,7 +8,7 @@
  * an integer-encodable value, an intset will be returned. Otherwise a regular
  * hash table. */
 robj *setTypeCreate(robj *value) {
-    if (getLongLongFromObject(value,NULL) == REDIS_OK)
+    if (isObjectRepresentableAsLongLong(value,NULL) == REDIS_OK)
         return createIntsetObject();
     return createSetObject();
 }
@@ -21,7 +21,7 @@ int setTypeAdd(robj *subject, robj *value) {
             return 1;
         }
     } else if (subject->encoding == REDIS_ENCODING_INTSET) {
-        if (getLongLongFromObject(value,&llval) == REDIS_OK) {
+        if (isObjectRepresentableAsLongLong(value,&llval) == REDIS_OK) {
             uint8_t success = 0;
             subject->ptr = intsetAdd(subject->ptr,llval,&success);
             if (success) {
@@ -47,17 +47,17 @@ int setTypeAdd(robj *subject, robj *value) {
     return 0;
 }
 
-int setTypeRemove(robj *subject, robj *value) {
+int setTypeRemove(robj *setobj, robj *value) {
     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;
         }
-    } else if (subject->encoding == REDIS_ENCODING_INTSET) {
-        if (getLongLongFromObject(value,&llval) == REDIS_OK) {
-            uint8_t success;
-            subject->ptr = intsetRemove(subject->ptr,llval,&success);
+    } else if (setobj->encoding == REDIS_ENCODING_INTSET) {
+        if (isObjectRepresentableAsLongLong(value,&llval) == REDIS_OK) {
+            int success;
+            setobj->ptr = intsetRemove(setobj->ptr,llval,&success);
             if (success) return 1;
         }
     } else {
@@ -71,7 +71,7 @@ int setTypeIsMember(robj *subject, robj *value) {
     if (subject->encoding == REDIS_ENCODING_HT) {
         return dictFind((dict*)subject->ptr,value) != NULL;
     } else if (subject->encoding == REDIS_ENCODING_INTSET) {
-        if (getLongLongFromObject(value,&llval) == REDIS_OK) {
+        if (isObjectRepresentableAsLongLong(value,&llval) == REDIS_OK) {
             return intsetFind((intset*)subject->ptr,llval);
         }
     } else {
@@ -80,8 +80,8 @@ int setTypeIsMember(robj *subject, robj *value) {
     return 0;
 }
 
-setIterator *setTypeInitIterator(robj *subject) {
-    setIterator *si = zmalloc(sizeof(setIterator));
+setTypeIterator *setTypeInitIterator(robj *subject) {
+    setTypeIterator *si = zmalloc(sizeof(setTypeIterator));
     si->subject = subject;
     si->encoding = subject->encoding;
     if (si->encoding == REDIS_ENCODING_HT) {
@@ -94,7 +94,7 @@ setIterator *setTypeInitIterator(robj *subject) {
     return si;
 }
 
-void setTypeReleaseIterator(setIterator *si) {
+void setTypeReleaseIterator(setTypeIterator *si) {
     if (si->encoding == REDIS_ENCODING_HT)
         dictReleaseIterator(si->di);
     zfree(si);
@@ -103,7 +103,7 @@ void setTypeReleaseIterator(setIterator *si) {
 /* 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(setIterator *si) {
+robj *setTypeNext(setTypeIterator *si) {
     robj *ret = NULL;
     if (si->encoding == REDIS_ENCODING_HT) {
         dictEntry *de = dictNext(si->di);
@@ -112,7 +112,7 @@ robj *setTypeNext(setIterator *si) {
             incrRefCount(ret);
         }
     } else if (si->encoding == REDIS_ENCODING_INTSET) {
-        long long llval;
+        int64_t llval;
         if (intsetGet(si->subject->ptr,si->ii++,&llval))
             ret = createStringObjectFromLongLong(llval);
     }
@@ -120,21 +120,29 @@ robj *setTypeNext(setIterator *si) {
 }
 
 
-/* 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 long long 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
+ * long long 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, long long *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");
     }
-    return ret;
+    return setobj->encoding;
 }
 
 unsigned long setTypeSize(robj *subject) {
@@ -151,7 +159,7 @@ unsigned long setTypeSize(robj *subject) {
  * to a hashtable) is presized to hold the number of elements in the original
  * set. */
 void setTypeConvert(robj *subject, int enc) {
-    setIterator *si;
+    setTypeIterator *si;
     robj *element;
     redisAssert(subject->type == REDIS_SET);
 
@@ -178,6 +186,7 @@ void saddCommand(redisClient *c) {
     robj *set;
 
     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);
@@ -188,6 +197,7 @@ void saddCommand(redisClient *c) {
         }
     }
     if (setTypeAdd(set,c->argv[2])) {
+        touchWatchedKey(c->db,c->argv[1]);
         server.dirty++;
         addReply(c,shared.cone);
     } else {
@@ -201,8 +211,10 @@ void sremCommand(redisClient *c) {
     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 {
@@ -214,7 +226,7 @@ void smoveCommand(redisClient *c) {
     robj *srcset, *dstset, *ele;
     srcset = lookupKeyWrite(c->db,c->argv[1]);
     dstset = lookupKeyWrite(c->db,c->argv[2]);
-    ele = c->argv[3];
+    ele = c->argv[3] = tryObjectEncoding(c->argv[3]);
 
     /* If the source key does not exist return 0 */
     if (srcset == NULL) {
@@ -241,6 +253,8 @@ void smoveCommand(redisClient *c) {
 
     /* 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]);
     server.dirty++;
 
     /* Create the destination set when it doesn't exist */
@@ -260,6 +274,7 @@ void sismemberCommand(redisClient *c) {
     if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
         checkType(c,set,REDIS_SET)) return;
 
+    c->argv[2] = tryObjectEncoding(c->argv[2]);
     if (setTypeIsMember(set,c->argv[2]))
         addReply(c,shared.cone);
     else
@@ -272,39 +287,43 @@ void scardCommand(redisClient *c) {
     if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
         checkType(c,o,REDIS_SET)) return;
 
-    addReplyUlong(c,setTypeSize(o));
+    addReplyLongLong(c,setTypeSize(o));
 }
 
 void spopCommand(redisClient *c) {
     robj *set, *ele;
+    long long llele;
+    int encoding;
 
     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) {
+        addReplyBulkLongLong(c,llele);
+        set->ptr = intsetRemove(set->ptr,llele,NULL);
     } else {
-        setTypeRemove(set,ele);
         addReplyBulk(c,ele);
-        decrRefCount(ele);
-        if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
-        server.dirty++;
+        setTypeRemove(set,ele);
     }
+    if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
+    touchWatchedKey(c->db,c->argv[1]);
+    server.dirty++;
 }
 
 void srandmemberCommand(redisClient *c) {
     robj *set, *ele;
+    long long llele;
+    int encoding;
 
     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);
-        decrRefCount(ele);
     }
 }
 
@@ -314,8 +333,9 @@ 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);
-    setIterator *si;
-    robj *ele, *lenobj = NULL, *dstset = NULL;
+    setTypeIterator *si;
+    robj *ele, *dstset = NULL;
+    void *replylen = NULL;
     unsigned long j, cardinality = 0;
 
     for (j = 0; j < setnum; j++) {
@@ -325,8 +345,10 @@ void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum,
         if (!setobj) {
             zfree(sets);
             if (dstkey) {
-                if (dbDelete(c->db,dstkey))
+                if (dbDelete(c->db,dstkey)) {
+                    touchWatchedKey(c->db,dstkey);
                     server.dirty++;
+                }
                 addReply(c,shared.czero);
             } else {
                 addReply(c,shared.emptymultibulk);
@@ -349,9 +371,7 @@ void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum,
      * 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);
+        replylen = addDeferredMultiBulkLength(c);
     } else {
         /* If we have a target key where to store the resulting set
          * create this key with an empty set inside */
@@ -390,9 +410,10 @@ void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum,
             decrRefCount(dstset);
             addReply(c,shared.czero);
         }
+        touchWatchedKey(c->db,dstkey);
         server.dirty++;
     } else {
-        lenobj->ptr = sdscatprintf(sdsempty(),"*%lu\r\n",cardinality);
+        setDeferredMultiBulkLength(c,replylen,cardinality);
     }
     zfree(sets);
 }
@@ -411,7 +432,7 @@ void sinterstoreCommand(redisClient *c) {
 
 void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *dstkey, int op) {
     robj **sets = zmalloc(sizeof(robj*)*setnum);
-    setIterator *si;
+    setTypeIterator *si;
     robj *ele, *dstset = NULL;
     int j, cardinality = 0;
 
@@ -462,7 +483,7 @@ void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *
 
     /* Output the content of the resulting set, if not in STORE mode */
     if (!dstkey) {
-        addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",cardinality));
+        addReplyMultiBulkLen(c,cardinality);
         si = setTypeInitIterator(dstset);
         while((ele = setTypeNext(si)) != NULL) {
             addReplyBulk(c,ele);
@@ -481,6 +502,7 @@ void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *
             decrRefCount(dstset);
             addReply(c,shared.czero);
         }
+        touchWatchedKey(c->db,dstkey);
         server.dirty++;
     }
     zfree(sets);