fluencelabs/redis
1
0
Fork 0
mirror of https://github.com/fluencelabs/redis synced 2025-04-21 08:32:14 +00:00
Code Issues Projects Releases Wiki Activity

Lazyfree: Convert Sets to use plains SDS (several commits squashed).

Browse Source
This commit is contained in:
antirez 2015-07-31 18:01:23 +02:00
parent 4ff3c17a20
commit 86d48efbfd
9 changed files with 183 additions and 163 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/db.c
View File

@ -415,8 +415,8 @@ void scanCallback(void *privdata, const dictEntry *de) {
sds sdskey = dictGetKey(de); sds sdskey = dictGetKey(de);
key = createStringObject(sdskey, sdslen(sdskey)); key = createStringObject(sdskey, sdslen(sdskey));
} else if (o->type == OBJ_SET) { } else if (o->type == OBJ_SET) {
key = dictGetKey(de); sds keysds = dictGetKey(de);
incrRefCount(key); key = createStringObject(keysds,sdslen(keysds));
} else if (o->type == OBJ_HASH) { } else if (o->type == OBJ_HASH) {
key = dictGetKey(de); key = dictGetKey(de);
incrRefCount(key); incrRefCount(key);

4
src/debug.c
View File

@ -164,7 +164,9 @@ void computeDatasetDigest(unsigned char *final) {
} else if (o->type == OBJ_SET) { } else if (o->type == OBJ_SET) {
setTypeIterator *si = setTypeInitIterator(o); setTypeIterator *si = setTypeInitIterator(o);
robj *ele; robj *ele;
while((ele = setTypeNextObject(si)) != NULL) { sds sdsele;
while((sdsele = setTypeNextObject(si)) != NULL) {
ele = createObject(OBJ_STRING,sdsele);
xorObjectDigest(digest,ele); xorObjectDigest(digest,ele);
decrRefCount(ele); decrRefCount(ele);
} }

4
src/networking.c
View File

@ -117,13 +117,13 @@ client *createClient(int fd) {
listSetDupMethod(c->reply,dupClientReplyValue); listSetDupMethod(c->reply,dupClientReplyValue);
c->btype = BLOCKED_NONE; c->btype = BLOCKED_NONE;
c->bpop.timeout = 0; c->bpop.timeout = 0;
c->bpop.keys = dictCreate(&setDictType,NULL); c->bpop.keys = dictCreate(&objectKeyPointerValueDictType,NULL);
c->bpop.target = NULL; c->bpop.target = NULL;
c->bpop.numreplicas = 0; c->bpop.numreplicas = 0;
c->bpop.reploffset = 0; c->bpop.reploffset = 0;
c->woff = 0; c->woff = 0;
c->watched_keys = listCreate(); c->watched_keys = listCreate();
c->pubsub_channels = dictCreate(&setDictType,NULL); c->pubsub_channels = dictCreate(&objectKeyPointerValueDictType,NULL);
c->pubsub_patterns = listCreate(); c->pubsub_patterns = listCreate();
c->peerid = NULL; c->peerid = NULL;
listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid); listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid);

6
src/object.c
View File

@ -364,13 +364,17 @@ int checkType(client *c, robj *o, int type) {
return 0; return 0;
} }
int isSdsRepresentableAsLongLong(sds s, long long *llval) {
return string2ll(s,sdslen(s),llval) ? C_OK : C_ERR;
}
int isObjectRepresentableAsLongLong(robj *o, long long *llval) { int isObjectRepresentableAsLongLong(robj *o, long long *llval) {
serverAssertWithInfo(NULL,o,o->type == OBJ_STRING); serverAssertWithInfo(NULL,o,o->type == OBJ_STRING);
if (o->encoding == OBJ_ENCODING_INT) { if (o->encoding == OBJ_ENCODING_INT) {
if (llval) *llval = (long) o->ptr; if (llval) *llval = (long) o->ptr;
return C_OK; return C_OK;
} else { } else {
return string2ll(o->ptr,sdslen(o->ptr),llval) ? C_OK : C_ERR; return isSdsRepresentableAsLongLong(o->ptr,llval);
} }
} }

60
src/rdb.c
View File

@ -43,6 +43,7 @@
#define RDB_LOAD_NONE 0 #define RDB_LOAD_NONE 0
#define RDB_LOAD_ENC (1<<0) #define RDB_LOAD_ENC (1<<0)
#define RDB_LOAD_PLAIN (1<<1) #define RDB_LOAD_PLAIN (1<<1)
#define RDB_LOAD_SDS (1<<2)
#define rdbExitReportCorruptRDB(reason) rdbCheckThenExit(reason, __LINE__); #define rdbExitReportCorruptRDB(reason) rdbCheckThenExit(reason, __LINE__);
@ -179,6 +180,7 @@ int rdbEncodeInteger(long long value, unsigned char *enc) {
* rdbGenerincLoadStringObject() for more info. */ * rdbGenerincLoadStringObject() for more info. */
void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags) { void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags) {
int plain = flags & RDB_LOAD_PLAIN; int plain = flags & RDB_LOAD_PLAIN;
int sds = flags & RDB_LOAD_SDS;
int encode = flags & RDB_LOAD_ENC; int encode = flags & RDB_LOAD_ENC;
unsigned char enc[4]; unsigned char enc[4];
long long val; long long val;
@ -200,10 +202,10 @@ void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags) {
val = 0; /* anti-warning */ val = 0; /* anti-warning */
rdbExitReportCorruptRDB("Unknown RDB integer encoding type"); rdbExitReportCorruptRDB("Unknown RDB integer encoding type");
} }
if (plain) { if (plain || sds) {
char buf[LONG_STR_SIZE], *p; char buf[LONG_STR_SIZE], *p;
int len = ll2string(buf,sizeof(buf),val); int len = ll2string(buf,sizeof(buf),val);
p = zmalloc(len); p = plain ? zmalloc(len) : sdsnewlen(NULL,len);
memcpy(p,buf,len); memcpy(p,buf,len);
return p; return p;
} else if (encode) { } else if (encode) {
@ -280,9 +282,10 @@ ssize_t rdbSaveLzfStringObject(rio *rdb, unsigned char *s, size_t len) {
* rdbGenericLoadStringObject() function. */ * rdbGenericLoadStringObject() function. */
void *rdbLoadLzfStringObject(rio *rdb, int flags) { void *rdbLoadLzfStringObject(rio *rdb, int flags) {
int plain = flags & RDB_LOAD_PLAIN; int plain = flags & RDB_LOAD_PLAIN;
int sds = flags & RDB_LOAD_SDS;
unsigned int len, clen; unsigned int len, clen;
unsigned char *c = NULL; unsigned char *c = NULL;
sds val = NULL; char *val = NULL;
if ((clen = rdbLoadLen(rdb,NULL)) == RDB_LENERR) return NULL; if ((clen = rdbLoadLen(rdb,NULL)) == RDB_LENERR) return NULL;
if ((len = rdbLoadLen(rdb,NULL)) == RDB_LENERR) return NULL; if ((len = rdbLoadLen(rdb,NULL)) == RDB_LENERR) return NULL;
@ -292,7 +295,7 @@ void *rdbLoadLzfStringObject(rio *rdb, int flags) {
if (plain) { if (plain) {
val = zmalloc(len); val = zmalloc(len);
} else { } else {
if ((val = sdsnewlen(NULL,len)) == NULL) goto err; val = sdsnewlen(NULL,len);
} }
/* Load the compressed representation and uncompress it to target. */ /* Load the compressed representation and uncompress it to target. */
@ -300,10 +303,11 @@ void *rdbLoadLzfStringObject(rio *rdb, int flags) {
if (lzf_decompress(c,clen,val,len) == 0) goto err; if (lzf_decompress(c,clen,val,len) == 0) goto err;
zfree(c); zfree(c);
if (plain) if (plain || sds) {
return val; return val;
else } else {
return createObject(OBJ_STRING,val); return createObject(OBJ_STRING,val);
}
err: err:
zfree(c); zfree(c);
if (plain) if (plain)
@ -366,7 +370,7 @@ ssize_t rdbSaveLongLongAsStringObject(rio *rdb, long long value) {
return nwritten; return nwritten;
} }
/* Like rdbSaveStringObjectRaw() but handle encoded objects */ /* Like rdbSaveRawString() gets a Redis object instead. */
int rdbSaveStringObject(rio *rdb, robj *obj) { int rdbSaveStringObject(rio *rdb, robj *obj) {
/* Avoid to decode the object, then encode it again, if the /* Avoid to decode the object, then encode it again, if the
* object is already integer encoded. */ * object is already integer encoded. */
@ -387,10 +391,12 @@ int rdbSaveStringObject(rio *rdb, robj *obj) {
* no longer guarantees that obj->ptr is an SDS string. * no longer guarantees that obj->ptr is an SDS string.
* RDB_LOAD_PLAIN: Return a plain string allocated with zmalloc() * RDB_LOAD_PLAIN: Return a plain string allocated with zmalloc()
* instead of a Redis object. * instead of a Redis object.
* RDB_LOAD_SDS: Return an SDS string instead of a Redis object.
*/ */
void *rdbGenericLoadStringObject(rio *rdb, int flags) { void *rdbGenericLoadStringObject(rio *rdb, int flags) {
int encode = flags & RDB_LOAD_ENC; int encode = flags & RDB_LOAD_ENC;
int plain = flags & RDB_LOAD_PLAIN; int plain = flags & RDB_LOAD_PLAIN;
int sds = flags & RDB_LOAD_SDS;
int isencoded; int isencoded;
uint32_t len; uint32_t len;
@ -409,7 +415,17 @@ void *rdbGenericLoadStringObject(rio *rdb, int flags) {
} }
if (len == RDB_LENERR) return NULL; if (len == RDB_LENERR) return NULL;
if (!plain) { if (plain || sds) {
void *buf = plain ? zmalloc(len) : sdsnewlen(NULL,len);
if (len && rioRead(rdb,buf,len) == 0) {
if (plain)
zfree(buf);
else
sdsfree(buf);
return NULL;
}
return buf;
} else {
robj *o = encode ? createStringObject(NULL,len) : robj *o = encode ? createStringObject(NULL,len) :
createRawStringObject(NULL,len); createRawStringObject(NULL,len);
if (len && rioRead(rdb,o->ptr,len) == 0) { if (len && rioRead(rdb,o->ptr,len) == 0) {
@ -417,13 +433,6 @@ void *rdbGenericLoadStringObject(rio *rdb, int flags) {
return NULL; return NULL;
} }
return o; return o;
} else {
void *buf = zmalloc(len);
if (len && rioRead(rdb,buf,len) == 0) {
zfree(buf);
return NULL;
}
return buf;
} }
} }
@ -583,8 +592,9 @@ ssize_t rdbSaveObject(rio *rdb, robj *o) {
nwritten += n; nwritten += n;
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
robj *eleobj = dictGetKey(de); sds ele = dictGetKey(de);
if ((n = rdbSaveStringObject(rdb,eleobj)) == -1) return -1; if ((n = rdbSaveRawString(rdb,(unsigned char*)ele,sdslen(ele)))
== -1) return -1;
nwritten += n; nwritten += n;
} }
dictReleaseIterator(di); dictReleaseIterator(di);
@ -959,7 +969,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
decrRefCount(ele); decrRefCount(ele);
} }
} else if (rdbtype == RDB_TYPE_SET) { } else if (rdbtype == RDB_TYPE_SET) {
/* Read list/set value */ /* Read Set value */
if ((len = rdbLoadLen(rdb,NULL)) == RDB_LENERR) return NULL; if ((len = rdbLoadLen(rdb,NULL)) == RDB_LENERR) return NULL;
/* Use a regular set when there are too many entries. */ /* Use a regular set when there are too many entries. */
@ -973,15 +983,17 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
o = createIntsetObject(); o = createIntsetObject();
} }
/* Load every single element of the list/set */ /* Load every single element of the set */
for (i = 0; i < len; i++) { for (i = 0; i < len; i++) {
long long llval; long long llval;
if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL; sds sdsele;
ele = tryObjectEncoding(ele);
if ((sdsele = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS)) == NULL)
return NULL;
if (o->encoding == OBJ_ENCODING_INTSET) { if (o->encoding == OBJ_ENCODING_INTSET) {
/* Fetch integer value from element */ /* Fetch integer value from element */
if (isObjectRepresentableAsLongLong(ele,&llval) == C_OK) { if (isSdsRepresentableAsLongLong(sdsele,&llval) == C_OK) {
o->ptr = intsetAdd(o->ptr,llval,NULL); o->ptr = intsetAdd(o->ptr,llval,NULL);
} else { } else {
setTypeConvert(o,OBJ_ENCODING_HT); setTypeConvert(o,OBJ_ENCODING_HT);
@ -992,9 +1004,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
/* This will also be called when the set was just converted /* This will also be called when the set was just converted
* to a regular hash table encoded set */ * to a regular hash table encoded set */
if (o->encoding == OBJ_ENCODING_HT) { if (o->encoding == OBJ_ENCODING_HT) {
dictAdd((dict*)o->ptr,ele,NULL); dictAdd((dict*)o->ptr,sdsele,NULL);
} else { } else {
decrRefCount(ele); sdsfree(sdsele);
} }
} }
} else if (rdbtype == RDB_TYPE_ZSET) { } else if (rdbtype == RDB_TYPE_ZSET) {

17
src/server.c
View File

@ -531,8 +531,9 @@ unsigned int dictEncObjHash(const void *key) {
} }
} }
/* Sets type hash table */ /* Generic hash table type where keys are Redis Objects, Values
dictType setDictType = { * dummy pointers. */
dictType objectKeyPointerValueDictType = {
dictEncObjHash, /* hash function */ dictEncObjHash, /* hash function */
NULL, /* key dup */ NULL, /* key dup */
NULL, /* val dup */ NULL, /* val dup */
@ -541,6 +542,16 @@ dictType setDictType = {
NULL /* val destructor */ NULL /* val destructor */
}; };
/* Set dictionary type. Keys are SDS strings, values are ot used. */
dictType setDictType = {
dictSdsHash, /* hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
dictSdsDestructor, /* key destructor */
NULL /* val destructor */
};
/* Sorted sets hash (note: a skiplist is used in addition to the hash table) */ /* Sorted sets hash (note: a skiplist is used in addition to the hash table) */
dictType zsetDictType = { dictType zsetDictType = {
dictEncObjHash, /* hash function */ dictEncObjHash, /* hash function */
@ -1834,7 +1845,7 @@ void initServer(void) {
server.db[j].dict = dictCreate(&dbDictType,NULL); server.db[j].dict = dictCreate(&dbDictType,NULL);
server.db[j].expires = dictCreate(&keyptrDictType,NULL); server.db[j].expires = dictCreate(&keyptrDictType,NULL);
server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL); server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
server.db[j].ready_keys = dictCreate(&setDictType,NULL); server.db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType,NULL);
server.db[j].watched_keys = dictCreate(&keylistDictType,NULL); server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);
server.db[j].eviction_pool = evictionPoolAlloc(); server.db[j].eviction_pool = evictionPoolAlloc();
server.db[j].id = j; server.db[j].id = j;

16
src/server.h
View File

@ -1043,6 +1043,7 @@ typedef struct {
extern struct redisServer server; extern struct redisServer server;
extern struct sharedObjectsStruct shared; extern struct sharedObjectsStruct shared;
extern dictType objectKeyPointerValueDictType;
extern dictType setDictType; extern dictType setDictType;
extern dictType zsetDictType; extern dictType zsetDictType;
extern dictType clusterNodesDictType; extern dictType clusterNodesDictType;
@ -1173,6 +1174,7 @@ robj *createStringObject(const char *ptr, size_t len);
robj *createRawStringObject(const char *ptr, size_t len); robj *createRawStringObject(const char *ptr, size_t len);
robj *createEmbeddedStringObject(const char *ptr, size_t len); robj *createEmbeddedStringObject(const char *ptr, size_t len);
robj *dupStringObject(robj *o); robj *dupStringObject(robj *o);
int isSdsRepresentableAsLongLong(sds s, long long *llval);
int isObjectRepresentableAsLongLong(robj *o, long long *llongval); int isObjectRepresentableAsLongLong(robj *o, long long *llongval);
robj *tryObjectEncoding(robj *o); robj *tryObjectEncoding(robj *o);
robj *getDecodedObject(robj *o); robj *getDecodedObject(robj *o);
@ -1313,15 +1315,15 @@ unsigned int getLRUClock(void);
const char *maxmemoryToString(void); const char *maxmemoryToString(void);
/* Set data type */ /* Set data type */
robj *setTypeCreate(robj *value); robj *setTypeCreate(sds value);
int setTypeAdd(robj *subject, robj *value); int setTypeAdd(robj *subject, sds value);
int setTypeRemove(robj *subject, robj *value); int setTypeRemove(robj *subject, sds value);
int setTypeIsMember(robj *subject, robj *value); int setTypeIsMember(robj *subject, sds value);
setTypeIterator *setTypeInitIterator(robj *subject); setTypeIterator *setTypeInitIterator(robj *subject);
void setTypeReleaseIterator(setTypeIterator *si); void setTypeReleaseIterator(setTypeIterator *si);
int setTypeNext(setTypeIterator *si, robj **objele, int64_t *llele); int setTypeNext(setTypeIterator *si, sds *sdsele, int64_t *llele);
robj *setTypeNextObject(setTypeIterator *si); sds setTypeNextObject(setTypeIterator *si);
int setTypeRandomElement(robj *setobj, robj **objele, int64_t *llele); int setTypeRandomElement(robj *setobj, sds *sdsele, int64_t *llele);
unsigned long setTypeRandomElements(robj *set, unsigned long count, robj *aux_set); unsigned long setTypeRandomElements(robj *set, unsigned long count, robj *aux_set);
unsigned long setTypeSize(robj *subject); unsigned long setTypeSize(robj *subject);
void setTypeConvert(robj *subject, int enc); void setTypeConvert(robj *subject, int enc);

6
src/sort.c
View File

@ -380,9 +380,9 @@ void sortCommand(client *c) {
listTypeReleaseIterator(li); listTypeReleaseIterator(li);
} else if (sortval->type == OBJ_SET) { } else if (sortval->type == OBJ_SET) {
setTypeIterator *si = setTypeInitIterator(sortval); setTypeIterator *si = setTypeInitIterator(sortval);
robj *ele; sds sdsele;
while((ele = setTypeNextObject(si)) != NULL) { while((sdsele = setTypeNextObject(si)) != NULL) {
vector[j].obj = ele; vector[j].obj = createObject(OBJ_STRING,sdsele);
vector[j].u.score = 0; vector[j].u.score = 0;
vector[j].u.cmpobj = NULL; vector[j].u.cmpobj = NULL;
j++; j++;

227
src/t_set.c
View File

@ -39,26 +39,28 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
/* Factory method to return a set that *can* hold "value". When the object has /* Factory method to return a set that *can* hold "value". When the object has
* an integer-encodable value, an intset will be returned. Otherwise a regular * an integer-encodable value, an intset will be returned. Otherwise a regular
* hash table. */ * hash table. */
robj *setTypeCreate(robj *value) { robj *setTypeCreate(sds value) {
if (isObjectRepresentableAsLongLong(value,NULL) == C_OK) if (isSdsRepresentableAsLongLong(value,NULL) == C_OK)
return createIntsetObject(); return createIntsetObject();
return createSetObject(); return createSetObject();
} }
/* Add the specified value into a set. The function takes care of incrementing /* Add the specified value into a set.
* the reference count of the object if needed in order to retain a copy.
* *
* If the value was already member of the set, nothing is done and 0 is * If the value was already member of the set, nothing is done and 0 is
* returned, otherwise the new element is added and 1 is returned. */ * returned, otherwise the new element is added and 1 is returned. */
int setTypeAdd(robj *subject, robj *value) { int setTypeAdd(robj *subject, sds value) {
long long llval; long long llval;
if (subject->encoding == OBJ_ENCODING_HT) { if (subject->encoding == OBJ_ENCODING_HT) {
if (dictAdd(subject->ptr,value,NULL) == DICT_OK) { dict *ht = subject->ptr;
incrRefCount(value); dictEntry *de = dictAddRaw(ht,value);
if (de) {
dictSetKey(ht,de,sdsdup(value));
dictSetVal(ht,de,NULL);
return 1; return 1;
} }
} else if (subject->encoding == OBJ_ENCODING_INTSET) { } else if (subject->encoding == OBJ_ENCODING_INTSET) {
if (isObjectRepresentableAsLongLong(value,&llval) == C_OK) { if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) {
uint8_t success = 0; uint8_t success = 0;
subject->ptr = intsetAdd(subject->ptr,llval,&success); subject->ptr = intsetAdd(subject->ptr,llval,&success);
if (success) { if (success) {
@ -74,9 +76,7 @@ int setTypeAdd(robj *subject, robj *value) {
/* The set *was* an intset and this value is not integer /* The set *was* an intset and this value is not integer
* encodable, so dictAdd should always work. */ * encodable, so dictAdd should always work. */
serverAssertWithInfo(NULL,value, serverAssert(dictAdd(subject->ptr,sdsdup(value),NULL) == DICT_OK);
dictAdd(subject->ptr,value,NULL) == DICT_OK);
incrRefCount(value);
return 1; return 1;
} }
} else { } else {
@ -85,7 +85,7 @@ int setTypeAdd(robj *subject, robj *value) {
return 0; return 0;
} }
int setTypeRemove(robj *setobj, robj *value) { int setTypeRemove(robj *setobj, sds value) {
long long llval; long long llval;
if (setobj->encoding == OBJ_ENCODING_HT) { if (setobj->encoding == OBJ_ENCODING_HT) {
if (dictDelete(setobj->ptr,value) == DICT_OK) { if (dictDelete(setobj->ptr,value) == DICT_OK) {
@ -93,7 +93,7 @@ int setTypeRemove(robj *setobj, robj *value) {
return 1; return 1;
} }
} else if (setobj->encoding == OBJ_ENCODING_INTSET) { } else if (setobj->encoding == OBJ_ENCODING_INTSET) {
if (isObjectRepresentableAsLongLong(value,&llval) == C_OK) { if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) {
int success; int success;
setobj->ptr = intsetRemove(setobj->ptr,llval,&success); setobj->ptr = intsetRemove(setobj->ptr,llval,&success);
if (success) return 1; if (success) return 1;
@ -104,12 +104,12 @@ int setTypeRemove(robj *setobj, robj *value) {
return 0; return 0;
} }
int setTypeIsMember(robj *subject, robj *value) { int setTypeIsMember(robj *subject, sds value) {
long long llval; long long llval;
if (subject->encoding == OBJ_ENCODING_HT) { if (subject->encoding == OBJ_ENCODING_HT) {
return dictFind((dict*)subject->ptr,value) != NULL; return dictFind((dict*)subject->ptr,value) != NULL;
} else if (subject->encoding == OBJ_ENCODING_INTSET) { } else if (subject->encoding == OBJ_ENCODING_INTSET) {
if (isObjectRepresentableAsLongLong(value,&llval) == C_OK) { if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) {
return intsetFind((intset*)subject->ptr,llval); return intsetFind((intset*)subject->ptr,llval);
} }
} else { } else {
@ -141,28 +141,26 @@ void setTypeReleaseIterator(setTypeIterator *si) {
/* Move to the next entry in the set. Returns the object at the current /* Move to the next entry in the set. Returns the object at the current
* position. * position.
* *
* Since set elements can be internally be stored as redis objects or * Since set elements can be internally be stored as SDS strings or
* simple arrays of integers, setTypeNext returns the encoding of the * simple arrays of integers, setTypeNext returns the encoding of the
* set object you are iterating, and will populate the appropriate pointer * set object you are iterating, and will populate the appropriate pointer
* (objele) or (llele) accordingly. * (sdsele) or (llele) accordingly.
* *
* Note that both the objele and llele pointers should be passed and cannot * Note that both the sdsele and llele pointers should be passed and cannot
* be NULL since the function will try to defensively populate the non * be NULL since the function will try to defensively populate the non
* used field with values which are easy to trap if misused. * used field with values which are easy to trap if misused.
* *
* When there are no longer elements -1 is returned. * When there are no longer elements -1 is returned. */
* Returned objects ref count is not incremented, so this function is int setTypeNext(setTypeIterator *si, sds *sdsele, int64_t *llele) {
* copy on write friendly. */
int setTypeNext(setTypeIterator *si, robj **objele, int64_t *llele) {
if (si->encoding == OBJ_ENCODING_HT) { if (si->encoding == OBJ_ENCODING_HT) {
dictEntry *de = dictNext(si->di); dictEntry *de = dictNext(si->di);
if (de == NULL) return -1; if (de == NULL) return -1;
*objele = dictGetKey(de); *sdsele = dictGetKey(de);
*llele = -123456789; /* Not needed. Defensive. */ *llele = -123456789; /* Not needed. Defensive. */
} else if (si->encoding == OBJ_ENCODING_INTSET) { } else if (si->encoding == OBJ_ENCODING_INTSET) {
if (!intsetGet(si->subject->ptr,si->ii++,llele)) if (!intsetGet(si->subject->ptr,si->ii++,llele))
return -1; return -1;
*objele = NULL; /* Not needed. Defensive. */ *sdsele = NULL; /* Not needed. Defensive. */
} else { } else {
serverPanic("Wrong set encoding in setTypeNext"); serverPanic("Wrong set encoding in setTypeNext");
} }
@ -170,25 +168,24 @@ int setTypeNext(setTypeIterator *si, robj **objele, int64_t *llele) {
} }
/* The not copy on write friendly version but easy to use version /* The not copy on write friendly version but easy to use version
* of setTypeNext() is setTypeNextObject(), returning new objects * of setTypeNext() is setTypeNextObject(), returning new SDS
* or incrementing the ref count of returned objects. So if you don't * strings. So if you don't retain a pointer to this object you should call
* retain a pointer to this object you should call decrRefCount() against it. * sdsfree() against it.
* *
* This function is the way to go for write operations where COW is not * This function is the way to go for write operations where COW is not
* an issue as the result will be anyway of incrementing the ref count. */ * an issue. */
robj *setTypeNextObject(setTypeIterator *si) { sds setTypeNextObject(setTypeIterator *si) {
int64_t intele; int64_t intele;
robj *objele; sds sdsele;
int encoding; int encoding;
encoding = setTypeNext(si,&objele,&intele); encoding = setTypeNext(si,&sdsele,&intele);
switch(encoding) { switch(encoding) {
case -1: return NULL; case -1: return NULL;
case OBJ_ENCODING_INTSET: case OBJ_ENCODING_INTSET:
return createStringObjectFromLongLong(intele); return sdsfromlonglong(intele);
case OBJ_ENCODING_HT: case OBJ_ENCODING_HT:
incrRefCount(objele); return sdsdup(sdsele);
return objele;
default: default:
serverPanic("Unsupported encoding"); serverPanic("Unsupported encoding");
} }
@ -197,7 +194,7 @@ robj *setTypeNextObject(setTypeIterator *si) {
/* Return random element from a non empty set. /* Return random element from a non empty set.
* The returned element can be a int64_t value if the set is encoded * The returned element can be a int64_t value if the set is encoded
* as an "intset" blob of integers, or a redis object if the set * as an "intset" blob of integers, or an SDS string if the set
* is a regular set. * is a regular set.
* *
* The caller provides both pointers to be populated with the right * The caller provides both pointers to be populated with the right
@ -205,21 +202,17 @@ robj *setTypeNextObject(setTypeIterator *si) {
* field of the object and is used by the caller to check if the * field of the object and is used by the caller to check if the
* int64_t pointer or the redis object pointer was populated. * int64_t pointer or the redis object pointer was populated.
* *
* Note that both the objele and llele pointers should be passed and cannot * Note that both the sdsele and llele pointers should be passed and cannot
* be NULL since the function will try to defensively populate the non * be NULL since the function will try to defensively populate the non
* used field with values which are easy to trap if misused. * used field with values which are easy to trap if misused. */
* int setTypeRandomElement(robj *setobj, sds *sdsele, int64_t *llele) {
* When an object is returned (the set was a real set) the ref count
* of the object is not incremented so this function can be considered
* copy on write friendly. */
int setTypeRandomElement(robj *setobj, robj **objele, int64_t *llele) {
if (setobj->encoding == OBJ_ENCODING_HT) { if (setobj->encoding == OBJ_ENCODING_HT) {
dictEntry *de = dictGetRandomKey(setobj->ptr); dictEntry *de = dictGetRandomKey(setobj->ptr);
*objele = dictGetKey(de); *sdsele = dictGetKey(de);
*llele = -123456789; /* Not needed. Defensive. */ *llele = -123456789; /* Not needed. Defensive. */
} else if (setobj->encoding == OBJ_ENCODING_INTSET) { } else if (setobj->encoding == OBJ_ENCODING_INTSET) {
*llele = intsetRandom(setobj->ptr); *llele = intsetRandom(setobj->ptr);
*objele = NULL; /* Not needed. Defensive. */ *sdsele = NULL; /* Not needed. Defensive. */
} else { } else {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
} }
@ -247,7 +240,7 @@ void setTypeConvert(robj *setobj, int enc) {
if (enc == OBJ_ENCODING_HT) { if (enc == OBJ_ENCODING_HT) {
int64_t intele; int64_t intele;
dict *d = dictCreate(&setDictType,NULL); dict *d = dictCreate(&setDictType,NULL);
robj *element; sds element;
/* Presize the dict to avoid rehashing */ /* Presize the dict to avoid rehashing */
dictExpand(d,intsetLen(setobj->ptr)); dictExpand(d,intsetLen(setobj->ptr));
@ -255,9 +248,8 @@ void setTypeConvert(robj *setobj, int enc) {
/* To add the elements we extract integers and create redis objects */ /* To add the elements we extract integers and create redis objects */
si = setTypeInitIterator(setobj); si = setTypeInitIterator(setobj);
while (setTypeNext(si,&element,&intele) != -1) { while (setTypeNext(si,&element,&intele) != -1) {
element = createStringObjectFromLongLong(intele); element = sdsfromlonglong(intele);
serverAssertWithInfo(NULL,element, serverAssert(dictAdd(d,element,NULL) == DICT_OK);
dictAdd(d,element,NULL) == DICT_OK);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
@ -275,7 +267,7 @@ void saddCommand(client *c) {
set = lookupKeyWrite(c->db,c->argv[1]); set = lookupKeyWrite(c->db,c->argv[1]);
if (set == NULL) { if (set == NULL) {
set = setTypeCreate(c->argv[2]); set = setTypeCreate(c->argv[2]->ptr);
dbAdd(c->db,c->argv[1],set); dbAdd(c->db,c->argv[1],set);
} else { } else {
if (set->type != OBJ_SET) { if (set->type != OBJ_SET) {
@ -285,8 +277,7 @@ void saddCommand(client *c) {
} }
for (j = 2; j < c->argc; j++) { for (j = 2; j < c->argc; j++) {
c->argv[j] = tryObjectEncoding(c->argv[j]); if (setTypeAdd(set,c->argv[j]->ptr)) added++;
if (setTypeAdd(set,c->argv[j])) added++;
} }
if (added) { if (added) {
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
@ -304,7 +295,7 @@ void sremCommand(client *c) {
checkType(c,set,OBJ_SET)) return; checkType(c,set,OBJ_SET)) return;
for (j = 2; j < c->argc; j++) { for (j = 2; j < c->argc; j++) {
if (setTypeRemove(set,c->argv[j])) { if (setTypeRemove(set,c->argv[j]->ptr)) {
deleted++; deleted++;
if (setTypeSize(set) == 0) { if (setTypeSize(set) == 0) {
dbDelete(c->db,c->argv[1]); dbDelete(c->db,c->argv[1]);
@ -328,7 +319,7 @@ void smoveCommand(client *c) {
robj *srcset, *dstset, *ele; robj *srcset, *dstset, *ele;
srcset = lookupKeyWrite(c->db,c->argv[1]); srcset = lookupKeyWrite(c->db,c->argv[1]);
dstset = lookupKeyWrite(c->db,c->argv[2]); dstset = lookupKeyWrite(c->db,c->argv[2]);
ele = c->argv[3] = tryObjectEncoding(c->argv[3]); ele = c->argv[3];
/* If the source key does not exist return 0 */ /* If the source key does not exist return 0 */
if (srcset == NULL) { if (srcset == NULL) {
@ -343,12 +334,13 @@ void smoveCommand(client *c) {
/* If srcset and dstset are equal, SMOVE is a no-op */ /* If srcset and dstset are equal, SMOVE is a no-op */
if (srcset == dstset) { if (srcset == dstset) {
addReply(c,setTypeIsMember(srcset,ele) ? shared.cone : shared.czero); addReply(c,setTypeIsMember(srcset,ele->ptr) ?
shared.cone : shared.czero);
return; return;
} }
/* If the element cannot be removed from the src set, return 0. */ /* If the element cannot be removed from the src set, return 0. */
if (!setTypeRemove(srcset,ele)) { if (!setTypeRemove(srcset,ele->ptr)) {
addReply(c,shared.czero); addReply(c,shared.czero);
return; return;
} }
@ -365,12 +357,12 @@ void smoveCommand(client *c) {
/* Create the destination set when it doesn't exist */ /* Create the destination set when it doesn't exist */
if (!dstset) { if (!dstset) {
dstset = setTypeCreate(ele); dstset = setTypeCreate(ele->ptr);
dbAdd(c->db,c->argv[2],dstset); dbAdd(c->db,c->argv[2],dstset);
} }
/* An extra key has changed when ele was successfully added to dstset */ /* An extra key has changed when ele was successfully added to dstset */
if (setTypeAdd(dstset,ele)) { if (setTypeAdd(dstset,ele->ptr)) {
server.dirty++; server.dirty++;
notifyKeyspaceEvent(NOTIFY_SET,"sadd",c->argv[2],c->db->id); notifyKeyspaceEvent(NOTIFY_SET,"sadd",c->argv[2],c->db->id);
} }
@ -383,8 +375,7 @@ void sismemberCommand(client *c) {
if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL || if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,set,OBJ_SET)) return; checkType(c,set,OBJ_SET)) return;
c->argv[2] = tryObjectEncoding(c->argv[2]); if (setTypeIsMember(set,c->argv[2]->ptr))
if (setTypeIsMember(set,c->argv[2]))
addReply(c,shared.cone); addReply(c,shared.cone);
else else
addReply(c,shared.czero); addReply(c,shared.czero);
@ -457,7 +448,7 @@ void spopWithCountCommand(client *c) {
return; return;
} }
/* Case 2 and 3 require to replicate SPOP as a set of SERM commands. /* Case 2 and 3 require to replicate SPOP as a set of SREM commands.
* Prepare our replication argument vector. Also send the array length * Prepare our replication argument vector. Also send the array length
* which is common to both the code paths. */ * which is common to both the code paths. */
robj *propargv[3]; robj *propargv[3];
@ -466,6 +457,7 @@ void spopWithCountCommand(client *c) {
addReplyMultiBulkLen(c,count); addReplyMultiBulkLen(c,count);
/* Common iteration vars. */ /* Common iteration vars. */
sds sdsele;
robj *objele; robj *objele;
int encoding; int encoding;
int64_t llele; int64_t llele;
@ -480,17 +472,18 @@ void spopWithCountCommand(client *c) {
* the set. */ * the set. */
if (remaining*SPOP_MOVE_STRATEGY_MUL > count) { if (remaining*SPOP_MOVE_STRATEGY_MUL > count) {
while(count--) { while(count--) {
encoding = setTypeRandomElement(set,&objele,&llele); /* Emit and remove. */
encoding = setTypeRandomElement(set,&sdsele,&llele);
if (encoding == OBJ_ENCODING_INTSET) { if (encoding == OBJ_ENCODING_INTSET) {
addReplyBulkLongLong(c,llele);
objele = createStringObjectFromLongLong(llele); objele = createStringObjectFromLongLong(llele);
set->ptr = intsetRemove(set->ptr,llele,NULL);
} else { } else {
incrRefCount(objele); addReplyBulkCBuffer(c,sdsele,sdslen(sdsele));
objele = createStringObject(sdsele,sdslen(sdsele));
setTypeRemove(set,sdsele);
} }
/* Return the element to the client and remove from the set. */
addReplyBulk(c,objele);
setTypeRemove(set,objele);
/* Replicate/AOF this command as an SREM operation */ /* Replicate/AOF this command as an SREM operation */
propargv[2] = objele; propargv[2] = objele;
alsoPropagate(server.sremCommand,c->db->id,propargv,3, alsoPropagate(server.sremCommand,c->db->id,propargv,3,
@ -510,16 +503,16 @@ void spopWithCountCommand(client *c) {
/* Create a new set with just the remaining elements. */ /* Create a new set with just the remaining elements. */
while(remaining--) { while(remaining--) {
encoding = setTypeRandomElement(set,&objele,&llele); encoding = setTypeRandomElement(set,&sdsele,&llele);
if (encoding == OBJ_ENCODING_INTSET) { if (encoding == OBJ_ENCODING_INTSET) {
objele = createStringObjectFromLongLong(llele); sdsele = sdsfromlonglong(llele);
} else { } else {
incrRefCount(objele); sdsele = sdsdup(sdsele);
} }
if (!newset) newset = setTypeCreate(objele); if (!newset) newset = setTypeCreate(sdsele);
setTypeAdd(newset,objele); setTypeAdd(newset,sdsele);
setTypeRemove(set,objele); setTypeRemove(set,sdsele);
decrRefCount(objele); sdsfree(sdsele);
} }
/* Assign the new set as the key value. */ /* Assign the new set as the key value. */
@ -529,19 +522,19 @@ void spopWithCountCommand(client *c) {
/* Tranfer the old set to the client and release it. */ /* Tranfer the old set to the client and release it. */
setTypeIterator *si; setTypeIterator *si;
si = setTypeInitIterator(set); si = setTypeInitIterator(set);
while((encoding = setTypeNext(si,&objele,&llele)) != -1) { while((encoding = setTypeNext(si,&sdsele,&llele)) != -1) {
if (encoding == OBJ_ENCODING_INTSET) { if (encoding == OBJ_ENCODING_INTSET) {
addReplyBulkLongLong(c,llele);
objele = createStringObjectFromLongLong(llele); objele = createStringObjectFromLongLong(llele);
} else { } else {
incrRefCount(objele); addReplyBulkCBuffer(c,sdsele,sdslen(sdsele));
objele = createStringObject(sdsele,sdslen(sdsele));
} }
addReplyBulk(c,objele);
/* Replicate/AOF this command as an SREM operation */ /* Replicate/AOF this command as an SREM operation */
propargv[2] = objele; propargv[2] = objele;
alsoPropagate(server.sremCommand,c->db->id,propargv,3, alsoPropagate(server.sremCommand,c->db->id,propargv,3,
PROPAGATE_AOF|PROPAGATE_REPL); PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(objele); decrRefCount(objele);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
@ -558,6 +551,7 @@ void spopWithCountCommand(client *c) {
void spopCommand(client *c) { void spopCommand(client *c) {
robj *set, *ele, *aux; robj *set, *ele, *aux;
sds sdsele;
int64_t llele; int64_t llele;
int encoding; int encoding;
@ -575,15 +569,15 @@ void spopCommand(client *c) {
checkType(c,set,OBJ_SET)) return; checkType(c,set,OBJ_SET)) return;
/* Get a random element from the set */ /* Get a random element from the set */
encoding = setTypeRandomElement(set,&ele,&llele); encoding = setTypeRandomElement(set,&sdsele,&llele);
/* Remove the element from the set */ /* Remove the element from the set */
if (encoding == OBJ_ENCODING_INTSET) { if (encoding == OBJ_ENCODING_INTSET) {
ele = createStringObjectFromLongLong(llele); ele = createStringObjectFromLongLong(llele);
set->ptr = intsetRemove(set->ptr,llele,NULL); set->ptr = intsetRemove(set->ptr,llele,NULL);
} else { } else {
incrRefCount(ele); ele = createStringObject(sdsele,sdslen(sdsele));
setTypeRemove(set,ele); setTypeRemove(set,ele->ptr);
} }
notifyKeyspaceEvent(NOTIFY_SET,"spop",c->argv[1],c->db->id); notifyKeyspaceEvent(NOTIFY_SET,"spop",c->argv[1],c->db->id);
@ -591,11 +585,11 @@ void spopCommand(client *c) {
/* Replicate/AOF this command as an SREM operation */ /* Replicate/AOF this command as an SREM operation */
aux = createStringObject("SREM",4); aux = createStringObject("SREM",4);
rewriteClientCommandVector(c,3,aux,c->argv[1],ele); rewriteClientCommandVector(c,3,aux,c->argv[1],ele);
decrRefCount(ele);
decrRefCount(aux); decrRefCount(aux);
/* Add the element to the reply */ /* Add the element to the reply */
addReplyBulk(c,ele); addReplyBulk(c,ele);
decrRefCount(ele);
/* Delete the set if it's empty */ /* Delete the set if it's empty */
if (setTypeSize(set) == 0) { if (setTypeSize(set) == 0) {
@ -620,7 +614,8 @@ void srandmemberWithCountCommand(client *c) {
long l; long l;
unsigned long count, size; unsigned long count, size;
int uniq = 1; int uniq = 1;
robj *set, *ele; robj *set;
sds ele;
int64_t llele; int64_t llele;
int encoding; int encoding;
@ -657,7 +652,7 @@ void srandmemberWithCountCommand(client *c) {
if (encoding == OBJ_ENCODING_INTSET) { if (encoding == OBJ_ENCODING_INTSET) {
addReplyBulkLongLong(c,llele); addReplyBulkLongLong(c,llele);
} else { } else {
addReplyBulk(c,ele); addReplyBulkCBuffer(c,ele,sdslen(ele));
} }
} }
return; return;
@ -672,7 +667,7 @@ void srandmemberWithCountCommand(client *c) {
} }
/* For CASE 3 and CASE 4 we need an auxiliary dictionary. */ /* For CASE 3 and CASE 4 we need an auxiliary dictionary. */
d = dictCreate(&setDictType,NULL); d = dictCreate(&objectKeyPointerValueDictType,NULL);
/* CASE 3: /* CASE 3:
* The number of elements inside the set is not greater than * The number of elements inside the set is not greater than
@ -694,7 +689,7 @@ void srandmemberWithCountCommand(client *c) {
if (encoding == OBJ_ENCODING_INTSET) { if (encoding == OBJ_ENCODING_INTSET) {
retval = dictAdd(d,createStringObjectFromLongLong(llele),NULL); retval = dictAdd(d,createStringObjectFromLongLong(llele),NULL);
} else { } else {
retval = dictAdd(d,dupStringObject(ele),NULL); retval = dictAdd(d,createStringObject(ele,sdslen(ele)),NULL);
} }
serverAssert(retval == DICT_OK); serverAssert(retval == DICT_OK);
} }
@ -717,21 +712,22 @@ void srandmemberWithCountCommand(client *c) {
* to reach the specified count. */ * to reach the specified count. */
else { else {
unsigned long added = 0; unsigned long added = 0;
robj *objele;
while(added < count) { while(added < count) {
encoding = setTypeRandomElement(set,&ele,&llele); encoding = setTypeRandomElement(set,&ele,&llele);
if (encoding == OBJ_ENCODING_INTSET) { if (encoding == OBJ_ENCODING_INTSET) {
ele = createStringObjectFromLongLong(llele); objele = createStringObjectFromLongLong(llele);
} else { } else {
ele = dupStringObject(ele); objele = createStringObject(ele,sdslen(ele));
} }
/* Try to add the object to the dictionary. If it already exists /* Try to add the object to the dictionary. If it already exists
* free it, otherwise increment the number of objects we have * free it, otherwise increment the number of objects we have
* in the result dictionary. */ * in the result dictionary. */
if (dictAdd(d,ele,NULL) == DICT_OK) if (dictAdd(d,objele,NULL) == DICT_OK)
added++; added++;
else else
decrRefCount(ele); decrRefCount(objele);
} }
} }
@ -750,7 +746,8 @@ void srandmemberWithCountCommand(client *c) {
} }
void srandmemberCommand(client *c) { void srandmemberCommand(client *c) {
robj *set, *ele; robj *set;
sds ele;
int64_t llele; int64_t llele;
int encoding; int encoding;
@ -769,7 +766,7 @@ void srandmemberCommand(client *c) {
if (encoding == OBJ_ENCODING_INTSET) { if (encoding == OBJ_ENCODING_INTSET) {
addReplyBulkLongLong(c,llele); addReplyBulkLongLong(c,llele);
} else { } else {
addReplyBulk(c,ele); addReplyBulkCBuffer(c,ele,sdslen(ele));
} }
} }
@ -789,7 +786,8 @@ void sinterGenericCommand(client *c, robj **setkeys,
unsigned long setnum, robj *dstkey) { unsigned long setnum, robj *dstkey) {
robj **sets = zmalloc(sizeof(robj*)*setnum); robj **sets = zmalloc(sizeof(robj*)*setnum);
setTypeIterator *si; setTypeIterator *si;
robj *eleobj, *dstset = NULL; robj *dstset = NULL;
sds elesds;
int64_t intobj; int64_t intobj;
void *replylen = NULL; void *replylen = NULL;
unsigned long j, cardinality = 0; unsigned long j, cardinality = 0;
@ -839,7 +837,7 @@ void sinterGenericCommand(client *c, robj **setkeys,
* the element against all the other sets, if at least one set does * the element against all the other sets, if at least one set does
* not include the element it is discarded */ * not include the element it is discarded */
si = setTypeInitIterator(sets[0]); si = setTypeInitIterator(sets[0]);
while((encoding = setTypeNext(si,&eleobj,&intobj)) != -1) { while((encoding = setTypeNext(si,&elesds,&intobj)) != -1) {
for (j = 1; j < setnum; j++) { for (j = 1; j < setnum; j++) {
if (sets[j] == sets[0]) continue; if (sets[j] == sets[0]) continue;
if (encoding == OBJ_ENCODING_INTSET) { if (encoding == OBJ_ENCODING_INTSET) {
@ -852,25 +850,15 @@ void sinterGenericCommand(client *c, robj **setkeys,
* have to use the generic function, creating an object * have to use the generic function, creating an object
* for this */ * for this */
} else if (sets[j]->encoding == OBJ_ENCODING_HT) { } else if (sets[j]->encoding == OBJ_ENCODING_HT) {
eleobj = createStringObjectFromLongLong(intobj); elesds = sdsfromlonglong(intobj);
if (!setTypeIsMember(sets[j],eleobj)) { if (!setTypeIsMember(sets[j],elesds)) {
decrRefCount(eleobj); sdsfree(elesds);
break; break;
} }
decrRefCount(eleobj); sdsfree(elesds);
} }
} else if (encoding == OBJ_ENCODING_HT) { } else if (encoding == OBJ_ENCODING_HT) {
/* Optimization... if the source object is integer if (!setTypeIsMember(sets[j],elesds)) {
* encoded AND the target set is an intset, we can get
* a much faster path. */
if (eleobj->encoding == OBJ_ENCODING_INT &&
sets[j]->encoding == OBJ_ENCODING_INTSET &&
!intsetFind((intset*)sets[j]->ptr,(long)eleobj->ptr))
{
break;
/* else... object to object check is easy as we use the
* type agnostic API here. */
} else if (!setTypeIsMember(sets[j],eleobj)) {
break; break;
} }
} }
@ -880,17 +868,17 @@ void sinterGenericCommand(client *c, robj **setkeys,
if (j == setnum) { if (j == setnum) {
if (!dstkey) { if (!dstkey) {
if (encoding == OBJ_ENCODING_HT) if (encoding == OBJ_ENCODING_HT)
addReplyBulk(c,eleobj); addReplyBulkCBuffer(c,elesds,sdslen(elesds));
else else
addReplyBulkLongLong(c,intobj); addReplyBulkLongLong(c,intobj);
cardinality++; cardinality++;
} else { } else {
if (encoding == OBJ_ENCODING_INTSET) { if (encoding == OBJ_ENCODING_INTSET) {
eleobj = createStringObjectFromLongLong(intobj); elesds = sdsfromlonglong(intobj);
setTypeAdd(dstset,eleobj); setTypeAdd(dstset,elesds);
decrRefCount(eleobj); sdsfree(elesds);
} else { } else {
setTypeAdd(dstset,eleobj); setTypeAdd(dstset,elesds);
} }
} }
} }
@ -937,7 +925,8 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
robj *dstkey, int op) { robj *dstkey, int op) {
robj **sets = zmalloc(sizeof(robj*)*setnum); robj **sets = zmalloc(sizeof(robj*)*setnum);
setTypeIterator *si; setTypeIterator *si;
robj *ele, *dstset = NULL; robj *dstset = NULL;
sds ele;
int j, cardinality = 0; int j, cardinality = 0;
int diff_algo = 1; int diff_algo = 1;
@ -1003,7 +992,7 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
si = setTypeInitIterator(sets[j]); si = setTypeInitIterator(sets[j]);
while((ele = setTypeNextObject(si)) != NULL) { while((ele = setTypeNextObject(si)) != NULL) {
if (setTypeAdd(dstset,ele)) cardinality++; if (setTypeAdd(dstset,ele)) cardinality++;
decrRefCount(ele); sdsfree(ele);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
} }
@ -1028,7 +1017,7 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
setTypeAdd(dstset,ele); setTypeAdd(dstset,ele);
cardinality++; cardinality++;
} }
decrRefCount(ele); sdsfree(ele);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
} else if (op == SET_OP_DIFF && sets[0] && diff_algo == 2) { } else if (op == SET_OP_DIFF && sets[0] && diff_algo == 2) {
@ -1049,7 +1038,7 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
} else { } else {
if (setTypeRemove(dstset,ele)) cardinality--; if (setTypeRemove(dstset,ele)) cardinality--;
} }
decrRefCount(ele); sdsfree(ele);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
@ -1064,8 +1053,8 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
addReplyMultiBulkLen(c,cardinality); addReplyMultiBulkLen(c,cardinality);
si = setTypeInitIterator(dstset); si = setTypeInitIterator(dstset);
while((ele = setTypeNextObject(si)) != NULL) { while((ele = setTypeNextObject(si)) != NULL) {
addReplyBulk(c,ele); addReplyBulkCBuffer(c,ele,sdslen(ele));
decrRefCount(ele); sdsfree(ele);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
decrRefCount(dstset); decrRefCount(dstset);