diff options
Diffstat (limited to 'include/framework')
| -rw-r--r-- | include/framework/DynamicExtension.h | 393 | ||||
| -rw-r--r-- | include/framework/InternalLevel.h | 162 | ||||
| -rw-r--r-- | include/framework/MutableBuffer.h | 115 | ||||
| -rw-r--r-- | include/framework/QueryInterface.h | 22 | ||||
| -rw-r--r-- | include/framework/ShardInterface.h | 9 |
5 files changed, 215 insertions, 486 deletions
diff --git a/include/framework/DynamicExtension.h b/include/framework/DynamicExtension.h index 53b55b1..2009344 100644 --- a/include/framework/DynamicExtension.h +++ b/include/framework/DynamicExtension.h @@ -17,6 +17,7 @@ #include "framework/MutableBuffer.h" #include "framework/InternalLevel.h" #include "framework/ShardInterface.h" +#include "framework/QueryInterface.h" #include "shard/WIRS.h" #include "ds/Alias.h" @@ -71,221 +72,88 @@ enum class DeletePolicy { typedef ssize_t level_index; -template <RecordInterface R, ShardInterface S> +template <RecordInterface R, ShardInterface S, QueryInterface Q, typename FQ=void> class DynamicExtension { - typedef decltype(R::key) K; - typedef decltype(R::value) V; - typedef decltype(R::weight) W; public: - DynamicExtension(size_t buffer_cap, size_t buffer_delete_cap, size_t scale_factor, - double max_delete_prop, double max_rejection_prop, gsl_rng *rng) - : m_active_buffer(0), - m_scale_factor(scale_factor), - m_max_delete_prop(max_delete_prop), - m_max_rejection_rate(max_rejection_prop), - m_last_level_idx(-1), - m_buffer_1(new MutableBuffer<R>(buffer_cap, LSM_REJ_SAMPLE, buffer_delete_cap, rng)), - m_buffer_2(new MutableBuffer<R>(buffer_cap, LSM_REJ_SAMPLE, buffer_delete_cap, rng)), - m_buffer_1_merging(false), m_buffer_2_merging(false) {} + DynamicExtension(size_t buffer_cap, size_t scale_factor, double max_delete_prop) + : m_scale_factor(scale_factor), m_max_delete_prop(max_delete_prop), + m_buffer(new MutableBuffer<R>(buffer_cap, LSM_REJ_SAMPLE, buffer_cap * max_delete_prop)) + {} ~DynamicExtension() { - delete m_buffer_1; - delete m_buffer_2; + delete m_buffer; for (size_t i=0; i<m_levels.size(); i++) { delete m_levels[i]; } } - int delete_record(const K& key, const V& val, gsl_rng *rng) { - assert(DELETE_TAGGING); - - auto buffer = get_buffer(); - // Check the levels first. This assumes there aren't - // any undeleted duplicate records. - for (auto level : m_levels) { - if (level && level->delete_record(key, val)) { - return 1; - } - } - - // the buffer will take the longest amount of time, and - // probably has the lowest probability of having the record, - // so we'll check it last. - return buffer->delete_record(key, val); + int insert(const R &rec) { + return internal_append(rec, false); } - int append(R &rec, gsl_rng *rng) { - // NOTE: single-threaded implementation only + int erase(const R &rec) { MutableBuffer<R> *buffer; - while (!(buffer = get_buffer())) - ; - - if (buffer->is_full()) { - merge_buffer(rng); - } - - return buffer->append(rec); - } - - void range_sample(R *sample_set, const K& lower_key, const K& upper_key, size_t sample_sz, gsl_rng *rng) { - auto buffer = get_buffer(); - Alias *buffer_alias = nullptr; - std::vector<R *> buffer_records; - size_t buffer_cutoff = 0; - - W buffer_weight; - if (LSM_REJ_SAMPLE) { - buffer_weight = buffer->get_total_weight(); - buffer_cutoff = buffer->get_record_count() - 1; - } else { - buffer_weight = buffer->get_sample_range(lower_key, upper_key, buffer_records, &buffer_alias, &buffer_cutoff); - } - - // Get the shard weights for each level. Index 0 is the buffer, - // represented by nullptr. - std::vector<std::pair<ShardID, S*>> shards; - std::vector<void*> states; - shards.push_back({{-1, -1}, nullptr}); - states.push_back(nullptr); - - 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_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) S::delete_query_state(x); - sampling_bailouts++; - return; // no records in the sampling range - } - - double tot_weight = std::accumulate(shard_weights.begin(), shard_weights.end(), 0); - std::vector<double> normalized_weights(shard_weights.size()); - for (size_t i=0; i<shard_weights.size(); i++) { - normalized_weights[i] = ((double) shard_weights[i]) / tot_weight; - } - - // Construct alias structure - auto alias = Alias(normalized_weights); - - std::vector<size_t> shard_samples(shard_weights.size(), 0); - - size_t rejections = sample_sz; - size_t sample_idx = 0; - - size_t buffer_rejections = 0; - - do { - for (size_t i=0; i<rejections; i++) { - shard_samples[alias.get(rng)] += 1; - } - - rejections = 0; - - while (shard_samples[0] > 0) { - const R *rec; - if (LSM_REJ_SAMPLE) { - rec = buffer->get_sample(lower_key, upper_key, rng); - } else { - rec = buffer_records[buffer_alias->get(rng)]; - } + if constexpr (DELETE_TAGGING) { + auto buffer = get_buffer(); - if (DELETE_TAGGING) { - if (rec && !rec->is_deleted()) { - sample_set[sample_idx++] = *rec; - } else { - rejections++; - } - } else { - if (rec && !buffer->check_tombstone(*rec)) { - sample_set[sample_idx++] = *rec; - } else { - rejections++; - } - } - - shard_samples[0]--; - - // Assume nothing in buffer and bail out. - // FIXME: rather than a bailout, we could switch to non-rejection - // sampling, but that would require rebuilding the full alias structure. - // Wouldn't be too hard to do, but for the moment I'll just do this. - if (LSM_REJ_SAMPLE && buffer_rejections >= sample_sz && sample_idx == 0 && shard_weights.size() == 1) { - if (buffer_alias) delete buffer_alias; - //for (auto& x: states) delete x; - sampling_bailouts++; - return; // no records in the sampling range + // Check the levels first. This assumes there aren't + // any undeleted duplicate records. + for (auto level : m_levels) { + if (level && level->delete_record(rec)) { + return 1; } } - std::vector<R> results; - for (size_t i=1; i<shard_samples.size(); i++) { - results.reserve(shard_samples[i]); - - shards[i].second->get_samples(states[i], results, lower_key, upper_key, shard_samples[i], rng); + // the buffer will take the longest amount of time, and + // probably has the lowest probability of having the record, + // so we'll check it last. + return buffer->delete_record(rec); + } - for (size_t j=0; j<results.size(); j++) { - if (rejection(results[j], shards[i].first, lower_key, upper_key, buffer, buffer_cutoff)) { - rejections++; - continue; - } + return internal_append(rec, true); + } - sample_set[sample_idx++] = results[j]; - } + std::vector<R> query(void *parms) { - shard_samples[i] = 0; - results.clear(); - } + // Use the provided top-level query function is one + // is specified. Otherwise, use the default framework + // behavior. + if constexpr (!std::is_same<FQ, void>::value) { + return FQ(parms); + } - } while (sample_idx < sample_sz); + auto buffer = get_buffer(); - if (buffer_alias) delete buffer_alias; - for (auto& x: states) S::delete_query_state(x); + // Get the buffer query state + auto buffer_state = Q::get_buffer_query_state(buffer, parms); - enforce_rejection_rate_maximum(rng); - } + // Get the shard query states + std::vector<std::pair<ShardID, S*>> shards; + std::vector<void*> states; - // Checks the tree and buffer for a tombstone corresponding to - // the provided record in any shard *above* the shid, which - // should correspond to the shard containing the record in question - // - // Passing INVALID_SHID indicates that the record exists within the buffer - bool is_deleted(const R &record, const ShardID &shid, MutableBuffer<R> *buffer, size_t buffer_cutoff) { - // If tagging is enabled, we just need to check if the record has the delete tag set - if (DELETE_TAGGING) { - return record.is_deleted(); + for (auto &level : m_levels) { + level->get_query_states(shards, states, parms); } - // Otherwise, we need to look for a tombstone. - - // check for tombstone in the buffer. This will require accounting for the cutoff eventually. - if (buffer->check_tombstone(record)) { - return true; - } + std::vector<std::vector<R>> query_results(shards.size() + 1); - // if the record is in the buffer, then we're done. - if (shid == INVALID_SHID) { - return false; - } + // Execute the query for the buffer + query_results[0] = Q::buffer_query(buffer, buffer_state, parms); - for (size_t lvl=0; lvl<=shid.level_idx; lvl++) { - if (m_levels[lvl]->check_tombstone(0, record)) { - return true; - } + // Execute the query for each shard + for (size_t i=0; i<shards.size(); i++) { + query_results[i] = post_process(Q::query(shards[i].second, states[i], parms)); } + + // Merge the results together + auto result = Q::merge(query_results); - // check the level containing the shard - return m_levels[shid.level_idx]->check_tombstone(shid.shard_idx + 1, record); + return result; } - size_t get_record_cnt() { size_t cnt = get_buffer()->get_record_count(); @@ -296,7 +164,6 @@ public: return cnt; } - size_t get_tombstone_cnt() { size_t cnt = get_buffer()->get_tombstone_count(); @@ -311,22 +178,22 @@ public: return m_levels.size(); } - size_t get_memory_utilization() { - size_t cnt = m_buffer_1->get_memory_utilization() + m_buffer_2->get_memory_utilization(); + size_t get_memory_usage() { + size_t cnt = m_buffer->get_memory_usage(); for (size_t i=0; i<m_levels.size(); i++) { - if (m_levels[i]) cnt += m_levels[i]->get_memory_utilization(); + if (m_levels[i]) cnt += m_levels[i]->get_memory_usage(); } return cnt; } - size_t get_aux_memory_utilization() { - size_t cnt = m_buffer_1->get_aux_memory_utilization() + m_buffer_2->get_aux_memory_utilization(); + size_t get_aux_memory_usage() { + size_t cnt = m_buffer->get_aux_memory_usage(); for (size_t i=0; i<m_levels.size(); i++) { if (m_levels[i]) { - cnt += m_levels[i]->get_aux_memory_utilization(); + cnt += m_levels[i]->get_aux_memory_usage(); } } @@ -348,10 +215,9 @@ public: } size_t get_buffer_capacity() { - return m_buffer_1->get_capacity(); + return m_buffer->get_capacity(); } - S *create_ssi() { std::vector<S *> shards; @@ -384,57 +250,72 @@ public: } private: - MutableBuffer<R> *m_buffer_1; - MutableBuffer<R> *m_buffer_2; - std::atomic<bool> m_active_buffer; - std::atomic<bool> m_buffer_1_merging; - std::atomic<bool> m_buffer_2_merging; + MutableBuffer<R> *m_buffer; size_t m_scale_factor; double m_max_delete_prop; - double m_max_rejection_rate; - std::vector<InternalLevel<R, S> *> m_levels; - - level_index m_last_level_idx; + std::vector<InternalLevel<R, S, Q> *> m_levels; MutableBuffer<R> *get_buffer() { - if (m_buffer_1_merging && m_buffer_2_merging) { - return nullptr; - } - - return (m_active_buffer) ? m_buffer_2 : m_buffer_1; + return m_buffer; } - inline bool rejection(const R &record, ShardID shid, const K& lower_bound, const K& upper_bound, MutableBuffer<R> *buffer, size_t buffer_cutoff) { - if (record.is_tombstone()) { - tombstone_rejections++; - return true; - } else if (record.key < lower_bound || record.key > upper_bound) { - bounds_rejections++; - return true; - } else if (is_deleted(record, shid, buffer, buffer_cutoff)) { - deletion_rejections++; - return true; + int internal_append(R &rec, bool ts) { + MutableBuffer<R> *buffer; + while (!(buffer = get_buffer())) + ; + + if (buffer->is_full()) { + merge_buffer(); } - return false; + return buffer->append(rec, ts); } - inline bool add_to_sample(const R &record, ShardID shid, const K& upper_key, const K& lower_key, char *io_buffer, - R *sample_buffer, size_t &sample_idx, MutableBuffer<R> *buffer, size_t buffer_cutoff) { - TIMER_INIT(); - TIMER_START(); - sampling_attempts++; - if (!record || rejection(record, shid, lower_key, upper_key, io_buffer, buffer, buffer_cutoff)) { - sampling_rejections++; - return false; + std::vector<R> post_process(std::vector<R> records, ShardID shid, MutableBuffer<R> *buffer) { + std::vector<R> processed_records; + processed_records.reserve(records.size()); + + // For delete tagging, we just need to check the delete bit on each + // record. + if constexpr (DELETE_TAGGING) { + for (auto &rec : records) { + if (rec.is_deleted()) { + continue; + } + + processed_records.emplace_back(rec.rec); + } + + return processed_records; } - TIMER_STOP(); - rejection_check_time += TIMER_RESULT(); - sample_buffer[sample_idx++] = *record; - return true; + // For tombstone deletes, we need to search for the corresponding + // tombstone for each record. + for (auto &rec : records) { + if (rec.is_tombstone()) { + continue; + } + + if (buffer->check_tombstone(rec.rec)) { + continue; + } + + if (shid != INVALID_SHID) { + for (size_t lvl=0; lvl<=shid.level_idx; lvl++) { + if (m_levels[lvl]->check_tombstone(0, rec.rec)) { + continue; + } + } + + if (m_levels[shid.level_idx]->check_tombstone(shid.shard_idx + 1, rec.rec)) { + continue; + } + } + + processed_records.emplace_back(rec.rec); + } } /* @@ -450,24 +331,23 @@ 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, S>(new_idx, new_shard_cnt)); + m_levels.emplace_back(new InternalLevel<R, S, Q>(new_idx, new_shard_cnt)); - m_last_level_idx++; return new_idx; } // Merge the memory table down into the tree, completing any required other // merges to make room for it. - inline void merge_buffer(gsl_rng *rng) { + inline void merge_buffer() { auto buffer = get_buffer(); if (!can_merge_with(0, buffer->get_record_count())) { - merge_down(0, rng); + merge_down(0); } - merge_buffer_into_l0(buffer, rng); - enforce_delete_maximum(0, rng); + merge_buffer_into_l0(buffer); + enforce_delete_maximum(0); buffer->truncate(); return; @@ -479,15 +359,15 @@ private: * routine will recursively perform any necessary merges to make room for the * specified level. */ - inline void merge_down(level_index idx, gsl_rng *rng) { + inline void merge_down(level_index idx) { level_index merge_base_level = find_mergable_level(idx); if (merge_base_level == -1) { merge_base_level = grow(); } for (level_index i=merge_base_level; i>idx; i--) { - merge_levels(i, i-1, rng); - enforce_delete_maximum(i, rng); + merge_levels(i, i-1); + enforce_delete_maximum(i); } return; @@ -509,7 +389,7 @@ private: level_index merge_level_idx; size_t incoming_rec_cnt = get_level_record_count(idx, buffer); - for (level_index i=idx+1; i<=m_last_level_idx; i++) { + for (level_index i=idx+1; i<=m_levels.size(); i++) { if (can_merge_with(i, incoming_rec_cnt)) { return i; } @@ -526,35 +406,35 @@ private: * are skipped in the merge process--otherwise the tombstone ordering * invariant may be violated by the merge operation. */ - inline void merge_levels(level_index base_level, level_index incoming_level, gsl_rng *rng) { + inline void merge_levels(level_index base_level, level_index incoming_level) { // merging two memory levels if (LSM_LEVELING) { auto tmp = m_levels[base_level]; - m_levels[base_level] = InternalLevel<R, S>::merge_levels(m_levels[base_level], m_levels[incoming_level], rng); + m_levels[base_level] = InternalLevel<R, S, Q>::merge_levels(m_levels[base_level], m_levels[incoming_level]); mark_as_unused(tmp); } else { - m_levels[base_level]->append_merged_shards(m_levels[incoming_level], rng); + m_levels[base_level]->append_merged_shards(m_levels[incoming_level]); } mark_as_unused(m_levels[incoming_level]); - m_levels[incoming_level] = new InternalLevel<R, S>(incoming_level, (LSM_LEVELING) ? 1 : m_scale_factor); + m_levels[incoming_level] = new InternalLevel<R, S, Q>(incoming_level, (LSM_LEVELING) ? 1 : m_scale_factor); } - inline void merge_buffer_into_l0(MutableBuffer<R> *buffer, gsl_rng *rng) { + inline void merge_buffer_into_l0(MutableBuffer<R> *buffer) { assert(m_levels[0]); if (LSM_LEVELING) { // FIXME: Kludgey implementation due to interface constraints. auto old_level = m_levels[0]; - auto temp_level = new InternalLevel<R, S>(0, 1); - temp_level->append_mem_table(buffer, rng); - auto new_level = InternalLevel<R, S>::merge_levels(old_level, temp_level, rng); + auto temp_level = new InternalLevel<R, S, Q>(0, 1); + temp_level->append_buffer(buffer); + auto new_level = InternalLevel<R, S, Q>::merge_levels(old_level, temp_level); m_levels[0] = new_level; delete temp_level; mark_as_unused(old_level); } else { - m_levels[0]->append_mem_table(buffer, rng); + m_levels[0]->append_buffer(buffer); } } @@ -564,7 +444,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, S> *level) { + inline void mark_as_unused(InternalLevel<R, S, Q> *level) { delete level; } @@ -573,31 +453,16 @@ private: * if the limit is exceeded, forcibly merge levels until all * levels below idx are below the limit. */ - inline void enforce_delete_maximum(level_index idx, gsl_rng *rng) { + inline void enforce_delete_maximum(level_index idx) { long double ts_prop = (long double) m_levels[idx]->get_tombstone_count() / (long double) calc_level_record_capacity(idx); if (ts_prop > (long double) m_max_delete_prop) { - merge_down(idx, rng); + merge_down(idx); } return; } - inline void enforce_rejection_rate_maximum(gsl_rng *rng) { - if (m_levels.size() == 0) { - return; - } - - for (size_t i=0; i<m_last_level_idx; i++) { - if (m_levels[i]) { - double ratio = m_levels[i]->get_rejection_rate(); - if (ratio > m_max_rejection_rate) { - merge_down(i, rng); - } - } - } - } - /* * Assume that level "0" should be larger than the buffer. The buffer * itself is index -1, which should return simply the buffer capacity. diff --git a/include/framework/InternalLevel.h b/include/framework/InternalLevel.h index 18b7de3..70da821 100644 --- a/include/framework/InternalLevel.h +++ b/include/framework/InternalLevel.h @@ -16,85 +16,52 @@ #include "util/bf_config.h" #include "framework/ShardInterface.h" #include "framework/MutableBuffer.h" +#include "framework/QueryInterface.h" #include "ds/BloomFilter.h" namespace de { -template <RecordInterface R, ShardInterface S> +template <RecordInterface R, ShardInterface S, QueryInterface Q> class InternalLevel { - static const size_t REJECTION_TRIGGER_THRESHOLD = 1024; - - typedef decltype(R::key) K; - typedef decltype(R::value) V; - //typedef WIRS<R> S; - -private: - struct InternalLevelStructure { - InternalLevelStructure(size_t cap) - : m_cap(cap) - , m_shards(new S*[cap]{nullptr}) - , m_bfs(new BloomFilter*[cap]{nullptr}) {} - - ~InternalLevelStructure() { - for (size_t i = 0; i < m_cap; ++i) { - if (m_shards[i]) delete m_shards[i]; - if (m_bfs[i]) delete m_bfs[i]; - } - - delete[] m_shards; - delete[] m_bfs; - } - - size_t m_cap; - S** m_shards; - BloomFilter** m_bfs; - }; - public: InternalLevel(ssize_t level_no, size_t shard_cap) - : m_level_no(level_no), m_shard_cnt(0) - , m_structure(new InternalLevelStructure(shard_cap)) {} + : m_level_no(level_no), m_shard_cnt(0), m_shards(new std::vector<S>(shard_cap, nullptr)) + {} // Create a new memory level sharing the shards and repurposing it as previous level_no + 1 // WARNING: for leveling only. InternalLevel(InternalLevel* level) : m_level_no(level->m_level_no + 1), m_shard_cnt(level->m_shard_cnt) - , m_structure(level->m_structure) { - assert(m_structure->m_cap == 1 && m_shard_cnt == 1); + , m_shards(level->m_shards) { + assert(m_shard_cnt == 1 && m_shards.size() == 1); } ~InternalLevel() {} // WARNING: for leveling only. // assuming the base level is the level new level is merging into. (base_level is larger.) - static InternalLevel* merge_levels(InternalLevel* base_level, InternalLevel* new_level, const gsl_rng* rng) { + static InternalLevel* merge_levels(InternalLevel* base_level, InternalLevel* new_level) { assert(base_level->m_level_no > new_level->m_level_no || (base_level->m_level_no == 0 && new_level->m_level_no == 0)); auto res = new InternalLevel(base_level->m_level_no, 1); res->m_shard_cnt = 1; - res->m_structure->m_bfs[0] = - new BloomFilter(BF_FPR, - new_level->get_tombstone_count() + base_level->get_tombstone_count(), - BF_HASH_FUNCS, rng); S* shards[2]; - shards[0] = base_level->m_structure->m_shards[0]; - shards[1] = new_level->m_structure->m_shards[0]; + shards[0] = base_level->m_shards[0]; + shards[1] = new_level->m_shards[0]; - res->m_structure->m_shards[0] = new S(shards, 2, res->m_structure->m_bfs[0]); + res->m_shards[0] = new S(shards, 2); 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 S(buffer, m_structure->m_bfs[m_shard_cnt]); + void append_buffer(MutableBuffer<R>* buffer) { + assert(m_shard_cnt < m_shards.size()); + m_shards[m_shard_cnt] = new S(buffer); ++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 S(level->m_structure->m_shards, level->m_shard_cnt, m_structure->m_bfs[m_shard_cnt]); + void append_merged_shards(InternalLevel* level) { + assert(m_shard_cnt < m_shards.size()); + m_shards[m_shard_cnt] = new S(level->m_shards, level->m_shard_cnt); ++m_shard_cnt; } @@ -102,50 +69,36 @@ public: 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; + shards[i] = m_shards[i]; } - return new S(shards, m_shard_cnt, nullptr); + return new S(shards, m_shard_cnt); } // Append the sample range in-order..... - void get_query_states(std::vector<uint64_t>& weights, std::vector<std::pair<ShardID, S *>> &shards, std::vector<void*>& shard_states, void *query_parms) { + void get_query_states(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_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 { - S::delete_state(shard_state); - } + if (m_shards[i]) { + auto shard_state = Q::get_query_state(m_shards[i], query_parms); + shards.push_back({{m_level_no, (ssize_t) i}, m_shards[i]}); + shard_states.emplace_back(shard_state); } } } - bool bf_rejection_check(size_t shard_stop, const K& key) { - for (size_t i = 0; i < shard_stop; ++i) { - if (m_structure->m_bfs[i] && m_structure->m_bfs[i]->lookup(key)) - return true; - } - return false; - } - bool check_tombstone(size_t shard_stop, const R& rec) { if (m_shard_cnt == 0) return false; for (int i = m_shard_cnt - 1; i >= (ssize_t) shard_stop; i--) { - if (m_structure->m_shards[i] && (m_structure->m_bfs[i]->lookup(rec.key)) - && m_structure->m_shards[i]->check_tombstone(rec)) + if (m_shards[i] && m_shards[i]->check_tombstone(rec)) return true; } return false; } - bool delete_record(const K& key, const V& val) { - for (size_t i = 0; i < m_structure->m_cap; ++i) { - if (m_structure->m_shards[i] && m_structure->m_shards[i]->delete_record(key, val)) { + bool delete_record(const R &rec) { + for (size_t i = 0; i < m_shards.size(); ++i) { + if (m_shards[i] && m_shards[i]->delete_record(rec)) { return true; } } @@ -153,12 +106,8 @@ public: return false; } - const R* get_record_at(size_t shard_no, size_t idx) { - return m_structure->m_shards[shard_no]->get_record_at(idx); - } - S* get_shard(size_t idx) { - return m_structure->m_shards[idx]; + return m_shards[idx]; } size_t get_shard_count() { @@ -168,7 +117,7 @@ public: size_t get_record_cnt() { size_t cnt = 0; for (size_t i=0; i<m_shard_cnt; i++) { - cnt += m_structure->m_shards[i]->get_record_count(); + cnt += m_shards[i]->get_record_count(); } return cnt; @@ -177,27 +126,25 @@ public: size_t get_tombstone_count() { size_t res = 0; for (size_t i = 0; i < m_shard_cnt; ++i) { - res += m_structure->m_shards[i]->get_tombstone_count(); + res += m_shards[i]->get_tombstone_count(); } return res; } - size_t get_aux_memory_utilization() { + size_t get_aux_memory_usage() { size_t cnt = 0; for (size_t i=0; i<m_shard_cnt; i++) { - if (m_structure->m_bfs[i]) { - cnt += m_structure->m_bfs[i]->get_memory_utilization(); - } + cnt += m_shards[i]->get_aux_memory_usage(); } return cnt; } - size_t get_memory_utilization() { + size_t get_memory_usage() { size_t cnt = 0; for (size_t i=0; i<m_shard_cnt; i++) { - if (m_structure->m_shards[i]) { - cnt += m_structure->m_shards[i]->get_memory_utilization(); + if (m_shards[i]) { + cnt += m_shards[i]->get_memory_usage(); } } @@ -208,51 +155,22 @@ public: size_t tscnt = 0; size_t reccnt = 0; for (size_t i=0; i<m_shard_cnt; i++) { - if (m_structure->m_shards[i]) { - tscnt += m_structure->m_shards[i]->get_tombstone_count(); - reccnt += m_structure->m_shards[i]->get_record_count(); + if (m_shards[i]) { + tscnt += m_shards[i]->get_tombstone_count(); + reccnt += m_shards[i]->get_record_count(); } } return (double) tscnt / (double) (tscnt + reccnt); } - size_t get_rejection_count() { - size_t rej_cnt = 0; - for (size_t i=0; i<m_shard_cnt; i++) { - if (m_structure->m_shards[i]) { - rej_cnt += m_structure->m_shards[i]->get_rejection_count(); - } - } - - return rej_cnt; - } - - double get_rejection_rate() { - size_t rej_cnt = 0; - size_t attempt_cnt = 0; - for (size_t i=0; i<m_shard_cnt; i++) { - if (m_structure->m_shards[i]) { - attempt_cnt += m_structure->m_shards[i]->get_ts_check_count(); - rej_cnt += m_structure->m_shards[i]->get_rejection_count(); - } - } - - if (attempt_cnt == 0) return 0; - - // the rejection rate is considered 0 until we exceed an - // absolute threshold of rejections. - if (rej_cnt <= REJECTION_TRIGGER_THRESHOLD) return 0; - - return (double) rej_cnt / (double) attempt_cnt; - } - private: ssize_t m_level_no; size_t m_shard_cnt; size_t m_shard_size_cap; - std::shared_ptr<InternalLevelStructure> m_structure; + + std::shared_ptr<std::vector<S*>> m_shards; }; } diff --git a/include/framework/MutableBuffer.h b/include/framework/MutableBuffer.h index 74838b8..4b45f20 100644 --- a/include/framework/MutableBuffer.h +++ b/include/framework/MutableBuffer.h @@ -28,17 +28,17 @@ namespace de { template <RecordInterface R> class MutableBuffer { + typedef WrappedRecord<R> WRec; public: - MutableBuffer(size_t capacity, bool rej_sampling, size_t max_tombstone_cap, const gsl_rng* rng) + MutableBuffer(size_t capacity, bool rej_sampling, size_t max_tombstone_cap) : m_cap(capacity), m_tombstone_cap(max_tombstone_cap), m_reccnt(0) , m_tombstonecnt(0), m_weight(0), m_max_weight(0) { - auto len = capacity * sizeof(R); + auto len = capacity * sizeof(WRec); size_t aligned_buffersize = len + (CACHELINE_SIZE - (len % CACHELINE_SIZE)); - m_data = (R*) std::aligned_alloc(CACHELINE_SIZE, aligned_buffersize); + m_data = (WRec*) std::aligned_alloc(CACHELINE_SIZE, aligned_buffersize); m_tombstone_filter = nullptr; if (max_tombstone_cap > 0) { - assert(rng != nullptr); - m_tombstone_filter = new BloomFilter(BF_FPR, max_tombstone_cap, BF_HASH_FUNCS, rng); + m_tombstone_filter = new BloomFilter<R>(BF_FPR, max_tombstone_cap, BF_HASH_FUNCS); } } @@ -48,18 +48,23 @@ public: } template <typename R_ = R> - int append(const R &rec) { - if (rec.is_tombstone() && m_tombstonecnt + 1 > m_tombstone_cap) return 0; + int append(const R &rec, bool tombstone=false) { + if (tombstone && m_tombstonecnt + 1 > m_tombstone_cap) return 0; int32_t pos = 0; if ((pos = try_advance_tail()) == -1) return 0; - m_data[pos] = rec; + WRec wrec; + wrec.rec = rec; + + if (tombstone) wrec.set_tombstone(); + + m_data[pos] = wrec; m_data[pos].header |= (pos << 2); - if (rec.is_tombstone()) { + if (tombstone) { m_tombstonecnt.fetch_add(1); - if (m_tombstone_filter) m_tombstone_filter->insert(rec.key); + if (m_tombstone_filter) m_tombstone_filter->insert(rec); } if constexpr (WeightedRecordInterface<R_>) { @@ -86,18 +91,6 @@ public: return true; } - R* sorted_output() { - TIMER_INIT(); - TIMER_START(); - std::sort(m_data, m_data + m_reccnt.load(), memtable_record_cmp<R>); - TIMER_STOP(); - - #ifdef INSTRUMENT_MERGING - fprintf(stderr, "sort\t%ld\n", TIMER_RESULT()); - #endif - return m_data; - } - size_t get_record_count() { return m_reccnt; } @@ -117,7 +110,7 @@ public: bool delete_record(const R& rec) { auto offset = 0; while (offset < m_reccnt.load()) { - if (m_data[offset] == rec) { + if (m_data[offset].rec == rec) { m_data[offset].set_delete(); return true; } @@ -128,11 +121,11 @@ public: } bool check_tombstone(const R& rec) { - if (m_tombstone_filter && !m_tombstone_filter->lookup(rec.key)) return false; + if (m_tombstone_filter && !m_tombstone_filter->lookup(rec)) return false; auto offset = 0; while (offset < m_reccnt.load()) { - if (m_data[offset] == rec && m_data[offset].is_tombstone()) { + if (m_data[offset].rec == rec && m_data[offset].is_tombstone()) { return true; } offset++;; @@ -140,73 +133,12 @@ public: return false; } - const R* get_record_at(size_t idx) { - return m_data + idx; - } - - size_t get_memory_utilization() { + size_t get_memory_usage() { return m_cap * sizeof(R); } - size_t get_aux_memory_utilization() { - return m_tombstone_filter->get_memory_utilization(); - } - // - // NOTE: This operation samples from records strictly between the upper and - // lower bounds, not including them - template <typename R_ = R> - double get_sample_range(const decltype(R_::key) lower, const decltype(R_::key)& upper, - std::vector<R *> &records, Alias **alias, size_t *cutoff) { - std::vector<double> weights; - - *cutoff = std::atomic_load(&m_reccnt) - 1; - records.clear(); - double tot_weight = 0.0; - for (size_t i = 0; i < (*cutoff) + 1; i++) { - R *rec = m_data + i; - - if (rec->key >= lower && rec->key <= upper && !rec->is_tombstone() && !rec->is_deleted()) { - weights.push_back(rec->weight); - records.push_back(rec); - tot_weight += rec->weight; - } - } - - for (size_t i = 0; i < weights.size(); i++) { - weights[i] = weights[i] / tot_weight; - } - - *alias = new Alias(weights); - - return tot_weight; - } - - // rejection sampling - template <typename R_ = R> - const R *get_sample(const decltype(R_::key)& lower, const decltype(R_::key)& upper, gsl_rng *rng) { - size_t reccnt = m_reccnt.load(); - if (reccnt == 0) { - return nullptr; - } - - auto idx = (reccnt == 1) ? 0 : gsl_rng_uniform_int(rng, reccnt - 1); - auto rec = get_record_at(idx); - - auto test = gsl_rng_uniform(rng) * m_max_weight.load(); - - if (test > rec->weight) { - return nullptr; - } - - if (test <= rec->weight && - rec->key >= lower && - rec->key <= upper && - !rec->is_tombstone() && !rec->is_deleted()) { - - return rec; - } - - return nullptr; + size_t get_aux_memory_usage() { + return m_tombstone_filter->get_memory_usage(); } size_t get_tombstone_capacity() { @@ -226,15 +158,12 @@ private: } size_t m_cap; - //size_t m_buffersize; size_t m_tombstone_cap; - //char* m_data; R* m_data; - BloomFilter* m_tombstone_filter; + BloomFilter<R>* m_tombstone_filter; alignas(64) std::atomic<size_t> m_tombstonecnt; - //alignas(64) std::atomic<uint32_t> m_current_tail; alignas(64) std::atomic<uint32_t> m_reccnt; alignas(64) std::atomic<double> m_weight; alignas(64) std::atomic<double> m_max_weight; diff --git a/include/framework/QueryInterface.h b/include/framework/QueryInterface.h new file mode 100644 index 0000000..eafeeb0 --- /dev/null +++ b/include/framework/QueryInterface.h @@ -0,0 +1,22 @@ +/* + * include/framework/QueryInterface.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 Q> +concept QueryInterface = requires(Q q, void *p) { + {q.get_query_state(p, p)} -> std::convertible_to<void*>; + {q.get_buffer_query_state(p, p)}; + {q.query(p, p)}; + {q.buffer_query(p, p)}; + {q.merge()}; + {q.delete_query_state(p)}; +}; diff --git a/include/framework/ShardInterface.h b/include/framework/ShardInterface.h index 5d07a99..1f48a45 100644 --- a/include/framework/ShardInterface.h +++ b/include/framework/ShardInterface.h @@ -1,5 +1,5 @@ /* - * include/shard/ShardInterface.h + * include/framework/ShardInterface.h * * Copyright (C) 2023 Douglas Rumbaugh <drumbaugh@psu.edu> * @@ -14,12 +14,7 @@ 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.point_lookup(); {s.get_record_count()} -> std::convertible_to<size_t>; {s.get_tombstone_count()} -> std::convertible_to<size_t>; |