From 829b00b3ed6a7f91c9b09895db5d5da9c23e0a2b Mon Sep 17 00:00:00 2001 From: Mehdi Hosseini Date: Thu, 10 Sep 2026 16:15:09 -0700 Subject: [PATCH 1/6] logstore: reschedule flush_if_necessary() on lost try_lock instead of dropping it Root cause ---------- LogDev::flush_if_necessary() uses std::try_to_lock on m_flush_mtx. If it loses that race to a concurrent flush() (another write's own flush_if_necessary() call, or HomeLogStore::truncate()'s internal flush()), it silently gives up -- no retry, no rescheduling. The concurrent flush's own snapshot of m_log_idx may not include the data that made *this* call decide to flush, so the dropped attempt can leave that data unflushed indefinitely, with nothing to notice. In production this is normally masked by the periodic flush timer (flush_timer_frequency_us) eventually retrying. But test_log_store.cpp disables that timer globally for the whole binary (flush_timer_frequency_us=0, comment: "Currently only tests set it to 0"), so flush_if_necessary()'s own try_lock is the *only* path that will ever flush pending data in these tests. LogStoreTest.VarRateInsertThenTruncate's Step 4 deliberately runs writes and repeated truncate_validate() calls concurrently; if the very last batch issued in a run has its own flush_if_necessary() call lose the try_lock race -- because a concurrent truncate()'s internal flush() is holding the mutex at that exact instant -- and nothing else ever calls flush_if_necessary() again for that logdev, that batch's data is never flushed, and any waiter on its completion (LogStoreTest::wait_for_inserts()) blocks forever. How this was diagnosed ----------------------- - Reproduced by looping test_log_store --gtest_filter= LogStoreTest.VarRateInsertThenTruncate:LogStoreTest.ThrottleSeqInsertThenRecover repeatedly; the hang appeared within 2-12 attempts, consistently. - gdb (thread apply all bt, all threads) on a stuck process showed only the main thread blocked in wait_for_inserts()'s condition_variable::wait -- every other thread (including the dedicated log_flush_thread) fully idle in epoll_wait. Nothing deadlocked; the process genuinely had nothing left to do. - Temporary diagnostic logging (issue/complete per LSN) showed exactly one clean insert_next_batch()-sized batch (10 consecutive lsns, one store) issued but never completed, with zero flush or I/O errors logged anywhere in the run. - Confirmed pre-existing and unrelated to any other in-flight change: an unmodified stable/v7.x checkout reproduced the identical signature (same "one store's last batch never completes, no errors, nothing deadlocked") under the same repro loop. - Confirmed the mechanism directly: added temporary logging inside flush_if_necessary() itself (try_lock=SUCCESS/FAILED/threshold_not_met per call) and observed no further flush attempt at all for the affected logdev after the stuck batch's own issuing thread finished (consistent with that thread's own flush_if_necessary() call losing the race and nothing else ever coming along to retry). Fix --- When the try_lock fails, reschedule a follow-up flush_if_necessary() attempt onto the log store flush thread via iomanager.run_on_forget(), mirroring the existing reschedule already used a few lines above for the !can_flush_in_this_thread() case. This does not change behavior during shutdown: LogDev::stop() already performs its own guaranteed, blocking flush_under_guard() call (a real blocking lock acquire, not try_to_lock) before teardown, which flushes everything up to m_log_idx - 1 regardless of whether a rescheduled flush_if_necessary() attempt bails out early via is_stopping(). Verification ------------ Stress-tested by looping the same two-test gtest filter 60 consecutive times after the fix: 0 hangs out of 60 attempts (previously reproduced within 2-12 attempts on the same machine/build, both with and without this fix absent). --- src/lib/logstore/log_dev.cpp | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/src/lib/logstore/log_dev.cpp b/src/lib/logstore/log_dev.cpp index 60d269509..eb2aabfd0 100644 --- a/src/lib/logstore/log_dev.cpp +++ b/src/lib/logstore/log_dev.cpp @@ -462,6 +462,17 @@ bool LogDev::flush_if_necessary(int64_t threshold_size) { 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 a follow-up + // attempt on the flush thread instead of dropping it, mirroring the !can_flush_in_this_thread() + // reschedule above. + iomanager.run_on_forget(logstore_service().flush_thread(), + [this, threshold_size]() { flush_if_necessary(threshold_size); }); } decr_pending_request_num(); return false; From 1a8e940c6b77699e395aeadc33448962be2c1769 Mon Sep 17 00:00:00 2001 From: Mehdi Hosseini Date: Fri, 11 Sep 2026 12:48:59 -0700 Subject: [PATCH 2/6] logstore: make the lost-race reschedule immune to a stale-clock re-check, and fix a latent stack overflow in it A reviewer (JacksonYao287) flagged that the reschedule added in the previous commit re-invokes flush_if_necessary(), which re-derives flush_by_size/ flush_by_time from scratch. LogDev::flush() unconditionally resets m_last_flush_time at its very start, so the concurrent flush that stole the race will, by the time the retry runs, likely have already reset that clock to "now" (flush_by_time re-evaluates false) while this call's own pending size may still be under threshold_size (flush_by_size false too) -- silently dropping the retry with nothing left to trigger it again. Verified with a deterministic repro (a temporary gate hook holding the flush lock for a controlled window) before and after: reliably reproduces the drop pre-fix, passes reliably post-fix. Fix: flush_if_necessary() gains a `force` parameter. The reschedule now passes force=true, which skips the size/time re-derivation entirely and goes straight to try_lock -- the retry's only job is to keep trying to actually acquire the lock, not re-litigate whether to. Still a non-blocking try_to_lock, so no new deadlock surface. Building that same repro also surfaced a separate, more severe, pre-existing bug in the original reschedule: IOReactor::deliver_msg takes a same-thread shortcut, calling the target handler inline instead of queuing it whenever sender and receiver are the same reactor. The reschedule always targets flush_thread(), and by construction we're always already on flush_thread() when we reach it (that's how we got past can_flush_in_this_thread()), so every failed retry was a real recursive call on the same stack, not a queued task. Under sustained lock contention (confirmed with a lock held for as little as ~200ms) that recursion runs tens of thousands of frames deep and stack-overflows the process -- reproduced identically with and without the force=true change above, confirming it predates this PR entirely. Fixed by bouncing the retry through iomgr::reactor_regex:: random_worker first (the same pattern already used a few lines below in this file for an analogous reentrancy concern), forcing a genuine cross-thread hop so the message is always queued and this stack frame unwinds before the retry runs, no matter how many times it fails. --- src/lib/logstore/log_dev.cpp | 37 ++++++++++++++++++++++++++++-------- src/lib/logstore/log_dev.hpp | 5 ++++- 2 files changed, 33 insertions(+), 9 deletions(-) diff --git a/src/lib/logstore/log_dev.cpp b/src/lib/logstore/log_dev.cpp index eb2aabfd0..ca064e56f 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,7 +456,7 @@ 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(); @@ -468,11 +468,32 @@ bool LogDev::flush_if_necessary(int64_t threshold_size) { // 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 a follow-up - // attempt on the flush thread instead of dropping it, mirroring the !can_flush_in_this_thread() - // reschedule above. - iomanager.run_on_forget(logstore_service().flush_thread(), - [this, threshold_size]() { flush_if_necessary(threshold_size); }); + // very last write issued in a run has no later trigger to fall back on. + // + // Reschedule a follow-up attempt -- with force=true, unlike the initial call. LogDev::flush() + // unconditionally resets m_last_flush_time at its very start, so the concurrent flush that just + // stole this race will, by the time the retry runs, likely have already reset that clock to "now" + // (making flush_by_time re-evaluate false) while this call's own pending size may still be under + // threshold_size (making flush_by_size false too) -- silently abandoning the retry with nothing + // left to trigger it again. force=true bypasses that re-derivation on the retry: we already + // decided this data needs flushing, so the retry's only job is to keep trying to actually acquire + // the lock, not re-litigate whether to. + // + // Reschedule onto a random *worker* reactor, not flush_thread() directly, even though flush_thread + // is where the retry ultimately needs to run (can_flush_in_this_thread() will bounce it back + // there). We're already executing on flush_thread at this point (that's how we got past the + // can_flush_in_this_thread() check above) -- IOReactor::deliver_msg takes a same-thread shortcut + // that calls the target inline instead of queuing it when sender and receiver are the same + // reactor, so posting straight back to flush_thread here would recurse synchronously on this same + // call stack for every failed try_lock, not queue a new task. Under sustained lock contention + // (verified: a lock held for as little as ~200ms is enough) that recursion runs thousands of + // frames deep and stack-overflows the process -- a real crash, reproduced with and without + // force=true, i.e. pre-existing in the original reschedule-on-lost-race fix, not introduced by + // force. Bouncing through random_worker first forces a genuine cross-thread hop (a real reactor, + // never flush_thread itself), so the message is queued and this stack frame unwinds before the + // retry runs, no matter how many times it 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; diff --git a/src/lib/logstore/log_dev.hpp b/src/lib/logstore/log_dev.hpp index 31d62c2c0..ff8b40547 100644 --- a/src/lib/logstore/log_dev.hpp +++ b/src/lib/logstore/log_dev.hpp @@ -653,9 +653,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. /// From e608b6e326b564b5d9bef9d56a5919fd92091d15 Mon Sep 17 00:00:00 2001 From: Mehdi Hosseini Date: Fri, 11 Sep 2026 14:05:04 -0700 Subject: [PATCH 3/6] logstore: add a regression test for the stale-clock reschedule fix Adds LogStoreTest.FlushIfNecessaryRetrySurvivesStaleClockReset, a deterministic repro for the gap fixed in the previous commit (flush_if_necessary()'s lost-try_lock-race reschedule silently dropped by a stale m_last_flush_time reset) and for the stack overflow that turned out to predate it. A real racer flush() call can't cleanly isolate this: whatever holds the lock long enough for the target write's try_lock to fail will, once released, take its own fresh snapshot of m_log_idx -- which by then includes the target write too, so releasing the racer would flush it regardless of whether the fix is present. Instead this adds two small, _PRERELEASE-only LogDev hooks: - test_acquire_flush_mtx(): grabs the real m_flush_mtx directly, without going through flush() at all, so releasing it later never triggers a 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), simulating the side effect of a concurrent flush completing without the side effect of actually flushing anything. Verified against all three relevant code states, same test unchanged: - force removed (random_worker kept): fails (assertion -- write never completes), 9/10 runs. - random_worker reverted to flush_thread() (force kept): crashes (SIGSEGV, the stack overflow from the previous commit), 5/5 runs. - both fixes present: passes, 20/20 runs. --- src/lib/logstore/log_dev.cpp | 6 ++ src/lib/logstore/log_dev.hpp | 16 ++++++ src/tests/test_log_store.cpp | 107 +++++++++++++++++++++++++++++++++++ 3 files changed, 129 insertions(+) diff --git a/src/lib/logstore/log_dev.cpp b/src/lib/logstore/log_dev.cpp index ca064e56f..a76c30f73 100644 --- a/src/lib/logstore/log_dev.cpp +++ b/src/lib/logstore/log_dev.cpp @@ -513,6 +513,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 ff8b40547..6783189b7 100644 --- a/src/lib/logstore/log_dev.hpp +++ b/src/lib/logstore/log_dev.hpp @@ -583,6 +583,22 @@ 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: directly acquire m_flush_mtx from test code, bypassing flush()/flush_under_guard() + // entirely -- lets a UT create a lost-try_lock race deterministically (holding the real lock + // flush_if_necessary() contends on) without going through a real flush cycle, which would + // unavoidably also flush any data appended while the lock was held (its snapshot of m_log_idx is + // taken fresh at flush() call time, so it would include anything appended before that call runs, + // regardless of when the lock was originally acquired). See + // LogStoreTest.FlushIfNecessaryRetrySurvivesStaleClockReset. + std::unique_lock< iomgr::FiberManagerLib::mutex > test_acquire_flush_mtx() { return std::unique_lock(m_flush_mtx); } + + // Test-only: simulates the side effect of an unrelated concurrent flush completing (resetting + // m_last_flush_time) without actually flushing anything. Only takes effect if the + // "test_touch_last_flush_time" flip is armed, so calling this is inert otherwise. + 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()); diff --git a/src/tests/test_log_store.cpp b/src/tests/test_log_store.cpp index 939c18d65..d745d410d 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,108 @@ TEST_F(LogStoreTest, FlushSync) { #endif } +#ifdef _PRERELEASE +// Regression test for a gap in flush_if_necessary()'s lost-try_lock-race reschedule: the reschedule +// re-invokes flush_if_necessary(), which re-derives flush_by_size/flush_by_time from scratch. Since +// LogDev::flush() unconditionally resets m_last_flush_time at its very start, the concurrent flush that +// won the race can reset that clock right as the retry runs, making flush_by_time false again -- and if +// this write's own size never crosses threshold_size on its own, flush_by_size stays false too, silently +// dropping the retry with nothing left to trigger it again. +// +// A real racer flush() call can't cleanly isolate this: whatever holds the lock long enough for our +// target write's try_lock to fail will, once released, take its own fresh snapshot of m_log_idx -- which +// by then includes the target write too (it was appended while the racer held the lock), so releasing +// the racer would flush the target write regardless of whether the fix is present. To test the reschedule +// mechanism in isolation, this uses two test-only LogDev hooks instead: +// - test_acquire_flush_mtx(): grabs the real m_flush_mtx directly, without going through flush() at all +// -- so releasing it later never triggers any 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), simulating the *side effect* of a concurrent flush completing, +// without the *other* side effect of actually flushing anything. +// +// Sequence: +// 1. Acquire m_flush_mtx directly (test_acquire_flush_mtx()). A fresh logdev's m_last_flush_time starts +// effectively infinitely stale, so flush_by_time would read true for anyone checking right now. +// 2. Issue the target write and its own flush_if_necessary() call. It decides to flush (flush_by_time +// true, since nothing has touched the clock yet) but loses the try_lock race (we hold m_flush_mtx) -- +// that loss is what queues the retry. +// 3. Arm the "test_touch_last_flush_time" flip and call test_touch_last_flush_time() -- resets the clock +// to *now*, exactly as a concurrent flush completing would, but with the lock still held by us and +// nothing actually flushed. +// 4. Keep holding m_flush_mtx for 200ms, with max_time_between_flush_us configured far larger (2s) than +// that hold. Every retry landing anywhere in this window sees a still-too-recent clock (flush_by_time +// false) and a pending size still under threshold_size (flush_by_size false) -- the exact condition +// the fix targets, held open long enough that timing luck can't save a retry that isn't robust to it. +// 5. Release m_flush_mtx via a plain unlock (not flush()) -- this does not touch m_log_idx or complete +// anything on its own. Pre-fix, whatever retry was dropped during step 4's window is gone for good, +// and the write never completes. Post-fix, force=true has kept retrying the whole time and now wins +// the lock, calling a real flush() that (for the first time) takes a snapshot including the target +// write and completes it. +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; From f9bbf11af8077b00c0234e519f149f21e2b09f65 Mon Sep 17 00:00:00 2001 From: Mehdi Hosseini Date: Fri, 11 Sep 2026 14:58:40 -0700 Subject: [PATCH 4/6] logstore: tighten comments on the stale-clock fix and its regression test No functional change -- trims the explanatory comments added in the previous two commits (force flag, random_worker routing, and the new UT's docstring) down to the essential why, without dropping any of the reasoning. --- src/lib/logstore/log_dev.cpp | 35 ++++++++++-------------- src/lib/logstore/log_dev.hpp | 14 ++++------ src/tests/test_log_store.cpp | 53 ++++++++++++++---------------------- 3 files changed, 39 insertions(+), 63 deletions(-) diff --git a/src/lib/logstore/log_dev.cpp b/src/lib/logstore/log_dev.cpp index a76c30f73..1a7b535cd 100644 --- a/src/lib/logstore/log_dev.cpp +++ b/src/lib/logstore/log_dev.cpp @@ -470,28 +470,21 @@ bool LogDev::flush_if_necessary(int64_t threshold_size, bool force) { // 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 a follow-up attempt -- with force=true, unlike the initial call. LogDev::flush() - // unconditionally resets m_last_flush_time at its very start, so the concurrent flush that just - // stole this race will, by the time the retry runs, likely have already reset that clock to "now" - // (making flush_by_time re-evaluate false) while this call's own pending size may still be under - // threshold_size (making flush_by_size false too) -- silently abandoning the retry with nothing - // left to trigger it again. force=true bypasses that re-derivation on the retry: we already - // decided this data needs flushing, so the retry's only job is to keep trying to actually acquire - // the lock, not re-litigate whether to. + // 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. // - // Reschedule onto a random *worker* reactor, not flush_thread() directly, even though flush_thread - // is where the retry ultimately needs to run (can_flush_in_this_thread() will bounce it back - // there). We're already executing on flush_thread at this point (that's how we got past the - // can_flush_in_this_thread() check above) -- IOReactor::deliver_msg takes a same-thread shortcut - // that calls the target inline instead of queuing it when sender and receiver are the same - // reactor, so posting straight back to flush_thread here would recurse synchronously on this same - // call stack for every failed try_lock, not queue a new task. Under sustained lock contention - // (verified: a lock held for as little as ~200ms is enough) that recursion runs thousands of - // frames deep and stack-overflows the process -- a real crash, reproduced with and without - // force=true, i.e. pre-existing in the original reschedule-on-lost-race fix, not introduced by - // force. Bouncing through random_worker first forces a genuine cross-thread hop (a real reactor, - // never flush_thread itself), so the message is queued and this stack frame unwinds before the - // retry runs, no matter how many times it fails. + // 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 + // just ~200ms; 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); }); } diff --git a/src/lib/logstore/log_dev.hpp b/src/lib/logstore/log_dev.hpp index 6783189b7..43688a23e 100644 --- a/src/lib/logstore/log_dev.hpp +++ b/src/lib/logstore/log_dev.hpp @@ -584,18 +584,14 @@ class LogDev : public std::enable_shared_from_this< LogDev > { } #ifdef _PRERELEASE - // Test-only: directly acquire m_flush_mtx from test code, bypassing flush()/flush_under_guard() - // entirely -- lets a UT create a lost-try_lock race deterministically (holding the real lock - // flush_if_necessary() contends on) without going through a real flush cycle, which would - // unavoidably also flush any data appended while the lock was held (its snapshot of m_log_idx is - // taken fresh at flush() call time, so it would include anything appended before that call runs, - // regardless of when the lock was originally acquired). See + // 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 the side effect of an unrelated concurrent flush completing (resetting - // m_last_flush_time) without actually flushing anything. Only takes effect if the - // "test_touch_last_flush_time" flip is armed, so calling this is inert otherwise. + // 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 diff --git a/src/tests/test_log_store.cpp b/src/tests/test_log_store.cpp index d745d410d..1852dadef 100644 --- a/src/tests/test_log_store.cpp +++ b/src/tests/test_log_store.cpp @@ -1191,42 +1191,29 @@ TEST_F(LogStoreTest, FlushSync) { } #ifdef _PRERELEASE -// Regression test for a gap in flush_if_necessary()'s lost-try_lock-race reschedule: the reschedule -// re-invokes flush_if_necessary(), which re-derives flush_by_size/flush_by_time from scratch. Since -// LogDev::flush() unconditionally resets m_last_flush_time at its very start, the concurrent flush that -// won the race can reset that clock right as the retry runs, making flush_by_time false again -- and if -// this write's own size never crosses threshold_size on its own, flush_by_size stays false too, silently -// dropping the retry with nothing left to trigger it again. +// 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() call can't cleanly isolate this: whatever holds the lock long enough for our -// target write's try_lock to fail will, once released, take its own fresh snapshot of m_log_idx -- which -// by then includes the target write too (it was appended while the racer held the lock), so releasing -// the racer would flush the target write regardless of whether the fix is present. To test the reschedule -// mechanism in isolation, this uses two test-only LogDev hooks instead: -// - test_acquire_flush_mtx(): grabs the real m_flush_mtx directly, without going through flush() at all -// -- so releasing it later never triggers any snapshot/flush of its own. +// 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), simulating the *side effect* of a concurrent flush completing, -// without the *other* side effect of actually flushing anything. +// "test_touch_last_flush_time" flip) -- simulates just the clock-reset side effect of a concurrent +// flush, without actually flushing anything. // -// Sequence: -// 1. Acquire m_flush_mtx directly (test_acquire_flush_mtx()). A fresh logdev's m_last_flush_time starts -// effectively infinitely stale, so flush_by_time would read true for anyone checking right now. -// 2. Issue the target write and its own flush_if_necessary() call. It decides to flush (flush_by_time -// true, since nothing has touched the clock yet) but loses the try_lock race (we hold m_flush_mtx) -- -// that loss is what queues the retry. -// 3. Arm the "test_touch_last_flush_time" flip and call test_touch_last_flush_time() -- resets the clock -// to *now*, exactly as a concurrent flush completing would, but with the lock still held by us and -// nothing actually flushed. -// 4. Keep holding m_flush_mtx for 200ms, with max_time_between_flush_us configured far larger (2s) than -// that hold. Every retry landing anywhere in this window sees a still-too-recent clock (flush_by_time -// false) and a pending size still under threshold_size (flush_by_size false) -- the exact condition -// the fix targets, held open long enough that timing luck can't save a retry that isn't robust to it. -// 5. Release m_flush_mtx via a plain unlock (not flush()) -- this does not touch m_log_idx or complete -// anything on its own. Pre-fix, whatever retry was dropped during step 4's window is gone for good, -// and the write never completes. Post-fix, force=true has kept retrying the whole time and now wins -// the lock, calling a real flush() that (for the first time) takes a snapshot including the target -// write and completes it. +// 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); From 7ac8ce411499ce7598ad34290afda41b2d8097f5 Mon Sep 17 00:00:00 2001 From: Mehdi Hosseini Date: Fri, 11 Sep 2026 14:59:48 -0700 Subject: [PATCH 5/6] bump version and clang --- conanfile.py | 2 +- .../homestore/btree/detail/btree_internal.hpp | 36 ++++----- src/include/homestore/meta_service.hpp | 2 +- src/lib/blkalloc/varsize_blk_allocator.h | 5 +- src/lib/checkpoint/cp_mgr.cpp | 12 ++- src/lib/common/homestore_assert.hpp | 21 +++-- src/lib/device/virtual_dev.hpp | 2 +- src/lib/logstore/log_store_service.cpp | 10 +-- src/lib/replication/repl_dev/solo_repl_dev.h | 4 +- .../replication/service/generic_repl_svc.cpp | 2 +- .../replication/service/raft_repl_service.cpp | 45 +++++------ src/tests/test_index_gc.cpp | 81 +++++++++---------- 12 files changed, 99 insertions(+), 123 deletions(-) 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_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"); } From a4a9802b559fa954532e2e1d62d306dadba53bda Mon Sep 17 00:00:00 2001 From: Mehdi Hosseini Date: Fri, 11 Sep 2026 15:07:26 -0700 Subject: [PATCH 6/6] logstore: fix an overclaim in the stack-overflow comment "held for just ~200ms" implied a verified lower bound; only 200ms was actually tested, not shorter durations. Reword to state what was actually confirmed. --- src/lib/logstore/log_dev.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/lib/logstore/log_dev.cpp b/src/lib/logstore/log_dev.cpp index 1a7b535cd..81caeb488 100644 --- a/src/lib/logstore/log_dev.cpp +++ b/src/lib/logstore/log_dev.cpp @@ -482,7 +482,7 @@ bool LogDev::flush_if_necessary(int64_t threshold_size, bool force) { // 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 - // just ~200ms; this predates force -- it's already present in the plain reschedule above). + // ~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,