Add event callbacks into listener for upper layer (#657)

some events need to be handled by upper layer. This PR add three event:
1 fetch_data: upper layer can decide which data to be returned

2 no_space_left: this error should be handled by upper layer if
  necessary

3 on_log_replay_done: after log replay is done and before joining raft
  group, upper layer might do something

Author: JacksonYao287

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomestoreConan(ConanFile):
     name = "homestore"
-    version = "6.6.25"
+    version = "6.7.0"
 
     homepage = "https://github.com/eBay/Homestore"
     description = "HomeStore Storage Engine"

src/include/homestore/replication/repl_dev.h

@@ -11,6 +11,7 @@
 #include <sisl/grpc/generic_service.hpp>
 #include <sisl/grpc/rpc_client.hpp>
 #include <homestore/replication/repl_decls.h>
+#include <homestore/blkdata_service.hpp>
 #include <libnuraft/snapshot.hxx>
 
 namespace nuraft {
@@ -367,6 +368,25 @@ class ReplDevListener {
     /// @brief Free up user-defined context inside the snapshot_obj that is allocated during read_snapshot_obj.
     virtual void free_user_snp_ctx(void*& user_snp_ctx) = 0;
 
+    /// @brief ask upper layer to decide which data should be returned.
+    // @param header - header of the log entry.
+    // @param blkid - original blkid of the log entry
+    // @param sgs - sgs to be filled with data
+    // @param lsn - lsn of the log entry
+    virtual folly::Future< std::error_code > on_fetch_data(const int64_t lsn, const sisl::blob& header,
+                                                           const MultiBlkId& blkid, sisl::sg_list& sgs) {
+        // default implementation is reading by blkid directly
+        return data_service().async_read(blkid, sgs, sgs.size);
+    }
+
+    /// @brief ask upper layer to handle no_space_left event
+    virtual folly::Future< std::error_code > on_no_space_left(uint32_t pdev_id, chunk_num_t chunk_id) {
+        return folly::makeFuture< std::error_code >(std::error_code{});
+    }
+
+    /// @brief when restart, after all the logs are replayed and before joining raft group, notify the upper layer
+    virtual void on_log_replay_done(const group_id_t& group_id){};
+
 private:
     std::weak_ptr< ReplDev > m_repl_dev;
 };
@@ -449,7 +469,7 @@ class ReplDev {
     /// @brief Clean up resources on this repl dev.
     virtual void purge() = 0;
 
-    virtual std::shared_ptr<snapshot_context> deserialize_snapshot_context(sisl::io_blob_safe &snp_ctx) = 0;
+    virtual std::shared_ptr< snapshot_context > deserialize_snapshot_context(sisl::io_blob_safe& snp_ctx) = 0;
 
     virtual void attach_listener(shared< ReplDevListener > listener) { m_listener = std::move(listener); }
 
@@ -460,6 +480,8 @@ class ReplDev {
         }
     }
 
+    virtual shared< ReplDevListener > get_listener() { return m_listener; }
+
     // we have no shutdown for repl_dev, since shutdown repl_dev is done by repl_service
     void stop() {
         start_stopping();

src/lib/replication/repl_dev/raft_repl_dev.cpp

@@ -335,7 +335,15 @@ void RaftReplDev::async_alloc_write(sisl::blob const& header, sisl::blob const&
             handle_error(rreq, ReplServiceError::DATA_DUPLICATED);
             return;
         }
+
+#ifdef _PRERELEASE
+        if (iomgr_flip::instance()->test_flip("disable_leader_push_data")) {
+            RD_LOGD("Simulating push data failure, so that all the follower will have to fetch data");
+        } else
+            push_data_to_all_followers(rreq, data);
+#else
         push_data_to_all_followers(rreq, data);
+#endif
 
         COUNTER_INCREMENT(m_metrics, total_write_cnt, 1);
         COUNTER_INCREMENT(m_metrics, outstanding_data_write_cnt, 1);
@@ -794,30 +802,8 @@ void RaftReplDev::on_fetch_data_received(intrusive< sisl::GenericRpcData >& rpc_
         auto const& lsn = req->lsn();
         auto const& originator = req->blkid_originator();
         auto const& remote_blkid = req->remote_blkid();
-
-        // Edit this check if in the future we want to fetch from non-originator;
-        if (originator != server_id()) {
-            auto const error_msg =
-                fmt::format("Did not expect to receive fetch data from "
-                            "remote when I am not the originator of this request, originator={}, my_server_id={}",
-                            originator, server_id());
-            RD_LOGW("{}", error_msg);
-            auto status = ::grpc::Status(::grpc::INVALID_ARGUMENT, error_msg);
-            rpc_data->set_status(status);
-            rpc_data->send_response();
-            return;
-        }
-
-        // fetch data based on the remote_blkid
-        // We are the originator of the blkid, read data locally;
         MultiBlkId local_blkid;
-
-        // convert remote_blkid serialized data to local blkid
         local_blkid.deserialize(sisl::blob{remote_blkid->Data(), remote_blkid->size()}, true /* copy */);
-
-        RD_LOGD("Data Channel: FetchData received: dsn={} lsn={} my_blkid={}", req->dsn(), lsn,
-                local_blkid.to_string());
-
         // prepare the sgs data buffer to read into;
         auto const total_size = local_blkid.blk_count() * get_blk_size();
         sisl::sg_list sgs;
@@ -827,7 +813,18 @@ void RaftReplDev::on_fetch_data_received(intrusive< sisl::GenericRpcData >& rpc_
 
         // accumulate the sgs for later use (send back to the requester));
         sgs_vec.push_back(sgs);
-        futs.emplace_back(async_read(local_blkid, sgs, total_size));
+
+        if (originator != server_id()) {
+            RD_LOGD("non-originator FetchData received:  dsn={} lsn={} originator={}, my_server_id={}", req->dsn(), lsn,
+                    originator, server_id());
+        } else {
+            RD_LOGD("Data Channel: FetchData received:  dsn={} lsn={}", req->dsn(), lsn);
+        }
+
+        auto const& header = req->user_header();
+        sisl::blob user_header = sisl::blob{header->Data(), header->size()};
+        RD_LOGD("Data Channel: FetchData handled, my_blkid={}", local_blkid.to_string());
+        futs.emplace_back(std::move(m_listener->on_fetch_data(lsn, user_header, local_blkid, sgs)));
     }
 
     folly::collectAllUnsafe(futs).thenValue(
@@ -1238,10 +1235,10 @@ void RaftReplDev::save_config(const nuraft::cluster_config& config) {
 
 void RaftReplDev::save_state(const nuraft::srv_state& state) {
     std::unique_lock lg{m_config_mtx};
-    (*m_raft_config_sb)["state"] = nlohmann::json{
-        {"term", state.get_term()}, {"voted_for", state.get_voted_for()},
-        {"election_timer_allowed", state.is_election_timer_allowed()}, {"catching_up", state.is_catching_up()}
-    };
+    (*m_raft_config_sb)["state"] = nlohmann::json{{"term", state.get_term()},
+                                                  {"voted_for", state.get_voted_for()},
+                                                  {"election_timer_allowed", state.is_election_timer_allowed()},
+                                                  {"catching_up", state.is_catching_up()}};
     m_raft_config_sb.write();
     RD_LOGI("Saved state {}", (*m_raft_config_sb)["state"].dump());
 }
@@ -1251,17 +1248,16 @@ nuraft::ptr< nuraft::srv_state > RaftReplDev::read_state() {
     auto& js = *m_raft_config_sb;
     auto state = nuraft::cs_new< nuraft::srv_state >();
     if (js["state"].empty()) {
-        js["state"] = nlohmann::json{
-            {"term", state->get_term()}, {"voted_for", state->get_voted_for()},
-            {"election_timer_allowed", state->is_election_timer_allowed()},
-            {"catching_up", state->is_catching_up()}
-        };
+        js["state"] = nlohmann::json{{"term", state->get_term()},
+                                     {"voted_for", state->get_voted_for()},
+                                     {"election_timer_allowed", state->is_election_timer_allowed()},
+                                     {"catching_up", state->is_catching_up()}};
     } else {
         try {
             state->set_term(uint64_cast(js["state"]["term"]));
             state->set_voted_for(static_cast< int >(js["state"]["voted_for"]));
-            state->allow_election_timer(static_cast<bool>(js["state"]["election_timer_allowed"]));
-            state->set_catching_up(static_cast<bool>(js["state"]["catching_up"]));
+            state->allow_election_timer(static_cast< bool >(js["state"]["election_timer_allowed"]));
+            state->set_catching_up(static_cast< bool >(js["state"]["catching_up"]));
         } catch (std::out_of_range const&) {
             LOGWARN("State data was not in the expected format [group_id={}]!", m_group_id)
         }

src/lib/replication/repl_dev/raft_repl_dev.h

@@ -115,22 +115,22 @@ struct ReplDevCPContext {
 
 class nuraft_snapshot_context : public snapshot_context {
 public:
-    nuraft_snapshot_context(nuraft::snapshot &snp) : snapshot_context(snp.get_last_log_idx()) {
+    nuraft_snapshot_context(nuraft::snapshot& snp) : snapshot_context(snp.get_last_log_idx()) {
         auto snp_buf = snp.serialize();
         snapshot_ = nuraft::snapshot::deserialize(*snp_buf);
     }
 
-    nuraft_snapshot_context(sisl::io_blob_safe const &snp_ctx) : snapshot_context(0) { deserialize(snp_ctx); }
+    nuraft_snapshot_context(sisl::io_blob_safe const& snp_ctx) : snapshot_context(0) { deserialize(snp_ctx); }
 
     sisl::io_blob_safe serialize() override {
         // Dump the context from nuraft buffer to the io blob.
         auto snp_buf = snapshot_->serialize();
-        sisl::io_blob_safe blob{s_cast<size_t>(snp_buf->size())};
+        sisl::io_blob_safe blob{s_cast< size_t >(snp_buf->size())};
         std::memcpy(blob.bytes(), snp_buf->data_begin(), snp_buf->size());
         return blob;
     }
 
-    void deserialize(const sisl::io_blob_safe &snp_ctx) {
+    void deserialize(const sisl::io_blob_safe& snp_ctx) {
         // Load the context from the io blob to nuraft buffer.
         auto snp_buf = nuraft::buffer::alloc(snp_ctx.size());
         snp_buf->put_raw(snp_ctx.cbytes(), snp_ctx.size());
@@ -139,10 +139,10 @@ class nuraft_snapshot_context : public snapshot_context {
         lsn_ = snapshot_->get_last_log_idx();
     }
 
-    nuraft::ptr<nuraft::snapshot> nuraft_snapshot() { return snapshot_; }
+    nuraft::ptr< nuraft::snapshot > nuraft_snapshot() { return snapshot_; }
 
 private:
-    nuraft::ptr<nuraft::snapshot> snapshot_;
+    nuraft::ptr< nuraft::snapshot > snapshot_;
 };
 
 class RaftReplDev : public ReplDev,
@@ -236,12 +236,12 @@ class RaftReplDev : public ReplDev,
         m_data_journal->purge_all_logs();
     }
 
-    std::shared_ptr<snapshot_context> deserialize_snapshot_context(sisl::io_blob_safe &snp_ctx) override {
-        return std::make_shared<nuraft_snapshot_context>(snp_ctx);
+    std::shared_ptr< snapshot_context > deserialize_snapshot_context(sisl::io_blob_safe& snp_ctx) override {
+        return std::make_shared< nuraft_snapshot_context >(snp_ctx);
     }
 
     //////////////// Accessor/shortcut methods ///////////////////////
-    nuraft_mesg::repl_service_ctx *group_msg_service();
+    nuraft_mesg::repl_service_ctx* group_msg_service();
 
     nuraft::raft_server* raft_server();
     RaftReplDevMetrics& metrics() { return m_metrics; }

src/lib/replication/service/raft_repl_service.cpp

@@ -156,16 +156,21 @@ void RaftReplService::start() {
     LOGINFO("Starting DataService");
     hs()->data_service().start();
 
-    // Step 6: Iterate all the repl dev and ask each one of the join the raft group.
-    for (auto it = m_rd_map.begin(); it != m_rd_map.end();) {
-        auto rdev = std::dynamic_pointer_cast< RaftReplDev >(it->second);
-        rdev->wait_for_logstore_ready();
-        if (!rdev->join_group()) {
-            HS_REL_ASSERT(false, "FAILED TO JOIN GROUP, PANIC HERE");
-            it = m_rd_map.erase(it);
-        } else {
-            ++it;
-        }
+    // Step 6: Iterate all the repl devs and ask each one of them to join the raft group concurrently.
+    std::vector< std::future< bool > > join_group_futures;
+    for (const auto& [_, repl_dev] : m_rd_map) {
+        join_group_futures.emplace_back(std::async(std::launch::async, [&repl_dev]() {
+            auto rdev = std::dynamic_pointer_cast< RaftReplDev >(repl_dev);
+            rdev->wait_for_logstore_ready();
+
+            // upper layer can register a callback to be notified when log replay is done.
+            if (auto listener = rdev->get_listener(); listener) listener->on_log_replay_done(rdev->group_id());
+            return rdev->join_group();
+        }));
+    }
+
+    for (auto& future : join_group_futures) {
+        if (!future.get()) HS_REL_ASSERT(false, "FAILED TO JOIN GROUP, PANIC HERE");
     }
 
     // Step 7: Register to CPManager to ensure we can flush the superblk.

src/tests/test_raft_repl_dev.cpp

@@ -33,8 +33,8 @@ TEST_F(RaftReplDevTest, Write_Duplicated_Data) {
         stored_key = dbs_[0]->inmem_db_.cbegin()->first;
         ASSERT_EQ(id, stored_key.id_);
     } else {
-        LOGINFO("I am not leader, leader_uuid={} my_uuid={}, do nothing",
-                boost::uuids::to_string(leader_uuid), boost::uuids::to_string(g_helper->my_replica_id()));
+        LOGINFO("I am not leader, leader_uuid={} my_uuid={}, do nothing", boost::uuids::to_string(leader_uuid),
+                boost::uuids::to_string(g_helper->my_replica_id()));
     }
     wait_for_commits(total_writes);
 
@@ -45,12 +45,12 @@ TEST_F(RaftReplDevTest, Write_Duplicated_Data) {
     if duplication found in leader proposal, reject it;
     if duplication found in the followers, skip it.
     */
-    //1. write the same data again on leader, should fail
+    // 1. write the same data again on leader, should fail
     if (leader_uuid == g_helper->my_replica_id()) {
         auto err = this->write_with_id(id, true /* wait_for_commit */);
         ASSERT_EQ(ReplServiceError::DATA_DUPLICATED, err);
 
-        //2. delete it from the db to simulate duplication in followers(skip the duplication check in leader side)
+        // 2. delete it from the db to simulate duplication in followers(skip the duplication check in leader side)
         dbs_[0]->inmem_db_.erase(stored_key);
         LOGINFO("data with id={} has been deleted from db", id);
         err = this->write_with_id(id, true /* wait_for_commit */);
@@ -109,6 +109,24 @@ TEST_F(RaftReplDevTest, Follower_Fetch_OnActive_ReplicaGroup) {
 
     g_helper->sync_for_cleanup_start();
 }
+
+TEST_F(RaftReplDevTest, Write_With_Diabled_Leader_Push_Data) {
+    g_helper->set_basic_flip("disable_leader_push_data");
+    LOGINFO("Homestore replica={} setup completed, all the push_data from leader are disabled",
+            g_helper->replica_num());
+    LOGINFO("Homestore replica={} setup completed", g_helper->replica_num());
+    g_helper->sync_for_test_start();
+
+    this->write_on_leader(100, true /* wait_for_commit */);
+
+    g_helper->sync_for_verify_start();
+
+    LOGINFO("Validate all data written so far by reading them");
+    this->validate_data();
+
+    g_helper->sync_for_cleanup_start();
+}
+
 #endif
 
 // do some io before restart;