From 5243ab00c213fb4d5c8feaf35f98e5f69470adc6 Mon Sep 17 00:00:00 2001 From: "matej.gradicek" Date: Wed, 5 Apr 2017 08:27:13 +0000 Subject: [PATCH] Scheduler and tests. Reviewers: dgleich, buda Reviewed By: dgleich, buda Subscribers: pullbot, matej.gradicek Differential Revision: https://phabricator.memgraph.io/D213 --- src/utils/scheduler.hpp | 83 ++++++++++++++++++++++++++++++++++++++++ tests/unit/scheduler.cpp | 47 +++++++++++++++++++++++ 2 files changed, 130 insertions(+) create mode 100644 src/utils/scheduler.hpp create mode 100644 tests/unit/scheduler.cpp diff --git a/src/utils/scheduler.hpp b/src/utils/scheduler.hpp new file mode 100644 index 000000000..0d90cc606 --- /dev/null +++ b/src/utils/scheduler.hpp @@ -0,0 +1,83 @@ +#include +#include +#include +#include +#include + +/** + * Class used to run given std::function. Function is ran and then sleep + * is called with parameter given in class constructor. Class is templated with + * mutex class TMutex which is used to synchronize threads. Default template + * value is std::mutex. + */ +template +class Scheduler { + public: + /** + * @param t - Time between executing function. If function is still running + * when it should be ran again, it will not be ran and next + * start time will be increased current time plus t. If t equals + * -1, scheduler will not run. + * @param f - Function which will be executed. + */ + Scheduler(const std::chrono::seconds &t, const std::function &f) + : pause_(t), func_(f) { + thread_ = std::thread([this]() { + // if pause_ equals -1, scheduler will not run + if (pause_ == std::chrono::seconds(-1)) return; + auto start_time = std::chrono::system_clock::now(); + for(;;) { + if (!is_working_.load()) break; + + func_(); + + std::unique_lock lk(mutex_); + + auto now = std::chrono::system_clock::now(); + while(now >= start_time) start_time += pause_; + + condition_variable_.wait_for(lk, start_time - now, [&] { + return is_working_.load() == false; + }); + lk.unlock(); + } + }); + } + + ~Scheduler() { + is_working_.store(false); + { + std::unique_lock lk(mutex_); + condition_variable_.notify_one(); + } + if (thread_.joinable()) thread_.join(); + } + + private: + /** + * Variable is true when thread is running. + */ + std::atomic is_working_{true}; + /** + * Time interval between function execution. + */ + const std::chrono::seconds pause_; + /** + * Function which scheduler executes. + */ + const std::function &func_; + /** + * Mutex used to synchronize threads using condition variable. + */ + TMutex mutex_; + + /** + * Condition variable is used to stop waiting until the end of the + * time interval if destructor is called. + */ + std::condition_variable condition_variable_; + /** + * Thread which runs function. + */ + std::thread thread_; +}; diff --git a/tests/unit/scheduler.cpp b/tests/unit/scheduler.cpp new file mode 100644 index 000000000..7013f59c0 --- /dev/null +++ b/tests/unit/scheduler.cpp @@ -0,0 +1,47 @@ +#include "gmock/gmock.h" +#include "gtest/gtest.h" + +#include "utils/scheduler.hpp" +#include + +/** + * Scheduler runs every 2 seconds and increases one variable. Test thread + * increases other variable. Scheduler checks if variables have the same + * value. + */ +TEST(Scheduler, TestFunctionExecuting) { + std::atomic x{0}, y{0}; + std::function func{[&x, &y]() { + EXPECT_EQ(y.load(), x.load()); + x++; + }}; + Scheduler<> scheduler(std::chrono::seconds(1), func); + + std::this_thread::sleep_for(std::chrono::milliseconds(980)); + y++; + EXPECT_EQ(x.load(), y.load()); + + + std::this_thread::sleep_for(std::chrono::seconds(1)); + y++; + EXPECT_EQ(x.load(), y.load()); + + std::this_thread::sleep_for(std::chrono::seconds(1)); + y++; + EXPECT_EQ(x.load(), y.load()); + + EXPECT_EQ(x.load(), 3); +} + +/** + * Scheduler will not run because time is set to -1. + */ +TEST(Scheduler, DoNotRunScheduler) { + std::atomic x{0}; + std::function func{[&x]() { x++; }}; + Scheduler<> scheduler(std::chrono::seconds(-1), func); + + std::this_thread::sleep_for(std::chrono::seconds(3)); + + EXPECT_EQ(x.load(), 0); +}