PG cleanup for moved out member

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomeObjectConan(ConanFile):
     name = "homeobject"
-    version = "2.1.17"
+    version = "2.1.18"
 
     homepage = "https://github.com/eBay/HomeObject"
     description = "Blob Store built on HomeReplication"

src/lib/homestore_backend/heap_chunk_selector.cpp

@@ -118,6 +118,54 @@ bool HeapChunkSelector::release_chunk(const pg_id_t pg_id, const chunk_num_t v_c
     return true;
 }
 
+bool HeapChunkSelector::reset_pg_chunks(pg_id_t pg_id) {
+    std::shared_lock lock_guard(m_chunk_selector_mtx);
+    auto pg_it = m_per_pg_chunks.find(pg_id);
+    if (pg_it == m_per_pg_chunks.end()) {
+        LOGWARNMOD(homeobject, "No pg found for pg_id {}", pg_id);
+        return false;
+    }
+    {
+        auto pg_chunk_collection = pg_it->second;
+        std::scoped_lock lock(pg_chunk_collection->mtx);
+        for (auto& chunk : pg_chunk_collection->m_pg_chunks) {
+            chunk->reset();
+        }
+    }
+    return true;
+}
+
+bool HeapChunkSelector::return_pg_chunks_to_dev_heap(const pg_id_t pg_id) {
+    std::unique_lock lock_guard(m_chunk_selector_mtx);
+    auto pg_it = m_per_pg_chunks.find(pg_id);
+    if (pg_it == m_per_pg_chunks.end()) {
+        LOGWARNMOD(homeobject, "No pg found for pg_id {}", pg_id);
+        return false;
+    }
+
+    auto pg_chunk_collection = pg_it->second;
+    auto pdev_id = pg_chunk_collection->m_pg_chunks[0]->get_pdev_id();
+    auto pdev_it = m_per_dev_heap.find(pdev_id);
+    RELEASE_ASSERT(pdev_it != m_per_dev_heap.end(), "pdev {} should in per dev heap", pdev_id);
+    auto pdev_heap = pdev_it->second;
+
+    {
+        std::scoped_lock lock(pdev_heap->mtx, pg_chunk_collection->mtx);
+        for (auto& chunk : pg_chunk_collection->m_pg_chunks) {
+            if (chunk->m_state == ChunkState::INUSE) {
+                chunk->m_state = ChunkState::AVAILABLE;
+            } // with shard which should be first
+            chunk->m_pg_id = std::nullopt;
+            chunk->m_v_chunk_id = std::nullopt;
+
+            pdev_heap->m_heap.emplace(chunk);
+            pdev_heap->available_blk_count += chunk->available_blks();
+        }
+    }
+    m_per_pg_chunks.erase(pg_it);
+    return true;
+}
+
 uint32_t HeapChunkSelector::get_chunk_size() const {
     const auto chunk = m_chunks.begin()->second;
     return chunk->size();

src/lib/homestore_backend/heap_chunk_selector.h

@@ -79,6 +79,20 @@ class HeapChunkSelector : public homestore::ChunkSelector {
     // It is used in two scenarios: 1. seal shard  2. create shard rollback
     bool release_chunk(const pg_id_t pg_id, const chunk_num_t v_chunk_id);
 
+    bool reset_pg_chunks(pg_id_t pg_id);
+
+    /**
+     * Releases all chunks associated with the specified pg_id.
+     *
+     * This function is used to return all chunks that are currently associated with a particular
+     * pg identified by the given pg_id. It is typically used in scenarios where
+     * all chunks associated with a pg need to be freed, such as pg move out.
+     *
+     * @param pg_id The ID of the protection group whose chunks are to be released.
+     * @return A boolean value indicating whether the operation was successful.
+     */
+    bool return_pg_chunks_to_dev_heap(pg_id_t pg_id);
+
     /**
      * select chunks for pg, chunks need to be in same pdev.
      *

src/lib/homestore_backend/hs_homeobject.hpp

@@ -30,6 +30,7 @@ static constexpr uint64_t io_align{512};
 PGError toPgError(homestore::ReplServiceError const&);
 BlobError toBlobError(homestore::ReplServiceError const&);
 ShardError toShardError(homestore::ReplServiceError const&);
+ENUM(PGState, uint8_t, ALIVE = 0, DESTROYED);
 
 class HSHomeObject : public HomeObjectImpl {
 private:
@@ -80,6 +81,7 @@ class HSHomeObject : public HomeObjectImpl {
 
     struct pg_info_superblk {
         pg_id_t id;
+        PGState state;
         uint32_t num_members;
         uint32_t num_chunks;
         peer_id_t replica_set_uuid;
@@ -109,6 +111,7 @@ class HSHomeObject : public HomeObjectImpl {
 
         pg_info_superblk& operator=(pg_info_superblk const& rhs) {
             id = rhs.id;
+            state = rhs.state;
             num_members = rhs.num_members;
             num_chunks = rhs.num_chunks;
             pg_size = rhs.pg_size;
@@ -411,6 +414,20 @@ class HSHomeObject : public HomeObjectImpl {
     void on_pg_replace_member(homestore::group_id_t group_id, const homestore::replica_member_info& member_out,
                               const homestore::replica_member_info& member_in);
 
+    /**
+     * @brief Cleans up and recycles resources for the PG identified by the given pg_id on the current node.
+     *
+     * This function is called when the replication device leaves or when a specific PG is destroyed on the current
+     * node. Note that this function does not perform Raft synchronization with other nodes.
+     *
+     * Possible scenarios for calling this function include:
+     * - A member-out node cleaning up resources for a specified PG.
+     * - During baseline rsync to clean up PG resources on the current node.
+     *
+     * @param pg_id The ID of the PG to be destroyed.
+     */
+    void pg_destroy(pg_id_t pg_id);
+
     /**
      * @brief Callback function invoked when a message is committed on a shard.
      *
@@ -488,6 +505,7 @@ class HSHomeObject : public HomeObjectImpl {
                                    size_t hash_len) const;
 
     std::shared_ptr< BlobIndexTable > recover_index_table(homestore::superblk< homestore::index_table_sb >&& sb);
+    std::optional< pg_id_t > get_pg_id_with_group_id(homestore::group_id_t group_id) const;
 
 private:
     std::shared_ptr< BlobIndexTable > create_index_table();
@@ -512,6 +530,36 @@ class HSHomeObject : public HomeObjectImpl {
     sisl::io_blob_safe& get_pad_buf(uint32_t pad_len);
 
     // void trigger_timed_events();
+
+    /**
+     * @brief Marks the PG as destroyed.
+     *
+     * Updates the internal state to indicate that the specified PG is destroyed and ensures its state is persisted.
+     *
+     * @param pg_id The ID of the PG to be marked as destroyed.
+     */
+    void mark_pg_destroyed(pg_id_t pg_id);
+
+    /**
+     * @brief Cleans up and recycles resources for shards in the PG located using a PG ID.
+     *
+     * @param pg_id The ID of the PG whose shards are to be destroyed.
+     */
+    void destroy_shards(pg_id_t pg_id);
+
+    /**
+     * @brief Resets the chunks for the given PG ID and triggers a checkpoint flush.
+     *
+     * @param pg_id The ID of the PG whose chunks are to be reset.
+     */
+    void reset_pg_chunks(pg_id_t pg_id);
+
+    /**
+     * @brief Cleans up and recycles resources for the PG located using a pg_id.
+     *
+     * @param pg_id The ID of the PG to be cleaned.
+     */
+    void cleanup_pg_resources(pg_id_t pg_id);
 };
 
 class BlobIndexServiceCallbacks : public homestore::IndexServiceCallbacks {

src/lib/homestore_backend/hs_pg_manager.cpp

@@ -250,6 +250,79 @@ void HSHomeObject::on_pg_replace_member(homestore::group_id_t group_id, const re
          boost::uuids::to_string(member_in.id));
 }
 
+std::optional< pg_id_t > HSHomeObject::get_pg_id_with_group_id(homestore::group_id_t group_id) const {
+    auto lg = std::shared_lock(_pg_lock);
+    auto iter = std::find_if(_pg_map.begin(), _pg_map.end(), [group_id](const auto& entry) {
+        return pg_repl_dev(*entry.second).group_id() == group_id;
+    });
+    if (iter != _pg_map.end()) {
+        return iter->first;
+    } else {
+        return std::nullopt;
+    }
+}
+
+void HSHomeObject::pg_destroy(pg_id_t pg_id) {
+    mark_pg_destroyed(pg_id);
+    destroy_shards(pg_id);
+    reset_pg_chunks(pg_id);
+    cleanup_pg_resources(pg_id);
+    LOGI("pg {} is destroyed", pg_id);
+}
+void HSHomeObject::mark_pg_destroyed(pg_id_t pg_id) {
+    auto lg = std::scoped_lock(_pg_lock);
+    auto iter = _pg_map.find(pg_id);
+    if (iter == _pg_map.end()) {
+        LOGW("on pg destroy with unknown pg_id {}", pg_id);
+        return;
+    }
+    auto& pg = iter->second;
+    auto hs_pg = s_cast< HS_PG* >(pg.get());
+    hs_pg->pg_sb_->state = PGState::DESTROYED;
+    hs_pg->pg_sb_.write();
+}
+
+void HSHomeObject::reset_pg_chunks(pg_id_t pg_id) {
+    bool res = chunk_selector_->reset_pg_chunks(pg_id);
+    RELEASE_ASSERT(res, "Failed to reset all chunks in pg {}", pg_id);
+    auto fut = homestore::hs()->cp_mgr().trigger_cp_flush(true /* force */);
+    auto on_complete = [&](auto success) {
+        RELEASE_ASSERT(success, "Failed to trigger CP flush");
+        LOGI("CP Flush completed");
+    };
+    on_complete(std::move(fut).get());
+}
+
+void HSHomeObject::cleanup_pg_resources(pg_id_t pg_id) {
+    auto lg = std::scoped_lock(_pg_lock);
+    auto iter = _pg_map.find(pg_id);
+    if (iter == _pg_map.end()) {
+        LOGW("on pg resource release with unknown pg_id {}", pg_id);
+        return;
+    }
+
+    // destroy index table
+    auto& pg = iter->second;
+    auto hs_pg = s_cast< HS_PG* >(pg.get());
+    if (nullptr != hs_pg->index_table_) {
+        auto uuid_str = boost::uuids::to_string(hs_pg->index_table_->uuid());
+        index_table_pg_map_.erase(uuid_str);
+        hs()->index_service().remove_index_table(hs_pg->index_table_);
+        hs_pg->index_table_->destroy();
+    }
+
+    // destroy pg super blk
+    hs_pg->pg_sb_.destroy();
+
+    // return pg chunks to dev heap
+    // which must be done after destroying pg super blk to avoid multiple pg use same chunks
+    bool res = chunk_selector_->return_pg_chunks_to_dev_heap(pg_id);
+    RELEASE_ASSERT(res, "Failed to return pg {} chunks to dev_heap", pg_id);
+
+    // erase pg in pg map
+    _pg_map.erase(iter);
+}
+
 void HSHomeObject::add_pg_to_map(unique< HS_PG > hs_pg) {
     RELEASE_ASSERT(hs_pg->pg_info_.replica_set_uuid == hs_pg->repl_dev_->group_id(),
                    "PGInfo replica set uuid mismatch with ReplDev instance for {}",
@@ -317,9 +390,14 @@ void HSHomeObject::on_pg_meta_blk_found(sisl::byte_view const& buf, void* meta_c
     // add entry in map, so that index recovery can update the PG.
     std::scoped_lock lg(index_lock_);
     auto it = index_table_pg_map_.find(uuid_str);
-    RELEASE_ASSERT(it != index_table_pg_map_.end(), "IndexTable should be recovered before PG");
-    hs_pg->index_table_ = it->second.index_table;
-    it->second.pg_id = pg_id;
+    if (it != index_table_pg_map_.end()) {
+        hs_pg->index_table_ = it->second.index_table;
+        it->second.pg_id = pg_id;
+    } else {
+        RELEASE_ASSERT(hs_pg->pg_sb_->state == PGState::DESTROYED, "IndexTable should be recovered before PG");
+        hs_pg->index_table_ = nullptr;
+        LOGI("Index table not found for destroyed pg_id={}, index_table_uuid={}", pg_id, uuid_str);
+    }
 
     add_pg_to_map(std::move(hs_pg));
 }
@@ -349,6 +427,7 @@ HSHomeObject::HS_PG::HS_PG(PGInfo info, shared< homestore::ReplDev > rdev, share
     pg_sb_.create(sizeof(pg_info_superblk) - sizeof(char) + pg_info_.members.size() * sizeof(pg_members) +
                   num_chunks * sizeof(homestore::chunk_num_t));
     pg_sb_->id = pg_info_.id;
+    pg_sb_->state = PGState::ALIVE;
     pg_sb_->num_members = pg_info_.members.size();
     pg_sb_->num_chunks = num_chunks;
     pg_sb_->pg_size = pg_info_.size;

src/lib/homestore_backend/hs_shard_manager.cpp

@@ -44,6 +44,7 @@ ShardError toShardError(ReplServiceError const& e) {
     }
 }
 
+
 uint64_t ShardManager::max_shard_size() { return Gi; }
 
 uint64_t ShardManager::max_shard_num_in_pg() { return ((uint64_t)0x01) << shard_width; }
@@ -526,6 +527,25 @@ bool HSHomeObject::release_chunk_based_on_create_shard_message(sisl::blob const&
     }
 }
 
+void HSHomeObject::destroy_shards(pg_id_t pg_id) {
+    auto lg = std::scoped_lock(_pg_lock, _shard_lock);
+    auto iter = _pg_map.find(pg_id);
+    if (iter == _pg_map.end()) {
+        LOGW("on shards destroy with unknown pg_id {}", pg_id);
+        return;
+    }
+
+    auto& pg = iter->second;
+    for (auto& shard : pg->shards_) {
+        // release open shard v_chunk
+        auto hs_shard = s_cast< HS_Shard* >(shard.get());
+        // destroy shard super blk
+        hs_shard->sb_.destroy();
+        // erase shard in shard map
+        _shard_map.erase(shard->info.id);
+    }
+}
+
 HSHomeObject::HS_Shard::HS_Shard(ShardInfo shard_info, homestore::chunk_num_t p_chunk_id,
                                  homestore::chunk_num_t v_chunk_id) :
         Shard(std::move(shard_info)), sb_(_shard_meta_name) {

src/lib/homestore_backend/replication_state_machine.cpp

@@ -193,8 +193,14 @@ void ReplicationStateMachine::on_replace_member(const homestore::replica_member_
 }
 
 void ReplicationStateMachine::on_destroy(const homestore::group_id_t& group_id) {
-    // TODO:: add the logic to handle destroy
-    LOGI("replica destroyed");
+    auto PG_ID = home_object_->get_pg_id_with_group_id(group_id);
+    if (!PG_ID.has_value()) {
+        LOGW("do not have pg mapped by group_id {}", boost::uuids::to_string(group_id));
+        return;
+    }
+    home_object_->pg_destroy(PG_ID.value());
+    LOGI("replica destroyed, cleared PG {} resources with group_id {}", PG_ID.value(),
+         boost::uuids::to_string(group_id));
 }
 
 homestore::AsyncReplResult<>

src/lib/homestore_backend/tests/homeobj_fixture.hpp

@@ -131,11 +131,11 @@ class HomeObjectFixture : public ::testing::Test {
         run_on_pg_leader(pg_id, [&]() {
             auto v_chunkID = _obj_inst->get_shard_v_chunk_id(shard_id);
             RELEASE_ASSERT(v_chunkID.has_value(), "failed to get shard v_chunk_id");
-            g_helper->set_v_chunk_id(v_chunkID.value());
+            g_helper->set_auxiliary_uint64_id(v_chunkID.value());
         });
 
         // get v_chunk_id from IPC and compare with local
-        auto leader_v_chunk_id = g_helper->get_v_chunk_id();
+        auto leader_v_chunk_id = g_helper->get_auxiliary_uint64_id();
         auto local_v_chunkID = _obj_inst->get_shard_v_chunk_id(shard_id);
         RELEASE_ASSERT(local_v_chunkID.has_value(), "failed to get shard v_chunk_id");
         RELEASE_ASSERT(leader_v_chunk_id == local_v_chunkID, "v_chunk_id supposed to be identical");
@@ -318,6 +318,22 @@ class HomeObjectFixture : public ::testing::Test {
         }
     }
 
+    void verify_pg_destroy(pg_id_t pg_id, const string& index_table_uuid_str,
+                           const std::vector< shard_id_t >& shard_id_vec) {
+        // check pg
+        ASSERT_FALSE(pg_exist(pg_id));
+        ASSERT_EQ(_obj_inst->index_table_pg_map_.find(index_table_uuid_str), _obj_inst->index_table_pg_map_.end());
+        // check shards
+        auto e = _obj_inst->shard_manager()->list_shards(pg_id).get();
+        ASSERT_EQ(e.error(), ShardError::UNKNOWN_PG);
+        for (const auto& shard_id : shard_id_vec) {
+            ASSERT_FALSE(shard_exist(shard_id));
+        }
+        // check chunk_selector
+        const auto& chunk_selector = _obj_inst->chunk_selector();
+        ASSERT_EQ(chunk_selector->m_per_pg_chunks.find(pg_id), chunk_selector->m_per_pg_chunks.end());
+    }
+
     void verify_hs_pg(HSHomeObject::HS_PG* lhs_pg, HSHomeObject::HS_PG* rhs_pg) {
         // verify index table
         EXPECT_EQ(lhs_pg->index_table_->uuid(), rhs_pg->index_table_->uuid());

src/lib/homestore_backend/tests/hs_blob_tests.cpp

@@ -14,6 +14,9 @@ TEST_F(HomeObjectFixture, BasicEquivalence) {
 }
 
 TEST_F(HomeObjectFixture, BasicPutGetDelBlobWRestart) {
+    // test recovery with pristine state firstly
+    restart();
+
     auto num_pgs = SISL_OPTIONS["num_pgs"].as< uint64_t >();
     auto num_shards_per_pg = SISL_OPTIONS["num_shards"].as< uint64_t >() / num_pgs;