diff --git a/cloud/src/common/bvars.h b/cloud/src/common/bvars.h index 7a02be09ae058d..bbd1f10adc5b5b 100644 --- a/cloud/src/common/bvars.h +++ b/cloud/src/common/bvars.h @@ -19,6 +19,7 @@ #include #include +#include #include #include #include @@ -30,7 +31,6 @@ #include #include -#include #include #include #include @@ -39,6 +39,7 @@ #include #include #include +#include #include "common/logging.h" @@ -345,17 +346,15 @@ class MBvarLatencyRecorderWithStatus { * @return true if the timer was successfully started, false otherwise */ bool start() { - if (!_started.load()) { - { - std::lock_guard l(init_mutex_); - if (!_started.load()) { - if (!schedule()) { - return false; - } - _started.store(true); - } - return true; - } + std::lock_guard l(lifecycle_mutex_); + if (_started) { + return true; + } + + _started = true; + if (!schedule_locked()) { + _started = false; + return false; } return true; } @@ -365,12 +364,14 @@ class MBvarLatencyRecorderWithStatus { * Scheduling a one-time task. * This is useful if you want to reset the timer interval. */ - bool schedule() { + // lifecycle_mutex_ must be held so stop() cannot race with replacing _timer. + bool schedule_locked() { if (bthread_timer_add(&_timer, butil::seconds_from_now(_interval_s), update, this) != 0) { LOG(WARNING) << "Failed to add bthread timer for ScheduledLatencyUpdater"; return false; } + _callback_pending = true; return true; } @@ -384,8 +385,8 @@ class MBvarLatencyRecorderWithStatus { */ static void update(void* arg) { auto* latency_updater = static_cast(arg); - if (!latency_updater || !latency_updater->_started) { - LOG(WARNING) << "Invalid ScheduledLatencyUpdater in timer callback"; + CHECK(latency_updater != nullptr); + if (!latency_updater->begin_callback()) { return; } @@ -395,53 +396,48 @@ class MBvarLatencyRecorderWithStatus { auto* parent = static_cast(latency_updater->_arg); if (!parent) { LOG(WARNING) << "Invalid parent container in timer callback"; - return; - } - - std::list current_dim_list; - { - std::lock_guard l(parent->recorder_mutex_); - for (const auto& it : parent->recorder_) { - if (it.second.get() == latency_updater) { - current_dim_list = it.first; - break; + } else { + std::list current_dim_list; + { + std::lock_guard l(parent->recorder_mutex_); + for (const auto& it : parent->recorder_) { + if (it.second.get() == latency_updater) { + current_dim_list = it.first; + break; + } } } - } - - if (current_dim_list.empty()) { - LOG(WARNING) << "Could not find dimension for ScheduledLatencyUpdater"; - return; - } - - { - std::lock_guard l(parent->timer_mutex_); - - bvar::Status* max_status = parent->max_status_.get_stats(current_dim_list); - bvar::Status* avg_status = parent->avg_status_.get_stats(current_dim_list); - bvar::Status* count_status = - parent->count_status_.get_stats(current_dim_list); - - VLOG_DEBUG << "Updating latency recorder status for dimension, " - << "max_latency: " << latency_updater->max_latency() - << ", avg_latency: " << latency_updater->latency(); - TEST_SYNC_POINT("mBvarLatencyRecorderWithStatus::update"); - if (max_status) { - max_status->set_value(latency_updater->max_latency()); - } - if (avg_status) { - avg_status->set_value(latency_updater->latency()); - } - if (count_status) { - count_status->set_value(latency_updater->count()); + if (current_dim_list.empty()) { + LOG(WARNING) << "Could not find dimension for ScheduledLatencyUpdater"; + } else { + std::lock_guard l(parent->timer_mutex_); + + bvar::Status* max_status = + parent->max_status_.get_stats(current_dim_list); + bvar::Status* avg_status = + parent->avg_status_.get_stats(current_dim_list); + bvar::Status* count_status = + parent->count_status_.get_stats(current_dim_list); + + VLOG_DEBUG << "Updating latency recorder status for dimension, " + << "max_latency: " << latency_updater->max_latency() + << ", avg_latency: " << latency_updater->latency(); + TEST_SYNC_POINT("mBvarLatencyRecorderWithStatus::update"); + + if (max_status) { + max_status->set_value(latency_updater->max_latency()); + } + if (avg_status) { + avg_status->set_value(latency_updater->latency()); + } + if (count_status) { + count_status->set_value(latency_updater->count()); + } } } - if (latency_updater->_started && !latency_updater->schedule()) { - LOG(WARNING) << "Failed to reschedule timer for ScheduledLatencyUpdater"; - latency_updater->_started = false; - } + latency_updater->finish_callback(); } /** @@ -451,18 +447,64 @@ class MBvarLatencyRecorderWithStatus { * any pending callbacks from accessing potentially freed resources. */ void stop() { - if (_started.load()) { - bthread_timer_del(_timer); - _started = false; + std::unique_lock l(lifecycle_mutex_); + if (!_started && !_callback_pending) { + return; + } + + // Prevent a running callback from scheduling the next timer before trying to + // cancel the current one. bthread_timer_del() returns 1 when the callback is + // already running; in that case the updater and its parent must stay alive until + // finish_callback() signals that the callback no longer accesses either object. + _started = false; + if (!_callback_pending) { + return; + } + + const int timer_state = bthread_timer_del(_timer); + if (timer_state == 0) { + _callback_pending = false; + return; + } + + CHECK_EQ(1, timer_state); + TEST_SYNC_POINT("mBvarLatencyRecorderWithStatus::stop"); + while (_callback_pending) { + callback_finished_.wait(l); } } private: - int _interval_s; // Timer interval in seconds - void* _arg; // Argument to pass to the callback - bthread_timer_t _timer; // The bthread timer handle - std::atomic_bool _started {false}; // Whether the timer has been started - bthread::Mutex init_mutex_; // Mutex for timer_map_ + bool begin_callback() { + std::lock_guard l(lifecycle_mutex_); + if (_started) { + return true; + } + + // stop() may observe the timer as running before this function acquires the + // lifecycle mutex. Acknowledge the canceled callback so stop() can finish. + _callback_pending = false; + callback_finished_.notify_all(); + return false; + } + + void finish_callback() { + std::lock_guard l(lifecycle_mutex_); + _callback_pending = false; + if (_started && !schedule_locked()) { + LOG(WARNING) << "Failed to reschedule timer for ScheduledLatencyUpdater"; + _started = false; + } + callback_finished_.notify_all(); + } + + int _interval_s; // Timer interval in seconds + void* _arg; // Argument to pass to the callback + bthread_timer_t _timer; // The bthread timer handle + bool _started = false; // Whether callbacks should keep running + bool _callback_pending = false; // A timer callback is scheduled or running + bthread::Mutex lifecycle_mutex_; + bthread::ConditionVariable callback_finished_; }; public: @@ -483,6 +525,24 @@ class MBvarLatencyRecorderWithStatus { const std::initializer_list& dim_names) : MBvarLatencyRecorderWithStatus(prefix + "_" + metric_name, dim_names) {} + ~MBvarLatencyRecorderWithStatus() { + // Members are destroyed in reverse declaration order, so recorder_ (which owns the + // timer callbacks) would otherwise be destroyed after the status bvars and mutexes + // used by those callbacks. Stop and drain every callback while all parent members are + // still alive. + std::vector> latency_updaters; + { + std::lock_guard l(recorder_mutex_); + latency_updaters.reserve(recorder_.size()); + for (const auto& entry : recorder_) { + latency_updaters.push_back(entry.second); + } + } + for (const auto& latency_updater : latency_updaters) { + latency_updater->stop(); + } + } + /** * @brief Record a latency value * diff --git a/cloud/test/bvars_test.cpp b/cloud/test/bvars_test.cpp index 95062465a4ef65..d82820823f8800 100644 --- a/cloud/test/bvars_test.cpp +++ b/cloud/test/bvars_test.cpp @@ -19,6 +19,8 @@ #include #include +#include +#include #include #include @@ -49,11 +51,17 @@ int main(int argc, char** argv) { class BvarsTest : public ::testing::Test { public: void SetUp() override { + auto* sp = SyncPoint::get_instance(); + sp->disable_processing(); + sp->clear_all_call_backs(); if (server.Start("0.0.0.0:0", &options) == -1) { perror("Start brpc server"); } } void TearDown() override { + auto* sp = SyncPoint::get_instance(); + sp->disable_processing(); + sp->clear_all_call_backs(); server.Stop(0); server.Join(); } @@ -126,4 +134,44 @@ TEST(BvarsTest, MultiThreadRecordMetrics) { ASSERT_GT(update_count.load(), 200); } -} // namespace doris::cloud \ No newline at end of file +TEST(BvarsTest, DestructorWaitsForRunningUpdate) { + constexpr int callback_timeout_s = 5; + int interval_s = 1; + bthread::CountdownEvent update_started; + bthread::CountdownEvent allow_update_to_finish; + bthread::CountdownEvent stop_waiting; + + auto* sp = SyncPoint::get_instance(); + sp->set_call_back("mBvarLatencyRecorderWithStatus::put", [&interval_s](auto&& args) { + auto* interval = try_any_cast(args[0]); + *interval = interval_s; + }); + sp->set_call_back("mBvarLatencyRecorderWithStatus::update", [&](auto&&) { + update_started.signal(); + allow_update_to_finish.wait(); + }); + sp->set_call_back("mBvarLatencyRecorderWithStatus::stop", + [&](auto&&) { stop_waiting.signal(); }); + sp->enable_processing(); + + auto recorder = std::make_unique>( + "destructor_wait_test", std::initializer_list {"instance_id"}); + recorder->put({"instance"}, 1); + const int update_result = + update_started.timed_wait(butil::seconds_from_now(callback_timeout_s)); + if (update_result != 0) { + // Keep cleanup non-blocking even when the timer thread did not invoke the callback. + allow_update_to_finish.signal(); + EXPECT_EQ(0, update_result); + return; + } + + std::thread destroyer([&] { recorder.reset(); }); + const int stop_result = stop_waiting.timed_wait(butil::seconds_from_now(callback_timeout_s)); + allow_update_to_finish.signal(); + destroyer.join(); + + EXPECT_EQ(0, stop_result); +} + +} // namespace doris::cloud