From a9ee3f2f7f557adbe5448e4f4fbf71c23fe808b1 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Mon, 7 Apr 2025 14:37:19 -0400 Subject: Implemented the legacy policies and fixed a few bugs --- include/framework/reconstruction/BSMPolicy.h | 38 +++++++++----- include/framework/reconstruction/LevelingPolicy.h | 62 +++++++++++++---------- 2 files changed, 60 insertions(+), 40 deletions(-) (limited to 'include/framework/reconstruction') diff --git a/include/framework/reconstruction/BSMPolicy.h b/include/framework/reconstruction/BSMPolicy.h index ae17182..42ae506 100644 --- a/include/framework/reconstruction/BSMPolicy.h +++ b/include/framework/reconstruction/BSMPolicy.h @@ -21,11 +21,13 @@ class BSMPolicy : public ReconstructionPolicy { LevelVector; public: - BSMPolicy(size_t buffer_size, size_t scale_factor, size_t modifier=0) - : m_scale_factor(scale_factor), m_buffer_size(buffer_size), m_size_modifier(modifier) {} + BSMPolicy(size_t buffer_size, size_t scale_factor, size_t modifier = 0) + : m_scale_factor(scale_factor), m_buffer_size(buffer_size), + m_size_modifier(modifier) {} std::vector - get_reconstruction_tasks(const Version *version, LockManager &lock_mngr) const override { + get_reconstruction_tasks(const Version *version, + LockManager &lock_mngr) const override { return {}; } @@ -45,23 +47,31 @@ public: ReconstructionTask task; task.target = target_level; - task.type = ReconstructionType::Merge; - std::vector source_shards; - size_t reccnt = 0; + if (target_level == 1 && + (levels.size() == 1 || levels[1]->get_record_count() == 0)) { + /* if the first level is empty, then we just append the buffer to it */ + task.type = ReconstructionType::Append; + } else { + /* otherwise, we'll need to do a merge of at least two shards */ + task.type = ReconstructionType::Merge; + } - source_shards.push_back({0, all_shards_idx}); + size_t reccnt = 0; + if (target_level < (ssize_t)levels.size() && levels[target_level]->get_record_count() > 0) { + task.sources.push_back({target_level, all_shards_idx}); + } - for (level_index i = target_level; i > source_level; i--) { - if (i < (level_index)levels.size()) { - source_shards.push_back({i-1, all_shards_idx}); - reccnt += levels[i-1]->get_record_count(); - } + for (level_index i = target_level - 1; i >= source_level; i--) { + assert(i < (ssize_t)levels.size()); + task.sources.push_back({i, all_shards_idx}); + reccnt += levels[i]->get_record_count(); } - assert(source_shards.size() > 0); + task.reccnt = reccnt; + assert(task.sources.size() > 0); + reconstructions.add_reconstruction(task); - reconstructions.add_reconstruction(source_shards, target_level, reccnt, ReconstructionType::Merge); return reconstructions; } diff --git a/include/framework/reconstruction/LevelingPolicy.h b/include/framework/reconstruction/LevelingPolicy.h index d448fee..6e20cdb 100644 --- a/include/framework/reconstruction/LevelingPolicy.h +++ b/include/framework/reconstruction/LevelingPolicy.h @@ -21,11 +21,13 @@ class LevelingPolicy : public ReconstructionPolicy { LevelVector; public: - LevelingPolicy(size_t scale_factor, size_t buffer_size, size_t modifier=0) - : m_scale_factor(scale_factor), m_buffer_size(buffer_size), m_size_modifier(modifier) {} + LevelingPolicy(size_t scale_factor, size_t buffer_size, size_t modifier = 0) + : m_scale_factor(scale_factor), m_buffer_size(buffer_size), + m_size_modifier(modifier) {} std::vector - get_reconstruction_tasks(const Version *version, LockManager &lock_mngr) const override { + get_reconstruction_tasks(const Version *version, + LockManager &lock_mngr) const override { return {}; } @@ -34,15 +36,8 @@ public: ReconstructionVector reconstructions; auto levels = version->get_structure()->get_level_vector(); - /* - * on the very first flush, the whole structure will be empty, so - * there isn't anything to merge into level 1 from level 0 yet. - */ - if (version->get_structure()->get_record_count() == 0) { - return reconstructions; - } - - level_index target_level = find_reconstruction_target(levels, version->get_structure()->get_record_count()); + level_index target_level = find_reconstruction_target( + levels, version->get_structure()->get_record_count()); assert(target_level != -1); level_index source_level = 0; @@ -51,29 +46,43 @@ public: target_level = levels.size(); } - for (level_index i = target_level; i > source_level; i--) { - size_t target_reccnt = - (i < (level_index)levels.size()) ? levels[i]->get_record_count() : 0; - size_t total_reccnt = - (i == 0) ? m_buffer_size + target_reccnt - : levels[i - 1]->get_record_count() + target_reccnt; + /* + * For leveling, the only "actual" reconstruction happens at the target + * level. All the other reconstructions simply shift the levels down + * without needing to do any rebuilding. + */ + size_t target_reccnt = (target_level < (level_index)levels.size()) + ? levels[target_level]->get_record_count() + : 0; + size_t total_reccnt = + (target_level == 1) + ? m_buffer_size + target_reccnt + : levels[target_level - 1]->get_record_count() + target_reccnt; + auto type = (target_level >= (level_index) levels.size()) ? ReconstructionType::Append + : ReconstructionType::Merge; + reconstructions.add_reconstruction(target_level - 1, target_level, + total_reccnt, type); - if (total_reccnt > 0) { - auto type = (i >= (level_index) levels.size()) ? ReconstructionType::Append : ReconstructionType::Merge; - reconstructions.add_reconstruction(i - 1, i, total_reccnt, type); - } + /* + * For all other levels, we'll shift them down by using a single-source + * append. + */ + for (level_index i = target_level - 1; i > source_level; i--) { + reconstructions.add_reconstruction(i - 1, i, 0, + ReconstructionType::Append); } - return reconstructions; } private: - level_index find_reconstruction_target(LevelVector &levels, size_t reccnt) const { + level_index find_reconstruction_target(LevelVector &levels, + size_t reccnt) const { level_index target_level = invalid_level_idx; size_t incoming_records = m_buffer_size; for (level_index i = 1; i < (level_index)levels.size(); i++) { - if (levels[i]->get_record_count() + incoming_records < capacity(i, reccnt)) { + if (levels[i]->get_record_count() + incoming_records < + capacity(i, reccnt)) { target_level = i; break; } @@ -85,7 +94,8 @@ private: } inline size_t capacity(level_index level, size_t reccnt) const { - return m_buffer_size * pow(m_scale_factor * pow(std::log(reccnt), m_size_modifier), level); + return m_buffer_size * + pow(m_scale_factor * pow(std::log(reccnt), m_size_modifier), level); } size_t m_scale_factor; -- cgit v1.2.3