Merge remote branch 'pietern/unstable-replnonblock' into unstable

[redis.git] / src / t_zset.c

diff --git a/src/t_zset.c b/src/t_zset.c
index 65c9962258f28857938440c6ddb1c8dbfd10d393..a5dc27c7685e7677f0f138d9284ddc41b25a68ab 100644 (file)
--- a/src/t_zset.c
+++ b/src/t_zset.c
@@ -174,12 +174,6 @@ int zslDelete(zskiplist *zsl, double score, robj *obj) {
     return 0; /* not found */
 }
 
     return 0; /* not found */
 }
 

-/* Struct to hold a inclusive/exclusive range spec. */
-typedef struct {
-    double min, max;
-    int minex, maxex; /* are min or max exclusive? */
-} zrangespec;
-

 static int zslValueGteMin(double value, zrangespec *spec) {
     return spec->minex ? (value > spec->min) : (value >= spec->min);
 }
 static int zslValueGteMin(double value, zrangespec *spec) {
     return spec->minex ? (value > spec->min) : (value >= spec->min);
 }


@@ -188,10 +182,6 @@ static int zslValueLteMax(double value, zrangespec *spec) {
     return spec->maxex ? (value < spec->max) : (value <= spec->max);
 }
 
     return spec->maxex ? (value < spec->max) : (value <= spec->max);
 }
 

-static int zslValueInRange(double value, zrangespec *spec) {
-    return zslValueGteMin(value,spec) && zslValueLteMax(value,spec);
-}
-

 /* Returns if there is a part of the zset is in range. */
 int zslIsInRange(zskiplist *zsl, zrangespec *range) {
     zskiplistNode *x;
 /* Returns if there is a part of the zset is in range. */
 int zslIsInRange(zskiplist *zsl, zrangespec *range) {
     zskiplistNode *x;


@@ -226,10 +216,12 @@ zskiplistNode *zslFirstInRange(zskiplist *zsl, zrangespec range) {
                 x = x->level[i].forward;
     }
 
                 x = x->level[i].forward;
     }
 

-    /* The tail is in range, so the previous block should always return a
-     * node that is non-NULL and the last one to be out of range. */
+    /* This is an inner range, so the next node cannot be NULL. */

     x = x->level[0].forward;
     x = x->level[0].forward;

-    redisAssert(x != NULL && zslValueInRange(x->score,&range));
+    redisAssert(x != NULL);
+
+    /* Check if score <= max. */
+    if (!zslValueLteMax(x->score,&range)) return NULL;

     return x;
 }
 
     return x;
 }
 


@@ -250,9 +242,11 @@ zskiplistNode *zslLastInRange(zskiplist *zsl, zrangespec range) {
                 x = x->level[i].forward;
     }
 
                 x = x->level[i].forward;
     }
 

-    /* The header is in range, so the previous block should always return a
-     * node that is non-NULL and in range. */
-    redisAssert(x != NULL && zslValueInRange(x->score,&range));
+    /* This is an inner range, so this node cannot be NULL. */
+    redisAssert(x != NULL);
+
+    /* Check if score >= min. */
+    if (!zslValueGteMin(x->score,&range)) return NULL;

     return x;
 }
 
     return x;
 }
 


@@ -531,8 +525,12 @@ unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec range) {
         redisAssert(sptr != NULL);
 
         score = zzlGetScore(sptr);
         redisAssert(sptr != NULL);
 
         score = zzlGetScore(sptr);

-        if (zslValueGteMin(score,&range))
-            return eptr;
+        if (zslValueGteMin(score,&range)) {
+            /* Check if score <= max. */
+            if (zslValueLteMax(score,&range))
+                return eptr;
+            return NULL;
+        }

 
         /* Move to next element. */
         eptr = ziplistNext(zl,sptr);
 
         /* Move to next element. */
         eptr = ziplistNext(zl,sptr);


@@ -555,8 +553,12 @@ unsigned char *zzlLastInRange(unsigned char *zl, zrangespec range) {
         redisAssert(sptr != NULL);
 
         score = zzlGetScore(sptr);
         redisAssert(sptr != NULL);
 
         score = zzlGetScore(sptr);

-        if (zslValueLteMax(score,&range))
-            return eptr;
+        if (zslValueLteMax(score,&range)) {
+            /* Check if score >= min. */
+            if (zslValueGteMin(score,&range))
+                return eptr;
+            return NULL;
+        }

 
         /* Move to previous element by moving to the score of previous element.
          * When this returns NULL, we know there also is no element. */
 
         /* Move to previous element by moving to the score of previous element.
          * When this returns NULL, we know there also is no element. */


@@ -608,7 +610,7 @@ unsigned char *zzlInsertAt(unsigned char *zl, unsigned char *eptr, robj *ele, do
     unsigned char *sptr;
     char scorebuf[128];
     int scorelen;
     unsigned char *sptr;
     char scorebuf[128];
     int scorelen;

-    int offset;
+    size_t offset;

 
     redisAssert(ele->encoding == REDIS_ENCODING_RAW);
     scorelen = d2string(scorebuf,sizeof(scorebuf),score);
 
     redisAssert(ele->encoding == REDIS_ENCODING_RAW);
     scorelen = d2string(scorebuf,sizeof(scorebuf),score);


@@ -713,7 +715,7 @@ unsigned int zsetLength(robj *zobj) {
     int length = -1;
     if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
         length = zzlLength(zobj->ptr);
     int length = -1;
     if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
         length = zzlLength(zobj->ptr);

-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {

         length = ((zset*)zobj->ptr)->zsl->length;
     } else {
         redisPanic("Unknown sorted set encoding");
         length = ((zset*)zobj->ptr)->zsl->length;
     } else {
         redisPanic("Unknown sorted set encoding");


@@ -735,7 +737,7 @@ void zsetConvert(robj *zobj, int encoding) {
         unsigned int vlen;
         long long vlong;
 
         unsigned int vlen;
         long long vlong;
 

-        if (encoding != REDIS_ENCODING_RAW)
+        if (encoding != REDIS_ENCODING_SKIPLIST)

             redisPanic("Unknown target encoding");
 
         zs = zmalloc(sizeof(*zs));
             redisPanic("Unknown target encoding");
 
         zs = zmalloc(sizeof(*zs));


@@ -764,8 +766,8 @@ void zsetConvert(robj *zobj, int encoding) {
 
         zfree(zobj->ptr);
         zobj->ptr = zs;
 
         zfree(zobj->ptr);
         zobj->ptr = zs;

-        zobj->encoding = REDIS_ENCODING_RAW;
-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+        zobj->encoding = REDIS_ENCODING_SKIPLIST;
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {

         unsigned char *zl = ziplistNew();
 
         if (encoding != REDIS_ENCODING_ZIPLIST)
         unsigned char *zl = ziplistNew();
 
         if (encoding != REDIS_ENCODING_ZIPLIST)


@@ -864,9 +866,9 @@ void zaddGenericCommand(redisClient *c, int incr) {
              * too long *before* executing zzlInsert. */
             zobj->ptr = zzlInsert(zobj->ptr,ele,score);
             if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
              * too long *before* executing zzlInsert. */
             zobj->ptr = zzlInsert(zobj->ptr,ele,score);
             if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)

-                zsetConvert(zobj,REDIS_ENCODING_RAW);
+                zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);

             if (sdslen(ele->ptr) > server.zset_max_ziplist_value)
             if (sdslen(ele->ptr) > server.zset_max_ziplist_value)

-                zsetConvert(zobj,REDIS_ENCODING_RAW);
+                zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);

 
             signalModifiedKey(c->db,key);
             server.dirty++;
 
             signalModifiedKey(c->db,key);
             server.dirty++;


@@ -876,7 +878,7 @@ void zaddGenericCommand(redisClient *c, int incr) {
             else /* ZADD */
                 addReply(c,shared.cone);
         }
             else /* ZADD */
                 addReply(c,shared.cone);
         }

-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {

         zset *zs = zobj->ptr;
         zskiplistNode *znode;
         dictEntry *de;
         zset *zs = zobj->ptr;
         zskiplistNode *znode;
         dictEntry *de;


@@ -959,7 +961,7 @@ void zremCommand(redisClient *c) {
             addReply(c,shared.czero);
             return;
         }
             addReply(c,shared.czero);
             return;
         }

-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {

         zset *zs = zobj->ptr;
         dictEntry *de;
         double score;
         zset *zs = zobj->ptr;
         dictEntry *de;
         double score;


@@ -1004,7 +1006,8 @@ void zremrangebyscoreCommand(redisClient *c) {
 
     if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
         zobj->ptr = zzlDeleteRangeByScore(zobj->ptr,range,&deleted);
 
     if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
         zobj->ptr = zzlDeleteRangeByScore(zobj->ptr,range,&deleted);

-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+        if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key);
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {

         zset *zs = zobj->ptr;
         deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict);
         if (htNeedsResize(zs->dict)) dictResize(zs->dict);
         zset *zs = zobj->ptr;
         deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict);
         if (htNeedsResize(zs->dict)) dictResize(zs->dict);


@@ -1049,7 +1052,8 @@ void zremrangebyrankCommand(redisClient *c) {
     if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
         /* Correct for 1-based rank. */
         zobj->ptr = zzlDeleteRangeByRank(zobj->ptr,start+1,end+1,&deleted);
     if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
         /* Correct for 1-based rank. */
         zobj->ptr = zzlDeleteRangeByRank(zobj->ptr,start+1,end+1,&deleted);

-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+        if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key);
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {

         zset *zs = zobj->ptr;
 
         /* Correct for 1-based rank. */
         zset *zs = zobj->ptr;
 
         /* Correct for 1-based rank. */


@@ -1149,7 +1153,7 @@ void zuiInitIterator(zsetopsrc *op) {
                 it->zl.sptr = ziplistNext(it->zl.zl,it->zl.eptr);
                 redisAssert(it->zl.sptr != NULL);
             }
                 it->zl.sptr = ziplistNext(it->zl.zl,it->zl.eptr);
                 redisAssert(it->zl.sptr != NULL);
             }

-        } else if (op->encoding == REDIS_ENCODING_RAW) {
+        } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {

             it->sl.zs = op->subject->ptr;
             it->sl.node = it->sl.zs->zsl->header->level[0].forward;
         } else {
             it->sl.zs = op->subject->ptr;
             it->sl.node = it->sl.zs->zsl->header->level[0].forward;
         } else {


@@ -1177,7 +1181,7 @@ void zuiClearIterator(zsetopsrc *op) {
         iterzset *it = &op->iter.zset;
         if (op->encoding == REDIS_ENCODING_ZIPLIST) {
             REDIS_NOTUSED(it); /* skip */
         iterzset *it = &op->iter.zset;
         if (op->encoding == REDIS_ENCODING_ZIPLIST) {
             REDIS_NOTUSED(it); /* skip */

-        } else if (op->encoding == REDIS_ENCODING_RAW) {
+        } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {

             REDIS_NOTUSED(it); /* skip */
         } else {
             redisPanic("Unknown sorted set encoding");
             REDIS_NOTUSED(it); /* skip */
         } else {
             redisPanic("Unknown sorted set encoding");


@@ -1204,7 +1208,7 @@ int zuiLength(zsetopsrc *op) {
         iterzset *it = &op->iter.zset;
         if (op->encoding == REDIS_ENCODING_ZIPLIST) {
             return zzlLength(it->zl.zl);
         iterzset *it = &op->iter.zset;
         if (op->encoding == REDIS_ENCODING_ZIPLIST) {
             return zzlLength(it->zl.zl);

-        } else if (op->encoding == REDIS_ENCODING_RAW) {
+        } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {

             return it->sl.zs->zsl->length;
         } else {
             redisPanic("Unknown sorted set encoding");
             return it->sl.zs->zsl->length;
         } else {
             redisPanic("Unknown sorted set encoding");


@@ -1229,7 +1233,7 @@ int zuiNext(zsetopsrc *op, zsetopval *val) {
     if (op->type == REDIS_SET) {
         iterset *it = &op->iter.set;
         if (op->encoding == REDIS_ENCODING_INTSET) {
     if (op->type == REDIS_SET) {
         iterset *it = &op->iter.set;
         if (op->encoding == REDIS_ENCODING_INTSET) {

-            if (!intsetGet(it->is.is,it->is.ii,&val->ell))
+            if (!intsetGet(it->is.is,it->is.ii,(int64_t*)&val->ell))

                 return 0;
             val->score = 1.0;
 
                 return 0;
             val->score = 1.0;
 


@@ -1257,7 +1261,7 @@ int zuiNext(zsetopsrc *op, zsetopval *val) {
 
             /* Move to next element. */
             zzlNext(it->zl.zl,&it->zl.eptr,&it->zl.sptr);
 
             /* Move to next element. */
             zzlNext(it->zl.zl,&it->zl.eptr,&it->zl.sptr);

-        } else if (op->encoding == REDIS_ENCODING_RAW) {
+        } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {

             if (it->sl.node == NULL)
                 return 0;
             val->ele = it->sl.node->obj;
             if (it->sl.node == NULL)
                 return 0;
             val->ele = it->sl.node->obj;


@@ -1369,7 +1373,7 @@ int zuiFind(zsetopsrc *op, zsetopval *val, double *score) {
             } else {
                 return 0;
             }
             } else {
                 return 0;
             }

-        } else if (op->encoding == REDIS_ENCODING_RAW) {
+        } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {

             dictEntry *de;
             if ((de = dictFind(it->sl.zs->dict,val->ele)) != NULL) {
                 *score = *(double*)dictGetEntryVal(de);
             dictEntry *de;
             if ((de = dictFind(it->sl.zs->dict,val->ele)) != NULL) {
                 *score = *(double*)dictGetEntryVal(de);


@@ -1505,6 +1509,7 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
 
     dstobj = createZsetObject();
     dstzset = dstobj->ptr;
 
     dstobj = createZsetObject();
     dstzset = dstobj->ptr;

+    memset(&zval, 0, sizeof(zval));

 
     if (op == REDIS_OP_INTER) {
         /* Skip everything if the smallest input is empty. */
 
     if (op == REDIS_OP_INTER) {
         /* Skip everything if the smallest input is empty. */


@@ -1516,7 +1521,12 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
 
                 score = src[0].weight * zval.score;
                 for (j = 1; j < setnum; j++) {
 
                 score = src[0].weight * zval.score;
                 for (j = 1; j < setnum; j++) {

-                    if (zuiFind(&src[j],&zval,&value)) {
+                    /* It is not safe to access the zset we are
+                     * iterating, so explicitly check for equal object. */
+                    if (src[j].subject == src[0].subject) {
+                        value = zval.score*src[j].weight;
+                        zunionInterAggregate(&score,value,aggregate);
+                    } else if (zuiFind(&src[j],&zval,&value)) {

                         value *= src[j].weight;
                         zunionInterAggregate(&score,value,aggregate);
                     } else {
                         value *= src[j].weight;
                         zunionInterAggregate(&score,value,aggregate);
                     } else {


@@ -1540,7 +1550,7 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
         }
     } else if (op == REDIS_OP_UNION) {
         for (i = 0; i < setnum; i++) {
         }
     } else if (op == REDIS_OP_UNION) {
         for (i = 0; i < setnum; i++) {

-            if (zuiLength(&src[0]) == 0)
+            if (zuiLength(&src[i]) == 0)

                 continue;
 
             while (zuiNext(&src[i],&zval)) {
                 continue;
 
             while (zuiNext(&src[i],&zval)) {


@@ -1556,7 +1566,12 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
                 /* Because the inputs are sorted by size, it's only possible
                  * for sets at larger indices to hold this element. */
                 for (j = (i+1); j < setnum; j++) {
                 /* Because the inputs are sorted by size, it's only possible
                  * for sets at larger indices to hold this element. */
                 for (j = (i+1); j < setnum; j++) {

-                    if (zuiFind(&src[j],&zval,&value)) {
+                    /* It is not safe to access the zset we are
+                     * iterating, so explicitly check for equal object. */
+                    if(src[j].subject == src[i].subject) {
+                        value = zval.score*src[j].weight;
+                        zunionInterAggregate(&score,value,aggregate);
+                    } else if (zuiFind(&src[j],&zval,&value)) {

                         value *= src[j].weight;
                         zunionInterAggregate(&score,value,aggregate);
                     }
                         value *= src[j].weight;
                         zunionInterAggregate(&score,value,aggregate);
                     }


@@ -1682,7 +1697,7 @@ void zrangeGenericCommand(redisClient *c, int reverse) {
                 zzlNext(zl,&eptr,&sptr);
         }
 
                 zzlNext(zl,&eptr,&sptr);
         }
 

-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {

         zset *zs = zobj->ptr;
         zskiplist *zsl = zs->zsl;
         zskiplistNode *ln;
         zset *zs = zobj->ptr;
         zskiplist *zsl = zs->zsl;
         zskiplistNode *ln;


@@ -1839,7 +1854,7 @@ void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) {
             else
                 zzlNext(zl,&eptr,&sptr);
         }
             else
                 zzlNext(zl,&eptr,&sptr);
         }

-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {

         zset *zs = zobj->ptr;
         zskiplist *zsl = zs->zsl;
         zskiplistNode *ln;
         zset *zs = zobj->ptr;
         zskiplist *zsl = zs->zsl;
         zskiplistNode *ln;


@@ -1933,7 +1948,7 @@ void zscoreCommand(redisClient *c) {
             addReplyDouble(c,score);
         else
             addReply(c,shared.nullbulk);
             addReplyDouble(c,score);
         else
             addReply(c,shared.nullbulk);

-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {

         zset *zs = zobj->ptr;
         dictEntry *de;
 
         zset *zs = zobj->ptr;
         dictEntry *de;
 


@@ -1987,7 +2002,7 @@ void zrankGenericCommand(redisClient *c, int reverse) {
         } else {
             addReply(c,shared.nullbulk);
         }
         } else {
             addReply(c,shared.nullbulk);
         }

-    } else if (zobj->encoding == REDIS_ENCODING_RAW) {
+    } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {

         zset *zs = zobj->ptr;
         zskiplist *zsl = zs->zsl;
         dictEntry *de;
         zset *zs = zobj->ptr;
         zskiplist *zsl = zs->zsl;
         dictEntry *de;