Rewrite comment that was no longer valid

[redis.git] / src / t_zset.c

diff --git a/src/t_zset.c b/src/t_zset.c
index 503486385adbd7700a697f436a9c8b29f5460ad1..ba05c27898b4d45bf0e823e4843c1278880860cf 100644 (file)
--- a/src/t_zset.c
+++ b/src/t_zset.c
@@ -71,7 +71,7 @@ int zslRandomLevel(void) {
     return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
 }
 
     return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
 }
 

-void zslInsert(zskiplist *zsl, double score, robj *obj) {
+zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {

     zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
     unsigned int rank[ZSKIPLIST_MAXLEVEL];
     int i, level;
     zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
     unsigned int rank[ZSKIPLIST_MAXLEVEL];
     int i, level;


@@ -123,6 +123,7 @@ void zslInsert(zskiplist *zsl, double score, robj *obj) {
     else
         zsl->tail = x;
     zsl->length++;
     else
         zsl->tail = x;
     zsl->length++;

+    return x;

 }
 
 /* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */
 }
 
 /* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */


@@ -193,7 +194,7 @@ unsigned long zslDeleteRangeByScore(zskiplist *zsl, double min, double max, dict
     x = x->level[0].forward;
     while (x && x->score <= max) {
         zskiplistNode *next = x->level[0].forward;
     x = x->level[0].forward;
     while (x && x->score <= max) {
         zskiplistNode *next = x->level[0].forward;

-        zslDeleteNode(zsl, x, update);
+        zslDeleteNode(zsl,x,update);

         dictDelete(dict,x->obj);
         zslFreeNode(x);
         removed++;
         dictDelete(dict,x->obj);
         zslFreeNode(x);
         removed++;


@@ -222,7 +223,7 @@ unsigned long zslDeleteRangeByRank(zskiplist *zsl, unsigned int start, unsigned
     x = x->level[0].forward;
     while (x && traversed <= end) {
         zskiplistNode *next = x->level[0].forward;
     x = x->level[0].forward;
     while (x && traversed <= end) {
         zskiplistNode *next = x->level[0].forward;

-        zslDeleteNode(zsl, x, update);
+        zslDeleteNode(zsl,x,update);

         dictDelete(dict,x->obj);
         zslFreeNode(x);
         removed++;
         dictDelete(dict,x->obj);
         zslFreeNode(x);
         removed++;


@@ -299,13 +300,11 @@ zskiplistNode* zslistTypeGetElementByRank(zskiplist *zsl, unsigned long rank) {
  * Sorted set commands 
  *----------------------------------------------------------------------------*/
 
  * Sorted set commands 
  *----------------------------------------------------------------------------*/
 

-/* This generic command implements both ZADD and ZINCRBY.
- * scoreval is the score if the operation is a ZADD (doincrement == 0) or
- * the increment if the operation is a ZINCRBY (doincrement == 1). */
-void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double scoreval, int doincrement) {
+/* This generic command implements both ZADD and ZINCRBY. */
+void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double score, int incr) {

     robj *zsetobj;
     zset *zs;
     robj *zsetobj;
     zset *zs;

-    double *score;
+    zskiplistNode *znode;

 
     zsetobj = lookupKeyWrite(c->db,key);
     if (zsetobj == NULL) {
 
     zsetobj = lookupKeyWrite(c->db,key);
     if (zsetobj == NULL) {


@@ -319,72 +318,72 @@ void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double scoreval, i
     }
     zs = zsetobj->ptr;
 
     }
     zs = zsetobj->ptr;
 

-    /* Ok now since we implement both ZADD and ZINCRBY here the code
-     * needs to handle the two different conditions. It's all about setting
-     * '*score', that is, the new score to set, to the right value. */
-    score = zmalloc(sizeof(double));
-    if (doincrement) {
-        dictEntry *de;
-
+    /* Since both ZADD and ZINCRBY are implemented here, we need to increment
+     * the score first by the current score if ZINCRBY is called. */
+    if (incr) {

         /* Read the old score. If the element was not present starts from 0 */
         /* Read the old score. If the element was not present starts from 0 */

-        de = dictFind(zs->dict,ele);
-        if (de) {
-            double *oldscore = dictGetEntryVal(de);
-            *score = *oldscore + scoreval;
-        } else {
-            *score = scoreval;
-        }
-        if (isnan(*score)) {
-            addReplySds(c,
-                sdsnew("-ERR resulting score is not a number (NaN)\r\n"));
-            zfree(score);
+        dictEntry *de = dictFind(zs->dict,ele);
+        if (de != NULL)
+            score += *(double*)dictGetEntryVal(de);
+
+        if (isnan(score)) {
+            addReplyError(c,"resulting score is not a number (NaN)");

             /* Note that we don't need to check if the zset may be empty and
              * should be removed here, as we can only obtain Nan as score if
              * there was already an element in the sorted set. */
             return;
         }
             /* Note that we don't need to check if the zset may be empty and
              * should be removed here, as we can only obtain Nan as score if
              * there was already an element in the sorted set. */
             return;
         }

-    } else {
-        *score = scoreval;

     }
 
     }
 

-    /* What follows is a simple remove and re-insert operation that is common
-     * to both ZADD and ZINCRBY... */
-    if (dictAdd(zs->dict,ele,score) == DICT_OK) {
-        /* case 1: New element */
+    /* We need to remove and re-insert the element when it was already present
+     * in the dictionary, to update the skiplist. Note that we delay adding a
+     * pointer to the score because we want to reference the score in the
+     * skiplist node. */
+    if (dictAdd(zs->dict,ele,NULL) == DICT_OK) {
+        dictEntry *de;
+
+        /* New element */

         incrRefCount(ele); /* added to hash */
         incrRefCount(ele); /* added to hash */

-        zslInsert(zs->zsl,*score,ele);
+        znode = zslInsert(zs->zsl,score,ele);

         incrRefCount(ele); /* added to skiplist */
         incrRefCount(ele); /* added to skiplist */

+
+        /* Update the score in the dict entry */
+        de = dictFind(zs->dict,ele);
+        redisAssert(de != NULL);
+        dictGetEntryVal(de) = &znode->score;

         touchWatchedKey(c->db,c->argv[1]);
         server.dirty++;
         touchWatchedKey(c->db,c->argv[1]);
         server.dirty++;

-        if (doincrement)
-            addReplyDouble(c,*score);
+        if (incr)
+            addReplyDouble(c,score);

         else
             addReply(c,shared.cone);
     } else {
         dictEntry *de;
         else
             addReply(c,shared.cone);
     } else {
         dictEntry *de;

-        double *oldscore;
+        robj *curobj;
+        double *curscore;
+        int deleted;

 
 

-        /* case 2: Score update operation */
+        /* Update score */

         de = dictFind(zs->dict,ele);
         redisAssert(de != NULL);
         de = dictFind(zs->dict,ele);
         redisAssert(de != NULL);

-        oldscore = dictGetEntryVal(de);
-        if (*score != *oldscore) {
-            int deleted;
+        curobj = dictGetEntryKey(de);
+        curscore = dictGetEntryVal(de);

 
 

-            /* Remove and insert the element in the skip list with new score */
-            deleted = zslDelete(zs->zsl,*oldscore,ele);
+        /* When the score is updated, reuse the existing string object to
+         * prevent extra alloc/dealloc of strings on ZINCRBY. */
+        if (score != *curscore) {
+            deleted = zslDelete(zs->zsl,*curscore,curobj);

             redisAssert(deleted != 0);
             redisAssert(deleted != 0);

-            zslInsert(zs->zsl,*score,ele);
-            incrRefCount(ele);
-            /* Update the score in the hash table */
-            dictReplace(zs->dict,ele,score);
+            znode = zslInsert(zs->zsl,score,curobj);
+            incrRefCount(curobj);
+
+            /* Update the score in the current dict entry */
+            dictGetEntryVal(de) = &znode->score;

             touchWatchedKey(c->db,c->argv[1]);
             server.dirty++;
             touchWatchedKey(c->db,c->argv[1]);
             server.dirty++;

-        } else {
-            zfree(score);

         }
         }

-        if (doincrement)
-            addReplyDouble(c,*score);
+        if (incr)
+            addReplyDouble(c,score);

         else
             addReply(c,shared.czero);
     }
         else
             addReply(c,shared.czero);
     }


@@ -393,12 +392,14 @@ void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double scoreval, i
 void zaddCommand(redisClient *c) {
     double scoreval;
     if (getDoubleFromObjectOrReply(c,c->argv[2],&scoreval,NULL) != REDIS_OK) return;
 void zaddCommand(redisClient *c) {
     double scoreval;
     if (getDoubleFromObjectOrReply(c,c->argv[2],&scoreval,NULL) != REDIS_OK) return;

+    c->argv[3] = tryObjectEncoding(c->argv[3]);

     zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,0);
 }
 
 void zincrbyCommand(redisClient *c) {
     double scoreval;
     if (getDoubleFromObjectOrReply(c,c->argv[2],&scoreval,NULL) != REDIS_OK) return;
     zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,0);
 }
 
 void zincrbyCommand(redisClient *c) {
     double scoreval;
     if (getDoubleFromObjectOrReply(c,c->argv[2],&scoreval,NULL) != REDIS_OK) return;

+    c->argv[3] = tryObjectEncoding(c->argv[3]);

     zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,1);
 }
 
     zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,1);
 }
 


@@ -406,21 +407,22 @@ void zremCommand(redisClient *c) {
     robj *zsetobj;
     zset *zs;
     dictEntry *de;
     robj *zsetobj;
     zset *zs;
     dictEntry *de;

-    double *oldscore;
+    double curscore;

     int deleted;
 
     if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
         checkType(c,zsetobj,REDIS_ZSET)) return;
 
     zs = zsetobj->ptr;
     int deleted;
 
     if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
         checkType(c,zsetobj,REDIS_ZSET)) return;
 
     zs = zsetobj->ptr;

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

     de = dictFind(zs->dict,c->argv[2]);
     if (de == NULL) {
         addReply(c,shared.czero);
         return;
     }
     /* Delete from the skiplist */
     de = dictFind(zs->dict,c->argv[2]);
     if (de == NULL) {
         addReply(c,shared.czero);
         return;
     }
     /* Delete from the skiplist */

-    oldscore = dictGetEntryVal(de);
-    deleted = zslDelete(zs->zsl,*oldscore,c->argv[2]);
+    curscore = *(double*)dictGetEntryVal(de);
+    deleted = zslDelete(zs->zsl,curscore,c->argv[2]);

     redisAssert(deleted != 0);
 
     /* Delete from the hash table */
     redisAssert(deleted != 0);
 
     /* Delete from the hash table */


@@ -534,6 +536,7 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
     zsetopsrc *src;
     robj *dstobj;
     zset *dstzset;
     zsetopsrc *src;
     robj *dstobj;
     zset *dstzset;

+    zskiplistNode *znode;

     dictIterator *di;
     dictEntry *de;
     int touched = 0;
     dictIterator *di;
     dictEntry *de;
     int touched = 0;


@@ -541,7 +544,8 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
     /* expect setnum input keys to be given */
     setnum = atoi(c->argv[2]->ptr);
     if (setnum < 1) {
     /* expect setnum input keys to be given */
     setnum = atoi(c->argv[2]->ptr);
     if (setnum < 1) {

-        addReplySds(c,sdsnew("-ERR at least 1 input key is needed for ZUNIONSTORE/ZINTERSTORE\r\n"));
+        addReplyError(c,
+            "at least 1 input key is needed for ZUNIONSTORE/ZINTERSTORE");

         return;
     }
 
         return;
     }
 


@@ -624,28 +628,26 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
              * from small to large, all src[i > 0].dict are non-empty too */
             di = dictGetIterator(src[0].dict);
             while((de = dictNext(di)) != NULL) {
              * from small to large, all src[i > 0].dict are non-empty too */
             di = dictGetIterator(src[0].dict);
             while((de = dictNext(di)) != NULL) {

-                double *score = zmalloc(sizeof(double)), value;
-                *score = src[0].weight * zunionInterDictValue(de);
+                double score, value;

 
 

+                score = src[0].weight * zunionInterDictValue(de);

                 for (j = 1; j < setnum; j++) {
                     dictEntry *other = dictFind(src[j].dict,dictGetEntryKey(de));
                     if (other) {
                         value = src[j].weight * zunionInterDictValue(other);
                 for (j = 1; j < setnum; j++) {
                     dictEntry *other = dictFind(src[j].dict,dictGetEntryKey(de));
                     if (other) {
                         value = src[j].weight * zunionInterDictValue(other);

-                        zunionInterAggregate(score, value, aggregate);
+                        zunionInterAggregate(&score, value, aggregate);

                     } else {
                         break;
                     }
                 }
 
                     } else {
                         break;
                     }
                 }
 

-                /* skip entry when not present in every source dict */
-                if (j != setnum) {
-                    zfree(score);
-                } else {
+                /* accept entry only when present in every source dict */
+                if (j == setnum) {

                     robj *o = dictGetEntryKey(de);
                     robj *o = dictGetEntryKey(de);

-                    dictAdd(dstzset->dict,o,score);
-                    incrRefCount(o); /* added to dictionary */
-                    zslInsert(dstzset->zsl,*score,o);
+                    znode = zslInsert(dstzset->zsl,score,o);

                     incrRefCount(o); /* added to skiplist */
                     incrRefCount(o); /* added to skiplist */

+                    dictAdd(dstzset->dict,o,&znode->score);
+                    incrRefCount(o); /* added to dictionary */

                 }
             }
             dictReleaseIterator(di);
                 }
             }
             dictReleaseIterator(di);


@@ -656,11 +658,12 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
 
             di = dictGetIterator(src[i].dict);
             while((de = dictNext(di)) != NULL) {
 
             di = dictGetIterator(src[i].dict);
             while((de = dictNext(di)) != NULL) {

-                /* skip key when already processed */
-                if (dictFind(dstzset->dict,dictGetEntryKey(de)) != NULL) continue;
+                double score, value;

 
 

-                double *score = zmalloc(sizeof(double)), value;
-                *score = src[i].weight * zunionInterDictValue(de);
+                /* skip key when already processed */
+                if (dictFind(dstzset->dict,dictGetEntryKey(de)) != NULL)
+                    continue;
+                score = src[i].weight * zunionInterDictValue(de);

 
                 /* because the zsets are sorted by size, its only possible
                  * for sets at larger indices to hold this entry */
 
                 /* because the zsets are sorted by size, its only possible
                  * for sets at larger indices to hold this entry */


@@ -668,15 +671,15 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
                     dictEntry *other = dictFind(src[j].dict,dictGetEntryKey(de));
                     if (other) {
                         value = src[j].weight * zunionInterDictValue(other);
                     dictEntry *other = dictFind(src[j].dict,dictGetEntryKey(de));
                     if (other) {
                         value = src[j].weight * zunionInterDictValue(other);

-                        zunionInterAggregate(score, value, aggregate);
+                        zunionInterAggregate(&score, value, aggregate);

                     }
                 }
 
                 robj *o = dictGetEntryKey(de);
                     }
                 }
 
                 robj *o = dictGetEntryKey(de);

-                dictAdd(dstzset->dict,o,score);
-                incrRefCount(o); /* added to dictionary */
-                zslInsert(dstzset->zsl,*score,o);
+                znode = zslInsert(dstzset->zsl,score,o);

                 incrRefCount(o); /* added to skiplist */
                 incrRefCount(o); /* added to skiplist */

+                dictAdd(dstzset->dict,o,&znode->score);
+                incrRefCount(o); /* added to dictionary */

             }
             dictReleaseIterator(di);
         }
             }
             dictReleaseIterator(di);
         }


@@ -762,8 +765,7 @@ void zrangeGenericCommand(redisClient *c, int reverse) {
     }
 
     /* Return the result in form of a multi-bulk reply */
     }
 
     /* Return the result in form of a multi-bulk reply */

-    addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",
-        withscores ? (rangelen*2) : rangelen));
+    addReplyMultiBulkLen(c,withscores ? (rangelen*2) : rangelen);

     for (j = 0; j < rangelen; j++) {
         ele = ln->obj;
         addReplyBulk(c,ele);
     for (j = 0; j < rangelen; j++) {
         ele = ln->obj;
         addReplyBulk(c,ele);


@@ -820,8 +822,7 @@ void genericZrangebyscoreCommand(redisClient *c, int justcount) {
     if (c->argc != (4 + withscores) && c->argc != (7 + withscores))
         badsyntax = 1;
     if (badsyntax) {
     if (c->argc != (4 + withscores) && c->argc != (7 + withscores))
         badsyntax = 1;
     if (badsyntax) {

-        addReplySds(c,
-            sdsnew("-ERR wrong number of arguments for ZRANGEBYSCORE\r\n"));
+        addReplyError(c,"wrong number of arguments for ZRANGEBYSCORE");

         return;
     }
 
         return;
     }
 


@@ -846,7 +847,8 @@ void genericZrangebyscoreCommand(redisClient *c, int justcount) {
             zset *zsetobj = o->ptr;
             zskiplist *zsl = zsetobj->zsl;
             zskiplistNode *ln;
             zset *zsetobj = o->ptr;
             zskiplist *zsl = zsetobj->zsl;
             zskiplistNode *ln;

-            robj *ele, *lenobj = NULL;
+            robj *ele;
+            void *replylen = NULL;

             unsigned long rangelen = 0;
 
             /* Get the first node with the score >= min, or with
             unsigned long rangelen = 0;
 
             /* Get the first node with the score >= min, or with


@@ -864,11 +866,8 @@ void genericZrangebyscoreCommand(redisClient *c, int justcount) {
              * are in the list, so we push this object that will represent
              * the multi-bulk length in the output buffer, and will "fix"
              * it later */
              * are in the list, so we push this object that will represent
              * the multi-bulk length in the output buffer, and will "fix"
              * it later */

-            if (!justcount) {
-                lenobj = createObject(REDIS_STRING,NULL);
-                addReply(c,lenobj);
-                decrRefCount(lenobj);
-            }
+            if (!justcount)
+                replylen = addDeferredMultiBulkLength(c);

 
             while(ln && (maxex ? (ln->score < max) : (ln->score <= max))) {
                 if (offset) {
 
             while(ln && (maxex ? (ln->score < max) : (ln->score <= max))) {
                 if (offset) {


@@ -890,7 +889,7 @@ void genericZrangebyscoreCommand(redisClient *c, int justcount) {
             if (justcount) {
                 addReplyLongLong(c,(long)rangelen);
             } else {
             if (justcount) {
                 addReplyLongLong(c,(long)rangelen);
             } else {

-                lenobj->ptr = sdscatprintf(sdsempty(),"*%lu\r\n",
+                setDeferredMultiBulkLength(c,replylen,

                      withscores ? (rangelen*2) : rangelen);
             }
         }
                      withscores ? (rangelen*2) : rangelen);
             }
         }


@@ -913,7 +912,7 @@ void zcardCommand(redisClient *c) {
         checkType(c,o,REDIS_ZSET)) return;
 
     zs = o->ptr;
         checkType(c,o,REDIS_ZSET)) return;
 
     zs = o->ptr;

-    addReplyUlong(c,zs->zsl->length);
+    addReplyLongLong(c,zs->zsl->length);

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


@@ -925,6 +924,7 @@ void zscoreCommand(redisClient *c) {
         checkType(c,o,REDIS_ZSET)) return;
 
     zs = o->ptr;
         checkType(c,o,REDIS_ZSET)) return;
 
     zs = o->ptr;

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

     de = dictFind(zs->dict,c->argv[2]);
     if (!de) {
         addReply(c,shared.nullbulk);
     de = dictFind(zs->dict,c->argv[2]);
     if (!de) {
         addReply(c,shared.nullbulk);


@@ -948,6 +948,7 @@ void zrankGenericCommand(redisClient *c, int reverse) {
 
     zs = o->ptr;
     zsl = zs->zsl;
 
     zs = o->ptr;
     zsl = zs->zsl;

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

     de = dictFind(zs->dict,c->argv[2]);
     if (!de) {
         addReply(c,shared.nullbulk);
     de = dictFind(zs->dict,c->argv[2]);
     if (!de) {
         addReply(c,shared.nullbulk);