From 2089a9ef6cd4ccbc686139d47a3edd99b2b6fdfa Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Sat, 3 May 2025 14:29:29 +0300 Subject: [PATCH] WIP Support "prefetch" in kv_db --- src/kv/kv_db.cpp | 107 ++++++++++++++++++++++++++++------------------- 1 file changed, 65 insertions(+), 42 deletions(-) diff --git a/src/kv/kv_db.cpp b/src/kv/kv_db.cpp index 2343a3aa..df508405 100644 --- a/src/kv/kv_db.cpp +++ b/src/kv/kv_db.cpp @@ -96,7 +96,7 @@ struct kv_block_t void set_data_size(); static int kv_size(const std::string & key, const std::string & value); - int parse(uint64_t offset, uint8_t *data, int size); + int parse(uint64_t offset, uint8_t *data, int size, bool allow_empty = false); bool serialize(uint8_t *data, int size); void apply_change(); void cancel_change(); @@ -243,13 +243,13 @@ static std::string read_string(uint8_t *data, int size, int *pos) return key; } -int kv_block_t::parse(uint64_t offset, uint8_t *data, int size) +int kv_block_t::parse(uint64_t offset, uint8_t *data, int size, bool allow_empty) { kv_stored_block_t *blk = (kv_stored_block_t *)data; if (blk->magic == 0 || blk->type == KV_EMPTY) { // empty block - if (offset != 0) + if (offset != 0 && !allow_empty) fprintf(stderr, "K/V: Block %ju is %s\n", offset, blk->magic == 0 ? "empty" : "cleared"); return -ENOTBLK; } @@ -884,7 +884,10 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p } else { - op->len = db->kv_block_size; + op->offset = (offset / db->ino_block_size) * db->ino_block_size; + op->len = db->ino_block_size; + //op->offset = offset; + //op->len = db->kv_block_size; op->iov.push_back(malloc_or_die(op->len), op->len); } op->callback = [=](cluster_op_t *op) @@ -898,59 +901,82 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p delete op; return; } + int blk_err = 0; + int blk_status = 0; invalidate(db, op->offset, op->version); - auto blk_it = db->block_cache.find(op->offset); - if (blk_it != db->block_cache.end() && - // read may start BEFORE update and end AFTER update, in this case known will be > returned version - (db->known_versions[op->offset/db->ino_block_size] >= op->version && !blk_it->second.invalidated || blk_it->second.updating > 0)) + for (uint64_t in_offset = op->offset; in_offset < (op->offset+op->len) || in_offset == offset; in_offset += db->kv_block_size) { - auto blk = &db->block_cache.at(op->offset); - if (blk->updating > 0 && recheck_policy == KV_RECHECK_WAIT) + auto blk_it = db->block_cache.find(in_offset); + if (blk_it != db->block_cache.end() && + // read may start BEFORE update and end AFTER update, in this case known will be > returned version + (db->known_versions[in_offset/db->ino_block_size] >= op->version && !blk_it->second.invalidated || blk_it->second.updating > 0)) { - // Wait until block update stops - if (op->len) - free(op->iov.buf[0].iov_base); - delete op; - db->continue_update.emplace(blk->offset, [=, blk_offset = blk->offset]() + auto blk = &db->block_cache.at(in_offset); + if (blk->updating > 0 && recheck_policy == KV_RECHECK_WAIT) { - get_block(db, offset, cur_level, recheck_policy, cb); - db->run_continue_update(blk_offset); - }); - return; + // Wait until block update stops + if (op->len) + free(op->iov.buf[0].iov_base); + delete op; + db->continue_update.emplace(in_offset, [=, blk_offset = blk->offset]() + { + get_block(db, offset, cur_level, recheck_policy, cb); + db->run_continue_update(blk_offset); + }); + return; + } } - blk->usage = db->usage_counter; - cb(0, blk->updating > 0 ? BLK_UPDATING : BLK_NOCHANGE); - } - else - { - if (!op->len) + else if (!op->len) { // Version check failed, re-read block delete op; get_block(db, offset, cur_level, recheck_policy, cb); return; } - auto blk = &db->block_cache[op->offset]; - if (blk_it != db->block_cache.end()) + } + for (uint64_t in_offset = op->offset; in_offset < (op->offset+op->len) || in_offset == offset; in_offset += db->kv_block_size) + { + auto blk_it = db->block_cache.find(in_offset); + if (blk_it != db->block_cache.end() && + // read may start BEFORE update and end AFTER update, in this case known will be > returned version + (db->known_versions[in_offset/db->ino_block_size] >= op->version && !blk_it->second.invalidated || blk_it->second.updating > 0)) { - del_block_level(db, blk); - *blk = {}; - } - int err = blk->parse(op->offset, (uint8_t*)op->iov.buf[0].iov_base, op->len); - if (err == 0) - { - blk->level = cur_level; + auto blk = &db->block_cache.at(in_offset); blk->usage = db->usage_counter; - add_block_level(db, blk); - cb(0, BLK_RELOADED); + if (in_offset == offset) + { + blk_err = 0; + blk_status = (blk->updating > 0 ? BLK_UPDATING : BLK_NOCHANGE); + } } else { - db->block_cache.erase(op->offset); - cb(err, BLK_NOCHANGE); + auto blk = &db->block_cache[in_offset]; + if (blk_it != db->block_cache.end()) + { + del_block_level(db, blk); + *blk = {}; + } + int err = blk->parse(in_offset, (uint8_t*)op->iov.buf[0].iov_base + (in_offset - op->offset), db->kv_block_size, in_offset != offset); + if (in_offset == offset) + { + blk_err = err; + blk_status = err == 0 ? BLK_RELOADED : BLK_NOCHANGE; + } + if (err == 0) + { + blk->level = cur_level; + blk->usage = db->usage_counter; + add_block_level(db, blk); + } + else + { + db->block_cache.erase(in_offset); + } + try_evict(db); } - try_evict(db); } + cb(blk_err, blk_status); if (op->len) free(op->iov.buf[0].iov_base); delete op; @@ -1860,7 +1886,6 @@ void kv_op_t::next_handle_block(int res, int refresh) else { // OK, leaf block found - recheck_policy = KV_RECHECK_NONE; next_get(); } } @@ -1887,7 +1912,6 @@ void kv_op_t::next_get() else if (blk->type == KV_LEAF_SPLIT) { // Left half finished, go to the right - recheck_policy = KV_RECHECK_LEAF; key = blk->right_half; skip_equal = false; prev_key_ge = blk->right_half; @@ -1921,7 +1945,6 @@ void kv_op_t::next_go_up() path.pop_back(); cur_level--; cur_block = path[path.size()-1].offset; - recheck_policy = KV_RECHECK_LEAF; // Check if we can resume listing from the next key auto pb_it = db->block_cache.find(cur_block); if (pb_it == db->block_cache.end())