Properly init/release iterators in zunionInterGenericCommand().

This commit does mainly two things:

1) It fixes zunionInterGenericCommand() by removing mass-initialization
of all the iterators used, so that we don't violate the unsafe iterator
API of dictionaries. This fixes issue #1240.

2) Since the zui* APIs required the allocator to be initialized in the
zsetopsrc structure in order to use non-iterator related APIs, this
commit fixes this strict requirement by accessing objects directly via
the op->subject->ptr pointer we have to the object.

1 files changed, 18 insertions, 19 deletions

diff --git a/src/t_zset.c b/src/t_zset.c
index 0d92f681f..f7ec467eb 100644
--- a/src/t_zset.c
+++ b/src/t_zset.c
@@ -1251,20 +1251,20 @@ int zuiLength(zsetopsrc *op) {
         return 0;
 
     if (op->type == REDIS_SET) {
-        iterset *it = &op->iter.set;
         if (op->encoding == REDIS_ENCODING_INTSET) {
-            return intsetLen(it->is.is);
+            return intsetLen(op->subject->ptr);
         } else if (op->encoding == REDIS_ENCODING_HT) {
-            return dictSize(it->ht.dict);
+            dict *ht = op->subject->ptr;
+            return dictSize(ht);
         } else {
             redisPanic("Unknown set encoding");
         }
     } else if (op->type == REDIS_ZSET) {
-        iterzset *it = &op->iter.zset;
         if (op->encoding == REDIS_ENCODING_ZIPLIST) {
-            return zzlLength(it->zl.zl);
+            return zzlLength(op->subject->ptr);
         } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
-            return it->sl.zs->zsl->length;
+            zset *zs = op->subject->ptr;
+            return zs->zsl->length;
         } else {
             redisPanic("Unknown sorted set encoding");
         }
@@ -1400,18 +1400,19 @@ int zuiFind(zsetopsrc *op, zsetopval *val, double *score) {
         return 0;
 
     if (op->type == REDIS_SET) {
-        iterset *it = &op->iter.set;
-
         if (op->encoding == REDIS_ENCODING_INTSET) {
-            if (zuiLongLongFromValue(val) && intsetFind(it->is.is,val->ell)) {
+            if (zuiLongLongFromValue(val) &&
+                intsetFind(op->subject->ptr,val->ell))
+            {
                 *score = 1.0;
                 return 1;
             } else {
                 return 0;
             }
         } else if (op->encoding == REDIS_ENCODING_HT) {
+            dict *ht = op->subject->ptr;
             zuiObjectFromValue(val);
-            if (dictFind(it->ht.dict,val->ele) != NULL) {
+            if (dictFind(ht,val->ele) != NULL) {
                 *score = 1.0;
                 return 1;
             } else {
@@ -1421,19 +1422,19 @@ int zuiFind(zsetopsrc *op, zsetopval *val, double *score) {
             redisPanic("Unknown set encoding");
         }
     } else if (op->type == REDIS_ZSET) {
-        iterzset *it = &op->iter.zset;
         zuiObjectFromValue(val);
 
         if (op->encoding == REDIS_ENCODING_ZIPLIST) {
-            if (zzlFind(it->zl.zl,val->ele,score) != NULL) {
+            if (zzlFind(op->subject->ptr,val->ele,score) != NULL) {
                 /* Score is already set by zzlFind. */
                 return 1;
             } else {
                 return 0;
             }
         } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
+            zset *zs = op->subject->ptr;
             dictEntry *de;
-            if ((de = dictFind(it->sl.zs->dict,val->ele)) != NULL) {
+            if ((de = dictFind(zs->dict,val->ele)) != NULL) {
                 *score = *(double*)dictGetVal(de);
                 return 1;
             } else {
@@ -1561,9 +1562,6 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
         }
     }
 
-    for (i = 0; i < setnum; i++)
-        zuiInitIterator(&src[i]);
-
     /* sort sets from the smallest to largest, this will improve our
      * algorithm's performance */
     qsort(src,setnum,sizeof(zsetopsrc),zuiCompareByCardinality);
@@ -1577,6 +1575,7 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
         if (zuiLength(&src[0]) > 0) {
             /* Precondition: as src[0] is non-empty and the inputs are ordered
              * by size, all src[i > 0] are non-empty too. */
+            zuiInitIterator(&src[0]);
             while (zuiNext(&src[0],&zval)) {
                 double score, value;
 
@@ -1610,12 +1609,14 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
                             maxelelen = sdslen(tmp->ptr);
                 }
             }
+            zuiClearIterator(&src[0]);
         }
     } else if (op == REDIS_OP_UNION) {
         for (i = 0; i < setnum; i++) {
             if (zuiLength(&src[i]) == 0)
                 continue;
 
+            zuiInitIterator(&src[i]);
             while (zuiNext(&src[i],&zval)) {
                 double score, value;
 
@@ -1651,14 +1652,12 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
                     if (sdslen(tmp->ptr) > maxelelen)
                         maxelelen = sdslen(tmp->ptr);
             }
+            zuiClearIterator(&src[i]);
         }
     } else {
         redisPanic("Unknown operator");
     }
 
-    for (i = 0; i < setnum; i++)
-        zuiClearIterator(&src[i]);
-
     if (dbDelete(c->db,dstkey)) {
         signalModifiedKey(c->db,dstkey);
         touched = 1;