Remove req from lsn map when committing. (#496)

* Remove req from lsn map when committing.

Signed-off-by: Xiaoxi Chen <[email protected]>

conanfile.py

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

src/lib/replication/repl_dev/raft_repl_dev.cpp

@@ -758,6 +758,8 @@ void RaftReplDev::handle_commit(repl_req_ptr_t rreq, bool recovery) {
 
     // Remove the request from repl_key map.
     m_repl_key_req_map.erase(rreq->rkey());
+    // Remove the request from lsn map.
+    m_state_machine->unlink_lsn_to_req(rreq->lsn());
 
     auto cur_dsn = m_next_dsn.load(std::memory_order_relaxed);
     while (cur_dsn <= rreq->dsn()) {
@@ -775,7 +777,6 @@ void RaftReplDev::handle_commit(repl_req_ptr_t rreq, bool recovery) {
         auto prev_lsn = m_commit_upto_lsn.exchange(rreq->lsn());
         RD_DBG_ASSERT_GT(rreq->lsn(), prev_lsn, "Out of order commit of lsns, it is not expected in RaftReplDev");
     }
-
     if (!rreq->is_proposer()) { rreq->clear(); }
 }
 

src/lib/replication/service/raft_repl_service.cpp

@@ -338,9 +338,11 @@ void RaftReplService::start_reaper_thread() {
             m_reaper_fiber = iomanager.iofiber_self();
 
             // Schedule the rdev garbage collector timer
+            LOGINFOMOD(replication, "Reaper Thread: scheduling GC every {} seconds", HS_DYNAMIC_CONFIG(generic.repl_dev_cleanup_interval_sec));
             m_rdev_gc_timer_hdl = iomanager.schedule_thread_timer(
                 HS_DYNAMIC_CONFIG(generic.repl_dev_cleanup_interval_sec) * 1000 * 1000 * 1000, true /* recurring */,
                 nullptr, [this](void*) {
+                    LOGINFOMOD(replication, "Reaper Thread: Doing GC");
                     gc_repl_reqs();
                     gc_repl_devs();
                 });

src/tests/test_raft_repl_dev.cpp

@@ -491,7 +491,7 @@ class RaftReplDevTest : public testing::Test {
 
                 LOGINFO("Run on worker threads to schedule append on repldev for {} Bytes.", block_size);
                 g_helper->runner().set_task([this, block_size, db]() {
-                    static std::normal_distribution<> num_blks_gen{3.0, 2.0};
+                    static std::normal_distribution<> num_blks_gen{128.0, 0.0};
                     this->generate_writes(std::abs(std::round(num_blks_gen(g_re))) * block_size, block_size, db);
                 });
                 if (wait_for_commit) { g_helper->runner().execute().get(); }