Dev/v8.x

conanfile.py

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

src/include/homestore/blkdata_service.hpp

@@ -187,9 +187,14 @@ class blk_data_service {
      * @brief Commits the block with the given multi_blk_id.
      *
      * @param bid The multi_blk_id of the block to commit.
+     * @param recommit If true, skip the allocation check and commit the blk unconditionally.
+     *        Use this when re-committing a blk that is known to be live but whose allocator state
+     *        may have been lost (e.g. after a crash where the blk exists in the index but the
+     *        allocator watermarks were not yet persisted). Safe to pass when the allocator already
+     *        tracks the blk, as the underlying operations are monotonically advancing.
      * @return success, or the error_condition describing the failure.
      */
-    status commit_blk(multi_blk_id const& bid);
+    status commit_blk(multi_blk_id const& bid, bool recommit = false);
 
     /**
      * @brief Allocates a contiguous block of disk space of the given size. Use this when the consumer expects the blks

src/lib/blkdata_svc/blkdata_service.cpp

@@ -288,17 +288,17 @@ result< std::vector< blk_id > > blk_data_service::alloc_blk_list(uint32_t size,
     return out_blkids;
 }
 
-status blk_data_service::commit_blk(multi_blk_id const& blkid) {
+status blk_data_service::commit_blk(multi_blk_id const& blkid, bool recommit) {
     if (is_stopping()) return std::unexpected(std::make_error_condition(std::errc::operation_canceled));
     incr_pending_request_num();
 
     BlkAllocStatus ret = BlkAllocStatus::SUCCESS;
     if (blkid.num_pieces() == 1) {
-        ret = m_vdev->commit_blk(blkid); // shortcut for the most common case
+        ret = m_vdev->commit_blk(blkid, recommit); // shortcut for the most common case
     } else {
         auto it = blkid.iterate();
         while (auto const bid = it.next()) {
-            ret = m_vdev->commit_blk(*bid);
+            ret = m_vdev->commit_blk(*bid, recommit);
             if (ret != BlkAllocStatus::SUCCESS) { break; }
         }
     }

src/lib/device/virtual_dev.cpp

@@ -173,7 +173,7 @@ bool VirtualDev::is_blk_alloced(blk_id const& blkid) const {
     return m_dmgr.get_chunk(blkid.chunk_num())->blk_allocator()->is_blk_alloced(blkid, true /* lock */);
 }
 
-BlkAllocStatus VirtualDev::commit_blk(blk_id const& blkid) {
+BlkAllocStatus VirtualDev::commit_blk(blk_id const& blkid, bool recommit) {
     Chunk* chunk = m_dmgr.get_chunk_mutable(blkid.chunk_num());
     // if we start with missing drive, we will have no chunk for this blkid;
     if (!chunk) {
@@ -187,12 +187,14 @@ BlkAllocStatus VirtualDev::commit_blk(blk_id const& blkid) {
     }
 
     auto const recovering = homestore::hs()->is_initializing();
-    if (!recovering) {
+    if (recovering || recommit) {
+        // Also advance the in-memory watermark so the allocator considers this blk allocated;
+        // required both during normal recovery and on an explicit recommit by the caller.
+        chunk->blk_allocator_mutable()->reserve_on_cache(blkid);
+    } else {
         // in non-recovery mode, if a blk is committed without allocating, it will cause data corruption
         HS_REL_ASSERT(is_blk_alloced(blkid), "commiting blkid {} is not allocated in non-recovery mode",
                       blkid.to_string());
-    } else {
-        chunk->blk_allocator_mutable()->reserve_on_cache(blkid);
     }
     return chunk->blk_allocator_mutable()->reserve_on_disk(blkid);
 }

src/lib/device/virtual_dev.hpp

@@ -172,11 +172,13 @@ class VirtualDev {
     /// @brief Commits the blkid in on-disk version of the blk allocator. The blkid is assumed to be allocated using
     /// alloc_blk or alloc_contiguous_blk method earlier (either after reboot or prior to reboot). It is not required
     /// to call this method if alloc_blk is called and system is not restarted. Typical use case of this method is
-    /// during recovery where alloc_blk is called but before it was checkpointed, it crashed and we are trying to
-    /// recover Please note that even calling this method is not guaranteed to persisted until checkpoint is taken.
+    /// during recovery where alloc_blk is called, but before it was checkpointed, it crashed, and we are trying to
+    /// recover. Please note that even calling this method is not guaranteed to be persisted until checkpoint is taken.
     /// @param blkid blk_id to commit explicitly.
+    /// @param recommit If true, skip the allocation check and commit unconditionally.
+    ///        See blk_data_service::commit_blk for details.
     /// @return Allocation Status
-    virtual BlkAllocStatus commit_blk(blk_id const& blkid);
+    virtual BlkAllocStatus commit_blk(blk_id const& blkid, bool recommit = false);
 
     virtual void free_blk(blk_id const& b, VDevCPContext* vctx = nullptr, bool free_now = false);
 

src/lib/replication/service/raft_repl_service.cpp

@@ -639,17 +639,6 @@ void RaftReplService::start_repl_service_timers() {
                     gc_repl_devs();
                 });
 
-            // Check for queued fetches at the minimum every second
-            uint64_t interval_ns =
-                std::min(HS_DYNAMIC_CONFIG(consensus.wait_data_write_timer_ms) * 1000 * 1000, 1ul * 1000 * 1000 * 1000);
-            m_rdev_fetch_timer_hdl = iomanager.schedule_thread_timer(
-                interval_ns, true /* recurring */, nullptr, [this](void*, uint64_t exp_count) {
-                    if (exp_count > 1) {
-                        LOGINFOMOD(replication, "fetch pending data timer expired {} times, running once", exp_count);
-                    }
-                    fetch_pending_data();
-                });
-
             // Flush durable commit lsns to superblock
             // FIXUP: what is the best value for flush_durable_commit_interval_ms?
             m_flush_durable_commit_timer_hdl = iomanager.schedule_thread_timer(
@@ -674,13 +663,38 @@ void RaftReplService::start_repl_service_timers() {
         }
     });
     latch.wait();
+
+    // Dedicated reactor for fetch_pending_data so that a blocking flush_durable_commit_lsn() or
+    // GC on the reaper fiber cannot delay queued fetch batches past consensus.data_receive_timeout_ms
+    // (default 10s), which would otherwise cause "Data fetch timeout" assertion / TIMEOUT errors.
+    std::latch fetcher_latch{1};
+    iomanager.create_reactor("raft_repl_fetcher", iomgr::INTERRUPT_LOOP, [this, &fetcher_latch](bool is_started) {
+        if (is_started) {
+            m_fetcher_reactor = iomanager.this_reactor();
+            // Check for queued fetches at the minimum every second
+            uint64_t interval_ns =
+                std::min(HS_DYNAMIC_CONFIG(consensus.wait_data_write_timer_ms) * 1000 * 1000, 1ul * 1000 * 1000 * 1000);
+            m_rdev_fetch_timer_hdl = iomanager.schedule_thread_timer(
+                interval_ns, true /* recurring */, nullptr, [this](void*, uint64_t exp_count) {
+                    if (exp_count > 1) {
+                        LOGINFOMOD(replication, "fetch pending data timer expired {} times, running once", exp_count);
+                    }
+                    fetch_pending_data();
+                });
+            fetcher_latch.count_down();
+        }
+    });
+    fetcher_latch.wait();
 }
 
 void RaftReplService::stop_repl_service_timers() {
+    iomanager.run_on_wait(m_fetcher_reactor, [this]() {
+        LOGINFOMOD(replication, "Fetcher Thread: Stopping timer");
+        iomanager.cancel_timer(m_rdev_fetch_timer_hdl, true);
+    });
     iomanager.run_on_wait(m_reaper_reactor, [this]() {
         LOGINFOMOD(replication, "Reaper Thread: Stopping timers");
         iomanager.cancel_timer(m_rdev_gc_timer_hdl, true);
-        iomanager.cancel_timer(m_rdev_fetch_timer_hdl, true);
         iomanager.cancel_timer(m_flush_durable_commit_timer_hdl, true);
         iomanager.cancel_timer(m_replace_member_sync_check_timer_hdl, true);
     });

src/lib/replication/service/raft_repl_service.h

@@ -53,6 +53,10 @@ class RaftReplService : public GenericReplService,
     iomgr::timer_handle_t m_flush_durable_commit_timer_hdl;
     iomgr::timer_handle_t m_replace_member_sync_check_timer_hdl;
     iomgr::IOReactor* m_reaper_reactor{nullptr};
+    // Dedicated reactor for the fetch_pending_data timer. Kept separate from the reaper reactor so that
+    // a slow flush_durable_commit_lsn() (which does synchronous metablk I/O under m_rd_map_mtx) cannot
+    // delay queued fetch batches past consensus.data_receive_timeout_ms.
+    iomgr::IOReactor* m_fetcher_reactor{nullptr};
     std::atomic< int32_t > restart_counter{0};
     std::mutex raft_restart_mutex;