src/t_set.c

   1 #include "redis.h"
   2
   3 /*-----------------------------------------------------------------------------
   4  * Set Commands
   5  *----------------------------------------------------------------------------*/
   6
   7 void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *dstkey, int op);
   8
   9 /* Factory method to return a set that *can* hold "value". When the object has
  10  * an integer-encodable value, an intset will be returned. Otherwise a regular
  11  * hash table. */
  12 robj *setTypeCreate(robj *value) {
  13     if (isObjectRepresentableAsLongLong(value,NULL) == REDIS_OK)
  14         return createIntsetObject();
  15     return createSetObject();
  16 }
  17
  18 int setTypeAdd(robj *subject, robj *value) {
  19     long long llval;
  20     if (subject->encoding == REDIS_ENCODING_HT) {
  21         if (dictAdd(subject->ptr,value,NULL) == DICT_OK) {
  22             incrRefCount(value);
  23             return 1;
  24         }
  25     } else if (subject->encoding == REDIS_ENCODING_INTSET) {
  26         if (isObjectRepresentableAsLongLong(value,&llval) == REDIS_OK) {
  27             uint8_t success = 0;
  28             subject->ptr = intsetAdd(subject->ptr,llval,&success);
  29             if (success) {
  30                 /* Convert to regular set when the intset contains
  31                  * too many entries. */
  32                 if (intsetLen(subject->ptr) > server.set_max_intset_entries)
  33                     setTypeConvert(subject,REDIS_ENCODING_HT);
  34                 return 1;
  35             }
  36         } else {
  37             /* Failed to get integer from object, convert to regular set. */
  38             setTypeConvert(subject,REDIS_ENCODING_HT);
  39
  40             /* The set *was* an intset and this value is not integer
  41              * encodable, so dictAdd should always work. */
  42             redisAssertWithInfo(NULL,value,dictAdd(subject->ptr,value,NULL) == DICT_OK);
  43             incrRefCount(value);
  44             return 1;
  45         }
  46     } else {
  47         redisPanic("Unknown set encoding");
  48     }
  49     return 0;
  50 }
  51
  52 int setTypeRemove(robj *setobj, robj *value) {
  53     long long llval;
  54     if (setobj->encoding == REDIS_ENCODING_HT) {
  55         if (dictDelete(setobj->ptr,value) == DICT_OK) {
  56             if (htNeedsResize(setobj->ptr)) dictResize(setobj->ptr);
  57             return 1;
  58         }
  59     } else if (setobj->encoding == REDIS_ENCODING_INTSET) {
  60         if (isObjectRepresentableAsLongLong(value,&llval) == REDIS_OK) {
  61             int success;
  62             setobj->ptr = intsetRemove(setobj->ptr,llval,&success);
  63             if (success) return 1;
  64         }
  65     } else {
  66         redisPanic("Unknown set encoding");
  67     }
  68     return 0;
  69 }
  70
  71 int setTypeIsMember(robj *subject, robj *value) {
  72     long long llval;
  73     if (subject->encoding == REDIS_ENCODING_HT) {
  74         return dictFind((dict*)subject->ptr,value) != NULL;
  75     } else if (subject->encoding == REDIS_ENCODING_INTSET) {
  76         if (isObjectRepresentableAsLongLong(value,&llval) == REDIS_OK) {
  77             return intsetFind((intset*)subject->ptr,llval);
  78         }
  79     } else {
  80         redisPanic("Unknown set encoding");
  81     }
  82     return 0;
  83 }
  84
  85 setTypeIterator *setTypeInitIterator(robj *subject) {
  86     setTypeIterator *si = zmalloc(sizeof(setTypeIterator));
  87     si->subject = subject;
  88     si->encoding = subject->encoding;
  89     if (si->encoding == REDIS_ENCODING_HT) {
  90         si->di = dictGetIterator(subject->ptr);
  91     } else if (si->encoding == REDIS_ENCODING_INTSET) {
  92         si->ii = 0;
  93     } else {
  94         redisPanic("Unknown set encoding");
  95     }
  96     return si;
  97 }
  98
  99 void setTypeReleaseIterator(setTypeIterator *si) {
 100     if (si->encoding == REDIS_ENCODING_HT)
 101         dictReleaseIterator(si->di);
 102     zfree(si);
 103 }
 104
 105 /* Move to the next entry in the set. Returns the object at the current
 106  * position.
 107  *
 108  * Since set elements can be internally be stored as redis objects or
 109  * simple arrays of integers, setTypeNext returns the encoding of the
 110  * set object you are iterating, and will populate the appropriate pointer
 111  * (eobj) or (llobj) accordingly.
 112  *
 113  * When there are no longer elements -1 is returned.
 114  * Returned objects ref count is not incremented, so this function is
 115  * copy on write friendly. */
 116 int setTypeNext(setTypeIterator *si, robj **objele, int64_t *llele) {
 117     if (si->encoding == REDIS_ENCODING_HT) {
 118         dictEntry *de = dictNext(si->di);
 119         if (de == NULL) return -1;
 120         *objele = dictGetKey(de);
 121     } else if (si->encoding == REDIS_ENCODING_INTSET) {
 122         if (!intsetGet(si->subject->ptr,si->ii++,llele))
 123             return -1;
 124     }
 125     return si->encoding;
 126 }
 127
 128 /* The not copy on write friendly version but easy to use version
 129  * of setTypeNext() is setTypeNextObject(), returning new objects
 130  * or incrementing the ref count of returned objects. So if you don't
 131  * retain a pointer to this object you should call decrRefCount() against it.
 132  *
 133  * This function is the way to go for write operations where COW is not
 134  * an issue as the result will be anyway of incrementing the ref count. */
 135 robj *setTypeNextObject(setTypeIterator *si) {
 136     int64_t intele;
 137     robj *objele;
 138     int encoding;
 139
 140     encoding = setTypeNext(si,&objele,&intele);
 141     switch(encoding) {
 142         case -1:    return NULL;
 143         case REDIS_ENCODING_INTSET:
 144             return createStringObjectFromLongLong(intele);
 145         case REDIS_ENCODING_HT:
 146             incrRefCount(objele);
 147             return objele;
 148         default:
 149             redisPanic("Unsupported encoding");
 150     }
 151     return NULL; /* just to suppress warnings */
 152 }
 153
 154 /* Return random element from a non empty set.
 155  * The returned element can be a int64_t value if the set is encoded
 156  * as an "intset" blob of integers, or a redis object if the set
 157  * is a regular set.
 158  *
 159  * The caller provides both pointers to be populated with the right
 160  * object. The return value of the function is the object->encoding
 161  * field of the object and is used by the caller to check if the
 162  * int64_t pointer or the redis object pointere was populated.
 163  *
 164  * When an object is returned (the set was a real set) the ref count
 165  * of the object is not incremented so this function can be considered
 166  * copy on write friendly. */
 167 int setTypeRandomElement(robj *setobj, robj **objele, int64_t *llele) {
 168     if (setobj->encoding == REDIS_ENCODING_HT) {
 169         dictEntry *de = dictGetRandomKey(setobj->ptr);
 170         *objele = dictGetKey(de);
 171     } else if (setobj->encoding == REDIS_ENCODING_INTSET) {
 172         *llele = intsetRandom(setobj->ptr);
 173     } else {
 174         redisPanic("Unknown set encoding");
 175     }
 176     return setobj->encoding;
 177 }
 178
 179 unsigned long setTypeSize(robj *subject) {
 180     if (subject->encoding == REDIS_ENCODING_HT) {
 181         return dictSize((dict*)subject->ptr);
 182     } else if (subject->encoding == REDIS_ENCODING_INTSET) {
 183         return intsetLen((intset*)subject->ptr);
 184     } else {
 185         redisPanic("Unknown set encoding");
 186     }
 187 }
 188
 189 /* Convert the set to specified encoding. The resulting dict (when converting
 190  * to a hashtable) is presized to hold the number of elements in the original
 191  * set. */
 192 void setTypeConvert(robj *setobj, int enc) {
 193     setTypeIterator *si;
 194     redisAssertWithInfo(NULL,setobj,setobj->type == REDIS_SET &&
 195                              setobj->encoding == REDIS_ENCODING_INTSET);
 196
 197     if (enc == REDIS_ENCODING_HT) {
 198         int64_t intele;
 199         dict *d = dictCreate(&setDictType,NULL);
 200         robj *element;
 201
 202         /* Presize the dict to avoid rehashing */
 203         dictExpand(d,intsetLen(setobj->ptr));
 204
 205         /* To add the elements we extract integers and create redis objects */
 206         si = setTypeInitIterator(setobj);
 207         while (setTypeNext(si,NULL,&intele) != -1) {
 208             element = createStringObjectFromLongLong(intele);
 209             redisAssertWithInfo(NULL,element,dictAdd(d,element,NULL) == DICT_OK);
 210         }
 211         setTypeReleaseIterator(si);
 212
 213         setobj->encoding = REDIS_ENCODING_HT;
 214         zfree(setobj->ptr);
 215         setobj->ptr = d;
 216     } else {
 217         redisPanic("Unsupported set conversion");
 218     }
 219 }
 220
 221 void saddCommand(redisClient *c) {
 222     robj *set;
 223     int j, added = 0;
 224
 225     set = lookupKeyWrite(c->db,c->argv[1]);
 226     if (set == NULL) {
 227         set = setTypeCreate(c->argv[2]);
 228         dbAdd(c->db,c->argv[1],set);
 229     } else {
 230         if (set->type != REDIS_SET) {
 231             addReply(c,shared.wrongtypeerr);
 232             return;
 233         }
 234     }
 235
 236     for (j = 2; j < c->argc; j++) {
 237         c->argv[j] = tryObjectEncoding(c->argv[j]);
 238         if (setTypeAdd(set,c->argv[j])) added++;
 239     }
 240     if (added) signalModifiedKey(c->db,c->argv[1]);
 241     server.dirty += added;
 242     addReplyLongLong(c,added);
 243 }
 244
 245 void sremCommand(redisClient *c) {
 246     robj *set;
 247     int j, deleted = 0;
 248
 249     if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
 250         checkType(c,set,REDIS_SET)) return;
 251
 252     for (j = 2; j < c->argc; j++) {
 253         if (setTypeRemove(set,c->argv[j])) {
 254             deleted++;
 255             if (setTypeSize(set) == 0) {
 256                 dbDelete(c->db,c->argv[1]);
 257                 break;
 258             }
 259         }
 260     }
 261     if (deleted) {
 262         signalModifiedKey(c->db,c->argv[1]);
 263         server.dirty += deleted;
 264     }
 265     addReplyLongLong(c,deleted);
 266 }
 267
 268 void smoveCommand(redisClient *c) {
 269     robj *srcset, *dstset, *ele;
 270     srcset = lookupKeyWrite(c->db,c->argv[1]);
 271     dstset = lookupKeyWrite(c->db,c->argv[2]);
 272     ele = c->argv[3] = tryObjectEncoding(c->argv[3]);
 273
 274     /* If the source key does not exist return 0 */
 275     if (srcset == NULL) {
 276         addReply(c,shared.czero);
 277         return;
 278     }
 279
 280     /* If the source key has the wrong type, or the destination key
 281      * is set and has the wrong type, return with an error. */
 282     if (checkType(c,srcset,REDIS_SET) ||
 283         (dstset && checkType(c,dstset,REDIS_SET))) return;
 284
 285     /* If srcset and dstset are equal, SMOVE is a no-op */
 286     if (srcset == dstset) {
 287         addReply(c,shared.cone);
 288         return;
 289     }
 290
 291     /* If the element cannot be removed from the src set, return 0. */
 292     if (!setTypeRemove(srcset,ele)) {
 293         addReply(c,shared.czero);
 294         return;
 295     }
 296
 297     /* Remove the src set from the database when empty */
 298     if (setTypeSize(srcset) == 0) dbDelete(c->db,c->argv[1]);
 299     signalModifiedKey(c->db,c->argv[1]);
 300     signalModifiedKey(c->db,c->argv[2]);
 301     server.dirty++;
 302
 303     /* Create the destination set when it doesn't exist */
 304     if (!dstset) {
 305         dstset = setTypeCreate(ele);
 306         dbAdd(c->db,c->argv[2],dstset);
 307     }
 308
 309     /* An extra key has changed when ele was successfully added to dstset */
 310     if (setTypeAdd(dstset,ele)) server.dirty++;
 311     addReply(c,shared.cone);
 312 }
 313
 314 void sismemberCommand(redisClient *c) {
 315     robj *set;
 316
 317     if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
 318         checkType(c,set,REDIS_SET)) return;
 319
 320     c->argv[2] = tryObjectEncoding(c->argv[2]);
 321     if (setTypeIsMember(set,c->argv[2]))
 322         addReply(c,shared.cone);
 323     else
 324         addReply(c,shared.czero);
 325 }
 326
 327 void scardCommand(redisClient *c) {
 328     robj *o;
 329
 330     if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
 331         checkType(c,o,REDIS_SET)) return;
 332
 333     addReplyLongLong(c,setTypeSize(o));
 334 }
 335
 336 void spopCommand(redisClient *c) {
 337     robj *set, *ele, *aux;
 338     int64_t llele;
 339     int encoding;
 340
 341     if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
 342         checkType(c,set,REDIS_SET)) return;
 343
 344     encoding = setTypeRandomElement(set,&ele,&llele);
 345     if (encoding == REDIS_ENCODING_INTSET) {
 346         ele = createStringObjectFromLongLong(llele);
 347         set->ptr = intsetRemove(set->ptr,llele,NULL);
 348     } else {
 349         incrRefCount(ele);
 350         setTypeRemove(set,ele);
 351     }
 352
 353     /* Replicate/AOF this command as an SREM operation */
 354     aux = createStringObject("SREM",4);
 355     rewriteClientCommandVector(c,3,aux,c->argv[1],ele);
 356     decrRefCount(ele);
 357     decrRefCount(aux);
 358
 359     addReplyBulk(c,ele);
 360     if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
 361     signalModifiedKey(c->db,c->argv[1]);
 362     server.dirty++;
 363 }
 364
 365 /* handle the "SRANDMEMBER key <count>" variant. The normal version of the
 366  * command is handled by the srandmemberCommand() function itself. */
 367
 368 /* How many times bigger should be the set compared to the requested size
 369  * for us to don't use the "remove elements" strategy? Read later in the
 370  * implementation for more info. */
 371 #define SRANDMEMBER_SUB_STRATEGY_MUL 3
 372
 373 void srandmemberWithCountCommand(redisClient *c) {
 374     long l;
 375     unsigned long count, size;
 376     int uniq = 1;
 377     robj *set, *ele;
 378     int64_t llele;
 379     int encoding;
 380
 381     dict *d;
 382
 383     if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != REDIS_OK) return;
 384     if (l >= 0) {
 385         count = (unsigned) l;
 386     } else {
 387         /* A negative count means: return the same elements multiple times
 388          * (i.e. don't remove the extracted element after every extraction). */
 389         count = -l;
 390         uniq = 0;
 391     }
 392
 393     if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk))
 394         == NULL || checkType(c,set,REDIS_SET)) return;
 395     size = setTypeSize(set);
 396
 397     /* If count is zero, serve it ASAP to avoid special cases later. */
 398     if (count == 0) {
 399         addReply(c,shared.emptymultibulk);
 400         return;
 401     }
 402
 403     /* CASE 1: The count was negative, so the extraction method is just:
 404      * "return N random elements" sampling the whole set every time.
 405      * This case is trivial and can be served without auxiliary data
 406      * structures. */
 407     if (!uniq) {
 408         addReplyMultiBulkLen(c,count);
 409         while(count--) {
 410             encoding = setTypeRandomElement(set,&ele,&llele);
 411             if (encoding == REDIS_ENCODING_INTSET) {
 412                 addReplyBulkLongLong(c,llele);
 413             } else {
 414                 addReplyBulk(c,ele);
 415             }
 416         }
 417         return;
 418     }
 419
 420     /* CASE 2:
 421      * The number of requested elements is greater than the number of
 422      * elements inside the set: simply return the whole set. */
 423     if (count >= size) {
 424         sunionDiffGenericCommand(c,c->argv,c->argc-1,NULL,REDIS_OP_UNION);
 425         return;
 426     }
 427
 428     /* For CASE 3 and CASE 4 we need an auxiliary dictionary. */
 429     d = dictCreate(&setDictType,NULL);
 430
 431     /* CASE 3:
 432      * The number of elements inside the set is not greater than
 433      * SRANDMEMBER_SUB_STRATEGY_MUL times the number of requested elements.
 434      * In this case we create a set from scratch with all the elements, and
 435      * subtract random elements to reach the requested number of elements.
 436      *
 437      * This is done because if the number of requsted elements is just
 438      * a bit less than the number of elements in the set, the natural approach
 439      * used into CASE 3 is highly inefficient. */
 440     if (count*SRANDMEMBER_SUB_STRATEGY_MUL > size) {
 441         setTypeIterator *si;
 442
 443         /* Add all the elements into the temporary dictionary. */
 444         si = setTypeInitIterator(set);
 445         while((encoding = setTypeNext(si,&ele,&llele)) != -1) {
 446             int retval;
 447
 448             if (encoding == REDIS_ENCODING_INTSET) {
 449                 retval = dictAdd(d,createStringObjectFromLongLong(llele),NULL);
 450             } else if (ele->encoding == REDIS_ENCODING_RAW) {
 451                 retval = dictAdd(d,dupStringObject(ele),NULL);
 452             } else if (ele->encoding == REDIS_ENCODING_INT) {
 453                 retval = dictAdd(d,
 454                     createStringObjectFromLongLong((long)ele->ptr),NULL);
 455             }
 456             redisAssert(retval == DICT_OK);
 457         }
 458         setTypeReleaseIterator(si);
 459         redisAssert(dictSize(d) == size);
 460
 461         /* Remove random elements to reach the right count. */
 462         while(size > count) {
 463             dictEntry *de;
 464
 465             de = dictGetRandomKey(d);
 466             dictDelete(d,dictGetKey(de));
 467             size--;
 468         }
 469     }
 470
 471     /* CASE 4: We have a big set compared to the requested number of elements.
 472      * In this case we can simply get random elements from the set and add
 473      * to the temporary set, trying to eventually get enough unique elements
 474      * to reach the specified count. */
 475     else {
 476         unsigned long added = 0;
 477
 478         while(added < count) {
 479             encoding = setTypeRandomElement(set,&ele,&llele);
 480             if (encoding == REDIS_ENCODING_INTSET) {
 481                 ele = createStringObjectFromLongLong(llele);
 482             } else if (ele->encoding == REDIS_ENCODING_RAW) {
 483                 ele = dupStringObject(ele);
 484             } else if (ele->encoding == REDIS_ENCODING_INT) {
 485                 ele = createStringObjectFromLongLong((long)ele->ptr);
 486             }
 487             /* Try to add the object to the dictionary. If it already exists
 488              * free it, otherwise increment the number of objects we have
 489              * in the result dictionary. */
 490             if (dictAdd(d,ele,NULL) == DICT_OK)
 491                 added++;
 492             else
 493                 decrRefCount(ele);
 494         }
 495     }
 496
 497     /* CASE 3 & 4: send the result to the user. */
 498     {
 499         dictIterator *di;
 500         dictEntry *de;
 501
 502         addReplyMultiBulkLen(c,count);
 503         di = dictGetIterator(d);
 504         while((de = dictNext(di)) != NULL)
 505             addReplyBulk(c,dictGetKey(de));
 506         dictReleaseIterator(di);
 507         dictRelease(d);
 508     }
 509 }
 510
 511 void srandmemberCommand(redisClient *c) {
 512     robj *set, *ele;
 513     int64_t llele;
 514     int encoding;
 515
 516     if (c->argc == 3) {
 517         srandmemberWithCountCommand(c);
 518         return;
 519     } else if (c->argc > 3) {
 520         addReply(c,shared.syntaxerr);
 521         return;
 522     }
 523
 524     if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
 525         checkType(c,set,REDIS_SET)) return;
 526
 527     encoding = setTypeRandomElement(set,&ele,&llele);
 528     if (encoding == REDIS_ENCODING_INTSET) {
 529         addReplyBulkLongLong(c,llele);
 530     } else {
 531         addReplyBulk(c,ele);
 532     }
 533 }
 534
 535 int qsortCompareSetsByCardinality(const void *s1, const void *s2) {
 536     return setTypeSize(*(robj**)s1)-setTypeSize(*(robj**)s2);
 537 }
 538
 539 void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum, robj *dstkey) {
 540     robj **sets = zmalloc(sizeof(robj*)*setnum);
 541     setTypeIterator *si;
 542     robj *eleobj, *dstset = NULL;
 543     int64_t intobj;
 544     void *replylen = NULL;
 545     unsigned long j, cardinality = 0;
 546     int encoding;
 547
 548     for (j = 0; j < setnum; j++) {
 549         robj *setobj = dstkey ?
 550             lookupKeyWrite(c->db,setkeys[j]) :
 551             lookupKeyRead(c->db,setkeys[j]);
 552         if (!setobj) {
 553             zfree(sets);
 554             if (dstkey) {
 555                 if (dbDelete(c->db,dstkey)) {
 556                     signalModifiedKey(c->db,dstkey);
 557                     server.dirty++;
 558                 }
 559                 addReply(c,shared.czero);
 560             } else {
 561                 addReply(c,shared.emptymultibulk);
 562             }
 563             return;
 564         }
 565         if (checkType(c,setobj,REDIS_SET)) {
 566             zfree(sets);
 567             return;
 568         }
 569         sets[j] = setobj;
 570     }
 571     /* Sort sets from the smallest to largest, this will improve our
 572      * algorithm's performace */
 573     qsort(sets,setnum,sizeof(robj*),qsortCompareSetsByCardinality);
 574
 575     /* The first thing we should output is the total number of elements...
 576      * since this is a multi-bulk write, but at this stage we don't know
 577      * the intersection set size, so we use a trick, append an empty object
 578      * to the output list and save the pointer to later modify it with the
 579      * right length */
 580     if (!dstkey) {
 581         replylen = addDeferredMultiBulkLength(c);
 582     } else {
 583         /* If we have a target key where to store the resulting set
 584          * create this key with an empty set inside */
 585         dstset = createIntsetObject();
 586     }
 587
 588     /* Iterate all the elements of the first (smallest) set, and test
 589      * the element against all the other sets, if at least one set does
 590      * not include the element it is discarded */
 591     si = setTypeInitIterator(sets[0]);
 592     while((encoding = setTypeNext(si,&eleobj,&intobj)) != -1) {
 593         for (j = 1; j < setnum; j++) {
 594             if (sets[j] == sets[0]) continue;
 595             if (encoding == REDIS_ENCODING_INTSET) {
 596                 /* intset with intset is simple... and fast */
 597                 if (sets[j]->encoding == REDIS_ENCODING_INTSET &&
 598                     !intsetFind((intset*)sets[j]->ptr,intobj))
 599                 {
 600                     break;
 601                 /* in order to compare an integer with an object we
 602                  * have to use the generic function, creating an object
 603                  * for this */
 604                 } else if (sets[j]->encoding == REDIS_ENCODING_HT) {
 605                     eleobj = createStringObjectFromLongLong(intobj);
 606                     if (!setTypeIsMember(sets[j],eleobj)) {
 607                         decrRefCount(eleobj);
 608                         break;
 609                     }
 610                     decrRefCount(eleobj);
 611                 }
 612             } else if (encoding == REDIS_ENCODING_HT) {
 613                 /* Optimization... if the source object is integer
 614                  * encoded AND the target set is an intset, we can get
 615                  * a much faster path. */
 616                 if (eleobj->encoding == REDIS_ENCODING_INT &&
 617                     sets[j]->encoding == REDIS_ENCODING_INTSET &&
 618                     !intsetFind((intset*)sets[j]->ptr,(long)eleobj->ptr))
 619                 {
 620                     break;
 621                 /* else... object to object check is easy as we use the
 622                  * type agnostic API here. */
 623                 } else if (!setTypeIsMember(sets[j],eleobj)) {
 624                     break;
 625                 }
 626             }
 627         }
 628
 629         /* Only take action when all sets contain the member */
 630         if (j == setnum) {
 631             if (!dstkey) {
 632                 if (encoding == REDIS_ENCODING_HT)
 633                     addReplyBulk(c,eleobj);
 634                 else
 635                     addReplyBulkLongLong(c,intobj);
 636                 cardinality++;
 637             } else {
 638                 if (encoding == REDIS_ENCODING_INTSET) {
 639                     eleobj = createStringObjectFromLongLong(intobj);
 640                     setTypeAdd(dstset,eleobj);
 641                     decrRefCount(eleobj);
 642                 } else {
 643                     setTypeAdd(dstset,eleobj);
 644                 }
 645             }
 646         }
 647     }
 648     setTypeReleaseIterator(si);
 649
 650     if (dstkey) {
 651         /* Store the resulting set into the target, if the intersection
 652          * is not an empty set. */
 653         dbDelete(c->db,dstkey);
 654         if (setTypeSize(dstset) > 0) {
 655             dbAdd(c->db,dstkey,dstset);
 656             addReplyLongLong(c,setTypeSize(dstset));
 657         } else {
 658             decrRefCount(dstset);
 659             addReply(c,shared.czero);
 660         }
 661         signalModifiedKey(c->db,dstkey);
 662         server.dirty++;
 663     } else {
 664         setDeferredMultiBulkLength(c,replylen,cardinality);
 665     }
 666     zfree(sets);
 667 }
 668
 669 void sinterCommand(redisClient *c) {
 670     sinterGenericCommand(c,c->argv+1,c->argc-1,NULL);
 671 }
 672
 673 void sinterstoreCommand(redisClient *c) {
 674     sinterGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
 675 }
 676
 677 #define REDIS_OP_UNION 0
 678 #define REDIS_OP_DIFF 1
 679 #define REDIS_OP_INTER 2
 680
 681 void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *dstkey, int op) {
 682     robj **sets = zmalloc(sizeof(robj*)*setnum);
 683     setTypeIterator *si;
 684     robj *ele, *dstset = NULL;
 685     int j, cardinality = 0;
 686
 687     for (j = 0; j < setnum; j++) {
 688         robj *setobj = dstkey ?
 689             lookupKeyWrite(c->db,setkeys[j]) :
 690             lookupKeyRead(c->db,setkeys[j]);
 691         if (!setobj) {
 692             sets[j] = NULL;
 693             continue;
 694         }
 695         if (checkType(c,setobj,REDIS_SET)) {
 696             zfree(sets);
 697             return;
 698         }
 699         sets[j] = setobj;
 700     }
 701
 702     /* We need a temp set object to store our union. If the dstkey
 703      * is not NULL (that is, we are inside an SUNIONSTORE operation) then
 704      * this set object will be the resulting object to set into the target key*/
 705     dstset = createIntsetObject();
 706
 707     /* Iterate all the elements of all the sets, add every element a single
 708      * time to the result set */
 709     for (j = 0; j < setnum; j++) {
 710         if (op == REDIS_OP_DIFF && j == 0 && !sets[j]) break; /* result set is empty */
 711         if (!sets[j]) continue; /* non existing keys are like empty sets */
 712
 713         si = setTypeInitIterator(sets[j]);
 714         while((ele = setTypeNextObject(si)) != NULL) {
 715             if (op == REDIS_OP_UNION || j == 0) {
 716                 if (setTypeAdd(dstset,ele)) {
 717                     cardinality++;
 718                 }
 719             } else if (op == REDIS_OP_DIFF) {
 720                 if (setTypeRemove(dstset,ele)) {
 721                     cardinality--;
 722                 }
 723             }
 724             decrRefCount(ele);
 725         }
 726         setTypeReleaseIterator(si);
 727
 728         /* Exit when result set is empty. */
 729         if (op == REDIS_OP_DIFF && cardinality == 0) break;
 730     }
 731
 732     /* Output the content of the resulting set, if not in STORE mode */
 733     if (!dstkey) {
 734         addReplyMultiBulkLen(c,cardinality);
 735         si = setTypeInitIterator(dstset);
 736         while((ele = setTypeNextObject(si)) != NULL) {
 737             addReplyBulk(c,ele);
 738             decrRefCount(ele);
 739         }
 740         setTypeReleaseIterator(si);
 741         decrRefCount(dstset);
 742     } else {
 743         /* If we have a target key where to store the resulting set
 744          * create this key with the result set inside */
 745         dbDelete(c->db,dstkey);
 746         if (setTypeSize(dstset) > 0) {
 747             dbAdd(c->db,dstkey,dstset);
 748             addReplyLongLong(c,setTypeSize(dstset));
 749         } else {
 750             decrRefCount(dstset);
 751             addReply(c,shared.czero);
 752         }
 753         signalModifiedKey(c->db,dstkey);
 754         server.dirty++;
 755     }
 756     zfree(sets);
 757 }
 758
 759 void sunionCommand(redisClient *c) {
 760     sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,REDIS_OP_UNION);
 761 }
 762
 763 void sunionstoreCommand(redisClient *c) {
 764     sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],REDIS_OP_UNION);
 765 }
 766
 767 void sdiffCommand(redisClient *c) {
 768     sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,REDIS_OP_DIFF);
 769 }
 770
 771 void sdiffstoreCommand(redisClient *c) {
 772     sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],REDIS_OP_DIFF);
 773 }