From e014e6b15493f4bccefe60d96e6023dd95ea4ca0 Mon Sep 17 00:00:00 2001 From: Alexander Shabalin Date: Thu, 10 Feb 2022 23:13:58 +0300 Subject: [PATCH] [K/N] Handle clock wrapping in RepeatedTimer. ^KT-48537 --- .../runtime/src/gc/common/cpp/GCScheduler.cpp | 6 +- .../runtime/src/main/cpp/RepeatedTimer.hpp | 48 ++++-- .../src/main/cpp/RepeatedTimerTest.cpp | 157 ++++++++++++------ 3 files changed, 146 insertions(+), 65 deletions(-) diff --git a/kotlin-native/runtime/src/gc/common/cpp/GCScheduler.cpp b/kotlin-native/runtime/src/gc/common/cpp/GCScheduler.cpp index 144510ba1c0..01f9a0d4436 100644 --- a/kotlin-native/runtime/src/gc/common/cpp/GCScheduler.cpp +++ b/kotlin-native/runtime/src/gc/common/cpp/GCScheduler.cpp @@ -106,11 +106,10 @@ public: heapGrowthController_(config), regularIntervalPacer_(config, currentTimeProvider), scheduleGC_(std::move(scheduleGC)), - timer_("GC Timer thread", config_.regularGcInterval(), [this]() { + timer_("GC Timer thread", config_.regularGcInterval(), [this] { if (regularIntervalPacer_.NeedsGC()) { scheduleGC_(); } - return config_.regularGcInterval(); }) {} void UpdateFromThreadData(gc::GCSchedulerThreadData& threadData) noexcept override { @@ -123,6 +122,7 @@ public: void OnPerformFullGC() noexcept override { heapGrowthController_.OnPerformFullGC(); regularIntervalPacer_.OnPerformFullGC(); + timer_.restart(config_.regularGcInterval()); } void UpdateAliveSetBytes(size_t bytes) noexcept override { heapGrowthController_.UpdateAliveSetBytes(bytes); } @@ -132,7 +132,7 @@ private: HeapGrowthController heapGrowthController_; RegularIntervalPacer regularIntervalPacer_; std::function scheduleGC_; - RepeatedTimer timer_; + RepeatedTimer<> timer_; }; #endif // !KONAN_NO_THREADS diff --git a/kotlin-native/runtime/src/main/cpp/RepeatedTimer.hpp b/kotlin-native/runtime/src/main/cpp/RepeatedTimer.hpp index 51c1264675b..2fc8b3896d7 100644 --- a/kotlin-native/runtime/src/main/cpp/RepeatedTimer.hpp +++ b/kotlin-native/runtime/src/main/cpp/RepeatedTimer.hpp @@ -7,23 +7,25 @@ #ifndef KONAN_NO_THREADS -#include #include #include #include +#include "Clock.hpp" #include "KAssert.h" #include "ScopedThread.hpp" #include "Utils.hpp" namespace kotlin { +template class RepeatedTimer : private Pinned { public: template RepeatedTimer(std::string_view name, std::chrono::duration interval, F&& f) noexcept : - thread_(ScopedThread::attributes().name(name), &RepeatedTimer::Run, this, std::move(interval), std::forward(f)) { - } + interval_(interval), + next_(Clock::now() + interval_), + thread_(ScopedThread::attributes().name(name), &RepeatedTimer::Run, this, std::forward(f)) {} template RepeatedTimer(std::chrono::duration interval, F&& f) noexcept : @@ -33,29 +35,47 @@ public: { std::unique_lock lock(mutex_); run_ = false; + scheduledInterrupt_ = true; + } + wait_.notify_all(); + // Make sure we wait for the thread to finish before starting to destroy the fields. + thread_.join(); + } + + template + void restart(std::chrono::duration interval) noexcept { + { + std::unique_lock lock(mutex_); + interval_ = interval; + next_ = Clock::now() + interval_; + scheduledInterrupt_ = true; } wait_.notify_all(); } private: - template - void Run(std::chrono::duration interval, F f) noexcept { - while (true) { - std::unique_lock lock(mutex_); - if (wait_.wait_for(lock, interval, [this]() noexcept { return !run_; })) { - RuntimeAssert(!run_, "Can only happen once run_ is set to false"); - return; + template + void Run(F&& f) noexcept { + std::unique_lock lock(mutex_); + while (run_) { + scheduledInterrupt_ = false; + if (Clock::wait_until(wait_, lock, next_, [this] { return scheduledInterrupt_; })) { + continue; } - RuntimeAssert(run_, "Can only happen if we timed out on waiting and run_ is still true"); - auto newInterval = f(); - // The next waiting will use the new interval. - interval = std::chrono::duration_cast>(newInterval); + // The function must be executed in the unlocked environment. + lock.unlock(); + std::invoke(std::forward(f)); + lock.lock(); + next_ = Clock::now() + interval_; } } std::mutex mutex_; std::condition_variable wait_; bool run_ = true; + typename Clock::duration interval_; + std::chrono::time_point next_; + bool scheduledInterrupt_ = false; ScopedThread thread_; }; diff --git a/kotlin-native/runtime/src/main/cpp/RepeatedTimerTest.cpp b/kotlin-native/runtime/src/main/cpp/RepeatedTimerTest.cpp index 8ca494e07e0..9d01a2cf249 100644 --- a/kotlin-native/runtime/src/main/cpp/RepeatedTimerTest.cpp +++ b/kotlin-native/runtime/src/main/cpp/RepeatedTimerTest.cpp @@ -10,63 +10,124 @@ #include "gmock/gmock.h" #include "gtest/gtest.h" +#include "ClockTestSupport.hpp" + using namespace kotlin; -TEST(RepeatedTimerTest, WillNotExecuteImmediately) { - std::atomic counter = 0; - RepeatedTimer timer(std::chrono::minutes(10), [&counter]() { - ++counter; - return std::chrono::minutes(10); - }); - // The function is not executed immediately. - EXPECT_THAT(counter.load(), 0); +class RepeatedTimerTest : public testing::Test { +public: + RepeatedTimerTest() { test_support::manual_clock::reset(); } +}; + +TEST_F(RepeatedTimerTest, WillNotExecuteImmediately) { + testing::StrictMock> f; + RepeatedTimer timer(minutes(10), f.AsStdFunction()); } -TEST(RepeatedTimerTest, WillRun) { - std::atomic counter = 0; - RepeatedTimer timer(std::chrono::milliseconds(10), [&counter]() { - ++counter; - return std::chrono::milliseconds(10); - }); - // Wait until the counter increases at least twice. - while (counter < 2) { - std::this_thread::sleep_for(std::chrono::milliseconds(10)); - } +TEST_F(RepeatedTimerTest, WillRun) { + testing::StrictMock> f; + RepeatedTimer timer(minutes(10), f.AsStdFunction()); + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(10)); + + EXPECT_CALL(f, Call()); + test_support::manual_clock::sleep_for(minutes(10)); + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(10)); + testing::Mock::VerifyAndClearExpectations(&f); + + EXPECT_CALL(f, Call()); + test_support::manual_clock::sleep_for(minutes(10)); + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(10)); + testing::Mock::VerifyAndClearExpectations(&f); } -TEST(RepeatedTimerTest, WillStopInDestructor) { - std::atomic counter = 0; +TEST_F(RepeatedTimerTest, WillStopInDestructor) { + testing::StrictMock> f; + std::promise promise; + std::future future = promise.get_future(); { - RepeatedTimer timer(std::chrono::milliseconds(1), [&counter]() { - // This lambda will only get executed once. - EXPECT_THAT(counter.load(), 0); - ++counter; - return std::chrono::minutes(10); - }); + RepeatedTimer timer(minutes(10), f.AsStdFunction()); // Wait until the counter increases once. - while (counter < 1) { - std::this_thread::sleep_for(std::chrono::milliseconds(10)); - } + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(10)); + + EXPECT_CALL(f, Call()).WillOnce([&] { promise.set_value_at_thread_exit(42); }); + test_support::manual_clock::sleep_for(minutes(10)); + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(10)); + testing::Mock::VerifyAndClearExpectations(&f); } - // The destructor was fired and cancelled the timer without executing the function. - EXPECT_THAT(counter.load(), 1); + future.wait(); + // The thread has definitely finished. + EXPECT_THAT(future.get(), 42); + // Timer is gone, so nothing can be pending. + EXPECT_THAT(test_support::manual_clock::pending(), std::nullopt); } -TEST(RepeatedTimerTest, AdjustInterval) { - std::atomic counter = 0; - RepeatedTimer timer(std::chrono::milliseconds(1), [&counter]() { - ++counter; - if (counter < 2) { - return std::chrono::milliseconds(1); - } else { - return std::chrono::duration_cast(std::chrono::minutes(10)); - } - }); - // Wait until counter grows to 2, when the waiting time changes to 10 minutes. - while (counter < 2) { - } - EXPECT_THAT(counter.load(), 2); - std::this_thread::sleep_for(std::chrono::milliseconds(10)); - // After we've slept for 10ms, we still haven't executed the function another time. - EXPECT_THAT(counter.load(), 2); +TEST_F(RepeatedTimerTest, InfiniteInterval) { + test_support::manual_clock::reset(test_support::manual_clock::time_point::max() - hours(365 * 24)); + + constexpr auto infinite = test_support::manual_clock::duration::max(); + testing::StrictMock> f; + RepeatedTimer timer(infinite, f.AsStdFunction()); + // test_support::manual_clock will wait the absolute maximum time. + test_support::manual_clock::waitForPending(test_support::manual_clock::time_point::max()); +} + +TEST_F(RepeatedTimerTest, Restart) { + test_support::manual_clock::reset(test_support::manual_clock::time_point::max() - hours(365 * 24)); + + testing::StrictMock> f; + RepeatedTimer timer(minutes(10), f.AsStdFunction()); + // Wait until the clock starts waiting. + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(10)); + + // Now restart the timer to fire in 10 seconds instead. + timer.restart(seconds(10)); + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + seconds(10)); + + // Now wait until task triggers once and is scheduled again with the same interval. + EXPECT_CALL(f, Call()); + test_support::manual_clock::sleep_for(seconds(10)); + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + seconds(10)); + testing::Mock::VerifyAndClearExpectations(&f); + + // Wait 5 seconds and restart the timer to fire in 1 minute time. + test_support::manual_clock::sleep_for(seconds(5)); + timer.restart(minutes(1)); + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(1)); + + // And immediately restart the timer again to run in 5 minutes instead. + timer.restart(minutes(5)); + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(5)); + + // And just restart the timer to fire after "infinite" interval. + timer.restart(test_support::manual_clock::duration::max()); + test_support::manual_clock::waitForPending(test_support::manual_clock::time_point::max()); + + // Wait for an hour and restart the timer to fire every 10 minutes. + test_support::manual_clock::sleep_for(hours(1)); + timer.restart(minutes(10)); + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(10)); + + // Wait for 5 minutes, and restart the timer keeping the same interval. + test_support::manual_clock::sleep_for(minutes(5)); + timer.restart(minutes(10)); + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(10)); +} + +TEST_F(RepeatedTimerTest, RestartFromTask) { + testing::StrictMock> f; + RepeatedTimer timer(minutes(10), f.AsStdFunction()); + + // Wait until the clock starts waiting. + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(10)); + + // Make the task execute once and restart the task to fire every minute. + EXPECT_CALL(f, Call()).WillOnce([&] { timer.restart(minutes(1)); }); + test_support::manual_clock::sleep_for(minutes(10)); + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(1)); + testing::Mock::VerifyAndClear(&f); + + // Now wait for that minute and make sure the next task is also scheduled 1 minute from now. + EXPECT_CALL(f, Call()); + test_support::manual_clock::sleep_for(minutes(1)); + test_support::manual_clock::waitForPending(test_support::manual_clock::now() + minutes(1)); }