summaryrefslogtreecommitdiffstats
path: root/include/framework
diff options
context:
space:
mode:
Diffstat (limited to 'include/framework')
-rw-r--r--include/framework/DynamicExtension.h41
-rw-r--r--include/framework/InternalLevel.h33
-rw-r--r--include/framework/ShardInterface.h27
3 files changed, 67 insertions, 34 deletions
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 <RecordInterface R>
+template <RecordInterface R, ShardInterface S>
class DynamicExtension {
- typedef WIRS<R> 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<std::pair<ShardID, Shard*>> shards;
+ std::vector<std::pair<ShardID, S*>> shards;
std::vector<void*> states;
shards.push_back({{-1, -1}, nullptr});
states.push_back(nullptr);
@@ -151,13 +153,15 @@ public:
std::vector<W> shard_weights;
shard_weights.push_back((double) buffer_weight);
+ WIRS<R>::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<Shard *> shards;
+ S *create_ssi() {
+ std::vector<S *> 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<shards.size(); i++) {
@@ -370,7 +374,7 @@ public:
}
}
- Shard *flattened = new Shard(shards_array, j, nullptr);
+ S *flattened = new S(shards_array, j, nullptr);
for (auto shard : shards) {
delete shard;
@@ -390,7 +394,7 @@ private:
double m_max_delete_prop;
double m_max_rejection_rate;
- std::vector<InternalLevel<R> *> m_levels;
+ std::vector<InternalLevel<R, S> *> 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<R>(new_idx, new_shard_cnt));
+ m_levels.emplace_back(new InternalLevel<R, S>(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<R>::merge_levels(m_levels[base_level], m_levels[incoming_level], rng);
+ m_levels[base_level] = InternalLevel<R, S>::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<R>(incoming_level, (LSM_LEVELING) ? 1 : m_scale_factor);
+ m_levels[incoming_level] = new InternalLevel<R, S>(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<R>(0, 1);
+ auto temp_level = new InternalLevel<R, S>(0, 1);
temp_level->append_mem_table(buffer, rng);
- auto new_level = InternalLevel<R>::merge_levels(old_level, temp_level, rng);
+ auto new_level = InternalLevel<R, S>::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<R> *level) {
+ inline void mark_as_unused(InternalLevel<R, S> *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<R> *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 <RecordInterface R>
+template <RecordInterface R, ShardInterface S>
class InternalLevel {
static const size_t REJECTION_TRIGGER_THRESHOLD = 1024;
typedef decltype(R::key) K;
typedef decltype(R::value) V;
- typedef WIRS<R> Shard;
+ //typedef WIRS<R> 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<R>* 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; i<m_shard_cnt; i++) {
shards[i] = (m_structure->m_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<uint64_t>& weights, std::vector<std::pair<ShardID, Shard *>> &shards, std::vector<void*>& shard_states, const K& low, const K& high) {
+ void get_query_states(std::vector<uint64_t>& weights, std::vector<std::pair<ShardID, S *>> &shards, std::vector<void*>& shard_states, void *query_parms) {
for (size_t i=0; i<m_shard_cnt; i++) {
if (m_structure->m_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 <drumbaugh@psu.edu>
+ *
+ * All rights reserved. Published under the Modified BSD License.
+ *
+ */
+#pragma once
+
+#include <concepts>
+
+#include "util/types.h"
+
+template <typename S>
+concept ShardInterface = requires(S s, void *p) {
+ //s.point_lookup();
+ //s.tombstone_lookup();
+ //s.delete_record();
+
+ {s.get_query_state(p)} -> std::convertible_to<void*>;
+ {s.delete_query_state(p)};
+
+ {s.get_record_count()} -> std::convertible_to<size_t>;
+ {s.get_tombstone_count()} -> std::convertible_to<size_t>;
+ {s.get_memory_usage()} -> std::convertible_to<size_t>;
+};