From 6bdcf74ad91e0efaa8c2e4339f5085fde8a7982b Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Sat, 5 Apr 2025 19:08:12 -0400 Subject: working commit (temporary progress, doesn't build) --- include/framework/DynamicExtension.h | 14 +++++----- include/framework/reconstruction/BSMPolicy.h | 10 +++++--- .../reconstruction/BackgroundTieringPolicy.h | 13 +++++----- include/framework/reconstruction/LevelingPolicy.h | 30 +++++++++++++++------- include/framework/reconstruction/TieringPolicy.h | 13 +++++----- include/framework/structure/ExtensionStructure.h | 3 +++ include/framework/structure/InternalLevel.h | 8 ++++-- 7 files changed, 57 insertions(+), 34 deletions(-) (limited to 'include/framework') diff --git a/include/framework/DynamicExtension.h b/include/framework/DynamicExtension.h index 762029e..9d76813 100644 --- a/include/framework/DynamicExtension.h +++ b/include/framework/DynamicExtension.h @@ -461,8 +461,8 @@ private: auto extension = (DynamicExtension *)args->extension; extension->SetThreadAffinity(); - // static std::atomic cnt = 0; - // size_t recon_id = cnt.fetch_add(1); + static std::atomic cnt = 0; + size_t recon_id = cnt.fetch_add(1); size_t new_head = 0; std::vector> reconstructions; @@ -570,7 +570,7 @@ private: args->version->set_structure(std::move(std::unique_ptr( active_version->get_structure()->copy()))); - // size_t cur_reccnt = args->version->get_structure()->get_record_count(); + size_t cur_reccnt = args->version->get_structure()->get_record_count(); /* apply our updates to the copied structure (adding/removing shards) */ for (auto recon : reconstructions) { @@ -582,7 +582,7 @@ private: } } - // size_t new_reccnt = args->version->get_structure()->get_record_count(); + size_t new_reccnt = args->version->get_structure()->get_record_count(); // fprintf(stderr, "\t[I] Post-reconstruction L0 Size\t%ld (%ld)\n", // args->version->get_structure()->get_level_vector()[0]->get_shard_count(), @@ -595,9 +595,9 @@ private: active_version->get_head()); // fprintf(stderr, "\t[I] Buffer head set to %ld (%ld)\n", // active_version->get_head(), recon_id); - // if (new_reccnt != cur_reccnt) { - // fprintf(stderr, "ERROR: invalid reccnt (%ld)\n", recon_id); - // } + if (new_reccnt != cur_reccnt) { + fprintf(stderr, "ERROR: invalid reccnt (%ld)\n", recon_id); + } } // fprintf(stderr, "\t[I] Record Counts: %ld %ld %ld (%ld)\n", old_reccnt, diff --git a/include/framework/reconstruction/BSMPolicy.h b/include/framework/reconstruction/BSMPolicy.h index 6d55a12..ae17182 100644 --- a/include/framework/reconstruction/BSMPolicy.h +++ b/include/framework/reconstruction/BSMPolicy.h @@ -21,8 +21,8 @@ class BSMPolicy : public ReconstructionPolicy { LevelVector; public: - BSMPolicy(size_t buffer_size) - : m_scale_factor(2), m_buffer_size(buffer_size) {} + 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 { @@ -79,11 +79,13 @@ private: return target_level; } - inline size_t capacity(level_index level) const { - return m_buffer_size * pow(m_scale_factor, level + 1); + inline size_t capacity(level_index level, size_t reccnt) const { + size_t base = m_scale_factor * pow(log(reccnt), m_size_modifier); + return m_buffer_size * (base - 1) * pow(base, level + 1); } size_t m_scale_factor; size_t m_buffer_size; + size_t m_size_modifier; }; } // namespace de diff --git a/include/framework/reconstruction/BackgroundTieringPolicy.h b/include/framework/reconstruction/BackgroundTieringPolicy.h index ab19e24..36556a2 100644 --- a/include/framework/reconstruction/BackgroundTieringPolicy.h +++ b/include/framework/reconstruction/BackgroundTieringPolicy.h @@ -21,8 +21,8 @@ class BackgroundTieringPolicy : public ReconstructionPolicy get_reconstruction_tasks( const Version *version, LockManager &lock_mngr) const override { @@ -34,7 +34,7 @@ public: return {}; } - level_index target_level = find_reconstruction_target(levels); + level_index target_level = find_reconstruction_target(levels, version->get_structure()->get_record_count()); assert(target_level != -1); level_index source_level = 0; @@ -68,11 +68,11 @@ public: } private: - level_index find_reconstruction_target(LevelVector &levels) const { + level_index find_reconstruction_target(LevelVector &levels, size_t reccnt) const { level_index target_level = invalid_level_idx; for (level_index i = 1; i < (level_index)levels.size(); i++) { - if (levels[i]->get_shard_count() + 1 <= capacity()) { + if (levels[i]->get_shard_count() + 1 <= capacity(reccnt)) { target_level = i; break; } @@ -81,9 +81,10 @@ private: return target_level; } - inline size_t capacity() const { return m_scale_factor; } + inline size_t capacity(size_t reccnt) const { return m_scale_factor * std::pow(std::log(reccnt), m_size_modifier); } size_t m_scale_factor; size_t m_buffer_size; + size_t m_size_modifier; }; } // namespace de diff --git a/include/framework/reconstruction/LevelingPolicy.h b/include/framework/reconstruction/LevelingPolicy.h index f0feb53..d448fee 100644 --- a/include/framework/reconstruction/LevelingPolicy.h +++ b/include/framework/reconstruction/LevelingPolicy.h @@ -21,8 +21,8 @@ class LevelingPolicy : public ReconstructionPolicy { LevelVector; public: - LevelingPolicy(size_t scale_factor, size_t buffer_size) - : m_scale_factor(scale_factor), m_buffer_size(buffer_size) {} + 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 { @@ -34,7 +34,16 @@ public: ReconstructionVector reconstructions; auto levels = version->get_structure()->get_level_vector(); - level_index target_level = find_reconstruction_target(levels); + /* + * 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()); + assert(target_level != -1); level_index source_level = 0; if (target_level == invalid_level_idx) { @@ -49,20 +58,22 @@ public: (i == 0) ? m_buffer_size + target_reccnt : levels[i - 1]->get_record_count() + target_reccnt; - reconstructions.add_reconstruction(i - 1, i, total_reccnt, - ReconstructionType::Merge); + if (total_reccnt > 0) { + auto type = (i >= (level_index) levels.size()) ? ReconstructionType::Append : ReconstructionType::Merge; + reconstructions.add_reconstruction(i - 1, i, total_reccnt, type); + } } return reconstructions; } private: - level_index find_reconstruction_target(LevelVector &levels) 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)) { + if (levels[i]->get_record_count() + incoming_records < capacity(i, reccnt)) { target_level = i; break; } @@ -73,11 +84,12 @@ private: return target_level; } - inline size_t capacity(level_index level) const { - return m_buffer_size * pow(m_scale_factor, level); + 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); } size_t m_scale_factor; size_t m_buffer_size; + size_t m_size_modifier; }; } // namespace de diff --git a/include/framework/reconstruction/TieringPolicy.h b/include/framework/reconstruction/TieringPolicy.h index ce8130e..55f5f47 100644 --- a/include/framework/reconstruction/TieringPolicy.h +++ b/include/framework/reconstruction/TieringPolicy.h @@ -21,8 +21,8 @@ class TieringPolicy : public ReconstructionPolicy { LevelVector; public: - TieringPolicy(size_t scale_factor, size_t buffer_size) - : m_scale_factor(scale_factor), m_buffer_size(buffer_size) {} + TieringPolicy(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 { @@ -34,7 +34,7 @@ public: ReconstructionVector reconstructions; auto levels = version->get_structure()->get_level_vector(); - level_index target_level = find_reconstruction_target(levels); + level_index target_level = find_reconstruction_target(levels, version->get_structure()->get_record_count()); assert(target_level != -1); level_index source_level = 0; @@ -60,11 +60,11 @@ public: } private: - level_index find_reconstruction_target(LevelVector &levels) const { + level_index find_reconstruction_target(LevelVector &levels, size_t reccnt) const { level_index target_level = invalid_level_idx; for (level_index i = 1; i < (level_index)levels.size(); i++) { - if (levels[i]->get_shard_count() + 1 <= capacity()) { + if (levels[i]->get_shard_count() + 1 <= capacity(reccnt)) { target_level = i; break; } @@ -73,9 +73,10 @@ private: return target_level; } - inline size_t capacity() const { return m_scale_factor; } + inline size_t capacity(size_t reccnt) const { return m_scale_factor * std::pow(std::log(reccnt), m_size_modifier); } size_t m_scale_factor; size_t m_buffer_size; + size_t m_size_modifier; }; } // namespace de diff --git a/include/framework/structure/ExtensionStructure.h b/include/framework/structure/ExtensionStructure.h index 521e68b..a77088f 100644 --- a/include/framework/structure/ExtensionStructure.h +++ b/include/framework/structure/ExtensionStructure.h @@ -231,6 +231,9 @@ public: if (shard_idx != -1) { m_levels[shards[i].first]->delete_shard(shard_idx); + } else { + fprintf(stderr, "ERROR: failed to delete shard %ld\t%p\n", shards[i].first, shards[i].second); + //exit(EXIT_FAILURE); } } } diff --git a/include/framework/structure/InternalLevel.h b/include/framework/structure/InternalLevel.h index 6e8b67e..5659c72 100644 --- a/include/framework/structure/InternalLevel.h +++ b/include/framework/structure/InternalLevel.h @@ -102,8 +102,12 @@ public: return false; } - const ShardType *get_shard(size_t idx) const { - if (idx >= m_shards.size()) { + const ShardType *get_shard(ssize_t idx) const { + if (idx == all_shards_idx) { + idx = 0; + } + + if (idx >= (ssize_t) m_shards.size()) { return nullptr; } -- cgit v1.2.3