From 20df560099e964c1679c8e365cafde3e2da6c11d Mon Sep 17 00:00:00 2001 From: Xiaoxi Chen Date: Mon, 22 Jun 2026 01:08:09 -0700 Subject: [PATCH 1/4] SDSTOR-22332 Run fetch_pending_data on a dedicated reactor Previously fetch_pending_data shared the raft_repl_svc_timer fiber with flush_durable_commit_lsn, gc_repl_*, and monitor_replace_member_*. When flush_durable_commit_lsn blocked on synchronous metablk I/O (under m_rd_map_mtx), queued fetch batches sat past consensus.data_receive_timeout_ms (10s), tripping the 'Data fetch timeout' assertion / TIMEOUT path. Spawn a dedicated raft_repl_fetcher reactor (1 fiber) that owns only the fetch timer. Queue and locking stay global; only the consumer thread changes. Signed-off-by: Xiaoxi Chen --- conanfile.py | 2 +- .../replication/service/raft_repl_service.cpp | 45 +++++++++++++------ .../replication/service/raft_repl_service.h | 4 ++ 3 files changed, 36 insertions(+), 15 deletions(-) diff --git a/conanfile.py b/conanfile.py index 87a52abba..4ec340dfb 100644 --- a/conanfile.py +++ b/conanfile.py @@ -9,7 +9,7 @@ class HomestoreConan(ConanFile): name = "homestore" - version = "7.5.10" + version = "7.5.11" homepage = "https://github.com/eBay/Homestore" description = "HomeStore Storage Engine" diff --git a/src/lib/replication/service/raft_repl_service.cpp b/src/lib/replication/service/raft_repl_service.cpp index f59c77d3e..a83d8e00a 100644 --- a/src/lib/replication/service/raft_repl_service.cpp +++ b/src/lib/replication/service/raft_repl_service.cpp @@ -645,19 +645,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( @@ -682,13 +669,43 @@ 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, 1u, + [this, &fetcher_latch](bool is_started) { + if (is_started) { + m_fetcher_fiber = iomanager.iofiber_self(); + // 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_fiber, [this]() { + LOGINFOMOD(replication, "Fetcher Thread: Stopping timer"); + iomanager.cancel_timer(m_rdev_fetch_timer_hdl, true); + }); iomanager.run_on_wait(m_reaper_fiber, [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); }); diff --git a/src/lib/replication/service/raft_repl_service.h b/src/lib/replication/service/raft_repl_service.h index 59fe786f8..1fdd62134 100644 --- a/src/lib/replication/service/raft_repl_service.h +++ b/src/lib/replication/service/raft_repl_service.h @@ -58,6 +58,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::io_fiber_t m_reaper_fiber; + // Dedicated fiber for the fetch_pending_data timer. Kept separate from the reaper fiber 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::io_fiber_t m_fetcher_fiber; std::atomic< int32_t > restart_counter{0}; std::mutex raft_restart_mutex; From 7eafb3a5f271f835e5fe1f86e20133191b42d8ca Mon Sep 17 00:00:00 2001 From: KOU Jilong <108138320+koujl@users.noreply.github.com> Date: Wed, 15 Jul 2026 16:31:47 +0800 Subject: [PATCH 2/4] [SDSTOR-23023] Add `recommit` option to commit_blk (#896) Normally commit_blk is called either during recovery (which advances both watermarks) or after a fresh alloc. However, in some scenarios, a blk_id may be recorded to be live by the caller but the allocator watermarks have not been successfully persisted. The new `recommit` parameter allows the caller to unconditionally advance both m_last_append_offset and m_commit_offset, covering such cases without requiring the system to be in initialising mode. Signed-off-by: Kou, Jilong --- conanfile.py | 2 +- src/include/homestore/blkdata_service.hpp | 7 ++++++- src/lib/blkdata_svc/blkdata_service.cpp | 6 +++--- src/lib/device/virtual_dev.cpp | 10 ++++++---- src/lib/device/virtual_dev.hpp | 8 +++++--- 5 files changed, 21 insertions(+), 12 deletions(-) diff --git a/conanfile.py b/conanfile.py index 4ec340dfb..ae6ed9383 100644 --- a/conanfile.py +++ b/conanfile.py @@ -9,7 +9,7 @@ class HomestoreConan(ConanFile): name = "homestore" - version = "7.5.11" + version = "7.5.12" homepage = "https://github.com/eBay/Homestore" description = "HomeStore Storage Engine" diff --git a/src/include/homestore/blkdata_service.hpp b/src/include/homestore/blkdata_service.hpp index 518ad26e7..e3e669175 100644 --- a/src/include/homestore/blkdata_service.hpp +++ b/src/include/homestore/blkdata_service.hpp @@ -166,8 +166,13 @@ class BlkDataService { * @brief Commits the block with the given MultiBlkId. * * @param bid The MultiBlkId 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. */ - BlkAllocStatus commit_blk(MultiBlkId const& bid); + BlkAllocStatus commit_blk(MultiBlkId const& bid, bool recommit = false); /** * @brief Allocates a contiguous block of disk space of the given size. This API should be called that when consumer diff --git a/src/lib/blkdata_svc/blkdata_service.cpp b/src/lib/blkdata_svc/blkdata_service.cpp index b0b73a5ef..bfc35d675 100644 --- a/src/lib/blkdata_svc/blkdata_service.cpp +++ b/src/lib/blkdata_svc/blkdata_service.cpp @@ -251,19 +251,19 @@ BlkAllocStatus BlkDataService::alloc_blks(uint32_t size, const blk_alloc_hints& return ret; } -BlkAllocStatus BlkDataService::commit_blk(MultiBlkId const& blkid) { +BlkAllocStatus BlkDataService::commit_blk(MultiBlkId const& blkid, bool recommit) { if (is_stopping()) return BlkAllocStatus::FAILED; incr_pending_request_num(); if (blkid.num_pieces() == 1) { // Shortcut to most common case - auto ret = m_vdev->commit_blk(blkid); + auto ret = m_vdev->commit_blk(blkid, recommit); decr_pending_request_num(); return ret; } auto it = blkid.iterate(); while (auto const bid = it.next()) { - auto alloc_status = m_vdev->commit_blk(*bid); + auto alloc_status = m_vdev->commit_blk(*bid, recommit); if (alloc_status != BlkAllocStatus::SUCCESS) { decr_pending_request_num(); return alloc_status; diff --git a/src/lib/device/virtual_dev.cpp b/src/lib/device/virtual_dev.cpp index b3c4bb3ac..b9b43bb89 100644 --- a/src/lib/device/virtual_dev.cpp +++ b/src/lib/device/virtual_dev.cpp @@ -174,7 +174,7 @@ bool VirtualDev::is_blk_alloced(BlkId const& blkid) const { return m_dmgr.get_chunk(blkid.chunk_num())->blk_allocator()->is_blk_alloced(blkid, true /* lock */); } -BlkAllocStatus VirtualDev::commit_blk(BlkId const& blkid) { +BlkAllocStatus VirtualDev::commit_blk(BlkId 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) { @@ -188,12 +188,14 @@ BlkAllocStatus VirtualDev::commit_blk(BlkId 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); } diff --git a/src/lib/device/virtual_dev.hpp b/src/lib/device/virtual_dev.hpp index 621411389..d9c79645b 100644 --- a/src/lib/device/virtual_dev.hpp +++ b/src/lib/device/virtual_dev.hpp @@ -170,11 +170,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 BlkId to commit explicitly. + /// @param recommit If true, skip the allocation check and commit unconditionally. + /// See BlkDataService::commit_blk for details. /// @return Allocation Status - virtual BlkAllocStatus commit_blk(BlkId const& blkid); + virtual BlkAllocStatus commit_blk(BlkId const& blkid, bool recommit = false); virtual void free_blk(BlkId const& b, VDevCPContext* vctx = nullptr, bool free_now = false); From c40afaa3c1bfd03b8b2592c32aea9a1c33a23389 Mon Sep 17 00:00:00 2001 From: Brian Szmyd Date: Wed, 15 Jul 2026 12:20:42 -0600 Subject: [PATCH 3/4] Bump version --- conanfile.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/conanfile.py b/conanfile.py index a34ad5fac..f05a6b406 100644 --- a/conanfile.py +++ b/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" From a69363402903b85b8a32aadadb9c2f01783a9034 Mon Sep 17 00:00:00 2001 From: Brian Szmyd Date: Wed, 15 Jul 2026 16:49:14 -0600 Subject: [PATCH 4/4] Adapt to iomgr v13 --- src/lib/replication/service/raft_repl_service.cpp | 8 ++++---- src/lib/replication/service/raft_repl_service.h | 6 +++--- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/lib/replication/service/raft_repl_service.cpp b/src/lib/replication/service/raft_repl_service.cpp index 98e62a448..c9525de82 100644 --- a/src/lib/replication/service/raft_repl_service.cpp +++ b/src/lib/replication/service/raft_repl_service.cpp @@ -668,9 +668,9 @@ void RaftReplService::start_repl_service_timers() { // 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, 1u, [this, &fetcher_latch](bool is_started) { + iomanager.create_reactor("raft_repl_fetcher", iomgr::INTERRUPT_LOOP, [this, &fetcher_latch](bool is_started) { if (is_started) { - m_fetcher_fiber = iomanager.iofiber_self(); + 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); @@ -688,11 +688,11 @@ void RaftReplService::start_repl_service_timers() { } void RaftReplService::stop_repl_service_timers() { - iomanager.run_on_wait(m_fetcher_fiber, [this]() { + 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_fiber, [this]() { + 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_flush_durable_commit_timer_hdl, true); diff --git a/src/lib/replication/service/raft_repl_service.h b/src/lib/replication/service/raft_repl_service.h index 652cb81cd..1d06d3527 100644 --- a/src/lib/replication/service/raft_repl_service.h +++ b/src/lib/replication/service/raft_repl_service.h @@ -52,11 +52,11 @@ class RaftReplService : public GenericReplService, iomgr::timer_handle_t m_rdev_gc_timer_hdl; iomgr::timer_handle_t m_flush_durable_commit_timer_hdl; iomgr::timer_handle_t m_replace_member_sync_check_timer_hdl; - iomgr::io_fiber_t m_reaper_fiber; - // Dedicated fiber for the fetch_pending_data timer. Kept separate from the reaper fiber so that + 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::io_fiber_t m_fetcher_fiber; + iomgr::IOReactor* m_fetcher_reactor{nullptr}; std::atomic< int32_t > restart_counter{0}; std::mutex raft_restart_mutex;