From 74c6f96c49212c0b385acd952b66c45da5211511 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 8 Sep 2026 11:53:36 +0200 Subject: [PATCH] fix race Signed-off-by: Konstantin Morozov --- src/Common/FailPoint.cpp | 4 +- .../ContentAddressed/Gc/CasGcScheduler.cpp | 75 +++++++++++++++---- .../ContentAddressed/Gc/CasGcScheduler.h | 17 ++++- src/Disks/tests/gtest_cas_gc_stop_start.cpp | 44 +++++++++++ 4 files changed, 120 insertions(+), 20 deletions(-) diff --git a/src/Common/FailPoint.cpp b/src/Common/FailPoint.cpp index b05271d4f860..19a646648bdc 100644 --- a/src/Common/FailPoint.cpp +++ b/src/Common/FailPoint.cpp @@ -241,7 +241,9 @@ static struct InitFiu REGULAR(cas_relink_receiver_force_mechanism_failure) \ PAUSEABLE_ONCE(cas_relink_receiver_pause_before_confirm) \ REGULAR(cas_relink_sender_omit_pool_cookie) \ - REGULAR(cas_relink_receiver_drop_forced_disk) + REGULAR(cas_relink_receiver_drop_forced_disk) \ + ONCE(cas_gc_scheduler_fail_before_heartbeat_worker_start) \ + ONCE(cas_gc_scheduler_fail_before_worker_start) namespace FailPoints { diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp index 610d13879779..e0dee3413310 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp @@ -3,11 +3,13 @@ #include #include #include +#include #include #include #include #include #include +#include #include #include #include @@ -20,6 +22,13 @@ namespace DB::ErrorCodes extern const int TIMEOUT_EXCEEDED; extern const int SOCKET_TIMEOUT; extern const int MEMORY_LIMIT_EXCEEDED; + extern const int FAULT_INJECTED; +} + +namespace DB::FailPoints +{ + extern const char cas_gc_scheduler_fail_before_heartbeat_worker_start[]; + extern const char cas_gc_scheduler_fail_before_worker_start[]; } namespace DB::Cas @@ -90,19 +99,53 @@ CasGcScheduler::~CasGcScheduler() void CasGcScheduler::start() { - std::lock_guard lock(mutex); - if (thread.joinable()) - return; - stopping = false; - thread = ThreadFromGlobalPool([this] { loop(); }); - hb_thread = ThreadFromGlobalPool([this] { heartbeatLoop(); }); + std::lock_guard threads_lock(threads_mutex); + { + std::lock_guard lock(mutex); + if (scheduler_state == SchedulerState::Running) + return; + scheduler_state = SchedulerState::Running; + } + try + { + fiu_do_on(FailPoints::cas_gc_scheduler_fail_before_heartbeat_worker_start, + { + throw Exception(ErrorCodes::FAULT_INJECTED, "Injected failure before starting CAS GC heartbeat worker"); + }); + hb_thread = ThreadFromGlobalPool([this] { heartbeatLoop(); }); + fiu_do_on(FailPoints::cas_gc_scheduler_fail_before_worker_start, + { + throw Exception(ErrorCodes::FAULT_INJECTED, "Injected failure before starting CAS GC worker"); + }); + thread = ThreadFromGlobalPool([this] { loop(); }); + } + catch (...) + { + { + std::lock_guard lock(mutex); + scheduler_state = SchedulerState::Stopped; + } + wake.notify_all(); + if (thread.joinable()) + thread.join(); + if (hb_thread.joinable()) + hb_thread.join(); + i_am_leader.store(false, std::memory_order_relaxed); + throw; + } } void CasGcScheduler::stop() { + std::lock_guard threads_lock(threads_mutex); { std::lock_guard lock(mutex); - stopping = true; + if (scheduler_state == SchedulerState::Stopped) + { + i_am_leader.store(false, std::memory_order_relaxed); + return; + } + scheduler_state = SchedulerState::Stopped; } wake.notify_all(); if (thread.joinable()) @@ -122,7 +165,7 @@ void CasGcScheduler::requestRoundSoon() { { std::lock_guard lock(mutex); - if (stopping || !thread.joinable()) + if (scheduler_state != SchedulerState::Running) return; round_requested = true; } @@ -299,9 +342,10 @@ void CasGcScheduler::loop() while (true) { { - std::unique_lock lock(mutex); - wake.wait_for(lock, interval, [this] { return stopping || round_requested; }); - if (stopping) + UniqueLock lock(mutex); + wake.wait_for(lock.getUnderlyingLock(), interval, [this]() TSA_NO_THREAD_SAFETY_ANALYSIS + { return scheduler_state == SchedulerState::Stopped || round_requested; }); + if (scheduler_state == SchedulerState::Stopped) return; round_requested = false; } @@ -332,9 +376,9 @@ void CasGcScheduler::loop() } try { - /// LOW/benign: if stop() flips `stopping` while we're blocked here (a concurrent manual + /// LOW/benign: if stop() flips `scheduler_state` while we're blocked here (a concurrent manual /// round holds gc_round_mutex), we still run one more Scheduled round once it unblocks, - /// before the next wait_for() observes `stopping` - an accepted extra round, not a + /// before the next wait_for() observes `scheduler_state` - an accepted extra round, not a /// correctness issue. std::lock_guard round_lock(gc_round_mutex); @@ -427,8 +471,9 @@ void CasGcScheduler::heartbeatLoop() while (true) { { - std::unique_lock lock(mutex); - if (wake.wait_for(lock, hb_interval, [this] { return stopping; })) + UniqueLock lock(mutex); + if (wake.wait_for(lock.getUnderlyingLock(), hb_interval, [this]() TSA_NO_THREAD_SAFETY_ANALYSIS + { return scheduler_state == SchedulerState::Stopped; })) return; } /// rev.7 §3 [C1] + rev.8 §9 item 8: self-exit on ANY terminal (or FORGET-intent) pool, same as diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h index 80b50f2fc448..cebd691dd76a 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h @@ -3,6 +3,7 @@ #include #include #include +#include #include #include #include @@ -169,6 +170,12 @@ class CasGcScheduler bool waitForTerminalSelfExitForTest(std::chrono::milliseconds timeout); private: + enum class SchedulerState + { + Stopped, + Running + }; + /// Waits for the configured interval, runs scheduled rounds while the scheduler is active, and /// logs exceptions before continuing with the next tick. The round lock serializes this worker /// with `runOneRoundNow` because the persistent `gc` object is not thread-safe. @@ -210,15 +217,17 @@ class CasGcScheduler /// the round so stop()/heartbeatLoop are not blocked, so the round cannot hold `mutex`. std::mutex gc_round_mutex; + std::mutex threads_mutex; + ThreadFromGlobalPool thread TSA_GUARDED_BY(threads_mutex); + ThreadFromGlobalPool hb_thread TSA_GUARDED_BY(threads_mutex); + std::mutex mutex; std::condition_variable wake; - bool stopping = false; - bool round_requested = false; /// guarded by `mutex`; coalesced external wake request - ThreadFromGlobalPool thread; + SchedulerState scheduler_state TSA_GUARDED_BY(mutex) = SchedulerState::Stopped; + bool round_requested TSA_GUARDED_BY(mutex) = false; /// coalesced external wake request /// Set by the round worker and read by the heartbeat worker. It is only an in-process hint: the /// durable lease remains the authority, and a failed round clears the hint before retrying. std::atomic i_am_leader{false}; - ThreadFromGlobalPool hb_thread; /// Set true for the whole body of one round (`runRoundLogged`, held across the `gc_round_mutex` /// critical section a scheduled or manual round runs under) and cleared when it returns, on the diff --git a/src/Disks/tests/gtest_cas_gc_stop_start.cpp b/src/Disks/tests/gtest_cas_gc_stop_start.cpp index 6079a2b3436d..485862c84e96 100644 --- a/src/Disks/tests/gtest_cas_gc_stop_start.cpp +++ b/src/Disks/tests/gtest_cas_gc_stop_start.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include @@ -34,9 +35,16 @@ namespace DB::ErrorCodes { +extern const int FAULT_INJECTED; extern const int INVALID_STATE; } +namespace DB::FailPoints +{ +extern const char cas_gc_scheduler_fail_before_heartbeat_worker_start[]; +extern const char cas_gc_scheduler_fail_before_worker_start[]; +} + using namespace DB; using DB::Cas::CasGcScheduler; using DB::Cas::GcRoundLogRecord; @@ -342,6 +350,42 @@ TEST(CASGCStopStart, StopAndStartAreIdempotent) sched.stop(); } +TEST(CASGCStopStart, StopClearsLeadershipAfterManualRoundWithoutStart) +{ + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + CasGcScheduler sched(store, std::chrono::seconds(3600), "CasGcManualStopTest", "ca-disk"); + + const RoundReport report = sched.runOneRoundNow(); + ASSERT_TRUE(report.acquired_lease); + ASSERT_TRUE(sched.gcHealth().is_leader); + + sched.stop(); + EXPECT_FALSE(sched.gcHealth().is_leader); +} + +TEST(CASGCStopStart, StartFailureRollsBackAndCanBeRetried) +{ + for (const char * failpoint : + {FailPoints::cas_gc_scheduler_fail_before_heartbeat_worker_start, + FailPoints::cas_gc_scheduler_fail_before_worker_start}) + { + SCOPED_TRACE(failpoint); + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + CasGcScheduler sched(store, std::chrono::seconds(3600), "CasGcStartFailureTest", "ca-disk"); + + FailPointInjection::enableFailPoint(failpoint); + Cas::tests::expectThrowsCode(ErrorCodes::FAULT_INJECTED, [&] { sched.start(); }); + FailPointInjection::disableFailPoint(failpoint); + EXPECT_TRUE(sched.isQuiescent()); + + EXPECT_NO_THROW(sched.start()); + sched.stop(); + EXPECT_TRUE(sched.isQuiescent()); + } +} + /// (d) START refuses on a Vanished disk with the typed 668 (`INVALID_STATE`) error -- restarting GC on a /// decommissioned pool is meaningless and would only spin failing rounds -- while STOP on the SAME /// Vanished disk (with a live scheduler present) SUCCEEDS: stopping the reclaimer on a sick disk is a