azalea/kotlin-fork
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[K/N] Handle clock wrapping in RepeatedTimer. ^KT-48537

Browse Source
This commit is contained in:
Alexander Shabalin
2022-02-10 23:13:58 +03:00
committed by Space
parent eae9d05f50
commit e014e6b154
3 changed files with 146 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kotlin-native/runtime/src/gc/common/cpp/GCScheduler.cpp
+3 -3
View File
@@ -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<void()> scheduleGC_;
RepeatedTimer timer_;
RepeatedTimer<> timer_;
};
#endif // !KONAN_NO_THREADS
kotlin-native/runtime/src/main/cpp/RepeatedTimer.hpp
+34 -14
View File
@@ -7,23 +7,25 @@
#ifndef KONAN_NO_THREADS
#include <chrono>
#include <condition_variable>
#include <mutex>
#include <string_view>
#include "Clock.hpp"
#include "KAssert.h"
#include "ScopedThread.hpp"
#include "Utils.hpp"
namespace kotlin {
template <typename Clock = steady_clock>
class RepeatedTimer : private Pinned {
public:
template <typename Rep, typename Period, typename F>
RepeatedTimer(std::string_view name, std::chrono::duration<Rep, Period> interval, F&& f) noexcept :
thread_(ScopedThread::attributes().name(name), &RepeatedTimer::Run<Rep, Period, F>, this, std::move(interval), std::forward<F>(f)) {
}
interval_(interval),
next_(Clock::now() + interval_),
thread_(ScopedThread::attributes().name(name), &RepeatedTimer::Run<F>, this, std::forward<F>(f)) {}
template <typename Rep, typename Period, typename F>
RepeatedTimer(std::chrono::duration<Rep, Period> 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 <typename Rep, typename Period>
void restart(std::chrono::duration<Rep, Period> interval) noexcept {
{
std::unique_lock lock(mutex_);
interval_ = interval;
next_ = Clock::now() + interval_;
scheduledInterrupt_ = true;
}
wait_.notify_all();
}
private:
template <typename Rep, typename Period, typename F>
void Run(std::chrono::duration<Rep, Period> 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 <typename F>
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<std::chrono::duration<Rep, Period>>(newInterval);
// The function must be executed in the unlocked environment.
lock.unlock();
std::invoke(std::forward<F>(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<Clock> next_;
bool scheduledInterrupt_ = false;
ScopedThread thread_;
};
kotlin-native/runtime/src/main/cpp/RepeatedTimerTest.cpp
+109 -48
View File
@@ -10,63 +10,124 @@
#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include "ClockTestSupport.hpp"
using namespace kotlin;
TEST(RepeatedTimerTest, WillNotExecuteImmediately) {
std::atomic<int> 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<testing::MockFunction<void()>> f;
RepeatedTimer<test_support::manual_clock> timer(minutes(10), f.AsStdFunction());
}
TEST(RepeatedTimerTest, WillRun) {
std::atomic<int> 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<testing::MockFunction<void()>> f;
RepeatedTimer<test_support::manual_clock> 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<int> counter = 0;
TEST_F(RepeatedTimerTest, WillStopInDestructor) {
testing::StrictMock<testing::MockFunction<void()>> f;
std::promise<int> promise;
std::future<int> 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<test_support::manual_clock> 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<int> 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::milliseconds>(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<testing::MockFunction<void()>> f;
RepeatedTimer<test_support::manual_clock> 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<testing::MockFunction<void()>> f;
RepeatedTimer<test_support::manual_clock> 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<testing::MockFunction<void()>> f;
RepeatedTimer<test_support::manual_clock> 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));
}