Unverified Commit 3ca6972e authored by sundb's avatar sundb Committed by GitHub
Browse files

Replace all usage of ziplist with listpack for t_zset (#9366)

Part two of implementing #8702 (zset), after #8887.

## Description of the feature
Replaced all uses of ziplist with listpack in t_zset, and optimized some of the code to optimize performance.

## Rdb format changes
New `RDB_TYPE_ZSET_LISTPACK` rdb type.

## Rdb loading improvements:
1) Pre-expansion of dict for validation of duplicate data for listpack and ziplist.
2) Simplifying the release of empty key objects when RDB loading.
3) Unify ziplist and listpack data verify methods for zset and hash, and move code to rdb.c.

## Interface changes
1) New `zset-max-listpack-entries` config is an alias for `zset-max-ziplist-entries` (same with `zset-max-listpack-value`).
2) OBJECT ENCODING will return listpack instead of ziplist.

## Listpack improvements:
1) Add `lpDeleteRange` and `lpDeleteRangeWithEntry` functions to delete a range of entries from listpack.
2) Improve the performance of `lpCompare`, converting from string to integer is faster than converting from integer to string.
3) Replace `snprintf` with `ll2string` to improve performance in converting numbers to strings in `lpGet()`.

## Zset improvements:
1) Improve the performance of `zzlFind` method, use `lpFind` instead of `lpCompare` in a loop.
2) Use `lpDeleteRangeWithEntry` instead of `lpDelete` twice to delete a element of zset.

## Tests
1) Add some unittests for `lpDeleteRange` and `lpDeleteRangeWithEntry` function.
2) Add zset RDB loading test.
3) Add benchmark test for `lpCompare` and `ziplsitCompare`.
4) Add empty listpack zset corrupt dump test.
parent 86b0de5c
...@@ -1748,8 +1748,8 @@ set-max-intset-entries 512 ...@@ -1748,8 +1748,8 @@ set-max-intset-entries 512
# 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:
zset-max-ziplist-entries 128 zset-max-listpack-entries 128
zset-max-ziplist-value 64 zset-max-listpack-value 64
# HyperLogLog sparse representation bytes limit. The limit includes the # HyperLogLog sparse representation bytes limit. The limit includes the
# 16 bytes header. When an HyperLogLog using the sparse representation crosses # 16 bytes header. When an HyperLogLog using the sparse representation crosses
......
...@@ -1023,7 +1023,7 @@ int rewriteSetObject(rio *r, robj *key, robj *o) { ...@@ -1023,7 +1023,7 @@ int rewriteSetObject(rio *r, robj *key, robj *o) {
int rewriteSortedSetObject(rio *r, robj *key, robj *o) { int rewriteSortedSetObject(rio *r, robj *key, robj *o) {
long long count = 0, items = zsetLength(o); long long count = 0, items = zsetLength(o);
if (o->encoding == OBJ_ENCODING_ZIPLIST) { if (o->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *zl = o->ptr; unsigned char *zl = o->ptr;
unsigned char *eptr, *sptr; unsigned char *eptr, *sptr;
unsigned char *vstr; unsigned char *vstr;
...@@ -1031,13 +1031,13 @@ int rewriteSortedSetObject(rio *r, robj *key, robj *o) { ...@@ -1031,13 +1031,13 @@ int rewriteSortedSetObject(rio *r, robj *key, robj *o) {
long long vll; long long vll;
double score; double score;
eptr = ziplistIndex(zl,0); eptr = lpSeek(zl,0);
serverAssert(eptr != NULL); serverAssert(eptr != NULL);
sptr = ziplistNext(zl,eptr); sptr = lpNext(zl,eptr);
serverAssert(sptr != NULL); serverAssert(sptr != NULL);
while (eptr != NULL) { while (eptr != NULL) {
serverAssert(ziplistGet(eptr,&vstr,&vlen,&vll)); vstr = lpGetValue(eptr,&vlen,&vll);
score = zzlGetScore(sptr); score = zzlGetScore(sptr);
if (count == 0) { if (count == 0) {
......
...@@ -2645,11 +2645,11 @@ standardConfig configs[] = { ...@@ -2645,11 +2645,11 @@ standardConfig 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("zset-max-ziplist-entries", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.zset_max_ziplist_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),
createSizeTConfig("stream-node-max-bytes", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.stream_node_max_bytes, 4096, MEMORY_CONFIG, NULL, NULL), createSizeTConfig("stream-node-max-bytes", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.stream_node_max_bytes, 4096, MEMORY_CONFIG, NULL, NULL),
createSizeTConfig("zset-max-ziplist-value", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.zset_max_ziplist_value, 64, MEMORY_CONFIG, NULL, NULL), createSizeTConfig("zset-max-listpack-value", "zset-max-ziplist-value", MODIFIABLE_CONFIG, 0, LONG_MAX, server.zset_max_listpack_value, 64, MEMORY_CONFIG, NULL, NULL),
createSizeTConfig("hll-sparse-max-bytes", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.hll_sparse_max_bytes, 3000, MEMORY_CONFIG, NULL, NULL), createSizeTConfig("hll-sparse-max-bytes", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.hll_sparse_max_bytes, 3000, MEMORY_CONFIG, NULL, NULL),
createSizeTConfig("tracking-table-max-keys", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.tracking_table_max_keys, 1000000, INTEGER_CONFIG, NULL, NULL), /* Default: 1 million keys max. */ createSizeTConfig("tracking-table-max-keys", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.tracking_table_max_keys, 1000000, INTEGER_CONFIG, NULL, NULL), /* Default: 1 million keys max. */
createSizeTConfig("client-query-buffer-limit", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, 1024*1024, LONG_MAX, server.client_max_querybuf_len, 1024*1024*1024, MEMORY_CONFIG, NULL, NULL), /* Default: 1GB max query buffer. */ createSizeTConfig("client-query-buffer-limit", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, 1024*1024, LONG_MAX, server.client_max_querybuf_len, 1024*1024*1024, MEMORY_CONFIG, NULL, NULL), /* Default: 1GB max query buffer. */
......
...@@ -914,21 +914,7 @@ void scanGenericCommand(client *c, robj *o, unsigned long cursor) { ...@@ -914,21 +914,7 @@ void scanGenericCommand(client *c, robj *o, unsigned long cursor) {
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_ZSET) { } else if (o->type == OBJ_HASH || o->type == OBJ_ZSET) {
unsigned char *p = ziplistIndex(o->ptr,0);
unsigned char *vstr;
unsigned int vlen;
long long vll;
while(p) {
ziplistGet(p,&vstr,&vlen,&vll);
listAddNodeTail(keys,
(vstr != NULL) ? createStringObject((char*)vstr,vlen) :
createStringObjectFromLongLong(vll));
p = ziplistNext(o->ptr,p);
}
cursor = 0;
} else if (o->type == OBJ_HASH) {
unsigned char *p = lpFirst(o->ptr); unsigned char *p = lpFirst(o->ptr);
unsigned char *vstr; unsigned char *vstr;
int64_t vlen; int64_t vlen;
......
...@@ -162,7 +162,7 @@ void xorObjectDigest(redisDb *db, robj *keyobj, unsigned char *digest, robj *o) ...@@ -162,7 +162,7 @@ void xorObjectDigest(redisDb *db, robj *keyobj, unsigned char *digest, robj *o)
} else if (o->type == OBJ_ZSET) { } else if (o->type == OBJ_ZSET) {
unsigned char eledigest[20]; unsigned char eledigest[20];
if (o->encoding == OBJ_ENCODING_ZIPLIST) { if (o->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *zl = o->ptr; unsigned char *zl = o->ptr;
unsigned char *eptr, *sptr; unsigned char *eptr, *sptr;
unsigned char *vstr; unsigned char *vstr;
...@@ -170,13 +170,13 @@ void xorObjectDigest(redisDb *db, robj *keyobj, unsigned char *digest, robj *o) ...@@ -170,13 +170,13 @@ void xorObjectDigest(redisDb *db, robj *keyobj, unsigned char *digest, robj *o)
long long vll; long long vll;
double score; double score;
eptr = ziplistIndex(zl,0); eptr = lpSeek(zl,0);
serverAssert(eptr != NULL); serverAssert(eptr != NULL);
sptr = ziplistNext(zl,eptr); sptr = lpNext(zl,eptr);
serverAssert(sptr != NULL); serverAssert(sptr != NULL);
while (eptr != NULL) { while (eptr != NULL) {
serverAssert(ziplistGet(eptr,&vstr,&vlen,&vll)); vstr = lpGetValue(eptr,&vlen,&vll);
score = zzlGetScore(sptr); score = zzlGetScore(sptr);
memset(eledigest,0,20); memset(eledigest,0,20);
......
...@@ -856,7 +856,7 @@ long defragKey(redisDb *db, dictEntry *de) { ...@@ -856,7 +856,7 @@ long defragKey(redisDb *db, dictEntry *de) {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
} }
} else if (ob->type == OBJ_ZSET) { } else if (ob->type == OBJ_ZSET) {
if (ob->encoding == OBJ_ENCODING_ZIPLIST) { if (ob->encoding == OBJ_ENCODING_LISTPACK) {
if ((newzl = activeDefragAlloc(ob->ptr))) if ((newzl = activeDefragAlloc(ob->ptr)))
defragged++, ob->ptr = newzl; defragged++, ob->ptr = newzl;
} else if (ob->encoding == OBJ_ENCODING_SKIPLIST) { } else if (ob->encoding == OBJ_ENCODING_SKIPLIST) {
......
...@@ -259,7 +259,7 @@ int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geo ...@@ -259,7 +259,7 @@ int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geo
size_t origincount = ga->used; size_t origincount = ga->used;
sds member; sds member;
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) { if (zobj->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *zl = zobj->ptr; unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr; unsigned char *eptr, *sptr;
unsigned char *vstr = NULL; unsigned char *vstr = NULL;
...@@ -272,7 +272,7 @@ int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geo ...@@ -272,7 +272,7 @@ int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geo
return 0; return 0;
} }
sptr = ziplistNext(zl, eptr); sptr = lpNext(zl, eptr);
while (eptr) { while (eptr) {
score = zzlGetScore(sptr); score = zzlGetScore(sptr);
...@@ -280,8 +280,7 @@ int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geo ...@@ -280,8 +280,7 @@ int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geo
if (!zslValueLteMax(score, &range)) if (!zslValueLteMax(score, &range))
break; break;
/* We know the element exists. ziplistGet should always succeed */ vstr = lpGetValue(eptr, &vlen, &vlong);
ziplistGet(eptr, &vstr, &vlen, &vlong);
member = (vstr == NULL) ? sdsfromlonglong(vlong) : member = (vstr == NULL) ? sdsfromlonglong(vlong) :
sdsnewlen(vstr,vlen); sdsnewlen(vstr,vlen);
if (geoAppendIfWithinShape(ga,shape,score,member) if (geoAppendIfWithinShape(ga,shape,score,member)
...@@ -819,7 +818,7 @@ void georadiusGeneric(client *c, int srcKeyIndex, int flags) { ...@@ -819,7 +818,7 @@ void georadiusGeneric(client *c, int srcKeyIndex, int flags) {
} }
if (returned_items) { if (returned_items) {
zsetConvertToZiplistIfNeeded(zobj,maxelelen); zsetConvertToListpackIfNeeded(zobj,maxelelen);
setKey(c,c->db,storekey,zobj); setKey(c,c->db,storekey,zobj);
decrRefCount(zobj); decrRefCount(zobj);
notifyKeyspaceEvent(NOTIFY_ZSET,flags & GEOSEARCH ? "geosearchstore" : "georadiusstore",storekey, notifyKeyspaceEvent(NOTIFY_ZSET,flags & GEOSEARCH ? "geosearchstore" : "georadiusstore",storekey,
......
...@@ -43,6 +43,7 @@ ...@@ -43,6 +43,7 @@
#include "listpack.h" #include "listpack.h"
#include "listpack_malloc.h" #include "listpack_malloc.h"
#include "redisassert.h" #include "redisassert.h"
#include "util.h"
#define LP_HDR_SIZE 6 /* 32 bit total len + 16 bit number of elements. */ #define LP_HDR_SIZE 6 /* 32 bit total len + 16 bit number of elements. */
#define LP_HDR_NUMELE_UNKNOWN UINT16_MAX #define LP_HDR_NUMELE_UNKNOWN UINT16_MAX
...@@ -642,7 +643,7 @@ static inline unsigned char *lpGetWithSize(unsigned char *p, int64_t *count, uns ...@@ -642,7 +643,7 @@ static inline unsigned char *lpGetWithSize(unsigned char *p, int64_t *count, uns
/* Return the string representation of the integer or the value itself /* Return the string representation of the integer or the value itself
* depending on intbuf being NULL or not. */ * depending on intbuf being NULL or not. */
if (intbuf) { if (intbuf) {
*count = snprintf((char*)intbuf,LP_INTBUF_SIZE,"%lld",(long long)val); *count = ll2string((char*)intbuf,LP_INTBUF_SIZE,(long long)val);
return intbuf; return intbuf;
} else { } else {
*count = val; *count = val;
...@@ -909,6 +910,13 @@ unsigned char *lpInsert(unsigned char *lp, unsigned char *elestr, unsigned char ...@@ -909,6 +910,13 @@ unsigned char *lpInsert(unsigned char *lp, unsigned char *elestr, unsigned char
return lp; return lp;
} }
/* This is just a wrapper for lpInsert() to directly use a string. */
unsigned char *lpInsertString(unsigned char *lp, unsigned char *s, uint32_t slen,
unsigned char *p, int where, unsigned char **newp)
{
return lpInsert(lp, s, NULL, slen, p, where, newp);
}
/* This is just a wrapper for lpInsert() to directly use a 64 bit integer /* This is just a wrapper for lpInsert() to directly use a 64 bit integer
* instead of a string. */ * instead of a string. */
unsigned char *lpInsertInteger(unsigned char *lp, long long lval, unsigned char *p, int where, unsigned char **newp) { unsigned char *lpInsertInteger(unsigned char *lp, long long lval, unsigned char *p, int where, unsigned char **newp) {
...@@ -972,6 +980,73 @@ unsigned char *lpDelete(unsigned char *lp, unsigned char *p, unsigned char **new ...@@ -972,6 +980,73 @@ unsigned char *lpDelete(unsigned char *lp, unsigned char *p, unsigned char **new
return lpInsert(lp,NULL,NULL,0,p,LP_REPLACE,newp); return lpInsert(lp,NULL,NULL,0,p,LP_REPLACE,newp);
} }
/* Delete a range of entries from the listpack start with the element pointed by 'p'. */
unsigned char *lpDeleteRangeWithEntry(unsigned char *lp, unsigned char **p, unsigned long num) {
size_t bytes = lpBytes(lp);
unsigned long deleted = 0;
unsigned char *eofptr = lp + bytes - 1;
unsigned char *first, *tail;
first = tail = *p;
if (num == 0) return lp; /* Nothing to delete, return ASAP. */
/* Find the next entry to the last entry that needs to be deleted.
* lpLength may be unreliable due to corrupt data, so we cannot
* treat 'num' as the number of elements to be deleted. */
while (num--) {
deleted++;
tail = lpSkip(tail);
if (tail[0] == LP_EOF) break;
lpAssertValidEntry(lp, bytes, tail);
}
/* Store the offset of the element 'first', so that we can obtain its
* address again after a reallocation. */
unsigned long poff = first-lp;
/* Move tail to the front of the ziplist */
memmove(first, tail, eofptr - tail + 1);
lpSetTotalBytes(lp, bytes - (tail - first));
uint32_t numele = lpGetNumElements(lp);
if (numele != LP_HDR_NUMELE_UNKNOWN)
lpSetNumElements(lp, numele-deleted);
lp = lpShrinkToFit(lp);
/* Store the entry. */
*p = lp+poff;
if ((*p)[0] == LP_EOF) *p = NULL;
return lp;
}
/* Delete a range of entries from the listpack. */
unsigned char *lpDeleteRange(unsigned char *lp, long index, unsigned long num) {
unsigned char *p;
uint32_t numele = lpGetNumElements(lp);
if (num == 0) return lp; /* Nothing to delete, return ASAP. */
if ((p = lpSeek(lp, index)) == NULL) return lp;
/* If we know we're gonna delete beyond the end of the listpack, we can just move
* the EOF marker, and there's no need to iterate through the entries,
* but if we can't be sure how many entries there are, we rather avoid calling lpLength
* since that means an additional iteration on all elements.
*
* Note that index could overflow, but we use the value after seek, so when we
* use it no overflow happens. */
if (numele != LP_HDR_NUMELE_UNKNOWN && index < 0) index = (long)numele + index;
if (numele != LP_HDR_NUMELE_UNKNOWN && (numele - (unsigned long)index) <= num) {
p[0] = LP_EOF;
lpSetTotalBytes(lp, p - lp + 1);
lpSetNumElements(lp, index);
lp = lpShrinkToFit(lp);
} else {
lp = lpDeleteRangeWithEntry(lp, &p, num);
}
return lp;
}
/* Return the total number of bytes the listpack is composed of. */ /* Return the total number of bytes the listpack is composed of. */
size_t lpBytes(unsigned char *lp) { size_t lpBytes(unsigned char *lp) {
return lpGetTotalBytes(lp); return lpGetTotalBytes(lp);
...@@ -1112,6 +1187,7 @@ int lpValidateIntegrity(unsigned char *lp, size_t size, int deep, ...@@ -1112,6 +1187,7 @@ int lpValidateIntegrity(unsigned char *lp, size_t size, int deep,
/* Validate the individual entries. */ /* Validate the individual entries. */
uint32_t count = 0; uint32_t count = 0;
uint32_t numele = lpGetNumElements(lp);
unsigned char *p = lp + LP_HDR_SIZE; unsigned char *p = lp + LP_HDR_SIZE;
while(p && p[0] != LP_EOF) { while(p && p[0] != LP_EOF) {
unsigned char *prev = p; unsigned char *prev = p;
...@@ -1122,28 +1198,39 @@ int lpValidateIntegrity(unsigned char *lp, size_t size, int deep, ...@@ -1122,28 +1198,39 @@ int lpValidateIntegrity(unsigned char *lp, size_t size, int deep,
return 0; return 0;
/* Optionally let the caller validate the entry too. */ /* Optionally let the caller validate the entry too. */
if (entry_cb && !entry_cb(prev, cb_userdata)) if (entry_cb && !entry_cb(prev, numele, cb_userdata))
return 0; return 0;
count++; count++;
} }
/* Check that the count in the header is correct */ /* Check that the count in the header is correct */
uint32_t numele = lpGetNumElements(lp);
if (numele != LP_HDR_NUMELE_UNKNOWN && numele != count) if (numele != LP_HDR_NUMELE_UNKNOWN && numele != count)
return 0; return 0;
return 1; return 1;
} }
/* Compare entry pointer to by 'p' with string 's' of length 'slen'.
* Return 1 if equal. */
unsigned int lpCompare(unsigned char *p, unsigned char *s, uint32_t slen) { unsigned int lpCompare(unsigned char *p, unsigned char *s, uint32_t slen) {
unsigned char buf[LP_INTBUF_SIZE];
unsigned char *value; unsigned char *value;
int64_t sz; int64_t sz;
if (p[0] == LP_EOF) return 0; if (p[0] == LP_EOF) return 0;
value = lpGet(p, &sz, buf);
return (slen == sz) && memcmp(value,s,slen) == 0; value = lpGet(p, &sz, NULL);
if (value) {
return (slen == sz) && memcmp(value,s,slen) == 0;
} else {
/* We use lpStringToInt64() to get an integer representation of the
* string 's' and compare it to 'sval', it's much faster than convert
* integer to string and comparing. */
int64_t sval;
if (lpStringToInt64((const char*)s, slen, &sval))
return sz == sval;
}
return 0;
} }
/* uint compare for qsort */ /* uint compare for qsort */
...@@ -1391,8 +1478,9 @@ static void verifyEntry(unsigned char *p, unsigned char *s, size_t slen) { ...@@ -1391,8 +1478,9 @@ static void verifyEntry(unsigned char *p, unsigned char *s, size_t slen) {
assert(lpCompare(p, s, slen)); assert(lpCompare(p, s, slen));
} }
static int lpValidation(unsigned char *p, void *userdata) { static int lpValidation(unsigned char *p, unsigned int head_count, void *userdata) {
UNUSED(p); UNUSED(p);
UNUSED(head_count);
int ret; int ret;
long *count = userdata; long *count = userdata;
...@@ -1537,6 +1625,114 @@ int listpackTest(int argc, char *argv[], int accurate) { ...@@ -1537,6 +1625,114 @@ int listpackTest(int argc, char *argv[], int accurate) {
lpFree(lp); lpFree(lp);
} }
TEST("Delete whole listpack when num == -1");
{
lp = createList();
lp = lpDeleteRange(lp, 0, -1);
assert(lpLength(lp) == 0);
assert(lp[LP_HDR_SIZE] == LP_EOF);
assert(lpBytes(lp) == (LP_HDR_SIZE + 1));
zfree(lp);
lp = createList();
unsigned char *ptr = lpFirst(lp);
lp = lpDeleteRangeWithEntry(lp, &ptr, -1);
assert(lpLength(lp) == 0);
assert(lp[LP_HDR_SIZE] == LP_EOF);
assert(lpBytes(lp) == (LP_HDR_SIZE + 1));
zfree(lp);
}
TEST("Delete whole listpack with negative index");
{
lp = createList();
lp = lpDeleteRange(lp, -4, 4);
assert(lpLength(lp) == 0);
assert(lp[LP_HDR_SIZE] == LP_EOF);
assert(lpBytes(lp) == (LP_HDR_SIZE + 1));
zfree(lp);
lp = createList();
unsigned char *ptr = lpSeek(lp, -4);
lp = lpDeleteRangeWithEntry(lp, &ptr, 4);
assert(lpLength(lp) == 0);
assert(lp[LP_HDR_SIZE] == LP_EOF);
assert(lpBytes(lp) == (LP_HDR_SIZE + 1));
zfree(lp);
}
TEST("Delete inclusive range 0,0");
{
lp = createList();
lp = lpDeleteRange(lp, 0, 1);
assert(lpLength(lp) == 3);
assert(lpSkip(lpLast(lp))[0] == LP_EOF); /* check set LP_EOF correctly */
zfree(lp);
lp = createList();
unsigned char *ptr = lpFirst(lp);
lp = lpDeleteRangeWithEntry(lp, &ptr, 1);
assert(lpLength(lp) == 3);
assert(lpSkip(lpLast(lp))[0] == LP_EOF); /* check set LP_EOF correctly */
zfree(lp);
}
TEST("Delete inclusive range 0,1");
{
lp = createList();
lp = lpDeleteRange(lp, 0, 2);
assert(lpLength(lp) == 2);
verifyEntry(lpFirst(lp), (unsigned char*)mixlist[2], strlen(mixlist[2]));
zfree(lp);
lp = createList();
unsigned char *ptr = lpFirst(lp);
lp = lpDeleteRangeWithEntry(lp, &ptr, 2);
assert(lpLength(lp) == 2);
verifyEntry(lpFirst(lp), (unsigned char*)mixlist[2], strlen(mixlist[2]));
zfree(lp);
}
TEST("Delete inclusive range 1,2");
{
lp = createList();
lp = lpDeleteRange(lp, 1, 2);
assert(lpLength(lp) == 2);
verifyEntry(lpFirst(lp), (unsigned char*)mixlist[0], strlen(mixlist[0]));
zfree(lp);
lp = createList();
unsigned char *ptr = lpSeek(lp, 1);
lp = lpDeleteRangeWithEntry(lp, &ptr, 2);
assert(lpLength(lp) == 2);
verifyEntry(lpFirst(lp), (unsigned char*)mixlist[0], strlen(mixlist[0]));
zfree(lp);
}
TEST("Delete with start index out of range");
{
lp = createList();
lp = lpDeleteRange(lp, 5, 1);
assert(lpLength(lp) == 4);
zfree(lp);
}
TEST("Delete with num overflow");
{
lp = createList();
lp = lpDeleteRange(lp, 1, 5);
assert(lpLength(lp) == 1);
verifyEntry(lpFirst(lp), (unsigned char*)mixlist[0], strlen(mixlist[0]));
zfree(lp);
lp = createList();
unsigned char *ptr = lpSeek(lp, 1);
lp = lpDeleteRangeWithEntry(lp, &ptr, 5);
assert(lpLength(lp) == 1);
verifyEntry(lpFirst(lp), (unsigned char*)mixlist[0], strlen(mixlist[0]));
zfree(lp);
}
TEST("Delete foo while iterating") { TEST("Delete foo while iterating") {
lp = createList(); lp = createList();
p = lpFirst(lp); p = lpFirst(lp);
...@@ -1817,6 +2013,21 @@ int listpackTest(int argc, char *argv[], int accurate) { ...@@ -1817,6 +2013,21 @@ int listpackTest(int argc, char *argv[], int accurate) {
lpFree(lp); lpFree(lp);
} }
TEST("Test number of elements exceeds LP_HDR_NUMELE_UNKNOWN") {
lp = lpNew(0);
for (int i = 0; i < LP_HDR_NUMELE_UNKNOWN + 1; i++)
lp = lpAppend(lp, (unsigned char*)"1", 1);
assert(lpGetNumElements(lp) == LP_HDR_NUMELE_UNKNOWN);
assert(lpLength(lp) == LP_HDR_NUMELE_UNKNOWN+1);
lp = lpDeleteRange(lp, -2, 2);
assert(lpGetNumElements(lp) == LP_HDR_NUMELE_UNKNOWN);
assert(lpLength(lp) == LP_HDR_NUMELE_UNKNOWN-1);
assert(lpGetNumElements(lp) == LP_HDR_NUMELE_UNKNOWN-1); /* update length after lpLength */
lpFree(lp);
}
TEST("Stress with random payloads of different encoding") { TEST("Stress with random payloads of different encoding") {
unsigned long long start = usec(); unsigned long long start = usec();
int i,j,len,where; int i,j,len,where;
...@@ -1951,6 +2162,30 @@ int listpackTest(int argc, char *argv[], int accurate) { ...@@ -1951,6 +2162,30 @@ int listpackTest(int argc, char *argv[], int accurate) {
printf("Done. usec=%lld\n", usec()-start); printf("Done. usec=%lld\n", usec()-start);
} }
TEST("Benchmark lpCompare with string") {
unsigned long long start = usec();
for (int i = 0; i < 2000; i++) {
unsigned char *eptr = lpSeek(lp,0);
while (eptr != NULL) {
lpCompare(eptr,(unsigned char*)"nothing",7);
eptr = lpNext(lp,eptr);
}
}
printf("Done. usec=%lld\n", usec()-start);
}
TEST("Benchmark lpCompare with number") {
unsigned long long start = usec();
for (int i = 0; i < 2000; i++) {
unsigned char *eptr = lpSeek(lp,0);
while (eptr != NULL) {
lpCompare(lp, (unsigned char*)"99999", 5);
eptr = lpNext(lp,eptr);
}
}
printf("Done. usec=%lld\n", usec()-start);
}
lpFree(lp); lpFree(lp);
} }
......
...@@ -57,6 +57,8 @@ typedef struct { ...@@ -57,6 +57,8 @@ typedef struct {
unsigned char *lpNew(size_t capacity); unsigned char *lpNew(size_t capacity);
void lpFree(unsigned char *lp); void lpFree(unsigned char *lp);
unsigned char* lpShrinkToFit(unsigned char *lp); unsigned char* lpShrinkToFit(unsigned char *lp);
unsigned char *lpInsertString(unsigned char *lp, unsigned char *s, uint32_t slen,
unsigned char *p, int where, unsigned char **newp);
unsigned char *lpPrepend(unsigned char *lp, unsigned char *s, uint32_t slen); unsigned char *lpPrepend(unsigned char *lp, unsigned char *s, uint32_t slen);
unsigned char *lpPrependInteger(unsigned char *lp, long long lval); unsigned char *lpPrependInteger(unsigned char *lp, long long lval);
unsigned char *lpAppend(unsigned char *lp, unsigned char *s, uint32_t slen); unsigned char *lpAppend(unsigned char *lp, unsigned char *s, uint32_t slen);
...@@ -64,6 +66,8 @@ unsigned char *lpAppendInteger(unsigned char *lp, long long lval); ...@@ -64,6 +66,8 @@ unsigned char *lpAppendInteger(unsigned char *lp, long long lval);
unsigned char *lpReplace(unsigned char *lp, unsigned char **p, unsigned char *s, uint32_t slen); unsigned char *lpReplace(unsigned char *lp, unsigned char **p, unsigned char *s, uint32_t slen);
unsigned char *lpReplaceInteger(unsigned char *lp, unsigned char **p, long long lval); unsigned char *lpReplaceInteger(unsigned char *lp, unsigned char **p, long long lval);
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 *lpDeleteRange(unsigned char *lp, long index, unsigned long num);
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);
unsigned char *lpGetValue(unsigned char *p, unsigned int *slen, long long *lval); unsigned char *lpGetValue(unsigned char *p, unsigned int *slen, long long *lval);
...@@ -74,7 +78,7 @@ unsigned char *lpNext(unsigned char *lp, unsigned char *p); ...@@ -74,7 +78,7 @@ unsigned char *lpNext(unsigned char *lp, unsigned char *p);
unsigned char *lpPrev(unsigned char *lp, unsigned char *p); unsigned char *lpPrev(unsigned char *lp, unsigned char *p);
size_t lpBytes(unsigned char *lp); size_t lpBytes(unsigned char *lp);
unsigned char *lpSeek(unsigned char *lp, long index); unsigned char *lpSeek(unsigned char *lp, long index);
typedef int (*listpackValidateEntryCB)(unsigned char *p, void *userdata); typedef int (*listpackValidateEntryCB)(unsigned char *p, unsigned int head_count, void *userdata);
int lpValidateIntegrity(unsigned char *lp, size_t size, int deep, int lpValidateIntegrity(unsigned char *lp, size_t size, int deep,
listpackValidateEntryCB entry_cb, void *cb_userdata); listpackValidateEntryCB entry_cb, void *cb_userdata);
unsigned char *lpValidateFirst(unsigned char *lp); unsigned char *lpValidateFirst(unsigned char *lp);
......
...@@ -522,7 +522,7 @@ int moduleCreateEmptyKey(RedisModuleKey *key, int type) { ...@@ -522,7 +522,7 @@ int moduleCreateEmptyKey(RedisModuleKey *key, int type) {
server.list_compress_depth); server.list_compress_depth);
break; break;
case REDISMODULE_KEYTYPE_ZSET: case REDISMODULE_KEYTYPE_ZSET:
obj = createZsetZiplistObject(); obj = createZsetListpackObject();
break; break;
case REDISMODULE_KEYTYPE_HASH: case REDISMODULE_KEYTYPE_HASH:
obj = createHashObject(); obj = createHashObject();
...@@ -2966,7 +2966,7 @@ int zsetInitScoreRange(RedisModuleKey *key, double min, double max, int minex, i ...@@ -2966,7 +2966,7 @@ int zsetInitScoreRange(RedisModuleKey *key, double min, double max, int minex, i
zrs->minex = minex; zrs->minex = minex;
zrs->maxex = maxex; zrs->maxex = maxex;
if (key->value->encoding == OBJ_ENCODING_ZIPLIST) { if (key->value->encoding == OBJ_ENCODING_LISTPACK) {
key->u.zset.current = first ? zzlFirstInRange(key->value->ptr,zrs) : key->u.zset.current = first ? zzlFirstInRange(key->value->ptr,zrs) :
zzlLastInRange(key->value->ptr,zrs); zzlLastInRange(key->value->ptr,zrs);
} else if (key->value->encoding == OBJ_ENCODING_SKIPLIST) { } else if (key->value->encoding == OBJ_ENCODING_SKIPLIST) {
...@@ -3030,7 +3030,7 @@ int zsetInitLexRange(RedisModuleKey *key, RedisModuleString *min, RedisModuleStr ...@@ -3030,7 +3030,7 @@ int zsetInitLexRange(RedisModuleKey *key, RedisModuleString *min, RedisModuleStr
* otherwise we don't want the zlexrangespec to be freed. */ * otherwise we don't want the zlexrangespec to be freed. */
key->u.zset.type = REDISMODULE_ZSET_RANGE_LEX; key->u.zset.type = REDISMODULE_ZSET_RANGE_LEX;
if (key->value->encoding == OBJ_ENCODING_ZIPLIST) { if (key->value->encoding == OBJ_ENCODING_LISTPACK) {
key->u.zset.current = first ? zzlFirstInLexRange(key->value->ptr,zlrs) : key->u.zset.current = first ? zzlFirstInLexRange(key->value->ptr,zlrs) :
zzlLastInLexRange(key->value->ptr,zlrs); zzlLastInLexRange(key->value->ptr,zlrs);
} else if (key->value->encoding == OBJ_ENCODING_SKIPLIST) { } else if (key->value->encoding == OBJ_ENCODING_SKIPLIST) {
...@@ -3076,12 +3076,12 @@ RedisModuleString *RM_ZsetRangeCurrentElement(RedisModuleKey *key, double *score ...@@ -3076,12 +3076,12 @@ RedisModuleString *RM_ZsetRangeCurrentElement(RedisModuleKey *key, double *score
if (!key->value || key->value->type != OBJ_ZSET) return NULL; if (!key->value || key->value->type != OBJ_ZSET) return NULL;
if (key->u.zset.current == NULL) return NULL; if (key->u.zset.current == NULL) return NULL;
if (key->value->encoding == OBJ_ENCODING_ZIPLIST) { if (key->value->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *eptr, *sptr; unsigned char *eptr, *sptr;
eptr = key->u.zset.current; eptr = key->u.zset.current;
sds ele = ziplistGetObject(eptr); sds ele = lpGetObject(eptr);
if (score) { if (score) {
sptr = ziplistNext(key->value->ptr,eptr); sptr = lpNext(key->value->ptr,eptr);
*score = zzlGetScore(sptr); *score = zzlGetScore(sptr);
} }
str = createObject(OBJ_STRING,ele); str = createObject(OBJ_STRING,ele);
...@@ -3103,12 +3103,12 @@ int RM_ZsetRangeNext(RedisModuleKey *key) { ...@@ -3103,12 +3103,12 @@ int RM_ZsetRangeNext(RedisModuleKey *key) {
if (!key->value || key->value->type != OBJ_ZSET) return 0; if (!key->value || key->value->type != OBJ_ZSET) return 0;
if (!key->u.zset.type || !key->u.zset.current) return 0; /* No active iterator. */ if (!key->u.zset.type || !key->u.zset.current) return 0; /* No active iterator. */
if (key->value->encoding == OBJ_ENCODING_ZIPLIST) { if (key->value->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *zl = key->value->ptr; unsigned char *zl = key->value->ptr;
unsigned char *eptr = key->u.zset.current; unsigned char *eptr = key->u.zset.current;
unsigned char *next; unsigned char *next;
next = ziplistNext(zl,eptr); /* Skip element. */ next = lpNext(zl,eptr); /* Skip element. */
if (next) next = ziplistNext(zl,next); /* Skip score. */ if (next) next = lpNext(zl,next); /* Skip score. */
if (next == NULL) { if (next == NULL) {
key->u.zset.er = 1; key->u.zset.er = 1;
return 0; return 0;
...@@ -3118,7 +3118,7 @@ int RM_ZsetRangeNext(RedisModuleKey *key) { ...@@ -3118,7 +3118,7 @@ int RM_ZsetRangeNext(RedisModuleKey *key) {
/* Fetch the next element score for the /* Fetch the next element score for the
* range check. */ * range check. */
unsigned char *saved_next = next; unsigned char *saved_next = next;
next = ziplistNext(zl,next); /* Skip next element. */ next = lpNext(zl,next); /* Skip next element. */
double score = zzlGetScore(next); /* Obtain the next score. */ double score = zzlGetScore(next); /* Obtain the next score. */
if (!zslValueLteMax(score,&key->u.zset.rs)) { if (!zslValueLteMax(score,&key->u.zset.rs)) {
key->u.zset.er = 1; key->u.zset.er = 1;
...@@ -3167,12 +3167,12 @@ int RM_ZsetRangePrev(RedisModuleKey *key) { ...@@ -3167,12 +3167,12 @@ int RM_ZsetRangePrev(RedisModuleKey *key) {
if (!key->value || key->value->type != OBJ_ZSET) return 0; if (!key->value || key->value->type != OBJ_ZSET) return 0;
if (!key->u.zset.type || !key->u.zset.current) return 0; /* No active iterator. */ if (!key->u.zset.type || !key->u.zset.current) return 0; /* No active iterator. */
if (key->value->encoding == OBJ_ENCODING_ZIPLIST) { if (key->value->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *zl = key->value->ptr; unsigned char *zl = key->value->ptr;
unsigned char *eptr = key->u.zset.current; unsigned char *eptr = key->u.zset.current;
unsigned char *prev; unsigned char *prev;
prev = ziplistPrev(zl,eptr); /* Go back to previous score. */ prev = lpPrev(zl,eptr); /* Go back to previous score. */
if (prev) prev = ziplistPrev(zl,prev); /* Back to previous ele. */ if (prev) prev = lpPrev(zl,prev); /* Back to previous ele. */
if (prev == NULL) { if (prev == NULL) {
key->u.zset.er = 1; key->u.zset.er = 1;
return 0; return 0;
...@@ -3182,7 +3182,7 @@ int RM_ZsetRangePrev(RedisModuleKey *key) { ...@@ -3182,7 +3182,7 @@ int RM_ZsetRangePrev(RedisModuleKey *key) {
/* Fetch the previous element score for the /* Fetch the previous element score for the
* range check. */ * range check. */
unsigned char *saved_prev = prev; unsigned char *saved_prev = prev;
prev = ziplistNext(zl,prev); /* Skip element to get the score.*/ prev = lpNext(zl,prev); /* Skip element to get the score.*/
double score = zzlGetScore(prev); /* Obtain the prev score. */ double score = zzlGetScore(prev); /* Obtain the prev score. */
if (!zslValueGteMin(score,&key->u.zset.rs)) { if (!zslValueGteMin(score,&key->u.zset.rs)) {
key->u.zset.er = 1; key->u.zset.er = 1;
...@@ -7991,22 +7991,22 @@ int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleSc ...@@ -7991,22 +7991,22 @@ int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleSc
cursor->done = 1; cursor->done = 1;
ret = 0; ret = 0;
} else if (o->type == OBJ_ZSET) { } else if (o->type == OBJ_ZSET) {
unsigned char *p = ziplistIndex(o->ptr,0); unsigned char *p = lpSeek(o->ptr,0);
unsigned char *vstr; unsigned char *vstr;
unsigned int vlen; unsigned int vlen;
long long vll; long long vll;
while(p) { while(p) {
ziplistGet(p,&vstr,&vlen,&vll); vstr = lpGetValue(p,&vlen,&vll);
robj *field = (vstr != NULL) ? robj *field = (vstr != NULL) ?
createStringObject((char*)vstr,vlen) : createStringObject((char*)vstr,vlen) :
createObject(OBJ_STRING,sdsfromlonglong(vll)); createObject(OBJ_STRING,sdsfromlonglong(vll));
p = ziplistNext(o->ptr,p); p = lpNext(o->ptr,p);
ziplistGet(p,&vstr,&vlen,&vll); vstr = lpGetValue(p,&vlen,&vll);
robj *value = (vstr != NULL) ? robj *value = (vstr != NULL) ?
createStringObject((char*)vstr,vlen) : createStringObject((char*)vstr,vlen) :
createObject(OBJ_STRING,sdsfromlonglong(vll)); createObject(OBJ_STRING,sdsfromlonglong(vll));
fn(key, field, value, privdata); fn(key, field, value, privdata);
p = ziplistNext(o->ptr,p); p = lpNext(o->ptr,p);
decrRefCount(field); decrRefCount(field);
decrRefCount(value); decrRefCount(value);
} }
......
...@@ -272,10 +272,10 @@ robj *createZsetObject(void) { ...@@ -272,10 +272,10 @@ robj *createZsetObject(void) {
return o; return o;
} }
robj *createZsetZiplistObject(void) { robj *createZsetListpackObject(void) {
unsigned char *zl = ziplistNew(); unsigned char *lp = lpNew(0);
robj *o = createObject(OBJ_ZSET,zl); robj *o = createObject(OBJ_ZSET,lp);
o->encoding = OBJ_ENCODING_ZIPLIST; o->encoding = OBJ_ENCODING_LISTPACK;
return o; return o;
} }
...@@ -329,7 +329,7 @@ void freeZsetObject(robj *o) { ...@@ -329,7 +329,7 @@ void freeZsetObject(robj *o) {
zslFree(zs->zsl); zslFree(zs->zsl);
zfree(zs); zfree(zs);
break; break;
case OBJ_ENCODING_ZIPLIST: case OBJ_ENCODING_LISTPACK:
zfree(o->ptr); zfree(o->ptr);
break; break;
default: default:
...@@ -473,8 +473,8 @@ void dismissZsetObject(robj *o, size_t size_hint) { ...@@ -473,8 +473,8 @@ void dismissZsetObject(robj *o, size_t size_hint) {
dict *d = zs->dict; dict *d = zs->dict;
dismissMemory(d->ht_table[0], DICTHT_SIZE(d->ht_size_exp[0])*sizeof(dictEntry*)); dismissMemory(d->ht_table[0], DICTHT_SIZE(d->ht_size_exp[0])*sizeof(dictEntry*));
dismissMemory(d->ht_table[1], DICTHT_SIZE(d->ht_size_exp[1])*sizeof(dictEntry*)); dismissMemory(d->ht_table[1], DICTHT_SIZE(d->ht_size_exp[1])*sizeof(dictEntry*));
} else if (o->encoding == OBJ_ENCODING_ZIPLIST) { } else if (o->encoding == OBJ_ENCODING_LISTPACK) {
dismissMemory(o->ptr, ziplistBlobLen((unsigned char*)o->ptr)); dismissMemory(o->ptr, lpBytes((unsigned char*)o->ptr));
} else { } else {
serverPanic("Unknown zset encoding type"); serverPanic("Unknown zset encoding type");
} }
...@@ -1027,7 +1027,7 @@ size_t objectComputeSize(robj *key, robj *o, size_t sample_size, int dbid) { ...@@ -1027,7 +1027,7 @@ size_t objectComputeSize(robj *key, robj *o, size_t sample_size, int dbid) {
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
} }
} else if (o->type == OBJ_ZSET) { } else if (o->type == OBJ_ZSET) {
if (o->encoding == OBJ_ENCODING_ZIPLIST) { if (o->encoding == OBJ_ENCODING_LISTPACK) {
asize = sizeof(*o)+zmalloc_size(o->ptr); asize = sizeof(*o)+zmalloc_size(o->ptr);
} else if (o->encoding == OBJ_ENCODING_SKIPLIST) { } else if (o->encoding == OBJ_ENCODING_SKIPLIST) {
d = ((zset*)o->ptr)->dict; d = ((zset*)o->ptr)->dict;
......
...@@ -669,8 +669,8 @@ int rdbSaveObjectType(rio *rdb, robj *o) { ...@@ -669,8 +669,8 @@ int rdbSaveObjectType(rio *rdb, robj *o) {
else else
serverPanic("Unknown set encoding"); serverPanic("Unknown set encoding");
case OBJ_ZSET: case OBJ_ZSET:
if (o->encoding == OBJ_ENCODING_ZIPLIST) if (o->encoding == OBJ_ENCODING_LISTPACK)
return rdbSaveType(rdb,RDB_TYPE_ZSET_ZIPLIST); return rdbSaveType(rdb,RDB_TYPE_ZSET_LISTPACK);
else if (o->encoding == OBJ_ENCODING_SKIPLIST) else if (o->encoding == OBJ_ENCODING_SKIPLIST)
return rdbSaveType(rdb,RDB_TYPE_ZSET_2); return rdbSaveType(rdb,RDB_TYPE_ZSET_2);
else else
...@@ -861,8 +861,8 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) { ...@@ -861,8 +861,8 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) {
} }
} else if (o->type == OBJ_ZSET) { } else if (o->type == OBJ_ZSET) {
/* Save a sorted set value */ /* Save a sorted set value */
if (o->encoding == OBJ_ENCODING_ZIPLIST) { if (o->encoding == OBJ_ENCODING_LISTPACK) {
size_t l = ziplistBlobLen((unsigned char*)o->ptr); size_t l = lpBytes((unsigned char*)o->ptr);
if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1; if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
nwritten += n; nwritten += n;
...@@ -1518,6 +1518,124 @@ robj *rdbLoadCheckModuleValue(rio *rdb, char *modulename) { ...@@ -1518,6 +1518,124 @@ robj *rdbLoadCheckModuleValue(rio *rdb, char *modulename) {
return createStringObject("module-dummy-value",18); return createStringObject("module-dummy-value",18);
} }
/* callback for hashZiplistConvertAndValidateIntegrity.
* Check that the ziplist doesn't have duplicate hash field names.
* The ziplist element pointed by 'p' will be converted and stored into listpack. */
static int _ziplistPairsEntryConvertAndValidate(unsigned char *p, unsigned int head_count, void *userdata) {
unsigned char *str;
unsigned int slen;
long long vll;
struct {
long count;
dict *fields;
unsigned char **lp;
} *data = userdata;
if (data->fields == NULL) {
data->fields = dictCreate(&hashDictType);
dictExpand(data->fields, head_count/2);
}
if (!ziplistGet(p, &str, &slen, &vll))
return 0;
/* Even records are field names, add to dict and check that's not a dup */
if (((data->count) & 1) == 0) {
sds field = str? sdsnewlen(str, slen): sdsfromlonglong(vll);
if (dictAdd(data->fields, field, NULL) != DICT_OK) {
/* Duplicate, return an error */
sdsfree(field);
return 0;
}
}
if (str) {
*(data->lp) = lpAppend(*(data->lp), (unsigned char*)str, slen);
} else {
*(data->lp) = lpAppendInteger(*(data->lp), vll);
}
(data->count)++;
return 1;
}
/* Validate the integrity of the data structure while converting it to
* listpack and storing it at 'lp'.
* The function is safe to call on non-validated ziplists, it returns 0
* when encounter an integrity validation issue. */
int ziplistPairsConvertAndValidateIntegrity(unsigned char *zl, size_t size, unsigned char **lp) {
/* Keep track of the field names to locate duplicate ones */
struct {
long count;
dict *fields; /* Initialisation at the first callback. */
unsigned char **lp;
} data = {0, NULL, lp};
int ret = ziplistValidateIntegrity(zl, size, 1, _ziplistPairsEntryConvertAndValidate, &data);
/* make sure we have an even number of records. */
if (data.count & 1)
ret = 0;
if (data.fields) dictRelease(data.fields);
return ret;
}
/* callback for to check the listpack doesn't have duplicate records */
static int _lpPairsEntryValidation(unsigned char *p, unsigned int head_count, void *userdata) {
struct {
long count;
dict *fields;
} *data = userdata;
if (data->fields == NULL) {
data->fields = dictCreate(&hashDictType);
dictExpand(data->fields, head_count/2);
}
/* Even records are field names, add to dict and check that's not a dup */
if (((data->count) & 1) == 0) {
unsigned char *str;
int64_t slen;
unsigned char buf[LP_INTBUF_SIZE];
str = lpGet(p, &slen, buf);
sds field = sdsnewlen(str, slen);
if (dictAdd(data->fields, field, NULL) != DICT_OK) {
/* Duplicate, return an error */
sdsfree(field);
return 0;
}
}
(data->count)++;
return 1;
}
/* Validate the integrity of the listpack structure.
* when `deep` is 0, only the integrity of the header is validated.
* when `deep` is 1, we scan all the entries one by one. */
int lpPairsValidateIntegrityAndDups(unsigned char *lp, size_t size, int deep) {
if (!deep)
return lpValidateIntegrity(lp, size, 0, NULL, NULL);
/* Keep track of the field names to locate duplicate ones */
struct {
long count;
dict *fields; /* Initialisation at the first callback. */
} data = {0, NULL};
int ret = lpValidateIntegrity(lp, size, 1, _lpPairsEntryValidation, &data);
/* make sure we have an even number of records. */
if (data.count & 1)
ret = 0;
if (data.fields) dictRelease(data.fields);
return ret;
}
/* Load a Redis object of the specified type from the specified file. /* Load a Redis object of the specified type from the specified file.
* On success a newly allocated object is returned, otherwise NULL. * On success a newly allocated object is returned, otherwise NULL.
* When the function returns NULL and if 'error' is not NULL, the * When the function returns NULL and if 'error' is not NULL, the
...@@ -1686,9 +1804,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -1686,9 +1804,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
} }
/* Convert *after* loading, since sorted sets are not stored ordered. */ /* Convert *after* loading, since sorted sets are not stored ordered. */
if (zsetLength(o) <= server.zset_max_ziplist_entries && if (zsetLength(o) <= server.zset_max_listpack_entries &&
maxelelen <= server.zset_max_ziplist_value) maxelelen <= server.zset_max_listpack_value)
zsetConvert(o,OBJ_ENCODING_ZIPLIST); zsetConvert(o,OBJ_ENCODING_LISTPACK);
} else if (rdbtype == RDB_TYPE_HASH) { } else if (rdbtype == RDB_TYPE_HASH) {
uint64_t len; uint64_t len;
int ret; int ret;
...@@ -1842,6 +1960,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -1842,6 +1960,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
rdbtype == RDB_TYPE_LIST_ZIPLIST || rdbtype == RDB_TYPE_LIST_ZIPLIST ||
rdbtype == RDB_TYPE_SET_INTSET || rdbtype == RDB_TYPE_SET_INTSET ||
rdbtype == RDB_TYPE_ZSET_ZIPLIST || rdbtype == RDB_TYPE_ZSET_ZIPLIST ||
rdbtype == RDB_TYPE_ZSET_LISTPACK ||
rdbtype == RDB_TYPE_HASH_ZIPLIST || rdbtype == RDB_TYPE_HASH_ZIPLIST ||
rdbtype == RDB_TYPE_HASH_LISTPACK) rdbtype == RDB_TYPE_HASH_LISTPACK)
{ {
...@@ -1947,30 +2066,53 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -1947,30 +2066,53 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
setTypeConvert(o,OBJ_ENCODING_HT); setTypeConvert(o,OBJ_ENCODING_HT);
break; break;
case RDB_TYPE_ZSET_ZIPLIST: case RDB_TYPE_ZSET_ZIPLIST:
{
unsigned char *lp = lpNew(encoded_len);
if (!ziplistPairsConvertAndValidateIntegrity(encoded, encoded_len, &lp)) {
rdbReportCorruptRDB("Zset ziplist integrity check failed.");
zfree(lp);
zfree(encoded);
o->ptr = NULL;
decrRefCount(o);
return NULL;
}
zfree(o->ptr);
o->type = OBJ_ZSET;
o->ptr = lp;
o->encoding = OBJ_ENCODING_LISTPACK;
if (zsetLength(o) == 0) {
decrRefCount(o);
goto emptykey;
}
if (zsetLength(o) > server.zset_max_listpack_entries)
zsetConvert(o,OBJ_ENCODING_SKIPLIST);
break;
}
case RDB_TYPE_ZSET_LISTPACK:
if (deep_integrity_validation) server.stat_dump_payload_sanitizations++; if (deep_integrity_validation) server.stat_dump_payload_sanitizations++;
if (!zsetZiplistValidateIntegrity(encoded, encoded_len, deep_integrity_validation)) { if (!lpPairsValidateIntegrityAndDups(encoded, encoded_len, deep_integrity_validation)) {
rdbReportCorruptRDB("Zset ziplist integrity check failed."); rdbReportCorruptRDB("Zset listpack integrity check failed.");
zfree(encoded); zfree(encoded);
o->ptr = NULL; o->ptr = NULL;
decrRefCount(o); decrRefCount(o);
return NULL; return NULL;
} }
o->type = OBJ_ZSET; o->type = OBJ_ZSET;
o->encoding = OBJ_ENCODING_ZIPLIST; o->encoding = OBJ_ENCODING_LISTPACK;
if (zsetLength(o) == 0) { if (zsetLength(o) == 0) {
zfree(encoded);
o->ptr = NULL;
decrRefCount(o); decrRefCount(o);
goto emptykey; goto emptykey;
} }
if (zsetLength(o) > server.zset_max_ziplist_entries) if (zsetLength(o) > server.zset_max_listpack_entries)
zsetConvert(o,OBJ_ENCODING_SKIPLIST); zsetConvert(o,OBJ_ENCODING_SKIPLIST);
break; break;
case RDB_TYPE_HASH_ZIPLIST: case RDB_TYPE_HASH_ZIPLIST:
{ {
unsigned char *lp = lpNew(encoded_len); unsigned char *lp = lpNew(encoded_len);
if (!hashZiplistConvertAndValidateIntegrity(encoded, encoded_len, &lp)) { if (!ziplistPairsConvertAndValidateIntegrity(encoded, encoded_len, &lp)) {
rdbReportCorruptRDB("Hash ziplist integrity check failed."); rdbReportCorruptRDB("Hash ziplist integrity check failed.");
zfree(lp); zfree(lp);
zfree(encoded); zfree(encoded);
...@@ -1984,8 +2126,6 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -1984,8 +2126,6 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
o->type = OBJ_HASH; o->type = OBJ_HASH;
o->encoding = OBJ_ENCODING_LISTPACK; o->encoding = OBJ_ENCODING_LISTPACK;
if (hashTypeLength(o) == 0) { if (hashTypeLength(o) == 0) {
zfree(o->ptr);
o->ptr = NULL;
decrRefCount(o); decrRefCount(o);
goto emptykey; goto emptykey;
} }
...@@ -1997,7 +2137,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -1997,7 +2137,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 (!hashListpackValidateIntegrity(encoded, encoded_len, deep_integrity_validation)) { if (!lpPairsValidateIntegrityAndDups(encoded, encoded_len, deep_integrity_validation)) {
rdbReportCorruptRDB("Hash listpack integrity check failed."); rdbReportCorruptRDB("Hash listpack integrity check failed.");
zfree(encoded); zfree(encoded);
o->ptr = NULL; o->ptr = NULL;
...@@ -2007,8 +2147,6 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -2007,8 +2147,6 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
o->type = OBJ_HASH; o->type = OBJ_HASH;
o->encoding = OBJ_ENCODING_LISTPACK; o->encoding = OBJ_ENCODING_LISTPACK;
if (hashTypeLength(o) == 0) { if (hashTypeLength(o) == 0) {
zfree(encoded);
o->ptr = NULL;
decrRefCount(o); decrRefCount(o);
goto emptykey; goto emptykey;
} }
......
...@@ -92,10 +92,11 @@ ...@@ -92,10 +92,11 @@
#define RDB_TYPE_LIST_QUICKLIST 14 #define RDB_TYPE_LIST_QUICKLIST 14
#define RDB_TYPE_STREAM_LISTPACKS 15 #define RDB_TYPE_STREAM_LISTPACKS 15
#define RDB_TYPE_HASH_LISTPACK 16 #define RDB_TYPE_HASH_LISTPACK 16
#define RDB_TYPE_ZSET_LISTPACK 17
/* 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 <= 16)) #define rdbIsObjectType(t) ((t >= 0 && t <= 7) || (t >= 9 && t <= 17))
/* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */ /* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
#define RDB_OPCODE_MODULE_AUX 247 /* Module auxiliary data. */ #define RDB_OPCODE_MODULE_AUX 247 /* Module auxiliary data. */
......
...@@ -92,6 +92,7 @@ char *rdb_type_string[] = { ...@@ -92,6 +92,7 @@ char *rdb_type_string[] = {
"quicklist", "quicklist",
"stream", "stream",
"hash-listpack" "hash-listpack"
"zset-listpack"
}; };
/* Show a few stats collected into 'rdbstate' */ /* Show a few stats collected into 'rdbstate' */
......
...@@ -1576,8 +1576,8 @@ struct redisServer { ...@@ -1576,8 +1576,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 zset_max_ziplist_entries; size_t zset_max_listpack_entries;
size_t zset_max_ziplist_value; size_t zset_max_listpack_value;
size_t hll_sparse_max_bytes; size_t hll_sparse_max_bytes;
size_t stream_node_max_bytes; size_t stream_node_max_bytes;
long long stream_node_max_entries; long long stream_node_max_entries;
...@@ -2047,7 +2047,7 @@ robj *createSetObject(void); ...@@ -2047,7 +2047,7 @@ robj *createSetObject(void);
robj *createIntsetObject(void); robj *createIntsetObject(void);
robj *createHashObject(void); robj *createHashObject(void);
robj *createZsetObject(void); robj *createZsetObject(void);
robj *createZsetZiplistObject(void); robj *createZsetListpackObject(void);
robj *createStreamObject(void); robj *createStreamObject(void);
robj *createModuleObject(moduleType *mt, void *value); robj *createModuleObject(moduleType *mt, void *value);
int getLongFromObjectOrReply(client *c, robj *o, long *target, const char *msg); int getLongFromObjectOrReply(client *c, robj *o, long *target, const char *msg);
...@@ -2233,16 +2233,15 @@ unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec *range); ...@@ -2233,16 +2233,15 @@ unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec *range);
unsigned char *zzlLastInRange(unsigned char *zl, zrangespec *range); unsigned char *zzlLastInRange(unsigned char *zl, zrangespec *range);
unsigned long zsetLength(const robj *zobj); unsigned long zsetLength(const robj *zobj);
void zsetConvert(robj *zobj, int encoding); void zsetConvert(robj *zobj, int encoding);
void zsetConvertToZiplistIfNeeded(robj *zobj, size_t maxelelen); void zsetConvertToListpackIfNeeded(robj *zobj, size_t maxelelen);
int zsetScore(robj *zobj, sds member, double *score); int zsetScore(robj *zobj, sds member, double *score);
unsigned long zslGetRank(zskiplist *zsl, double score, sds o); unsigned long zslGetRank(zskiplist *zsl, double score, sds o);
int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, double *newscore); int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, double *newscore);
long zsetRank(robj *zobj, sds ele, int reverse); long zsetRank(robj *zobj, sds ele, int reverse);
int zsetDel(robj *zobj, sds ele); int zsetDel(robj *zobj, sds ele);
robj *zsetDup(robj *o); robj *zsetDup(robj *o);
int zsetZiplistValidateIntegrity(unsigned char *zl, size_t size, int deep);
void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey, robj *countarg); void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey, robj *countarg);
sds ziplistGetObject(unsigned char *sptr); sds lpGetObject(unsigned char *sptr);
int zslValueGteMin(double value, zrangespec *spec); int zslValueGteMin(double value, zrangespec *spec);
int zslValueLteMax(double value, zrangespec *spec); int zslValueLteMax(double value, zrangespec *spec);
void zslFreeLexRange(zlexrangespec *spec); void zslFreeLexRange(zlexrangespec *spec);
...@@ -2359,8 +2358,6 @@ robj *hashTypeLookupWriteOrCreate(client *c, robj *key); ...@@ -2359,8 +2358,6 @@ robj *hashTypeLookupWriteOrCreate(client *c, robj *key);
robj *hashTypeGetValueObject(robj *o, sds field); robj *hashTypeGetValueObject(robj *o, sds field);
int hashTypeSet(robj *o, sds field, sds value, int flags); int hashTypeSet(robj *o, sds field, sds value, int flags);
robj *hashTypeDup(robj *o); robj *hashTypeDup(robj *o);
int hashZiplistConvertAndValidateIntegrity(unsigned char *zl, size_t size, unsigned char **lp);
int hashListpackValidateIntegrity(unsigned char *lp, size_t size, int deep);
/* Pub / Sub */ /* Pub / Sub */
int pubsubUnsubscribeAllChannels(client *c, int notify); int pubsubUnsubscribeAllChannels(client *c, int notify);
......
...@@ -279,8 +279,8 @@ int hashTypeDelete(robj *o, sds field) { ...@@ -279,8 +279,8 @@ int hashTypeDelete(robj *o, sds field) {
if (fptr != NULL) { if (fptr != NULL) {
fptr = lpFind(zl, fptr, (unsigned char*)field, sdslen(field), 1); fptr = lpFind(zl, fptr, (unsigned char*)field, sdslen(field), 1);
if (fptr != NULL) { if (fptr != NULL) {
zl = lpDelete(zl,fptr,&fptr); /* Delete the key. */ /* Delete both of the key and the value. */
zl = lpDelete(zl,fptr,&fptr); /* Delete the value. */ zl = lpDeleteRangeWithEntry(zl,&fptr,2);
o->ptr = zl; o->ptr = zl;
deleted = 1; deleted = 1;
} }
...@@ -541,114 +541,6 @@ robj *hashTypeDup(robj *o) { ...@@ -541,114 +541,6 @@ robj *hashTypeDup(robj *o) {
return hobj; return hobj;
} }
/* callback for hashZiplistConvertAndValidateIntegrity.
* Check that the ziplist doesn't have duplicate hash field names.
* The ziplist element pointed by 'p' will be converted and stored into listpack. */
static int _hashZiplistEntryConvertAndValidation(unsigned char *p, void *userdata) {
unsigned char *str;
unsigned int slen;
long long vll;
struct {
long count;
dict *fields;
unsigned char **lp;
} *data = userdata;
if (!ziplistGet(p, &str, &slen, &vll))
return 0;
/* Even records are field names, add to dict and check that's not a dup */
if (((data->count) & 1) == 0) {
sds field = str? sdsnewlen(str, slen): sdsfromlonglong(vll);
if (dictAdd(data->fields, field, NULL) != DICT_OK) {
/* Duplicate, return an error */
sdsfree(field);
return 0;
}
}
if (str) {
*(data->lp) = lpAppend(*(data->lp), (unsigned char*)str, slen);
} else {
*(data->lp) = lpAppendInteger(*(data->lp), vll);
}
(data->count)++;
return 1;
}
/* callback for to check the listpack doesn't have duplicate records */
static int _hashListpackEntryValidation(unsigned char *p, void *userdata) {
struct {
long count;
dict *fields;
} *data = userdata;
/* Even records are field names, add to dict and check that's not a dup */
if (((data->count) & 1) == 0) {
unsigned char *str;
int64_t slen;
unsigned char buf[LP_INTBUF_SIZE];
str = lpGet(p, &slen, buf);
sds field = sdsnewlen(str, slen);
if (dictAdd(data->fields, field, NULL) != DICT_OK) {
/* Duplicate, return an error */
sdsfree(field);
return 0;
}
}
(data->count)++;
return 1;
}
/* Validate the integrity of the data structure while converting it to
* listpack and storing it at 'lp'.
* The function is safe to call on non-validated ziplists, it returns 0
* when encounter an integrity validation issue. */
int hashZiplistConvertAndValidateIntegrity(unsigned char *zl, size_t size, unsigned char **lp) {
/* Keep track of the field names to locate duplicate ones */
struct {
long count;
dict *fields;
unsigned char **lp;
} data = {0, dictCreate(&hashDictType), lp};
int ret = ziplistValidateIntegrity(zl, size, 1, _hashZiplistEntryConvertAndValidation, &data);
/* make sure we have an even number of records. */
if (data.count & 1)
ret = 0;
dictRelease(data.fields);
return ret;
}
/* Validate the integrity of the listpack structure.
* when `deep` is 0, only the integrity of the header is validated.
* when `deep` is 1, we scan all the entries one by one. */
int hashListpackValidateIntegrity(unsigned char *lp, size_t size, int deep) {
if (!deep)
return lpValidateIntegrity(lp, size, 0, NULL, NULL);
/* Keep track of the field names to locate duplicate ones */
struct {
long count;
dict *fields;
} data = {0, dictCreate(&hashDictType)};
int ret = lpValidateIntegrity(lp, size, 1, _hashListpackEntryValidation, &data);
/* make sure we have an even number of records. */
if (data.count & 1)
ret = 0;
dictRelease(data.fields);
return ret;
}
/* Create a new sds string from the listpack entry. */ /* Create a new sds string from the listpack entry. */
sds hashSdsFromListpackEntry(listpackEntry *e) { sds hashSdsFromListpackEntry(listpackEntry *e) {
return e->sval ? sdsnewlen(e->sval, e->slen) : sdsfromlonglong(e->lval); return e->sval ? sdsnewlen(e->sval, e->slen) : sdsfromlonglong(e->lval);
......
This diff is collapsed.
...@@ -1498,6 +1498,7 @@ int ziplistValidateIntegrity(unsigned char *zl, size_t size, int deep, ...@@ -1498,6 +1498,7 @@ int ziplistValidateIntegrity(unsigned char *zl, size_t size, int deep,
return 1; return 1;
unsigned int count = 0; unsigned int count = 0;
unsigned int header_count = intrev16ifbe(ZIPLIST_LENGTH(zl));
unsigned char *p = ZIPLIST_ENTRY_HEAD(zl); unsigned char *p = ZIPLIST_ENTRY_HEAD(zl);
unsigned char *prev = NULL; unsigned char *prev = NULL;
size_t prev_raw_size = 0; size_t prev_raw_size = 0;
...@@ -1512,7 +1513,7 @@ int ziplistValidateIntegrity(unsigned char *zl, size_t size, int deep, ...@@ -1512,7 +1513,7 @@ int ziplistValidateIntegrity(unsigned char *zl, size_t size, int deep,
return 0; return 0;
/* Optionally let the caller validate the entry too. */ /* Optionally let the caller validate the entry too. */
if (entry_cb && !entry_cb(p, cb_userdata)) if (entry_cb && !entry_cb(p, header_count, cb_userdata))
return 0; return 0;
/* Move to the next entry */ /* Move to the next entry */
...@@ -1531,7 +1532,6 @@ int ziplistValidateIntegrity(unsigned char *zl, size_t size, int deep, ...@@ -1531,7 +1532,6 @@ int ziplistValidateIntegrity(unsigned char *zl, size_t size, int deep,
return 0; return 0;
/* Check that the count in the header is correct */ /* Check that the count in the header is correct */
unsigned int header_count = intrev16ifbe(ZIPLIST_LENGTH(zl));
if (header_count != UINT16_MAX && count != header_count) if (header_count != UINT16_MAX && count != header_count)
return 0; return 0;
...@@ -2464,6 +2464,32 @@ int ziplistTest(int argc, char **argv, int accurate) { ...@@ -2464,6 +2464,32 @@ int ziplistTest(int argc, char **argv, int accurate) {
printf("%lld\n", usec()-start); printf("%lld\n", usec()-start);
} }
printf("Benchmark ziplistCompare with string\n");
{
unsigned long long start = usec();
for (int i = 0; i < 2000; i++) {
unsigned char *eptr = ziplistIndex(zl,0);
while (eptr != NULL) {
ziplistCompare(eptr,(unsigned char*)"nothing",7);
eptr = ziplistNext(zl,eptr);
}
}
printf("Done. usec=%lld\n", usec()-start);
}
printf("Benchmark ziplistCompare with number\n");
{
unsigned long long start = usec();
for (int i = 0; i < 2000; i++) {
unsigned char *eptr = ziplistIndex(zl,0);
while (eptr != NULL) {
ziplistCompare(eptr,(unsigned char*)"99999",5);
eptr = ziplistNext(zl,eptr);
}
}
printf("Done. usec=%lld\n", usec()-start);
}
zfree(zl); zfree(zl);
} }
......
...@@ -59,7 +59,7 @@ unsigned char *ziplistFind(unsigned char *zl, unsigned char *p, unsigned char *v ...@@ -59,7 +59,7 @@ unsigned char *ziplistFind(unsigned char *zl, unsigned char *p, unsigned char *v
unsigned int ziplistLen(unsigned char *zl); unsigned int ziplistLen(unsigned char *zl);
size_t ziplistBlobLen(unsigned char *zl); size_t ziplistBlobLen(unsigned char *zl);
void ziplistRepr(unsigned char *zl); void ziplistRepr(unsigned char *zl);
typedef int (*ziplistValidateEntryCB)(unsigned char* p, void* userdata); typedef int (*ziplistValidateEntryCB)(unsigned char* p, unsigned int head_count, void* userdata);
int ziplistValidateIntegrity(unsigned char *zl, size_t size, int deep, int ziplistValidateIntegrity(unsigned char *zl, size_t size, int deep,
ziplistValidateEntryCB entry_cb, void *cb_userdata); ziplistValidateEntryCB entry_cb, void *cb_userdata);
void ziplistRandomPair(unsigned char *zl, unsigned long total_count, ziplistEntry *key, ziplistEntry *val); void ziplistRandomPair(unsigned char *zl, unsigned long total_count, ziplistEntry *key, ziplistEntry *val);
......
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