From 5617bed5257506d3dfda8537b16f44b3e40f1b42 Mon Sep 17 00:00:00 2001 From: "Douglas B. Rumbaugh" Date: Sun, 22 Dec 2024 13:00:19 -0500 Subject: Began overhauling reconstruction mechanism --- include/framework/reconstruction/BSMPolicy.h | 83 +++++++++++++++++++++ include/framework/reconstruction/LevelingPolicy.h | 84 ++++++++++++++++++++++ .../reconstruction/ReconstructionPolicy.h | 30 ++++++++ include/framework/reconstruction/TieringPolicy.h | 79 ++++++++++++++++++++ 4 files changed, 276 insertions(+) create mode 100644 include/framework/reconstruction/BSMPolicy.h create mode 100644 include/framework/reconstruction/LevelingPolicy.h create mode 100644 include/framework/reconstruction/ReconstructionPolicy.h create mode 100644 include/framework/reconstruction/TieringPolicy.h (limited to 'include/framework/reconstruction') diff --git a/include/framework/reconstruction/BSMPolicy.h b/include/framework/reconstruction/BSMPolicy.h new file mode 100644 index 0000000..9138bd1 --- /dev/null +++ b/include/framework/reconstruction/BSMPolicy.h @@ -0,0 +1,83 @@ +/* + * include/framework/reconstruction/LevelingPolicy.h + * + * Copyright (C) 2023-2024 Douglas B. Rumbaugh + * Dong Xie + * + * Distributed under the Modified BSD License. + * + */ +#pragma once +#include + +#include "framework/reconstruction/ReconstructionPolicy.h" +#include "framework/scheduling/Epoch.h" +#include "util/types.h" + +namespace de { +template QueryType> +class BSMPolicy : ReconstructionPolicy { + typedef std::vector>> + LevelVector; + +public: + BSMPolicy(size_t scale_factor, size_t buffer_size) + : m_scale_factor(scale_factor), m_buffer_size(buffer_size) {} + + ReconstructionVector + get_reconstruction_tasks(Epoch *epoch, + size_t incoming_reccnt) override { + ReconstructionVector reconstructions; + auto levels = epoch->get_structure()->get_level_vector(); + + level_index target_level = find_reconstruction_target(levels); + assert(target_level != -1); + level_index source_level = 0; + + if (target_level == invalid_level_idx) { + /* grow */ + target_level = levels.size(); + } + + ReconstructionTask task; + task.target = target_level; + task.type = ReconstructionType::Merge; + + for (level_index i = target_level; i > source_level; i--) { + if (i < levels.size()) { + task.add_shard({i, all_shards_idx}, levels[i]->get_record_count()); + } + } + + reconstructions.add_reconstruction(task); + return reconstructions; + } + + ReconstructionTask + get_flush_task(Epoch *epoch) override { + return ReconstructionTask{ + {{buffer_shid}}, 0, m_buffer_size, ReconstructionType::Flush}; + } + +private: + level_index find_reconstruction_target(LevelVector &levels) { + level_index target_level = 0; + + for (size_t i = 0; i < (level_index)levels.size(); i++) { + if (levels[i].get_record_count() + 1 <= capacity(i)) { + target_level = i; + break; + } + } + + return target_level; + } + + inline size_t capacity(level_index level) { + return m_buffer_size * pow(m_scale_factor, level + 1); + } + + size_t m_scale_factor; + size_t m_buffer_size; +}; +} // namespace de diff --git a/include/framework/reconstruction/LevelingPolicy.h b/include/framework/reconstruction/LevelingPolicy.h new file mode 100644 index 0000000..00f2cff --- /dev/null +++ b/include/framework/reconstruction/LevelingPolicy.h @@ -0,0 +1,84 @@ +/* + * include/framework/reconstruction/LevelingPolicy.h + * + * Copyright (C) 2023-2024 Douglas B. Rumbaugh + * Dong Xie + * + * Distributed under the Modified BSD License. + * + */ +#pragma once +#include + +#include "framework/reconstruction/ReconstructionPolicy.h" +#include "framework/scheduling/Epoch.h" +#include "util/types.h" + +namespace de { +template QueryType> +class LevelingPolicy : ReconstructionPolicy { + typedef std::vector>> + LevelVector; + +public: + LevelingPolicy(size_t scale_factor, size_t buffer_size) + : m_scale_factor(scale_factor), m_buffer_size(buffer_size) {} + + ReconstructionVector + get_reconstruction_tasks(Epoch *epoch, + size_t incoming_reccnt) override { + ReconstructionVector reconstructions; + auto levels = epoch->get_structure()->get_level_vector(); + + level_index target_level = find_reconstruction_target(levels); + assert(target_level != -1); + level_index source_level = 0; + + if (target_level == invalid_level_idx) { + /* grow */ + target_level = levels.size(); + } + + for (level_index i = target_level; i > source_level; i--) { + size_t target_reccnt = + (i < levels.size()) ? levels[i]->get_record_count() : 0; + size_t total_reccnt = levels[i - 1]->get_record_count() + target_reccnt; + + reconstructions.add_reconstruction(i - 1, i, total_reccnt, + ReconstructionType::Merge); + } + + return reconstructions; + } + + ReconstructionTask + get_flush_task(Epoch *epoch) override { + return ReconstructionTask{ + {{buffer_shid}}, 0, m_buffer_size, ReconstructionType::Flush}; + } + +private: + level_index find_reconstruction_target(LevelVector &levels) { + level_index target_level = 0; + size_t incoming_records = m_buffer_size; + + for (size_t i = 0; i < (level_index)levels.size(); i++) { + if (levels[i].get_record_count() + incoming_records < capacity(i)) { + target_level = i; + break; + } + + incoming_records = levels[i].get_record_count(); + } + + return target_level; + } + + inline size_t capacity(level_index level) { + return m_buffer_size * pow(m_scale_factor, level + 1); + } + + size_t m_scale_factor; + size_t m_buffer_size; +}; +} // namespace de diff --git a/include/framework/reconstruction/ReconstructionPolicy.h b/include/framework/reconstruction/ReconstructionPolicy.h new file mode 100644 index 0000000..976091e --- /dev/null +++ b/include/framework/reconstruction/ReconstructionPolicy.h @@ -0,0 +1,30 @@ +/* + * include/framework/reconstruction/ReconstructionPolicy.h + * + * Reconstruction class interface, used to implement custom reconstruction + * policies. + * + * Copyright (C) 2023-2024 Douglas B. Rumbaugh + * Dong Xie + * + * Distributed under the Modified BSD License. + * + */ +#pragma once + +#include "util/types.h" +#include "framework/structure/ExtensionStructure.h" +#include "framework/scheduling/Epoch.h" + +namespace de { +template QueryType> +class ReconstructionPolicy { + typedef ExtensionStructure StructureType; + +public: + ReconstructionPolicy() {} + virtual ReconstructionVector get_reconstruction_tasks(Epoch *epoch, + size_t incoming_reccnt) = 0; + virtual ReconstructionTask get_flush_task(Epoch *epoch) = 0; + }; +} diff --git a/include/framework/reconstruction/TieringPolicy.h b/include/framework/reconstruction/TieringPolicy.h new file mode 100644 index 0000000..120bcb5 --- /dev/null +++ b/include/framework/reconstruction/TieringPolicy.h @@ -0,0 +1,79 @@ +/* + * include/framework/reconstruction/LevelingPolicy.h + * + * Copyright (C) 2023-2024 Douglas B. Rumbaugh + * Dong Xie + * + * Distributed under the Modified BSD License. + * + */ +#pragma once +#include + +#include "framework/reconstruction/ReconstructionPolicy.h" +#include "framework/scheduling/Epoch.h" +#include "util/types.h" + +namespace de { +template QueryType> +class TieringPolicy : ReconstructionPolicy { + typedef std::vector>> LevelVector; +public: + TieringPolicy(size_t scale_factor, size_t buffer_size) + : m_scale_factor(scale_factor), m_buffer_size(buffer_size) {} + + ReconstructionVector + get_reconstruction_tasks(Epoch *epoch, + size_t incoming_reccnt) override { + ReconstructionVector reconstructions; + auto levels = epoch->get_structure()->get_level_vector(); + + level_index target_level = find_reconstruction_target(levels); + assert(target_level != -1); + level_index source_level = 0; + + if (target_level == invalid_level_idx) { + /* grow */ + target_level = levels.size(); + } + + for (level_index i = target_level; i > source_level; i--) { + size_t target_reccnt = + (i < levels.size()) ? levels[i]->get_record_count() : 0; + size_t total_reccnt = levels[i - 1]->get_record_count() + target_reccnt; + + reconstructions.add_reconstruction(i - 1, i, total_reccnt, + ReconstructionType::Compact); + } + + return reconstructions; + } + + ReconstructionTask + get_flush_task(Epoch *epoch) override { + return ReconstructionTask{ + {{buffer_shid}}, 0, m_buffer_size, ReconstructionType::Flush}; + } + +private: + level_index find_reconstruction_target(LevelVector &levels) { + level_index target_level = 0; + + for (size_t i = 0; i < (level_index) levels.size(); i++) { + if (levels[i].get_shard_count() + 1 <= capacity()) { + target_level = i; + break; + } + } + + return target_level; + } + + inline size_t capacity() { + return m_scale_factor; + } + + size_t m_scale_factor; + size_t m_buffer_size; +}; +} // namespace de -- cgit v1.2.3