diff options
Diffstat (limited to 'include/framework/reconstruction')
5 files changed, 125 insertions, 34 deletions
diff --git a/include/framework/reconstruction/BSMPolicy.h b/include/framework/reconstruction/BSMPolicy.h index eaa374a..9ddd150 100644 --- a/include/framework/reconstruction/BSMPolicy.h +++ b/include/framework/reconstruction/BSMPolicy.h @@ -25,8 +25,7 @@ public: : m_scale_factor(2), m_buffer_size(buffer_size) {} ReconstructionVector - get_reconstruction_tasks(const Version<ShardType, QueryType> *version, - size_t incoming_reccnt) const override { + get_reconstruction_tasks(const Version<ShardType, QueryType> *version) const override { ReconstructionVector reconstructions; return reconstructions; } @@ -49,13 +48,21 @@ public: task.target = target_level; task.type = ReconstructionType::Merge; + std::vector<ShardID> source_shards; + size_t reccnt = 0; + + source_shards.push_back({0, all_shards_idx}); + for (level_index i = target_level; i > source_level; i--) { if (i < (level_index)levels.size()) { - task.add_shard({i, all_shards_idx}, levels[i]->get_record_count()); + source_shards.push_back({i-1, all_shards_idx}); + reccnt += levels[i-1]->get_record_count(); } } - reconstructions.add_reconstruction(task); + assert(source_shards.size() > 0); + + reconstructions.add_reconstruction(source_shards, target_level, reccnt, ReconstructionType::Merge); return reconstructions; } @@ -63,8 +70,8 @@ private: level_index find_reconstruction_target(LevelVector &levels) const { level_index target_level = invalid_level_idx; - for (level_index i = 0; i < (level_index)levels.size(); i++) { - if (levels[i]->get_record_count() + m_buffer_size <= capacity(i)) { + for (level_index i = 1; i < (level_index)levels.size(); i++) { + if (levels[i]->get_record_count() == 0) { target_level = i; break; } diff --git a/include/framework/reconstruction/BackgroundTieringPolicy.h b/include/framework/reconstruction/BackgroundTieringPolicy.h new file mode 100644 index 0000000..4c266fd --- /dev/null +++ b/include/framework/reconstruction/BackgroundTieringPolicy.h @@ -0,0 +1,88 @@ +/* + * include/framework/reconstruction/LevelingPolicy.h + * + * Copyright (C) 2023-2024 Douglas B. Rumbaugh <drumbaugh@psu.edu> + * Dong Xie <dongx@psu.edu> + * + * Distributed under the Modified BSD License. + * + */ +#pragma once +#include <cmath> + +#include "framework/reconstruction/ReconstructionPolicy.h" +#include "framework/scheduling/Version.h" +#include "util/types.h" + +namespace de { +template <ShardInterface ShardType, QueryInterface<ShardType> QueryType> +class BackgroundTieringPolicy : public ReconstructionPolicy<ShardType, QueryType> { + typedef std::vector<std::shared_ptr<InternalLevel<ShardType, QueryType>>> + LevelVector; + +public: + BackgroundTieringPolicy(size_t scale_factor, size_t buffer_size) + : m_scale_factor(scale_factor), m_buffer_size(buffer_size) {} + + ReconstructionVector get_reconstruction_tasks( + const Version<ShardType, QueryType> *version) const override { + ReconstructionVector reconstructions; + + auto levels = version->get_structure()->get_level_vector(); + + if (levels[0]->get_shard_count() < m_scale_factor) { + return reconstructions; + } + + 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 < (level_index)levels.size()) ? levels[i]->get_record_count() : 0; + size_t total_reccnt = levels[i - 1]->get_record_count() + target_reccnt; + + std::vector<ShardID> shards; + for (ssize_t j=0; j<(ssize_t)levels[i-1]->get_shard_count(); j++) { + shards.push_back({i-1, j}); + } + + reconstructions.add_reconstruction(shards, i, total_reccnt, ReconstructionType::Compact); + } + + return reconstructions; + } + + ReconstructionVector + get_flush_tasks(const Version<ShardType, QueryType> *version) const override { + ReconstructionVector reconstructions; + + return reconstructions; + } + +private: + level_index find_reconstruction_target(LevelVector &levels) const { + level_index target_level = invalid_level_idx; + + for (level_index 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() const { return m_scale_factor; } + + size_t m_scale_factor; + size_t m_buffer_size; +}; +} // namespace de diff --git a/include/framework/reconstruction/FixedShardCountPolicy.h b/include/framework/reconstruction/FixedShardCountPolicy.h index 0768daa..a181052 100644 --- a/include/framework/reconstruction/FixedShardCountPolicy.h +++ b/include/framework/reconstruction/FixedShardCountPolicy.h @@ -25,8 +25,7 @@ public: : m_buffer_size(buffer_size), m_shard_count(shard_count), m_max_reccnt(max_record_count) {} ReconstructionVector - get_reconstruction_tasks(const Version<ShardType, QueryType> *version, - size_t incoming_reccnt) const override { + get_reconstruction_tasks(const Version<ShardType, QueryType> *version) const override { ReconstructionVector reconstructions; return reconstructions; @@ -36,26 +35,25 @@ public: get_flush_tasks(const Version<ShardType, QueryType> *version) const override { auto levels = version->get_structure()->get_level_vector(); - ReconstructionVector v; - if (levels.size() == 0) { - v.add_reconstruction(ReconstructionTask{ - {{buffer_shid}}, 0, m_buffer_size, ReconstructionType::Append}); - return v; - } - - ShardID last_shid = {0, (shard_index) (levels[0]->get_shard_count() - 1)}; - - if (levels[0]->get_shard(last_shid.shard_idx)->get_record_count() + m_buffer_size <= capacity()) { - v.add_reconstruction(ReconstructionTask{ - {{buffer_shid, last_shid}}, 0, m_buffer_size, ReconstructionType::Merge}); - return v; - } else { - v.add_reconstruction(ReconstructionTask{ - {{buffer_shid}}, 0, m_buffer_size, ReconstructionType::Append}); + /* if this is the very first flush, there won't be an L1 yet */ + if (levels.size() > 1 && levels[1]->get_shard_count() > 0) { + ShardID last_shid = {1, (shard_index) (levels[1]->get_shard_count() - 1)}; + if (levels[1]->get_shard(last_shid.shard_idx)->get_record_count() + m_buffer_size <= capacity()) { + auto task = ReconstructionTask { + {{0, 0}, last_shid}, 1, m_buffer_size,ReconstructionType::Merge + }; + v.add_reconstruction(task); return v; + } } + + auto task = ReconstructionTask { + {{0, 0}}, 1, m_buffer_size, ReconstructionType::Append + }; + v.add_reconstruction(task); + return v; } private: diff --git a/include/framework/reconstruction/FloodL0Policy.h b/include/framework/reconstruction/FloodL0Policy.h index 94bed70..b4f453b 100644 --- a/include/framework/reconstruction/FloodL0Policy.h +++ b/include/framework/reconstruction/FloodL0Policy.h @@ -24,8 +24,7 @@ public: FloodL0Policy(size_t buffer_size) : m_buffer_size(buffer_size) {} ReconstructionVector - get_reconstruction_tasks(const Version<ShardType, QueryType> *version, - size_t incoming_reccnt) const override { + get_reconstruction_tasks(const Version<ShardType, QueryType> *version) const override { ReconstructionVector reconstructions; return reconstructions; } diff --git a/include/framework/reconstruction/TieringPolicy.h b/include/framework/reconstruction/TieringPolicy.h index dce5c3c..ae215db 100644 --- a/include/framework/reconstruction/TieringPolicy.h +++ b/include/framework/reconstruction/TieringPolicy.h @@ -17,14 +17,15 @@ namespace de { template <ShardInterface ShardType, QueryInterface<ShardType> QueryType> class TieringPolicy : public ReconstructionPolicy<ShardType, QueryType> { - typedef std::vector<std::shared_ptr<InternalLevel<ShardType, QueryType>>> LevelVector; + typedef std::vector<std::shared_ptr<InternalLevel<ShardType, QueryType>>> + 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(const Version<ShardType, QueryType> *version, - size_t incoming_reccnt) const override { + ReconstructionVector get_reconstruction_tasks( + const Version<ShardType, QueryType> *version) const override { ReconstructionVector reconstructions; return reconstructions; } @@ -59,7 +60,7 @@ private: level_index find_reconstruction_target(LevelVector &levels) const { level_index target_level = invalid_level_idx; - for (level_index i = 0; i < (level_index) levels.size(); i++) { + for (level_index i = 0; i < (level_index)levels.size(); i++) { if (levels[i]->get_shard_count() + 1 <= capacity()) { target_level = i; break; @@ -69,9 +70,7 @@ private: return target_level; } - inline size_t capacity() const { - return m_scale_factor; - } + inline size_t capacity() const { return m_scale_factor; } size_t m_scale_factor; size_t m_buffer_size; |