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" diff --git a/src/include/homestore/blkdata_service.hpp b/src/include/homestore/blkdata_service.hpp index 3c9acc8a3..3f6cbb10a 100644 --- a/src/include/homestore/blkdata_service.hpp +++ b/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 diff --git a/src/lib/blkdata_svc/blkdata_service.cpp b/src/lib/blkdata_svc/blkdata_service.cpp index 112018d8c..e8cc02ad7 100644 --- a/src/lib/blkdata_svc/blkdata_service.cpp +++ b/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; } } } diff --git a/src/lib/device/virtual_dev.cpp b/src/lib/device/virtual_dev.cpp index 545cd8d0f..0b9570902 100644 --- a/src/lib/device/virtual_dev.cpp +++ b/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); } diff --git a/src/lib/device/virtual_dev.hpp b/src/lib/device/virtual_dev.hpp index 89a5f3397..5fcc8902c 100644 --- a/src/lib/device/virtual_dev.hpp +++ b/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); diff --git a/src/lib/replication/service/raft_repl_service.cpp b/src/lib/replication/service/raft_repl_service.cpp index 9d7c5e080..c9525de82 100644 --- a/src/lib/replication/service/raft_repl_service.cpp +++ b/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); }); diff --git a/src/lib/replication/service/raft_repl_service.h b/src/lib/replication/service/raft_repl_service.h index 15d348c45..1d06d3527 100644 --- a/src/lib/replication/service/raft_repl_service.h +++ b/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;