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