Pause state machine before pg_destroy

Author: yuwmao

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomeObjectConan(ConanFile):
     name = "homeobject"
-    version = "2.6.2"
+    version = "2.6.3"
 
     homepage = "https://github.com/eBay/HomeObject"
     description = "Blob Store built on HomeReplication"
@@ -49,7 +49,7 @@ def build_requirements(self):
 
     def requirements(self):
         self.requires("sisl/[^12.2]@oss/master", transitive_headers=True)
-        self.requires("homestore/[~6.18.0]@oss/master")
+        self.requires("homestore/[~6.18.4]@oss/master")
         self.requires("iomgr/[^11.3]@oss/master")
         self.requires("lz4/1.9.4", override=True)
         self.requires("openssl/3.3.1", override=True)

src/lib/homestore_backend/hs_homeobject.hpp

@@ -764,7 +764,11 @@ class HSHomeObject : public HomeObjectImpl {
      *
      * @param pg_id The ID of the PG to be destroyed.
      */
-    void pg_destroy(pg_id_t pg_id);
+    bool pg_destroy(pg_id_t pg_id, bool need_to_pause_pg_state_machine = false);
+
+    bool pause_pg_state_machine(pg_id_t pg_id);
+
+    bool resume_pg_state_machine(pg_id_t pg_id);
 
     /**
      * @brief Get HS_PG object from given pg_id.

src/lib/homestore_backend/hs_pg_manager.cpp

@@ -434,7 +434,11 @@ std::optional< pg_id_t > HSHomeObject::get_pg_id_with_group_id(group_id_t group_
     }
 }
 
-void HSHomeObject::pg_destroy(pg_id_t pg_id) {
+bool HSHomeObject::pg_destroy(pg_id_t pg_id, bool need_to_pause_pg_state_machine) {
+    if (need_to_pause_pg_state_machine && !pause_pg_state_machine(pg_id)) {
+        LOGI("Failed to pause pg state machine, pg_id={}", pg_id);
+        return false;
+    }
     LOGI("Destroying pg={}", pg_id);
     mark_pg_destroyed(pg_id);
     destroy_shards(pg_id);
@@ -448,6 +452,35 @@ void HSHomeObject::pg_destroy(pg_id_t pg_id) {
     RELEASE_ASSERT(res, "Failed to return pg={} chunks to dev_heap", pg_id);
 
     LOGI("pg={} is destroyed", pg_id);
+    return true;
+}
+
+bool HSHomeObject::pause_pg_state_machine(pg_id_t pg_id) {
+    LOGI("Pause pg state machine, pg={}", pg_id);
+    auto hs_pg = const_cast< HS_PG* >(_get_hs_pg_unlocked(pg_id));
+    auto repl_dev = hs_pg ? hs_pg->repl_dev_ : nullptr;
+    auto max_wait_time_ms = 2000;
+    for (auto i = 0; i < 2000 / 500; i++) {
+        hs_pg->repl_dev_->pause_state_machine(500 /* ms */);
+        if (repl_dev->is_state_machine_paused()) {
+            LOGI("pg={} state machine is paused", pg_id);
+            return true;
+        }
+    }
+    LOGE("Failed to pause pg={} state machine after {} ms", pg_id, max_wait_time_ms);
+    return false;
+}
+
+bool HSHomeObject::resume_pg_state_machine(pg_id_t pg_id) {
+    LOGI("Resuming pg state machine, pg={}", pg_id);
+    auto hs_pg = const_cast< HS_PG* >(_get_hs_pg_unlocked(pg_id));
+    auto repl_dev = hs_pg ? hs_pg->repl_dev_ : nullptr;
+    while (repl_dev->is_state_machine_paused()) {
+        hs_pg->repl_dev_->resume_state_machine();
+        std::this_thread::sleep_for(std::chrono::milliseconds(500));
+    }
+    LOGE("Resumed pg state machine, pg=", pg_id);
+    return true;
 }
 
 void HSHomeObject::mark_pg_destroyed(pg_id_t pg_id) {

src/lib/homestore_backend/replication_state_machine.cpp

@@ -311,6 +311,8 @@ bool ReplicationStateMachine::apply_snapshot(std::shared_ptr< homestore::snapsho
         LOGI("Simulating apply snapshot with delay, delay={}", delay.get());
         std::this_thread::sleep_for(std::chrono::milliseconds(delay.get()));
     }
+    // It's an inifinite loop until the state machine resumes.
+    home_object_->resume_pg_state_machine(m_snp_rcv_handler->get_context_pg_id());
 #endif
     m_snp_rcv_handler->destroy_context_and_metrics();
 
@@ -486,7 +488,11 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna
         // If PG already exists, clean the stale pg resources. Let's resync on a pristine base
         if (home_object_->pg_exists(pg_data->pg_id())) {
             LOGI("pg already exists, clean pg resources before snapshot, pg={} {}", pg_data->pg_id(), log_suffix);
-            home_object_->pg_destroy(pg_data->pg_id());
+            // Need to pause state machine before destroying the PG, if fail, let raft retry.
+            if (!home_object_->pg_destroy(pg_data->pg_id(), true /* pause state machine */)) {
+                LOGE("failed to destroy existing pg, let raft retry, pg={} {}", pg_data->pg_id(), log_suffix);
+                return;
+            }
         }
         LOGI("reset context from lsn={} to lsn={}", m_snp_rcv_handler->get_context_lsn(), context->get_lsn());
         m_snp_rcv_handler->reset_context_and_metrics(context->get_lsn(), pg_data->pg_id());