summaryrefslogtreecommitdiffstats
path: root/include
diff options
context:
space:
mode:
Diffstat (limited to 'include')
-rw-r--r--include/framework/DynamicExtension.h14
-rw-r--r--include/framework/reconstruction/BSMPolicy.h10
-rw-r--r--include/framework/reconstruction/BackgroundTieringPolicy.h13
-rw-r--r--include/framework/reconstruction/LevelingPolicy.h30
-rw-r--r--include/framework/reconstruction/TieringPolicy.h13
-rw-r--r--include/framework/structure/ExtensionStructure.h3
-rw-r--r--include/framework/structure/InternalLevel.h8
7 files changed, 57 insertions, 34 deletions
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<size_t> cnt = 0;
- // size_t recon_id = cnt.fetch_add(1);
+ static std::atomic<size_t> cnt = 0;
+ size_t recon_id = cnt.fetch_add(1);
size_t new_head = 0;
std::vector<reconstruction_results<ShardType>> reconstructions;
@@ -570,7 +570,7 @@ private:
args->version->set_structure(std::move(std::unique_ptr<StructureType>(
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<ShardType, QueryType> {
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<ReconstructionVector>
get_reconstruction_tasks(const Version<ShardType, QueryType> *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<ShardType, QueryType
LevelVector;
public:
- BackgroundTieringPolicy(size_t scale_factor, size_t buffer_size)
- : m_scale_factor(scale_factor), m_buffer_size(buffer_size) {}
+ BackgroundTieringPolicy(size_t scale_factor, size_t buffer_size, size_t size_modifier=0)
+ : m_scale_factor(scale_factor), m_buffer_size(buffer_size), m_size_modifier(size_modifier) {}
std::vector<ReconstructionVector> get_reconstruction_tasks(
const Version<ShardType, QueryType> *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<ShardType, QueryType> {
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<ReconstructionVector>
get_reconstruction_tasks(const Version<ShardType, QueryType> *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<ShardType, QueryType> {
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<ReconstructionVector> get_reconstruction_tasks(
const Version<ShardType, QueryType> *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;
}