Unverified Commit 494ee2f1 authored by sundb's avatar sundb Committed by GitHub
Browse files

Fix abnormal compression due to out-of-control recompress (#9849)

This pr is following #9779 .

## Describe of feature
Now when we turn on the `list-compress-depth` configuration, the list will compress
the ziplist between `[list-compress-depth, -list-compress-depth]`.
When we need to use the compressed data, we will first decompress it, then use it,
and finally compress it again.
It's controlled by `quicklistNode->recompress`, which is designed to avoid the need to
re-traverse the entire quicklist for compression after each decompression, we only need
to recompress the quicklsitNode being used.
In order to ensure the correctness of recompressing, we should normally let
quicklistDecompressNodeForUse and quicklistCompress appear in pairs, otherwise,
it may lead to the head and tail being compressed or the middle ziplist not being
compressed correctly, which is exactly the problem this pr needs to solve.

## Solution
1. Reset `quicklistIter` after insert and replace.
    The quicklist node ...
parent 8759c1e1
This diff is collapsed.
...@@ -114,7 +114,7 @@ typedef struct quicklist { ...@@ -114,7 +114,7 @@ typedef struct quicklist {
} quicklist; } quicklist;
typedef struct quicklistIter { typedef struct quicklistIter {
const quicklist *quicklist; quicklist *quicklist;
quicklistNode *current; quicklistNode *current;
unsigned char *zi; unsigned char *zi;
long offset; /* offset in current listpack */ long offset; /* offset in current listpack */
...@@ -163,25 +163,25 @@ void quicklistPush(quicklist *quicklist, void *value, const size_t sz, ...@@ -163,25 +163,25 @@ void quicklistPush(quicklist *quicklist, void *value, const size_t sz,
int where); int where);
void quicklistAppendListpack(quicklist *quicklist, unsigned char *zl); void quicklistAppendListpack(quicklist *quicklist, unsigned char *zl);
void quicklistAppendPlainNode(quicklist *quicklist, unsigned char *data, size_t sz); void quicklistAppendPlainNode(quicklist *quicklist, unsigned char *data, size_t sz);
void quicklistInsertAfter(quicklist *quicklist, quicklistEntry *entry, void quicklistInsertAfter(quicklistIter *iter, quicklistEntry *entry,
void *value, const size_t sz); void *value, const size_t sz);
void quicklistInsertBefore(quicklist *quicklist, quicklistEntry *entry, void quicklistInsertBefore(quicklistIter *iter, 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(quicklistIter *iter, quicklistEntry *entry,
void *data, size_t sz); void *data, size_t sz);
int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data, int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data,
const size_t 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(quicklist *quicklist, int direction);
quicklistIter *quicklistGetIteratorAtIdx(const quicklist *quicklist, quicklistIter *quicklistGetIteratorAtIdx(quicklist *quicklist,
int direction, const long long idx); int direction, const long long idx);
quicklistIter *quicklistGetIteratorEntryAtIdx(quicklist *quicklist, const long long index,
quicklistEntry *entry);
int quicklistNext(quicklistIter *iter, quicklistEntry *entry); int quicklistNext(quicklistIter *iter, quicklistEntry *entry);
void quicklistSetDirection(quicklistIter *iter, int direction); void quicklistSetDirection(quicklistIter *iter, int direction);
void quicklistReleaseIterator(quicklistIter *iter); void quicklistReleaseIterator(quicklistIter *iter);
quicklist *quicklistDup(quicklist *orig); quicklist *quicklistDup(quicklist *orig);
int quicklistIndex(const quicklist *quicklist, const long long index,
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,
size_t *sz, long long *sval, size_t *sz, long long *sval,
......
...@@ -112,7 +112,7 @@ void listTypeSetIteratorDirection(listTypeIterator *li, unsigned char direction) ...@@ -112,7 +112,7 @@ void listTypeSetIteratorDirection(listTypeIterator *li, unsigned char direction)
/* Clean up the iterator. */ /* Clean up the iterator. */
void listTypeReleaseIterator(listTypeIterator *li) { void listTypeReleaseIterator(listTypeIterator *li) {
zfree(li->iter); quicklistReleaseIterator(li->iter);
zfree(li); zfree(li);
} }
...@@ -154,11 +154,9 @@ void listTypeInsert(listTypeEntry *entry, robj *value, int where) { ...@@ -154,11 +154,9 @@ void listTypeInsert(listTypeEntry *entry, robj *value, int where) {
sds str = value->ptr; sds str = value->ptr;
size_t len = sdslen(str); size_t len = sdslen(str);
if (where == LIST_TAIL) { if (where == LIST_TAIL) {
quicklistInsertAfter((quicklist *)entry->entry.quicklist, quicklistInsertAfter(entry->li->iter, &entry->entry, str, len);
&entry->entry, str, len);
} else if (where == LIST_HEAD) { } else if (where == LIST_HEAD) {
quicklistInsertBefore((quicklist *)entry->entry.quicklist, quicklistInsertBefore(entry->li->iter, &entry->entry, str, len);
&entry->entry, str, len);
} }
decrRefCount(value); decrRefCount(value);
} else { } else {
...@@ -172,8 +170,7 @@ void listTypeReplace(listTypeEntry *entry, robj *value) { ...@@ -172,8 +170,7 @@ void listTypeReplace(listTypeEntry *entry, robj *value) {
value = getDecodedObject(value); value = getDecodedObject(value);
sds str = value->ptr; sds str = value->ptr;
size_t len = sdslen(str); size_t len = sdslen(str);
quicklistReplaceEntry((quicklist *)entry->entry.quicklist, quicklistReplaceEntry(entry->li->iter, &entry->entry, str, len);
&entry->entry, str, len);
decrRefCount(value); decrRefCount(value);
} else { } else {
serverPanic("Unknown list encoding"); serverPanic("Unknown list encoding");
...@@ -348,7 +345,8 @@ void lindexCommand(client *c) { ...@@ -348,7 +345,8 @@ void lindexCommand(client *c) {
if (o->encoding == OBJ_ENCODING_QUICKLIST) { if (o->encoding == OBJ_ENCODING_QUICKLIST) {
quicklistEntry entry; quicklistEntry entry;
if (quicklistIndex(o->ptr, index, &entry)) { quicklistIter *iter = quicklistGetIteratorEntryAtIdx(o->ptr, index, &entry);
if (iter) {
if (entry.value) { if (entry.value) {
addReplyBulkCBuffer(c, entry.value, entry.sz); addReplyBulkCBuffer(c, entry.value, entry.sz);
} else { } else {
...@@ -357,6 +355,7 @@ void lindexCommand(client *c) { ...@@ -357,6 +355,7 @@ void lindexCommand(client *c) {
} else { } else {
addReplyNull(c); addReplyNull(c);
} }
quicklistReleaseIterator(iter);
} else { } else {
serverPanic("Unknown list encoding"); serverPanic("Unknown list encoding");
} }
......
proc generate_cmd_on_list_key {key} {
set op [randomInt 7]
set small_signed_count [expr 5-[randomInt 10]]
if {[randomInt 2] == 0} {
set ele [randomInt 1000]
} else {
set ele [string repeat x [randomInt 10000]][randomInt 1000]
}
switch $op {
0 {return "lpush $key $ele"}
1 {return "rpush $key $ele"}
2 {return "lpop $key"}
3 {return "rpop $key"}
4 {
return "lset $key $small_signed_count $ele"
}
5 {
set otherele [randomInt 1000]
if {[randomInt 2] == 0} {
set where before
} else {
set where after
}
return "linsert $key $where $otherele $ele"
}
6 {
set otherele ""
catch {
set index [randomInt [r llen $key]]
set otherele [r lindex $key $index]
}
return "lrem $key 1 $otherele"
}
}
}
start_server { start_server {
tags {"list ziplist"} tags {"list ziplist"}
overrides { overrides {
...@@ -38,7 +74,24 @@ start_server { ...@@ -38,7 +74,24 @@ start_server {
r rpush key [string repeat e 5000] r rpush key [string repeat e 5000]
r linsert key before f 1 r linsert key before f 1
r rpush key g r rpush key g
} r ping
}
test {Crash due to wrongly recompress after lrem} {
r del key
config_set list-compress-depth 2
r lpush key a
r lpush key [string repeat a 5000]
r lpush key [string repeat b 5000]
r lpush key [string repeat c 5000]
r rpush key [string repeat x 10000]"969"
r rpush key b
r lrem key 1 a
r rpop key
r lrem key 1 [string repeat x 10000]"969"
r rpush key crash
r ping
}
foreach comp {2 1 0} { foreach comp {2 1 0} {
set cycles 1000 set cycles 1000
...@@ -47,38 +100,44 @@ foreach comp {2 1 0} { ...@@ -47,38 +100,44 @@ foreach comp {2 1 0} {
test "Stress tester for #3343-alike bugs comp: $comp" { test "Stress tester for #3343-alike bugs comp: $comp" {
r del key r del key
set sent {}
for {set j 0} {$j < $cycles} {incr j} { for {set j 0} {$j < $cycles} {incr j} {
set op [randomInt 7] catch {
set small_signed_count [expr 5-[randomInt 10]] set cmd [generate_cmd_on_list_key key]
if {[randomInt 2] == 0} { lappend sent $cmd
set ele [randomInt 1000]
} else { # execute the command, we expect commands to fail on syntax errors
set ele [string repeat x [randomInt 10000]][randomInt 1000] r {*}$cmd
} }
switch $op { }
0 {r lpush key $ele}
1 {r rpush key $ele} set print_commands false
2 {r lpop key} set crash false
3 {r rpop key} if {[catch {r ping}]} {
4 { puts "Server crashed"
catch {r lset key $small_signed_count $ele} set print_commands true
} set crash true
5 { }
set otherele [randomInt 1000]
if {[randomInt 2] == 0} { if {!$::external} {
set where before # check valgrind and asan report for invalid reads after execute
} else { # command so that we have a report that is easier to reproduce
set where after set valgrind_errors [find_valgrind_errors [srv 0 stderr] false]
} set asan_errors [sanitizer_errors_from_file [srv 0 stderr]]
r linsert key $where $otherele $ele if {$valgrind_errors != "" || $asan_errors != ""} {
} puts "valgrind or asan found an issue"
6 { set print_commands true
set index [randomInt [r llen key]]
set otherele [r lindex key $index]
r lrem key 1 $otherele
}
} }
} }
if {$print_commands} {
puts "violating commands:"
foreach cmd $sent {
puts $cmd
}
}
assert_equal $crash false
} }
} ;# foreach comp } ;# foreach comp
......
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