Unverified Commit f27083a4 authored by perryitay's avatar perryitay Committed by GitHub
Browse files

Add support for list type to store elements larger than 4GB (#9357)



Redis lists are stored in quicklist, which is currently a linked list of ziplists.
Ziplists are limited to storing elements no larger than 4GB, so when bigger
items are added they're getting truncated.
This PR changes quicklists so that they're capable of storing large items
in quicklist nodes that are plain string buffers rather than ziplist.

As part of the PR there were few other changes in redis: 
1. new DEBUG sub-commands: 
   - QUICKLIST-PACKED-THRESHOLD - set the threshold of for the node type to
     be plan or ziplist. default (1GB)
   - QUICKLIST <key> - Shows low level info about the quicklist encoding of <key>
2. rdb format change:
   - A new type was added - RDB_TYPE_LIST_QUICKLIST_2 . 
   - container type (packed / plain) was added to the beginning of the rdb object
     (before the actual node list).
3. testing:
   - Tests that requires over 100MB will be by default skipped. a new flag was
     added to 'runtest' to run the large memory tests (not used by default)
Co-authored-by: default avatarsundb <sundbcn@gmail.com>
Co-authored-by: default avatarOran Agra <oran@redislabs.com>
parent f11a2d4d
...@@ -254,7 +254,7 @@ jobs: ...@@ -254,7 +254,7 @@ jobs:
sudo apt-get install tcl8.6 tclx valgrind -y sudo apt-get install tcl8.6 tclx valgrind -y
- name: test - name: test
if: true && !contains(github.event.inputs.skiptests, 'redis') if: true && !contains(github.event.inputs.skiptests, 'redis')
run: ./runtest --valgrind --verbose --clients 1 --tags -large-memory --dump-logs ${{github.event.inputs.test_args}} run: ./runtest --valgrind --verbose --clients 1 --dump-logs ${{github.event.inputs.test_args}}
- name: module api test - name: module api test
if: true && !contains(github.event.inputs.skiptests, 'modules') if: true && !contains(github.event.inputs.skiptests, 'modules')
run: ./runtest-moduleapi --valgrind --no-latency --verbose --clients 1 ${{github.event.inputs.test_args}} run: ./runtest-moduleapi --valgrind --no-latency --verbose --clients 1 ${{github.event.inputs.test_args}}
...@@ -285,7 +285,7 @@ jobs: ...@@ -285,7 +285,7 @@ jobs:
sudo apt-get install tcl8.6 tclx valgrind -y sudo apt-get install tcl8.6 tclx valgrind -y
- name: test - name: test
if: true && !contains(github.event.inputs.skiptests, 'redis') if: true && !contains(github.event.inputs.skiptests, 'redis')
run: ./runtest --valgrind --verbose --clients 1 --tags -large-memory --dump-logs ${{github.event.inputs.test_args}} run: ./runtest --valgrind --verbose --clients 1 --dump-logs ${{github.event.inputs.test_args}}
- name: module api test - name: module api test
if: true && !contains(github.event.inputs.skiptests, 'modules') if: true && !contains(github.event.inputs.skiptests, 'modules')
run: ./runtest-moduleapi --valgrind --no-latency --verbose --clients 1 ${{github.event.inputs.test_args}} run: ./runtest-moduleapi --valgrind --no-latency --verbose --clients 1 ${{github.event.inputs.test_args}}
...@@ -420,7 +420,7 @@ jobs: ...@@ -420,7 +420,7 @@ jobs:
prepare: pkg install -y bash gmake lang/tcl86 lang/tclx prepare: pkg install -y bash gmake lang/tcl86 lang/tclx
run: > run: >
gmake || exit 1 ; gmake || exit 1 ;
if echo "${{github.event.inputs.skiptests}}" | grep -vq redis ; then ./runtest --accurate --verbose --no-latency --tags -large-memory --dump-logs ${{github.event.inputs.test_args}} || exit 1 ; fi ; if echo "${{github.event.inputs.skiptests}}" | grep -vq redis ; then ./runtest --accurate --verbose --no-latency --dump-logs ${{github.event.inputs.test_args}} || exit 1 ; fi ;
if echo "${{github.event.inputs.skiptests}}" | grep -vq modules ; then MAKE=gmake ./runtest-moduleapi --verbose ${{github.event.inputs.test_args}} || exit 1 ; fi ; if echo "${{github.event.inputs.skiptests}}" | grep -vq modules ; then MAKE=gmake ./runtest-moduleapi --verbose ${{github.event.inputs.test_args}} || exit 1 ; fi ;
if echo "${{github.event.inputs.skiptests}}" | grep -vq sentinel ; then ./runtest-sentinel ${{github.event.inputs.cluster_test_args}} || exit 1 ; fi ; if echo "${{github.event.inputs.skiptests}}" | grep -vq sentinel ; then ./runtest-sentinel ${{github.event.inputs.cluster_test_args}} || exit 1 ; fi ;
if echo "${{github.event.inputs.skiptests}}" | grep -vq cluster ; then ./runtest-cluster ${{github.event.inputs.cluster_test_args}} || exit 1 ; fi ; if echo "${{github.event.inputs.skiptests}}" | grep -vq cluster ; then ./runtest-cluster ${{github.event.inputs.cluster_test_args}} || exit 1 ; fi ;
......
...@@ -29,9 +29,11 @@ ...@@ -29,9 +29,11 @@
*/ */
#include "server.h" #include "server.h"
#include "util.h"
#include "sha1.h" /* SHA1 is used for DEBUG DIGEST */ #include "sha1.h" /* SHA1 is used for DEBUG DIGEST */
#include "crc64.h" #include "crc64.h"
#include "bio.h" #include "bio.h"
#include "quicklist.h"
#include <arpa/inet.h> #include <arpa/inet.h>
#include <signal.h> #include <signal.h>
...@@ -459,6 +461,9 @@ void debugCommand(client *c) { ...@@ -459,6 +461,9 @@ void debugCommand(client *c) {
" Setting it to 0 disables expiring keys in background when they are not", " Setting it to 0 disables expiring keys in background when they are not",
" accessed (otherwise the Redis behavior). Setting it to 1 reenables back the", " accessed (otherwise the Redis behavior). Setting it to 1 reenables back the",
" default.", " default.",
"QUICKLIST-PACKED-THRESHOLD <size>",
" Sets the threshold for elements to be inserted as plain vs packed nodes",
" Default value is 1GB, allows values up to 4GB",
"SET-SKIP-CHECKSUM-VALIDATION <0|1>", "SET-SKIP-CHECKSUM-VALIDATION <0|1>",
" Enables or disables checksum checks for RDB files and RESTORE's payload.", " Enables or disables checksum checks for RDB files and RESTORE's payload.",
"SLEEP <seconds>", "SLEEP <seconds>",
...@@ -469,6 +474,9 @@ void debugCommand(client *c) { ...@@ -469,6 +474,9 @@ void debugCommand(client *c) {
" Return the size of different Redis core C structures.", " Return the size of different Redis core C structures.",
"ZIPLIST <key>", "ZIPLIST <key>",
" Show low level info about the ziplist encoding of <key>.", " Show low level info about the ziplist encoding of <key>.",
"QUICKLIST <key> [<0|1>]",
" Show low level info about the quicklist encoding of <key>."
" The optional argument (0 by default) sets the level of detail",
"CLIENT-EVICTION", "CLIENT-EVICTION",
" Show low level client eviction pools info (maxmemory-clients).", " Show low level client eviction pools info (maxmemory-clients).",
"PAUSE-CRON <0|1>", "PAUSE-CRON <0|1>",
...@@ -652,6 +660,21 @@ NULL ...@@ -652,6 +660,21 @@ NULL
ziplistRepr(o->ptr); ziplistRepr(o->ptr);
addReplyStatus(c,"Ziplist structure printed on stdout"); addReplyStatus(c,"Ziplist structure printed on stdout");
} }
} else if (!strcasecmp(c->argv[1]->ptr,"quicklist") && (c->argc == 3 || c->argc == 4)) {
robj *o;
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nokeyerr))
== NULL) return;
int full = 0;
if (c->argc == 4)
full = atoi(c->argv[3]->ptr);
if (o->encoding != OBJ_ENCODING_QUICKLIST) {
addReplyError(c,"Not a quicklist encoded object.");
} else {
quicklistRepr(o->ptr, full);
addReplyStatus(c,"Quicklist structure printed on stdout");
}
} else if (!strcasecmp(c->argv[1]->ptr,"populate") && } else if (!strcasecmp(c->argv[1]->ptr,"populate") &&
c->argc >= 3 && c->argc <= 5) { c->argc >= 3 && c->argc <= 5) {
long keys, j; long keys, j;
...@@ -782,6 +805,16 @@ NULL ...@@ -782,6 +805,16 @@ NULL
{ {
server.active_expire_enabled = atoi(c->argv[2]->ptr); server.active_expire_enabled = atoi(c->argv[2]->ptr);
addReply(c,shared.ok); addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"quicklist-packed-threshold") &&
c->argc == 3)
{
int memerr;
unsigned long long sz = memtoull((const char *)c->argv[2]->ptr, &memerr);
if (memerr || !quicklistisSetPackedThreshold(sz)) {
addReplyError(c, "argument must be a memory value bigger then 1 and smaller than 4gb");
} else {
addReply(c,shared.ok);
}
} else if (!strcasecmp(c->argv[1]->ptr,"set-skip-checksum-validation") && } else if (!strcasecmp(c->argv[1]->ptr,"set-skip-checksum-validation") &&
c->argc == 3) c->argc == 3)
{ {
......
...@@ -416,8 +416,8 @@ long activeDefragQuickListNode(quicklist *ql, quicklistNode **node_ref) { ...@@ -416,8 +416,8 @@ long activeDefragQuickListNode(quicklist *ql, quicklistNode **node_ref) {
*node_ref = node = newnode; *node_ref = node = newnode;
defragged++; defragged++;
} }
if ((newzl = activeDefragAlloc(node->zl))) if ((newzl = activeDefragAlloc(node->entry)))
defragged++, node->zl = newzl; defragged++, node->entry = newzl;
return defragged; return defragged;
} }
......
...@@ -415,9 +415,9 @@ void dismissListObject(robj *o, size_t size_hint) { ...@@ -415,9 +415,9 @@ void dismissListObject(robj *o, size_t size_hint) {
quicklistNode *node = ql->head; quicklistNode *node = ql->head;
while (node) { while (node) {
if (quicklistNodeIsCompressed(node)) { if (quicklistNodeIsCompressed(node)) {
dismissMemory(node->zl, ((quicklistLZF*)node->zl)->sz); dismissMemory(node->entry, ((quicklistLZF*)node->entry)->sz);
} else { } else {
dismissMemory(node->zl, node->sz); dismissMemory(node->entry, node->sz);
} }
node = node->next; node = node->next;
} }
...@@ -1000,7 +1000,7 @@ size_t objectComputeSize(robj *key, robj *o, size_t sample_size, int dbid) { ...@@ -1000,7 +1000,7 @@ size_t objectComputeSize(robj *key, robj *o, size_t sample_size, int dbid) {
quicklistNode *node = ql->head; quicklistNode *node = ql->head;
asize = sizeof(*o)+sizeof(quicklist); asize = sizeof(*o)+sizeof(quicklist);
do { do {
elesize += sizeof(quicklistNode)+zmalloc_size(node->zl); elesize += sizeof(quicklistNode)+zmalloc_size(node->entry);
samples++; samples++;
} while ((node = node->next) && samples < sample_size); } while ((node = node->next) && samples < sample_size);
asize += (double)elesize/samples*ql->len; asize += (double)elesize/samples*ql->len;
......
...@@ -28,6 +28,7 @@ ...@@ -28,6 +28,7 @@
* POSSIBILITY OF SUCH DAMAGE. * POSSIBILITY OF SUCH DAMAGE.
*/ */
#include <stdio.h>
#include <string.h> /* for memcpy */ #include <string.h> /* for memcpy */
#include "quicklist.h" #include "quicklist.h"
#include "zmalloc.h" #include "zmalloc.h"
...@@ -37,10 +38,6 @@ ...@@ -37,10 +38,6 @@
#include "lzf.h" #include "lzf.h"
#include "redisassert.h" #include "redisassert.h"
#if defined(REDIS_TEST) || defined(REDIS_TEST_VERBOSE)
#include <stdio.h> /* for printf (debug printing), snprintf (genstr) */
#endif
#ifndef REDIS_STATIC #ifndef REDIS_STATIC
#define REDIS_STATIC static #define REDIS_STATIC static
#endif #endif
...@@ -50,6 +47,21 @@ ...@@ -50,6 +47,21 @@
* just one byte, it still won't overflow the 16 bit count field. */ * just one byte, it still won't overflow the 16 bit count field. */
static const size_t optimization_level[] = {4096, 8192, 16384, 32768, 65536}; static const size_t optimization_level[] = {4096, 8192, 16384, 32768, 65536};
/* packed_threshold is initialized to 1gb*/
static size_t packed_threshold = (1 << 30);
/* set threshold for PLAIN nodes, the real limit is 4gb */
#define isLargeElement(size) ((size) >= packed_threshold)
int quicklistisSetPackedThreshold(size_t sz) {
/* Don't allow threshold to be set above or even slightly below 4GB */
if (sz > (1ull<<32) - (1<<20)) {
return 0;
}
packed_threshold = sz;
return 1;
}
/* Maximum size in bytes of any multi-element ziplist. /* Maximum size in bytes of any multi-element ziplist.
* Larger values will live in their own isolated ziplists. * Larger values will live in their own isolated ziplists.
* This is used only if we're limited by record count. when we're limited by * This is used only if we're limited by record count. when we're limited by
...@@ -144,7 +156,7 @@ quicklist *quicklistNew(int fill, int compress) { ...@@ -144,7 +156,7 @@ quicklist *quicklistNew(int fill, int compress) {
REDIS_STATIC quicklistNode *quicklistCreateNode(void) { REDIS_STATIC quicklistNode *quicklistCreateNode(void) {
quicklistNode *node; quicklistNode *node;
node = zmalloc(sizeof(*node)); node = zmalloc(sizeof(*node));
node->zl = NULL; node->entry = NULL;
node->count = 0; node->count = 0;
node->sz = 0; node->sz = 0;
node->next = node->prev = NULL; node->next = node->prev = NULL;
...@@ -167,7 +179,7 @@ void quicklistRelease(quicklist *quicklist) { ...@@ -167,7 +179,7 @@ void quicklistRelease(quicklist *quicklist) {
while (len--) { while (len--) {
next = current->next; next = current->next;
zfree(current->zl); zfree(current->entry);
quicklist->count -= current->count; quicklist->count -= current->count;
zfree(current); zfree(current);
...@@ -194,7 +206,7 @@ REDIS_STATIC int __quicklistCompressNode(quicklistNode *node) { ...@@ -194,7 +206,7 @@ REDIS_STATIC int __quicklistCompressNode(quicklistNode *node) {
quicklistLZF *lzf = zmalloc(sizeof(*lzf) + node->sz); quicklistLZF *lzf = zmalloc(sizeof(*lzf) + node->sz);
/* Cancel if compression fails or doesn't compress small enough */ /* Cancel if compression fails or doesn't compress small enough */
if (((lzf->sz = lzf_compress(node->zl, node->sz, lzf->compressed, if (((lzf->sz = lzf_compress(node->entry, node->sz, lzf->compressed,
node->sz)) == 0) || node->sz)) == 0) ||
lzf->sz + MIN_COMPRESS_IMPROVE >= node->sz) { lzf->sz + MIN_COMPRESS_IMPROVE >= node->sz) {
/* lzf_compress aborts/rejects compression if value not compressible. */ /* lzf_compress aborts/rejects compression if value not compressible. */
...@@ -202,8 +214,8 @@ REDIS_STATIC int __quicklistCompressNode(quicklistNode *node) { ...@@ -202,8 +214,8 @@ REDIS_STATIC int __quicklistCompressNode(quicklistNode *node) {
return 0; return 0;
} }
lzf = zrealloc(lzf, sizeof(*lzf) + lzf->sz); lzf = zrealloc(lzf, sizeof(*lzf) + lzf->sz);
zfree(node->zl); zfree(node->entry);
node->zl = (unsigned char *)lzf; node->entry = (unsigned char *)lzf;
node->encoding = QUICKLIST_NODE_ENCODING_LZF; node->encoding = QUICKLIST_NODE_ENCODING_LZF;
node->recompress = 0; node->recompress = 0;
return 1; return 1;
...@@ -225,14 +237,14 @@ REDIS_STATIC int __quicklistDecompressNode(quicklistNode *node) { ...@@ -225,14 +237,14 @@ REDIS_STATIC int __quicklistDecompressNode(quicklistNode *node) {
#endif #endif
void *decompressed = zmalloc(node->sz); void *decompressed = zmalloc(node->sz);
quicklistLZF *lzf = (quicklistLZF *)node->zl; quicklistLZF *lzf = (quicklistLZF *)node->entry;
if (lzf_decompress(lzf->compressed, lzf->sz, decompressed, node->sz) == 0) { if (lzf_decompress(lzf->compressed, lzf->sz, decompressed, node->sz) == 0) {
/* Someone requested decompress, but we can't decompress. Not good. */ /* Someone requested decompress, but we can't decompress. Not good. */
zfree(decompressed); zfree(decompressed);
return 0; return 0;
} }
zfree(lzf); zfree(lzf);
node->zl = decompressed; node->entry = decompressed;
node->encoding = QUICKLIST_NODE_ENCODING_RAW; node->encoding = QUICKLIST_NODE_ENCODING_RAW;
return 1; return 1;
} }
...@@ -258,11 +270,41 @@ REDIS_STATIC int __quicklistDecompressNode(quicklistNode *node) { ...@@ -258,11 +270,41 @@ REDIS_STATIC int __quicklistDecompressNode(quicklistNode *node) {
* Pointer to LZF data is assigned to '*data'. * Pointer to LZF data is assigned to '*data'.
* Return value is the length of compressed LZF data. */ * Return value is the length of compressed LZF data. */
size_t quicklistGetLzf(const quicklistNode *node, void **data) { size_t quicklistGetLzf(const quicklistNode *node, void **data) {
quicklistLZF *lzf = (quicklistLZF *)node->zl; quicklistLZF *lzf = (quicklistLZF *)node->entry;
*data = lzf->compressed; *data = lzf->compressed;
return lzf->sz; return lzf->sz;
} }
void quicklistRepr(unsigned char *ql, int full) {
int i = 0;
quicklist *quicklist = (struct quicklist*) ql;
printf("{count : %ld}\n", quicklist->count);
printf("{len : %ld}\n", quicklist->len);
printf("{fill : %d}\n", quicklist->fill);
printf("{compress : %d}\n", quicklist->compress);
printf("{bookmark_count : %d}\n", quicklist->bookmark_count);
quicklistNode* node = quicklist->head;
while(node != NULL) {
printf("{quicklist node(%d)\n", i++);
printf("{container : %s, encoding: %s, size: %zu}\n", QL_NODE_IS_PLAIN(node) ? "PLAIN": "PACKED",
(node->encoding == QUICKLIST_NODE_ENCODING_RAW) ? "RAW": "LZF", node->sz);
if (full) {
if (node->container == QUICKLIST_NODE_CONTAINER_ZIPLIST) {
printf("{ ziplist:\n");
ziplistRepr(node->entry);
printf("}\n");
} else if (QL_NODE_IS_PLAIN(node)) {
printf("{ entry : %s }\n", node->entry);
}
printf("}\n");
}
node = node->next;
}
}
#define quicklistAllowsCompression(_ql) ((_ql)->compress != 0) #define quicklistAllowsCompression(_ql) ((_ql)->compress != 0)
/* Force 'quicklist' to meet compression guidelines set by compress depth. /* Force 'quicklist' to meet compression guidelines set by compress depth.
...@@ -430,6 +472,9 @@ REDIS_STATIC int _quicklistNodeAllowInsert(const quicklistNode *node, ...@@ -430,6 +472,9 @@ REDIS_STATIC int _quicklistNodeAllowInsert(const quicklistNode *node,
if (unlikely(!node)) if (unlikely(!node))
return 0; return 0;
if (unlikely(QL_NODE_IS_PLAIN(node) || isLargeElement(sz)))
return 0;
int ziplist_overhead; int ziplist_overhead;
/* size of previous offset */ /* size of previous offset */
if (sz < 254) if (sz < 254)
...@@ -465,6 +510,9 @@ REDIS_STATIC int _quicklistNodeAllowMerge(const quicklistNode *a, ...@@ -465,6 +510,9 @@ REDIS_STATIC int _quicklistNodeAllowMerge(const quicklistNode *a,
if (!a || !b) if (!a || !b)
return 0; return 0;
if (unlikely(QL_NODE_IS_PLAIN(a) || QL_NODE_IS_PLAIN(b)))
return 0;
/* approximate merged ziplist size (- 11 to remove one ziplist /* approximate merged ziplist size (- 11 to remove one ziplist
* header/trailer) */ * header/trailer) */
unsigned int merge_sz = a->sz + b->sz - 11; unsigned int merge_sz = a->sz + b->sz - 11;
...@@ -482,24 +530,45 @@ REDIS_STATIC int _quicklistNodeAllowMerge(const quicklistNode *a, ...@@ -482,24 +530,45 @@ REDIS_STATIC int _quicklistNodeAllowMerge(const quicklistNode *a,
#define quicklistNodeUpdateSz(node) \ #define quicklistNodeUpdateSz(node) \
do { \ do { \
(node)->sz = ziplistBlobLen((node)->zl); \ (node)->sz = ziplistBlobLen((node)->entry); \
} while (0) } while (0)
static quicklistNode* __quicklistCreatePlainNode(void *value, size_t sz) {
quicklistNode *new_node = quicklistCreateNode();
new_node->entry = zmalloc(sz);
new_node->container = QUICKLIST_NODE_CONTAINER_PLAIN;
memcpy(new_node->entry, value, sz);
new_node->sz = sz;
new_node->count++;
return new_node;
}
static void __quicklistInsertPlainNode(quicklist *quicklist, quicklistNode *old_node,
void *value, size_t sz, int after) {
__quicklistInsertNode(quicklist, old_node, __quicklistCreatePlainNode(value, sz), after);
quicklist->count++;
}
/* Add new entry to head node of quicklist. /* Add new entry to head node of quicklist.
* *
* Returns 0 if used existing head. * Returns 0 if used existing head.
* Returns 1 if new head created. */ * Returns 1 if new head created. */
int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) { int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) {
quicklistNode *orig_head = quicklist->head; quicklistNode *orig_head = quicklist->head;
assert(sz < UINT32_MAX); /* TODO: add support for quicklist nodes that are sds encoded (not zipped) */
if (unlikely(isLargeElement(sz))) {
__quicklistInsertPlainNode(quicklist, quicklist->head, value, sz, 0);
return 1;
}
if (likely( if (likely(
_quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) { _quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) {
quicklist->head->zl = quicklist->head->entry =
ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD); ziplistPush(quicklist->head->entry, value, sz, ZIPLIST_HEAD);
quicklistNodeUpdateSz(quicklist->head); quicklistNodeUpdateSz(quicklist->head);
} else { } else {
quicklistNode *node = quicklistCreateNode(); quicklistNode *node = quicklistCreateNode();
node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD); node->entry = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
quicklistNodeUpdateSz(node); quicklistNodeUpdateSz(node);
_quicklistInsertNodeBefore(quicklist, quicklist->head, node); _quicklistInsertNodeBefore(quicklist, quicklist->head, node);
...@@ -515,15 +584,19 @@ int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) { ...@@ -515,15 +584,19 @@ int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) {
* Returns 1 if new tail created. */ * Returns 1 if new tail created. */
int quicklistPushTail(quicklist *quicklist, void *value, size_t sz) { int quicklistPushTail(quicklist *quicklist, void *value, size_t sz) {
quicklistNode *orig_tail = quicklist->tail; quicklistNode *orig_tail = quicklist->tail;
assert(sz < UINT32_MAX); /* TODO: add support for quicklist nodes that are sds encoded (not zipped) */ if (unlikely(isLargeElement(sz))) {
__quicklistInsertPlainNode(quicklist, quicklist->tail, value, sz, 1);
return 1;
}
if (likely( if (likely(
_quicklistNodeAllowInsert(quicklist->tail, quicklist->fill, sz))) { _quicklistNodeAllowInsert(quicklist->tail, quicklist->fill, sz))) {
quicklist->tail->zl = quicklist->tail->entry =
ziplistPush(quicklist->tail->zl, value, sz, ZIPLIST_TAIL); ziplistPush(quicklist->tail->entry, value, sz, ZIPLIST_TAIL);
quicklistNodeUpdateSz(quicklist->tail); quicklistNodeUpdateSz(quicklist->tail);
} else { } else {
quicklistNode *node = quicklistCreateNode(); quicklistNode *node = quicklistCreateNode();
node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_TAIL); node->entry = ziplistPush(ziplistNew(), value, sz, ZIPLIST_TAIL);
quicklistNodeUpdateSz(node); quicklistNodeUpdateSz(node);
_quicklistInsertNodeAfter(quicklist, quicklist->tail, node); _quicklistInsertNodeAfter(quicklist, quicklist->tail, node);
...@@ -539,14 +612,22 @@ int quicklistPushTail(quicklist *quicklist, void *value, size_t sz) { ...@@ -539,14 +612,22 @@ int quicklistPushTail(quicklist *quicklist, void *value, size_t sz) {
void quicklistAppendZiplist(quicklist *quicklist, unsigned char *zl) { void quicklistAppendZiplist(quicklist *quicklist, unsigned char *zl) {
quicklistNode *node = quicklistCreateNode(); quicklistNode *node = quicklistCreateNode();
node->zl = zl; node->entry = zl;
node->count = ziplistLen(node->zl); node->count = ziplistLen(node->entry);
node->sz = ziplistBlobLen(zl); node->sz = ziplistBlobLen(zl);
_quicklistInsertNodeAfter(quicklist, quicklist->tail, node); _quicklistInsertNodeAfter(quicklist, quicklist->tail, node);
quicklist->count += node->count; quicklist->count += node->count;
} }
/* Create new node consisting of a pre-formed plain node.
* Used for loading RDBs where entire plain node has been stored
* to be retrieved later.
* data - the data to add (pointer becomes the responsibility of quicklist) */
void quicklistAppendPlainNode(quicklist *quicklist, unsigned char *data, size_t sz) {
__quicklistInsertPlainNode(quicklist, quicklist->tail, data, sz, 1);
}
/* Append all values of ziplist 'zl' individually into 'quicklist'. /* Append all values of ziplist 'zl' individually into 'quicklist'.
* *
* This allows us to restore old RDB ziplists into new quicklists * This allows us to restore old RDB ziplists into new quicklists
...@@ -622,7 +703,7 @@ REDIS_STATIC void __quicklistDelNode(quicklist *quicklist, ...@@ -622,7 +703,7 @@ REDIS_STATIC void __quicklistDelNode(quicklist *quicklist,
* now have compressed nodes needing to be decompressed. */ * now have compressed nodes needing to be decompressed. */
__quicklistCompress(quicklist, NULL); __quicklistCompress(quicklist, NULL);
zfree(node->zl); zfree(node->entry);
zfree(node); zfree(node);
} }
...@@ -638,7 +719,11 @@ REDIS_STATIC int quicklistDelIndex(quicklist *quicklist, quicklistNode *node, ...@@ -638,7 +719,11 @@ REDIS_STATIC int quicklistDelIndex(quicklist *quicklist, quicklistNode *node,
unsigned char **p) { unsigned char **p) {
int gone = 0; int gone = 0;
node->zl = ziplistDelete(node->zl, p); if (unlikely(QL_NODE_IS_PLAIN(node))) {
__quicklistDelNode(quicklist, node);
return 1;
}
node->entry = ziplistDelete(node->entry, p);
node->count--; node->count--;
if (node->count == 0) { if (node->count == 0) {
gone = 1; gone = 1;
...@@ -686,11 +771,33 @@ void quicklistDelEntry(quicklistIter *iter, quicklistEntry *entry) { ...@@ -686,11 +771,33 @@ void quicklistDelEntry(quicklistIter *iter, quicklistEntry *entry) {
/* Replace quicklist entry by 'data' with length 'sz'. */ /* Replace quicklist entry by 'data' with length 'sz'. */
void quicklistReplaceEntry(quicklist *quicklist, quicklistEntry *entry, void quicklistReplaceEntry(quicklist *quicklist, quicklistEntry *entry,
void *data, int sz) { void *data, size_t sz) {
/* quicklistNext() and quicklistIndex() provide an uncompressed node */ if (likely(!QL_NODE_IS_PLAIN(entry->node) && !isLargeElement(sz))) {
entry->node->zl = ziplistReplace(entry->node->zl, entry->zi, data, sz); entry->node->entry = ziplistReplace(entry->node->entry, entry->zi, data, sz);
quicklistNodeUpdateSz(entry->node); quicklistNodeUpdateSz(entry->node);
quicklistCompress(quicklist, entry->node); /* quicklistNext() and quicklistIndex() provide an uncompressed node */
quicklistCompress(quicklist, entry->node);
} else if (QL_NODE_IS_PLAIN(entry->node)) {
if (isLargeElement(sz)) {
zfree(entry->node->entry);
entry->node->entry = zmalloc(sz);
entry->node->sz = sz;
memcpy(entry->node->entry, data, sz);
quicklistCompress(quicklist, entry->node);
} else {
quicklistInsertAfter(quicklist, entry, data, sz);
__quicklistDelNode(quicklist, entry->node);
}
} else {
quicklistInsertAfter(quicklist, entry, data, sz);
if (entry->node->count == 1)
__quicklistDelNode(quicklist, entry->node);
else {
unsigned char *p = ziplistIndex(entry->node->entry, -1);
quicklistDelIndex(quicklist, entry->node, &p);
quicklistCompress(quicklist, entry->node->next);
}
}
} }
/* Replace quicklist entry at offset 'index' by 'data' with length 'sz'. /* Replace quicklist entry at offset 'index' by 'data' with length 'sz'.
...@@ -698,7 +805,7 @@ void quicklistReplaceEntry(quicklist *quicklist, quicklistEntry *entry, ...@@ -698,7 +805,7 @@ void quicklistReplaceEntry(quicklist *quicklist, quicklistEntry *entry,
* Returns 1 if replace happened. * Returns 1 if replace happened.
* Returns 0 if replace failed and no changes happened. */ * Returns 0 if replace failed and no changes happened. */
int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data, int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data,
int sz) { size_t sz) {
quicklistEntry entry; quicklistEntry entry;
if (likely(quicklistIndex(quicklist, index, &entry))) { if (likely(quicklistIndex(quicklist, index, &entry))) {
quicklistReplaceEntry(quicklist, &entry, data, sz); quicklistReplaceEntry(quicklist, &entry, data, sz);
...@@ -728,17 +835,17 @@ REDIS_STATIC quicklistNode *_quicklistZiplistMerge(quicklist *quicklist, ...@@ -728,17 +835,17 @@ REDIS_STATIC quicklistNode *_quicklistZiplistMerge(quicklist *quicklist,
quicklistDecompressNode(a); quicklistDecompressNode(a);
quicklistDecompressNode(b); quicklistDecompressNode(b);
if ((ziplistMerge(&a->zl, &b->zl))) { if ((ziplistMerge(&a->entry, &b->entry))) {
/* We merged ziplists! Now remove the unused quicklistNode. */ /* We merged ziplists! Now remove the unused quicklistNode. */
quicklistNode *keep = NULL, *nokeep = NULL; quicklistNode *keep = NULL, *nokeep = NULL;
if (!a->zl) { if (!a->entry) {
nokeep = a; nokeep = a;
keep = b; keep = b;
} else if (!b->zl) { } else if (!b->entry) {
nokeep = b; nokeep = b;
keep = a; keep = a;
} }
keep->count = ziplistLen(keep->zl); keep->count = ziplistLen(keep->entry);
quicklistNodeUpdateSz(keep); quicklistNodeUpdateSz(keep);
nokeep->count = 0; nokeep->count = 0;
...@@ -828,10 +935,10 @@ REDIS_STATIC quicklistNode *_quicklistSplitNode(quicklistNode *node, int offset, ...@@ -828,10 +935,10 @@ REDIS_STATIC quicklistNode *_quicklistSplitNode(quicklistNode *node, int offset,
size_t zl_sz = node->sz; size_t zl_sz = node->sz;
quicklistNode *new_node = quicklistCreateNode(); quicklistNode *new_node = quicklistCreateNode();
new_node->zl = zmalloc(zl_sz); new_node->entry = zmalloc(zl_sz);
/* Copy original ziplist so we can split it */ /* Copy original ziplist so we can split it */
memcpy(new_node->zl, node->zl, zl_sz); memcpy(new_node->entry, node->entry, zl_sz);
/* Ranges to be trimmed: -1 here means "continue deleting until the list ends" */ /* Ranges to be trimmed: -1 here means "continue deleting until the list ends" */
int orig_start = after ? offset + 1 : 0; int orig_start = after ? offset + 1 : 0;
...@@ -842,12 +949,12 @@ REDIS_STATIC quicklistNode *_quicklistSplitNode(quicklistNode *node, int offset, ...@@ -842,12 +949,12 @@ REDIS_STATIC quicklistNode *_quicklistSplitNode(quicklistNode *node, int offset,
D("After %d (%d); ranges: [%d, %d], [%d, %d]", after, offset, orig_start, D("After %d (%d); ranges: [%d, %d], [%d, %d]", after, offset, orig_start,
orig_extent, new_start, new_extent); orig_extent, new_start, new_extent);
node->zl = ziplistDeleteRange(node->zl, orig_start, orig_extent); node->entry = ziplistDeleteRange(node->entry, orig_start, orig_extent);
node->count = ziplistLen(node->zl); node->count = ziplistLen(node->entry);
quicklistNodeUpdateSz(node); quicklistNodeUpdateSz(node);
new_node->zl = ziplistDeleteRange(new_node->zl, new_start, new_extent); new_node->entry = ziplistDeleteRange(new_node->entry, new_start, new_extent);
new_node->count = ziplistLen(new_node->zl); new_node->count = ziplistLen(new_node->entry);
quicklistNodeUpdateSz(new_node); quicklistNodeUpdateSz(new_node);
D("After split lengths: orig (%d), new (%d)", node->count, new_node->count); D("After split lengths: orig (%d), new (%d)", node->count, new_node->count);
...@@ -864,13 +971,16 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry, ...@@ -864,13 +971,16 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry,
int fill = quicklist->fill; int fill = quicklist->fill;
quicklistNode *node = entry->node; quicklistNode *node = entry->node;
quicklistNode *new_node = NULL; quicklistNode *new_node = NULL;
assert(sz < UINT32_MAX); /* TODO: add support for quicklist nodes that are sds encoded (not zipped) */
if (!node) { if (!node) {
/* we have no reference node, so let's create only node in the list */ /* we have no reference node, so let's create only node in the list */
D("No node given!"); D("No node given!");
if (unlikely(isLargeElement(sz))) {
__quicklistInsertPlainNode(quicklist, quicklist->tail, value, sz, after);
return;
}
new_node = quicklistCreateNode(); new_node = quicklistCreateNode();
new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD); new_node->entry = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
__quicklistInsertNode(quicklist, NULL, new_node, after); __quicklistInsertNode(quicklist, NULL, new_node, after);
new_node->count++; new_node->count++;
quicklist->count++; quicklist->count++;
...@@ -902,15 +1012,29 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry, ...@@ -902,15 +1012,29 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry,
} }
} }
if (unlikely(isLargeElement(sz))) {
if (QL_NODE_IS_PLAIN(node) || (at_tail && after) || (at_head && !after)) {
__quicklistInsertPlainNode(quicklist, node, value, sz, after);
} else {
quicklistDecompressNodeForUse(node);
new_node = _quicklistSplitNode(node, entry->offset, after);
quicklistNode *entry_node = __quicklistCreatePlainNode(value, sz);
__quicklistInsertNode(quicklist, node, entry_node, after);
__quicklistInsertNode(quicklist, entry_node, new_node, after);
quicklist->count++;
}
return;
}
/* Now determine where and how to insert the new element */ /* Now determine where and how to insert the new element */
if (!full && after) { if (!full && after) {
D("Not full, inserting after current position."); D("Not full, inserting after current position.");
quicklistDecompressNodeForUse(node); quicklistDecompressNodeForUse(node);
unsigned char *next = ziplistNext(node->zl, entry->zi); unsigned char *next = ziplistNext(node->entry, entry->zi);
if (next == NULL) { if (next == NULL) {
node->zl = ziplistPush(node->zl, value, sz, ZIPLIST_TAIL); node->entry = ziplistPush(node->entry, value, sz, ZIPLIST_TAIL);
} else { } else {
node->zl = ziplistInsert(node->zl, next, value, sz); node->entry = ziplistInsert(node->entry, next, value, sz);
} }
node->count++; node->count++;
quicklistNodeUpdateSz(node); quicklistNodeUpdateSz(node);
...@@ -918,7 +1042,7 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry, ...@@ -918,7 +1042,7 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry,
} else if (!full && !after) { } else if (!full && !after) {
D("Not full, inserting before current position."); D("Not full, inserting before current position.");
quicklistDecompressNodeForUse(node); quicklistDecompressNodeForUse(node);
node->zl = ziplistInsert(node->zl, entry->zi, value, sz); node->entry = ziplistInsert(node->entry, entry->zi, value, sz);
node->count++; node->count++;
quicklistNodeUpdateSz(node); quicklistNodeUpdateSz(node);
quicklistRecompressOnly(quicklist, node); quicklistRecompressOnly(quicklist, node);
...@@ -928,7 +1052,7 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry, ...@@ -928,7 +1052,7 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry,
D("Full and tail, but next isn't full; inserting next node head"); D("Full and tail, but next isn't full; inserting next node head");
new_node = node->next; new_node = node->next;
quicklistDecompressNodeForUse(new_node); quicklistDecompressNodeForUse(new_node);
new_node->zl = ziplistPush(new_node->zl, value, sz, ZIPLIST_HEAD); new_node->entry = ziplistPush(new_node->entry, value, sz, ZIPLIST_HEAD);
new_node->count++; new_node->count++;
quicklistNodeUpdateSz(new_node); quicklistNodeUpdateSz(new_node);
quicklistRecompressOnly(quicklist, new_node); quicklistRecompressOnly(quicklist, new_node);
...@@ -938,7 +1062,7 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry, ...@@ -938,7 +1062,7 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry,
D("Full and head, but prev isn't full, inserting prev node tail"); D("Full and head, but prev isn't full, inserting prev node tail");
new_node = node->prev; new_node = node->prev;
quicklistDecompressNodeForUse(new_node); quicklistDecompressNodeForUse(new_node);
new_node->zl = ziplistPush(new_node->zl, value, sz, ZIPLIST_TAIL); new_node->entry = ziplistPush(new_node->entry, value, sz, ZIPLIST_TAIL);
new_node->count++; new_node->count++;
quicklistNodeUpdateSz(new_node); quicklistNodeUpdateSz(new_node);
quicklistRecompressOnly(quicklist, new_node); quicklistRecompressOnly(quicklist, new_node);
...@@ -948,7 +1072,7 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry, ...@@ -948,7 +1072,7 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry,
* - create new node and attach to quicklist */ * - create new node and attach to quicklist */
D("\tprovisioning new node..."); D("\tprovisioning new node...");
new_node = quicklistCreateNode(); new_node = quicklistCreateNode();
new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD); new_node->entry = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
new_node->count++; new_node->count++;
quicklistNodeUpdateSz(new_node); quicklistNodeUpdateSz(new_node);
__quicklistInsertNode(quicklist, node, new_node, after); __quicklistInsertNode(quicklist, node, new_node, after);
...@@ -958,7 +1082,7 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry, ...@@ -958,7 +1082,7 @@ REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry,
D("\tsplitting node..."); D("\tsplitting node...");
quicklistDecompressNodeForUse(node); quicklistDecompressNodeForUse(node);
new_node = _quicklistSplitNode(node, entry->offset, after); new_node = _quicklistSplitNode(node, entry->offset, after);
new_node->zl = ziplistPush(new_node->zl, value, sz, new_node->entry = ziplistPush(new_node->entry, value, sz,
after ? ZIPLIST_HEAD : ZIPLIST_TAIL); after ? ZIPLIST_HEAD : ZIPLIST_TAIL);
new_node->count++; new_node->count++;
quicklistNodeUpdateSz(new_node); quicklistNodeUpdateSz(new_node);
...@@ -1046,11 +1170,11 @@ int quicklistDelRange(quicklist *quicklist, const long start, ...@@ -1046,11 +1170,11 @@ int quicklistDelRange(quicklist *quicklist, const long start,
"node count: %u", "node count: %u",
extent, del, entry.offset, delete_entire_node, node->count); extent, del, entry.offset, delete_entire_node, node->count);
if (delete_entire_node) { if (delete_entire_node || QL_NODE_IS_PLAIN(node)) {
__quicklistDelNode(quicklist, node); __quicklistDelNode(quicklist, node);
} else { } else {
quicklistDecompressNodeForUse(node); quicklistDecompressNodeForUse(node);
node->zl = ziplistDeleteRange(node->zl, entry.offset, del); node->entry = ziplistDeleteRange(node->entry, entry.offset, del);
quicklistNodeUpdateSz(node); quicklistNodeUpdateSz(node);
node->count -= del; node->count -= del;
quicklist->count -= del; quicklist->count -= del;
...@@ -1068,9 +1192,12 @@ int quicklistDelRange(quicklist *quicklist, const long start, ...@@ -1068,9 +1192,12 @@ int quicklistDelRange(quicklist *quicklist, const long start,
return 1; return 1;
} }
/* Passthrough to ziplistCompare() */ /* compare between a two entries */
int quicklistCompare(unsigned char *p1, unsigned char *p2, int p2_len) { int quicklistCompare(quicklistEntry* entry, unsigned char *p2, const size_t p2_len) {
return ziplistCompare(p1, p2, p2_len); if (unlikely(QL_NODE_IS_PLAIN(entry->node))) {
return ((entry->sz == p2_len) && (memcmp(entry->value, p2, p2_len) == 0));
}
return ziplistCompare(entry->zi, p2, p2_len);
} }
/* Returns a quicklist iterator 'iter'. After the initialization every /* Returns a quicklist iterator 'iter'. After the initialization every
...@@ -1163,10 +1290,16 @@ int quicklistNext(quicklistIter *iter, quicklistEntry *entry) { ...@@ -1163,10 +1290,16 @@ int quicklistNext(quicklistIter *iter, quicklistEntry *entry) {
unsigned char *(*nextFn)(unsigned char *, unsigned char *) = NULL; unsigned char *(*nextFn)(unsigned char *, unsigned char *) = NULL;
int offset_update = 0; int offset_update = 0;
int plain = QL_NODE_IS_PLAIN(iter->current);
if (!iter->zi) { if (!iter->zi) {
/* If !zi, use current index. */ /* If !zi, use current index. */
quicklistDecompressNodeForUse(iter->current); quicklistDecompressNodeForUse(iter->current);
iter->zi = ziplistIndex(iter->current->zl, iter->offset); if (unlikely(plain))
iter->zi = iter->current->entry;
else
iter->zi = ziplistIndex(iter->current->entry, iter->offset);
} else if (unlikely(plain)) {
iter->zi = NULL;
} else { } else {
/* else, use existing iterator offset and get prev/next as necessary. */ /* else, use existing iterator offset and get prev/next as necessary. */
if (iter->direction == AL_START_HEAD) { if (iter->direction == AL_START_HEAD) {
...@@ -1176,7 +1309,7 @@ int quicklistNext(quicklistIter *iter, quicklistEntry *entry) { ...@@ -1176,7 +1309,7 @@ int quicklistNext(quicklistIter *iter, quicklistEntry *entry) {
nextFn = ziplistPrev; nextFn = ziplistPrev;
offset_update = -1; offset_update = -1;
} }
iter->zi = nextFn(iter->current->zl, iter->zi); iter->zi = nextFn(iter->current->entry, iter->zi);
iter->offset += offset_update; iter->offset += offset_update;
} }
...@@ -1184,8 +1317,15 @@ int quicklistNext(quicklistIter *iter, quicklistEntry *entry) { ...@@ -1184,8 +1317,15 @@ int quicklistNext(quicklistIter *iter, quicklistEntry *entry) {
entry->offset = iter->offset; entry->offset = iter->offset;
if (iter->zi) { if (iter->zi) {
if (unlikely(plain)) {
entry->value = entry->node->entry;
entry->sz = entry->node->sz;
return 1;
}
/* Populate value from existing ziplist position */ /* Populate value from existing ziplist position */
ziplistGet(entry->zi, &entry->value, &entry->sz, &entry->longval); unsigned int sz = 0;
ziplistGet(entry->zi, &entry->value, &sz, &entry->longval);
entry->sz = sz;
return 1; return 1;
} else { } else {
/* We ran out of ziplist entries. /* We ran out of ziplist entries.
...@@ -1228,19 +1368,20 @@ quicklist *quicklistDup(quicklist *orig) { ...@@ -1228,19 +1368,20 @@ quicklist *quicklistDup(quicklist *orig) {
quicklistNode *node = quicklistCreateNode(); quicklistNode *node = quicklistCreateNode();
if (current->encoding == QUICKLIST_NODE_ENCODING_LZF) { if (current->encoding == QUICKLIST_NODE_ENCODING_LZF) {
quicklistLZF *lzf = (quicklistLZF *)current->zl; quicklistLZF *lzf = (quicklistLZF *)current->entry;
size_t lzf_sz = sizeof(*lzf) + lzf->sz; size_t lzf_sz = sizeof(*lzf) + lzf->sz;
node->zl = zmalloc(lzf_sz); node->entry = zmalloc(lzf_sz);
memcpy(node->zl, current->zl, lzf_sz); memcpy(node->entry, current->entry, lzf_sz);
} else if (current->encoding == QUICKLIST_NODE_ENCODING_RAW) { } else if (current->encoding == QUICKLIST_NODE_ENCODING_RAW) {
node->zl = zmalloc(current->sz); node->entry = zmalloc(current->sz);
memcpy(node->zl, current->zl, current->sz); memcpy(node->entry, current->entry, current->sz);
} }
node->count = current->count; node->count = current->count;
copy->count += node->count; copy->count += node->count;
node->sz = current->sz; node->sz = current->sz;
node->encoding = current->encoding; node->encoding = current->encoding;
node->container = current->container;
_quicklistInsertNodeAfter(copy, copy->tail, node); _quicklistInsertNodeAfter(copy, copy->tail, node);
} }
...@@ -1311,21 +1452,46 @@ int quicklistIndex(const quicklist *quicklist, const long long idx, ...@@ -1311,21 +1452,46 @@ int quicklistIndex(const quicklist *quicklist, const long long idx,
} }
quicklistDecompressNodeForUse(entry->node); quicklistDecompressNodeForUse(entry->node);
entry->zi = ziplistIndex(entry->node->zl, entry->offset);
if (!ziplistGet(entry->zi, &entry->value, &entry->sz, &entry->longval)) if (unlikely(QL_NODE_IS_PLAIN(entry->node))) {
entry->value = entry->node->entry;
entry->sz = entry->node->sz;
return 1;
}
entry->zi = ziplistIndex(entry->node->entry, entry->offset);
unsigned int sz = 0;
if (!ziplistGet(entry->zi, &entry->value, &sz, &entry->longval))
assert(0); /* This can happen on corrupt ziplist with fake entry count. */ assert(0); /* This can happen on corrupt ziplist with fake entry count. */
/* The caller will use our result, so we don't re-compress here. /* The caller will use our result, so we don't re-compress here.
* The caller can recompress or delete the node as needed. */ * The caller can recompress or delete the node as needed. */
entry->sz = sz;
return 1; return 1;
} }
static void quicklistRotatePlain(quicklist *quicklist) {
quicklistNode *new_head = quicklist->tail;
quicklistNode *new_tail = quicklist->tail->prev;
quicklist->head->prev = new_head;
new_tail->next = NULL;
new_head->next = quicklist->head;
new_head->prev = NULL;
quicklist->head = new_head;
quicklist->tail = new_tail;
}
/* Rotate quicklist by moving the tail element to the head. */ /* Rotate quicklist by moving the tail element to the head. */
void quicklistRotate(quicklist *quicklist) { void quicklistRotate(quicklist *quicklist) {
if (quicklist->count <= 1) if (quicklist->count <= 1)
return; return;
if (unlikely(QL_NODE_IS_PLAIN(quicklist->tail))) {
quicklistRotatePlain(quicklist);
return;
}
/* First, get the tail entry */ /* First, get the tail entry */
unsigned char *p = ziplistIndex(quicklist->tail->zl, -1); unsigned char *p = ziplistIndex(quicklist->tail->entry, -1);
unsigned char *value, *tmp; unsigned char *value, *tmp;
long long longval; long long longval;
unsigned int sz; unsigned int sz;
...@@ -1353,7 +1519,7 @@ void quicklistRotate(quicklist *quicklist) { ...@@ -1353,7 +1519,7 @@ void quicklistRotate(quicklist *quicklist) {
* tail ziplist and PushHead() could have reallocated our single ziplist, * tail ziplist and PushHead() could have reallocated our single ziplist,
* which would make our pre-existing 'p' unusable. */ * which would make our pre-existing 'p' unusable. */
if (quicklist->len == 1) { if (quicklist->len == 1) {
p = ziplistIndex(quicklist->tail->zl, -1); p = ziplistIndex(quicklist->tail->entry, -1);
} }
/* Remove tail entry. */ /* Remove tail entry. */
...@@ -1372,8 +1538,8 @@ void quicklistRotate(quicklist *quicklist) { ...@@ -1372,8 +1538,8 @@ void quicklistRotate(quicklist *quicklist) {
* Return value of 1 means check 'data' and 'sval' for values. * Return value of 1 means check 'data' and 'sval' for values.
* If 'data' is set, use 'data' and 'sz'. Otherwise, use 'sval'. */ * If 'data' is set, use 'data' and 'sz'. Otherwise, use 'sval'. */
int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data, int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
unsigned int *sz, long long *sval, size_t *sz, long long *sval,
void *(*saver)(unsigned char *data, unsigned int sz)) { void *(*saver)(unsigned char *data, size_t sz)) {
unsigned char *p; unsigned char *p;
unsigned char *vstr; unsigned char *vstr;
unsigned int vlen; unsigned int vlen;
...@@ -1399,7 +1565,16 @@ int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data, ...@@ -1399,7 +1565,16 @@ int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
return 0; return 0;
} }
p = ziplistIndex(node->zl, pos); if (unlikely(QL_NODE_IS_PLAIN(node))) {
if (data)
*data = saver(node->entry, node->sz);
if (sz)
*sz = node->sz;
quicklistDelIndex(quicklist, node, NULL);
return 1;
}
p = ziplistIndex(node->entry, pos);
if (ziplistGet(p, &vstr, &vlen, &vlong)) { if (ziplistGet(p, &vstr, &vlen, &vlong)) {
if (vstr) { if (vstr) {
if (data) if (data)
...@@ -1419,7 +1594,7 @@ int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data, ...@@ -1419,7 +1594,7 @@ int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
} }
/* Return a malloc'd copy of data passed in */ /* Return a malloc'd copy of data passed in */
REDIS_STATIC void *_quicklistSaver(unsigned char *data, unsigned int sz) { REDIS_STATIC void *_quicklistSaver(unsigned char *data, size_t sz) {
unsigned char *vstr; unsigned char *vstr;
if (data) { if (data) {
vstr = zmalloc(sz); vstr = zmalloc(sz);
...@@ -1433,9 +1608,9 @@ REDIS_STATIC void *_quicklistSaver(unsigned char *data, unsigned int sz) { ...@@ -1433,9 +1608,9 @@ REDIS_STATIC void *_quicklistSaver(unsigned char *data, unsigned int sz) {
* *
* Returns malloc'd value from quicklist */ * Returns malloc'd value from quicklist */
int quicklistPop(quicklist *quicklist, int where, unsigned char **data, int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
unsigned int *sz, long long *slong) { size_t *sz, long long *slong) {
unsigned char *vstr; unsigned char *vstr;
unsigned int vlen; size_t vlen;
long long vlong; long long vlong;
if (quicklist->count == 0) if (quicklist->count == 0)
return 0; return 0;
...@@ -1613,7 +1788,8 @@ static int _itrprintr(quicklist *ql, int print, int forward) { ...@@ -1613,7 +1788,8 @@ static int _itrprintr(quicklist *ql, int print, int forward) {
prev = entry.node; prev = entry.node;
} }
if (print) { if (print) {
printf("[%3d (%2d)]: [%.*s] (%lld)\n", i, p, entry.sz, int size = (entry.sz > (1<<20)) ? 1<<20 : entry.sz;
printf("[%3d (%2d)]: [%.*s] (%lld)\n", i, p, size,
(char *)entry.value, entry.longval); (char *)entry.value, entry.longval);
} }
i++; i++;
...@@ -1671,18 +1847,18 @@ static int _ql_verify(quicklist *ql, uint32_t len, uint32_t count, ...@@ -1671,18 +1847,18 @@ static int _ql_verify(quicklist *ql, uint32_t len, uint32_t count,
} }
if (ql->head && head_count != ql->head->count && if (ql->head && head_count != ql->head->count &&
head_count != ziplistLen(ql->head->zl)) { head_count != ziplistLen(ql->head->entry)) {
yell("quicklist head count wrong: expected %d, " yell("quicklist head count wrong: expected %d, "
"got cached %d vs. actual %d", "got cached %d vs. actual %d",
head_count, ql->head->count, ziplistLen(ql->head->zl)); head_count, ql->head->count, ziplistLen(ql->head->entry));
errors++; errors++;
} }
if (ql->tail && tail_count != ql->tail->count && if (ql->tail && tail_count != ql->tail->count &&
tail_count != ziplistLen(ql->tail->zl)) { tail_count != ziplistLen(ql->tail->entry)) {
yell("quicklist tail count wrong: expected %d, " yell("quicklist tail count wrong: expected %d, "
"got cached %u vs. actual %d", "got cached %u vs. actual %d",
tail_count, ql->tail->count, ziplistLen(ql->tail->zl)); tail_count, ql->tail->count, ziplistLen(ql->tail->entry));
errors++; errors++;
} }
...@@ -1696,7 +1872,7 @@ static int _ql_verify(quicklist *ql, uint32_t len, uint32_t count, ...@@ -1696,7 +1872,7 @@ static int _ql_verify(quicklist *ql, uint32_t len, uint32_t count,
if (node->encoding != QUICKLIST_NODE_ENCODING_RAW) { if (node->encoding != QUICKLIST_NODE_ENCODING_RAW) {
yell("Incorrect compression: node %d is " yell("Incorrect compression: node %d is "
"compressed at depth %d ((%u, %u); total " "compressed at depth %d ((%u, %u); total "
"nodes: %lu; size: %u; recompress: %d)", "nodes: %lu; size: %zu; recompress: %d)",
at, ql->compress, low_raw, high_raw, ql->len, node->sz, at, ql->compress, low_raw, high_raw, ql->len, node->sz,
node->recompress); node->recompress);
errors++; errors++;
...@@ -1706,7 +1882,7 @@ static int _ql_verify(quicklist *ql, uint32_t len, uint32_t count, ...@@ -1706,7 +1882,7 @@ static int _ql_verify(quicklist *ql, uint32_t len, uint32_t count,
!node->attempted_compress) { !node->attempted_compress) {
yell("Incorrect non-compression: node %d is NOT " yell("Incorrect non-compression: node %d is NOT "
"compressed at depth %d ((%u, %u); total " "compressed at depth %d ((%u, %u); total "
"nodes: %lu; size: %u; recompress: %d; attempted: %d)", "nodes: %lu; size: %zu; recompress: %d; attempted: %d)",
at, ql->compress, low_raw, high_raw, ql->len, node->sz, at, ql->compress, low_raw, high_raw, ql->len, node->sz,
node->recompress, node->attempted_compress); node->recompress, node->attempted_compress);
errors++; errors++;
...@@ -1829,6 +2005,74 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -1829,6 +2005,74 @@ int quicklistTest(int argc, char *argv[], int accurate) {
quicklistRelease(ql); quicklistRelease(ql);
} }
TEST("Comprassion Plain node") {
quicklistisSetPackedThreshold(1);
quicklist *ql = quicklistNew(-2, 1);
for (int i = 0; i < 500; i++)
/* Set to 256 to allow the node to be triggered to compress,
* if it is less than 48(nocompress), the test will be successful. */
quicklistPushHead(ql, genstr("hello", i), 256);
quicklistIter *iter = quicklistGetIterator(ql, AL_START_TAIL);
quicklistEntry entry;
int i = 0;
while (quicklistNext(iter, &entry)) {
char *h = genstr("hello", i);
if (strcmp((char *)entry.value, h))
ERR("value [%s] didn't match [%s] at position %d",
entry.value, h, i);
i++;
}
quicklistReleaseIterator(iter);
quicklistRelease(ql);
}
TEST("NEXT plain node")
{
packed_threshold = 3;
quicklist *ql = quicklistNew(-2, options[_i]);
char *strings[] = {"hello1", "hello2", "h3", "h4", "hello5"};
for (int i = 0; i < 5; ++i)
quicklistPushHead(ql, strings[i], strlen(strings[i]));
quicklistEntry entry;
quicklistIter *iter = quicklistGetIterator(ql, AL_START_TAIL);
int j = 0;
while(quicklistNext(iter, &entry) != 0) {
assert(strncmp(strings[j], (char *)entry.value, strlen(strings[j])) == 0);
j++;
}
quicklistReleaseIterator(iter);
quicklistRelease(ql);
}
TEST("rotate plain node ") {
unsigned char *data = NULL;
size_t sz;
long long lv;
int i =0;
packed_threshold = 5;
quicklist *ql = quicklistNew(-2, options[_i]);
quicklistPushHead(ql, "hello1", 6);
quicklistPushHead(ql, "hello4", 6);
quicklistPushHead(ql, "hello3", 6);
quicklistPushHead(ql, "hello2", 6);
quicklistRotate(ql);
for(i = 1 ; i < 5; i++) {
quicklistPop(ql, QUICKLIST_HEAD, &data, &sz, &lv);
int temp_char = data[5];
zfree(data);
assert(temp_char == ('0' + i));
}
ql_verify(ql, 0, 0, 0, 0);
quicklistRelease(ql);
packed_threshold = (1 << 30);
}
TEST("rotate one val once") { TEST("rotate one val once") {
for (int f = 0; f < fill_count; f++) { for (int f = 0; f < fill_count; f++) {
quicklist *ql = quicklistNew(fills[f], options[_i]); quicklist *ql = quicklistNew(fills[f], options[_i]);
...@@ -1877,15 +2121,17 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -1877,15 +2121,17 @@ int quicklistTest(int argc, char *argv[], int accurate) {
char *populate = genstr("hello", 331); char *populate = genstr("hello", 331);
quicklistPushHead(ql, populate, 32); quicklistPushHead(ql, populate, 32);
unsigned char *data; unsigned char *data;
unsigned int sz; size_t sz;
long long lv; long long lv;
ql_info(ql); ql_info(ql);
assert(quicklistPop(ql, QUICKLIST_HEAD, &data, &sz, &lv)); assert(quicklistPop(ql, QUICKLIST_HEAD, &data, &sz, &lv));
assert(data != NULL); assert(data != NULL);
assert(sz == 32); assert(sz == 32);
if (strcmp(populate, (char *)data)) if (strcmp(populate, (char *)data)) {
ERR("Pop'd value (%.*s) didn't equal original value (%s)", sz, int size = sz;
ERR("Pop'd value (%.*s) didn't equal original value (%s)", size,
data, populate); data, populate);
}
zfree(data); zfree(data);
ql_verify(ql, 0, 0, 0, 0); ql_verify(ql, 0, 0, 0, 0);
quicklistRelease(ql); quicklistRelease(ql);
...@@ -1895,7 +2141,7 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -1895,7 +2141,7 @@ int quicklistTest(int argc, char *argv[], int accurate) {
quicklist *ql = quicklistNew(-2, options[_i]); quicklist *ql = quicklistNew(-2, options[_i]);
quicklistPushHead(ql, "55513", 5); quicklistPushHead(ql, "55513", 5);
unsigned char *data; unsigned char *data;
unsigned int sz; size_t sz;
long long lv; long long lv;
ql_info(ql); ql_info(ql);
assert(quicklistPop(ql, QUICKLIST_HEAD, &data, &sz, &lv)); assert(quicklistPop(ql, QUICKLIST_HEAD, &data, &sz, &lv));
...@@ -1912,15 +2158,17 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -1912,15 +2158,17 @@ int quicklistTest(int argc, char *argv[], int accurate) {
ql_info(ql); ql_info(ql);
for (int i = 0; i < 500; i++) { for (int i = 0; i < 500; i++) {
unsigned char *data; unsigned char *data;
unsigned int sz; size_t sz;
long long lv; long long lv;
int ret = quicklistPop(ql, QUICKLIST_HEAD, &data, &sz, &lv); int ret = quicklistPop(ql, QUICKLIST_HEAD, &data, &sz, &lv);
assert(ret == 1); assert(ret == 1);
assert(data != NULL); assert(data != NULL);
assert(sz == 32); assert(sz == 32);
if (strcmp(genstr("hello", 499 - i), (char *)data)) if (strcmp(genstr("hello", 499 - i), (char *)data)) {
int size = sz;
ERR("Pop'd value (%.*s) didn't equal original value (%s)", ERR("Pop'd value (%.*s) didn't equal original value (%s)",
sz, data, genstr("hello", 499 - i)); size, data, genstr("hello", 499 - i));
}
zfree(data); zfree(data);
} }
ql_verify(ql, 0, 0, 0, 0); ql_verify(ql, 0, 0, 0, 0);
...@@ -1933,17 +2181,19 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -1933,17 +2181,19 @@ int quicklistTest(int argc, char *argv[], int accurate) {
quicklistPushHead(ql, genstr("hello", i), 32); quicklistPushHead(ql, genstr("hello", i), 32);
for (int i = 0; i < 5000; i++) { for (int i = 0; i < 5000; i++) {
unsigned char *data; unsigned char *data;
unsigned int sz; size_t sz;
long long lv; long long lv;
int ret = quicklistPop(ql, QUICKLIST_HEAD, &data, &sz, &lv); int ret = quicklistPop(ql, QUICKLIST_HEAD, &data, &sz, &lv);
if (i < 500) { if (i < 500) {
assert(ret == 1); assert(ret == 1);
assert(data != NULL); assert(data != NULL);
assert(sz == 32); assert(sz == 32);
if (strcmp(genstr("hello", 499 - i), (char *)data)) if (strcmp(genstr("hello", 499 - i), (char *)data)) {
int size = sz;
ERR("Pop'd value (%.*s) didn't equal original value " ERR("Pop'd value (%.*s) didn't equal original value "
"(%s)", "(%s)",
sz, data, genstr("hello", 499 - i)); size, data, genstr("hello", 499 - i));
}
zfree(data); zfree(data);
} else { } else {
assert(ret == 0); assert(ret == 0);
...@@ -2008,7 +2258,8 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2008,7 +2258,8 @@ int quicklistTest(int argc, char *argv[], int accurate) {
/* verify results */ /* verify results */
quicklistIndex(ql, 0, &entry); quicklistIndex(ql, 0, &entry);
if (strncmp((char *)entry.value, "abc", 3)) { if (strncmp((char *)entry.value, "abc", 3)) {
ERR("Value 0 didn't match, instead got: %.*s", entry.sz, int sz = entry.sz;
ERR("Value 0 didn't match, instead got: %.*s", sz,
entry.value); entry.value);
} }
quicklistRelease(ql); quicklistRelease(ql);
...@@ -2023,8 +2274,9 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2023,8 +2274,9 @@ int quicklistTest(int argc, char *argv[], int accurate) {
/* verify results */ /* verify results */
quicklistIndex(ql, 0, &entry); quicklistIndex(ql, 0, &entry);
int sz = entry.sz;
if (strncmp((char *)entry.value, "abc", 3)) { if (strncmp((char *)entry.value, "abc", 3)) {
ERR("Value 0 didn't match, instead got: %.*s", entry.sz, ERR("Value 0 didn't match, instead got: %.*s", sz,
entry.value); entry.value);
} }
quicklistRelease(ql); quicklistRelease(ql);
...@@ -2040,13 +2292,15 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2040,13 +2292,15 @@ int quicklistTest(int argc, char *argv[], int accurate) {
/* verify results */ /* verify results */
quicklistIndex(ql, 0, &entry); quicklistIndex(ql, 0, &entry);
int sz = entry.sz;
if (strncmp((char *)entry.value, "hello", 5)) { if (strncmp((char *)entry.value, "hello", 5)) {
ERR("Value 0 didn't match, instead got: %.*s", entry.sz, ERR("Value 0 didn't match, instead got: %.*s", sz,
entry.value); entry.value);
} }
quicklistIndex(ql, 1, &entry); quicklistIndex(ql, 1, &entry);
sz = entry.sz;
if (strncmp((char *)entry.value, "abc", 3)) { if (strncmp((char *)entry.value, "abc", 3)) {
ERR("Value 1 didn't match, instead got: %.*s", entry.sz, ERR("Value 1 didn't match, instead got: %.*s", sz,
entry.value); entry.value);
} }
...@@ -2063,13 +2317,15 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2063,13 +2317,15 @@ int quicklistTest(int argc, char *argv[], int accurate) {
/* verify results */ /* verify results */
quicklistIndex(ql, 0, &entry); quicklistIndex(ql, 0, &entry);
int sz = entry.sz;
if (strncmp((char *)entry.value, "abc", 3)) { if (strncmp((char *)entry.value, "abc", 3)) {
ERR("Value 0 didn't match, instead got: %.*s", entry.sz, ERR("Value 0 didn't match, instead got: %.*s", sz,
entry.value); entry.value);
} }
quicklistIndex(ql, 1, &entry); quicklistIndex(ql, 1, &entry);
sz = entry.sz;
if (strncmp((char *)entry.value, "hello", 5)) { if (strncmp((char *)entry.value, "hello", 5)) {
ERR("Value 1 didn't match, instead got: %.*s", entry.sz, ERR("Value 1 didn't match, instead got: %.*s", sz,
entry.value); entry.value);
} }
...@@ -2129,28 +2385,30 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2129,28 +2385,30 @@ int quicklistTest(int argc, char *argv[], int accurate) {
/* verify results */ /* verify results */
quicklistIndex(ql, 0, &entry); quicklistIndex(ql, 0, &entry);
int sz = entry.sz;
if (strncmp((char *)entry.value, "abc", 3)) if (strncmp((char *)entry.value, "abc", 3))
ERR("Value 0 didn't match, instead got: %.*s", entry.sz, ERR("Value 0 didn't match, instead got: %.*s", sz,
entry.value); entry.value);
quicklistIndex(ql, 1, &entry); quicklistIndex(ql, 1, &entry);
if (strncmp((char *)entry.value, "def", 3)) if (strncmp((char *)entry.value, "def", 3))
ERR("Value 1 didn't match, instead got: %.*s", entry.sz, ERR("Value 1 didn't match, instead got: %.*s", sz,
entry.value); entry.value);
quicklistIndex(ql, 2, &entry); quicklistIndex(ql, 2, &entry);
if (strncmp((char *)entry.value, "bar", 3)) if (strncmp((char *)entry.value, "bar", 3))
ERR("Value 2 didn't match, instead got: %.*s", entry.sz, ERR("Value 2 didn't match, instead got: %.*s", sz,
entry.value); entry.value);
quicklistIndex(ql, 3, &entry); quicklistIndex(ql, 3, &entry);
if (strncmp((char *)entry.value, "bob", 3)) if (strncmp((char *)entry.value, "bob", 3))
ERR("Value 3 didn't match, instead got: %.*s", entry.sz, ERR("Value 3 didn't match, instead got: %.*s", sz,
entry.value); entry.value);
quicklistIndex(ql, 4, &entry); quicklistIndex(ql, 4, &entry);
if (strncmp((char *)entry.value, "foo", 3)) if (strncmp((char *)entry.value, "foo", 3))
ERR("Value 4 didn't match, instead got: %.*s", entry.sz, ERR("Value 4 didn't match, instead got: %.*s", sz,
entry.value); entry.value);
quicklistIndex(ql, 5, &entry); quicklistIndex(ql, 5, &entry);
if (strncmp((char *)entry.value, "zoo", 3)) if (strncmp((char *)entry.value, "zoo", 3))
ERR("Value 5 didn't match, instead got: %.*s", entry.sz, ERR("Value 5 didn't match, instead got: %.*s", sz,
entry.value); entry.value);
quicklistReleaseIterator(iter); quicklistReleaseIterator(iter);
quicklistRelease(ql); quicklistRelease(ql);
...@@ -2276,8 +2534,9 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2276,8 +2534,9 @@ int quicklistTest(int argc, char *argv[], int accurate) {
for (int i = 0; i < 50; i++) for (int i = 0; i < 50; i++)
quicklistPushTail(ql, genstr("hello", i + 1), 32); quicklistPushTail(ql, genstr("hello", i + 1), 32);
quicklistEntry entry; quicklistEntry entry;
int sz = entry.sz;
if (quicklistIndex(ql, 50, &entry)) if (quicklistIndex(ql, 50, &entry))
ERR("Index found at 50 with 50 list: %.*s", entry.sz, ERR("Index found at 50 with 50 list: %.*s", sz,
entry.value); entry.value);
quicklistRelease(ql); quicklistRelease(ql);
} }
...@@ -2469,7 +2728,7 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2469,7 +2728,7 @@ int quicklistTest(int argc, char *argv[], int accurate) {
quicklistEntry entry; quicklistEntry entry;
int i = 0; int i = 0;
while (quicklistNext(iter, &entry)) { while (quicklistNext(iter, &entry)) {
if (quicklistCompare(entry.zi, (unsigned char *)"bar", 3)) { if (quicklistCompare(&entry, (unsigned char *)"bar", 3)) {
quicklistDelEntry(iter, &entry); quicklistDelEntry(iter, &entry);
} }
i++; i++;
...@@ -2482,9 +2741,10 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2482,9 +2741,10 @@ int quicklistTest(int argc, char *argv[], int accurate) {
while (quicklistNext(iter, &entry)) { while (quicklistNext(iter, &entry)) {
/* Result must be: abc, foo, foobar, foobared, zap, test, /* Result must be: abc, foo, foobar, foobared, zap, test,
* foo */ * foo */
int sz = entry.sz;
if (strncmp((char *)entry.value, result[i], entry.sz)) { if (strncmp((char *)entry.value, result[i], entry.sz)) {
ERR("No match at position %d, got %.*s instead of %s", ERR("No match at position %d, got %.*s instead of %s",
i, entry.sz, entry.value, result[i]); i, sz, entry.value, result[i]);
} }
i++; i++;
} }
...@@ -2497,7 +2757,7 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2497,7 +2757,7 @@ int quicklistTest(int argc, char *argv[], int accurate) {
i = 0; i = 0;
int del = 2; int del = 2;
while (quicklistNext(iter, &entry)) { while (quicklistNext(iter, &entry)) {
if (quicklistCompare(entry.zi, (unsigned char *)"foo", 3)) { if (quicklistCompare(&entry, (unsigned char *)"foo", 3)) {
quicklistDelEntry(iter, &entry); quicklistDelEntry(iter, &entry);
del--; del--;
} }
...@@ -2517,10 +2777,11 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2517,10 +2777,11 @@ int quicklistTest(int argc, char *argv[], int accurate) {
while (quicklistNext(iter, &entry)) { while (quicklistNext(iter, &entry)) {
/* Result must be: abc, foo, foobar, foobared, zap, test, /* Result must be: abc, foo, foobar, foobared, zap, test,
* foo */ * foo */
int sz = entry.sz;
if (strncmp((char *)entry.value, resultB[resB - 1 - i], if (strncmp((char *)entry.value, resultB[resB - 1 - i],
entry.sz)) { sz)) {
ERR("No match at position %d, got %.*s instead of %s", ERR("No match at position %d, got %.*s instead of %s",
i, entry.sz, entry.value, resultB[resB - 1 - i]); i, sz, entry.value, resultB[resB - 1 - i]);
} }
i++; i++;
} }
...@@ -2543,7 +2804,7 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2543,7 +2804,7 @@ int quicklistTest(int argc, char *argv[], int accurate) {
quicklistIter *iter = quicklistGetIterator(ql, AL_START_TAIL); quicklistIter *iter = quicklistGetIterator(ql, AL_START_TAIL);
int i = 0; int i = 0;
while (quicklistNext(iter, &entry)) { while (quicklistNext(iter, &entry)) {
if (quicklistCompare(entry.zi, (unsigned char *)"hij", 3)) { if (quicklistCompare(&entry, (unsigned char *)"hij", 3)) {
quicklistDelEntry(iter, &entry); quicklistDelEntry(iter, &entry);
} }
i++; i++;
...@@ -2558,7 +2819,7 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2558,7 +2819,7 @@ int quicklistTest(int argc, char *argv[], int accurate) {
i = 0; i = 0;
char *vals[] = {"abc", "def", "jkl", "oop"}; char *vals[] = {"abc", "def", "jkl", "oop"};
while (quicklistNext(iter, &entry)) { while (quicklistNext(iter, &entry)) {
if (!quicklistCompare(entry.zi, (unsigned char *)vals[i], if (!quicklistCompare(&entry, (unsigned char *)vals[i],
3)) { 3)) {
ERR("Value at %d didn't match %s\n", i, vals[i]); ERR("Value at %d didn't match %s\n", i, vals[i]);
} }
...@@ -2652,9 +2913,10 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2652,9 +2913,10 @@ int quicklistTest(int argc, char *argv[], int accurate) {
ERR("B! got instead: %lld", entry.longval); ERR("B! got instead: %lld", entry.longval);
quicklistPushTail(ql, "bobobob", 7); quicklistPushTail(ql, "bobobob", 7);
quicklistIndex(ql, -1, &entry); quicklistIndex(ql, -1, &entry);
int sz = entry.sz;
if (strncmp((char *)entry.value, "bobobob", 7)) if (strncmp((char *)entry.value, "bobobob", 7))
ERR("Tail doesn't match bobobob, it's %.*s instead", ERR("Tail doesn't match bobobob, it's %.*s instead",
entry.sz, entry.value); sz, entry.value);
for (int i = 0; i < 12; i++) { for (int i = 0; i < 12; i++) {
quicklistIndex(ql, i, &entry); quicklistIndex(ql, i, &entry);
if (entry.longval != nums[5 + i]) if (entry.longval != nums[5 + i])
...@@ -2781,7 +3043,7 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2781,7 +3043,7 @@ int quicklistTest(int argc, char *argv[], int accurate) {
if (node->encoding != QUICKLIST_NODE_ENCODING_RAW) { if (node->encoding != QUICKLIST_NODE_ENCODING_RAW) {
ERR("Incorrect compression: node %d is " ERR("Incorrect compression: node %d is "
"compressed at depth %d ((%u, %u); total " "compressed at depth %d ((%u, %u); total "
"nodes: %lu; size: %u)", "nodes: %lu; size: %zu)",
at, depth, low_raw, high_raw, ql->len, at, depth, low_raw, high_raw, ql->len,
node->sz); node->sz);
} }
...@@ -2789,7 +3051,7 @@ int quicklistTest(int argc, char *argv[], int accurate) { ...@@ -2789,7 +3051,7 @@ int quicklistTest(int argc, char *argv[], int accurate) {
if (node->encoding != QUICKLIST_NODE_ENCODING_LZF) { if (node->encoding != QUICKLIST_NODE_ENCODING_LZF) {
ERR("Incorrect non-compression: node %d is NOT " ERR("Incorrect non-compression: node %d is NOT "
"compressed at depth %d ((%u, %u); total " "compressed at depth %d ((%u, %u); total "
"nodes: %lu; size: %u; attempted: %d)", "nodes: %lu; size: %zu; attempted: %d)",
at, depth, low_raw, high_raw, ql->len, at, depth, low_raw, high_raw, ql->len,
node->sz, node->attempted_compress); node->sz, node->attempted_compress);
} }
......
...@@ -46,8 +46,8 @@ ...@@ -46,8 +46,8 @@
typedef struct quicklistNode { typedef struct quicklistNode {
struct quicklistNode *prev; struct quicklistNode *prev;
struct quicklistNode *next; struct quicklistNode *next;
unsigned char *zl; unsigned char *entry;
unsigned int sz; /* ziplist size in bytes */ size_t sz; /* entry size in bytes */
unsigned int count : 16; /* count of items in ziplist */ unsigned int count : 16; /* count of items in ziplist */
unsigned int encoding : 2; /* RAW==1 or LZF==2 */ unsigned int encoding : 2; /* RAW==1 or LZF==2 */
unsigned int container : 2; /* NONE==1 or ZIPLIST==2 */ unsigned int container : 2; /* NONE==1 or ZIPLIST==2 */
...@@ -56,13 +56,13 @@ typedef struct quicklistNode { ...@@ -56,13 +56,13 @@ typedef struct quicklistNode {
unsigned int extra : 10; /* more bits to steal for future usage */ unsigned int extra : 10; /* more bits to steal for future usage */
} quicklistNode; } quicklistNode;
/* quicklistLZF is a 4+N byte struct holding 'sz' followed by 'compressed'. /* quicklistLZF is a 8+N byte struct holding 'sz' followed by 'compressed'.
* 'sz' is byte length of 'compressed' field. * 'sz' is byte length of 'compressed' field.
* 'compressed' is LZF data with total (compressed) length 'sz' * 'compressed' is LZF data with total (compressed) length 'sz'
* NOTE: uncompressed length is stored in quicklistNode->sz. * NOTE: uncompressed length is stored in quicklistNode->sz.
* When quicklistNode->zl is compressed, node->zl points to a quicklistLZF */ * When quicklistNode->entry is compressed, node->entry points to a quicklistLZF */
typedef struct quicklistLZF { typedef struct quicklistLZF {
unsigned int sz; /* LZF size in bytes*/ size_t sz; /* LZF size in bytes*/
char compressed[]; char compressed[];
} quicklistLZF; } quicklistLZF;
...@@ -127,7 +127,7 @@ typedef struct quicklistEntry { ...@@ -127,7 +127,7 @@ typedef struct quicklistEntry {
unsigned char *zi; unsigned char *zi;
unsigned char *value; unsigned char *value;
long long longval; long long longval;
unsigned int sz; size_t sz;
int offset; int offset;
} quicklistEntry; } quicklistEntry;
...@@ -142,9 +142,11 @@ typedef struct quicklistEntry { ...@@ -142,9 +142,11 @@ typedef struct quicklistEntry {
#define QUICKLIST_NOCOMPRESS 0 #define QUICKLIST_NOCOMPRESS 0
/* quicklist container formats */ /* quicklist container formats */
#define QUICKLIST_NODE_CONTAINER_NONE 1 #define QUICKLIST_NODE_CONTAINER_PLAIN 1
#define QUICKLIST_NODE_CONTAINER_ZIPLIST 2 #define QUICKLIST_NODE_CONTAINER_ZIPLIST 2
#define QL_NODE_IS_PLAIN(node) ((node)->container == QUICKLIST_NODE_CONTAINER_PLAIN)
#define quicklistNodeIsCompressed(node) \ #define quicklistNodeIsCompressed(node) \
((node)->encoding == QUICKLIST_NODE_ENCODING_LZF) ((node)->encoding == QUICKLIST_NODE_ENCODING_LZF)
...@@ -160,6 +162,7 @@ int quicklistPushTail(quicklist *quicklist, void *value, const size_t sz); ...@@ -160,6 +162,7 @@ int quicklistPushTail(quicklist *quicklist, void *value, const size_t sz);
void quicklistPush(quicklist *quicklist, void *value, const size_t sz, void quicklistPush(quicklist *quicklist, void *value, const size_t sz,
int where); int where);
void quicklistAppendZiplist(quicklist *quicklist, unsigned char *zl); void quicklistAppendZiplist(quicklist *quicklist, unsigned char *zl);
void quicklistAppendPlainNode(quicklist *quicklist, unsigned char *data, size_t sz);
quicklist *quicklistAppendValuesFromZiplist(quicklist *quicklist, quicklist *quicklistAppendValuesFromZiplist(quicklist *quicklist,
unsigned char *zl); unsigned char *zl);
quicklist *quicklistCreateFromZiplist(int fill, int compress, quicklist *quicklistCreateFromZiplist(int fill, int compress,
...@@ -170,9 +173,9 @@ void quicklistInsertBefore(quicklist *quicklist, quicklistEntry *entry, ...@@ -170,9 +173,9 @@ void quicklistInsertBefore(quicklist *quicklist, quicklistEntry *entry,
void *value, const size_t sz); void *value, const size_t sz);
void quicklistDelEntry(quicklistIter *iter, quicklistEntry *entry); void quicklistDelEntry(quicklistIter *iter, quicklistEntry *entry);
void quicklistReplaceEntry(quicklist *quicklist, quicklistEntry *entry, void quicklistReplaceEntry(quicklist *quicklist, quicklistEntry *entry,
void *data, int sz); void *data, size_t sz);
int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data, int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data,
int sz); const size_t sz);
int quicklistDelRange(quicklist *quicklist, const long start, const long stop); int quicklistDelRange(quicklist *quicklist, const long start, const long stop);
quicklistIter *quicklistGetIterator(const quicklist *quicklist, int direction); quicklistIter *quicklistGetIterator(const quicklist *quicklist, int direction);
quicklistIter *quicklistGetIteratorAtIdx(const quicklist *quicklist, quicklistIter *quicklistGetIteratorAtIdx(const quicklist *quicklist,
...@@ -185,19 +188,21 @@ int quicklistIndex(const quicklist *quicklist, const long long index, ...@@ -185,19 +188,21 @@ int quicklistIndex(const quicklist *quicklist, const long long index,
quicklistEntry *entry); quicklistEntry *entry);
void quicklistRotate(quicklist *quicklist); void quicklistRotate(quicklist *quicklist);
int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data, int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
unsigned int *sz, long long *sval, size_t *sz, long long *sval,
void *(*saver)(unsigned char *data, unsigned int sz)); void *(*saver)(unsigned char *data, size_t sz));
int quicklistPop(quicklist *quicklist, int where, unsigned char **data, int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
unsigned int *sz, long long *slong); size_t *sz, long long *slong);
unsigned long quicklistCount(const quicklist *ql); unsigned long quicklistCount(const quicklist *ql);
int quicklistCompare(unsigned char *p1, unsigned char *p2, int p2_len); int quicklistCompare(quicklistEntry *entry, unsigned char *p2, const size_t p2_len);
size_t quicklistGetLzf(const quicklistNode *node, void **data); size_t quicklistGetLzf(const quicklistNode *node, void **data);
void quicklistRepr(unsigned char *ql, int full);
/* bookmarks */ /* bookmarks */
int quicklistBookmarkCreate(quicklist **ql_ref, const char *name, quicklistNode *node); int quicklistBookmarkCreate(quicklist **ql_ref, const char *name, quicklistNode *node);
int quicklistBookmarkDelete(quicklist *ql, const char *name); int quicklistBookmarkDelete(quicklist *ql, const char *name);
quicklistNode *quicklistBookmarkFind(quicklist *ql, const char *name); quicklistNode *quicklistBookmarkFind(quicklist *ql, const char *name);
void quicklistBookmarksClear(quicklist *ql); void quicklistBookmarksClear(quicklist *ql);
int quicklistisSetPackedThreshold(size_t sz);
#ifdef REDIS_TEST #ifdef REDIS_TEST
int quicklistTest(int argc, char *argv[], int accurate); int quicklistTest(int argc, char *argv[], int accurate);
......
...@@ -658,7 +658,7 @@ int rdbSaveObjectType(rio *rdb, robj *o) { ...@@ -658,7 +658,7 @@ int rdbSaveObjectType(rio *rdb, robj *o) {
return rdbSaveType(rdb,RDB_TYPE_STRING); return rdbSaveType(rdb,RDB_TYPE_STRING);
case OBJ_LIST: case OBJ_LIST:
if (o->encoding == OBJ_ENCODING_QUICKLIST) if (o->encoding == OBJ_ENCODING_QUICKLIST)
return rdbSaveType(rdb,RDB_TYPE_LIST_QUICKLIST); return rdbSaveType(rdb, RDB_TYPE_LIST_QUICKLIST_2);
else else
serverPanic("Unknown list encoding"); serverPanic("Unknown list encoding");
case OBJ_SET: case OBJ_SET:
...@@ -813,13 +813,16 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) { ...@@ -813,13 +813,16 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) {
nwritten += n; nwritten += n;
while(node) { while(node) {
if ((n = rdbSaveLen(rdb,node->container)) == -1) return -1;
nwritten += n;
if (quicklistNodeIsCompressed(node)) { if (quicklistNodeIsCompressed(node)) {
void *data; void *data;
size_t compress_len = quicklistGetLzf(node, &data); size_t compress_len = quicklistGetLzf(node, &data);
if ((n = rdbSaveLzfBlob(rdb,data,compress_len,node->sz)) == -1) return -1; if ((n = rdbSaveLzfBlob(rdb,data,compress_len,node->sz)) == -1) return -1;
nwritten += n; nwritten += n;
} else { } else {
if ((n = rdbSaveRawString(rdb,node->zl,node->sz)) == -1) return -1; if ((n = rdbSaveRawString(rdb,node->entry,node->sz)) == -1) return -1;
nwritten += n; nwritten += n;
} }
node = node->next; node = node->next;
...@@ -1934,36 +1937,58 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) { ...@@ -1934,36 +1937,58 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
/* All pairs should be read by now */ /* All pairs should be read by now */
serverAssert(len == 0); serverAssert(len == 0);
} else if (rdbtype == RDB_TYPE_LIST_QUICKLIST) { } else if (rdbtype == RDB_TYPE_LIST_QUICKLIST || rdbtype == RDB_TYPE_LIST_QUICKLIST_2) {
if ((len = rdbLoadLen(rdb,NULL)) == RDB_LENERR) return NULL; if ((len = rdbLoadLen(rdb,NULL)) == RDB_LENERR) return NULL;
if (len == 0) goto emptykey; if (len == 0) goto emptykey;
o = createQuicklistObject(); o = createQuicklistObject();
quicklistSetOptions(o->ptr, server.list_max_ziplist_size, quicklistSetOptions(o->ptr, server.list_max_ziplist_size,
server.list_compress_depth); server.list_compress_depth);
uint64_t container = QUICKLIST_NODE_CONTAINER_ZIPLIST;
while (len--) { while (len--) {
size_t encoded_len; size_t encoded_len;
unsigned char *zl =
if (rdbtype == RDB_TYPE_LIST_QUICKLIST_2) {
if ((container = rdbLoadLen(rdb,NULL)) == RDB_LENERR) {
decrRefCount(o);
return NULL;
}
if (container != QUICKLIST_NODE_CONTAINER_ZIPLIST && container != QUICKLIST_NODE_CONTAINER_PLAIN) {
rdbReportCorruptRDB("Quicklist integrity check failed.");
decrRefCount(o);
return NULL;
}
}
unsigned char *data =
rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,&encoded_len); rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,&encoded_len);
if (zl == NULL) { if (data == NULL || (encoded_len == 0)) {
zfree(data);
decrRefCount(o); decrRefCount(o);
return NULL; return NULL;
} }
if (container == QUICKLIST_NODE_CONTAINER_PLAIN) {
quicklistAppendPlainNode(o->ptr, data, encoded_len);
zfree(data);
continue;
}
if (deep_integrity_validation) server.stat_dump_payload_sanitizations++; if (deep_integrity_validation) server.stat_dump_payload_sanitizations++;
if (!ziplistValidateIntegrity(zl, encoded_len, deep_integrity_validation, NULL, NULL)) { if (!ziplistValidateIntegrity(data, encoded_len, deep_integrity_validation, NULL, NULL)) {
rdbReportCorruptRDB("Ziplist integrity check failed."); rdbReportCorruptRDB("Ziplist integrity check failed.");
decrRefCount(o); decrRefCount(o);
zfree(zl); zfree(data);
return NULL; return NULL;
} }
/* Silently skip empty ziplists, if we'll end up with empty quicklist we'll fail later. */ /* Silently skip empty ziplists, if we'll end up with empty quicklist we'll fail later. */
if (ziplistLen(zl) == 0) { if (ziplistLen(data) == 0) {
zfree(zl); zfree(data);
continue; continue;
} else { } else {
quicklistAppendZiplist(o->ptr, zl); quicklistAppendZiplist(o->ptr, data);
} }
} }
......
...@@ -93,10 +93,11 @@ ...@@ -93,10 +93,11 @@
#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 #define RDB_TYPE_ZSET_LISTPACK 17
#define RDB_TYPE_LIST_QUICKLIST_2 18
/* 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 <= 17)) #define rdbIsObjectType(t) ((t >= 0 && t <= 7) || (t >= 9 && t <= 18))
/* 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,7 +92,8 @@ char *rdb_type_string[] = { ...@@ -92,7 +92,8 @@ char *rdb_type_string[] = {
"quicklist", "quicklist",
"stream", "stream",
"hash-listpack", "hash-listpack",
"zset-listpack" "zset-listpack",
"quicklist-v2"
}; };
/* Show a few stats collected into 'rdbstate' */ /* Show a few stats collected into 'rdbstate' */
......
...@@ -29,8 +29,6 @@ ...@@ -29,8 +29,6 @@
#include "server.h" #include "server.h"
#define LIST_MAX_ITEM_SIZE ((1ull<<32)-1024)
/*----------------------------------------------------------------------------- /*-----------------------------------------------------------------------------
* List API * List API
*----------------------------------------------------------------------------*/ *----------------------------------------------------------------------------*/
...@@ -55,7 +53,7 @@ void listTypePush(robj *subject, robj *value, int where) { ...@@ -55,7 +53,7 @@ void listTypePush(robj *subject, robj *value, int where) {
} }
} }
void *listPopSaver(unsigned char *data, unsigned int sz) { void *listPopSaver(unsigned char *data, size_t sz) {
return createStringObject((char*)data,sz); return createStringObject((char*)data,sz);
} }
...@@ -186,7 +184,7 @@ void listTypeReplace(listTypeEntry *entry, robj *value) { ...@@ -186,7 +184,7 @@ void listTypeReplace(listTypeEntry *entry, robj *value) {
int listTypeEqual(listTypeEntry *entry, robj *o) { int listTypeEqual(listTypeEntry *entry, robj *o) {
if (entry->li->encoding == OBJ_ENCODING_QUICKLIST) { if (entry->li->encoding == OBJ_ENCODING_QUICKLIST) {
serverAssertWithInfo(NULL,o,sdsEncodedObject(o)); serverAssertWithInfo(NULL,o,sdsEncodedObject(o));
return quicklistCompare(entry->entry.zi,o->ptr,sdslen(o->ptr)); return quicklistCompare(&entry->entry,o->ptr,sdslen(o->ptr));
} else { } else {
serverPanic("Unknown list encoding"); serverPanic("Unknown list encoding");
} }
...@@ -210,7 +208,7 @@ void listTypeConvert(robj *subject, int enc) { ...@@ -210,7 +208,7 @@ void listTypeConvert(robj *subject, int enc) {
size_t zlen = server.list_max_ziplist_size; size_t zlen = server.list_max_ziplist_size;
int depth = server.list_compress_depth; int depth = server.list_compress_depth;
subject->ptr = quicklistCreateFromZiplist(zlen, depth, subject->ptr); subject->ptr = quicklistCreateFromZiplist(zlen, depth, subject->ptr);
subject->encoding = OBJ_ENCODING_QUICKLIST; subject->encoding = enc;
} else { } else {
serverPanic("Unsupported list conversion"); serverPanic("Unsupported list conversion");
} }
...@@ -229,7 +227,7 @@ robj *listTypeDup(robj *o) { ...@@ -229,7 +227,7 @@ robj *listTypeDup(robj *o) {
switch (o->encoding) { switch (o->encoding) {
case OBJ_ENCODING_QUICKLIST: case OBJ_ENCODING_QUICKLIST:
lobj = createObject(OBJ_LIST, quicklistDup(o->ptr)); lobj = createObject(OBJ_LIST, quicklistDup(o->ptr));
lobj->encoding = OBJ_ENCODING_QUICKLIST; lobj->encoding = o->encoding;
break; break;
default: default:
serverPanic("Unknown list encoding"); serverPanic("Unknown list encoding");
...@@ -256,13 +254,6 @@ int listTypeDelRange(robj *subject, long start, long count) { ...@@ -256,13 +254,6 @@ int listTypeDelRange(robj *subject, long start, long count) {
void pushGenericCommand(client *c, int where, int xx) { void pushGenericCommand(client *c, int where, int xx) {
int j; int j;
for (j = 2; j < c->argc; j++) {
if (sdslen(c->argv[j]->ptr) > LIST_MAX_ITEM_SIZE) {
addReplyError(c, "Element too large");
return;
}
}
robj *lobj = lookupKeyWrite(c->db, c->argv[1]); robj *lobj = lookupKeyWrite(c->db, c->argv[1]);
if (checkType(c,lobj,OBJ_LIST)) return; if (checkType(c,lobj,OBJ_LIST)) return;
if (!lobj) { if (!lobj) {
...@@ -326,11 +317,6 @@ void linsertCommand(client *c) { ...@@ -326,11 +317,6 @@ void linsertCommand(client *c) {
return; return;
} }
if (sdslen(c->argv[4]->ptr) > LIST_MAX_ITEM_SIZE) {
addReplyError(c, "Element too large");
return;
}
if ((subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL || if ((subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,subject,OBJ_LIST)) return; checkType(c,subject,OBJ_LIST)) return;
...@@ -398,11 +384,6 @@ void lsetCommand(client *c) { ...@@ -398,11 +384,6 @@ void lsetCommand(client *c) {
long index; long index;
robj *value = c->argv[3]; robj *value = c->argv[3];
if (sdslen(value->ptr) > LIST_MAX_ITEM_SIZE) {
addReplyError(c, "Element too large");
return;
}
if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != C_OK)) if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != C_OK))
return; return;
...@@ -702,11 +683,6 @@ void lposCommand(client *c) { ...@@ -702,11 +683,6 @@ void lposCommand(client *c) {
int direction = LIST_TAIL; int direction = LIST_TAIL;
long rank = 1, count = -1, maxlen = 0; /* Count -1: option not given. */ long rank = 1, count = -1, maxlen = 0; /* Count -1: option not given. */
if (sdslen(ele->ptr) > LIST_MAX_ITEM_SIZE) {
addReplyError(c, "Element too large");
return;
}
/* Parse the optional arguments. */ /* Parse the optional arguments. */
for (int j = 3; j < c->argc; j++) { for (int j = 3; j < c->argc; j++) {
char *opt = c->argv[j]->ptr; char *opt = c->argv[j]->ptr;
...@@ -802,11 +778,6 @@ void lremCommand(client *c) { ...@@ -802,11 +778,6 @@ void lremCommand(client *c) {
long toremove; long toremove;
long removed = 0; long removed = 0;
if (sdslen(obj->ptr) > LIST_MAX_ITEM_SIZE) {
addReplyError(c, "Element too large");
return;
}
if ((getLongFromObjectOrReply(c, c->argv[2], &toremove, NULL) != C_OK)) if ((getLongFromObjectOrReply(c, c->argv[2], &toremove, NULL) != C_OK))
return; return;
......
...@@ -17,6 +17,7 @@ match different external server configurations: ...@@ -17,6 +17,7 @@ match different external server configurations:
| `--ignore-encoding` | Skip all checks for specific encoding. | | `--ignore-encoding` | Skip all checks for specific encoding. |
| `--ignore-digest` | Skip key value digest validations. | | `--ignore-digest` | Skip key value digest validations. |
| `--cluster-mode` | Run in strict Redis Cluster compatibility mode. | | `--cluster-mode` | Run in strict Redis Cluster compatibility mode. |
| `--large-memory` | Enables tests that consume more than 100mb |
Tags Tags
---- ----
...@@ -36,6 +37,7 @@ The following compatibility and capability tags are currently used: ...@@ -36,6 +37,7 @@ The following compatibility and capability tags are currently used:
| --------------------- | --------- | | --------------------- | --------- |
| `external:skip` | Not compatible with external servers. | | `external:skip` | Not compatible with external servers. |
| `cluster:skip` | Not compatible with `--cluster-mode`. | | `cluster:skip` | Not compatible with `--cluster-mode`. |
| `large-memory` | Test that requires more than 100mb |
| `tls:skip` | Not campatible with `--tls`. | | `tls:skip` | Not campatible with `--tls`. |
| `needs:repl` | Uses replication and needs to be able to `SYNC` from server. | | `needs:repl` | Uses replication and needs to be able to `SYNC` from server. |
| `needs:debug` | Uses the `DEBUG` command or other debugging focused commands (like `OBJECT`). | | `needs:debug` | Uses the `DEBUG` command or other debugging focused commands (like `OBJECT`). |
...@@ -51,6 +53,9 @@ When using an external server (`--host` and `--port`), filtering using the ...@@ -51,6 +53,9 @@ When using an external server (`--host` and `--port`), filtering using the
When using `--cluster-mode`, filtering using the `cluster:skip` tag is done When using `--cluster-mode`, filtering using the `cluster:skip` tag is done
automatically. automatically.
When not using `--large-memory`, filtering using the `largemem:skip` tag is done
automatically.
In addition, it is possible to specify additional configuration. For example, to In addition, it is possible to specify additional configuration. For example, to
run tests on a server that does not permit `SYNC` use: run tests on a server that does not permit `SYNC` use:
......
...@@ -135,6 +135,14 @@ test {corrupt payload: quicklist with empty ziplist} { ...@@ -135,6 +135,14 @@ test {corrupt payload: quicklist with empty ziplist} {
} }
} }
test {corrupt payload: quicklist encoded_len is 0} {
start_server [list overrides [list loglevel verbose use-exit-on-panic yes crash-memcheck-enabled no] ] {
catch { r restore _list 0 "\x12\x01\x01\x00\n\x00\x8f\xc6\xc0W\x1c\n\xb3<" replace } err
assert_match "*Bad data format*" $err
r ping
}
}
test {corrupt payload: #3080 - ziplist} { test {corrupt payload: #3080 - ziplist} {
start_server [list overrides [list loglevel verbose use-exit-on-panic yes crash-memcheck-enabled no] ] { start_server [list overrides [list loglevel verbose use-exit-on-panic yes crash-memcheck-enabled no] ] {
# shallow sanitization is enough for restore to safely reject the payload with wrong size # shallow sanitization is enough for restore to safely reject the payload with wrong size
......
...@@ -4,6 +4,15 @@ set server_path [tmpdir "server.rdb-encoding-test"] ...@@ -4,6 +4,15 @@ set server_path [tmpdir "server.rdb-encoding-test"]
# Copy RDB with different encodings in server path # Copy RDB with different encodings in server path
exec cp tests/assets/encodings.rdb $server_path exec cp tests/assets/encodings.rdb $server_path
exec cp tests/assets/list-quicklist.rdb $server_path
start_server [list overrides [list "dir" $server_path "dbfilename" "list-quicklist.rdb"]] {
test "test old version rdb file" {
r select 0
assert_equal [r get x] 7
r lpop list
} {7}
}
start_server [list overrides [list "dir" $server_path "dbfilename" "encodings.rdb"]] { start_server [list overrides [list "dir" $server_path "dbfilename" "encodings.rdb"]] {
test "RDB encoding loading test" { test "RDB encoding loading test" {
......
...@@ -146,6 +146,10 @@ proc ::redis::__method__read {id fd} { ...@@ -146,6 +146,10 @@ proc ::redis::__method__read {id fd} {
::redis::redis_read_reply $id $fd ::redis::redis_read_reply $id $fd
} }
proc ::redis::__method__rawread {id fd len} {
return [read $fd $len]
}
proc ::redis::__method__write {id fd buf} { proc ::redis::__method__write {id fd buf} {
::redis::redis_write $fd $buf ::redis::redis_write $fd $buf
} }
......
...@@ -210,6 +210,11 @@ proc tags_acceptable {tags err_return} { ...@@ -210,6 +210,11 @@ proc tags_acceptable {tags err_return} {
return 0 return 0
} }
if {!$::large_memory && [lsearch $tags "large-memory"] >= 0} {
set err "large memory flag not provided"
return 0
}
return 1 return 1
} }
......
...@@ -127,6 +127,7 @@ set ::singledb 0 ...@@ -127,6 +127,7 @@ set ::singledb 0
set ::cluster_mode 0 set ::cluster_mode 0
set ::ignoreencoding 0 set ::ignoreencoding 0
set ::ignoredigest 0 set ::ignoredigest 0
set ::large_memory 0
# Set to 1 when we are running in client mode. The Redis test uses a # Set to 1 when we are running in client mode. The Redis test uses a
# server-client model to run tests simultaneously. The server instance # server-client model to run tests simultaneously. The server instance
...@@ -592,6 +593,7 @@ proc print_help_screen {} { ...@@ -592,6 +593,7 @@ proc print_help_screen {} {
"--cluster-mode Run tests in cluster protocol compatible mode." "--cluster-mode Run tests in cluster protocol compatible mode."
"--ignore-encoding Don't validate object encoding." "--ignore-encoding Don't validate object encoding."
"--ignore-digest Don't use debug digest validations." "--ignore-digest Don't use debug digest validations."
"--large-memory Run tests using over 100mb."
"--help Print this help screen." "--help Print this help screen."
} "\n"] } "\n"]
} }
...@@ -703,6 +705,8 @@ for {set j 0} {$j < [llength $argv]} {incr j} { ...@@ -703,6 +705,8 @@ for {set j 0} {$j < [llength $argv]} {incr j} {
} elseif {$opt eq {--cluster-mode}} { } elseif {$opt eq {--cluster-mode}} {
set ::cluster_mode 1 set ::cluster_mode 1
set ::singledb 1 set ::singledb 1
} elseif {$opt eq {--large-memory}} {
set ::large_memory 1
} elseif {$opt eq {--ignore-encoding}} { } elseif {$opt eq {--ignore-encoding}} {
set ::ignoreencoding 1 set ::ignoreencoding 1
} elseif {$opt eq {--ignore-digest}} { } elseif {$opt eq {--ignore-digest}} {
......
set ::str500 [string repeat x 500000000] ;# 500mb
# Utility function to write big argument into redis client connection
proc write_big_bulk {size} {
r write "\$$size\r\n"
while {$size >= 500000000} {
r write $::str500
incr size -500000000
}
if {$size > 0} {
r write [string repeat x $size]
}
r write "\r\n"
r flush
r read
}
# Utility to read big bulk response (work around Tcl limitations)
proc read_big_bulk {code} {
r readraw 1
set resp_len [uplevel 1 $code] ;# get the first line of the RESP response
assert_equal [string range $resp_len 0 0] "$"
set resp_len [string range $resp_len 1 end]
set remaining $resp_len
while {$remaining > 0} {
set l $remaining
if {$l > 2147483647} {set l 2147483647}
set nbytes [string length [r rawread $l]]
incr remaining [expr {- $nbytes}]
}
assert_equal [r rawread 2] "\r\n"
r readraw 0
return $resp_len
}
# check functionality compression of plain and zipped nodes
start_server [list overrides [list save ""] ] {
r config set list-compress-depth 2
r config set list-max-ziplist-size 1
# 3 test to check compression with regular ziplist nodes
# 1. using push + insert
# 2. using push + insert + trim
# 3. using push + insert + set
test {reg node check compression with insert and pop} {
r lpush list1 [string repeat a 500]
r lpush list1 [string repeat b 500]
r lpush list1 [string repeat c 500]
r lpush list1 [string repeat d 500]
r linsert list1 after [string repeat d 500] [string repeat e 500]
r linsert list1 after [string repeat d 500] [string repeat f 500]
r linsert list1 after [string repeat d 500] [string repeat g 500]
r linsert list1 after [string repeat d 500] [string repeat j 500]
assert_equal [r lpop list1] [string repeat d 500]
assert_equal [r lpop list1] [string repeat j 500]
assert_equal [r lpop list1] [string repeat g 500]
assert_equal [r lpop list1] [string repeat f 500]
assert_equal [r lpop list1] [string repeat e 500]
assert_equal [r lpop list1] [string repeat c 500]
assert_equal [r lpop list1] [string repeat b 500]
assert_equal [r lpop list1] [string repeat a 500]
};
test {reg node check compression combined with trim} {
r lpush list2 [string repeat a 500]
r linsert list2 after [string repeat a 500] [string repeat b 500]
r rpush list2 [string repeat c 500]
assert_equal [string repeat b 500] [r lindex list2 1]
r LTRIM list2 1 -1
r llen list2
} {2}
test {reg node check compression with lset} {
r lpush list3 [string repeat a 500]
r LSET list3 0 [string repeat b 500]
assert_equal [string repeat b 500] [r lindex list3 0]
r lpush list3 [string repeat c 500]
r LSET list3 0 [string repeat d 500]
assert_equal [string repeat d 500] [r lindex list3 0]
}
# repeating the 3 tests with plain nodes
# (by adjusting quicklist-packed-threshold)
test {plain node check compression} {
r debug quicklist-packed-threshold 1b
r lpush list4 [string repeat a 500]
r lpush list4 [string repeat b 500]
r lpush list4 [string repeat c 500]
r lpush list4 [string repeat d 500]
r linsert list4 after [string repeat d 500] [string repeat e 500]
r linsert list4 after [string repeat d 500] [string repeat f 500]
r linsert list4 after [string repeat d 500] [string repeat g 500]
r linsert list4 after [string repeat d 500] [string repeat j 500]
assert_equal [r lpop list4] [string repeat d 500]
assert_equal [r lpop list4] [string repeat j 500]
assert_equal [r lpop list4] [string repeat g 500]
assert_equal [r lpop list4] [string repeat f 500]
assert_equal [r lpop list4] [string repeat e 500]
assert_equal [r lpop list4] [string repeat c 500]
assert_equal [r lpop list4] [string repeat b 500]
assert_equal [r lpop list4] [string repeat a 500]
} {} {needs:debug}
test {plain node check compression with ltrim} {
r debug quicklist-packed-threshold 1b
r lpush list5 [string repeat a 500]
r linsert list5 after [string repeat a 500] [string repeat b 500]
r rpush list5 [string repeat c 500]
assert_equal [string repeat b 500] [r lindex list5 1]
r LTRIM list5 1 -1
r llen list5
} {2} {needs:debug}
test {plain node check compression using lset} {
r debug quicklist-packed-threshold 1b
r lpush list6 [string repeat a 500]
r LSET list6 0 [string repeat b 500]
assert_equal [string repeat b 500] [r lindex list6 0]
r lpush list6 [string repeat c 500]
r LSET list6 0 [string repeat d 500]
assert_equal [string repeat d 500] [r lindex list6 0]
} {} {needs:debug}
}
# check functionality of plain nodes using low packed-threshold
start_server [list overrides [list save ""] ] {
# basic command check for plain nodes - "LPUSH & LPOP"
test {Test LPUSH and LPOP on plain nodes} {
r flushdb
r debug quicklist-packed-threshold 1b
r lpush lst 9
r lpush lst xxxxxxxxxx
r lpush lst xxxxxxxxxx
set s0 [s used_memory]
assert {$s0 > 10}
assert {[r llen lst] == 3}
set s0 [r rpop lst]
set s1 [r rpop lst]
assert {$s0 eq "9"}
assert {[r llen lst] == 1}
r lpop lst
assert {[string length $s1] == 10}
# check rdb
r lpush lst xxxxxxxxxx
r lpush lst bb
r debug reload
assert_equal [r rpop lst] "xxxxxxxxxx"
} {} {needs:debug}
# basic command check for plain nodes - "LINDEX & LINSERT"
test {Test LINDEX and LINSERT on plain nodes} {
r flushdb
r debug quicklist-packed-threshold 1b
r lpush lst xxxxxxxxxxx
r lpush lst 9
r lpush lst xxxxxxxxxxx
r linsert lst before "9" "8"
assert {[r lindex lst 1] eq "8"}
r linsert lst BEFORE "9" "7"
r linsert lst BEFORE "9" "xxxxxxxxxxx"
assert {[r lindex lst 3] eq "xxxxxxxxxxx"}
} {} {needs:debug}
# basic command check for plain nodes - "LTRIM"
test {Test LTRIM on plain nodes} {
r flushdb
r debug quicklist-packed-threshold 1b
r lpush lst1 9
r lpush lst1 xxxxxxxxxxx
r lpush lst1 9
r LTRIM lst1 1 -1
assert_equal [r llen lst1] 2
} {} {needs:debug}
# basic command check for plain nodes - "LREM"
test {Test LREM on plain nodes} {
r flushdb
r debug quicklist-packed-threshold 1b
r lpush lst one
r lpush lst xxxxxxxxxxx
set s0 [s used_memory]
assert {$s0 > 10}
r lpush lst 9
r LREM lst -2 "one"
assert_equal [r llen lst] 2
} {} {needs:debug}
# basic command check for plain nodes - "LPOS"
test {Test LPOS on plain nodes} {
r flushdb
r debug quicklist-packed-threshold 1b
r RPUSH lst "aa"
r RPUSH lst "bb"
r RPUSH lst "cc"
r LSET lst 0 "xxxxxxxxxxx"
assert_equal [r LPOS lst "xxxxxxxxxxx"] 0
} {} {needs:debug}
# basic command check for plain nodes - "LMOVE"
test {Test LMOVE on plain nodes} {
r flushdb
r debug quicklist-packed-threshold 1b
r RPUSH lst2{t} "aa"
r RPUSH lst2{t} "bb"
r LSET lst2{t} 0 xxxxxxxxxxx
r RPUSH lst2{t} "cc"
r RPUSH lst2{t} "dd"
r LMOVE lst2{t} lst{t} RIGHT LEFT
r LMOVE lst2{t} lst{t} LEFT RIGHT
assert_equal [r llen lst{t}] 2
assert_equal [r llen lst2{t}] 2
assert_equal [r lpop lst2{t}] "bb"
assert_equal [r lpop lst2{t}] "cc"
assert_equal [r lpop lst{t}] "dd"
assert_equal [r lpop lst{t}] "xxxxxxxxxxx"
} {} {needs:debug}
# testing LSET with combinations of node types
# plain->packed , packed->plain, plain->plain, packed->packed
test {Test LSET with packed / plain combinations} {
r debug quicklist-packed-threshold 5b
r RPUSH lst "aa"
r RPUSH lst "bb"
r lset lst 0 [string repeat d 50001]
set s1 [r lpop lst]
assert_equal $s1 [string repeat d 50001]
r RPUSH lst [string repeat f 50001]
r lset lst 0 [string repeat e 50001]
set s1 [r lpop lst]
assert_equal $s1 [string repeat e 50001]
r RPUSH lst [string repeat m 50001]
r lset lst 0 "bb"
set s1 [r lpop lst]
assert_equal $s1 "bb"
r RPUSH lst "bb"
r lset lst 0 "cc"
set s1 [r lpop lst]
assert_equal $s1 "cc"
} {} {needs:debug}
# checking LSET in case ziplist needs to be split
test {Test LSET with packed is split in the middle} {
r flushdb
r debug quicklist-packed-threshold 5b
r RPUSH lst "aa"
r RPUSH lst "bb"
r RPUSH lst "cc"
r RPUSH lst "dd"
r RPUSH lst "ee"
r lset lst 2 [string repeat e 10]
assert_equal [r lpop lst] "aa"
assert_equal [r lpop lst] "bb"
assert_equal [r lpop lst] [string repeat e 10]
assert_equal [r lpop lst] "dd"
assert_equal [r lpop lst] "ee"
} {} {needs:debug}
# repeating "plain check LSET with combinations"
# but now with single item in each ziplist
test {Test LSET with packed consist only one item} {
r flushdb
r config set list-max-ziplist-size 1
r debug quicklist-packed-threshold 1b
r RPUSH lst "aa"
r RPUSH lst "bb"
r lset lst 0 [string repeat d 50001]
set s1 [r lpop lst]
assert_equal $s1 [string repeat d 50001]
r RPUSH lst [string repeat f 50001]
r lset lst 0 [string repeat e 50001]
set s1 [r lpop lst]
assert_equal $s1 [string repeat e 50001]
r RPUSH lst [string repeat m 50001]
r lset lst 0 "bb"
set s1 [r lpop lst]
assert_equal $s1 "bb"
r RPUSH lst "bb"
r lset lst 0 "cc"
set s1 [r lpop lst]
assert_equal $s1 "cc"
} {} {needs:debug}
}
start_server [list overrides [list save ""] ] {
# test if the server supports such large configs (avoid 32 bit builds)
catch {
r config set proto-max-bulk-len 10000000000 ;#10gb
r config set client-query-buffer-limit 10000000000 ;#10gb
}
if {[lindex [r config get proto-max-bulk-len] 1] == 10000000000} {
set str_length 5000000000
# repeating all the plain nodes basic checks with 5gb values
test {Test LPUSH and LPOP on plain nodes over 4GB} {
r flushdb
r lpush lst 9
r write "*3\r\n\$5\r\nLPUSH\r\n\$3\r\nlst\r\n"
write_big_bulk $str_length;
r write "*3\r\n\$5\r\nLPUSH\r\n\$3\r\nlst\r\n"
write_big_bulk $str_length;
set s0 [s used_memory]
assert {$s0 > $str_length}
assert {[r llen lst] == 3}
assert_equal [r rpop lst] "9"
assert_equal [read_big_bulk {r rpop lst}] $str_length
assert {[r llen lst] == 1}
assert_equal [read_big_bulk {r rpop lst}] $str_length
} {} {large-memory}
test {Test LINDEX and LINSERT on plain nodes over 4GB} {
r flushdb
r write "*3\r\n\$5\r\nLPUSH\r\n\$3\r\nlst\r\n"
write_big_bulk $str_length;
r lpush lst 9
r write "*3\r\n\$5\r\nLPUSH\r\n\$3\r\nlst\r\n"
write_big_bulk $str_length;
r linsert lst before "9" "8"
assert_equal [r lindex lst 1] "8"
r LINSERT lst BEFORE "9" "7"
r write "*5\r\n\$7\r\nLINSERT\r\n\$3\r\nlst\r\n\$6\r\nBEFORE\r\n\$3\r\n\"9\"\r\n"
write_big_bulk 10;
assert_equal [read_big_bulk {r rpop lst}] $str_length
} {} {large-memory}
test {Test LTRIM on plain nodes over 4GB} {
r flushdb
r lpush lst 9
r write "*3\r\n\$5\r\nLPUSH\r\n\$3\r\nlst\r\n"
write_big_bulk $str_length;
r lpush lst 9
r LTRIM lst 1 -1
assert_equal [r llen lst] 2
assert_equal [r rpop lst] 9
assert_equal [read_big_bulk {r rpop lst}] $str_length
} {} {large-memory}
test {Test LREM on plain nodes over 4GB} {
r flushdb
r lpush lst one
r write "*3\r\n\$5\r\nLPUSH\r\n\$3\r\nlst\r\n"
write_big_bulk $str_length;
r lpush lst 9
r LREM lst -2 "one"
assert_equal [read_big_bulk {r rpop lst}] $str_length
r llen lst
} {1} {large-memory}
test {Test LSET on plain nodes over 4GB} {
r flushdb
r RPUSH lst "aa"
r RPUSH lst "bb"
r RPUSH lst "cc"
r write "*4\r\n\$4\r\nLSET\r\n\$3\r\nlst\r\n\$1\r\n0\r\n"
write_big_bulk $str_length;
assert_equal [r rpop lst] "cc"
assert_equal [r rpop lst] "bb"
assert_equal [read_big_bulk {r rpop lst}] $str_length
} {} {large-memory}
test {Test LMOVE on plain nodes over 4GB} {
r flushdb
r RPUSH lst2{t} "aa"
r RPUSH lst2{t} "bb"
r write "*4\r\n\$4\r\nLSET\r\n\$7\r\nlst2{t}\r\n\$1\r\n0\r\n"
write_big_bulk $str_length;
r RPUSH lst2{t} "cc"
r RPUSH lst2{t} "dd"
r LMOVE lst2{t} lst{t} RIGHT LEFT
assert_equal [read_big_bulk {r LMOVE lst2{t} lst{t} LEFT RIGHT}] $str_length
assert_equal [r llen lst{t}] 2
assert_equal [r llen lst2{t}] 2
assert_equal [r lpop lst2{t}] "bb"
assert_equal [r lpop lst2{t}] "cc"
assert_equal [r lpop lst{t}] "dd"
assert_equal [read_big_bulk {r rpop lst{t}}] $str_length
} {} {large-memory}
} ;# skip 32bit builds
}
start_server { start_server {
tags {"list"} tags {"list"}
overrides { overrides {
...@@ -1405,5 +1789,4 @@ foreach {pop} {BLPOP BLMPOP_RIGHT} { ...@@ -1405,5 +1789,4 @@ foreach {pop} {BLPOP BLMPOP_RIGHT} {
assert_equal [lpop k] [string repeat x 31] assert_equal [lpop k] [string repeat x 31]
set _ $k set _ $k
} {12 0 9223372036854775808 2147483647 32767 127} } {12 0 9223372036854775808 2147483647 32767 127}
} }
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