From d28f2cfcd4249fc7d984762a326e3f2d6dcba7dc Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Thu, 13 Feb 2025 18:13:33 -0500 Subject: progress towards resolving asynch structure merges --- include/framework/structure/ExtensionStructure.h | 132 +++++++++-------------- include/framework/structure/InternalLevel.h | 6 +- 2 files changed, 56 insertions(+), 82 deletions(-) (limited to 'include/framework/structure') diff --git a/include/framework/structure/ExtensionStructure.h b/include/framework/structure/ExtensionStructure.h index fa713af..521e68b 100644 --- a/include/framework/structure/ExtensionStructure.h +++ b/include/framework/structure/ExtensionStructure.h @@ -55,6 +55,8 @@ public: new_struct->m_levels.push_back(m_levels[i]->clone()); } + new_struct->m_deleted_shards = m_deleted_shards; + return new_struct; } @@ -159,65 +161,29 @@ public: return cnt; } + /* * Perform the reconstruction described by task. If the resulting * reconstruction grows the structure (i.e., adds a level), returns * true. Otherwise, returns false. */ - inline bool perform_reconstruction(ReconstructionTask task, size_t version=0) { - /* perform the reconstruction itself */ + inline reconstruction_results perform_reconstruction(ReconstructionTask task) const { + reconstruction_results result; + result.target_level = task.target; + std::vector shards; for (ShardID shid : task.sources) { - assert(shid.level_idx <= (level_index) m_levels.size()); + assert(shid.level_idx < (level_index) m_levels.size()); assert(shid.shard_idx >= -1); - if (shid.level_idx == (level_index) m_levels.size()) { - continue; - } - - /* if unspecified, push all shards into the vector */ - if (shid.shard_idx == all_shards_idx) { - for (size_t i = 0; i < m_levels[shid.level_idx]->get_shard_count(); - i++) { - if (m_levels[shid.level_idx]->get_shard(i)) { - shards.push_back(m_levels[shid.level_idx]->get_shard(i)); - } - } - } else { - shards.push_back(m_levels[shid.level_idx]->get_shard(shid.shard_idx)); - } + auto raw_shard_ptr = m_levels[shid.level_idx]->get_shard(shid.shard_idx); + shards.push_back(raw_shard_ptr); + result.source_shards.emplace_back(shid.level_idx, raw_shard_ptr); } - auto new_shard = new ShardType(shards); + result.new_shard = std::make_shared(shards); - /* - * Remove all of the shards processed by the operation - */ - for (ShardID shid : task.sources) { - if (shid.level_idx == (level_index) m_levels.size()) { - continue; - } else if (shid.shard_idx == all_shards_idx) { - m_levels[shid.level_idx]->truncate(); - } else if (shid != buffer_shid) { - m_levels[shid.level_idx]->delete_shard(shid.shard_idx); - } - } - - // fprintf(stderr, "Target: %ld\tLevels:%ld\n", task.target, m_levels.size()); - - /* - * Append the new shard to the target level - */ - if (task.target < (level_index)m_levels.size()) { - m_levels[task.target]->append(std::shared_ptr(new_shard), version); - return false; - // fprintf(stderr, "append (no growth)\n"); - } else { /* grow the structure if needed */ - m_levels.push_back(std::make_shared>(task.target)); - m_levels[task.target]->append(std::shared_ptr(new_shard), version); - return true; - // fprintf(stderr, "grow and append\n"); - } + return result; } std::vector @@ -237,8 +203,36 @@ public: return m_levels[0]->get_shard_count(); } - void append_l0(std::shared_ptr shard, size_t version) { - m_levels[0]->append(shard, version); + bool append_shard(std::shared_ptr shard, size_t version, size_t level) { + assert(level <= m_levels.size()); + auto rc = false; + + if (level == m_levels.size()) { + /* grow the structure */ + m_levels.push_back(std::make_shared>(level)); + rc = true; + } + + m_levels[level]->append(shard, version); + + return rc; + } + + void delete_shards(std::vector> shards) { + for (size_t i=0; iget_shard_count(); j++) { + if (m_levels[shards[i].first]->get_shard_ptr(j).first.get() == shards[i].second) { + shard_idx = j; + break; + } + } + + if (shard_idx != -1) { + m_levels[shards[i].first]->delete_shard(shard_idx); + } + } } LevelVector const &get_level_vector() const { return m_levels; } @@ -269,13 +263,17 @@ public: return ts_prop <= (long double) max_delete_prop; } - void print_structure() const { + void print_structure(bool debug=false) const { for (size_t i=0; iget_shard_count(); j++) { - fprintf(stdout, "(%ld: %ld) ", j, m_levels[i]->get_shard(j)->get_record_count()); + fprintf(stdout, "(%ld, %ld, %p: %ld) ", j, m_levels[i]->get_shard_ptr(j).second, m_levels[i]->get_shard_ptr(j).first.get(), m_levels[i]->get_shard(j)->get_record_count()); } } else { fprintf(stdout, "[Empty]"); @@ -285,37 +283,9 @@ public: } } - - void merge_structure(const ExtensionStructure* old_structure, size_t version_id = 0) { - assert(version_id > 0); - - for (size_t i=0; im_levels.size(); i++) { - if (m_levels.size() <= i) { - m_levels.push_back(old_structure->m_levels[i]); - } else { - for (size_t j=0; jm_levels[i]->get_shard_count(); j++) { - if (old_structure->m_levels[i]->get_shard_version(j) > version_id) { - m_levels[i]->append(old_structure->m_levels[i]->get_shard_ptr(j)); - } - } - } - } - } - - void update_shard_version(size_t version) { - assert(version != 0); - - for (size_t i=0; iget_shard_count(); j++) { - if (m_levels[i]->get_shard_version(j) == 0) { - m_levels[i]->set_shard_version(j, version); - } - } - } - } - private: LevelVector m_levels; + std::vector m_deleted_shards; }; } // namespace de diff --git a/include/framework/structure/InternalLevel.h b/include/framework/structure/InternalLevel.h index 37b2b40..7e8e87d 100644 --- a/include/framework/structure/InternalLevel.h +++ b/include/framework/structure/InternalLevel.h @@ -198,8 +198,12 @@ public: void truncate() { m_shards.erase(m_shards.begin(), m_shards.end()); } - void delete_shard(shard_index shard) { + void delete_shard(shard_index shard, bool log_delete=true) { + size_t before = m_shards.size(); + fprintf(stderr, "[D]\tReconstruction deleting shard %ld %p\n", shard, m_shards[shard].first.get()); m_shards.erase(m_shards.begin() + shard); + size_t after = m_shards.size(); + assert( before > after); } void append(std::shared_ptr shard, size_t version=0) { -- cgit v1.2.3