Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: swap_version not correctly resumed after load fix. #240

Merged
merged 3 commits into from
Sep 3, 2023
Merged

fix: swap_version not correctly resumed after load fix. #240

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a153b79
fix: swap_version not correctly resumed after load fix.
patpatbear Aug 30, 2023
9f606a2
fix: rocksdb data leak because object.persistent not set after persis…
patpatbear Sep 3, 2023
086dc66
minor fix test case.
patpatbear Sep 3, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions src/ctrip_swap.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -387,6 +387,20 @@ int submitEvictClientRequest(client *c, robj *key, int persist_keep, uint64_t pe

#define getObjectPersistKeep(o) ((o) ? o->persist_keep : 0)

#define setObjectPersistent(o) do { \
if (o) o->persistent = 1; \
} while(0)

#define clearObjectPersistent(o) do { \
if (o) o->persistent = 0; \
} while(0)

#define overwriteObjectPersistent(o,pk) do { \
if (o) o->persistent = pk; \
} while(0)

#define getObjectPersistent(o) ((o) ? o->persistent : 0)

#define setObjectMetaDirty(o) do { \
if (o) o->dirty_meta = 1; \
} while(0)
Expand Down
2 changes: 1 addition & 1 deletion src/ctrip_swap_data.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ int swapDataAna(swapData *d, int thd, struct keyRequest *key_request,
key_request->cmd_intention_flags = 0;
}

if (key_request->cmd_intention == SWAP_DEL && d->value && !d->value->persistent) {
if (key_request->cmd_intention == SWAP_DEL && d->value && !getObjectPersistent(d->value)) {
// no persistent data, skip del
key_request->cmd_intention = SWAP_NOP;
key_request->cmd_intention_flags = 0;
Expand Down
13 changes: 8 additions & 5 deletions src/ctrip_swap_hash.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -451,7 +451,7 @@ int hashSwapIn(swapData *data, void *result, void *datactx) {
robj *swapin = createSwapInObject(result);
/* mark persistent after data swap in without
* persistence deleted, or mark non-persistent else */
swapin->persistent = !data->persistence_deleted;
overwriteObjectPersistent(swapin,!data->persistence_deleted);
dbAdd(data->db,data->key,swapin);
/* expire will be swapped in later by swap framework. */
if (data->cold_meta) {
Expand All @@ -460,7 +460,7 @@ int hashSwapIn(swapData *data, void *result, void *datactx) {
}
} else {
if (result) decrRefCount(result);
if (data->value) data->value->persistent = !data->persistence_deleted;
if (data->value) overwriteObjectPersistent(data->value,!data->persistence_deleted);
}

return 0;
Expand All @@ -469,7 +469,7 @@ int hashSwapIn(swapData *data, void *result, void *datactx) {
/* subkeys already cleaned by cleanObject(to save cpu usage of main thread),
* swapout only updates db.dict keyspace, meta (db.meta/db.expire) swapped
* out by swap framework. */
int hashSwapOut(swapData *data, void *datactx, int clear_dirty, int *totally_out) {
int hashSwapOut(swapData *data, void *datactx, int keep_data, int *totally_out) {
UNUSED(datactx);
serverAssert(!swapDataIsCold(data));

Expand All @@ -478,7 +478,10 @@ int hashSwapOut(swapData *data, void *datactx, int clear_dirty, int *totally_out
dbDeleteDirtySubkeys(data->db,data->key);
}

if (clear_dirty) clearObjectDataDirty(data->value);
if (keep_data) {
clearObjectDataDirty(data->value);
setObjectPersistent(data->value);
}

if (hashTypeLength(data->value) == 0) {
/* all fields swapped out, key turnning into cold:
Expand All @@ -497,7 +500,7 @@ int hashSwapOut(swapData *data, void *datactx, int clear_dirty, int *totally_out
if (data->new_meta) {
dbAddMeta(data->db,data->key,data->new_meta);
data->new_meta = NULL; /* moved to db.meta */
data->value->persistent = 1; /* loss pure hot and persistent data exist. */
setObjectPersistent(data->value); /* loss pure hot and persistent data exist. */
}
if (totally_out) *totally_out = 0;
}
Expand Down
10 changes: 5 additions & 5 deletions src/ctrip_swap_list.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1718,7 +1718,7 @@ int listSwapIn(swapData *data, MOVE void *result_, void *datactx) {
metaListMerge(&main,result);
/* mark persistent after data swap in without
* persistence deleted, or mark non-persistent else */
main.list->persistent = !data->persistence_deleted;
overwriteObjectPersistent(main.list,!data->persistence_deleted);
/* cold key swapped in result (may be empty). */
dbAdd(data->db,data->key,main.list);
/* expire will be swapped in later by swap framework. */
Expand All @@ -1728,7 +1728,7 @@ int listSwapIn(swapData *data, MOVE void *result_, void *datactx) {
metaListDestroy(result);
} else {
if (result) metaListDestroy(result);
if (data->value) data->value->persistent = !data->persistence_deleted;
if (data->value) overwriteObjectPersistent(data->value,!data->persistence_deleted);
}

return 0;
Expand Down Expand Up @@ -1763,8 +1763,8 @@ int listCleanObject(swapData *data, void *datactx_, int keep_data) {
/* subkeys already cleaned by cleanObject(to save cpu usage of main thread),
* swapout only updates db.dict keyspace, meta (db.meta/db.expire) swapped
* out by swap framework. */
int listSwapOut(swapData *data, void *datactx, int clear_dirty, int *totally_out) {
UNUSED(datactx), UNUSED(clear_dirty);
int listSwapOut(swapData *data, void *datactx, int keep_data, int *totally_out) {
UNUSED(datactx), UNUSED(keep_data);
serverAssert(!swapDataIsCold(data));

if (listTypeLength(data->value) == 0) {
Expand All @@ -1784,7 +1784,7 @@ int listSwapOut(swapData *data, void *datactx, int clear_dirty, int *totally_out
if (data->new_meta) {
dbAddMeta(data->db,data->key,data->new_meta);
data->new_meta = NULL; /* moved to db.meta */
data->value->persistent = 1; /* loss pure hot and persistent data exist. */
setObjectPersistent(data->value); /* loss pure hot and persistent data exist. */
}
if (totally_out) *totally_out = 0;
}
Expand Down
11 changes: 8 additions & 3 deletions src/ctrip_swap_persist.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -439,8 +439,16 @@ sds genSwapPersistInfoString(sds info) {
struct listMeta *listMetaCreate();

int keyLoadFixDataInit(keyLoadFixData *fix, redisDb *db, decodedResult *dr) {
uint64_t version;
serverAssert(db->id == dr->dbid);

if (dr->cf != META_CF)
version = ((decodedData*)dr)->version;
else
version = ((decodedMeta*)dr)->version;
server.swap_persist_load_fix_version = MAX(version,
server.swap_persist_load_fix_version);

if (dr->cf != META_CF) {
/* skip orphan (sub)data keys: note that meta key is prefix of data
* subkey, so rocksIter always start init with meta key, except for
Expand All @@ -451,9 +459,6 @@ int keyLoadFixDataInit(keyLoadFixData *fix, redisDb *db, decodedResult *dr) {
objectMeta *cold_meta, *rebuild_meta;
decodedMeta *dm = (decodedMeta*)dr;

server.swap_persist_load_fix_version = MAX(dm->version,
server.swap_persist_load_fix_version);

size_t extlen = dm->extend ? sdslen(dm->extend) : 0;
if (buildObjectMeta(dm->object_type,dm->version,dm->extend,
extlen, &cold_meta)) {
Expand Down
11 changes: 7 additions & 4 deletions src/ctrip_swap_set.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -427,7 +427,7 @@ int setSwapIn(swapData *data, void *result_, void *datactx) {
robj *swapin = createSwapInObject(result);
/* mark persistent after data swap in without
* persistence deleted, or mark non-persistent else */
swapin->persistent = !data->persistence_deleted;
overwriteObjectPersistent(swapin,!data->persistence_deleted);
dbAdd(data->db,data->key,swapin);
/* expire will be swapped in later by swap framework. */
if (data->cold_meta) {
Expand All @@ -436,7 +436,7 @@ int setSwapIn(swapData *data, void *result_, void *datactx) {
}
} else {
if (result) decrRefCount(result);
if (data->value) data->value->persistent = !data->persistence_deleted;
if (data->value) overwriteObjectPersistent(data->value,!data->persistence_deleted);
}

return 0;
Expand All @@ -454,7 +454,10 @@ int setSwapOut(swapData *data, void *datactx, int clear_dirty, int *totally_out)
dbDeleteDirtySubkeys(data->db,data->key);
}

if (clear_dirty) clearObjectDataDirty(data->value);
if (clear_dirty) {
clearObjectDataDirty(data->value);
setObjectPersistent(data->value);
}

if (setTypeSize(data->value) == 0) {
/* all fields swapped out, key turnning into cold:
Expand All @@ -473,7 +476,7 @@ int setSwapOut(swapData *data, void *datactx, int clear_dirty, int *totally_out)
if (data->new_meta) {
dbAddMeta(data->db,data->key,data->new_meta);
data->new_meta = NULL; /* moved to db.meta */
data->value->persistent = 1; /* loss pure hot and persistent data exist. */
setObjectPersistent(data->value); /* loss pure hot and persistent data exist. */
}
if (totally_out) *totally_out = 0;
}
Expand Down
7 changes: 4 additions & 3 deletions src/ctrip_swap_string.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -214,16 +214,17 @@ int wholeKeySwapIn(swapData *data, MOVE void *result, void *datactx) {
swapin = createSwapInObject(result);
/* mark persistent after data swap in without
* persistence deleted, or mark non-persistent else */
swapin->persistent = !data->persistence_deleted;
overwriteObjectPersistent(swapin,!data->persistence_deleted);
dbAdd(data->db,data->key,swapin);
return 0;
}

int wholeKeySwapOut(swapData *data, void *datactx, int clear_dirty, int *totally_out) {
int wholeKeySwapOut(swapData *data, void *datactx, int keep_data, int *totally_out) {
UNUSED(datactx);
redisDb *db = data->db;
robj *key = data->key;
if (clear_dirty) {
if (keep_data) {
setObjectPersistent(data->value);
clearObjectDataDirty(data->value);
if (totally_out) *totally_out = 0;
} else {
Expand Down
13 changes: 8 additions & 5 deletions src/ctrip_swap_zset.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -649,7 +649,7 @@ int zsetSwapIn(swapData *data_, void *result_, void *datactx_) {
robj *swapin = createSwapInObject(result);
/* mark persistent after data swap in without
* persistence deleted, or mark non-persistent else */
swapin->persistent = !data->sd.persistence_deleted;
overwriteObjectPersistent(swapin,!data->sd.persistence_deleted);
dbAdd(data->sd.db,data->sd.key,swapin);
/* expire will be swapped in later by swap framework. */
if (data->sd.cold_meta) {
Expand All @@ -658,15 +658,15 @@ int zsetSwapIn(swapData *data_, void *result_, void *datactx_) {
}
} else {
if (result) decrRefCount(result);
if (data->sd.value) data->sd.value->persistent = !data->sd.persistence_deleted;
if (data->sd.value) overwriteObjectPersistent(data->sd.value,!data->sd.persistence_deleted);
}
return 0;
}

/* subkeys already cleaned by cleanObject(to save cpu usage of main thread),
* swapout only updates db.dict keyspace, meta (db.meta/db.expire) swapped
* out by swap framework. */
int zsetSwapOut(swapData *data, void *datactx, int clear_dirty, int *totally_out) {
int zsetSwapOut(swapData *data, void *datactx, int keep_data, int *totally_out) {
UNUSED(datactx);
serverAssert(!swapDataIsCold(data));

Expand All @@ -675,7 +675,10 @@ int zsetSwapOut(swapData *data, void *datactx, int clear_dirty, int *totally_out
dbDeleteDirtySubkeys(data->db,data->key);
}

if (clear_dirty) clearObjectDataDirty(data->value);
if (keep_data) {
clearObjectDataDirty(data->value);
setObjectPersistent(data->value);
}

if (zsetLength(data->value) == 0) {
/* all fields swapped out, key turnning into cold:
Expand All @@ -694,7 +697,7 @@ int zsetSwapOut(swapData *data, void *datactx, int clear_dirty, int *totally_out
if (data->new_meta) {
dbAddMeta(data->db,data->key,data->new_meta);
data->new_meta = NULL; /* moved to db.meta */
data->value->persistent = 1; /* loss pure hot and persistent data exist. */
setObjectPersistent(data->value); /* loss pure hot and persistent data exist. */
}
if (totally_out) *totally_out = 0;
}
Expand Down
35 changes: 34 additions & 1 deletion tests/swap/integration/persist.tcl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -198,7 +198,7 @@ start_server {tags {persist} overrides {swap-persist-enabled yes swap-dirty-subk
wait_key_clean r myhash5
set now [clock milliseconds]
# wont persist across writes
assert {[expr $now - $start] < 1000}
assert {[expr $now - $start] < 2000}

restart_server 0 true false

Expand Down Expand Up @@ -273,3 +273,36 @@ start_server {tags {persist} overrides {swap-persist-enabled yes swap-dirty-subk
assert {[expr $dbsize_before - $dbsize_after] < 1000}
}
}

start_server {tags {persist} overrides {swap-persist-enabled yes swap-dirty-subkeys-enabled yes}} {
test {swap_version not correctly resumed after load fix} {
r hmset myhash0 a a b b c c
r swap.evict myhash0
wait_key_cold r myhash0
r del myhash0

restart_server 0 true false

r hmset myhash0 e e
r swap.evict myhash0
wait_key_cold r myhash0
r hgetall myhash0
}
}

start_server {tags {persist} overrides {swap-persist-enabled yes swap-dirty-subkeys-enabled yes}} {
r config set swap-debug-evict-keys 0
test {persist keep data did not flag key persisted causing rocksdb data leak} {
r set mystring val
r hmset myhash a a b b c c
r sadd myset a b c
r zadd myzset 1 a 2 b 3 c
r expire mystring 1
r expire myhash 1
r expire myset 1
r expire myzset 1
after 1500
assert_equal [llength [r swap rio-scan meta {}]] 0
}
}

2 changes: 1 addition & 1 deletion tests/swap/support/util.tcl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,7 @@ proc keyspace_is_cold {r} {
}

proc wait_key_clean {r key} {
wait_for_condition 500 40 {
wait_for_condition 500 100 {
![object_is_dirty $r $key]
} else {
fail "wait $key clean failed."
Expand Down
Loading