New version of scheduler.

Summary: New scheduler declaration.

Reviewers: matej.gradicek, buda

Reviewed By: matej.gradicek

Differential Revision: https://phabricator.memgraph.io/D246

src/utils/scheduler.hpp

@@ -4,37 +4,37 @@
 #include <ctime>
 #include <thread>
 
+#include "utils/assert.hpp"
+
 /**
- * Class used to run given std::function<void()>. Function is ran and then sleep
- * is called with parameter given in class constructor. Class is templated with
+ * Class used to run scheduled function execution. Class is templated with
  * mutex class TMutex which is used to synchronize threads. Default template
  * value is std::mutex.
  */
 template <typename TMutex = std::mutex>
 class Scheduler {
  public:
+  Scheduler() {}
   /**
-   * @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 pause - 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 to current time plus pause.
    * @param f - Function which will be executed.
    */
-  Scheduler(const std::chrono::seconds &t, const std::function<void()> &f)
-      : pause_(t), func_(f) {
-    thread_ = std::thread([this]() {
-      // if pause_ equals -1, scheduler will not run
-      if (pause_ == std::chrono::seconds(-1)) return;
+  void Run(const std::chrono::seconds &pause, const std::function<void()> &f) {
+    debug_assert(is_working_ == false, "Thread already running.");
+    is_working_ = true;
+    thread_ = std::thread([this, pause, f]() {
       auto start_time = std::chrono::system_clock::now();
       for (;;) {
         if (!is_working_.load()) break;
 
-        func_();
+        f();
 
         std::unique_lock<std::mutex> lk(mutex_);
 
         auto now = std::chrono::system_clock::now();
-        while (now >= start_time) start_time += pause_;
+        while (now >= start_time) start_time += pause;
 
         condition_variable_.wait_for(
             lk, start_time - now, [&] { return is_working_.load() == false; });
@@ -43,7 +43,11 @@ class Scheduler {
     });
   }
 
-  ~Scheduler() {
+  /**
+   * @brief Stops the thread execution. This is a blocking call and may take as
+   * much time as one call to the function given previously to Run takes.
+   */
+  void Stop() {
     is_working_.store(false);
     {
       std::unique_lock<std::mutex> lk(mutex_);
@@ -52,19 +56,14 @@ class Scheduler {
     if (thread_.joinable()) thread_.join();
   }
 
+  ~Scheduler() { Stop(); }
+
  private:
   /**
    * Variable is true when thread is running.
    */
-  std::atomic<bool> is_working_{true};
-  /**
-   * Time interval between function execution.
-   */
-  const std::chrono::seconds pause_;
-  /**
-   * Function which scheduler executes.
-   */
-  const std::function<void()> func_;
+  std::atomic<bool> is_working_{false};
+
   /**
    * Mutex used to synchronize threads using condition variable.
    */
@@ -72,9 +71,10 @@ class Scheduler {
 
   /**
    * Condition variable is used to stop waiting until the end of the
-   * time interval if destructor is  called.
+   * time interval if destructor is called.
    */
   std::condition_variable condition_variable_;
+
   /**
    * Thread which runs function.
    */

tests/unit/scheduler.cpp

@@ -1,47 +1,29 @@
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
 
-#include "utils/scheduler.hpp"
 #include <atomic>
+#include "utils/scheduler.hpp"
 
 /**
- * Scheduler runs every 2 seconds and increases one variable. Test thread 
+ * 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<int> x{0}, y{0}; 
-  std::function<void()> func{[&x, &y]() { 
-    EXPECT_EQ(y.load(), x.load()); 
-    x++; 
+  std::atomic<int> x{0}, y{0};
+  std::function<void()> 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)); 
+  Scheduler<> scheduler;
+  scheduler.Run(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));
+  std::this_thread::sleep_for(std::chrono::milliseconds(100));
+  scheduler.Stop();
   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<int> x{0}; 
-  std::function<void()> func{[&x]() { x++; }};
-  Scheduler<> scheduler(std::chrono::seconds(-1), func);
-
-  std::this_thread::sleep_for(std::chrono::seconds(3));
-
-  EXPECT_EQ(x.load(), 0);
 }