From 6fd50506d2e50d2faf2478a2883a2ef1b4840a78 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Wed, 17 May 2023 16:00:20 -0400 Subject: Started implementing shard interface (not finished yet) --- include/framework/DynamicExtension.h | 41 ++++++++++++++++------------ include/framework/InternalLevel.h | 33 +++++++++++----------- include/framework/ShardInterface.h | 27 ++++++++++++++++++ include/shard/WIRS.h | 53 ++++++++++++++---------------------- 4 files changed, 87 insertions(+), 67 deletions(-) create mode 100644 include/framework/ShardInterface.h (limited to 'include') diff --git a/include/framework/DynamicExtension.h b/include/framework/DynamicExtension.h index fd42c5f..53b55b1 100644 --- a/include/framework/DynamicExtension.h +++ b/include/framework/DynamicExtension.h @@ -16,6 +16,9 @@ #include "framework/MutableBuffer.h" #include "framework/InternalLevel.h" +#include "framework/ShardInterface.h" + +#include "shard/WIRS.h" #include "ds/Alias.h" #include "util/timer.h" @@ -68,9 +71,8 @@ enum class DeletePolicy { typedef ssize_t level_index; -template +template class DynamicExtension { - typedef WIRS Shard; typedef decltype(R::key) K; typedef decltype(R::value) V; typedef decltype(R::weight) W; @@ -143,7 +145,7 @@ public: // Get the shard weights for each level. Index 0 is the buffer, // represented by nullptr. - std::vector> shards; + std::vector> shards; std::vector states; shards.push_back({{-1, -1}, nullptr}); states.push_back(nullptr); @@ -151,13 +153,15 @@ public: std::vector shard_weights; shard_weights.push_back((double) buffer_weight); + WIRS::wirs_query_parms parms = {lower_key, upper_key}; + for (auto &level : m_levels) { - level->get_shard_weights(shard_weights, shards, states, lower_key, upper_key); + level->get_query_states(shard_weights, shards, states, &parms); } if (shard_weights.size() == 1 && shard_weights[0] == 0) { if (buffer_alias) delete buffer_alias; - for (auto& x: states) Shard::delete_state(x); + for (auto& x: states) S::delete_query_state(x); sampling_bailouts++; return; // no records in the sampling range } @@ -243,7 +247,7 @@ public: } while (sample_idx < sample_sz); if (buffer_alias) delete buffer_alias; - for (auto& x: states) Shard::delete_state(x); + for (auto& x: states) S::delete_query_state(x); enforce_rejection_rate_maximum(rng); } @@ -348,8 +352,8 @@ public: } - Shard *create_ssi() { - std::vector shards; + S *create_ssi() { + std::vector shards; if (m_levels.size() > 0) { for (int i=m_levels.size() - 1; i>= 0; i--) { @@ -359,9 +363,9 @@ public: } } - shards.emplace_back(new Shard(get_buffer(), nullptr)); + shards.emplace_back(new S(get_buffer(), nullptr)); - Shard *shards_array[shards.size()]; + S *shards_array[shards.size()]; size_t j = 0; for (size_t i=0; i *> m_levels; + std::vector *> m_levels; level_index m_last_level_idx; @@ -446,7 +450,7 @@ private: if (new_idx > 0) { assert(m_levels[new_idx - 1]->get_shard(0)->get_tombstone_count() == 0); } - m_levels.emplace_back(new InternalLevel(new_idx, new_shard_cnt)); + m_levels.emplace_back(new InternalLevel(new_idx, new_shard_cnt)); m_last_level_idx++; return new_idx; @@ -526,14 +530,14 @@ private: // merging two memory levels if (LSM_LEVELING) { auto tmp = m_levels[base_level]; - m_levels[base_level] = InternalLevel::merge_levels(m_levels[base_level], m_levels[incoming_level], rng); + m_levels[base_level] = InternalLevel::merge_levels(m_levels[base_level], m_levels[incoming_level], rng); mark_as_unused(tmp); } else { m_levels[base_level]->append_merged_shards(m_levels[incoming_level], rng); } mark_as_unused(m_levels[incoming_level]); - m_levels[incoming_level] = new InternalLevel(incoming_level, (LSM_LEVELING) ? 1 : m_scale_factor); + m_levels[incoming_level] = new InternalLevel(incoming_level, (LSM_LEVELING) ? 1 : m_scale_factor); } @@ -542,9 +546,9 @@ private: if (LSM_LEVELING) { // FIXME: Kludgey implementation due to interface constraints. auto old_level = m_levels[0]; - auto temp_level = new InternalLevel(0, 1); + auto temp_level = new InternalLevel(0, 1); temp_level->append_mem_table(buffer, rng); - auto new_level = InternalLevel::merge_levels(old_level, temp_level, rng); + auto new_level = InternalLevel::merge_levels(old_level, temp_level, rng); m_levels[0] = new_level; delete temp_level; @@ -560,7 +564,7 @@ private: * level may not be able to immediately be deleted, depending upon who * else is using it. */ - inline void mark_as_unused(InternalLevel *level) { + inline void mark_as_unused(InternalLevel *level) { delete level; } @@ -609,6 +613,7 @@ private: * no guarantees about which buffer will be accessed if level_index is -1. */ inline size_t get_level_record_count(level_index idx, MutableBuffer *buffer=nullptr) { + assert(idx >= -1); if (idx == -1) { return (buffer) ? buffer->get_record_count() : get_buffer()->get_record_count(); diff --git a/include/framework/InternalLevel.h b/include/framework/InternalLevel.h index 19bfe9f..18b7de3 100644 --- a/include/framework/InternalLevel.h +++ b/include/framework/InternalLevel.h @@ -14,25 +14,26 @@ #include "util/types.h" #include "util/bf_config.h" -#include "shard/WIRS.h" +#include "framework/ShardInterface.h" +#include "framework/MutableBuffer.h" #include "ds/BloomFilter.h" namespace de { -template +template class InternalLevel { static const size_t REJECTION_TRIGGER_THRESHOLD = 1024; typedef decltype(R::key) K; typedef decltype(R::value) V; - typedef WIRS Shard; + //typedef WIRS S; private: struct InternalLevelStructure { InternalLevelStructure(size_t cap) : m_cap(cap) - , m_shards(new Shard*[cap]{nullptr}) + , m_shards(new S*[cap]{nullptr}) , m_bfs(new BloomFilter*[cap]{nullptr}) {} ~InternalLevelStructure() { @@ -46,7 +47,7 @@ private: } size_t m_cap; - Shard** m_shards; + S** m_shards; BloomFilter** m_bfs; }; @@ -75,49 +76,49 @@ public: new BloomFilter(BF_FPR, new_level->get_tombstone_count() + base_level->get_tombstone_count(), BF_HASH_FUNCS, rng); - Shard* shards[2]; + S* shards[2]; shards[0] = base_level->m_structure->m_shards[0]; shards[1] = new_level->m_structure->m_shards[0]; - res->m_structure->m_shards[0] = new Shard(shards, 2, res->m_structure->m_bfs[0]); + res->m_structure->m_shards[0] = new S(shards, 2, res->m_structure->m_bfs[0]); return res; } void append_mem_table(MutableBuffer* buffer, const gsl_rng* rng) { assert(m_shard_cnt < m_structure->m_cap); m_structure->m_bfs[m_shard_cnt] = new BloomFilter(BF_FPR, buffer->get_tombstone_count(), BF_HASH_FUNCS, rng); - m_structure->m_shards[m_shard_cnt] = new Shard(buffer, m_structure->m_bfs[m_shard_cnt]); + m_structure->m_shards[m_shard_cnt] = new S(buffer, m_structure->m_bfs[m_shard_cnt]); ++m_shard_cnt; } void append_merged_shards(InternalLevel* level, const gsl_rng* rng) { assert(m_shard_cnt < m_structure->m_cap); m_structure->m_bfs[m_shard_cnt] = new BloomFilter(BF_FPR, level->get_tombstone_count(), BF_HASH_FUNCS, rng); - m_structure->m_shards[m_shard_cnt] = new Shard(level->m_structure->m_shards, level->m_shard_cnt, m_structure->m_bfs[m_shard_cnt]); + m_structure->m_shards[m_shard_cnt] = new S(level->m_structure->m_shards, level->m_shard_cnt, m_structure->m_bfs[m_shard_cnt]); ++m_shard_cnt; } - Shard *get_merged_shard() { - Shard *shards[m_shard_cnt]; + S *get_merged_shard() { + S *shards[m_shard_cnt]; for (size_t i=0; im_shards[i]) ? m_structure->m_shards[i] : nullptr; } - return new Shard(shards, m_shard_cnt, nullptr); + return new S(shards, m_shard_cnt, nullptr); } // Append the sample range in-order..... - void get_shard_weights(std::vector& weights, std::vector> &shards, std::vector& shard_states, const K& low, const K& high) { + void get_query_states(std::vector& weights, std::vector> &shards, std::vector& shard_states, void *query_parms) { for (size_t i=0; im_shards[i]) { - auto shard_state = m_structure->m_shards[i]->get_sample_shard_state(low, high); + auto shard_state = m_structure->m_shards[i]->get_query_state(query_parms); if (shard_state->tot_weight > 0) { shards.push_back({{m_level_no, (ssize_t) i}, m_structure->m_shards[i]}); weights.push_back(shard_state->tot_weight); shard_states.emplace_back(shard_state); } else { - Shard::delete_state(shard_state); + S::delete_state(shard_state); } } } @@ -156,7 +157,7 @@ public: return m_structure->m_shards[shard_no]->get_record_at(idx); } - Shard* get_shard(size_t idx) { + S* get_shard(size_t idx) { return m_structure->m_shards[idx]; } diff --git a/include/framework/ShardInterface.h b/include/framework/ShardInterface.h new file mode 100644 index 0000000..5d07a99 --- /dev/null +++ b/include/framework/ShardInterface.h @@ -0,0 +1,27 @@ +/* + * include/shard/ShardInterface.h + * + * Copyright (C) 2023 Douglas Rumbaugh + * + * All rights reserved. Published under the Modified BSD License. + * + */ +#pragma once + +#include + +#include "util/types.h" + +template +concept ShardInterface = requires(S s, void *p) { + //s.point_lookup(); + //s.tombstone_lookup(); + //s.delete_record(); + + {s.get_query_state(p)} -> std::convertible_to; + {s.delete_query_state(p)}; + + {s.get_record_count()} -> std::convertible_to; + {s.get_tombstone_count()} -> std::convertible_to; + {s.get_memory_usage()} -> std::convertible_to; +}; diff --git a/include/shard/WIRS.h b/include/shard/WIRS.h index 2572caf..9f37396 100644 --- a/include/shard/WIRS.h +++ b/include/shard/WIRS.h @@ -25,6 +25,7 @@ namespace de { thread_local size_t wirs_cancelations = 0; + template class WIRS { private: @@ -53,6 +54,11 @@ private: }; public: + struct wirs_query_parms { + K lower_bound; + K upper_bound; + }; + WIRS(MutableBuffer* buffer, BloomFilter* bf) : m_reccnt(0), m_tombstone_cnt(0), m_deleted_cnt(0), m_total_weight(0), m_rejection_cnt(0), m_ts_check_cnt(0), m_root(nullptr) { @@ -178,14 +184,6 @@ public: return false; } - void free_tree(struct wirs_node* node) { - if (node) { - delete node->alias; - free_tree(node->left); - free_tree(node->right); - delete node; - } - } R* sorted_output() const { return m_data; @@ -206,8 +204,10 @@ public: // low - high -> decompose to a set of nodes. // Build Alias across the decomposed nodes. - WIRSState* get_sample_shard_state(const K& lower_key, const K& upper_key) { + WIRSState* get_query_state(void *query_parameters) { auto res = new WIRSState(); + K lower_key = ((wirs_query_parms *) query_parameters)->lower_bound; + K upper_key = ((wirs_query_parms *) query_parameters)->upper_bound; // Simulate a stack to unfold recursion. double tot_weight = 0.0; @@ -236,7 +236,7 @@ public: return res; } - static void delete_state(void *state) { + static void delete_query_state(void *state) { WIRSState *s = (WIRSState *) state; delete s; } @@ -293,26 +293,6 @@ public: return min; } - /* - bool check_delete(K key, V val) { - size_t idx = get_lower_bound(key); - if (idx >= m_reccnt) { - return false; - } - - auto ptr = m_data + get_lower_bound(key); - - while (ptr < m_data + m_reccnt && *ptr < R {key, val}) { - ptr ++; - } - - bool result = (m_tagging) ? ptr->is_deleted() - : *ptr == R {key, val} && ptr->is_tombstone(); - m_rejection_cnt += result; - return result; - } - */ - bool check_tombstone(const R& rec) { m_ts_check_cnt++; size_t idx = get_lower_bound(rec.key); @@ -332,12 +312,10 @@ public: return result; } - - size_t get_memory_utilization() { + size_t get_memory_usage() { return 0; } - size_t get_rejection_count() { return m_rejection_cnt; } @@ -418,6 +396,15 @@ private: m_root = construct_wirs_node(weights, 0, n_groups-1); } + void free_tree(struct wirs_node* node) { + if (node) { + delete node->alias; + free_tree(node->left); + free_tree(node->right); + delete node; + } + } + R* m_data; std::vector m_alias; wirs_node* m_root; -- cgit v1.2.3