Unverified Commit 4e472a1a authored by Viktor Söderqvist's avatar Viktor Söderqvist Committed by GitHub
Browse files

Listpack encoding for sets (#11290)

Small sets with not only integer elements are listpack encoded, by default
up to 128 elements, max 64 bytes per element, new config `set-max-listpack-entries`
and `set-max-listpack-value`. This saves memory for small sets compared to using a hashtable.

Sets with only integers, even very small sets, are still intset encoded (up to 1G
limit, etc.). Larger sets are hashtable encoded.

This PR increments the RDB version, and has an effect on OBJECT ENCODING

Possible conversions when elements are added:

    intset -> listpack
    listpack -> hashtable
    intset -> hashtable

Note: No conversion happens when elements are deleted. If all elements are
deleted and then added again, the set is deleted and recreated, thus implicitly
converted to a smaller encoding.
parent 07d18706
...@@ -1951,13 +1951,20 @@ list-max-listpack-size -2 ...@@ -1951,13 +1951,20 @@ list-max-listpack-size -2
# etc. # etc.
list-compress-depth 0 list-compress-depth 0
# Sets have a special encoding in just one case: when a set is composed # Sets have a special encoding when a set is composed
# of just strings that happen to be integers in radix 10 in the range # of just strings that happen to be integers in radix 10 in the range
# of 64 bit signed integers. # of 64 bit signed integers.
# The following configuration setting sets the limit in the size of the # The following configuration setting sets the limit in the size of the
# set in order to use this special memory saving encoding. # set in order to use this special memory saving encoding.
set-max-intset-entries 512 set-max-intset-entries 512
# Sets containing non-integer values are also encoded using a memory efficient
# data structure when they have a small number of entries, and the biggest entry
# does not exceed a given threshold. These thresholds can be configured using
# the following directives.
set-max-listpack-entries 128
set-max-listpack-value 64
# Similarly to hashes and lists, sorted sets are also specially encoded in # Similarly to hashes and lists, sorted sets are also specially encoded in
# order to save a lot of space. This encoding is only used when the length and # order to save a lot of space. This encoding is only used when the length and
# elements of a sorted set are below the following limits: # elements of a sorted set are below the following limits:
......
...@@ -1818,56 +1818,31 @@ int rewriteListObject(rio *r, robj *key, robj *o) { ...@@ -1818,56 +1818,31 @@ int rewriteListObject(rio *r, robj *key, robj *o) {
* The function returns 0 on error, 1 on success. */ * The function returns 0 on error, 1 on success. */
int rewriteSetObject(rio *r, robj *key, robj *o) { int rewriteSetObject(rio *r, robj *key, robj *o) {
long long count = 0, items = setTypeSize(o); long long count = 0, items = setTypeSize(o);
setTypeIterator *si = setTypeInitIterator(o);
if (o->encoding == OBJ_ENCODING_INTSET) { char *str;
int ii = 0; size_t len;
int64_t llval; int64_t llval;
while (setTypeNext(si, &str, &len, &llval) != -1) {
while(intsetGet(o->ptr,ii++,&llval)) { if (count == 0) {
if (count == 0) { int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ?
int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ? AOF_REWRITE_ITEMS_PER_CMD : items;
AOF_REWRITE_ITEMS_PER_CMD : items; if (!rioWriteBulkCount(r,'*',2+cmd_items) ||
!rioWriteBulkString(r,"SADD",4) ||
if (!rioWriteBulkCount(r,'*',2+cmd_items) || !rioWriteBulkObject(r,key))
!rioWriteBulkString(r,"SADD",4) || {
!rioWriteBulkObject(r,key)) return 0;
{
return 0;
}
} }
if (!rioWriteBulkLongLong(r,llval)) return 0;
if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0;
items--;
} }
} else if (o->encoding == OBJ_ENCODING_HT) { size_t written = str ?
dictIterator *di = dictGetIterator(o->ptr); rioWriteBulkString(r, str, len) : rioWriteBulkLongLong(r, llval);
dictEntry *de; if (!written) {
setTypeReleaseIterator(si);
while((de = dictNext(di)) != NULL) { return 0;
sds ele = dictGetKey(de);
if (count == 0) {
int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ?
AOF_REWRITE_ITEMS_PER_CMD : items;
if (!rioWriteBulkCount(r,'*',2+cmd_items) ||
!rioWriteBulkString(r,"SADD",4) ||
!rioWriteBulkObject(r,key))
{
dictReleaseIterator(di);
return 0;
}
}
if (!rioWriteBulkString(r,ele,sdslen(ele))) {
dictReleaseIterator(di);
return 0;
}
if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0;
items--;
} }
dictReleaseIterator(di); if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0;
} else { items--;
serverPanic("Unknown set encoding");
} }
setTypeReleaseIterator(si);
return 1; return 1;
} }
......
...@@ -3130,6 +3130,8 @@ standardConfig static_configs[] = { ...@@ -3130,6 +3130,8 @@ standardConfig static_configs[] = {
/* Size_t configs */ /* Size_t configs */
createSizeTConfig("hash-max-listpack-entries", "hash-max-ziplist-entries", MODIFIABLE_CONFIG, 0, LONG_MAX, server.hash_max_listpack_entries, 512, INTEGER_CONFIG, NULL, NULL), createSizeTConfig("hash-max-listpack-entries", "hash-max-ziplist-entries", MODIFIABLE_CONFIG, 0, LONG_MAX, server.hash_max_listpack_entries, 512, INTEGER_CONFIG, NULL, NULL),
createSizeTConfig("set-max-intset-entries", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.set_max_intset_entries, 512, INTEGER_CONFIG, NULL, NULL), createSizeTConfig("set-max-intset-entries", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.set_max_intset_entries, 512, INTEGER_CONFIG, NULL, NULL),
createSizeTConfig("set-max-listpack-entries", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.set_max_listpack_entries, 128, INTEGER_CONFIG, NULL, NULL),
createSizeTConfig("set-max-listpack-value", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.set_max_listpack_value, 64, INTEGER_CONFIG, NULL, NULL),
createSizeTConfig("zset-max-listpack-entries", "zset-max-ziplist-entries", MODIFIABLE_CONFIG, 0, LONG_MAX, server.zset_max_listpack_entries, 128, INTEGER_CONFIG, NULL, NULL), createSizeTConfig("zset-max-listpack-entries", "zset-max-ziplist-entries", MODIFIABLE_CONFIG, 0, LONG_MAX, server.zset_max_listpack_entries, 128, INTEGER_CONFIG, NULL, NULL),
createSizeTConfig("active-defrag-ignore-bytes", NULL, MODIFIABLE_CONFIG, 1, LLONG_MAX, server.active_defrag_ignore_bytes, 100<<20, MEMORY_CONFIG, NULL, NULL), /* Default: don't defrag if frag overhead is below 100mb */ createSizeTConfig("active-defrag-ignore-bytes", NULL, MODIFIABLE_CONFIG, 1, LLONG_MAX, server.active_defrag_ignore_bytes, 100<<20, MEMORY_CONFIG, NULL, NULL), /* Default: don't defrag if frag overhead is below 100mb */
createSizeTConfig("hash-max-listpack-value", "hash-max-ziplist-value", MODIFIABLE_CONFIG, 0, LONG_MAX, server.hash_max_listpack_value, 64, MEMORY_CONFIG, NULL, NULL), createSizeTConfig("hash-max-listpack-value", "hash-max-ziplist-value", MODIFIABLE_CONFIG, 0, LONG_MAX, server.hash_max_listpack_value, 64, MEMORY_CONFIG, NULL, NULL),
......
...@@ -915,14 +915,16 @@ void scanGenericCommand(client *c, robj *o, unsigned long cursor) { ...@@ -915,14 +915,16 @@ void scanGenericCommand(client *c, robj *o, unsigned long cursor) {
} while (cursor && } while (cursor &&
maxiterations-- && maxiterations-- &&
listLength(keys) < (unsigned long)count); listLength(keys) < (unsigned long)count);
} else if (o->type == OBJ_SET) { } else if (o->type == OBJ_SET && o->encoding == OBJ_ENCODING_INTSET) {
int pos = 0; int pos = 0;
int64_t ll; int64_t ll;
while(intsetGet(o->ptr,pos++,&ll)) while(intsetGet(o->ptr,pos++,&ll))
listAddNodeTail(keys,createStringObjectFromLongLong(ll)); listAddNodeTail(keys,createStringObjectFromLongLong(ll));
cursor = 0; cursor = 0;
} else if (o->type == OBJ_HASH || o->type == OBJ_ZSET) { } else if ((o->type == OBJ_HASH || o->type == OBJ_ZSET || o->type == OBJ_SET) &&
o->encoding == OBJ_ENCODING_LISTPACK)
{
unsigned char *p = lpFirst(o->ptr); unsigned char *p = lpFirst(o->ptr);
unsigned char *vstr; unsigned char *vstr;
int64_t vlen; int64_t vlen;
......
...@@ -874,10 +874,12 @@ long defragKey(redisDb *db, dictEntry *de) { ...@@ -874,10 +874,12 @@ long defragKey(redisDb *db, dictEntry *de) {
} else if (ob->type == OBJ_SET) { } else if (ob->type == OBJ_SET) {
if (ob->encoding == OBJ_ENCODING_HT) { if (ob->encoding == OBJ_ENCODING_HT) {
defragged += defragSet(db, de); defragged += defragSet(db, de);
} else if (ob->encoding == OBJ_ENCODING_INTSET) { } else if (ob->encoding == OBJ_ENCODING_INTSET ||
intset *newis, *is = ob->ptr; ob->encoding == OBJ_ENCODING_LISTPACK)
if ((newis = activeDefragAlloc(is))) {
defragged++, ob->ptr = newis; void *newptr, *ptr = ob->ptr;
if ((newptr = activeDefragAlloc(ptr)))
defragged++, ob->ptr = newptr;
} else { } else {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
} }
......
...@@ -265,6 +265,17 @@ int64_t intsetRandom(intset *is) { ...@@ -265,6 +265,17 @@ int64_t intsetRandom(intset *is) {
return _intsetGet(is,rand()%len); return _intsetGet(is,rand()%len);
} }
/* Return the largest member. */
int64_t intsetMax(intset *is) {
uint32_t len = intrev32ifbe(is->length);
return _intsetGet(is, len - 1);
}
/* Return the smallest member. */
int64_t intsetMin(intset *is) {
return _intsetGet(is, 0);
}
/* Get the value at the given position. When this position is /* Get the value at the given position. When this position is
* out of range the function returns 0, when in range it returns 1. */ * out of range the function returns 0, when in range it returns 1. */
uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value) { uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value) {
...@@ -425,6 +436,8 @@ int intsetTest(int argc, char **argv, int flags) { ...@@ -425,6 +436,8 @@ int intsetTest(int argc, char **argv, int flags) {
is = intsetAdd(is,6,&success); assert(success); is = intsetAdd(is,6,&success); assert(success);
is = intsetAdd(is,4,&success); assert(success); is = intsetAdd(is,4,&success); assert(success);
is = intsetAdd(is,4,&success); assert(!success); is = intsetAdd(is,4,&success); assert(!success);
assert(6 == intsetMax(is));
assert(4 == intsetMin(is));
ok(); ok();
zfree(is); zfree(is);
} }
......
...@@ -43,6 +43,8 @@ intset *intsetAdd(intset *is, int64_t value, uint8_t *success); ...@@ -43,6 +43,8 @@ intset *intsetAdd(intset *is, int64_t value, uint8_t *success);
intset *intsetRemove(intset *is, int64_t value, int *success); intset *intsetRemove(intset *is, int64_t value, int *success);
uint8_t intsetFind(intset *is, int64_t value); uint8_t intsetFind(intset *is, int64_t value);
int64_t intsetRandom(intset *is); int64_t intsetRandom(intset *is);
int64_t intsetMax(intset *is);
int64_t intsetMin(intset *is);
uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value); uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value);
uint32_t intsetLen(const intset *is); uint32_t intsetLen(const intset *is);
size_t intsetBlobLen(intset *is); size_t intsetBlobLen(intset *is);
......
...@@ -1063,6 +1063,55 @@ unsigned char *lpDeleteRange(unsigned char *lp, long index, unsigned long num) { ...@@ -1063,6 +1063,55 @@ unsigned char *lpDeleteRange(unsigned char *lp, long index, unsigned long num) {
return lp; return lp;
} }
/* Delete the elements 'ps' passed as an array of 'count' element pointers and
* return the resulting listpack. The elements must be given in the same order
* as they apper in the listpack. */
unsigned char *lpBatchDelete(unsigned char *lp, unsigned char **ps, unsigned long count) {
if (count == 0) return lp;
unsigned char *dst = ps[0];
size_t total_bytes = lpGetTotalBytes(lp);
unsigned char *lp_end = lp + total_bytes; /* After the EOF element. */
assert(lp_end[-1] == LP_EOF);
/*
* ----+--------+-----------+--------+---------+-----+---+
* ... | Delete | Keep | Delete | Keep | ... |EOF|
* ... |xxxxxxxx| |xxxxxxxx| | ... | |
* ----+--------+-----------+--------+---------+-----+---+
* ^ ^ ^ ^
* | | | |
* ps[i] | ps[i+1] |
* skip keep_start keep_end lp_end
*
* The loop memmoves the bytes between keep_start and keep_end to dst.
*/
for (unsigned long i = 0; i < count; i++) {
unsigned char *skip = ps[i];
assert(skip != NULL && skip[0] != LP_EOF);
unsigned char *keep_start = lpSkip(skip);
unsigned char *keep_end;
if (i + 1 < count) {
keep_end = ps[i + 1];
/* Deleting consecutive elements. Nothing to keep between them. */
if (keep_start == keep_end) continue;
} else {
/* Keep the rest of the listpack including the EOF marker. */
keep_end = lp_end;
}
assert(keep_end > keep_start);
size_t bytes_to_keep = keep_end - keep_start;
memmove(dst, keep_start, bytes_to_keep);
dst += bytes_to_keep;
}
/* Update total size and num elements. */
size_t deleted_bytes = lp_end - dst;
total_bytes -= deleted_bytes;
assert(lp[total_bytes - 1] == LP_EOF);
lpSetTotalBytes(lp, total_bytes);
uint32_t numele = lpGetNumElements(lp);
if (numele != LP_HDR_NUMELE_UNKNOWN) lpSetNumElements(lp, numele - count);
return lpShrinkToFit(lp);
}
/* Merge listpacks 'first' and 'second' by appending 'second' to 'first'. /* Merge listpacks 'first' and 'second' by appending 'second' to 'first'.
* *
* NOTE: The larger listpack is reallocated to contain the new merged listpack. * NOTE: The larger listpack is reallocated to contain the new merged listpack.
...@@ -1383,6 +1432,43 @@ void lpRandomPair(unsigned char *lp, unsigned long total_count, listpackEntry *k ...@@ -1383,6 +1432,43 @@ void lpRandomPair(unsigned char *lp, unsigned long total_count, listpackEntry *k
val->sval = lpGetValue(p, &(val->slen), &(val->lval)); val->sval = lpGetValue(p, &(val->slen), &(val->lval));
} }
/* Randomly select 'count' entries and store them in the 'entries' array, which
* needs to have space for 'count' listpackEntry structs. The order is random
* and duplicates are possible. */
void lpRandomEntries(unsigned char *lp, unsigned int count, listpackEntry *entries) {
struct pick {
unsigned int index;
unsigned int order;
} *picks = lp_malloc(count * sizeof(struct pick));
unsigned int total_size = lpLength(lp);
assert(total_size);
for (unsigned int i = 0; i < count; i++) {
picks[i].index = rand() % total_size;
picks[i].order = i;
}
/* Sort by index. */
qsort(picks, count, sizeof(struct pick), uintCompare);
/* Iterate over listpack in index order and store the values in the entries
* array respecting the original order. */
unsigned char *p = lpFirst(lp);
unsigned int j = 0; /* index in listpack */
for (unsigned int i = 0; i < count; i++) {
/* Advance listpack pointer to until we reach 'index' listpack. */
while (j < picks[i].index) {
p = lpNext(lp, p);
j++;
}
int storeorder = picks[i].order;
unsigned int len = 0;
long long llval = 0;
unsigned char *str = lpGetValue(p, &len, &llval);
lpSaveValue(str, len, llval, &entries[storeorder]);
}
lp_free(picks);
}
/* Randomly select count of key value pairs and store into 'keys' and /* Randomly select count of key value pairs and store into 'keys' and
* 'vals' args. The order of the picked entries is random, and the selections * 'vals' args. The order of the picked entries is random, and the selections
* are non-unique (repetitions are possible). * are non-unique (repetitions are possible).
...@@ -1449,34 +1535,83 @@ unsigned int lpRandomPairsUnique(unsigned char *lp, unsigned int count, listpack ...@@ -1449,34 +1535,83 @@ unsigned int lpRandomPairsUnique(unsigned char *lp, unsigned int count, listpack
if (count > total_size) if (count > total_size)
count = total_size; count = total_size;
/* To only iterate once, every time we try to pick a member, the probability
* we pick it is the quotient of the count left we want to pick and the
* count still we haven't visited in the dict, this way, we could make every
* member be equally picked.*/
p = lpFirst(lp); p = lpFirst(lp);
unsigned int picked = 0, remaining = count; unsigned int picked = 0, remaining = count;
while (picked < count && p) { while (picked < count && p) {
double randomDouble = ((double)rand()) / RAND_MAX; assert((p = lpNextRandom(lp, p, &index, remaining, 1)));
double threshold = ((double)remaining) / (total_size - index); key = lpGetValue(p, &klen, &klval);
if (randomDouble <= threshold) { lpSaveValue(key, klen, klval, &keys[picked]);
assert((p = lpNext(lp, p)));
index++;
if (vals) {
key = lpGetValue(p, &klen, &klval); key = lpGetValue(p, &klen, &klval);
lpSaveValue(key, klen, klval, &keys[picked]); lpSaveValue(key, klen, klval, &vals[picked]);
assert((p = lpNext(lp, p)));
if (vals) {
key = lpGetValue(p, &klen, &klval);
lpSaveValue(key, klen, klval, &vals[picked]);
}
remaining--;
picked++;
} else {
assert((p = lpNext(lp, p)));
} }
p = lpNext(lp, p); p = lpNext(lp, p);
remaining--;
picked++;
index++; index++;
} }
return picked; return picked;
} }
/* Iterates forward to the "next random" element, given we are yet to pick
* 'remaining' unique elements between the starting element 'p' (inclusive) and
* the end of the list. The 'index' needs to be initialized according to the
* current zero-based index matching the position of the starting element 'p'
* and is updated to match the returned element's zero-based index. If
* 'even_only' is nonzero, an element with an even index is picked, which is
* useful if the listpack represents a key-value pair sequence.
*
* Note that this function can return p. In order to skip the previously
* returned element, you need to call lpNext() or lpDelete() after each call to
* lpNextRandom(). Idea:
*
* assert(remaining <= lpLength(lp));
* p = lpFirst(lp);
* i = 0;
* while (remaining > 0) {
* p = lpNextRandom(lp, p, &i, remaining--, 0);
*
* // ... Do stuff with p ...
*
* p = lpNext(lp, p);
* i++;
* }
*/
unsigned char *lpNextRandom(unsigned char *lp, unsigned char *p, unsigned int *index,
unsigned int remaining, int even_only)
{
/* To only iterate once, every time we try to pick a member, the probability
* we pick it is the quotient of the count left we want to pick and the
* count still we haven't visited. This way, we could make every member be
* equally likely to be picked. */
unsigned int i = *index;
unsigned int total_size = lpLength(lp);
while (i < total_size && p != NULL) {
if (even_only && i % 2 != 0) {
p = lpNext(lp, p);
i++;
continue;
}
/* Do we pick this element? */
unsigned int available = total_size - i;
if (even_only) available /= 2;
double randomDouble = ((double)rand()) / RAND_MAX;
double threshold = ((double)remaining) / available;
if (randomDouble <= threshold) {
*index = i;
return p;
}
p = lpNext(lp, p);
i++;
}
return NULL;
}
/* Print info of listpack which is used in debugCommand */ /* Print info of listpack which is used in debugCommand */
void lpRepr(unsigned char *lp) { void lpRepr(unsigned char *lp) {
unsigned char *p, *vstr; unsigned char *p, *vstr;
...@@ -1902,6 +2037,21 @@ int listpackTest(int argc, char *argv[], int flags) { ...@@ -1902,6 +2037,21 @@ int listpackTest(int argc, char *argv[], int flags) {
zfree(lp); zfree(lp);
} }
TEST("Batch delete") {
unsigned char *lp = createList(); /* char *mixlist[] = {"hello", "foo", "quux", "1024"} */
assert(lpLength(lp) == 4); /* Pre-condition */
unsigned char *p0 = lpFirst(lp),
*p1 = lpNext(lp, p0),
*p2 = lpNext(lp, p1),
*p3 = lpNext(lp, p2);
unsigned char *ps[] = {p0, p1, p3};
lp = lpBatchDelete(lp, ps, 3);
assert(lpLength(lp) == 1);
verifyEntry(lpFirst(lp), (unsigned char*)mixlist[2], strlen(mixlist[2]));
assert(lpValidateIntegrity(lp, lpBytes(lp), 1, NULL, NULL) == 1);
lpFree(lp);
}
TEST("Delete foo while iterating") { TEST("Delete foo while iterating") {
lp = createList(); lp = createList();
p = lpFirst(lp); p = lpFirst(lp);
...@@ -2048,6 +2198,82 @@ int listpackTest(int argc, char *argv[], int flags) { ...@@ -2048,6 +2198,82 @@ int listpackTest(int argc, char *argv[], int flags) {
zfree(lp3); zfree(lp3);
} }
TEST("lpNextRandom normal usage") {
/* Create some data */
unsigned char *lp = lpNew(0);
unsigned char buf[100] = "asdf";
unsigned int size = 100;
for (size_t i = 0; i < size; i++) {
lp = lpAppend(lp, buf, i);
}
assert(lpLength(lp) == size);
/* Pick a subset of the elements of every possible subset size */
for (unsigned int count = 0; count <= size; count++) {
unsigned int remaining = count;
unsigned char *p = lpFirst(lp);
unsigned char *prev = NULL;
unsigned index = 0;
while (remaining > 0) {
assert(p != NULL);
p = lpNextRandom(lp, p, &index, remaining--, 0);
assert(p != NULL);
assert(p != prev);
prev = p;
p = lpNext(lp, p);
index++;
}
}
}
TEST("lpNextRandom corner cases") {
unsigned char *lp = lpNew(0);
unsigned i = 0;
/* Pick from empty listpack returns NULL. */
assert(lpNextRandom(lp, NULL, &i, 2, 0) == NULL);
/* Add some elements and find their pointers within the listpack. */
lp = lpAppend(lp, (unsigned char *)"abc", 3);
lp = lpAppend(lp, (unsigned char *)"def", 3);
lp = lpAppend(lp, (unsigned char *)"ghi", 3);
assert(lpLength(lp) == 3);
unsigned char *p0 = lpFirst(lp);
unsigned char *p1 = lpNext(lp, p0);
unsigned char *p2 = lpNext(lp, p1);
assert(lpNext(lp, p2) == NULL);
/* Pick zero elements returns NULL. */
i = 0; assert(lpNextRandom(lp, lpFirst(lp), &i, 0, 0) == NULL);
/* Pick all returns all. */
i = 0; assert(lpNextRandom(lp, p0, &i, 3, 0) == p0 && i == 0);
i = 1; assert(lpNextRandom(lp, p1, &i, 2, 0) == p1 && i == 1);
i = 2; assert(lpNextRandom(lp, p2, &i, 1, 0) == p2 && i == 2);
/* Pick more than one when there's only one left returns the last one. */
i = 2; assert(lpNextRandom(lp, p2, &i, 42, 0) == p2 && i == 2);
/* Pick all even elements returns p0 and p2. */
i = 0; assert(lpNextRandom(lp, p0, &i, 10, 1) == p0 && i == 0);
i = 1; assert(lpNextRandom(lp, p1, &i, 10, 1) == p2 && i == 2);
/* Don't crash even for bad index. */
for (int j = 0; j < 100; j++) {
unsigned char *p;
switch (j % 4) {
case 0: p = p0; break;
case 1: p = p1; break;
case 2: p = p2; break;
case 3: p = NULL; break;
}
i = j % 7;
unsigned int remaining = j % 5;
p = lpNextRandom(lp, p, &i, remaining, 0);
assert(p == p0 || p == p1 || p == p2 || p == NULL);
}
}
TEST("Random pair with one element") { TEST("Random pair with one element") {
listpackEntry key, val; listpackEntry key, val;
unsigned char *lp = lpNew(0); unsigned char *lp = lpNew(0);
......
...@@ -70,6 +70,7 @@ unsigned char *lpReplaceInteger(unsigned char *lp, unsigned char **p, long long ...@@ -70,6 +70,7 @@ unsigned char *lpReplaceInteger(unsigned char *lp, unsigned char **p, long long
unsigned char *lpDelete(unsigned char *lp, unsigned char *p, unsigned char **newp); unsigned char *lpDelete(unsigned char *lp, unsigned char *p, unsigned char **newp);
unsigned char *lpDeleteRangeWithEntry(unsigned char *lp, unsigned char **p, unsigned long num); unsigned char *lpDeleteRangeWithEntry(unsigned char *lp, unsigned char **p, unsigned long num);
unsigned char *lpDeleteRange(unsigned char *lp, long index, unsigned long num); unsigned char *lpDeleteRange(unsigned char *lp, long index, unsigned long num);
unsigned char *lpBatchDelete(unsigned char *lp, unsigned char **ps, unsigned long count);
unsigned char *lpMerge(unsigned char **first, unsigned char **second); unsigned char *lpMerge(unsigned char **first, unsigned char **second);
unsigned long lpLength(unsigned char *lp); unsigned long lpLength(unsigned char *lp);
unsigned char *lpGet(unsigned char *p, int64_t *count, unsigned char *intbuf); unsigned char *lpGet(unsigned char *p, int64_t *count, unsigned char *intbuf);
...@@ -90,6 +91,9 @@ unsigned int lpCompare(unsigned char *p, unsigned char *s, uint32_t slen); ...@@ -90,6 +91,9 @@ unsigned int lpCompare(unsigned char *p, unsigned char *s, uint32_t slen);
void lpRandomPair(unsigned char *lp, unsigned long total_count, listpackEntry *key, listpackEntry *val); void lpRandomPair(unsigned char *lp, unsigned long total_count, listpackEntry *key, listpackEntry *val);
void lpRandomPairs(unsigned char *lp, unsigned int count, listpackEntry *keys, listpackEntry *vals); void lpRandomPairs(unsigned char *lp, unsigned int count, listpackEntry *keys, listpackEntry *vals);
unsigned int lpRandomPairsUnique(unsigned char *lp, unsigned int count, listpackEntry *keys, listpackEntry *vals); unsigned int lpRandomPairsUnique(unsigned char *lp, unsigned int count, listpackEntry *keys, listpackEntry *vals);
void lpRandomEntries(unsigned char *lp, unsigned int count, listpackEntry *entries);
unsigned char *lpNextRandom(unsigned char *lp, unsigned char *p, unsigned int *index,
unsigned int remaining, int even_only);
int lpSafeToAdd(unsigned char* lp, size_t add); int lpSafeToAdd(unsigned char* lp, size_t add);
void lpRepr(unsigned char *lp); void lpRepr(unsigned char *lp);
......
...@@ -247,6 +247,13 @@ robj *createIntsetObject(void) { ...@@ -247,6 +247,13 @@ robj *createIntsetObject(void) {
return o; return o;
} }
robj *createSetListpackObject(void) {
unsigned char *lp = lpNew(0);
robj *o = createObject(OBJ_SET, lp);
o->encoding = OBJ_ENCODING_LISTPACK;
return o;
}
robj *createHashObject(void) { robj *createHashObject(void) {
unsigned char *zl = lpNew(0); unsigned char *zl = lpNew(0);
robj *o = createObject(OBJ_HASH, zl); robj *o = createObject(OBJ_HASH, zl);
...@@ -306,6 +313,7 @@ void freeSetObject(robj *o) { ...@@ -306,6 +313,7 @@ void freeSetObject(robj *o) {
dictRelease((dict*) o->ptr); dictRelease((dict*) o->ptr);
break; break;
case OBJ_ENCODING_INTSET: case OBJ_ENCODING_INTSET:
case OBJ_ENCODING_LISTPACK:
zfree(o->ptr); zfree(o->ptr);
break; break;
default: default:
...@@ -441,6 +449,8 @@ void dismissSetObject(robj *o, size_t size_hint) { ...@@ -441,6 +449,8 @@ void dismissSetObject(robj *o, size_t size_hint) {
dismissMemory(set->ht_table[1], DICTHT_SIZE(set->ht_size_exp[1])*sizeof(dictEntry*)); dismissMemory(set->ht_table[1], DICTHT_SIZE(set->ht_size_exp[1])*sizeof(dictEntry*));
} else if (o->encoding == OBJ_ENCODING_INTSET) { } else if (o->encoding == OBJ_ENCODING_INTSET) {
dismissMemory(o->ptr, intsetBlobLen((intset*)o->ptr)); dismissMemory(o->ptr, intsetBlobLen((intset*)o->ptr));
} else if (o->encoding == OBJ_ENCODING_LISTPACK) {
dismissMemory(o->ptr, lpBytes((unsigned char *)o->ptr));
} else { } else {
serverPanic("Unknown set encoding type"); serverPanic("Unknown set encoding type");
} }
......
...@@ -665,6 +665,8 @@ int rdbSaveObjectType(rio *rdb, robj *o) { ...@@ -665,6 +665,8 @@ int rdbSaveObjectType(rio *rdb, robj *o) {
return rdbSaveType(rdb,RDB_TYPE_SET_INTSET); return rdbSaveType(rdb,RDB_TYPE_SET_INTSET);
else if (o->encoding == OBJ_ENCODING_HT) else if (o->encoding == OBJ_ENCODING_HT)
return rdbSaveType(rdb,RDB_TYPE_SET); return rdbSaveType(rdb,RDB_TYPE_SET);
else if (o->encoding == OBJ_ENCODING_LISTPACK)
return rdbSaveType(rdb,RDB_TYPE_SET_LISTPACK);
else else
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
case OBJ_ZSET: case OBJ_ZSET:
...@@ -858,6 +860,10 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) { ...@@ -858,6 +860,10 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) {
if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1; if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
nwritten += n; nwritten += n;
} else if (o->encoding == OBJ_ENCODING_LISTPACK) {
size_t l = lpBytes((unsigned char *)o->ptr);
if ((n = rdbSaveRawString(rdb, o->ptr, l)) == -1) return -1;
nwritten += n;
} else { } else {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
} }
...@@ -1690,19 +1696,21 @@ static int _listZiplistEntryConvertAndValidate(unsigned char *p, unsigned int he ...@@ -1690,19 +1696,21 @@ static int _listZiplistEntryConvertAndValidate(unsigned char *p, unsigned int he
} }
/* callback for to check the listpack doesn't have duplicate records */ /* callback for to check the listpack doesn't have duplicate records */
static int _lpPairsEntryValidation(unsigned char *p, unsigned int head_count, void *userdata) { static int _lpEntryValidation(unsigned char *p, unsigned int head_count, void *userdata) {
struct { struct {
int pairs;
long count; long count;
dict *fields; dict *fields;
} *data = userdata; } *data = userdata;
if (data->fields == NULL) { if (data->fields == NULL) {
data->fields = dictCreate(&hashDictType); data->fields = dictCreate(&hashDictType);
dictExpand(data->fields, head_count/2); dictExpand(data->fields, data->pairs ? head_count/2 : head_count);
} }
/* Even records are field names, add to dict and check that's not a dup */ /* If we're checking pairs, then even records are field names. Otherwise
if (((data->count) & 1) == 0) { * we're checking all elements. Add to dict and check that's not a dup */
if (!data->pairs || ((data->count) & 1) == 0) {
unsigned char *str; unsigned char *str;
int64_t slen; int64_t slen;
unsigned char buf[LP_INTBUF_SIZE]; unsigned char buf[LP_INTBUF_SIZE];
...@@ -1722,21 +1730,24 @@ static int _lpPairsEntryValidation(unsigned char *p, unsigned int head_count, vo ...@@ -1722,21 +1730,24 @@ static int _lpPairsEntryValidation(unsigned char *p, unsigned int head_count, vo
/* Validate the integrity of the listpack structure. /* Validate the integrity of the listpack structure.
* when `deep` is 0, only the integrity of the header is validated. * when `deep` is 0, only the integrity of the header is validated.
* when `deep` is 1, we scan all the entries one by one. */ * when `deep` is 1, we scan all the entries one by one.
int lpPairsValidateIntegrityAndDups(unsigned char *lp, size_t size, int deep) { * when `pairs` is 0, all elements need to be unique (it's a set)
* when `pairs` is 1, odd elements need to be unique (it's a key-value map) */
int lpValidateIntegrityAndDups(unsigned char *lp, size_t size, int deep, int pairs) {
if (!deep) if (!deep)
return lpValidateIntegrity(lp, size, 0, NULL, NULL); return lpValidateIntegrity(lp, size, 0, NULL, NULL);
/* Keep track of the field names to locate duplicate ones */ /* Keep track of the field names to locate duplicate ones */
struct { struct {
int pairs;
long count; long count;
dict *fields; /* Initialisation at the first callback. */ dict *fields; /* Initialisation at the first callback. */
} data = {0, NULL}; } data = {pairs, 0, NULL};
int ret = lpValidateIntegrity(lp, size, 1, _lpPairsEntryValidation, &data); int ret = lpValidateIntegrity(lp, size, 1, _lpEntryValidation, &data);
/* make sure we have an even number of records. */ /* make sure we have an even number of records. */
if (data.count & 1) if (pairs && data.count & 1)
ret = 0; ret = 0;
if (data.fields) dictRelease(data.fields); if (data.fields) dictRelease(data.fields);
...@@ -1813,6 +1824,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -1813,6 +1824,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
} }
/* Load every single element of the set */ /* Load every single element of the set */
size_t maxelelen = 0, sumelelen = 0;
for (i = 0; i < len; i++) { for (i = 0; i < len; i++) {
long long llval; long long llval;
sds sdsele; sds sdsele;
...@@ -1821,6 +1833,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -1821,6 +1833,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
decrRefCount(o); decrRefCount(o);
return NULL; return NULL;
} }
size_t elelen = sdslen(sdsele);
sumelelen += elelen;
if (elelen > maxelelen) maxelelen = elelen;
if (o->encoding == OBJ_ENCODING_INTSET) { if (o->encoding == OBJ_ENCODING_INTSET) {
/* Fetch integer value from element. */ /* Fetch integer value from element. */
...@@ -1833,6 +1848,14 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -1833,6 +1848,14 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
sdsfree(sdsele); sdsfree(sdsele);
return NULL; return NULL;
} }
} else if (setTypeSize(o) < server.set_max_listpack_entries &&
maxelelen <= server.set_max_listpack_value &&
lpSafeToAdd(NULL, sumelelen))
{
/* We checked if it's safe to add one large element instead
* of many small ones. It's OK since lpSafeToAdd doesn't
* care about individual elements, only the total size. */
setTypeConvert(o, OBJ_ENCODING_LISTPACK);
} else { } else {
setTypeConvert(o,OBJ_ENCODING_HT); setTypeConvert(o,OBJ_ENCODING_HT);
if (dictTryExpand(o->ptr,len) != DICT_OK) { if (dictTryExpand(o->ptr,len) != DICT_OK) {
...@@ -1844,6 +1867,33 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -1844,6 +1867,33 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
} }
} }
/* This will also be called when the set was just converted
* to a listpack encoded set. */
if (o->encoding == OBJ_ENCODING_LISTPACK) {
if (setTypeSize(o) < server.set_max_listpack_entries &&
elelen <= server.set_max_listpack_value &&
lpSafeToAdd(o->ptr, elelen))
{
unsigned char *p = lpFirst(o->ptr);
if (p && lpFind(o->ptr, p, (unsigned char*)sdsele, elelen, 0)) {
rdbReportCorruptRDB("Duplicate set members detected");
decrRefCount(o);
sdsfree(sdsele);
return NULL;
}
o->ptr = lpAppend(o->ptr, (unsigned char *)sdsele, elelen);
} else {
setTypeConvert(o, OBJ_ENCODING_HT);
if (dictTryExpand(o->ptr, len) != DICT_OK) {
rdbReportCorruptRDB("OOM in dictTryExpand %llu",
(unsigned long long)len);
sdsfree(sdsele);
decrRefCount(o);
return NULL;
}
}
}
/* 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) {
...@@ -2126,6 +2176,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -2126,6 +2176,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
} else if (rdbtype == RDB_TYPE_HASH_ZIPMAP || } else if (rdbtype == RDB_TYPE_HASH_ZIPMAP ||
rdbtype == RDB_TYPE_LIST_ZIPLIST || rdbtype == RDB_TYPE_LIST_ZIPLIST ||
rdbtype == RDB_TYPE_SET_INTSET || rdbtype == RDB_TYPE_SET_INTSET ||
rdbtype == RDB_TYPE_SET_LISTPACK ||
rdbtype == RDB_TYPE_ZSET_ZIPLIST || rdbtype == RDB_TYPE_ZSET_ZIPLIST ||
rdbtype == RDB_TYPE_ZSET_LISTPACK || rdbtype == RDB_TYPE_ZSET_LISTPACK ||
rdbtype == RDB_TYPE_HASH_ZIPLIST || rdbtype == RDB_TYPE_HASH_ZIPLIST ||
...@@ -2243,6 +2294,20 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -2243,6 +2294,20 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
if (intsetLen(o->ptr) > server.set_max_intset_entries) if (intsetLen(o->ptr) > server.set_max_intset_entries)
setTypeConvert(o,OBJ_ENCODING_HT); setTypeConvert(o,OBJ_ENCODING_HT);
break; break;
case RDB_TYPE_SET_LISTPACK:
if (deep_integrity_validation) server.stat_dump_payload_sanitizations++;
if (!lpValidateIntegrityAndDups(encoded, encoded_len, deep_integrity_validation, 0)) {
rdbReportCorruptRDB("Set listpack integrity check failed.");
zfree(encoded);
o->ptr = NULL;
decrRefCount(o);
return NULL;
}
o->type = OBJ_SET;
o->encoding = OBJ_ENCODING_LISTPACK;
if (setTypeSize(o) > server.set_max_listpack_entries)
setTypeConvert(o, OBJ_ENCODING_HT);
break;
case RDB_TYPE_ZSET_ZIPLIST: case RDB_TYPE_ZSET_ZIPLIST:
{ {
unsigned char *lp = lpNew(encoded_len); unsigned char *lp = lpNew(encoded_len);
...@@ -2272,7 +2337,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -2272,7 +2337,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
} }
case RDB_TYPE_ZSET_LISTPACK: case RDB_TYPE_ZSET_LISTPACK:
if (deep_integrity_validation) server.stat_dump_payload_sanitizations++; if (deep_integrity_validation) server.stat_dump_payload_sanitizations++;
if (!lpPairsValidateIntegrityAndDups(encoded, encoded_len, deep_integrity_validation)) { if (!lpValidateIntegrityAndDups(encoded, encoded_len, deep_integrity_validation, 1)) {
rdbReportCorruptRDB("Zset listpack integrity check failed."); rdbReportCorruptRDB("Zset listpack integrity check failed.");
zfree(encoded); zfree(encoded);
o->ptr = NULL; o->ptr = NULL;
...@@ -2318,7 +2383,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -2318,7 +2383,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
} }
case RDB_TYPE_HASH_LISTPACK: case RDB_TYPE_HASH_LISTPACK:
if (deep_integrity_validation) server.stat_dump_payload_sanitizations++; if (deep_integrity_validation) server.stat_dump_payload_sanitizations++;
if (!lpPairsValidateIntegrityAndDups(encoded, encoded_len, deep_integrity_validation)) { if (!lpValidateIntegrityAndDups(encoded, encoded_len, deep_integrity_validation, 1)) {
rdbReportCorruptRDB("Hash listpack integrity check failed."); rdbReportCorruptRDB("Hash listpack integrity check failed.");
zfree(encoded); zfree(encoded);
o->ptr = NULL; o->ptr = NULL;
......
...@@ -38,7 +38,7 @@ ...@@ -38,7 +38,7 @@
/* The current RDB version. When the format changes in a way that is no longer /* The current RDB version. When the format changes in a way that is no longer
* backward compatible this number gets incremented. */ * backward compatible this number gets incremented. */
#define RDB_VERSION 10 #define RDB_VERSION 11
/* Defines related to the dump file format. To store 32 bits lengths for short /* Defines related to the dump file format. To store 32 bits lengths for short
* keys requires a lot of space, so we check the most significant 2 bits of * keys requires a lot of space, so we check the most significant 2 bits of
...@@ -95,10 +95,11 @@ ...@@ -95,10 +95,11 @@
#define RDB_TYPE_ZSET_LISTPACK 17 #define RDB_TYPE_ZSET_LISTPACK 17
#define RDB_TYPE_LIST_QUICKLIST_2 18 #define RDB_TYPE_LIST_QUICKLIST_2 18
#define RDB_TYPE_STREAM_LISTPACKS_2 19 #define RDB_TYPE_STREAM_LISTPACKS_2 19
#define RDB_TYPE_SET_LISTPACK 20
/* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */ /* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */
/* Test if a type is an object type. */ /* Test if a type is an object type. */
#define rdbIsObjectType(t) (((t) >= 0 && (t) <= 7) || ((t) >= 9 && (t) <= 19)) #define rdbIsObjectType(t) (((t) >= 0 && (t) <= 7) || ((t) >= 9 && (t) <= 20))
/* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */ /* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
#define RDB_OPCODE_FUNCTION2 245 /* function library data */ #define RDB_OPCODE_FUNCTION2 245 /* function library data */
......
...@@ -97,7 +97,8 @@ char *rdb_type_string[] = { ...@@ -97,7 +97,8 @@ char *rdb_type_string[] = {
"stream", "stream",
"hash-listpack", "hash-listpack",
"zset-listpack", "zset-listpack",
"quicklist-v2" "quicklist-v2",
"set-listpack",
}; };
/* Show a few stats collected into 'rdbstate' */ /* Show a few stats collected into 'rdbstate' */
......
...@@ -1850,6 +1850,8 @@ struct redisServer { ...@@ -1850,6 +1850,8 @@ struct redisServer {
size_t hash_max_listpack_entries; size_t hash_max_listpack_entries;
size_t hash_max_listpack_value; size_t hash_max_listpack_value;
size_t set_max_intset_entries; size_t set_max_intset_entries;
size_t set_max_listpack_entries;
size_t set_max_listpack_value;
size_t zset_max_listpack_entries; size_t zset_max_listpack_entries;
size_t zset_max_listpack_value; size_t zset_max_listpack_value;
size_t hll_sparse_max_bytes; size_t hll_sparse_max_bytes;
...@@ -2331,6 +2333,7 @@ typedef struct { ...@@ -2331,6 +2333,7 @@ typedef struct {
int encoding; int encoding;
int ii; /* intset iterator */ int ii; /* intset iterator */
dictIterator *di; dictIterator *di;
unsigned char *lpi; /* listpack iterator */
} setTypeIterator; } setTypeIterator;
/* Structure to hold hash iteration abstraction. Note that iteration over /* Structure to hold hash iteration abstraction. Note that iteration over
...@@ -2655,6 +2658,7 @@ robj *createStringObjectFromLongDouble(long double value, int humanfriendly); ...@@ -2655,6 +2658,7 @@ robj *createStringObjectFromLongDouble(long double value, int humanfriendly);
robj *createQuicklistObject(void); robj *createQuicklistObject(void);
robj *createSetObject(void); robj *createSetObject(void);
robj *createIntsetObject(void); robj *createIntsetObject(void);
robj *createSetListpackObject(void);
robj *createHashObject(void); robj *createHashObject(void);
robj *createZsetObject(void); robj *createZsetObject(void);
robj *createZsetListpackObject(void); robj *createZsetListpackObject(void);
...@@ -2980,9 +2984,9 @@ int setTypeRemove(robj *subject, sds value); ...@@ -2980,9 +2984,9 @@ int setTypeRemove(robj *subject, sds value);
int setTypeIsMember(robj *subject, sds 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, sds *sdsele, int64_t *llele); int setTypeNext(setTypeIterator *si, char **str, size_t *len, int64_t *llele);
sds setTypeNextObject(setTypeIterator *si); sds setTypeNextObject(setTypeIterator *si);
int setTypeRandomElement(robj *setobj, sds *sdsele, int64_t *llele); int setTypeRandomElement(robj *setobj, char **str, size_t *len, 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(const robj *subject); unsigned long setTypeSize(const robj *subject);
void setTypeConvert(robj *subject, int enc); void setTypeConvert(robj *subject, int enc);
......
...@@ -29,6 +29,12 @@ ...@@ -29,6 +29,12 @@
#include "server.h" #include "server.h"
/* Internal prototypes */
int setTypeAddAux(robj *set, char *str, size_t len, int64_t llval, int str_is_sds);
int setTypeRemoveAux(robj *set, char *str, size_t len, int64_t llval, int str_is_sds);
int setTypeIsMemberAux(robj *set, char *str, size_t len, int64_t llval, int str_is_sds);
/*----------------------------------------------------------------------------- /*-----------------------------------------------------------------------------
* Set Commands * Set Commands
*----------------------------------------------------------------------------*/ *----------------------------------------------------------------------------*/
...@@ -42,45 +48,154 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum, ...@@ -42,45 +48,154 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
robj *setTypeCreate(sds value) { robj *setTypeCreate(sds value) {
if (isSdsRepresentableAsLongLong(value,NULL) == C_OK) if (isSdsRepresentableAsLongLong(value,NULL) == C_OK)
return createIntsetObject(); return createIntsetObject();
return createSetObject(); return createSetListpackObject();
}
/* Return the maximum number of entries to store in an intset. */
static size_t intsetMaxEntries(void) {
size_t max_entries = server.set_max_intset_entries;
/* limit to 1G entries due to intset internals. */
if (max_entries >= 1<<30) max_entries = 1<<30;
return max_entries;
} }
/* Add the specified value into a set. /* Converts intset to HT if it contains too many entries. */
static void maybeConvertIntset(robj *subject) {
serverAssert(subject->encoding == OBJ_ENCODING_INTSET);
if (intsetLen(subject->ptr) > intsetMaxEntries())
setTypeConvert(subject,OBJ_ENCODING_HT);
}
/* When you know all set elements are integers, call this to convert the set to
* an intset. No conversion happens if the set contains too many entries for an
* intset. */
static void maybeConvertToIntset(robj *set) {
if (set->encoding == OBJ_ENCODING_INTSET) return; /* already intset */
if (setTypeSize(set) > intsetMaxEntries()) return; /* can't use intset */
intset *is = intsetNew();
char *str;
size_t len;
int64_t llval;
setTypeIterator *si = setTypeInitIterator(set);
while (setTypeNext(si, &str, &len, &llval) != -1) {
if (str) {
/* If the element is returned as a string, we may be able to convert
* it to integer. This happens for OBJ_ENCODING_HT. */
serverAssert(string2ll(str, len, (long long *)&llval));
}
uint8_t success = 0;
is = intsetAdd(is, llval, &success);
serverAssert(success);
}
setTypeReleaseIterator(si);
freeSetObject(set); /* frees the internals but not robj itself */
set->ptr = is;
set->encoding = OBJ_ENCODING_INTSET;
}
/* Add the specified sds value into a set.
* *
* 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, sds value) { int setTypeAdd(robj *subject, sds value) {
long long llval; return setTypeAddAux(subject, value, sdslen(value), 0, 1);
if (subject->encoding == OBJ_ENCODING_HT) { }
dict *ht = subject->ptr;
dictEntry *de = dictAddRaw(ht,value,NULL); /* Add member. This function is optimized for the different encodings. The
if (de) { * value can be provided as an sds string (indicated by passing str_is_sds =
dictSetKey(ht,de,sdsdup(value)); * 1), as string and length (str_is_sds = 0) or as an integer in which case str
* is set to NULL and llval is provided instead.
*
* Returns 1 if the value was added and 0 if it was already a member. */
int setTypeAddAux(robj *set, char *str, size_t len, int64_t llval, int str_is_sds) {
char tmpbuf[LONG_STR_SIZE];
if (!str) {
if (set->encoding == OBJ_ENCODING_INTSET) {
uint8_t success = 0;
set->ptr = intsetAdd(set->ptr, llval, &success);
if (success) maybeConvertIntset(set);
return success;
}
/* Convert int to string. */
len = ll2string(tmpbuf, sizeof tmpbuf, llval);
str = tmpbuf;
str_is_sds = 0;
}
serverAssert(str);
if (set->encoding == OBJ_ENCODING_HT) {
/* Avoid duping the string if it is an sds string. */
sds sdsval = str_is_sds ? (sds)str : sdsnewlen(str, len);
dict *ht = set->ptr;
dictEntry *de = dictAddRaw(ht,sdsval,NULL);
if (de && sdsval == str) {
/* String was added but we don't own this sds string. Dup it and
* replace it in the dict entry. */
dictSetKey(ht,de,sdsdup((sds)str));
dictSetVal(ht,de,NULL); dictSetVal(ht,de,NULL);
} else if (!de && sdsval != str) {
/* String was already a member. Free our temporary sds copy. */
sdsfree(sdsval);
}
return (de != NULL);
} else if (set->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *lp = set->ptr;
unsigned char *p = lpFirst(lp);
if (p != NULL)
p = lpFind(lp, p, (unsigned char*)str, len, 0);
if (p == NULL) {
/* Not found. */
if (lpLength(lp) < server.set_max_listpack_entries &&
len <= server.set_max_listpack_value &&
lpSafeToAdd(lp, len))
{
if (str == tmpbuf) {
/* This came in as integer so we can avoid parsing it again.
* TODO: Create and use lpFindInteger; don't go via string. */
lp = lpAppendInteger(lp, llval);
} else {
lp = lpAppend(lp, (unsigned char*)str, len);
}
set->ptr = lp;
} else {
/* Size limit is reached. Convert to hashtable and add. */
setTypeConvert(set, OBJ_ENCODING_HT);
serverAssert(dictAdd(set->ptr,sdsnewlen(str,len),NULL) == DICT_OK);
}
return 1; return 1;
} }
} else if (subject->encoding == OBJ_ENCODING_INTSET) { } else if (set->encoding == OBJ_ENCODING_INTSET) {
if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) { long long value;
if (string2ll(str, len, &value)) {
uint8_t success = 0; uint8_t success = 0;
subject->ptr = intsetAdd(subject->ptr,llval,&success); set->ptr = intsetAdd(set->ptr,value,&success);
if (success) { if (success) {
/* Convert to regular set when the intset contains maybeConvertIntset(set);
* too many entries. */
size_t max_entries = server.set_max_intset_entries;
/* limit to 1G entries due to intset internals. */
if (max_entries >= 1<<30) max_entries = 1<<30;
if (intsetLen(subject->ptr) > max_entries)
setTypeConvert(subject,OBJ_ENCODING_HT);
return 1; return 1;
} }
} else { } else {
/* Failed to get integer from object, convert to regular set. */ size_t maxelelen = intsetLen(set->ptr) == 0 ?
setTypeConvert(subject,OBJ_ENCODING_HT); 0 : max(sdigits10(intsetMax(set->ptr)),
sdigits10(intsetMin(set->ptr)));
/* The set *was* an intset and this value is not integer if (intsetLen((const intset*)set->ptr) < server.set_max_listpack_entries &&
* encodable, so dictAdd should always work. */ len <= server.set_max_listpack_value &&
serverAssert(dictAdd(subject->ptr,sdsdup(value),NULL) == DICT_OK); maxelelen <= server.set_max_listpack_value &&
return 1; lpSafeToAdd(NULL, maxelelen * intsetLen(set->ptr) + len))
{
/* In the "safe to add" check above we assumed all elements in
* the intset are of size maxelelen. This is an upper bound. */
setTypeConvert(set, OBJ_ENCODING_LISTPACK);
unsigned char *lp = set->ptr;
lp = lpAppend(lp, (unsigned char *)str, len);
set->ptr = lp;
return 1;
} else {
setTypeConvert(set, OBJ_ENCODING_HT);
/* The set *was* an intset and this value is not integer
* encodable, so dictAdd should always work. */
serverAssert(dictAdd(set->ptr,sdsnewlen(str,len),NULL) == DICT_OK);
return 1;
}
} }
} else { } else {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
...@@ -88,15 +203,50 @@ int setTypeAdd(robj *subject, sds value) { ...@@ -88,15 +203,50 @@ int setTypeAdd(robj *subject, sds value) {
return 0; return 0;
} }
/* Deletes a value provided as an sds string from the set. Returns 1 if the
* value was deleted and 0 if it was not a member of the set. */
int setTypeRemove(robj *setobj, sds value) { int setTypeRemove(robj *setobj, sds value) {
long long llval; return setTypeRemoveAux(setobj, value, sdslen(value), 0, 1);
}
/* Remove a member. This function is optimized for the different encodings. The
* value can be provided as an sds string (indicated by passing str_is_sds =
* 1), as string and length (str_is_sds = 0) or as an integer in which case str
* is set to NULL and llval is provided instead.
*
* Returns 1 if the value was deleted and 0 if it was not a member of the set. */
int setTypeRemoveAux(robj *setobj, char *str, size_t len, int64_t llval, int str_is_sds) {
char tmpbuf[LONG_STR_SIZE];
if (!str) {
if (setobj->encoding == OBJ_ENCODING_INTSET) {
int success;
setobj->ptr = intsetRemove(setobj->ptr,llval,&success);
return success;
}
len = ll2string(tmpbuf, sizeof tmpbuf, llval);
str = tmpbuf;
str_is_sds = 0;
}
if (setobj->encoding == OBJ_ENCODING_HT) { if (setobj->encoding == OBJ_ENCODING_HT) {
if (dictDelete(setobj->ptr,value) == DICT_OK) { sds sdsval = str_is_sds ? (sds)str : sdsnewlen(str, len);
if (htNeedsResize(setobj->ptr)) dictResize(setobj->ptr); int deleted = (dictDelete(setobj->ptr, sdsval) == DICT_OK);
if (deleted && htNeedsResize(setobj->ptr)) dictResize(setobj->ptr);
if (sdsval != str) sdsfree(sdsval); /* free temp copy */
return deleted;
} else if (setobj->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *lp = setobj->ptr;
unsigned char *p = lpFirst(lp);
if (p == NULL) return 0;
p = lpFind(lp, p, (unsigned char*)str, len, 0);
if (p != NULL) {
lp = lpDelete(lp, p, NULL);
setobj->ptr = lp;
return 1; return 1;
} }
} else if (setobj->encoding == OBJ_ENCODING_INTSET) { } else if (setobj->encoding == OBJ_ENCODING_INTSET) {
if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) { long long llval;
if (string2ll(str, len, &llval)) {
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;
...@@ -107,18 +257,45 @@ int setTypeRemove(robj *setobj, sds value) { ...@@ -107,18 +257,45 @@ int setTypeRemove(robj *setobj, sds value) {
return 0; return 0;
} }
/* Check if an sds string is a member of the set. Returns 1 if the value is a
* member of the set and 0 if it isn't. */
int setTypeIsMember(robj *subject, sds value) { int setTypeIsMember(robj *subject, sds value) {
long long llval; return setTypeIsMemberAux(subject, value, sdslen(value), 0, 1);
if (subject->encoding == OBJ_ENCODING_HT) { }
return dictFind((dict*)subject->ptr,value) != NULL;
} else if (subject->encoding == OBJ_ENCODING_INTSET) { /* Membership checking optimized for the different encodings. The value can be
if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) { * provided as an sds string (indicated by passing str_is_sds = 1), as string
return intsetFind((intset*)subject->ptr,llval); * and length (str_is_sds = 0) or as an integer in which case str is set to NULL
} * and llval is provided instead.
*
* Returns 1 if the value is a member of the set and 0 if it isn't. */
int setTypeIsMemberAux(robj *set, char *str, size_t len, int64_t llval, int str_is_sds) {
char tmpbuf[LONG_STR_SIZE];
if (!str) {
if (set->encoding == OBJ_ENCODING_INTSET)
return intsetFind(set->ptr, llval);
len = ll2string(tmpbuf, sizeof tmpbuf, llval);
str = tmpbuf;
str_is_sds = 0;
}
if (set->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *lp = set->ptr;
unsigned char *p = lpFirst(lp);
return p && lpFind(lp, p, (unsigned char*)str, len, 0);
} else if (set->encoding == OBJ_ENCODING_INTSET) {
long long llval;
return string2ll(str, len, &llval) && intsetFind(set->ptr, llval);
} else if (set->encoding == OBJ_ENCODING_HT && str_is_sds) {
return dictFind(set->ptr, (sds)str) != NULL;
} else if (set->encoding == OBJ_ENCODING_HT) {
sds sdsval = sdsnewlen(str, len);
int result = dictFind(set->ptr, sdsval) != NULL;
sdsfree(sdsval);
return result;
} else { } else {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
} }
return 0;
} }
setTypeIterator *setTypeInitIterator(robj *subject) { setTypeIterator *setTypeInitIterator(robj *subject) {
...@@ -129,6 +306,8 @@ setTypeIterator *setTypeInitIterator(robj *subject) { ...@@ -129,6 +306,8 @@ setTypeIterator *setTypeInitIterator(robj *subject) {
si->di = dictGetIterator(subject->ptr); si->di = dictGetIterator(subject->ptr);
} else if (si->encoding == OBJ_ENCODING_INTSET) { } else if (si->encoding == OBJ_ENCODING_INTSET) {
si->ii = 0; si->ii = 0;
} else if (si->encoding == OBJ_ENCODING_LISTPACK) {
si->lpi = NULL;
} else { } else {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
} }
...@@ -142,28 +321,50 @@ void setTypeReleaseIterator(setTypeIterator *si) { ...@@ -142,28 +321,50 @@ 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, as a string or as an integer.
* *
* Since set elements can be internally be stored as SDS strings or * Since set elements can be internally be stored as SDS strings, char buffers 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 pointers
* (sdsele) or (llele) accordingly. * (str and len) or (llele) depending on whether the value is stored as a string
* or as an integer internally.
*
* If OBJ_ENCODING_HT is returned, then str points to an sds string and can be
* used as such. If OBJ_ENCODING_INTSET, then llele is populated and str is
* pointed to NULL. If OBJ_ENCODING_LISTPACK is returned, the value can be
* either a string or an integer. If *str is not NULL, then str and len are
* populated with the string content and length. Otherwise, llele populated with
* an integer value.
* *
* Note that both the sdsele and llele pointers should be passed and cannot * Note that str, len and llele pointers should all 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 more elements -1 is returned. */
int setTypeNext(setTypeIterator *si, sds *sdsele, int64_t *llele) { int setTypeNext(setTypeIterator *si, char **str, size_t *len, 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;
*sdsele = dictGetKey(de); *str = dictGetKey(de);
*len = sdslen(*str);
*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;
*sdsele = NULL; /* Not needed. Defensive. */ *str = NULL;
} else if (si->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *lp = si->subject->ptr;
unsigned char *lpi = si->lpi;
if (lpi == NULL) {
lpi = lpFirst(lp);
} else {
lpi = lpNext(lp, lpi);
}
if (lpi == NULL) return -1;
si->lpi = lpi;
unsigned int l;
*str = (char *)lpGetValue(lpi, &l, (long long *)llele);
*len = (size_t)l;
} else { } else {
serverPanic("Wrong set encoding in setTypeNext"); serverPanic("Wrong set encoding in setTypeNext");
} }
...@@ -179,54 +380,85 @@ int setTypeNext(setTypeIterator *si, sds *sdsele, int64_t *llele) { ...@@ -179,54 +380,85 @@ int setTypeNext(setTypeIterator *si, sds *sdsele, int64_t *llele) {
* an issue. */ * an issue. */
sds setTypeNextObject(setTypeIterator *si) { sds setTypeNextObject(setTypeIterator *si) {
int64_t intele; int64_t intele;
sds sdsele; char *str;
int encoding; size_t len;
encoding = setTypeNext(si,&sdsele,&intele); if (setTypeNext(si, &str, &len, &intele) == -1) return NULL;
switch(encoding) { if (str != NULL) return sdsnewlen(str, len);
case -1: return NULL; return sdsfromlonglong(intele);
case OBJ_ENCODING_INTSET:
return sdsfromlonglong(intele);
case OBJ_ENCODING_HT:
return sdsdup(sdsele);
default:
serverPanic("Unsupported encoding");
}
return NULL; /* just to suppress warnings */
} }
/* Return random element from a non empty set. /* Return random element from a non empty set.
* The returned element can be an int64_t value if the set is encoded * The returned element can be an int64_t value if the set is encoded
* as an "intset" blob of integers, or an SDS string if the set * as an "intset" blob of integers, or an string.
* is a regular set.
* *
* The caller provides both pointers to be populated with the right * The caller provides three pointers to be populated with the right
* object. The return value of the function is the object->encoding * object. The return value of the function is the object->encoding
* field of the object and is used by the caller to check if the * field of the object and can be used by the caller to check if the
* int64_t pointer or the sds pointer was populated. * int64_t pointer or the str and len pointers were populated, as for
* setTypeNext. If OBJ_ENCODING_HT is returned, str is pointed to a
* string which is actually an sds string and it can be used as such.
* *
* Note that both the sdsele and llele pointers should be passed and cannot * Note that both the str, len and llele pointers should be passed and cannot
* be NULL since the function will try to defensively populate the non * be NULL. If str is set to NULL, the value is an integer stored in llele. */
* used field with values which are easy to trap if misused. */ int setTypeRandomElement(robj *setobj, char **str, size_t *len, int64_t *llele) {
int setTypeRandomElement(robj *setobj, sds *sdsele, int64_t *llele) {
if (setobj->encoding == OBJ_ENCODING_HT) { if (setobj->encoding == OBJ_ENCODING_HT) {
dictEntry *de = dictGetFairRandomKey(setobj->ptr); dictEntry *de = dictGetFairRandomKey(setobj->ptr);
*sdsele = dictGetKey(de); *str = dictGetKey(de);
*len = sdslen(*str);
*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);
*sdsele = NULL; /* Not needed. Defensive. */ *str = NULL; /* Not needed. Defensive. */
} else if (setobj->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *lp = setobj->ptr;
int r = rand() % lpLength(lp);
unsigned char *p = lpSeek(lp, r);
unsigned int l;
*str = (char *)lpGetValue(p, &l, (long long *)llele);
*len = (size_t)l;
} else { } else {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
} }
return setobj->encoding; return setobj->encoding;
} }
/* Pops a random element and returns it as an object. */
robj *setTypePopRandom(robj *set) {
robj *obj;
if (set->encoding == OBJ_ENCODING_LISTPACK) {
/* Find random and delete it without re-seeking the listpack. */
unsigned int i = 0;
unsigned char *p = lpNextRandom(set->ptr, lpFirst(set->ptr), &i, 1, 0);
unsigned int len = 0; /* initialize to silence warning */
long long llele = 0; /* initialize to silence warning */
char *str = (char *)lpGetValue(p, &len, &llele);
if (str)
obj = createStringObject(str, len);
else
obj = createStringObjectFromLongLong(llele);
set->ptr = lpDelete(set->ptr, p, NULL);
} else {
char *str;
size_t len = 0;
int64_t llele = 0;
int encoding = setTypeRandomElement(set, &str, &len, &llele);
if (str)
obj = createStringObject(str, len);
else
obj = createStringObjectFromLongLong(llele);
setTypeRemoveAux(set, str, len, llele, encoding == OBJ_ENCODING_HT);
}
return obj;
}
unsigned long setTypeSize(const robj *subject) { unsigned long setTypeSize(const robj *subject) {
if (subject->encoding == OBJ_ENCODING_HT) { if (subject->encoding == OBJ_ENCODING_HT) {
return dictSize((const dict*)subject->ptr); return dictSize((const dict*)subject->ptr);
} else if (subject->encoding == OBJ_ENCODING_INTSET) { } else if (subject->encoding == OBJ_ENCODING_INTSET) {
return intsetLen((const intset*)subject->ptr); return intsetLen((const intset*)subject->ptr);
} else if (subject->encoding == OBJ_ENCODING_LISTPACK) {
return lpLength((unsigned char *)subject->ptr);
} else { } else {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
} }
...@@ -238,27 +470,44 @@ unsigned long setTypeSize(const robj *subject) { ...@@ -238,27 +470,44 @@ unsigned long setTypeSize(const robj *subject) {
void setTypeConvert(robj *setobj, int enc) { void setTypeConvert(robj *setobj, int enc) {
setTypeIterator *si; setTypeIterator *si;
serverAssertWithInfo(NULL,setobj,setobj->type == OBJ_SET && serverAssertWithInfo(NULL,setobj,setobj->type == OBJ_SET &&
setobj->encoding == OBJ_ENCODING_INTSET); setobj->encoding != enc);
if (enc == OBJ_ENCODING_HT) { if (enc == OBJ_ENCODING_HT) {
int64_t intele;
dict *d = dictCreate(&setDictType); dict *d = dictCreate(&setDictType);
sds element; sds element;
/* Presize the dict to avoid rehashing */ /* Presize the dict to avoid rehashing */
dictExpand(d,intsetLen(setobj->ptr)); dictExpand(d, setTypeSize(setobj));
/* 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 ((element = setTypeNextObject(si)) != NULL) {
element = sdsfromlonglong(intele);
serverAssert(dictAdd(d,element,NULL) == DICT_OK); serverAssert(dictAdd(d,element,NULL) == DICT_OK);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
freeSetObject(setobj); /* frees the internals but not setobj itself */
setobj->encoding = OBJ_ENCODING_HT; setobj->encoding = OBJ_ENCODING_HT;
zfree(setobj->ptr);
setobj->ptr = d; setobj->ptr = d;
} else if (enc == OBJ_ENCODING_LISTPACK) {
/* Preallocate the minimum one byte per element */
size_t estcap = setTypeSize(setobj);
unsigned char *lp = lpNew(estcap);
char *str;
size_t len;
int64_t llele;
si = setTypeInitIterator(setobj);
while (setTypeNext(si, &str, &len, &llele) != -1) {
if (str != NULL)
lp = lpAppend(lp, (unsigned char *)str, len);
else
lp = lpAppendInteger(lp, llele);
}
setTypeReleaseIterator(si);
freeSetObject(setobj); /* frees the internals but not setobj itself */
setobj->encoding = OBJ_ENCODING_LISTPACK;
setobj->ptr = lp;
} else { } else {
serverPanic("Unsupported set conversion"); serverPanic("Unsupported set conversion");
} }
...@@ -272,8 +521,6 @@ void setTypeConvert(robj *setobj, int enc) { ...@@ -272,8 +521,6 @@ void setTypeConvert(robj *setobj, int enc) {
robj *setTypeDup(robj *o) { robj *setTypeDup(robj *o) {
robj *set; robj *set;
setTypeIterator *si; setTypeIterator *si;
sds elesds;
int64_t intobj;
serverAssert(o->type == OBJ_SET); serverAssert(o->type == OBJ_SET);
...@@ -285,13 +532,23 @@ robj *setTypeDup(robj *o) { ...@@ -285,13 +532,23 @@ robj *setTypeDup(robj *o) {
memcpy(newis,is,size); memcpy(newis,is,size);
set = createObject(OBJ_SET, newis); set = createObject(OBJ_SET, newis);
set->encoding = OBJ_ENCODING_INTSET; set->encoding = OBJ_ENCODING_INTSET;
} else if (o->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *lp = o->ptr;
size_t sz = lpBytes(lp);
unsigned char *new_lp = zmalloc(sz);
memcpy(new_lp, lp, sz);
set = createObject(OBJ_SET, new_lp);
set->encoding = OBJ_ENCODING_LISTPACK;
} else if (o->encoding == OBJ_ENCODING_HT) { } else if (o->encoding == OBJ_ENCODING_HT) {
set = createSetObject(); set = createSetObject();
dict *d = o->ptr; dict *d = o->ptr;
dictExpand(set->ptr, dictSize(d)); dictExpand(set->ptr, dictSize(d));
si = setTypeInitIterator(o); si = setTypeInitIterator(o);
while (setTypeNext(si, &elesds, &intobj) != -1) { char *str;
setTypeAdd(set, elesds); size_t len;
int64_t intobj;
while (setTypeNext(si, &str, &len, &intobj) != -1) {
setTypeAdd(set, (sds)str);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
} else { } else {
...@@ -509,9 +766,9 @@ void spopWithCountCommand(client *c) { ...@@ -509,9 +766,9 @@ void spopWithCountCommand(client *c) {
addReplySetLen(c,count); addReplySetLen(c,count);
/* Common iteration vars. */ /* Common iteration vars. */
sds sdsele;
robj *objele; robj *objele;
int encoding; char *str;
size_t len;
int64_t llele; int64_t llele;
unsigned long remaining = size-count; /* Elements left after SPOP. */ unsigned long remaining = size-count; /* Elements left after SPOP. */
...@@ -522,24 +779,49 @@ void spopWithCountCommand(client *c) { ...@@ -522,24 +779,49 @@ void spopWithCountCommand(client *c) {
* CASE 2: The number of elements to return is small compared to the * CASE 2: The number of elements to return is small compared to the
* set size. We can just extract random elements and return them to * set size. We can just extract random elements and return them to
* the set. */ * the set. */
if (remaining*SPOP_MOVE_STRATEGY_MUL > count) { if (remaining*SPOP_MOVE_STRATEGY_MUL > count &&
while(count--) { set->encoding == OBJ_ENCODING_LISTPACK)
/* Emit and remove. */ {
encoding = setTypeRandomElement(set,&sdsele,&llele); /* Specialized case for listpack. Traverse it only once. */
if (encoding == OBJ_ENCODING_INTSET) { unsigned char *lp = set->ptr;
addReplyBulkLongLong(c,llele); unsigned char *p = lpFirst(lp);
objele = createStringObjectFromLongLong(llele); unsigned int index = 0;
set->ptr = intsetRemove(set->ptr,llele,NULL); unsigned char **ps = zmalloc(sizeof(char *) * count);
for (unsigned long i = 0; i < count; i++) {
p = lpNextRandom(lp, p, &index, count - i, 0);
unsigned int len;
str = (char *)lpGetValue(p, &len, (long long *)&llele);
if (str) {
addReplyBulkCBuffer(c, str, len);
objele = createStringObject(str, len);
} else { } else {
addReplyBulkCBuffer(c,sdsele,sdslen(sdsele)); addReplyBulkLongLong(c, llele);
objele = createStringObject(sdsele,sdslen(sdsele)); objele = createStringObjectFromLongLong(llele);
setTypeRemove(set,sdsele);
} }
/* Replicate/AOF this command as an SREM operation */ /* Replicate/AOF this command as an SREM operation */
propargv[2] = objele; propargv[2] = objele;
alsoPropagate(c->db->id,propargv,3,PROPAGATE_AOF|PROPAGATE_REPL); alsoPropagate(c->db->id,propargv,3,PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(objele); decrRefCount(objele);
/* Store pointer for later deletion and move to next. */
ps[i] = p;
p = lpNext(lp, p);
index++;
}
lp = lpBatchDelete(lp, ps, count);
zfree(ps);
set->ptr = lp;
} else if (remaining*SPOP_MOVE_STRATEGY_MUL > count) {
while(count--) {
objele = setTypePopRandom(set);
addReplyBulk(c, objele);
/* Replicate/AOF this command as an SREM operation */
propargv[2] = objele;
alsoPropagate(c->db->id,propargv,3,PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(objele);
} }
} else { } else {
/* CASE 3: The number of elements to return is very big, approaching /* CASE 3: The number of elements to return is very big, approaching
...@@ -553,29 +835,46 @@ void spopWithCountCommand(client *c) { ...@@ -553,29 +835,46 @@ void spopWithCountCommand(client *c) {
robj *newset = NULL; robj *newset = NULL;
/* Create a new set with just the remaining elements. */ /* Create a new set with just the remaining elements. */
while(remaining--) { if (set->encoding == OBJ_ENCODING_LISTPACK) {
encoding = setTypeRandomElement(set,&sdsele,&llele); /* Specialized case for listpack. Traverse it only once. */
if (encoding == OBJ_ENCODING_INTSET) { newset = createSetListpackObject();
sdsele = sdsfromlonglong(llele); unsigned char *lp = set->ptr;
} else { unsigned char *p = lpFirst(lp);
sdsele = sdsdup(sdsele); unsigned int index = 0;
unsigned char **ps = zmalloc(sizeof(char *) * remaining);
for (unsigned long i = 0; i < remaining; i++) {
p = lpNextRandom(lp, p, &index, remaining - i, 0);
unsigned int len;
str = (char *)lpGetValue(p, &len, (long long *)&llele);
setTypeAddAux(newset, str, len, llele, 0);
ps[i] = p;
p = lpNext(lp, p);
index++;
}
lp = lpBatchDelete(lp, ps, remaining);
zfree(ps);
set->ptr = lp;
} else {
while(remaining--) {
int encoding = setTypeRandomElement(set, &str, &len, &llele);
if (!newset) {
newset = str ? createSetListpackObject() : createIntsetObject();
}
setTypeAddAux(newset, str, len, llele, encoding == OBJ_ENCODING_HT);
setTypeRemoveAux(set, str, len, llele, encoding == OBJ_ENCODING_HT);
} }
if (!newset) newset = setTypeCreate(sdsele);
setTypeAdd(newset,sdsele);
setTypeRemove(set,sdsele);
sdsfree(sdsele);
} }
/* Transfer the old set to the client. */ /* Transfer the old set to the client. */
setTypeIterator *si; setTypeIterator *si;
si = setTypeInitIterator(set); si = setTypeInitIterator(set);
while((encoding = setTypeNext(si,&sdsele,&llele)) != -1) { while (setTypeNext(si, &str, &len, &llele) != -1) {
if (encoding == OBJ_ENCODING_INTSET) { if (str == NULL) {
addReplyBulkLongLong(c,llele); addReplyBulkLongLong(c,llele);
objele = createStringObjectFromLongLong(llele); objele = createStringObjectFromLongLong(llele);
} else { } else {
addReplyBulkCBuffer(c,sdsele,sdslen(sdsele)); addReplyBulkCBuffer(c, str, len);
objele = createStringObject(sdsele,sdslen(sdsele)); objele = createStringObject(str, len);
} }
/* Replicate/AOF this command as an SREM operation */ /* Replicate/AOF this command as an SREM operation */
...@@ -599,9 +898,6 @@ void spopWithCountCommand(client *c) { ...@@ -599,9 +898,6 @@ void spopWithCountCommand(client *c) {
void spopCommand(client *c) { void spopCommand(client *c) {
robj *set, *ele; robj *set, *ele;
sds sdsele;
int64_t llele;
int encoding;
if (c->argc == 3) { if (c->argc == 3) {
spopWithCountCommand(c); spopWithCountCommand(c);
...@@ -616,17 +912,8 @@ void spopCommand(client *c) { ...@@ -616,17 +912,8 @@ void spopCommand(client *c) {
if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.null[c->resp])) if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.null[c->resp]))
== NULL || checkType(c,set,OBJ_SET)) return; == NULL || checkType(c,set,OBJ_SET)) return;
/* Get a random element from the set */ /* Pop a random element from the set */
encoding = setTypeRandomElement(set,&sdsele,&llele); ele = setTypePopRandom(set);
/* Remove the element from the set */
if (encoding == OBJ_ENCODING_INTSET) {
ele = createStringObjectFromLongLong(llele);
set->ptr = intsetRemove(set->ptr,llele,NULL);
} else {
ele = createStringObject(sdsele,sdslen(sdsele));
setTypeRemove(set,ele->ptr);
}
notifyKeyspaceEvent(NOTIFY_SET,"spop",c->argv[1],c->db->id); notifyKeyspaceEvent(NOTIFY_SET,"spop",c->argv[1],c->db->id);
...@@ -634,7 +921,7 @@ void spopCommand(client *c) { ...@@ -634,7 +921,7 @@ void spopCommand(client *c) {
rewriteClientCommandVector(c,3,shared.srem,c->argv[1],ele); rewriteClientCommandVector(c,3,shared.srem,c->argv[1],ele);
/* Add the element to the reply */ /* Add the element to the reply */
addReplyBulk(c,ele); addReplyBulk(c, ele);
decrRefCount(ele); decrRefCount(ele);
/* Delete the set if it's empty */ /* Delete the set if it's empty */
...@@ -661,9 +948,9 @@ void srandmemberWithCountCommand(client *c) { ...@@ -661,9 +948,9 @@ void srandmemberWithCountCommand(client *c) {
unsigned long count, size; unsigned long count, size;
int uniq = 1; int uniq = 1;
robj *set; robj *set;
sds ele; char *str;
size_t len;
int64_t llele; int64_t llele;
int encoding;
dict *d; dict *d;
...@@ -694,12 +981,27 @@ void srandmemberWithCountCommand(client *c) { ...@@ -694,12 +981,27 @@ void srandmemberWithCountCommand(client *c) {
* elements in random order. */ * elements in random order. */
if (!uniq || count == 1) { if (!uniq || count == 1) {
addReplyArrayLen(c,count); addReplyArrayLen(c,count);
if (set->encoding == OBJ_ENCODING_LISTPACK && count > 1) {
/* Specialized case for listpack, traversing it only once. */
listpackEntry *entries = zmalloc(count * sizeof(listpackEntry));
lpRandomEntries(set->ptr, count, entries);
for (unsigned long i = 0; i < count; i++) {
if (entries[i].sval)
addReplyBulkCBuffer(c, entries[i].sval, entries[i].slen);
else
addReplyBulkLongLong(c, entries[i].lval);
}
zfree(entries);
return;
}
while(count--) { while(count--) {
encoding = setTypeRandomElement(set,&ele,&llele); setTypeRandomElement(set, &str, &len, &llele);
if (encoding == OBJ_ENCODING_INTSET) { if (str == NULL) {
addReplyBulkLongLong(c,llele); addReplyBulkLongLong(c,llele);
} else { } else {
addReplyBulkCBuffer(c,ele,sdslen(ele)); addReplyBulkCBuffer(c, str, len);
} }
} }
return; return;
...@@ -712,11 +1014,11 @@ void srandmemberWithCountCommand(client *c) { ...@@ -712,11 +1014,11 @@ void srandmemberWithCountCommand(client *c) {
setTypeIterator *si; setTypeIterator *si;
addReplyArrayLen(c,size); addReplyArrayLen(c,size);
si = setTypeInitIterator(set); si = setTypeInitIterator(set);
while ((encoding = setTypeNext(si,&ele,&llele)) != -1) { while (setTypeNext(si, &str, &len, &llele) != -1) {
if (encoding == OBJ_ENCODING_INTSET) { if (str == NULL) {
addReplyBulkLongLong(c,llele); addReplyBulkLongLong(c,llele);
} else { } else {
addReplyBulkCBuffer(c,ele,sdslen(ele)); addReplyBulkCBuffer(c, str, len);
} }
size--; size--;
} }
...@@ -725,6 +1027,31 @@ void srandmemberWithCountCommand(client *c) { ...@@ -725,6 +1027,31 @@ void srandmemberWithCountCommand(client *c) {
return; return;
} }
/* CASE 2.5 listpack only. Sampling unique elements, in non-random order.
* Listpack encoded sets are meant to be relatively small, so
* SRANDMEMBER_SUB_STRATEGY_MUL isn't necessary and we rather not make
* copies of the entries. Instead, we emit them directly to the output
* buffer. */
if (set->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *lp = set->ptr;
unsigned char *p = lpFirst(lp);
unsigned int i = 0;
addReplyArrayLen(c, count);
while (count) {
p = lpNextRandom(lp, p, &i, count--, 0);
unsigned int len;
str = (char *)lpGetValue(p, &len, (long long *)&llele);
if (str == NULL) {
addReplyBulkLongLong(c, llele);
} else {
addReplyBulkCBuffer(c, str, len);
}
p = lpNext(lp, p);
i++;
}
return;
}
/* For CASE 3 and CASE 4 we need an auxiliary dictionary. */ /* For CASE 3 and CASE 4 we need an auxiliary dictionary. */
d = dictCreate(&sdsReplyDictType); d = dictCreate(&sdsReplyDictType);
...@@ -743,13 +1070,13 @@ void srandmemberWithCountCommand(client *c) { ...@@ -743,13 +1070,13 @@ void srandmemberWithCountCommand(client *c) {
/* Add all the elements into the temporary dictionary. */ /* Add all the elements into the temporary dictionary. */
si = setTypeInitIterator(set); si = setTypeInitIterator(set);
dictExpand(d, size); dictExpand(d, size);
while ((encoding = setTypeNext(si,&ele,&llele)) != -1) { while (setTypeNext(si, &str, &len, &llele) != -1) {
int retval = DICT_ERR; int retval = DICT_ERR;
if (encoding == OBJ_ENCODING_INTSET) { if (str == NULL) {
retval = dictAdd(d,sdsfromlonglong(llele),NULL); retval = dictAdd(d,sdsfromlonglong(llele),NULL);
} else { } else {
retval = dictAdd(d,sdsdup(ele),NULL); retval = dictAdd(d, sdsnewlen(str, len), NULL);
} }
serverAssert(retval == DICT_OK); serverAssert(retval == DICT_OK);
} }
...@@ -777,11 +1104,11 @@ void srandmemberWithCountCommand(client *c) { ...@@ -777,11 +1104,11 @@ void srandmemberWithCountCommand(client *c) {
dictExpand(d, count); dictExpand(d, count);
while (added < count) { while (added < count) {
encoding = setTypeRandomElement(set,&ele,&llele); setTypeRandomElement(set, &str, &len, &llele);
if (encoding == OBJ_ENCODING_INTSET) { if (str == NULL) {
sdsele = sdsfromlonglong(llele); sdsele = sdsfromlonglong(llele);
} else { } else {
sdsele = sdsdup(ele); sdsele = sdsnewlen(str, len);
} }
/* 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
...@@ -810,9 +1137,9 @@ void srandmemberWithCountCommand(client *c) { ...@@ -810,9 +1137,9 @@ void srandmemberWithCountCommand(client *c) {
/* SRANDMEMBER <key> [<count>] */ /* SRANDMEMBER <key> [<count>] */
void srandmemberCommand(client *c) { void srandmemberCommand(client *c) {
robj *set; robj *set;
sds ele; char *str;
size_t len;
int64_t llele; int64_t llele;
int encoding;
if (c->argc == 3) { if (c->argc == 3) {
srandmemberWithCountCommand(c); srandmemberWithCountCommand(c);
...@@ -826,11 +1153,11 @@ void srandmemberCommand(client *c) { ...@@ -826,11 +1153,11 @@ void srandmemberCommand(client *c) {
if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp])) if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp]))
== NULL || checkType(c,set,OBJ_SET)) return; == NULL || checkType(c,set,OBJ_SET)) return;
encoding = setTypeRandomElement(set,&ele,&llele); setTypeRandomElement(set, &str, &len, &llele);
if (encoding == OBJ_ENCODING_INTSET) { if (str == NULL) {
addReplyBulkLongLong(c,llele); addReplyBulkLongLong(c,llele);
} else { } else {
addReplyBulkCBuffer(c,ele,sdslen(ele)); addReplyBulkCBuffer(c, str, len);
} }
} }
...@@ -866,7 +1193,8 @@ void sinterGenericCommand(client *c, robj **setkeys, ...@@ -866,7 +1193,8 @@ void sinterGenericCommand(client *c, robj **setkeys,
robj **sets = zmalloc(sizeof(robj*)*setnum); robj **sets = zmalloc(sizeof(robj*)*setnum);
setTypeIterator *si; setTypeIterator *si;
robj *dstset = NULL; robj *dstset = NULL;
sds elesds; char *str;
size_t len;
int64_t intobj; int64_t intobj;
void *replylen = NULL; void *replylen = NULL;
unsigned long j, cardinality = 0; unsigned long j, cardinality = 0;
...@@ -918,7 +1246,24 @@ void sinterGenericCommand(client *c, robj **setkeys, ...@@ -918,7 +1246,24 @@ void sinterGenericCommand(client *c, robj **setkeys,
if (dstkey) { if (dstkey) {
/* If we have a target key where to store the resulting set /* If we have a target key where to store the resulting set
* create this key with an empty set inside */ * create this key with an empty set inside */
dstset = createIntsetObject(); if (sets[0]->encoding == OBJ_ENCODING_INTSET) {
/* The first set is an intset, so the result is an intset too. The
* elements are inserted in ascending order which is efficient in an
* intset. */
dstset = createIntsetObject();
} else if (sets[0]->encoding == OBJ_ENCODING_LISTPACK) {
/* To avoid many reallocs, we estimate that the result is a listpack
* of approximately the same size as the first set. Then we shrink
* it or possibly convert it to intset it in the end. */
unsigned char *lp = lpNew(lpBytes(sets[0]->ptr));
dstset = createObject(OBJ_SET, lp);
dstset->encoding = OBJ_ENCODING_LISTPACK;
} else {
/* We start off with a listpack, since it's more efficient to append
* to than an intset. Later we can convert it to intset or a
* hashtable. */
dstset = createSetListpackObject();
}
} else if (!cardinality_only) { } else if (!cardinality_only) {
replylen = addReplyDeferredLen(c); replylen = addReplyDeferredLen(c);
} }
...@@ -926,32 +1271,14 @@ void sinterGenericCommand(client *c, robj **setkeys, ...@@ -926,32 +1271,14 @@ void sinterGenericCommand(client *c, robj **setkeys,
/* Iterate all the elements of the first (smallest) set, and test /* Iterate all the elements of the first (smallest) set, and test
* 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 */
int only_integers = 1;
si = setTypeInitIterator(sets[0]); si = setTypeInitIterator(sets[0]);
while((encoding = setTypeNext(si,&elesds,&intobj)) != -1) { while((encoding = setTypeNext(si, &str, &len, &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 (!setTypeIsMemberAux(sets[j], str, len, intobj,
/* intset with intset is simple... and fast */ encoding == OBJ_ENCODING_HT))
if (sets[j]->encoding == OBJ_ENCODING_INTSET && break;
!intsetFind((intset*)sets[j]->ptr,intobj))
{
break;
/* in order to compare an integer with an object we
* have to use the generic function, creating an object
* for this */
} else if (sets[j]->encoding == OBJ_ENCODING_HT) {
elesds = sdsfromlonglong(intobj);
if (!setTypeIsMember(sets[j],elesds)) {
sdsfree(elesds);
break;
}
sdsfree(elesds);
}
} else if (encoding == OBJ_ENCODING_HT) {
if (!setTypeIsMember(sets[j],elesds)) {
break;
}
}
} }
/* Only take action when all sets contain the member */ /* Only take action when all sets contain the member */
...@@ -963,19 +1290,29 @@ void sinterGenericCommand(client *c, robj **setkeys, ...@@ -963,19 +1290,29 @@ void sinterGenericCommand(client *c, robj **setkeys,
if (limit && cardinality >= limit) if (limit && cardinality >= limit)
break; break;
} else if (!dstkey) { } else if (!dstkey) {
if (encoding == OBJ_ENCODING_HT) if (str != NULL)
addReplyBulkCBuffer(c,elesds,sdslen(elesds)); addReplyBulkCBuffer(c, str, len);
else else
addReplyBulkLongLong(c,intobj); addReplyBulkLongLong(c,intobj);
cardinality++; cardinality++;
} else { } else {
if (encoding == OBJ_ENCODING_INTSET) { if (str && only_integers) {
elesds = sdsfromlonglong(intobj); /* It may be an integer although we got it as a string. */
setTypeAdd(dstset,elesds); if (encoding == OBJ_ENCODING_HT &&
sdsfree(elesds); string2ll(str, len, (long long *)&intobj))
} else { {
setTypeAdd(dstset,elesds); if (dstset->encoding == OBJ_ENCODING_LISTPACK ||
dstset->encoding == OBJ_ENCODING_INTSET)
{
/* Adding it as an integer is more efficient. */
str = NULL;
}
} else {
/* It's not an integer */
only_integers = 0;
}
} }
setTypeAddAux(dstset, str, len, intobj, encoding == OBJ_ENCODING_HT);
} }
} }
} }
...@@ -987,6 +1324,12 @@ void sinterGenericCommand(client *c, robj **setkeys, ...@@ -987,6 +1324,12 @@ void sinterGenericCommand(client *c, robj **setkeys,
/* Store the resulting set into the target, if the intersection /* Store the resulting set into the target, if the intersection
* is not an empty set. */ * is not an empty set. */
if (setTypeSize(dstset) > 0) { if (setTypeSize(dstset) > 0) {
if (only_integers) maybeConvertToIntset(dstset);
if (dstset->encoding == OBJ_ENCODING_LISTPACK) {
/* We allocated too much memory when we created it to avoid
* frequent reallocs. Therefore, we shrink it now. */
dstset->ptr = lpShrinkToFit(dstset->ptr);
}
setKey(c,c->db,dstkey,dstset,0); setKey(c,c->db,dstkey,dstset,0);
addReplyLongLong(c,setTypeSize(dstset)); addReplyLongLong(c,setTypeSize(dstset));
notifyKeyspaceEvent(NOTIFY_SET,"sinterstore", notifyKeyspaceEvent(NOTIFY_SET,"sinterstore",
...@@ -1054,7 +1397,10 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum, ...@@ -1054,7 +1397,10 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
robj **sets = zmalloc(sizeof(robj*)*setnum); robj **sets = zmalloc(sizeof(robj*)*setnum);
setTypeIterator *si; setTypeIterator *si;
robj *dstset = NULL; robj *dstset = NULL;
sds ele; char *str;
size_t len;
int64_t llval;
int encoding;
int j, cardinality = 0; int j, cardinality = 0;
int diff_algo = 1; int diff_algo = 1;
int sameset = 0; int sameset = 0;
...@@ -1120,9 +1466,8 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum, ...@@ -1120,9 +1466,8 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
if (!sets[j]) continue; /* non existing keys are like empty sets */ if (!sets[j]) continue; /* non existing keys are like empty sets */
si = setTypeInitIterator(sets[j]); si = setTypeInitIterator(sets[j]);
while((ele = setTypeNextObject(si)) != NULL) { while ((encoding = setTypeNext(si, &str, &len, &llval)) != -1) {
if (setTypeAdd(dstset,ele)) cardinality++; cardinality += setTypeAddAux(dstset, str, len, llval, encoding == OBJ_ENCODING_HT);
sdsfree(ele);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
} }
...@@ -1138,18 +1483,19 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum, ...@@ -1138,18 +1483,19 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
* This way we perform at max N*M operations, where N is the size of * This way we perform at max N*M operations, where N is the size of
* the first set, and M the number of sets. */ * the first set, and M the number of sets. */
si = setTypeInitIterator(sets[0]); si = setTypeInitIterator(sets[0]);
while((ele = setTypeNextObject(si)) != NULL) { while ((encoding = setTypeNext(si, &str, &len, &llval)) != -1) {
for (j = 1; j < setnum; j++) { for (j = 1; j < setnum; j++) {
if (!sets[j]) continue; /* no key is an empty set. */ if (!sets[j]) continue; /* no key is an empty set. */
if (sets[j] == sets[0]) break; /* same set! */ if (sets[j] == sets[0]) break; /* same set! */
if (setTypeIsMember(sets[j],ele)) break; if (setTypeIsMemberAux(sets[j], str, len, llval,
encoding == OBJ_ENCODING_HT))
break;
} }
if (j == setnum) { if (j == setnum) {
/* There is no other set with this element. Add it. */ /* There is no other set with this element. Add it. */
setTypeAdd(dstset,ele); setTypeAddAux(dstset, str, len, llval, encoding == OBJ_ENCODING_HT);
cardinality++; cardinality++;
} }
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) {
...@@ -1164,13 +1510,14 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum, ...@@ -1164,13 +1510,14 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
if (!sets[j]) continue; /* non existing keys are like empty sets */ if (!sets[j]) continue; /* non existing keys are like empty sets */
si = setTypeInitIterator(sets[j]); si = setTypeInitIterator(sets[j]);
while((ele = setTypeNextObject(si)) != NULL) { while((encoding = setTypeNext(si, &str, &len, &llval)) != -1) {
if (j == 0) { if (j == 0) {
if (setTypeAdd(dstset,ele)) cardinality++; cardinality += setTypeAddAux(dstset, str, len, llval,
encoding == OBJ_ENCODING_HT);
} else { } else {
if (setTypeRemove(dstset,ele)) cardinality--; cardinality -= setTypeRemoveAux(dstset, str, len, llval,
encoding == OBJ_ENCODING_HT);
} }
sdsfree(ele);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
...@@ -1184,9 +1531,11 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum, ...@@ -1184,9 +1531,11 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
if (!dstkey) { if (!dstkey) {
addReplySetLen(c,cardinality); addReplySetLen(c,cardinality);
si = setTypeInitIterator(dstset); si = setTypeInitIterator(dstset);
while((ele = setTypeNextObject(si)) != NULL) { while (setTypeNext(si, &str, &len, &llval) != -1) {
addReplyBulkCBuffer(c,ele,sdslen(ele)); if (str)
sdsfree(ele); addReplyBulkCBuffer(c, str, len);
else
addReplyBulkLongLong(c, llval);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
server.lazyfree_lazy_server_del ? freeObjAsync(NULL, dstset, -1) : server.lazyfree_lazy_server_del ? freeObjAsync(NULL, dstset, -1) :
......
...@@ -1971,6 +1971,10 @@ typedef struct { ...@@ -1971,6 +1971,10 @@ typedef struct {
dictIterator *di; dictIterator *di;
dictEntry *de; dictEntry *de;
} ht; } ht;
struct {
unsigned char *lp;
unsigned char *p;
} lp;
} set; } set;
/* Sorted set iterators. */ /* Sorted set iterators. */
...@@ -2025,6 +2029,9 @@ void zuiInitIterator(zsetopsrc *op) { ...@@ -2025,6 +2029,9 @@ void zuiInitIterator(zsetopsrc *op) {
it->ht.dict = op->subject->ptr; it->ht.dict = op->subject->ptr;
it->ht.di = dictGetIterator(op->subject->ptr); it->ht.di = dictGetIterator(op->subject->ptr);
it->ht.de = dictNext(it->ht.di); it->ht.de = dictNext(it->ht.di);
} else if (op->encoding == OBJ_ENCODING_LISTPACK) {
it->lp.lp = op->subject->ptr;
it->lp.p = lpFirst(it->lp.lp);
} else { } else {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
} }
...@@ -2061,6 +2068,8 @@ void zuiClearIterator(zsetopsrc *op) { ...@@ -2061,6 +2068,8 @@ void zuiClearIterator(zsetopsrc *op) {
UNUSED(it); /* skip */ UNUSED(it); /* skip */
} else if (op->encoding == OBJ_ENCODING_HT) { } else if (op->encoding == OBJ_ENCODING_HT) {
dictReleaseIterator(it->ht.di); dictReleaseIterator(it->ht.di);
} else if (op->encoding == OBJ_ENCODING_LISTPACK) {
UNUSED(it);
} else { } else {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
} }
...@@ -2091,14 +2100,7 @@ unsigned long zuiLength(zsetopsrc *op) { ...@@ -2091,14 +2100,7 @@ unsigned long zuiLength(zsetopsrc *op) {
return 0; return 0;
if (op->type == OBJ_SET) { if (op->type == OBJ_SET) {
if (op->encoding == OBJ_ENCODING_INTSET) { return setTypeSize(op->subject);
return intsetLen(op->subject->ptr);
} else if (op->encoding == OBJ_ENCODING_HT) {
dict *ht = op->subject->ptr;
return dictSize(ht);
} else {
serverPanic("Unknown set encoding");
}
} else if (op->type == OBJ_ZSET) { } else if (op->type == OBJ_ZSET) {
if (op->encoding == OBJ_ENCODING_LISTPACK) { if (op->encoding == OBJ_ENCODING_LISTPACK) {
return zzlLength(op->subject->ptr); return zzlLength(op->subject->ptr);
...@@ -2144,6 +2146,14 @@ int zuiNext(zsetopsrc *op, zsetopval *val) { ...@@ -2144,6 +2146,14 @@ int zuiNext(zsetopsrc *op, zsetopval *val) {
/* Move to next element. */ /* Move to next element. */
it->ht.de = dictNext(it->ht.di); it->ht.de = dictNext(it->ht.di);
} else if (op->encoding == OBJ_ENCODING_LISTPACK) {
if (it->lp.p == NULL)
return 0;
val->estr = lpGetValue(it->lp.p, &val->elen, &val->ell);
val->score = 1.0;
/* Move to next element. */
it->lp.p = lpNext(it->lp.lp, it->lp.p);
} else { } else {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
} }
......
...@@ -256,30 +256,27 @@ start_server {tags {"keyspace"}} { ...@@ -256,30 +256,27 @@ start_server {tags {"keyspace"}} {
assert_equal $digest [debug_digest_value mynewlist{t}] assert_equal $digest [debug_digest_value mynewlist{t}]
} }
test {COPY basic usage for intset set} { foreach type {intset listpack hashtable} {
r del set1{t} newset1{t} test {COPY basic usage for $type set} {
r sadd set1{t} 1 2 3 r del set1{t} newset1{t}
assert_encoding intset set1{t} r sadd set1{t} 1 2 3
r copy set1{t} newset1{t} if {$type ne "intset"} {
set digest [debug_digest_value set1{t}] r sadd set1{t} a
assert_equal $digest [debug_digest_value newset1{t}] }
assert_refcount 1 set1{t} if {$type eq "hashtable"} {
assert_refcount 1 newset1{t} for {set i 4} {$i < 200} {incr i} {
r del set1{t} r sadd set1{t} $i
assert_equal $digest [debug_digest_value newset1{t}] }
} }
assert_encoding $type set1{t}
test {COPY basic usage for hashtable set} { r copy set1{t} newset1{t}
r del set2{t} newset2{t} set digest [debug_digest_value set1{t}]
r sadd set2{t} 1 2 3 a assert_equal $digest [debug_digest_value newset1{t}]
assert_encoding hashtable set2{t} assert_refcount 1 set1{t}
r copy set2{t} newset2{t} assert_refcount 1 newset1{t}
set digest [debug_digest_value set2{t}] r del set1{t}
assert_equal $digest [debug_digest_value newset2{t}] assert_equal $digest [debug_digest_value newset1{t}]
assert_refcount 1 set2{t} }
assert_refcount 1 newset2{t}
r del set2{t}
assert_equal $digest [debug_digest_value newset2{t}]
} }
test {COPY basic usage for listpack sorted set} { test {COPY basic usage for listpack sorted set} {
......
...@@ -98,7 +98,7 @@ start_server {tags {"scan network"}} { ...@@ -98,7 +98,7 @@ start_server {tags {"scan network"}} {
assert_equal 1000 [llength $keys] assert_equal 1000 [llength $keys]
} }
foreach enc {intset hashtable} { foreach enc {intset listpack hashtable} {
test "SSCAN with encoding $enc" { test "SSCAN with encoding $enc" {
# Create the Set # Create the Set
r del set r del set
...@@ -107,8 +107,9 @@ start_server {tags {"scan network"}} { ...@@ -107,8 +107,9 @@ start_server {tags {"scan network"}} {
} else { } else {
set prefix "ele:" set prefix "ele:"
} }
set count [expr {$enc eq "hashtable" ? 200 : 100}]
set elements {} set elements {}
for {set j 0} {$j < 100} {incr j} { for {set j 0} {$j < $count} {incr j} {
lappend elements ${prefix}${j} lappend elements ${prefix}${j}
} }
r sadd set {*}$elements r sadd set {*}$elements
...@@ -128,7 +129,7 @@ start_server {tags {"scan network"}} { ...@@ -128,7 +129,7 @@ start_server {tags {"scan network"}} {
} }
set keys [lsort -unique $keys] set keys [lsort -unique $keys]
assert_equal 100 [llength $keys] assert_equal $count [llength $keys]
} }
} }
......
...@@ -2,6 +2,8 @@ start_server { ...@@ -2,6 +2,8 @@ start_server {
tags {"set"} tags {"set"}
overrides { overrides {
"set-max-intset-entries" 512 "set-max-intset-entries" 512
"set-max-listpack-entries" 128
"set-max-listpack-value" 32
} }
} { } {
proc create_set {key entries} { proc create_set {key entries} {
...@@ -9,12 +11,19 @@ start_server { ...@@ -9,12 +11,19 @@ start_server {
foreach entry $entries { r sadd $key $entry } foreach entry $entries { r sadd $key $entry }
} }
test {SADD, SCARD, SISMEMBER, SMISMEMBER, SMEMBERS basics - regular set} { # Values for initialing sets, per encoding.
create_set myset {foo} array set initelems {listpack {foo} hashtable {foo}}
assert_encoding hashtable myset for {set i 0} {$i < 130} {incr i} {
lappend initelems(hashtable) [format "i%03d" $i]
}
foreach type {listpack hashtable} {
test "SADD, SCARD, SISMEMBER, SMISMEMBER, SMEMBERS basics - $type" {
create_set myset $initelems($type)
assert_encoding $type myset
assert_equal 1 [r sadd myset bar] assert_equal 1 [r sadd myset bar]
assert_equal 0 [r sadd myset bar] assert_equal 0 [r sadd myset bar]
assert_equal 2 [r scard myset] assert_equal [expr [llength $initelems($type)] + 1] [r scard myset]
assert_equal 1 [r sismember myset foo] assert_equal 1 [r sismember myset foo]
assert_equal 1 [r sismember myset bar] assert_equal 1 [r sismember myset bar]
assert_equal 0 [r sismember myset bla] assert_equal 0 [r sismember myset bla]
...@@ -23,7 +32,8 @@ start_server { ...@@ -23,7 +32,8 @@ start_server {
assert_equal {1 0} [r smismember myset foo bla] assert_equal {1 0} [r smismember myset foo bla]
assert_equal {0 1} [r smismember myset bla foo] assert_equal {0 1} [r smismember myset bla foo]
assert_equal {0} [r smismember myset bla] assert_equal {0} [r smismember myset bla]
assert_equal {bar foo} [lsort [r smembers myset]] assert_equal "bar $initelems($type)" [lsort [r smembers myset]]
}
} }
test {SADD, SCARD, SISMEMBER, SMISMEMBER, SMEMBERS basics - intset} { test {SADD, SCARD, SISMEMBER, SMISMEMBER, SMEMBERS basics - intset} {
...@@ -67,15 +77,33 @@ start_server { ...@@ -67,15 +77,33 @@ start_server {
assert_error WRONGTYPE* {r sadd mylist bar} assert_error WRONGTYPE* {r sadd mylist bar}
} }
test "SADD a non-integer against an intset" { test "SADD a non-integer against a small intset" {
create_set myset {1 2 3} create_set myset {1 2 3}
assert_encoding intset myset assert_encoding intset myset
assert_equal 1 [r sadd myset a] assert_equal 1 [r sadd myset a]
assert_encoding listpack myset
}
test "SADD a non-integer against a large intset" {
create_set myset {0}
for {set i 1} {$i < 130} {incr i} {r sadd myset $i}
assert_encoding intset myset
assert_equal 1 [r sadd myset a]
assert_encoding hashtable myset assert_encoding hashtable myset
} }
test "SADD an integer larger than 64 bits" { test "SADD an integer larger than 64 bits" {
create_set myset {213244124402402314402033402} create_set myset {213244124402402314402033402}
assert_encoding listpack myset
assert_equal 1 [r sismember myset 213244124402402314402033402]
assert_equal {1} [r smismember myset 213244124402402314402033402]
}
test "SADD an integer larger than 64 bits to a large intset" {
create_set myset {0}
for {set i 1} {$i < 130} {incr i} {r sadd myset $i}
assert_encoding intset myset
r sadd myset 213244124402402314402033402
assert_encoding hashtable myset assert_encoding hashtable myset
assert_equal 1 [r sismember myset 213244124402402314402033402] assert_equal 1 [r sismember myset 213244124402402314402033402]
assert_equal {1} [r smismember myset 213244124402402314402033402] assert_equal {1} [r smismember myset 213244124402402314402033402]
...@@ -100,25 +128,32 @@ start_server { ...@@ -100,25 +128,32 @@ start_server {
r del myintset r del myintset
r del myhashset r del myhashset
r del mylargeintset r del mylargeintset
r del mysmallset
for {set i 0} {$i < 100} {incr i} { r sadd myintset $i } for {set i 0} {$i < 100} {incr i} { r sadd myintset $i }
for {set i 0} {$i < 1280} {incr i} { r sadd mylargeintset $i } for {set i 0} {$i < 1280} {incr i} { r sadd mylargeintset $i }
for {set i 0} {$i < 50} {incr i} { r sadd mysmallset [format "i%03d" $i] }
for {set i 0} {$i < 256} {incr i} { r sadd myhashset [format "i%03d" $i] } for {set i 0} {$i < 256} {incr i} { r sadd myhashset [format "i%03d" $i] }
assert_encoding intset myintset assert_encoding intset myintset
assert_encoding hashtable mylargeintset assert_encoding hashtable mylargeintset
assert_encoding listpack mysmallset
assert_encoding hashtable myhashset assert_encoding hashtable myhashset
r debug reload r debug reload
assert_encoding intset myintset assert_encoding intset myintset
assert_encoding hashtable mylargeintset assert_encoding hashtable mylargeintset
assert_encoding listpack mysmallset
assert_encoding hashtable myhashset assert_encoding hashtable myhashset
} {} {needs:debug} } {} {needs:debug}
test {SREM basics - regular set} { foreach type {listpack hashtable} {
create_set myset {foo bar ciao} test {SREM basics - $type} {
assert_encoding hashtable myset create_set myset $initelems($type)
assert_equal 0 [r srem myset qux] r sadd myset ciao
assert_equal 1 [r srem myset foo] assert_encoding $type myset
assert_equal {bar ciao} [lsort [r smembers myset]] assert_equal 0 [r srem myset qux]
assert_equal 1 [r srem myset ciao]
assert_equal $initelems($type) [lsort [r smembers myset]]
}
} }
test {SREM basics - intset} { test {SREM basics - intset} {
...@@ -177,7 +212,18 @@ start_server { ...@@ -177,7 +212,18 @@ start_server {
assert_equal 0 [r sintercard 1 non-existing-key limit 10] assert_equal 0 [r sintercard 1 non-existing-key limit 10]
} }
foreach {type} {hashtable intset} { foreach {type} {regular intset} {
# Create sets setN{t} where N = 1..5
if {$type eq "regular"} {
set smallenc listpack
set bigenc hashtable
} else {
set smallenc intset
set bigenc intset
}
# Sets 1, 2 and 4 are big; sets 3 and 5 are small.
array set encoding "1 $bigenc 2 $bigenc 3 $smallenc 4 $bigenc 5 $smallenc"
for {set i 1} {$i <= 5} {incr i} { for {set i 1} {$i <= 5} {incr i} {
r del [format "set%d{t}" $i] r del [format "set%d{t}" $i]
} }
...@@ -198,7 +244,7 @@ start_server { ...@@ -198,7 +244,7 @@ start_server {
# while the tests are running -- an extra element is added to every # while the tests are running -- an extra element is added to every
# set that determines its encoding. # set that determines its encoding.
set large 200 set large 200
if {$type eq "hashtable"} { if {$type eq "regular"} {
set large foo set large foo
} }
...@@ -206,9 +252,9 @@ start_server { ...@@ -206,9 +252,9 @@ start_server {
r sadd [format "set%d{t}" $i] $large r sadd [format "set%d{t}" $i] $large
} }
test "Generated sets must be encoded as $type" { test "Generated sets must be encoded correctly - $type" {
for {set i 1} {$i <= 5} {incr i} { for {set i 1} {$i <= 5} {incr i} {
assert_encoding $type [format "set%d{t}" $i] assert_encoding $encoding($i) [format "set%d{t}" $i]
} }
} }
...@@ -225,14 +271,14 @@ start_server { ...@@ -225,14 +271,14 @@ start_server {
test "SINTERSTORE with two sets - $type" { test "SINTERSTORE with two sets - $type" {
r sinterstore setres{t} set1{t} set2{t} r sinterstore setres{t} set1{t} set2{t}
assert_encoding $type setres{t} assert_encoding $smallenc setres{t}
assert_equal [list 195 196 197 198 199 $large] [lsort [r smembers setres{t}]] assert_equal [list 195 196 197 198 199 $large] [lsort [r smembers setres{t}]]
} }
test "SINTERSTORE with two sets, after a DEBUG RELOAD - $type" { test "SINTERSTORE with two sets, after a DEBUG RELOAD - $type" {
r debug reload r debug reload
r sinterstore setres{t} set1{t} set2{t} r sinterstore setres{t} set1{t} set2{t}
assert_encoding $type setres{t} assert_encoding $smallenc setres{t}
assert_equal [list 195 196 197 198 199 $large] [lsort [r smembers setres{t}]] assert_equal [list 195 196 197 198 199 $large] [lsort [r smembers setres{t}]]
} {} {needs:debug} } {} {needs:debug}
...@@ -243,7 +289,7 @@ start_server { ...@@ -243,7 +289,7 @@ start_server {
test "SUNIONSTORE with two sets - $type" { test "SUNIONSTORE with two sets - $type" {
r sunionstore setres{t} set1{t} set2{t} r sunionstore setres{t} set1{t} set2{t}
assert_encoding $type setres{t} assert_encoding $bigenc setres{t}
set expected [lsort -uniq "[r smembers set1{t}] [r smembers set2{t}]"] set expected [lsort -uniq "[r smembers set1{t}] [r smembers set2{t}]"]
assert_equal $expected [lsort [r smembers setres{t}]] assert_equal $expected [lsort [r smembers setres{t}]]
} }
...@@ -294,6 +340,46 @@ start_server { ...@@ -294,6 +340,46 @@ start_server {
} }
} }
test "SINTERSTORE with two listpack sets where result is intset" {
r del setres{t} set1{t} set2{t}
r sadd set1{t} a b c 1 3 6 x y z
r sadd set2{t} e f g 1 2 3 u v w
assert_encoding listpack set1{t}
assert_encoding listpack set2{t}
r sinterstore setres{t} set1{t} set2{t}
assert_equal [list 1 3] [lsort [r smembers setres{t}]]
assert_encoding intset setres{t}
}
test "SINTERSTORE with two hashtable sets where result is intset" {
r del setres{t} set1{t} set2{t}
r sadd set1{t} a b c 444 555 666
r sadd set2{t} e f g 111 222 333
set expected {}
for {set i 1} {$i < 130} {incr i} {
r sadd set1{t} $i
r sadd set2{t} $i
lappend expected $i
}
assert_encoding hashtable set1{t}
assert_encoding hashtable set2{t}
r sinterstore setres{t} set1{t} set2{t}
assert_equal [lsort $expected] [lsort [r smembers setres{t}]]
assert_encoding intset setres{t}
}
test "SUNION hashtable and listpack" {
# This adds code coverage for adding a non-sds string to a hashtable set
# which already contains the string.
r del set1{t} set2{t}
set union {abcdefghijklmnopqrstuvwxyz1234567890 a b c 1 2 3}
create_set set1{t} $union
create_set set2{t} {a b c}
assert_encoding hashtable set1{t}
assert_encoding listpack set2{t}
assert_equal [lsort $union] [lsort [r sunion set1{t} set2{t}]]
}
test "SDIFF with first set empty" { test "SDIFF with first set empty" {
r del set1{t} set2{t} set3{t} r del set1{t} set2{t} set3{t}
r sadd set2{t} 1 2 3 4 r sadd set2{t} 1 2 3 4
...@@ -428,7 +514,7 @@ start_server { ...@@ -428,7 +514,7 @@ start_server {
r sadd set2{t} 1 2 3 a r sadd set2{t} 1 2 3 a
r srem set2{t} a r srem set2{t} a
assert_encoding intset set1{t} assert_encoding intset set1{t}
assert_encoding hashtable set2{t} assert_encoding listpack set2{t}
lsort [r sinter set1{t} set2{t}] lsort [r sinter set1{t} set2{t}]
} {1 2 3} } {1 2 3}
...@@ -549,7 +635,7 @@ start_server { ...@@ -549,7 +635,7 @@ start_server {
assert_equal 0 [r exists setres{t}] assert_equal 0 [r exists setres{t}]
} }
foreach {type contents} {hashtable {a b c} intset {1 2 3}} { foreach {type contents} {listpack {a b c} intset {1 2 3}} {
test "SPOP basics - $type" { test "SPOP basics - $type" {
create_set myset $contents create_set myset $contents
assert_encoding $type myset assert_encoding $type myset
...@@ -575,11 +661,20 @@ start_server { ...@@ -575,11 +661,20 @@ start_server {
} }
} }
test "SPOP integer from listpack set" {
create_set myset {a 1 2 3 4 5 6 7}
assert_encoding listpack myset
set a [r spop myset]
set b [r spop myset]
assert {[string is digit $a] || [string is digit $b]}
}
foreach {type contents} { foreach {type contents} {
hashtable {a b c d e f g h i j k l m n o p q r s t u v w x y z} listpack {a b c d e f g h i j k l m n o p q r s t u v w x y z}
intset {1 10 11 12 13 14 15 16 17 18 19 2 20 21 22 23 24 25 26 3 4 5 6 7 8 9} intset {1 10 11 12 13 14 15 16 17 18 19 2 20 21 22 23 24 25 26 3 4 5 6 7 8 9}
hashtable {ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789 b c d e f g h i j k l m n o p q r s t u v w x y z}
} { } {
test "SPOP with <count>" { test "SPOP with <count> - $type" {
create_set myset $contents create_set myset $contents
assert_encoding $type myset assert_encoding $type myset
assert_equal $contents [lsort [concat [r spop myset 11] [r spop myset 9] [r spop myset 0] [r spop myset 4] [r spop myset 1] [r spop myset 0] [r spop myset 1] [r spop myset 0]]] assert_equal $contents [lsort [concat [r spop myset 11] [r spop myset 9] [r spop myset 0] [r spop myset 4] [r spop myset 1] [r spop myset 0] [r spop myset 1] [r spop myset 0]]]
...@@ -610,16 +705,20 @@ start_server { ...@@ -610,16 +705,20 @@ start_server {
r spop nonexisting_key 100 r spop nonexisting_key 100
} {} } {}
test "SPOP new implementation: code path #1" { foreach {type content} {
set content {1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20} intset {1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20}
listpack {a 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20}
} {
test "SPOP new implementation: code path #1 $type" {
create_set myset $content create_set myset $content
assert_encoding $type myset
set res [r spop myset 30] set res [r spop myset 30]
assert {[lsort $content] eq [lsort $res]} assert {[lsort $content] eq [lsort $res]}
} }
test "SPOP new implementation: code path #2" { test "SPOP new implementation: code path #2 $type" {
set content {1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20}
create_set myset $content create_set myset $content
assert_encoding $type myset
set res [r spop myset 2] set res [r spop myset 2]
assert {[llength $res] == 2} assert {[llength $res] == 2}
assert {[r scard myset] == 18} assert {[r scard myset] == 18}
...@@ -627,15 +726,16 @@ start_server { ...@@ -627,15 +726,16 @@ start_server {
assert {[lsort $union] eq [lsort $content]} assert {[lsort $union] eq [lsort $content]}
} }
test "SPOP new implementation: code path #3" { test "SPOP new implementation: code path #3 $type" {
set content {1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20}
create_set myset $content create_set myset $content
assert_encoding $type myset
set res [r spop myset 18] set res [r spop myset 18]
assert {[llength $res] == 18} assert {[llength $res] == 18}
assert {[r scard myset] == 2} assert {[r scard myset] == 2}
set union [concat [r smembers myset] $res] set union [concat [r smembers myset] $res]
assert {[lsort $union] eq [lsort $content]} assert {[lsort $union] eq [lsort $content]}
} }
}
test "SRANDMEMBER count of 0 is handled correctly" { test "SRANDMEMBER count of 0 is handled correctly" {
r srandmember myset 0 r srandmember myset 0
...@@ -659,7 +759,7 @@ start_server { ...@@ -659,7 +759,7 @@ start_server {
r readraw 0 r readraw 0
foreach {type contents} { foreach {type contents} {
hashtable { listpack {
1 5 10 50 125 50000 33959417 4775547 65434162 1 5 10 50 125 50000 33959417 4775547 65434162
12098459 427716 483706 2726473884 72615637475 12098459 427716 483706 2726473884 72615637475
MARY PATRICIA LINDA BARBARA ELIZABETH JENNIFER MARIA MARY PATRICIA LINDA BARBARA ELIZABETH JENNIFER MARIA
...@@ -674,9 +774,20 @@ start_server { ...@@ -674,9 +774,20 @@ start_server {
30 31 32 33 34 35 36 37 38 39 30 31 32 33 34 35 36 37 38 39
40 41 42 43 44 45 46 47 48 49 40 41 42 43 44 45 46 47 48 49
} }
hashtable {
ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789
1 5 10 50 125 50000 33959417 4775547 65434162
12098459 427716 483706 2726473884 72615637475
MARY PATRICIA LINDA BARBARA ELIZABETH JENNIFER MARIA
SUSAN MARGARET DOROTHY LISA NANCY KAREN BETTY HELEN
SANDRA DONNA CAROL RUTH SHARON MICHELLE LAURA SARAH
KIMBERLY DEBORAH JESSICA SHIRLEY CYNTHIA ANGELA MELISSA
BRENDA AMY ANNA REBECCA VIRGINIA
}
} { } {
test "SRANDMEMBER with <count> - $type" { test "SRANDMEMBER with <count> - $type" {
create_set myset $contents create_set myset $contents
assert_encoding $type myset
unset -nocomplain myset unset -nocomplain myset
array set myset {} array set myset {}
foreach ele [r smembers myset] { foreach ele [r smembers myset] {
...@@ -767,16 +878,22 @@ start_server { ...@@ -767,16 +878,22 @@ start_server {
} }
foreach {type contents} { foreach {type contents} {
hashtable { listpack {
1 5 10 50 125 1 5 10 50 125
MARY PATRICIA LINDA BARBARA ELIZABETH MARY PATRICIA LINDA BARBARA ELIZABETH
} }
intset { intset {
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9
} }
hashtable {
ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789
1 5 10 50 125
MARY PATRICIA LINDA BARBARA
}
} { } {
test "SRANDMEMBER histogram distribution - $type" { test "SRANDMEMBER histogram distribution - $type" {
create_set myset $contents create_set myset $contents
assert_encoding $type myset
unset -nocomplain myset unset -nocomplain myset
array set myset {} array set myset {}
foreach ele [r smembers myset] { foreach ele [r smembers myset] {
...@@ -809,7 +926,7 @@ start_server { ...@@ -809,7 +926,7 @@ start_server {
r del myset3{t} myset4{t} r del myset3{t} myset4{t}
create_set myset1{t} {1 a b} create_set myset1{t} {1 a b}
create_set myset2{t} {2 3 4} create_set myset2{t} {2 3 4}
assert_encoding hashtable myset1{t} assert_encoding listpack myset1{t}
assert_encoding intset myset2{t} assert_encoding intset myset2{t}
} }
...@@ -819,7 +936,7 @@ start_server { ...@@ -819,7 +936,7 @@ start_server {
assert_equal 1 [r smove myset1{t} myset2{t} a] assert_equal 1 [r smove myset1{t} myset2{t} a]
assert_equal {1 b} [lsort [r smembers myset1{t}]] assert_equal {1 b} [lsort [r smembers myset1{t}]]
assert_equal {2 3 4 a} [lsort [r smembers myset2{t}]] assert_equal {2 3 4 a} [lsort [r smembers myset2{t}]]
assert_encoding hashtable myset2{t} assert_encoding listpack myset2{t}
# move an integer element should not convert the encoding # move an integer element should not convert the encoding
setup_move setup_move
...@@ -855,7 +972,7 @@ start_server { ...@@ -855,7 +972,7 @@ start_server {
assert_equal 1 [r smove myset1{t} myset3{t} a] assert_equal 1 [r smove myset1{t} myset3{t} a]
assert_equal {1 b} [lsort [r smembers myset1{t}]] assert_equal {1 b} [lsort [r smembers myset1{t}]]
assert_equal {a} [lsort [r smembers myset3{t}]] assert_equal {a} [lsort [r smembers myset3{t}]]
assert_encoding hashtable myset3{t} assert_encoding listpack myset3{t}
} }
test "SMOVE from intset to non existing destination set" { test "SMOVE from intset to non existing destination set" {
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment