From a6c5489ce505df490410f80e6a70065c025de59d Mon Sep 17 00:00:00 2001 From: Jie Yao Date: Wed, 16 Sep 2026 14:44:45 +0800 Subject: [PATCH] SDSTOR-25689: using sisl common headers in homestore v8 --- README.md | 3 +- conanfile.py | 6 +- src/lib/blkalloc/blk_cache_queue.cpp | 2 +- src/lib/blkalloc/blk_cache_queue.h | 4 +- src/lib/blkalloc/bounded_mpmc_queue.hpp | 64 -------- src/lib/blkalloc/fixed_blk_allocator.h | 5 +- src/lib/checkpoint/cp_mgr.cpp | 23 +-- src/lib/common/coro_helpers.hpp | 123 -------------- src/lib/device/journal_vdev.cpp | 4 +- src/lib/homestore.cpp | 6 +- src/lib/index/index_service.cpp | 4 +- src/lib/index/wb_cache.cpp | 24 +-- src/lib/logstore/log_dev.cpp | 8 +- .../log_store/home_raft_log_store.cpp | 8 +- .../replication/log_store/repl_log_store.cpp | 4 +- .../replication/repl_dev/raft_repl_dev.cpp | 86 +++++----- .../repl_dev/raft_state_machine.cpp | 6 +- .../replication/repl_dev/solo_repl_dev.cpp | 15 +- .../replication/service/generic_repl_svc.cpp | 4 +- .../replication/service/raft_repl_service.cpp | 7 +- src/tests/test_append_blkalloc.cpp | 12 +- .../test_common/homestore_test_common.hpp | 6 +- src/tests/test_common/hs_repl_test_common.hpp | 2 +- src/tests/test_common/raft_repl_test_base.hpp | 50 +++--- src/tests/test_cp_mgr.cpp | 6 +- src/tests/test_data_service.cpp | 154 +++++++++--------- src/tests/test_index_crash_recovery.cpp | 6 +- src/tests/test_log_dev.cpp | 32 ++-- src/tests/test_log_store.cpp | 9 +- src/tests/test_log_store_long_run.cpp | 9 +- src/tests/test_raft_repl_dev_dynamic.cpp | 2 +- src/tests/test_solo_repl_dev.cpp | 18 +- 32 files changed, 262 insertions(+), 450 deletions(-) delete mode 100644 src/lib/blkalloc/bounded_mpmc_queue.hpp delete mode 100644 src/lib/common/coro_helpers.hpp diff --git a/README.md b/README.md index 830d14cc6..ea90d5e7f 100644 --- a/README.md +++ b/README.md @@ -170,8 +170,7 @@ if (!r) { } ``` -Bridges into non-coroutine code (`detail::detach_then`, `sync_get`, …) live in homestore's coroutine -helpers; the underlying stdexec sender/receiver machinery is hidden - consumers never depend on stdexec +the underlying stdexec sender/receiver machinery is hidden - consumers never depend on stdexec directly. Errors propagate as `std::error_condition`; exceptions are reserved for precondition bugs. ## 🖥️ Usage diff --git a/conanfile.py b/conanfile.py index d02de6580..41c093642 100644 --- a/conanfile.py +++ b/conanfile.py @@ -5,11 +5,11 @@ from conan.tools.files import copy from os.path import join -required_conan_version = ">=1.60.0" +required_conan_version = ">=2.0" class HomestoreConan(ConanFile): name = "homestore" - version = "8.2.0" + version = "8.3.0" homepage = "https://github.com/eBay/Homestore" description = "HomeStore Storage Engine" @@ -53,7 +53,7 @@ def build_requirements(self): def requirements(self): self.requires("iomgr/[^13.0]@oss/dev", transitive_headers=True) - self.requires("sisl/[^14.5]@oss/dev", transitive_headers=True) + self.requires("sisl/[^14.9]@oss/dev", transitive_headers=True) self.requires("nuraft_mesg/[^5.0]@oss/dev", transitive_headers=True) if self.settings.arch in ['x86', 'x86_64']: self.requires("isa-l/[^2.30]", transitive_headers=True) diff --git a/src/lib/blkalloc/blk_cache_queue.cpp b/src/lib/blkalloc/blk_cache_queue.cpp index fc60b663c..d46791822 100644 --- a/src/lib/blkalloc/blk_cache_queue.cpp +++ b/src/lib/blkalloc/blk_cache_queue.cpp @@ -303,7 +303,7 @@ SlabCacheQueue::SlabCacheQueue(const blk_count_t slab_size, const std::vector< b const float refill_pct, BlkAllocMetrics* parent_metrics) : m_slab_size{slab_size}, m_metrics{m_slab_size, this, parent_metrics} { for (auto& limit : level_limits) { - auto ptr{std::make_unique< BoundedMPMCQueue< blk_cache_entry > >(limit)}; + auto ptr{std::make_unique< sisl::BoundedMPMCQueue< blk_cache_entry > >(limit)}; m_level_queues.push_back(std::move(ptr)); m_total_capacity += limit; } diff --git a/src/lib/blkalloc/blk_cache_queue.h b/src/lib/blkalloc/blk_cache_queue.h index 9e4dfc88a..302aa04b6 100644 --- a/src/lib/blkalloc/blk_cache_queue.h +++ b/src/lib/blkalloc/blk_cache_queue.h @@ -22,7 +22,7 @@ #include #include -#include "bounded_mpmc_queue.hpp" +#include #include "blk_cache.h" @@ -75,7 +75,7 @@ class SlabCacheQueue { private: blk_count_t m_slab_size; // Slab size in-terms of number of pages - std::vector< std::unique_ptr< BoundedMPMCQueue< blk_cache_entry > > > m_level_queues; + std::vector< std::unique_ptr< sisl::BoundedMPMCQueue< blk_cache_entry > > > m_level_queues; std::atomic< uint64_t > m_refill_session{0}; // Is a refill pending for this slab blk_num_t m_total_capacity{0}; blk_num_t m_refill_threshold_limits; // For every level whats their threshold limit size diff --git a/src/lib/blkalloc/bounded_mpmc_queue.hpp b/src/lib/blkalloc/bounded_mpmc_queue.hpp deleted file mode 100644 index f856a70ad..000000000 --- a/src/lib/blkalloc/bounded_mpmc_queue.hpp +++ /dev/null @@ -1,64 +0,0 @@ -/********************************************************************************* - * Modifications Copyright 2017-2019 eBay Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed - * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR - * CONDITIONS OF ANY KIND, either express or implied. See the License for the - * specific language governing permissions and limitations under the License. - * - *********************************************************************************/ -#pragma once - -#include -#include - -#include - -namespace homestore { - -// Folly-free replacement for folly::MPMCQueue: a capacity-bounded boost::lockfree MPMC queue plus an -// approximate size counter (folly exposed this as sizeGuess()). The queue is bounded -- write() returns false -// when full -- which blkalloc relies on (the slab cache spills to the next level on a full level, and bounds -// total cached free-blocks). We use the default (pointer-freelist) boost::lockfree::queue pre-reserved to -// `capacity` and push only via bounded_push() (never grows past the reserved nodes): this gives bounded -// behavior WITHOUT boost::lockfree::fixed_sized's hard 65535-element cap (16-bit freelist indices), -// which the blkalloc free-block / slab-cache capacities exceed. boost::lockfree::queue requires a -// trivially-copyable element type (blk_num_t, blk_cache_entry both satisfy this). The size counter is the -// central accounting a bounded+queryable queue inherently needs; boost::lockfree itself keeps no size. -template < typename T > -class BoundedMPMCQueue { -public: - explicit BoundedMPMCQueue(const size_t capacity) : m_q{capacity} {} - - // Non-blocking enqueue; returns false if the queue is full. (folly::MPMCQueue::write) - bool write(const T& value) { - if (m_q.bounded_push(value)) { - m_size.fetch_add(1, std::memory_order_relaxed); - return true; - } - return false; - } - - // Non-blocking dequeue; returns false if the queue is empty. (folly::MPMCQueue::read) - bool read(T& out_value) { - if (m_q.pop(out_value)) { - m_size.fetch_sub(1, std::memory_order_relaxed); - return true; - } - return false; - } - - // Approximate number of elements (racy under concurrency, like folly's). (folly::MPMCQueue::sizeGuess) - size_t sizeGuess() const { return m_size.load(std::memory_order_relaxed); } - -private: - boost::lockfree::queue< T > m_q; - std::atomic< size_t > m_size{0}; -}; - -} // namespace homestore diff --git a/src/lib/blkalloc/fixed_blk_allocator.h b/src/lib/blkalloc/fixed_blk_allocator.h index d63d763c9..4d8f453dd 100644 --- a/src/lib/blkalloc/fixed_blk_allocator.h +++ b/src/lib/blkalloc/fixed_blk_allocator.h @@ -14,8 +14,9 @@ *********************************************************************************/ #pragma once +#include + #include "bitmap_blk_allocator.h" -#include "bounded_mpmc_queue.hpp" namespace homestore { /* FixedBlkAllocator is a fast allocator where it allocates only 1 size block and ALL free blocks are cached instead @@ -55,6 +56,6 @@ class FixedBlkAllocator : public BitmapBlkAllocator { state_t m_state{state_t::RECOVERING}; std::unordered_set< blk_num_t > m_reserved_blks; // Keep track of all blks which are reserved as allocated std::mutex m_reserve_blk_mtx; // Mutex used while removing marked_blks from blk_q - BoundedMPMCQueue< blk_num_t > m_free_blk_q; + sisl::BoundedMPMCQueue< blk_num_t > m_free_blk_q; }; } // namespace homestore diff --git a/src/lib/checkpoint/cp_mgr.cpp b/src/lib/checkpoint/cp_mgr.cpp index 3d9faa1a2..efaf15a1c 100644 --- a/src/lib/checkpoint/cp_mgr.cpp +++ b/src/lib/checkpoint/cp_mgr.cpp @@ -25,18 +25,13 @@ #include "common/homestore_assert.hpp" #include "common/homestore_config.hpp" #include "common/resource_mgr.hpp" -#include "common/coro_helpers.hpp" // detail::detach (fire-and-forget the flush coroutine) +#include #include "cp_internal.hpp" namespace homestore { thread_local std::stack< CP* > CPGuard::t_cp_stack; namespace { -// trigger_cp_flush returns a task awaiting the CP's broadcast completion. do_trigger_cp_flush does its -// switchover synchronously (callers fire-and-forget that side effect) and hands back one of these awaiters. -sisl::async::task< bool > await_shared(std::shared_ptr< sisl::async::shared_awaitable< bool > > comp) { - co_return co_await *comp; -} sisl::async::task< bool > ready_bool(bool v) { co_return v; } } // namespace @@ -52,7 +47,7 @@ CPManager::CPManager() : resource_mgr().register_dirty_buf_exceed_cb([this]([[maybe_unused]] int64_t dirty_buf_count, bool critical) { LOGINFO("Dirty buffer exceeded count {} critical {}", dirty_buf_count, critical); - detail::detach(this->trigger_cp_flush(false /* force */)); + sisl::async::detach(this->trigger_cp_flush(false /* force */)); }); start_timer_thread(); @@ -113,7 +108,7 @@ void CPManager::start_timer() { m_cp_timer_hdl = iomanager.schedule_thread_timer( usecs * 1000, true /* recurring */, nullptr /* cookie */, [this](void*, uint64_t exp_count) { if (exp_count > 1) { LOGINFO("cp timer expired {} times, running once", exp_count); } - detail::detach(trigger_cp_flush(false)); + sisl::async::detach(trigger_cp_flush(false)); }); }); } @@ -142,11 +137,11 @@ void CPManager::shutdown(bool require_extra_cp) { } LOGINFO("Trigger cp flush at CP shutdown"); - auto success = detail::sync_get(do_trigger_cp_flush(true /* force */, true /* flush_on_shutdown */)); + auto success = sisl::async::sync_get(do_trigger_cp_flush(true /* force */, true /* flush_on_shutdown */)); HS_REL_ASSERT_EQ(success, true, "CP Flush failed"); if (require_extra_cp) { - success = detail::sync_get(do_trigger_cp_flush(true /* force */, true /* flush_on_shutdown */)); + success = sisl::async::sync_get(do_trigger_cp_flush(true /* force */, true /* flush_on_shutdown */)); HS_REL_ASSERT_EQ(success, true, "CP Flush failed"); } @@ -201,7 +196,7 @@ void CPManager::cp_io_exit(CP* cp) { HS_DBG_ASSERT_NE(cp->m_cp_status, cp_status_t::cp_flushing); if (cp->m_enter_cnt.decrement_testz(1) && (cp->m_cp_status == cp_status_t::cp_flush_prepare)) { m_wd_cp->set_cp(cp); - detail::detach(cp_start_flush(cp)); // fire-and-forget the flush coroutine + sisl::async::detach(cp_start_flush(cp)); // fire-and-forget the flush coroutine } } @@ -228,7 +223,7 @@ sisl::async::task< bool > CPManager::do_trigger_cp_flush(bool force, bool flush_ } // If multiple threads call trigger, they all await the same shared_awaitable (broadcast). - return await_shared(m_pending_trigger_cp_comp); + return sisl::async::await_shared(m_pending_trigger_cp_comp); } else { return ready_bool(false); } @@ -281,7 +276,7 @@ sisl::async::task< bool > CPManager::do_trigger_cp_flush(bool force, bool flush_ lk.unlock(); HS_PERIODIC_LOG(DEBUG, cp, "CP critical section done, doing cp_io_exit"); - return await_shared(comp); + return sisl::async::await_shared(comp); } sisl::async::task< void > CPManager::cp_start_flush(CP* cp) { @@ -342,7 +337,7 @@ void CPManager::on_cp_flush_done(CP* cp) { if (trigger_back_2_back_cp) { HS_PERIODIC_LOG(INFO, cp, "Triggering back to back CP"); COUNTER_INCREMENT(*m_metrics, back_to_back_cps, 1); - detail::detach(trigger_cp_flush(false)); + sisl::async::detach(trigger_cp_flush(false)); } } diff --git a/src/lib/common/coro_helpers.hpp b/src/lib/common/coro_helpers.hpp deleted file mode 100644 index 18dedabde..000000000 --- a/src/lib/common/coro_helpers.hpp +++ /dev/null @@ -1,123 +0,0 @@ -/********************************************************************************* - * Modifications Copyright 2017-2019 eBay Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed - * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR - * CONDITIONS OF ANY KIND, either express or implied. See the License for the - * specific language governing permissions and limitations under the License. - * - *********************************************************************************/ -#pragma once - -#include -#include -#include -#include -#include -#include - -#include -#include - -#include -#include -#include -#include - -namespace homestore::detail { - -// Return a task that awaits a heap-held completion. Used by the "do a synchronous side effect, then return a -// task that awaits the eventual completion" pattern (e.g. CP trigger / open_log_store: the switchover or map -// insert must run at call time, only the wait is deferred). The shared_ptr is copied into the task frame so the -// completion outlives the await. -template < typename T > -inline sisl::async::task< T > await_shared(std::shared_ptr< sisl::async::shared_awaitable< T > > aw) { - co_return co_await *aw; -} -template < typename T > -inline sisl::async::task< T > await_value(std::shared_ptr< sisl::async::value_awaitable< T > > aw) { - co_return co_await *aw; -} - -// Await a value_awaitable held by reference (e.g. a long-lived member of a heap object that outlives the await). -// The reference is bound into the returned task's frame; the awaitable must stay alive until this task completes. -template < typename T > -inline sisl::async::task< T > await_value_ref(sisl::async::value_awaitable< T >& aw) { - co_return co_await aw; -} - -// write_env injects an inline scheduler so the sticky-affinity exec::task can be started without an enclosing -// scheduler context (it resumes inline on whatever thread completes its awaited sender); start_detached owns the -// operation-state on the heap and frees it on completion. Same pattern as iomgr's io_launch.hpp and sisl's -// when_all. -template < typename Task > -inline void start_coro(Task&& t) { - stdexec::start_detached( - stdexec::write_env(std::forward< Task >(t), stdexec::prop{stdexec::get_scheduler, exec::inline_scheduler{}})); -} - -// Block the calling thread until the task completes and return its value. For the infrequent control-plane and -// shutdown paths that are synchronous today (e.g. a forced CP flush awaited before proceeding). The task is -// fulfilled by other threads, so this drains its run_loop here without self-deadlocking a data path. -template < typename Task > -inline auto sync_get(Task&& task) { - auto result = stdexec::sync_wait(std::forward< Task >(task)).value(); - // result is a tuple of the task's completion values; for task it is empty (nothing to return). - if constexpr (std::tuple_size_v< decltype(result) > == 0) { - return; - } else { - return std::get< 0 >(std::move(result)); - } -} - -// Block the calling thread until the task completes OR `timeout` elapses; returns true iff it completed in time. -// The task is started detached and signals a std::promise on completion, which we time-wait on via its future. -// On timeout the detached task remains pending -- it must keep alive whatever it awaits (e.g. by holding a strong -// ref into its frame); we simply stop waiting and leave it to complete (or leak) later. Used by the data-receive -// timeout path, which then inspects per-item readiness and remediates the stragglers. -template < typename Task > -inline bool sync_wait_for(Task&& task, std::chrono::milliseconds timeout) { - auto done = std::make_shared< std::promise< void > >(); - auto fut = done->get_future(); - start_coro([](std::decay_t< Task > t, std::shared_ptr< std::promise< void > > d) -> sisl::async::task< void > { - try { - co_await std::move(t); - } catch (...) {} - d->set_value(); - }(std::forward< Task >(task), std::move(done))); - return fut.wait_for(timeout) == std::future_status::ready; -} - -// Take the task BY VALUE so it is copied into the self-owning coroutine frame; a captured-by-reference task -// would dangle once start_detached returns. Swallows exceptions so a throwing body cannot reach start_detached's -// receiver (which would std::terminate) -- tasks normally complete errors-as-values, so this is a backstop. -template < typename T > -inline sisl::async::task< void > detach_wrapper(sisl::async::task< T > t) { - try { - co_await std::move(t); - } catch (const std::exception& e) { LOGERROR("Detached task threw, swallowing: {}", e.what()); } catch (...) { - LOGERROR("Detached task threw an unknown exception, swallowing"); - } -} - -// Fire-and-forget a task whose result is not needed (e.g. a non-forced CP trigger). Starts it detached. -template < typename T > -inline void detach(sisl::async::task< T > task) { - start_coro(detach_wrapper< T >(std::move(task))); -} - -// Fire-and-forget a task but invoke fn(result) when it completes (the non-blocking ".thenValue(cb)" shape). -// fn runs on whatever thread completes the task. Both task and fn are copied into the self-owning frame. -template < typename T, typename Fn > -inline void detach_then(sisl::async::task< T > task, Fn fn) { - start_coro([](sisl::async::task< T > t, Fn f) -> sisl::async::task< void > { - f(co_await std::move(t)); - }(std::move(task), std::move(fn))); -} - -} // namespace homestore::detail diff --git a/src/lib/device/journal_vdev.cpp b/src/lib/device/journal_vdev.cpp index c3b8fe4b6..6602d67f9 100644 --- a/src/lib/device/journal_vdev.cpp +++ b/src/lib/device/journal_vdev.cpp @@ -28,7 +28,7 @@ #include "replication/repl_dev/raft_repl_dev.h" #include "device/chunk.h" #include "device/device.h" -#include "common/coro_helpers.hpp" // detail::detach (fire-and-forget the journal-exceed CP flush) +#include #include "device/physical_dev.hpp" #include "device/journal_vdev.hpp" #include "common/error.h" @@ -59,7 +59,7 @@ JournalVirtualDev::JournalVirtualDev(DeviceManager& dmgr, const vdev_info& vinfo resource_mgr().register_journal_vdev_exceed_cb([this]([[maybe_unused]] int64_t dirty_buf_count, bool critical) { // either it is critical or non-critical, call cp_flush; - detail::detach(hs()->cp_mgr().trigger_cp_flush(false /* force */)); + sisl::async::detach(hs()->cp_mgr().trigger_cp_flush(false /* force */)); if (critical) { LOGINFO("Critical journal vdev size threshold reached. Triggering truncate."); diff --git a/src/lib/homestore.cpp b/src/lib/homestore.cpp index 7ea355587..a783eefa8 100644 --- a/src/lib/homestore.cpp +++ b/src/lib/homestore.cpp @@ -34,7 +34,7 @@ #include "index/wb_cache.hpp" #include "common/homestore_utils.hpp" -#include "common/coro_helpers.hpp" // detail::sync_get (block on the first-boot CP flush) +#include #include // sisl::async::when_all (concurrent vdev format) #include "common/homestore_config.hpp" #include "common/homestore_assert.hpp" @@ -266,7 +266,7 @@ void home_store::format_and_start(std::map< uint32_t, hs_format_params >&& forma } if (!futs.empty()) { - auto const results = detail::sync_get(sisl::async::when_all(std::move(futs))); + auto const results = sisl::async::sync_get(sisl::async::when_all(std::move(futs))); for (auto const& r : results) { HS_REL_ASSERT(bool(r), "IO error during format of vdev, error={}", r ? std::string{} : r.error().message()); } @@ -306,7 +306,7 @@ void home_store::do_start() { // boot going forward on next reboot. if (m_dev_mgr->is_first_time_boot()) { // Take the first CP after we have initialized all subsystems and wait for it to complete. - detail::sync_get(m_cp_mgr->trigger_cp_flush(true /* force */)); + sisl::async::sync_get(m_cp_mgr->trigger_cp_flush(true /* force */)); m_dev_mgr->commit_formatting(); } diff --git a/src/lib/index/index_service.cpp b/src/lib/index/index_service.cpp index cdb104b77..9ce941d68 100644 --- a/src/lib/index/index_service.cpp +++ b/src/lib/index/index_service.cpp @@ -20,7 +20,7 @@ #include "index/wb_cache.hpp" #include "index/index_cp.hpp" #include "common/homestore_utils.hpp" -#include "common/coro_helpers.hpp" // detail::detach (fire-and-forget the post-recovery CP flush) +#include #include "common/homestore_assert.hpp" #include "device/virtual_dev.hpp" #include "device/physical_dev.hpp" @@ -130,7 +130,7 @@ void IndexService::start() { #endif // Force taking cp after recovery done. This makes sure that the index table is in consistent state and dirty // buffer after recovery can be added to dirty list for flushing in the new cp - detail::detach(hs()->cp_mgr().trigger_cp_flush(true /* force */)); + sisl::async::detach(hs()->cp_mgr().trigger_cp_flush(true /* force */)); } void IndexService::write_sb(uint32_t ordinal) { diff --git a/src/lib/index/wb_cache.cpp b/src/lib/index/wb_cache.cpp index fe7942e82..819cc67c8 100644 --- a/src/lib/index/wb_cache.cpp +++ b/src/lib/index/wb_cache.cpp @@ -24,7 +24,7 @@ #include "wb_cache.hpp" #include "index_cp.hpp" #include "device/virtual_dev.hpp" -#include "common/coro_helpers.hpp" // detail::detach_then +#include #include // sisl::async::when_all (collectAllUnsafe replacement) #include "common/resource_mgr.hpp" @@ -1018,15 +1018,15 @@ sisl::async::task< bool > IndexWBCache::async_cp_flush(IndexCPContext* cp_ctx) { auto const buf = preflush_bufs.front(); LOGINFO("Simulating crash after partially preflushing root-transition node {}", buf->to_string()); // buf is captured by value in the completion so the node memory outlives the in-flight write. - detail::detach_then(m_vdev->async_write(r_cast< const char* >(buf->raw_buffer()), m_node_size, - buf->m_blkid, true /* part_of_batch */), - [cp_ctx, buf](iomgr::io_result const& result) { - HS_REL_ASSERT(result, - "Partial root-transition preflush failed with error={} ({})", - result.error().value(), result.error().message()); - hs()->crash_simulator().crash(); - cp_ctx->complete(true); - }); + sisl::async::detach_then( + m_vdev->async_write(r_cast< const char* >(buf->raw_buffer()), m_node_size, buf->m_blkid, + true /* part_of_batch */), + [cp_ctx, buf](iomgr::io_result const& result) { + HS_REL_ASSERT(result, "Partial root-transition preflush failed with error={} ({})", + result.error().value(), result.error().message()); + hs()->crash_simulator().crash(); + cp_ctx->complete(true); + }); m_vdev->submit_batch(); return; } @@ -1035,7 +1035,7 @@ sisl::async::task< bool > IndexWBCache::async_cp_flush(IndexCPContext* cp_ctx) { // SQE) synchronously before suspending on the fan-out latch, so every preflush write is already // queued by the time detach() returns and submit_batch() rings the doorbell for all of them -- // the same start-then-submit order start_buffer_flush() uses. - detail::detach(preflush_root_nodes(cp_ctx, std::move(preflush_bufs))); + sisl::async::detach(preflush_root_nodes(cp_ctx, std::move(preflush_bufs))); m_vdev->submit_batch(); }); } @@ -1084,7 +1084,7 @@ void IndexWBCache::do_flush_one_buf(IndexCPContext* cp_ctx, IndexBufferPtr const LOGTRACEMOD(wbcache, "Flushing cp {} new node buf {} blkid {}", cp_ctx->id(), buf->to_string(), buf->blkid().to_string()); } - detail::detach_then( + sisl::async::detach_then( m_vdev->async_write(r_cast< const char* >(buf->raw_buffer()), m_node_size, buf->m_blkid, part_of_batch), [buf, cp_ctx](iomgr::io_result const&) { try { diff --git a/src/lib/logstore/log_dev.cpp b/src/lib/logstore/log_dev.cpp index 3056d55c6..4ace891c7 100644 --- a/src/lib/logstore/log_dev.cpp +++ b/src/lib/logstore/log_dev.cpp @@ -31,7 +31,7 @@ #include "common/homestore_assert.hpp" #include "common/homestore_config.hpp" #include "common/homestore_utils.hpp" -#include "common/coro_helpers.hpp" // detail::await_shared / await_value / sync_get +#include #include "common/crash_simulator.hpp" #include "replication/service/generic_repl_svc.h" @@ -172,7 +172,7 @@ void LogDev::stop() { // after we call stop, we need to do any pending device truncations truncate(); m_id_logstore_map.clear(); - if (allow_timer_flush()) { detail::sync_get(stop_timer()); } + if (allow_timer_flush()) { sisl::async::sync_get(stop_timer()); } } void LogDev::destroy() { @@ -203,7 +203,7 @@ sisl::async::task< int > LogDev::stop_timer() { } aw->complete(0); }); - return detail::await_value(std::move(aw)); + return sisl::async::await_value(std::move(aw)); } void LogDev::do_load(off_t device_cursor) { @@ -765,7 +765,7 @@ sisl::async::task< shared< home_log_store > > LogDev::open_log_store(logstore_id } comp = it->second.promise; } - return detail::await_shared(std::move(comp)); + return sisl::async::await_shared(std::move(comp)); } bool LogDev::remove_log_store(logstore_id_t store_id) { diff --git a/src/lib/replication/log_store/home_raft_log_store.cpp b/src/lib/replication/log_store/home_raft_log_store.cpp index 80b85c0f4..3a9842ec3 100644 --- a/src/lib/replication/log_store/home_raft_log_store.cpp +++ b/src/lib/replication/log_store/home_raft_log_store.cpp @@ -18,7 +18,7 @@ #include #include "common/homestore_assert.hpp" #include -#include "common/coro_helpers.hpp" // detail::detach_then / sync_get / await_value_ref +#include #include using namespace homestore; @@ -104,7 +104,7 @@ HomeRaftLogStore::HomeRaftLogStore(logdev_id_t logdev_id, logstore_id_t logstore m_logstore_id = logstore_id; LOGDEBUGMOD(replication, "Opening existing home log_dev={} log_store={}", m_logdev_id, logstore_id); logstore_service().open_logdev(m_logdev_id, flush_mode_t::EXPLICIT); - detail::detach_then( + sisl::async::detach_then( logstore_service().open_log_store(m_logdev_id, logstore_id, true, log_found_cb, log_replay_done_cb), [this](auto log_store) { m_log_store = std::move(log_store); @@ -393,7 +393,9 @@ void HomeRaftLogStore::purge_all_logs() { m_log_store->truncate(last_lsn, false /* in_memory_truncate_only */); } -void HomeRaftLogStore::wait_for_log_store_ready() { detail::sync_get(detail::await_value_ref(m_log_store_ready)); } +void HomeRaftLogStore::wait_for_log_store_ready() { + sisl::async::sync_get(sisl::async::await_value_ref(m_log_store_ready)); +} void HomeRaftLogStore::set_last_durable_lsn(repl_lsn_t lsn) { m_last_durable_lsn = to_store_lsn(lsn); } diff --git a/src/lib/replication/log_store/repl_log_store.cpp b/src/lib/replication/log_store/repl_log_store.cpp index effeb4efe..ad7ac1645 100644 --- a/src/lib/replication/log_store/repl_log_store.cpp +++ b/src/lib/replication/log_store/repl_log_store.cpp @@ -3,7 +3,7 @@ #include "replication/repl_dev/raft_state_machine.h" #include "replication/repl_dev/raft_repl_dev.h" #include "replication/repl_dev/common.h" -#include "common/coro_helpers.hpp" // detail::sync_get +#include namespace homestore { @@ -70,7 +70,7 @@ void ReplLogStore::end_of_append_batch(ulong start_lsn, ulong count) { // before the data is written, a restart and subsequent log replay occurs, as the in-memory state is lost, // it leaves us uncertain about whether the data was actually written, potentially leading to data // inconsistency. - detail::sync_get(m_rd.notify_after_data_written(reqs)); + sisl::async::sync_get(m_rd.notify_after_data_written(reqs)); HISTOGRAM_OBSERVE(m_rd.metrics(), data_channel_wait_latency_us, get_elapsed_time_us(cur_time)); } diff --git a/src/lib/replication/repl_dev/raft_repl_dev.cpp b/src/lib/replication/repl_dev/raft_repl_dev.cpp index b72c4bef3..187d6f8c4 100644 --- a/src/lib/replication/repl_dev/raft_repl_dev.cpp +++ b/src/lib/replication/repl_dev/raft_repl_dev.cpp @@ -12,7 +12,7 @@ #include #include -#include "common/coro_helpers.hpp" +#include #include "common/homestore_assert.hpp" #include "common/homestore_config.hpp" #include "common/homestore_utils.hpp" @@ -61,7 +61,7 @@ RaftReplDev::RaftReplDev(RaftReplService& svc, superblk< raft_repl_dev_superblk } if (m_rd_sb->is_timeline_consistent) { - detail::detach_then( + sisl::async::detach_then( logstore_service().open_log_store(m_rd_sb->logdev_id, m_rd_sb->free_blks_journal_id, false), [this](auto log_store) { m_free_blks_journal = std::move(log_store); @@ -520,7 +520,7 @@ ReplServiceError RaftReplDev::do_add_member(const replica_member_info& member, u // add_member now retries config-changing and is idempotent on already-exists internally, returning a // collapsed std::error_condition. Block for the result (control-plane, infrequent; the task is fulfilled // by nuraft/gRPC threads, not a homestore reactor, so the wait does not deadlock the data path). - auto e = detail::sync_get(m_msg_mgr.add_member(m_group_id, srv_config)); + auto e = sisl::async::sync_get(m_msg_mgr.add_member(m_group_id, srv_config)); if (!e) { RD_LOGE(trace_id, "Add member failed, member={}, err={}", boost::uuids::to_string(member.id), e.error().message()); @@ -592,7 +592,7 @@ ReplServiceError RaftReplDev::do_remove_member(const replica_id_t& member, bool } // rem_member now retries config-changing and is idempotent on member-not-found internally, returning a // collapsed std::error_condition. Block for its result (see do_add_member). - auto e = detail::sync_get(m_msg_mgr.rem_member(m_group_id, member)); + auto e = sisl::async::sync_get(m_msg_mgr.rem_member(m_group_id, member)); if (!e) { // retryable -- replace member is idempotent. RD_LOGE(trace_id, "Replace member failed to remove member, member={}, err={}", boost::uuids::to_string(member), @@ -851,7 +851,7 @@ void RaftReplDev::use_config(json_superblk raft_config_sb) { m_raft_config_sb = void RaftReplDev::on_create_snapshot(nuraft::snapshot& s, nuraft::async_result< bool >::handler_type& when_done) { RD_LOGD(NO_TRACE_ID, "create_snapshot last_idx={}/term={}", s.get_last_log_idx(), s.get_last_log_term()); auto snp_ctx = std::make_shared< nuraft_snapshot_context >(s); - auto result = detail::sync_get(m_listener->create_snapshot(snp_ctx)); + auto result = sisl::async::sync_get(m_listener->create_snapshot(snp_ctx)); auto null_except = std::shared_ptr< std::exception >(); HS_REL_ASSERT(bool(result), "Not expecting creating snapshot to return false. "); @@ -874,7 +874,7 @@ void RaftReplDev::trigger_snapshot_creation(repl_lsn_t compact_lsn, bool wait_fo } } // Step 1.2 trigger cp_flush to make sure all changes are flushed to disk before updating truncation boundary - detail::sync_get(hs()->cp_mgr().trigger_cp_flush(true /*force*/)); + sisl::async::sync_get(hs()->cp_mgr().trigger_cp_flush(true /*force*/)); RD_LOGI(NO_TRACE_ID, "cp_flush completed before updating truncation boundary to lsn={}", compact_lsn); // Step 1.3 Update truncation boundary RD_LOGI(NO_TRACE_ID, "Updating truncation boundary to lsn={}, current_truncation_boundary={}", compact_lsn, @@ -905,7 +905,7 @@ void RaftReplDev::trigger_snapshot_creation(repl_lsn_t compact_lsn, bool wait_fo } // Step 4. trigger cp_flush to make sure all changes are flushed to disk after snapshot creation and log compaction - detail::sync_get(hs()->cp_mgr().trigger_cp_flush(true /*force*/)); + sisl::async::sync_get(hs()->cp_mgr().trigger_cp_flush(true /*force*/)); RD_LOGI(NO_TRACE_ID, "cp_flush completed after snapshot creation and log compaction"); RD_LOGI(NO_TRACE_ID, "snapshot creation and compaction completed"); } @@ -1026,30 +1026,30 @@ void RaftReplDev::async_alloc_write(sisl::blob const& header, sisl::blob const& auto const data_write_start_time = Clock::now(); // Write the data - detail::detach_then(data_service().async_write(data, rreq->local_blkid()), - [this, rreq, data_write_start_time](iomgr::io_result const& r) { - // update outstanding no matter error or not; - COUNTER_DECREMENT(m_metrics, outstanding_data_write_cnt, 1); - - if (!r) { - auto const& err = r.error(); - HS_DBG_ASSERT(false, - "Error in writing data, err_code={}, category={}, err_message={}", - err.value(), err.category().name(), err.message()); - handle_error(rreq, ReplServiceError::DRIVE_WRITE_ERROR); - } else { - // update metrics for originated rreq; - const auto write_num_pieces = rreq->local_blkid().num_pieces(); - HISTOGRAM_OBSERVE(m_metrics, rreq_pieces_per_write, write_num_pieces); - HISTOGRAM_OBSERVE(m_metrics, rreq_data_write_latency_us, - get_elapsed_time_us(data_write_start_time)); - HISTOGRAM_OBSERVE(m_metrics, rreq_total_data_write_latency_us, - get_elapsed_time_us(rreq->created_time())); - - auto raft_status = m_state_machine->propose_to_raft(rreq); - if (raft_status != ReplServiceError::OK) { handle_error(rreq, raft_status); } - } - }); + sisl::async::detach_then(data_service().async_write(data, rreq->local_blkid()), + [this, rreq, data_write_start_time](iomgr::io_result const& r) { + // update outstanding no matter error or not; + COUNTER_DECREMENT(m_metrics, outstanding_data_write_cnt, 1); + + if (!r) { + auto const& err = r.error(); + HS_DBG_ASSERT( + false, "Error in writing data, err_code={}, category={}, err_message={}", + err.value(), err.category().name(), err.message()); + handle_error(rreq, ReplServiceError::DRIVE_WRITE_ERROR); + } else { + // update metrics for originated rreq; + const auto write_num_pieces = rreq->local_blkid().num_pieces(); + HISTOGRAM_OBSERVE(m_metrics, rreq_pieces_per_write, write_num_pieces); + HISTOGRAM_OBSERVE(m_metrics, rreq_data_write_latency_us, + get_elapsed_time_us(data_write_start_time)); + HISTOGRAM_OBSERVE(m_metrics, rreq_total_data_write_latency_us, + get_elapsed_time_us(rreq->created_time())); + + auto raft_status = m_state_machine->propose_to_raft(rreq); + if (raft_status != ReplServiceError::OK) { handle_error(rreq, raft_status); } + } + }); } else { RD_LOGT(tid, "Skipping data channel send since value size is 0"); rreq->add_state(repl_req_state_t::DATA_WRITTEN); @@ -1076,7 +1076,7 @@ void RaftReplDev::push_data_to_all_followers(repl_req_ptr_t rreq, sisl::sg_list // Broadcast the push to every follower and release the packet buffers once all replies are in. This is // fire-and-forget: detach() starts the coroutine and returns immediately. - detail::detach(push_data_coro(std::move(rreq), get_active_peers())); + sisl::async::detach(push_data_coro(std::move(rreq), get_active_peers())); } // Fans the push out to all followers via when_all and releases the rreq packet buffers when every reply is @@ -1166,7 +1166,7 @@ void RaftReplDev::on_push_data_received(intrusive< sisl::GenericRpcData >& rpc_d COUNTER_INCREMENT(m_metrics, outstanding_data_write_cnt, 1); // Schedule a write and upon completion, mark the data as written. - detail::detach_then( + sisl::async::detach_then( data_service().async_write(r_cast< const char* >(rreq->data()), push_req->data_size(), rreq->local_blkid()), [this, rreq, push_data_rcv_time](iomgr::io_result const& r) { // update outstanding no matter error or not; @@ -1368,7 +1368,7 @@ bool RaftReplDev::wait_for_data_receive(std::vector< repl_req_ptr_t > const& rre // block waiting here until all the futs are ready (data channel filled in and promises are made); auto all_futs_ready = - detail::sync_wait_for(sisl::async::when_all(std::move(futs)), std::chrono::milliseconds(timeout_ms)); + sisl::async::sync_wait_for(sisl::async::when_all(std::move(futs)), std::chrono::milliseconds(timeout_ms)); if (!all_futs_ready && timeout_rreqs != nullptr) { timeout_rreqs->clear(); // await_ready() == true iff that rreq's data-received promise has already completed. @@ -1458,7 +1458,7 @@ void RaftReplDev::fetch_data_from_remote(std::vector< repl_req_ptr_t > rreqs) { // Fetch is fire-and-forget: detach() starts the coroutine and returns. Copy originator out of // rreqs.front() before rreqs is moved into the coroutine frame -- the reference would otherwise dangle. nuraft_mesg::svr_id_t const originator_id = originator; - detail::detach(fetch_data_coro(std::move(builder), originator_id, std::move(rreqs))); + sisl::async::detach(fetch_data_coro(std::move(builder), originator_id, std::move(rreqs))); } // Single bidirectional fetch to the originator: on success hands the response to handle_fetch_data_response @@ -1585,7 +1585,7 @@ void RaftReplDev::on_fetch_data_received(intrusive< sisl::GenericRpcData >& rpc_ } // Fan out the reads concurrently; respond once all complete (non-blocking -- this is an RPC handler). - detail::detach_then( + sisl::async::detach_then( sisl::async::when_all(std::move(futs)), [this, rpc_data = std::move(rpc_data), sgs_vec = std::move(sgs_vec), blkids_vec = std::move(blkids_vec), headers_vec = std::move(headers_vec)](std::vector< iomgr::io_result > const& results) { @@ -1663,7 +1663,7 @@ void RaftReplDev::handle_fetch_data_response(sisl::GenericClientResponse respons auto const data_write_start_time = Clock::now(); COUNTER_INCREMENT(m_metrics, total_write_cnt, 1); COUNTER_INCREMENT(m_metrics, outstanding_data_write_cnt, 1); - detail::detach_then( + sisl::async::detach_then( data_service().async_write(r_cast< const char* >(rreq->data()), data_size, rreq->local_blkid()), [this, rreq, data_write_start_time](iomgr::io_result const& r) { // update outstanding no matter error or not; @@ -1722,7 +1722,7 @@ void RaftReplDev::handle_rollback(repl_req_ptr_t rreq) { // 3. free the allocated blocks if (rreq->has_state(repl_req_state_t::BLK_ALLOCATED)) { auto blkid = rreq->local_blkid(); - detail::detach_then(data_service().async_free_blk(blkid), [this, blkid, rreq](iomgr::io_result const& r) { + sisl::async::detach_then(data_service().async_free_blk(blkid), [this, blkid, rreq](iomgr::io_result const& r) { HS_LOG_ASSERT(bool(r), "freeing blkid={} upon error failed, potential to cause blk leak", blkid.to_string()); RD_LOGD(rreq->traceID(), "Releasing blkid={} freed successfully", blkid.to_string()); @@ -1834,7 +1834,7 @@ void RaftReplDev::handle_error(repl_req_ptr_t const& rreq, ReplServiceError err) // Free the blks which is allocated already if (rreq->has_state(repl_req_state_t::BLK_ALLOCATED)) { auto blkid = rreq->local_blkid(); - detail::detach_then(data_service().async_free_blk(blkid), [blkid](iomgr::io_result const& r) { + sisl::async::detach_then(data_service().async_free_blk(blkid), [blkid](iomgr::io_result const& r) { HS_LOG_ASSERT(bool(r), "freeing blkid={} upon error failed, potential to cause blk leak", blkid.to_string()); }); @@ -2029,7 +2029,7 @@ async_status RaftReplDev::become_leader() { // become_leader is control-plane; block for its result and wrap into the async_result task this method // returns. counter lives on this frame until sync_get returns, the same span the old continuation kept alive. - auto e = detail::sync_get(m_msg_mgr.become_leader(m_group_id)); + auto e = sisl::async::sync_get(m_msg_mgr.become_leader(m_group_id)); if (!e) { RD_LOGE(NO_TRACE_ID, "Error in becoming leader: {}", e.error().message()); return make_async_error<>(RaftReplService::to_repl_error(e.error())); @@ -2614,7 +2614,7 @@ void RaftReplDev::monitor_replace_member_replication_status() { replica_member_info out{replica_out, ""}; replica_member_info in{replica_in, ""}; - auto ret = detail::sync_get(complete_replace_member(task_id, out, in, 0, trace_id)); + auto ret = sisl::async::sync_get(complete_replace_member(task_id, out, in, 0, trace_id)); if (!ret) { RD_LOGE(trace_id, "Failed to complete replace member, next time will retry it, task_id={}, error={}", task_id, ret.error()); @@ -2738,7 +2738,7 @@ void RaftReplDev::gc_repl_reqs() { RD_LOGD(removing_rreq->traceID(), "Removing rreq [{}]", removing_rreq->to_string()); if (removing_rreq->has_state(repl_req_state_t::BLK_ALLOCATED)) { auto blkid = removing_rreq->local_blkid(); - detail::detach_then( + sisl::async::detach_then( data_service().async_free_blk(blkid), [this, blkid, removing_rreq](iomgr::io_result const& r) { if (r) { RD_LOGD(removing_rreq->traceID(), "GC rreq: Releasing blkid={} freed successfully", @@ -2994,7 +2994,7 @@ void RaftReplDev::clear_chunk_req(chunk_num_t chunk_id) { } // need to wait for the completion before returning - detail::sync_get(sisl::async::when_all(std::move(futs))); + sisl::async::sync_get(sisl::async::when_all(std::move(futs))); // TODO:: handle the error in freeing blk if necessary in the future. // for nuobject case, error for freeing blk in the emergent chunk can be ingored RD_LOGD(NO_TRACE_ID, diff --git a/src/lib/replication/repl_dev/raft_state_machine.cpp b/src/lib/replication/repl_dev/raft_state_machine.cpp index 87342e998..4ffff56c7 100644 --- a/src/lib/replication/repl_dev/raft_state_machine.cpp +++ b/src/lib/replication/repl_dev/raft_state_machine.cpp @@ -11,7 +11,7 @@ #include #include "common/homestore_config.hpp" #include "common/crash_simulator.hpp" -#include "common/coro_helpers.hpp" // detail::sync_get (block on the DSN-flush CP trigger) +#include SISL_LOGGING_DECL(replication) @@ -416,7 +416,7 @@ void RaftStateMachine::save_logical_snp_obj(nuraft::snapshot& s, ulong& obj_id, // Nuraft will compact and truncate all logs when processeing the last obj. // Update the truncation upper limit here to ensure all stale logs are truncated. m_rd.m_truncation_upper_limit.exchange(s_cast< repl_lsn_t >(s.get_last_log_idx())); - detail::sync_get(hs()->cp_mgr().trigger_cp_flush(true)); // ensure DSN is flushed to disk + sisl::async::sync_get(hs()->cp_mgr().trigger_cp_flush(true)); // ensure DSN is flushed to disk } // Update the object offset. @@ -439,7 +439,7 @@ bool RaftStateMachine::apply_snapshot(nuraft::snapshot& s) { auto snp_ctx = std::make_shared< nuraft_snapshot_context >(s); auto res = m_rd.m_listener->apply_snapshot(snp_ctx); - detail::sync_get(hs()->cp_mgr().trigger_cp_flush(true /* force */)); + sisl::async::sync_get(hs()->cp_mgr().trigger_cp_flush(true /* force */)); return res; } diff --git a/src/lib/replication/repl_dev/solo_repl_dev.cpp b/src/lib/replication/repl_dev/solo_repl_dev.cpp index 9569e8525..38582ff1a 100644 --- a/src/lib/replication/repl_dev/solo_repl_dev.cpp +++ b/src/lib/replication/repl_dev/solo_repl_dev.cpp @@ -2,7 +2,7 @@ #include #include "replication/repl_dev/solo_repl_dev.h" #include "replication/repl_dev/common.h" -#include "common/coro_helpers.hpp" // detail::detach_then +#include #include // sisl::async::when_all #include #include @@ -21,7 +21,7 @@ SoloReplDev::SoloReplDev(superblk< solo_repl_dev_superblk >&& rd_sb, bool load_e if (load_existing) { m_logdev_id = m_rd_sb->logdev_id; logstore_service().open_logdev(m_rd_sb->logdev_id, flush_mode_t::TIMER | flush_mode_t::INLINE, gid); - detail::detach_then( + sisl::async::detach_then( logstore_service().open_log_store(m_rd_sb->logdev_id, m_rd_sb->logstore_id, true /* append_mode */), [this](auto log_store) { m_data_journal = std::move(log_store); @@ -56,11 +56,12 @@ void SoloReplDev::async_alloc_write(sisl::blob const& header, sisl::blob const& // they are sibling arguments to detach_then, whose evaluation order is unspecified, so moving rreq first // would null it before local_blkids() runs. auto const blkids = rreq->local_blkids(); - detail::detach_then(data_service().async_write(value, blkids), - [this, rreq = std::move(rreq)](iomgr::io_result const& r) mutable { - HS_REL_ASSERT(bool(r), "Error in writing data"); // TODO: return error to the Listener - write_journal(std::move(rreq)); - }); + sisl::async::detach_then(data_service().async_write(value, blkids), + [this, rreq = std::move(rreq)](iomgr::io_result const& r) mutable { + HS_REL_ASSERT(bool(r), + "Error in writing data"); // TODO: return error to the Listener + write_journal(std::move(rreq)); + }); } else { write_journal(std::move(rreq)); } diff --git a/src/lib/replication/service/generic_repl_svc.cpp b/src/lib/replication/service/generic_repl_svc.cpp index b7f463685..814249b21 100644 --- a/src/lib/replication/service/generic_repl_svc.cpp +++ b/src/lib/replication/service/generic_repl_svc.cpp @@ -18,7 +18,7 @@ #include #include #include "common/homestore_assert.hpp" -#include "common/coro_helpers.hpp" // detail::sync_get +#include #include "replication/service/generic_repl_svc.h" #include "replication/service/raft_repl_service.h" #include "replication/repl_dev/solo_repl_dev.h" @@ -232,7 +232,7 @@ ReplaceMemberStatus SoloReplService::get_replace_member_status(group_id_t group_ } status SoloReplService::destroy_repl_dev(group_id_t group_id, uint64_t trace_id) { - return detail::sync_get(remove_repl_dev(group_id)); + return sisl::async::sync_get(remove_repl_dev(group_id)); } void SoloReplService::trigger_snapshot_creation(group_id_t group_id, repl_lsn_t compact_lsn, bool wait_for_commit) {} diff --git a/src/lib/replication/service/raft_repl_service.cpp b/src/lib/replication/service/raft_repl_service.cpp index d08933d6a..a1ff8ad99 100644 --- a/src/lib/replication/service/raft_repl_service.cpp +++ b/src/lib/replication/service/raft_repl_service.cpp @@ -21,7 +21,7 @@ #include #include "common/homestore_config.hpp" #include "common/homestore_assert.hpp" -#include "common/coro_helpers.hpp" +#include #include "replication/service/raft_repl_service.h" #include @@ -395,7 +395,8 @@ async_result< shared< repl_dev > > RaftReplService::create_repl_dev(group_id_t g if (members.size() > 0) { // Create a new RAFT group and add all members. create_group() will call the create_state_mgr which will create // the repl_dev instance and add it to the map. - if (auto const status = detail::sync_get(m_msg_mgr->create_group(group_id, "homestore_replication")); !status) { + if (auto const status = sisl::async::sync_get(m_msg_mgr->create_group(group_id, "homestore_replication")); + !status) { return make_async_error< shared< repl_dev > >(to_repl_error(status.error())); } @@ -407,7 +408,7 @@ async_result< shared< repl_dev > > RaftReplService::create_repl_dev(group_id_t g auto srv_config = nuraft::srv_config(nuraft_mesg::to_server_id(member), 0, boost::uuids::to_string(member), "", false, follower_priority); // add_member retries config-changing internally now, so a single call settles it. - auto const result = detail::sync_get(m_msg_mgr->add_member(group_id, srv_config)); + auto const result = sisl::async::sync_get(m_msg_mgr->add_member(group_id, srv_config)); if (result) { LOGINFOMOD(replication, "Groupid={}, new member={} added with priority={}", boost::uuids::to_string(group_id), boost::uuids::to_string(member), follower_priority); diff --git a/src/tests/test_append_blkalloc.cpp b/src/tests/test_append_blkalloc.cpp index 7b32557cb..406cff832 100644 --- a/src/tests/test_append_blkalloc.cpp +++ b/src/tests/test_append_blkalloc.cpp @@ -182,7 +182,7 @@ TEST_F(AppendBlkAllocatorTest, TestBasicWrite) { const auto io_size = 4 * Ki; LOGINFO("Step 1: run on worker thread to schedule write for {} Bytes.", io_size); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_io(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_io(io_size)); }); LOGINFO("Step 2: Wait for I/O to complete."); wait_for_all_io_complete(); @@ -195,7 +195,7 @@ TEST_F(AppendBlkAllocatorTest, TestWriteThenReadVerify) { auto io_size = 4 * Ki; LOGINFO("Step 1: run on worker thread to schedule write for {} Bytes.", io_size); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_io_verify(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_io_verify(io_size)); }); LOGINFO("Step 2: Wait for I/O to complete."); wait_for_all_io_complete(); @@ -208,7 +208,7 @@ TEST_F(AppendBlkAllocatorTest, TestWriteThenFreeBlk) { auto io_size = 4 * Mi; LOGINFO("Step 1: run on worker thread to schedule write for {} Bytes, then free blk.", io_size); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_io_free_blk(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_io_free_blk(io_size)); }); LOGINFO("Step 2: Wait for I/O to complete."); wait_for_all_io_complete(); @@ -220,7 +220,7 @@ TEST_F(AppendBlkAllocatorTest, TestCPFlush) { const auto io_size = 4 * Ki; LOGINFO("Step 1: run on worker thread to schedule write for {} Bytes.", io_size); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_io(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_io(io_size)); }); LOGINFO("Step 2: Wait for I/O to complete."); wait_for_all_io_complete(); @@ -236,7 +236,7 @@ TEST_F(AppendBlkAllocatorTest, TestWriteThenRecovey) { auto io_size = 4 * Mi; LOGINFO("Step 1: run on worker thread to schedule write for {} Bytes, then free blk.", io_size); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_io_free_blk(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_io_free_blk(io_size)); }); LOGINFO("Step 2: Wait for I/O to complete."); wait_for_all_io_complete(); @@ -254,7 +254,7 @@ TEST_F(AppendBlkAllocatorTest, TestWriteThenRecovey) { LOGINFO("Step 6: run on worker thread to schedule write for {} Bytes.", io_size); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_io(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_io(io_size)); }); LOGINFO("Step 7: Wait for I/O to complete."); wait_for_all_io_complete(); diff --git a/src/tests/test_common/homestore_test_common.hpp b/src/tests/test_common/homestore_test_common.hpp index 444bc857a..9ff4844a3 100644 --- a/src/tests/test_common/homestore_test_common.hpp +++ b/src/tests/test_common/homestore_test_common.hpp @@ -26,7 +26,7 @@ #include #include #include -#include "common/coro_helpers.hpp" // detail::sync_get / detach_then +#include #include #include #include @@ -362,9 +362,9 @@ class HSTestHelper { }; if (wait) { - on_complete(homestore::detail::sync_get(std::move(fut))); + on_complete(sisl::async::sync_get(std::move(fut))); } else { - homestore::detail::detach_then(std::move(fut), on_complete); + sisl::async::detach_then(std::move(fut), on_complete); } } diff --git a/src/tests/test_common/hs_repl_test_common.hpp b/src/tests/test_common/hs_repl_test_common.hpp index 5d4151048..b23bd5c85 100644 --- a/src/tests/test_common/hs_repl_test_common.hpp +++ b/src/tests/test_common/hs_repl_test_common.hpp @@ -307,7 +307,7 @@ class HSReplTestHelper : public HSTestHelper { repl_groups_.insert({repl_group_id, std::move(listener)}); } - auto v = homestore::detail::sync_get(hs()->repl_service().create_repl_dev(repl_group_id, members)); + auto v = sisl::async::sync_get(hs()->repl_service().create_repl_dev(repl_group_id, members)); ASSERT_EQ(v.has_value(), true) << "Error in creating repl dev for group_id=" << boost::uuids::to_string(repl_group_id).c_str() << ", err=" << v.error().message(); diff --git a/src/tests/test_common/raft_repl_test_base.hpp b/src/tests/test_common/raft_repl_test_base.hpp index 5760bc04b..647fb393c 100644 --- a/src/tests/test_common/raft_repl_test_base.hpp +++ b/src/tests/test_common/raft_repl_test_base.hpp @@ -277,8 +277,8 @@ class TestReplicatedDB : public homestore::repl_dev_listener { void snapshot_obj_write(uint64_t data_size, uint64_t data_pattern, multi_blk_id& out_blkids) { auto block_size = SISL_OPTIONS["block_size"].as< uint32_t >(); auto write_sgs = test_common::HSTestHelper::create_sgs(data_size, block_size, data_pattern); - [[maybe_unused]] auto const r = - detail::sync_get(homestore::data_service().async_alloc_write(write_sgs, blk_alloc_hints{}, out_blkids)); + [[maybe_unused]] auto const r = sisl::async::sync_get( + homestore::data_service().async_alloc_write(write_sgs, blk_alloc_hints{}, out_blkids)); for (auto const& iov : write_sgs.iovs) { iomanager.iobuf_free(uintptr_cast(iov.iov_base)); } @@ -437,20 +437,20 @@ class TestReplicatedDB : public homestore::repl_dev_listener { auto block_size = SISL_OPTIONS["block_size"].as< uint32_t >(); auto read_sgs = test_common::HSTestHelper::create_sgs(v.data_size_, block_size); - detail::detach_then(device()->async_read(v.blkid_, read_sgs, v.data_size_), - [read_sgs, k, v](iomgr::io_result const& r) { - LOGINFOMOD(replication, "Validating key={} value[blkid={} pattern={}]", k.id_, - v.blkid_.to_string(), v.data_pattern_); - RELEASE_ASSERT(bool(r), "Read of blkid={} for key={} error={}", - v.blkid_.to_string(), k.id_, - r ? std::string{} : r.error().message()); - for (auto const& iov : read_sgs.iovs) { - test_common::HSTestHelper::validate_data_buf(uintptr_cast(iov.iov_base), - iov.iov_len, v.data_pattern_); - iomanager.iobuf_free(uintptr_cast(iov.iov_base)); - } - g_helper->runner().next_task(); - }); + sisl::async::detach_then(device()->async_read(v.blkid_, read_sgs, v.data_size_), + [read_sgs, k, v](iomgr::io_result const& r) { + LOGINFOMOD(replication, "Validating key={} value[blkid={} pattern={}]", + k.id_, v.blkid_.to_string(), v.data_pattern_); + RELEASE_ASSERT(bool(r), "Read of blkid={} for key={} error={}", + v.blkid_.to_string(), k.id_, + r ? std::string{} : r.error().message()); + for (auto const& iov : read_sgs.iovs) { + test_common::HSTestHelper::validate_data_buf( + uintptr_cast(iov.iov_base), iov.iov_len, v.data_pattern_); + iomanager.iobuf_free(uintptr_cast(iov.iov_base)); + } + g_helper->runner().next_task(); + }); } else { g_helper->runner().next_task(); } @@ -519,7 +519,7 @@ class RaftReplDevTestBase : public testing::Test { for (auto const& db : dbs_) { if (db->is_zombie()) { continue; } run_on_leader(db, [this, db]() { - auto err = detail::sync_get(hs()->repl_service().remove_repl_dev(db->device()->group_id())); + auto err = sisl::async::sync_get(hs()->repl_service().remove_repl_dev(db->device()->group_id())); ASSERT_TRUE(err.has_value()) << "Error in destroying the group: " << (err ? "" : err.error().message()); }); } @@ -586,7 +586,7 @@ class RaftReplDevTestBase : public testing::Test { if (g_helper->replica_num() == replica) { for (auto const& db : dbs_) { do { - auto result = detail::sync_get(db->device()->become_leader()); + auto result = sisl::async::sync_get(db->device()->become_leader()); if (!result) { std::this_thread::sleep_for(std::chrono::milliseconds(1000)); } else { @@ -732,7 +732,7 @@ class RaftReplDevTestBase : public testing::Test { void remove_db(std::shared_ptr< TestReplicatedDB > db, bool wait_for_removal) { this->run_on_leader(db, [this, db]() { - auto err = detail::sync_get(hs()->repl_service().remove_repl_dev(db->device()->group_id())); + auto err = sisl::async::sync_get(hs()->repl_service().remove_repl_dev(db->device()->group_id())); ASSERT_TRUE(err.has_value()) << "Error in destroying the group: " << (err ? "" : err.error().message()); }); @@ -794,7 +794,7 @@ class RaftReplDevTestBase : public testing::Test { boost::uuids::to_string(member_in)); replica_member_info out{member_out, ""}; replica_member_info in{member_in, ""}; - auto result = detail::sync_get( + auto result = sisl::async::sync_get( hs()->repl_service().replace_member(db->device()->group_id(), task_id, out, in, commit_quorum)); if (error == ReplServiceError::OK) { ASSERT_EQ(result.has_value(), true) << "Error in replacing member, err=" << result.error().message(); @@ -821,7 +821,7 @@ class RaftReplDevTestBase : public testing::Test { LOGINFO("remove member, member={}", boost::uuids::to_string(member_id)); while (true) { auto result = - detail::sync_get(hs()->repl_service().remove_member(db->device()->group_id(), member_id, 0)); + sisl::async::sync_get(hs()->repl_service().remove_member(db->device()->group_id(), member_id, 0)); if (result.has_value() || result.error() == ReplServiceError::OK) { LOGINFO("Member {} already removed", boost::uuids::to_string(member_id)); break; @@ -839,8 +839,8 @@ class RaftReplDevTestBase : public testing::Test { replica_member_info member{member_id, ""}; this->run_on_leader(db, [this, error, db, member, target]() { LOGINFO("flip learner to {}, member={}", target, boost::uuids::to_string(member.id)); - auto result = - detail::sync_get(hs()->repl_service().flip_learner_flag(db->device()->group_id(), member, target, 0)); + auto result = sisl::async::sync_get( + hs()->repl_service().flip_learner_flag(db->device()->group_id(), member, target, 0)); if (error == ReplServiceError::OK) { ASSERT_EQ(result.has_value(), true) << "Error in flip_learner, err=" << result.error().message(); } else { @@ -854,8 +854,8 @@ class RaftReplDevTestBase : public testing::Test { ReplServiceError error = ReplServiceError::OK) { this->run_on_leader(db, [this, error, db, task_id]() { LOGINFO("clean replace member task, task_id={}", task_id); - auto result = - detail::sync_get(hs()->repl_service().clean_replace_member_task(db->device()->group_id(), task_id, 0)); + auto result = sisl::async::sync_get( + hs()->repl_service().clean_replace_member_task(db->device()->group_id(), task_id, 0)); if (error == ReplServiceError::OK) { ASSERT_EQ(result.has_value(), true) << "Error in clean_replace_member_task, err=" << result.error().message(); diff --git a/src/tests/test_cp_mgr.cpp b/src/tests/test_cp_mgr.cpp index f4a02a5e1..ce680c281 100644 --- a/src/tests/test_cp_mgr.cpp +++ b/src/tests/test_cp_mgr.cpp @@ -22,7 +22,7 @@ #include #include #include -#include "common/coro_helpers.hpp" // detail::sync_get / detach_then +#include #include #include "test_common/homestore_test_common.hpp" @@ -135,9 +135,9 @@ class TestCPMgr : public ::testing::Test { }; if (wait) { - on_complete(homestore::detail::sync_get(std::move(fut))); + on_complete(sisl::async::sync_get(std::move(fut))); } else { - homestore::detail::detach_then(std::move(fut), on_complete); + sisl::async::detach_then(std::move(fut), on_complete); } } diff --git a/src/tests/test_data_service.cpp b/src/tests/test_data_service.cpp index a3d685624..8f44ae702 100644 --- a/src/tests/test_data_service.cpp +++ b/src/tests/test_data_service.cpp @@ -154,23 +154,25 @@ class BlkDataServiceTest : public testing::Test { // asserts that ordering. Both are detached; the read continuation does finish_and_notify. LOGINFO("Step 2a: inject read delay and read on blkid: {}", test_blkid_ptr->to_string()); add_read_delay(); - detail::detach_then(inst().async_read(*test_blkid_ptr, *sg_read_ptr, sg_read_ptr->size), - [this, sg_read_ptr](iomgr::io_result const& r) { - RELEASE_ASSERT(bool(r), "Read error"); - - // if we are here, free_blk callback must have been called already, because data service - // layer triggers the free_blk cb firstly then sends read complete cb back to caller; - m_read_blk_done = true; - LOGINFO("read completed;"); - HS_REL_ASSERT_EQ(m_free_blk_done.load(), true, - "free blk callback should not be called before read blk completes"); - - free(*sg_read_ptr); - this->finish_and_notify(); - }); + sisl::async::detach_then(inst().async_read(*test_blkid_ptr, *sg_read_ptr, sg_read_ptr->size), + [this, sg_read_ptr](iomgr::io_result const& r) { + RELEASE_ASSERT(bool(r), "Read error"); + + // if we are here, free_blk callback must have been called already, because data + // service layer triggers the free_blk cb firstly then sends read complete cb back + // to caller; + m_read_blk_done = true; + LOGINFO("read completed;"); + HS_REL_ASSERT_EQ( + m_free_blk_done.load(), true, + "free blk callback should not be called before read blk completes"); + + free(*sg_read_ptr); + this->finish_and_notify(); + }); LOGINFO("Step 3: started async_free_blk: {}", test_blkid_ptr->to_string()); - detail::detach_then(inst().async_free_blk(*test_blkid_ptr), [this](iomgr::io_result const& r) { + sisl::async::detach_then(inst().async_free_blk(*test_blkid_ptr), [this](iomgr::io_result const& r) { RELEASE_ASSERT(bool(r), "free_blk error"); LOGINFO("completed async_free_blk"); HS_REL_ASSERT_EQ(m_free_blk_done.load(), false, "Duplicate free blk completion"); @@ -275,24 +277,24 @@ class BlkDataServiceTest : public testing::Test { auto sg_write_ptr1 = std::make_shared< sisl::sg_list >(); hints.chunk_id_hint = chunk_in_living_pdev->chunk_id(); ++m_outstanding_io_cnt; - detail::detach_then(write_sgs(io_size, sg_write_ptr1, 4, living_drive_blk, hints), - [this](iomgr::io_result const& r) { - RELEASE_ASSERT(bool(r), "Write error"); - // do not free , use it when test write - --m_outstanding_io_cnt; - ++m_total_io_comp_cnt; - }); + sisl::async::detach_then(write_sgs(io_size, sg_write_ptr1, 4, living_drive_blk, hints), + [this](iomgr::io_result const& r) { + RELEASE_ASSERT(bool(r), "Write error"); + // do not free , use it when test write + --m_outstanding_io_cnt; + ++m_total_io_comp_cnt; + }); hints.chunk_id_hint = chunk_in_missing_pdev->chunk_id(); auto sg_write_ptr2 = std::make_shared< sisl::sg_list >(); ++m_outstanding_io_cnt; - detail::detach_then(write_sgs(io_size, sg_write_ptr2, 4, missing_drive_blk, hints), - [this](iomgr::io_result const& r) { - RELEASE_ASSERT(bool(r), "Write error"); - // free(*sg_write_ptr2); do not free , use it when test write - --m_outstanding_io_cnt; - ++m_total_io_comp_cnt; - }); + sisl::async::detach_then(write_sgs(io_size, sg_write_ptr2, 4, missing_drive_blk, hints), + [this](iomgr::io_result const& r) { + RELEASE_ASSERT(bool(r), "Write error"); + // free(*sg_write_ptr2); do not free , use it when test write + --m_outstanding_io_cnt; + ++m_total_io_comp_cnt; + }); // Wait for write operations to complete wait_for_outstanding_io_done(); @@ -326,7 +328,7 @@ class BlkDataServiceTest : public testing::Test { sg->iovs.push_back(iov); ++m_outstanding_io_cnt; - detail::detach_then(inst().async_read(missing_drive_blk, *sg, io_size), [this](iomgr::io_result const& r) { + sisl::async::detach_then(inst().async_read(missing_drive_blk, *sg, io_size), [this](iomgr::io_result const& r) { RELEASE_ASSERT_EQ(!r && r.error() == std::make_error_condition(std::errc::resource_unavailable_try_again), true, "should not be able to read blk on missing drive"); --m_outstanding_io_cnt; @@ -335,28 +337,29 @@ class BlkDataServiceTest : public testing::Test { ++m_outstanding_io_cnt; LOGINFO("Step 5: read the blk from living data drive"); - detail::detach_then(inst().async_read(living_drive_blk, *sg, io_size), [this, sg](iomgr::io_result const& r) { - RELEASE_ASSERT(bool(r), "should be able to read blk on living drive"); - free(*sg); - --m_outstanding_io_cnt; - ++m_total_io_comp_cnt; - }); + sisl::async::detach_then(inst().async_read(living_drive_blk, *sg, io_size), + [this, sg](iomgr::io_result const& r) { + RELEASE_ASSERT(bool(r), "should be able to read blk on living drive"); + free(*sg); + --m_outstanding_io_cnt; + ++m_total_io_comp_cnt; + }); wait_for_outstanding_io_done(); LOGINFO("Step 6: write the blk to living data drive"); ++m_outstanding_io_cnt; - detail::detach_then(inst().async_write(*(sg_write_ptr1.get()), living_drive_blk), - [this, sg_write_ptr1](iomgr::io_result const& r) { - RELEASE_ASSERT(bool(r), "should not be able to write blk on living drive"); - free(*sg_write_ptr1); - --m_outstanding_io_cnt; - ++m_total_io_comp_cnt; - }); + sisl::async::detach_then(inst().async_write(*(sg_write_ptr1.get()), living_drive_blk), + [this, sg_write_ptr1](iomgr::io_result const& r) { + RELEASE_ASSERT(bool(r), "should not be able to write blk on living drive"); + free(*sg_write_ptr1); + --m_outstanding_io_cnt; + ++m_total_io_comp_cnt; + }); LOGINFO("Step 7: write the blk to missing data drive"); ++m_outstanding_io_cnt; - detail::detach_then( + sisl::async::detach_then( inst().async_write(*(sg_write_ptr2.get()), missing_drive_blk), [this, sg_write_ptr2](iomgr::io_result const& r) { RELEASE_ASSERT_EQ(!r && @@ -371,7 +374,7 @@ class BlkDataServiceTest : public testing::Test { LOGINFO("Step 8: free the blk from missing data drive"); ++m_outstanding_io_cnt; - detail::detach_then(inst().async_free_blk(missing_drive_blk), [this](iomgr::io_result const& r) { + sisl::async::detach_then(inst().async_free_blk(missing_drive_blk), [this](iomgr::io_result const& r) { RELEASE_ASSERT_EQ(!r && r.error() == std::make_error_condition(std::errc::resource_unavailable_try_again), true, "should not be able to free blk on living drive"); --m_outstanding_io_cnt; @@ -380,7 +383,7 @@ class BlkDataServiceTest : public testing::Test { LOGINFO("Step 9: free the blk from living data drive"); ++m_outstanding_io_cnt; - detail::detach_then(inst().async_free_blk(living_drive_blk), [this](iomgr::io_result const& r) { + sisl::async::detach_then(inst().async_free_blk(living_drive_blk), [this](iomgr::io_result const& r) { RELEASE_ASSERT(bool(r), "should be able to free blk on living drive"); --m_outstanding_io_cnt; ++m_total_io_comp_cnt; @@ -419,12 +422,13 @@ class BlkDataServiceTest : public testing::Test { auto out_bids = std::make_shared< multi_blk_id >(); ++m_outstanding_io_cnt; // out_bids are populated by write_sgs before the write completes; read them in the continuation. - detail::detach_then(write_sgs(io_size, sg, num_iovs, *out_bids), [this, sg, out_bids](iomgr::io_result const&) { - cal_write_blk_crc(*sg, *out_bids); - free(*sg); - --m_outstanding_io_cnt; - ++m_total_io_comp_cnt; - }); + sisl::async::detach_then(write_sgs(io_size, sg, num_iovs, *out_bids), + [this, sg, out_bids](iomgr::io_result const&) { + cal_write_blk_crc(*sg, *out_bids); + free(*sg); + --m_outstanding_io_cnt; + ++m_total_io_comp_cnt; + }); } // read_io has to process and send async_read all the blkids before it can exit and yielf to next io; @@ -462,7 +466,7 @@ class BlkDataServiceTest : public testing::Test { RELEASE_ASSERT(bid.is_valid(), "expecting valid bid and single blkid, is_valid: {}", bid.is_valid()); ++m_outstanding_io_cnt; - detail::detach_then(inst().async_free_blk(bid), [this, bid](iomgr::io_result const& r) { + sisl::async::detach_then(inst().async_free_blk(bid), [this, bid](iomgr::io_result const& r) { RELEASE_ASSERT(bool(r), "Free error"); LOGINFO("completed async_free_blks, bid freed: {}", bid.to_string()); // remove from ouststanding free blk set and written blk crc map; @@ -723,20 +727,20 @@ class BlkDataServiceTest : public testing::Test { iov.iov_base = iomanager.iobuf_alloc(512, iov.iov_len); sg->iovs.push_back(iov); ++m_outstanding_io_cnt; - detail::detach_then(inst().async_read(bid, *sg, io_size), - [this, bid, sg, read_crc_vec](iomgr::io_result const& r) { - // if there is any pending free blk on this read, and if we arrive here, the free blk - // callback has already been called; - RELEASE_ASSERT(bool(r), "Read error"); - // LOGINFO("read completed, bid: {}", bid.to_string()); - - // now verify read data crc equals which was previous saved on write; - verify_read_blk_crc(*sg, *read_crc_vec); - - free(*sg); - --m_outstanding_io_cnt; - ++m_total_io_comp_cnt; - }); + sisl::async::detach_then(inst().async_read(bid, *sg, io_size), + [this, bid, sg, read_crc_vec](iomgr::io_result const& r) { + // if there is any pending free blk on this read, and if we arrive here, the free + // blk callback has already been called; + RELEASE_ASSERT(bool(r), "Read error"); + // LOGINFO("read completed, bid: {}", bid.to_string()); + + // now verify read data crc equals which was previous saved on write; + verify_read_blk_crc(*sg, *read_crc_vec); + + free(*sg); + --m_outstanding_io_cnt; + ++m_total_io_comp_cnt; + }); } /** @@ -830,7 +834,7 @@ TEST_F(BlkDataServiceTest, TestBasicWrite) { const auto io_size = 4 * Ki; LOGINFO("Step 1: run on worker thread to schedule write for {} Bytes.", io_size); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_io(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_io(io_size)); }); LOGINFO("Step 2: Wait for I/O to complete."); wait_for_all_io_complete(); @@ -845,7 +849,7 @@ TEST_F(BlkDataServiceTest, TestUsedCapacity) { // check initial capacity EXPECT_EQ(inst().get_used_capacity(), 0); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_io(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_io(io_size)); }); LOGINFO("Step 2: Wait for I/O to complete."); wait_for_all_io_complete(); @@ -870,7 +874,7 @@ TEST_F(BlkDataServiceTest, TestWriteMultiplePagesSingleIov) { const auto io_size = 4 * Mi; LOGINFO("Step 1: run on worker thread to schedule write for {} Bytes.", io_size); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_io(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_io(io_size)); }); LOGINFO("Step 2: Wait for I/O to complete."); wait_for_all_io_complete(); @@ -884,7 +888,7 @@ TEST_F(BlkDataServiceTest, TestWriteMultiplePagesMultiIovs) { const auto num_iovs = 4; LOGINFO("Step 1: run on worker thread to schedule write for {} Bytes, and {} iovs", io_size, num_iovs); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size, num_iovs]() { detail::detach(this->write_io(io_size, num_iovs)); }); + [this, io_size, num_iovs]() { sisl::async::detach(this->write_io(io_size, num_iovs)); }); LOGINFO("Step 2: Wait for I/O to complete."); wait_for_all_io_complete(); @@ -897,7 +901,7 @@ TEST_F(BlkDataServiceTest, TestWriteThenReadVerify) { auto io_size = 4 * Ki; LOGINFO("Step 1: run on worker thread to schedule write for {} Bytes.", io_size); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_io_verify(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_io_verify(io_size)); }); LOGINFO("Step 3: Wait for I/O to complete."); wait_for_all_io_complete(); @@ -911,7 +915,7 @@ TEST_F(BlkDataServiceTest, TestWriteThenFreeBlk) { auto io_size = 4 * Mi; LOGINFO("Step 1: run on worker thread to schedule write for {} Bytes, then free blk.", io_size); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_io_free_blk(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_io_free_blk(io_size)); }); LOGINFO("Step 3: Wait for I/O to complete."); wait_for_all_io_complete(); @@ -927,7 +931,7 @@ TEST_F(BlkDataServiceTest, TestWriteReadThenFreeBlkAfterReadComp) { auto io_size = 4 * Ki; LOGINFO("Step 1: Run on worker thread to schedule write for {} Bytes.", io_size); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_read_free_blk(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_read_free_blk(io_size)); }); LOGINFO("Step 2: Wait for I/O to complete."); wait_for_all_io_complete(); @@ -940,7 +944,7 @@ TEST_F(BlkDataServiceTest, TestWriteReadThenFreeBeforeReadComp) { auto io_size = 4 * Ki; LOGINFO("Step 1: Run on worker thread to schedule write for {} Bytes.", io_size); iomanager.run_on_forget(iomgr::reactor_regex::random_worker, - [this, io_size]() { detail::detach(this->write_free_blk_before_read_comp(io_size)); }); + [this, io_size]() { sisl::async::detach(this->write_free_blk_before_read_comp(io_size)); }); LOGINFO("Step 4: Wait for I/O to complete."); wait_for_all_io_complete(); diff --git a/src/tests/test_index_crash_recovery.cpp b/src/tests/test_index_crash_recovery.cpp index c937a787b..40f908754 100644 --- a/src/tests/test_index_crash_recovery.cpp +++ b/src/tests/test_index_crash_recovery.cpp @@ -358,7 +358,7 @@ struct IndexCrashTest : public test_common::HSTestHelper, BtreeTestHelper< TestT LOGINFO("Destroying index btree with uuid {} root id {}", boost::uuids::to_string(this->m_bt->uuid()), this->m_bt->root_node_id()); hs()->index_service().remove_index_table(this->m_bt); - homestore::detail::sync_get(this->m_bt->destroy()); + sisl::async::sync_get(this->m_bt->destroy()); this->trigger_cp(true); ASSERT_EQ(hs()->index_service().num_tables(), 0) << "After destroying the index table, some table still exists"; @@ -374,7 +374,7 @@ struct IndexCrashTest : public test_common::HSTestHelper, BtreeTestHelper< TestT void destroy_btree() { hs()->index_service().remove_index_table(this->m_bt); - homestore::detail::sync_get(this->m_bt->destroy()); + sisl::async::sync_get(this->m_bt->destroy()); this->trigger_cp(true); this->m_shadow_map.range_erase(0, SISL_OPTIONS["num_entries"].as< uint32_t >() - 1); this->m_shadow_map.save(m_shadow_filename); @@ -1308,7 +1308,7 @@ TYPED_TEST(IndexCrashTest, DestroyTableWithPendingCpCrash) { // dirty list because free_buf only marks m_node_freed — it does not touch the list. LOGINFO("Step 4: Destroy index table — meta superblock removed from disk"); hs()->index_service().remove_index_table(this->m_bt); - homestore::detail::sync_get(this->m_bt->destroy()); + sisl::async::sync_get(this->m_bt->destroy()); // Step 5: Trigger the CP (do not wait). async_cp_flush will: // (a) write the txn_journal to disk (entries include the destroyed table's ordinal), diff --git a/src/tests/test_log_dev.cpp b/src/tests/test_log_dev.cpp index 15e10c3de..60a7750de 100644 --- a/src/tests/test_log_dev.cpp +++ b/src/tests/test_log_dev.cpp @@ -37,7 +37,7 @@ #include "common/homestore_assert.hpp" #include "logstore/log_dev.hpp" #include "test_common/homestore_test_common.hpp" -#include "common/coro_helpers.hpp" // detail::detach_then +#include using namespace homestore; @@ -295,11 +295,11 @@ TEST_F(LogDevTest, Rollback) { std::promise< bool > p; auto starting_cb = [&]() { logstore_service().open_logdev(logdev_id, flush_mode_t::EXPLICIT); - homestore::detail::detach_then( - logstore_service().open_log_store(logdev_id, store_id, false /* append_mode */), [&](auto store) { - log_store = store; - p.set_value(true); - }); + sisl::async::detach_then(logstore_service().open_log_store(logdev_id, store_id, false /* append_mode */), + [&](auto store) { + log_store = store; + p.set_value(true); + }); }; start_homestore(true /* restart */, starting_cb); p.get_future().get(); @@ -434,11 +434,11 @@ TEST_F(LogDevTest, TruncateAfterRestart) { std::promise< bool > p; auto starting_cb = [&]() { logstore_service().open_logdev(logdev_id, flush_mode_t::EXPLICIT); - homestore::detail::detach_then( - logstore_service().open_log_store(logdev_id, store_id, false /* append_mode */), [&](auto store) { - log_store = store; - p.set_value(true); - }); + sisl::async::detach_then(logstore_service().open_log_store(logdev_id, store_id, false /* append_mode */), + [&](auto store) { + log_store = store; + p.set_value(true); + }); }; start_homestore(true /* restart */, starting_cb); p.get_future().get(); @@ -652,11 +652,11 @@ TEST_F(LogDevTest, TruncateLogsAfterFlushAndRestart) { std::promise< bool > p; auto starting_cb = [&]() { logstore_service().open_logdev(logdev_id, flush_mode_t::EXPLICIT); - homestore::detail::detach_then( - logstore_service().open_log_store(logdev_id, store_id, false /* append_mode */), [&](auto store) { - log_store = store; - p.set_value(true); - }); + sisl::async::detach_then(logstore_service().open_log_store(logdev_id, store_id, false /* append_mode */), + [&](auto store) { + log_store = store; + p.set_value(true); + }); }; start_homestore(true /* restart */, starting_cb); p.get_future().get(); diff --git a/src/tests/test_log_store.cpp b/src/tests/test_log_store.cpp index f0add8b1b..801a48d0d 100644 --- a/src/tests/test_log_store.cpp +++ b/src/tests/test_log_store.cpp @@ -45,7 +45,7 @@ #include #include -#include "common/coro_helpers.hpp" // detail::detach_then +#include #include #include "logstore/log_dev.hpp" @@ -465,10 +465,9 @@ class SampleDB { for (uint32_t i{0}; i < n_log_stores; ++i) { SampleLogStoreClient* client = m_log_store_clients[i].get(); logstore_service().open_logdev(client->m_logdev_id, flush_mode_t::EXPLICIT); - homestore::detail::detach_then( - logstore_service().open_log_store(client->m_logdev_id, client->m_store_id, - false /* append_mode */), - [i, this, client](auto log_store) { client->set_log_store(log_store); }); + sisl::async::detach_then(logstore_service().open_log_store(client->m_logdev_id, client->m_store_id, + false /* append_mode */), + [i, this, client](auto log_store) { client->set_log_store(log_store); }); } }); m_helper.restart_homestore(); diff --git a/src/tests/test_log_store_long_run.cpp b/src/tests/test_log_store_long_run.cpp index 9805807f3..27e7bcb03 100644 --- a/src/tests/test_log_store_long_run.cpp +++ b/src/tests/test_log_store_long_run.cpp @@ -44,7 +44,7 @@ #include #include -#include "common/coro_helpers.hpp" // detail::detach_then +#include #include #include "logstore/log_dev.hpp" @@ -294,10 +294,9 @@ class LogStoreLongRun : public ::testing::Test { for (uint32_t i{0}; i < n_log_stores; ++i) { SampleLogStoreClient* client = m_log_store_clients[i].get(); logstore_service().open_logdev(client->m_logdev_id, flush_mode_t::EXPLICIT); - homestore::detail::detach_then( - logstore_service().open_log_store(client->m_logdev_id, client->m_store_id, - false /* append_mode */), - [i, this, client](auto log_store) { client->set_log_store(log_store); }); + sisl::async::detach_then(logstore_service().open_log_store(client->m_logdev_id, client->m_store_id, + false /* append_mode */), + [i, this, client](auto log_store) { client->set_log_store(log_store); }); } }); m_helper.restart_homestore(); diff --git a/src/tests/test_raft_repl_dev_dynamic.cpp b/src/tests/test_raft_repl_dev_dynamic.cpp index 9ee7b132e..c06204ab5 100644 --- a/src/tests/test_raft_repl_dev_dynamic.cpp +++ b/src/tests/test_raft_repl_dev_dynamic.cpp @@ -210,7 +210,7 @@ TEST_F(ReplDevDynamicTest, TwoMemberDown) { constexpr int max_retries = 3; bool succeeded = false; for (int i = 0; i < max_retries; ++i) { - auto result = detail::sync_get(hs()->repl_service().replace_member( + auto result = sisl::async::sync_get(hs()->repl_service().replace_member( db->device()->group_id(), task_id, replica_member_info{g_helper->replica_id(member_out), ""}, replica_member_info{g_helper->replica_id(member_in), ""}, 1)); if (result.has_value()) { diff --git a/src/tests/test_solo_repl_dev.cpp b/src/tests/test_solo_repl_dev.cpp index cc70deeef..27ba0e5c3 100644 --- a/src/tests/test_solo_repl_dev.cpp +++ b/src/tests/test_solo_repl_dev.cpp @@ -29,7 +29,7 @@ #include #include -#include "common/coro_helpers.hpp" // detail::sync_get +#include #include #include #include @@ -186,8 +186,8 @@ class SoloReplDevTest : public testing::Test { .vdev_size_type = vdev_size_type_t::VDEV_SIZE_DYNAMIC}}}); m_uuid1 = hs_utils::gen_random_uuid(); m_uuid2 = hs_utils::gen_random_uuid(); - m_repl_dev1 = detail::sync_get(hs()->repl_service().create_repl_dev(m_uuid1, {})).value(); - m_repl_dev2 = detail::sync_get(hs()->repl_service().create_repl_dev(m_uuid2, {})).value(); + m_repl_dev1 = sisl::async::sync_get(hs()->repl_service().create_repl_dev(m_uuid1, {})).value(); + m_repl_dev2 = sisl::async::sync_get(hs()->repl_service().create_repl_dev(m_uuid2, {})).value(); } shared< repl_dev > repl_dev1() { return m_repl_dev1; } @@ -264,7 +264,7 @@ class SoloReplDevTest : public testing::Test { RELEASE_ASSERT(err.has_value(), "Error during alloc_blks"); RELEASE_ASSERT(!blkids.empty(), "Empty blkids"); - detail::detach_then( + sisl::async::detach_then( rdev->async_write(blkids, req->write_sgs), [this, rdev, blkids, data_size, req](iomgr::io_result const& r) { RELEASE_ASSERT(bool(r), "Error during async_write"); rdev->async_write_journal(blkids, *req->header, req->key ? *req->key : sisl::blob{}, data_size, req); @@ -289,7 +289,7 @@ class SoloReplDevTest : public testing::Test { auto read_sgs = HSTestHelper::create_sgs(size, size); LOGDEBUG("[{}] Validating replay of lsn={} blkid = {}", boost::uuids::to_string(rdev.group_id()), lsn, blkid.to_string()); - detail::detach_then( + sisl::async::detach_then( rdev.async_read(blkid, read_sgs, size), [this, io_count, total_io, hdr = *jhdr, read_sgs, lsn, blkid, &rdev](iomgr::io_result const& r) { RELEASE_ASSERT(bool(r), "Error during async_read"); @@ -317,7 +317,7 @@ class SoloReplDevTest : public testing::Test { for (const auto& blkid : req->written_blkids) { uint32_t size = blkid.blk_count() * g_block_size; auto read_sgs = HSTestHelper::create_sgs(size, size); - auto r = detail::sync_get(rdev->async_read(blkid, read_sgs, size)); + auto r = sisl::async::sync_get(rdev->async_read(blkid, read_sgs, size)); RELEASE_ASSERT(bool(r), "Error during async_read"); for (auto const& iov : read_sgs.iovs) { HSTestHelper::validate_data_buf(uintptr_cast(iov.iov_base), iov.iov_len, hdr->data_pattern); @@ -343,7 +343,7 @@ class SoloReplDevTest : public testing::Test { auto sgs_size = blkid.blk_count() * g_block_size; auto read_sgs = HSTestHelper::create_sgs(sgs_size, sgs_size); - detail::detach_then( + sisl::async::detach_then( rdev.async_read(blkid, read_sgs, read_sgs.size), [this, io_count, blkid, &rdev, sgs_size, read_sgs, req](iomgr::io_result const& r) { RELEASE_ASSERT(bool(r), "Error during async_read"); @@ -369,9 +369,7 @@ class SoloReplDevTest : public testing::Test { } } - void trigger_cp_flush() { - homestore::detail::sync_get(homestore::hs()->cp_mgr().trigger_cp_flush(true /* force */)); - } + void trigger_cp_flush() { sisl::async::sync_get(homestore::hs()->cp_mgr().trigger_cp_flush(true /* force */)); } void truncate_and_verify(shared< repl_dev > repl_dev) { auto solo_dev = std::dynamic_pointer_cast< SoloReplDev >(repl_dev); // Truncate and verify the CP LSN's