diff --git a/conanfile.py b/conanfile.py index 607dd9cab..5c7490ead 100644 --- a/conanfile.py +++ b/conanfile.py @@ -9,7 +9,7 @@ class HomestoreConan(ConanFile): name = "homestore" - version = "7.5.19" + version = "7.5.20" homepage = "https://github.com/eBay/Homestore" description = "HomeStore Storage Engine" diff --git a/src/include/homestore/btree/detail/btree_internal.hpp b/src/include/homestore/btree/detail/btree_internal.hpp index 0c970a563..d00356c37 100644 --- a/src/include/homestore/btree/detail/btree_internal.hpp +++ b/src/include/homestore/btree/detail/btree_internal.hpp @@ -315,33 +315,29 @@ class BtreeMetrics : public sisl::MetricsGroup { REGISTER_COUNTER(btree_num_pc_gen_mismatch, "Number of gen mismatches to recover"); REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(btree_int_node_occupancy, "Interior node occupancy", - "btree_node_occupancy", {"node_type", "interior"}, - HistogramBucketsType(PercentileBuckets)); + "btree_node_occupancy", {"node_type", "interior"}, + HistogramBucketsType(PercentileBuckets)); REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(btree_leaf_node_occupancy, "Leaf node occupancy", - "btree_node_occupancy", {"node_type", "leaf"}, - HistogramBucketsType(PercentileBuckets)); + "btree_node_occupancy", {"node_type", "leaf"}, + HistogramBucketsType(PercentileBuckets)); REGISTER_COUNTER(btree_retry_count, "number of retries"); REGISTER_COUNTER(write_err_cnt, "number of errors in write"); REGISTER_COUNTER(query_err_cnt, "number of errors in query"); REGISTER_COUNTER(btree_write_ops_count, "number of btree operations"); REGISTER_COUNTER(btree_query_ops_count, "number of btree operations"); REGISTER_COUNTER(btree_remove_ops_count, "number of btree operations"); - REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(btree_exclusive_time_in_int_node, - "Exclusive time spent (Write locked) on interior node (ns)", - "btree_exclusive_time_in_node", {"node_type", "interior"}, - HistogramBucketsType(OpLatecyBuckets)); - REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(btree_exclusive_time_in_leaf_node, - "Exclusive time spent (Write locked) on leaf node (ns)", - "btree_exclusive_time_in_node", {"node_type", "leaf"}, - HistogramBucketsType(OpLatecyBuckets)); - REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(btree_inclusive_time_in_int_node, - "Inclusive time spent (Read locked) on interior node (ns)", - "btree_inclusive_time_in_node", {"node_type", "interior"}, - HistogramBucketsType(OpLatecyBuckets)); - REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(btree_inclusive_time_in_leaf_node, - "Inclusive time spent (Read locked) on leaf node (ns)", - "btree_inclusive_time_in_node", {"node_type", "leaf"}, - HistogramBucketsType(OpLatecyBuckets)); + REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION( + btree_exclusive_time_in_int_node, "Exclusive time spent (Write locked) on interior node (ns)", + "btree_exclusive_time_in_node", {"node_type", "interior"}, HistogramBucketsType(OpLatecyBuckets)); + REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION( + btree_exclusive_time_in_leaf_node, "Exclusive time spent (Write locked) on leaf node (ns)", + "btree_exclusive_time_in_node", {"node_type", "leaf"}, HistogramBucketsType(OpLatecyBuckets)); + REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION( + btree_inclusive_time_in_int_node, "Inclusive time spent (Read locked) on interior node (ns)", + "btree_inclusive_time_in_node", {"node_type", "interior"}, HistogramBucketsType(OpLatecyBuckets)); + REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION( + btree_inclusive_time_in_leaf_node, "Inclusive time spent (Read locked) on leaf node (ns)", + "btree_inclusive_time_in_node", {"node_type", "leaf"}, HistogramBucketsType(OpLatecyBuckets)); register_me_to_farm(); } diff --git a/src/include/homestore/meta_service.hpp b/src/include/homestore/meta_service.hpp index 7baf400c1..8a87f41c3 100644 --- a/src/include/homestore/meta_service.hpp +++ b/src/include/homestore/meta_service.hpp @@ -63,7 +63,7 @@ class MetablkMetrics : public sisl::MetricsGroupWrapper { REGISTER_COUNTER(compress_backoff_ratio_cnt, "compression back-off cnt because of exceeding ratio limit"); REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(compress_ratio_percent, "compression ration percentage", - HistogramBucketsType(PercentileBuckets)); + HistogramBucketsType(PercentileBuckets)); register_me_to_farm(); } diff --git a/src/lib/blkalloc/varsize_blk_allocator.h b/src/lib/blkalloc/varsize_blk_allocator.h index 6aa459ada..666f72d6b 100644 --- a/src/lib/blkalloc/varsize_blk_allocator.h +++ b/src/lib/blkalloc/varsize_blk_allocator.h @@ -179,9 +179,8 @@ class BlkAllocMetrics : public sisl::MetricsGroup { REGISTER_COUNTER(num_retries, "Number of times it retried because of empty cache"); REGISTER_COUNTER(num_blks_alloc_direct, "Number of blks alloc attempt directly because of empty cache"); - REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(frag_pct_distribution, - "Distribution of fragmentation percentage", - HistogramBucketsType(PercentileBuckets)); + REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(frag_pct_distribution, "Distribution of fragmentation percentage", + HistogramBucketsType(PercentileBuckets)); register_me_to_farm(); } diff --git a/src/lib/checkpoint/cp_mgr.cpp b/src/lib/checkpoint/cp_mgr.cpp index ea09e0b4e..57e93148f 100644 --- a/src/lib/checkpoint/cp_mgr.cpp +++ b/src/lib/checkpoint/cp_mgr.cpp @@ -97,13 +97,11 @@ void CPManager::start_timer() { auto usecs = cp_timer_us(); LOGINFO("cp timer is set to {} usec", usecs); iomanager.run_on_wait(m_timer_fiber, [this, usecs]() { - 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); - } - trigger_cp_flush(false /* false */); - }); + 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); } + trigger_cp_flush(false /* false */); + }); }); } diff --git a/src/lib/common/homestore_assert.hpp b/src/lib/common/homestore_assert.hpp index 690021f1a..428aa73ba 100644 --- a/src/lib/common/homestore_assert.hpp +++ b/src/lib/common/homestore_assert.hpp @@ -144,7 +144,7 @@ fmt::make_format_args(detail_name, detail_val)))) \ (); \ fmt::vformat_to(fmt::appender{buf}, fmt::string_view{msgcb}, fmt::make_format_args(args...)); \ - return check_and_format_log(buf, freq, 0); \ + return check_and_format_log(buf, freq, 0); \ }), \ msg, ##__VA_ARGS__); \ } @@ -152,7 +152,7 @@ #define HS_LOG_EVERY_N(level, mod, freq, msg, ...) HS_DETAILED_LOG_EVERY_N(level, mod, freq, , , , , msg, ##__VA_ARGS__) #define HS_DETAILED_LOG_EVERY_N_SEC(level, mod, interval_sec, submod_name, submod_val, detail_name, detail_val, msg, \ - ...) \ + ...) \ { \ LOG##level##MOD_FMT( \ BOOST_PP_IF(BOOST_VMD_IS_EMPTY(mod), base, mod), \ @@ -168,7 +168,7 @@ fmt::make_format_args(detail_name, detail_val)))) \ (); \ fmt::vformat_to(fmt::appender{buf}, fmt::string_view{msgcb}, fmt::make_format_args(args...)); \ - return check_and_format_log(buf, 0, interval_sec); \ + return check_and_format_log(buf, 0, interval_sec); \ }), \ msg, ##__VA_ARGS__); \ } @@ -176,8 +176,8 @@ #define HS_LOG_EVERY_N_SEC(level, mod, interval_sec, msg, ...) \ HS_DETAILED_LOG_EVERY_N_SEC(level, mod, interval_sec, , , , , msg, ##__VA_ARGS__) -#define HS_DETAILED_LOG_EVERY_N_OR_SEC(level, mod, freq, interval_sec, submod_name, submod_val, detail_name, \ - detail_val, msg, ...) \ +#define HS_DETAILED_LOG_EVERY_N_OR_SEC(level, mod, freq, interval_sec, submod_name, submod_val, detail_name, \ + detail_val, msg, ...) \ { \ LOG##level##MOD_FMT( \ BOOST_PP_IF(BOOST_VMD_IS_EMPTY(mod), base, mod), \ @@ -193,7 +193,7 @@ fmt::make_format_args(detail_name, detail_val)))) \ (); \ fmt::vformat_to(fmt::appender{buf}, fmt::string_view{msgcb}, fmt::make_format_args(args...)); \ - return check_and_format_log(buf, freq, interval_sec); \ + return check_and_format_log(buf, freq, interval_sec); \ }), \ msg, ##__VA_ARGS__); \ } @@ -361,7 +361,7 @@ * If interval_sec >= 300, the behavior effectively becomes "log first occurrence after each 5min reset." */ [[maybe_unused]] static bool check_and_format_log(fmt::memory_buffer& buf, uint64_t freq = 0, - uint64_t interval_sec = 0) { + uint64_t interval_sec = 0) { static constexpr uint64_t COUNTER_RESET_SEC{300}; // Reset every 5 minutes static thread_local Clock::time_point last_cleanup{Clock::now()}; static thread_local std::unordered_map< size_t, std::pair< uint32_t, uint64_t > > log_map{}; @@ -388,8 +388,7 @@ const size_t msg_hash = std::hash< std::string_view >{}(msg); // Milliseconds since last cleanup (max ~49 days with uint32_t) - const uint32_t now_ms = - std::chrono::duration_cast< std::chrono::milliseconds >(now - last_cleanup).count(); + const uint32_t now_ms = std::chrono::duration_cast< std::chrono::milliseconds >(now - last_cleanup).count(); auto [it, happened] = log_map.emplace(msg_hash, std::make_pair(now_ms, 0)); uint32_t elapsed_ms = 0; @@ -429,8 +428,8 @@ // Update state after logging (so next log shows "since this log") if (!happened) { - it->second.first = now_ms; // Update timestamp - it->second.second = 0; // Reset count (next occurrence will be "1 since this log") + it->second.first = now_ms; // Update timestamp + it->second.second = 0; // Reset count (next occurrence will be "1 since this log") } } diff --git a/src/lib/device/virtual_dev.hpp b/src/lib/device/virtual_dev.hpp index d9c79645b..000ce8b97 100644 --- a/src/lib/device/virtual_dev.hpp +++ b/src/lib/device/virtual_dev.hpp @@ -55,7 +55,7 @@ class VirtualDevMetrics : public sisl::MetricsGroupWrapper { REGISTER_COUNTER(random_chunk_allocation_cnt, "random chunk allocation count"); // ideally it should be zero for hdd REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(blk_alloc_latency, "Blk allocation latency", "blk_alloc_latency", - {}, HistogramBucketsType(OpLatecyBuckets)); + {}, HistogramBucketsType(OpLatecyBuckets)); register_me_to_farm(); } diff --git a/src/lib/logstore/log_dev.cpp b/src/lib/logstore/log_dev.cpp index 60d269509..81caeb488 100644 --- a/src/lib/logstore/log_dev.cpp +++ b/src/lib/logstore/log_dev.cpp @@ -437,12 +437,12 @@ bool LogDev::can_flush_in_this_thread() { return (!HS_DYNAMIC_CONFIG(logstore.flush_only_in_dedicated_thread) && iomanager.am_i_worker_reactor()); } -bool LogDev::flush_if_necessary(int64_t threshold_size) { +bool LogDev::flush_if_necessary(int64_t threshold_size, bool force) { if (is_stopping()) return false; incr_pending_request_num(); if (!can_flush_in_this_thread()) { iomanager.run_on_forget(logstore_service().flush_thread(), - [this, threshold_size]() { flush_if_necessary(threshold_size); }); + [this, threshold_size, force]() { flush_if_necessary(threshold_size, force); }); decr_pending_request_num(); return false; } @@ -456,12 +456,37 @@ bool LogDev::flush_if_necessary(int64_t threshold_size) { bool const flush_by_size = (pending_sz >= threshold_size); bool const flush_by_time = !flush_by_size && pending_sz && (elapsed_time > HS_DYNAMIC_CONFIG(logstore.max_time_between_flush_us)); - if (flush_by_size || flush_by_time) { + if (force || flush_by_size || flush_by_time) { std::unique_lock lck(m_flush_mtx, std::try_to_lock); if (lck.owns_lock()) { decr_pending_request_num(); return flush(); } + // Lost the race to a concurrent flush() (e.g. another write's own flush_if_necessary() call, or + // HomeLogStore::truncate()'s internal flush()). That concurrent flush's own snapshot of m_log_idx + // may not include the data that made this call decide to flush, so giving up here silently can + // leave that data unflushed indefinitely if nothing else ever calls flush_if_necessary() again + // for this logdev -- normally masked by the periodic flush timer eventually retrying, but with + // it disabled (flush_timer_frequency_us=0, e.g. in tests) this is a real, reproducible hang: the + // very last write issued in a run has no later trigger to fall back on. + // + // Reschedule with force=true. LogDev::flush() unconditionally resets m_last_flush_time at its + // start, so the flush that just won this race may have already reset that clock by the time the + // retry runs -- re-deriving flush_by_size/flush_by_time here could then read false again (this + // write's own size may still be under threshold_size) and silently abandon the retry for good. + // force=true skips that re-derivation: we already decided to flush, so the retry only needs to + // keep trying the lock, not re-litigate whether to. + // + // Target a random *worker* reactor, not flush_thread() directly, even though the retry ultimately + // needs to run there (can_flush_in_this_thread() bounces it back). We're already ON flush_thread + // here, and IOReactor::deliver_msg runs same-reactor targets inline instead of queuing them -- + // so posting straight back to flush_thread would recurse synchronously on every failed try_lock. + // Under sustained contention this stack-overflows the process (confirmed with the lock held for + // ~200ms in testing; this predates force -- it's already present in the plain reschedule above). + // Routing through random_worker first forces a real queued hop, so the stack unwinds between + // attempts no matter how many times try_lock fails. + iomanager.run_on_forget(iomgr::reactor_regex::random_worker, + [this, threshold_size]() { flush_if_necessary(threshold_size, /* force = */ true); }); } decr_pending_request_num(); return false; @@ -481,6 +506,12 @@ bool LogDev::flush_under_guard() { return flush(); } +#ifdef _PRERELEASE +void LogDev::test_touch_last_flush_time() { + if (iomgr_flip::instance()->test_flip("test_touch_last_flush_time")) { m_last_flush_time = Clock::now(); } +} +#endif + bool LogDev::flush() { if (!is_ready()) { THIS_LOGDEV_LOG(INFO, "LogDev is not ready to flush, log_dev={}", m_logdev_id); diff --git a/src/lib/logstore/log_dev.hpp b/src/lib/logstore/log_dev.hpp index 31d62c2c0..43688a23e 100644 --- a/src/lib/logstore/log_dev.hpp +++ b/src/lib/logstore/log_dev.hpp @@ -583,6 +583,18 @@ class LogDev : public std::enable_shared_from_this< LogDev > { return HS_DYNAMIC_CONFIG(logstore.flush_threshold_size) - sizeof(log_group_header); } +#ifdef _PRERELEASE + // Test-only: acquires m_flush_mtx directly, bypassing flush() -- lets a UT hold the real lock + // deterministically without a real flush cycle, which would unavoidably flush any data appended + // while held (its m_log_idx snapshot is taken fresh at call time). See + // LogStoreTest.FlushIfNecessaryRetrySurvivesStaleClockReset. + std::unique_lock< iomgr::FiberManagerLib::mutex > test_acquire_flush_mtx() { return std::unique_lock(m_flush_mtx); } + + // Test-only: simulates a concurrent flush's clock-reset side effect (resets m_last_flush_time) + // without actually flushing. Only active if the "test_touch_last_flush_time" flip is armed. + void test_touch_last_flush_time(); +#endif + LogDev(logdev_id_t logdev_id, flush_mode_t flush_mode = static_cast< flush_mode_t >(HS_DYNAMIC_CONFIG(logstore.flush_mode)), uuid_t pid = boost::uuids::nil_uuid()); @@ -653,9 +665,12 @@ class LogDev : public std::enable_shared_from_this< LogDev > { /// redirect the flush to a flush thread and run there. /// /// @param threshold_size [Optional]: Size in bytes after which it will flush, if set to -1, will use default size + /// @param force [Optional]: Skip the size/time threshold check and go straight to the try_lock. Used internally + /// when rescheduling a retry after losing the try_lock race, so the retry can't be silently talked out of + /// trying again by a stale-clock re-derivation of the threshold check. /// /// @return bool : True if it has flushed the data, false otherwise - bool flush_if_necessary(int64_t threshold_size = -1); + bool flush_if_necessary(int64_t threshold_size = -1, bool force = false); /// @brief : Look at all logstore and find out the safest point upto which it can truncate and truncate them. /// diff --git a/src/lib/logstore/log_store_service.cpp b/src/lib/logstore/log_store_service.cpp index b7a2be946..75bf51b8b 100644 --- a/src/lib/logstore/log_store_service.cpp +++ b/src/lib/logstore/log_store_service.cpp @@ -417,13 +417,13 @@ LogStoreServiceMetrics::LogStoreServiceMetrics() : sisl::MetricsGroup("LogStores REGISTER_HISTOGRAM(logdev_flush_size_distribution, "Distribution of flush data size", HistogramBucketsType(ExponentialOfTwoBuckets)); REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(logdev_flush_records_distribution, - "Distribution of num records to flush", - HistogramBucketsType(LinearUpto128Buckets)); + "Distribution of num records to flush", + HistogramBucketsType(LinearUpto128Buckets)); REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(logstore_record_size, "Distribution of log record size", - HistogramBucketsType(ExponentialOfTwoBuckets)); + HistogramBucketsType(ExponentialOfTwoBuckets)); REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(logdev_post_flush_processing_latency, - "Logdev post flush processing (including callbacks) latency", - HistogramBucketsType(OpLatecyBuckets)); + "Logdev post flush processing (including callbacks) latency", + HistogramBucketsType(OpLatecyBuckets)); REGISTER_HISTOGRAM(logdev_flush_time_us, "time elapsed since last flush time in us", HistogramBucketsType(OpLatecyBuckets)); diff --git a/src/lib/replication/repl_dev/solo_repl_dev.h b/src/lib/replication/repl_dev/solo_repl_dev.h index d777e16b7..9d9e20cf7 100644 --- a/src/lib/replication/repl_dev/solo_repl_dev.h +++ b/src/lib/replication/repl_dev/solo_repl_dev.h @@ -78,9 +78,7 @@ class SoloReplDev : public ReplDev { return std::vector< peer_info >{ peer_info{.id_ = m_group_id, .replication_idx_ = 0, .last_succ_resp_us_ = 0, .priority_ = 1}}; } - std::vector< replica_id_t > get_replication_quorum() override { - return std::vector< replica_id_t >{m_group_id}; - } + std::vector< replica_id_t > get_replication_quorum() override { return std::vector< replica_id_t >{m_group_id}; } void reconcile_leader() override {} void yield_leadership(bool immediate_yield, replica_id_t candidate) override {} bool is_ready_for_traffic() const override { return true; } diff --git a/src/lib/replication/service/generic_repl_svc.cpp b/src/lib/replication/service/generic_repl_svc.cpp index 21d4780c3..4e34a94c3 100644 --- a/src/lib/replication/service/generic_repl_svc.cpp +++ b/src/lib/replication/service/generic_repl_svc.cpp @@ -79,7 +79,7 @@ hs_stats GenericReplService::get_cap_stats() const { ///////////////////// SoloReplService specializations and CP Callbacks ///////////////////////////// SoloReplService::SoloReplService(cshared< ReplApplication >& repl_app) : GenericReplService{repl_app} {} -SoloReplService::~SoloReplService(){}; +SoloReplService::~SoloReplService() {}; void SoloReplService::start() { for (auto const& [buf, mblk] : m_sb_bufs) { diff --git a/src/lib/replication/service/raft_repl_service.cpp b/src/lib/replication/service/raft_repl_service.cpp index 02bfd1602..72f43cceb 100644 --- a/src/lib/replication/service/raft_repl_service.cpp +++ b/src/lib/replication/service/raft_repl_service.cpp @@ -660,8 +660,8 @@ void RaftReplService::start_repl_service_timers() { HS_DYNAMIC_CONFIG(consensus.replace_member_sync_check_interval_ms) * 1000 * 1000, true /* recurring */, nullptr, [this](void*, uint64_t exp_count) { if (exp_count > 1) { - LOGINFOMOD(replication, - "replace member sync check timer expired {} times, running once", exp_count); + LOGINFOMOD(replication, "replace member sync check timer expired {} times, running once", + exp_count); } monitor_replace_member_replication_status(); }); @@ -674,27 +674,22 @@ 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) { - 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(); - } - }); + 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(); } @@ -711,7 +706,7 @@ void RaftReplService::stop_repl_service_timers() { }); } -void RaftReplService::add_to_fetch_queue(cshared &rdev, std::vector rreqs) { +void RaftReplService::add_to_fetch_queue(cshared< RaftReplDev >& rdev, std::vector< repl_req_ptr_t > rreqs) { std::unique_lock lg(m_pending_fetch_mtx); m_pending_fetch_batches.push(std::make_pair(rdev, std::move(rreqs))); } @@ -719,7 +714,7 @@ void RaftReplService::add_to_fetch_queue(cshared &rdev, std::vector void RaftReplService::fetch_pending_data() { std::unique_lock lg(m_pending_fetch_mtx); while (!m_pending_fetch_batches.empty()) { - auto const &[d, rreqs] = m_pending_fetch_batches.front(); + auto const& [d, rreqs] = m_pending_fetch_batches.front(); if (get_elapsed_time_ms(rreqs.at(0)->created_time()) < HS_DYNAMIC_CONFIG(consensus.wait_data_write_timer_ms)) { break; } diff --git a/src/tests/test_index_gc.cpp b/src/tests/test_index_gc.cpp index 712f56202..af6205e9c 100644 --- a/src/tests/test_index_gc.cpp +++ b/src/tests/test_index_gc.cpp @@ -12,7 +12,6 @@ #include "btree_helpers/btree_decls.h" #include "btree_helpers/blob_route.h" - using namespace homestore; SISL_LOGGING_INIT(HOMESTORE_LOG_MODS) @@ -22,11 +21,11 @@ SISL_LOGGING_DECL(test_index_gc) SISL_OPTION_GROUP( test_index_gc, (num_iters, "", "num_iters", "number of iterations for rand ops", - ::cxxopts::value< uint32_t >()->default_value("500000"), "number"), + ::cxxopts::value< uint32_t >()->default_value("500000"), "number"), (num_entries, "", "num_entries", "number of entries to test with", ::cxxopts::value< uint32_t >()->default_value("7000"), "number"), - (num_put, "", "num_put", "number of entries to test with", - ::cxxopts::value< uint32_t >()->default_value("20000"), "number"), + (num_put, "", "num_put", "number of entries to test with", ::cxxopts::value< uint32_t >()->default_value("20000"), + "number"), (run_time, "", "run_time", "run time for io", ::cxxopts::value< uint64_t >()->default_value("36000"), "seconds"), (disable_merge, "", "disable_merge", "disable_merge", ::cxxopts::value< bool >()->default_value("0"), ""), (operation_list, "", "operation_list", "operation list instead of default created following by percentage", @@ -34,7 +33,8 @@ SISL_OPTION_GROUP( (preload_size, "", "preload_size", "number of entries to preload tree with", ::cxxopts::value< uint32_t >()->default_value("1000"), "number"), (init_device, "", "init_device", "init device", ::cxxopts::value< bool >()->default_value("1"), ""), - (ignore_node_lock_refresh, "", "ignore_node_lock_refresh", "ignore node lock refresh", ::cxxopts::value< bool >(), ""), + (ignore_node_lock_refresh, "", "ignore_node_lock_refresh", "ignore node lock refresh", ::cxxopts::value< bool >(), + ""), (cleanup_after_shutdown, "", "cleanup_after_shutdown", "cleanup after shutdown", ::cxxopts::value< bool >()->default_value("1"), ""), (max_merge_level, "", "max_merge_level", "max merge level", ::cxxopts::value< uint8_t >()->default_value("127"), @@ -42,7 +42,6 @@ SISL_OPTION_GROUP( (seed, "", "seed", "random engine seed, use random if not defined", ::cxxopts::value< uint64_t >()->default_value("0"), "number")) - using BtreeType = IndexTable< BlobRouteByChunkKey, TestFixedValue >; using op_func_t = std::function< void(void) >; static constexpr uint32_t g_num_fibers{4}; @@ -93,10 +92,10 @@ class TestIndexGC : public ::testing::Test { create_io_reactors(g_num_fibers); m_run_time = SISL_OPTIONS["run_time"].as< uint64_t >(); - //m_operations["put"] = std::bind(&BtreeTestHelper::put_random, this); - //m_operations["range_remove"] = std::bind(&BtreeTestHelper::range_remove_existing_random, this); - //m_operations["range_query"] = std::bind(&BtreeTestHelper::query_random, this); - //m_operations["get"] = std::bind(&BtreeTestHelper::get_random, this); + // m_operations["put"] = std::bind(&BtreeTestHelper::put_random, this); + // m_operations["range_remove"] = std::bind(&BtreeTestHelper::range_remove_existing_random, this); + // m_operations["range_query"] = std::bind(&BtreeTestHelper::query_random, this); + // m_operations["get"] = std::bind(&BtreeTestHelper::get_random, this); m_bt = std::make_shared< BtreeType >(uuid, parent_uuid, 0, m_cfg); hs()->index_service().add_index_table(m_bt); LOGINFO("Added index table to index service"); @@ -131,17 +130,17 @@ class TestIndexGC : public ::testing::Test { }; auto ctx = std::make_shared< Context >(); for (uint32_t i{0}; i < num_io_reactors; ++i) { - iomanager.create_reactor("homeblks_long_running_io" + std::to_string(i), iomgr::INTERRUPT_LOOP, 1u, - [this, ctx](bool is_started) { - if (is_started) { - { - std::unique_lock< std::mutex > lk{ctx->mtx}; - m_fibers.push_back(iomanager.iofiber_self()); - ++(ctx->thread_cnt); + iomanager.create_reactor("homeblks_long_running_io" + std::to_string(i), iomgr::INTERRUPT_LOOP, 1u, + [this, ctx](bool is_started) { + if (is_started) { + { + std::unique_lock< std::mutex > lk{ctx->mtx}; + m_fibers.push_back(iomanager.iofiber_self()); + ++(ctx->thread_cnt); + } + ctx->cv.notify_one(); } - ctx->cv.notify_one(); - } - }); + }); } { std::unique_lock< std::mutex > lk{ctx->mtx}; @@ -152,7 +151,8 @@ class TestIndexGC : public ::testing::Test { void put_many_random(uint16_t chunk_id, uint32_t num_put) { for (uint16_t i = 0; i < num_put; ++i) { - auto key = BlobRouteByChunkKey{BlobRouteByChunk(chunk_id, g_randval_generator(g_re), g_randval_generator(g_re))}; + auto key = + BlobRouteByChunkKey{BlobRouteByChunk(chunk_id, g_randval_generator(g_re), g_randval_generator(g_re))}; auto value = TestFixedValue::generate_rand(); auto sreq = BtreeSinglePutRequest{&key, &value, btree_put_type::UPSERT}; sreq.enable_route_tracing(); @@ -189,13 +189,13 @@ class TestIndexGC : public ::testing::Test { status = m_bt->query(query_req, valid_blob_indexes); if (status != homestore::btree_status_t::success) { LOGERROR("Failed to query blobs after purging reserved chunk={} in gc index table, index ret={}", chunk_id, - status); + status); return false; } if (!valid_blob_indexes.empty()) { LOGERROR("gc index table is not empty for chunk={} after purging, valid_blob_indexes.size={}", chunk_id, - valid_blob_indexes.size()); + valid_blob_indexes.size()); return SISL_OPTIONS["ignore_node_lock_refresh"].as< bool >(); } @@ -205,18 +205,18 @@ class TestIndexGC : public ::testing::Test { void gc_task(uint32_t idx) { LOGINFO("GC task {} started", idx); auto num_puts = SISL_OPTIONS["num_put"].as< uint32_t >(); - while(!time_to_stop()) { + while (!time_to_stop()) { // Step 1: preload chunks with some data for (uint16_t i = 0; i < 20; ++i) { - uint16_t chunk_id = 20*idx + i; + uint16_t chunk_id = 20 * idx + i; put_many_random(chunk_id, num_puts); } LOGDEBUG("Preload done for index {}", idx); // Step 2: start chunk gc for (uint16_t i = 0; i < 20; ++i) { - uint16_t chunk_id = 20*idx + i; - ASSERT_TRUE(do_gc(chunk_id)); + uint16_t chunk_id = 20 * idx + i; + ASSERT_TRUE(do_gc(chunk_id)); } LOGDEBUG("GC done for index {}", idx); auto elapsed_time = get_elapsed_time_sec(m_start_time); @@ -234,7 +234,7 @@ class TestIndexGC : public ::testing::Test { LOGINFO("Put task {} started", idx); while (!time_to_stop()) { for (uint16_t i = 0; i < 1000; ++i) { - uint16_t chunk_id = 20*idx + i; + uint16_t chunk_id = 20 * idx + i; put_many_random(chunk_id, 10); } std::this_thread::sleep_for(std::chrono::milliseconds(1000)); @@ -247,8 +247,9 @@ class TestIndexGC : public ::testing::Test { LOGINFO("Get task {} started", idx); while (!time_to_stop()) { for (uint16_t i = 0; i < 1000; ++i) { - uint16_t chunk_id = 20*idx + i; - auto key = BlobRouteByChunkKey{BlobRouteByChunk(chunk_id, g_randval_generator(g_re), g_randval_generator(g_re))}; + uint16_t chunk_id = 20 * idx + i; + auto key = BlobRouteByChunkKey{ + BlobRouteByChunk(chunk_id, g_randval_generator(g_re), g_randval_generator(g_re))}; TestFixedValue value; homestore::BtreeSingleGetRequest get_req{&key, &value}; m_bt->get(get_req); @@ -259,9 +260,7 @@ class TestIndexGC : public ::testing::Test { m_test_done_latch.count_down(); } - bool time_to_stop() const { - return (get_elapsed_time_sec(m_start_time) > m_run_time); - } + bool time_to_stop() const { return (get_elapsed_time_sec(m_start_time) > m_run_time); } BtreeConfig m_cfg{g_node_size}; std::shared_ptr< BtreeType > m_bt; @@ -278,18 +277,10 @@ class TestIndexGC : public ::testing::Test { TEST_F(TestIndexGC, chunk_gc_test) { LOGINFO("Chunk GC test start"); m_start_time = Clock::now(); - iomanager.run_on_forget(m_fibers[0], [this]() { - gc_task(0); - }); - iomanager.run_on_forget(m_fibers[1], [this]() { - gc_task(1); - }); - iomanager.run_on_forget(m_fibers[2], [this]() { - put_task(2); - }); - iomanager.run_on_forget(m_fibers[3], [this]() { - get_task(3); - }); + iomanager.run_on_forget(m_fibers[0], [this]() { gc_task(0); }); + iomanager.run_on_forget(m_fibers[1], [this]() { gc_task(1); }); + iomanager.run_on_forget(m_fibers[2], [this]() { put_task(2); }); + iomanager.run_on_forget(m_fibers[3], [this]() { get_task(3); }); m_test_done_latch.wait(); LOGINFO("Chunk GC test passed"); } diff --git a/src/tests/test_log_store.cpp b/src/tests/test_log_store.cpp index 939c18d65..1852dadef 100644 --- a/src/tests/test_log_store.cpp +++ b/src/tests/test_log_store.cpp @@ -23,6 +23,7 @@ #include #include #include +#include #include #include #include @@ -566,6 +567,10 @@ class LogStoreTest : public ::testing::Test { set_store_workload_freq(inp_freqs); // Equal distribution by default } + HomeLogStore* raw_log_store(size_t idx = 0) { + return SampleDB::instance().m_log_store_clients[idx]->m_log_store.get(); + } + void kickstart_inserts(uint32_t batch_size, uint32_t q_depth, uint32_t holes_per_batch = 0) { m_batch_size = batch_size; m_q_depth = q_depth; @@ -1185,6 +1190,95 @@ TEST_F(LogStoreTest, FlushSync) { #endif } +#ifdef _PRERELEASE +// Regression test for a gap in flush_if_necessary()'s lost-try_lock-race reschedule: it re-invokes +// flush_if_necessary(), re-deriving flush_by_size/flush_by_time from scratch. LogDev::flush() +// unconditionally resets m_last_flush_time at its start, so a concurrent flush that won the race can +// reset the clock right as the retry runs -- if this write's own size never crosses threshold_size, +// both conditions can read false and the retry silently drops. +// +// A real racer flush() can't isolate this: whoever holds the lock long enough for our write to lose +// its try_lock will, once released, take a fresh snapshot that includes the write anyway (it was +// appended while the lock was held) -- so releasing the racer completes the write regardless of the +// fix. Instead this uses two test-only LogDev hooks: +// - test_acquire_flush_mtx(): grabs m_flush_mtx directly, bypassing flush() -- releasing it later +// triggers no snapshot/flush of its own. +// - test_touch_last_flush_time(): resets m_last_flush_time on demand (gated by the +// "test_touch_last_flush_time" flip) -- simulates just the clock-reset side effect of a concurrent +// flush, without actually flushing anything. +// +// Sequence: acquire the lock directly (m_last_flush_time starts infinitely stale on a fresh logdev, so +// flush_by_time reads true) -> issue the write, whose flush_if_necessary() decides to flush but loses +// the try_lock race, queuing the retry -> touch the clock (simulating the racing flush completing) and +// hold the lock 200ms (with max_time_between_flush_us set far larger, so every retry in this window +// sees a still-too-recent clock and a too-small pending size -- exactly the condition the fix targets) +// -> release the lock via plain unlock (not flush()). Pre-fix, the retry dropped during that window is +// gone for good and the write hangs; post-fix, force=true has kept retrying and now wins the lock. +TEST_F(LogStoreTest, FlushIfNecessaryRetrySurvivesStaleClockReset) { + LOGINFO("Step 1: Reinit with no records -- this test issues its own write directly"); + this->init(0); + auto* log_store = this->raw_log_store(0); + auto logdev = log_store->get_logdev(); + + HS_SETTINGS_FACTORY().modifiable_settings([](auto& s) { + s.logstore.max_time_between_flush_us = 2000000ul; /* 2s -- see step 4 above */ + }); + HS_SETTINGS_FACTORY().save(); + static constexpr int64_t threshold = 4096; // bigger than one small write, so flush_by_size alone + // can't save it -- flush_by_time must do the work. + + LOGINFO("Step 2: Directly acquire the flush lock, bypassing flush() entirely"); + auto flush_lock = logdev->test_acquire_flush_mtx(); + + LOGINFO("Step 3: Issue our target write and its own flush_if_necessary() call -- guaranteed to lose " + "the try_lock race since we hold the lock directly"); + bool io_memory{false}; + const auto lsn = log_store->get_contiguous_issued_seq_num(-1) + 1; + auto* d = SampleLogStoreClient::prepare_data(lsn, io_memory); + auto completed = std::make_shared< std::promise< void > >(); + auto fut = completed->get_future(); + log_store->write_async(lsn, {uintptr_cast(d), d->total_size(), false}, nullptr, + [completed, d, io_memory](logstore_seq_num_t, const sisl::io_blob&, logdev_key, void*) { + if (io_memory) { + iomanager.iobuf_free(uintptr_cast(d)); + } else { + std::free(voidptr_cast(d)); + } + completed->set_value(); + }); + logdev->flush_if_necessary(threshold); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); // let the try_lock fail and the retry + // get queued before we touch the clock + + LOGINFO("Step 4: Simulate a concurrent flush completing (clock reset, nothing else), then hold the " + "lock for 200ms -- long enough that no amount of retry-latency luck can save a non-robust " + "retry"); + flip::FlipClient* fc = iomgr_flip::client_instance(); + flip::FlipFrequency freq; + freq.set_count(1); + freq.set_percent(100); + flip::FlipCondition dont_care_cond; + fc->create_condition("", flip::Operator::DONT_CARE, (int)1, &dont_care_cond); + fc->inject_noreturn_flip("test_touch_last_flush_time", {dont_care_cond}, freq); + logdev->test_touch_last_flush_time(); + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + + LOGINFO("Step 5: Release the lock directly (not via flush()) -- must not touch m_log_idx or complete " + "anything on its own"); + flush_lock.unlock(); + + LOGINFO("Step 6: The write must still complete -- pre-fix, the retry silently drops during step 4's " + "window and never comes back"); + auto status = fut.wait_for(std::chrono::seconds(10)); + + HS_SETTINGS_FACTORY().modifiable_settings([](auto& s) { s.logstore.max_time_between_flush_us = 300ul; }); + HS_SETTINGS_FACTORY().save(); + + ASSERT_EQ(status, std::future_status::ready) + << "write never completed -- lost-race reschedule was silently dropped by a stale clock reset"; +} +#endif + TEST_F(LogStoreTest, DeleteMultipleLogStores) { const auto nrecords = (SISL_OPTIONS["num_records"].as< uint32_t >() * 5) / 100;