Implement Timekeeper class to wrap a 'static' clock as found in StateMachine

src/CMakeLists.txt

@@ -13,23 +13,25 @@ add_definitions(${GCOV_CFLAGS})
 add_library(XrdQuarkDB SHARED
   XrdPlugin.cc
   XrdQuarkDB.cc                       XrdQuarkDB.hh
-  Configuration.cc                    Configuration.hh
   Utils.cc                            Utils.hh
-                                      Common.hh
-  Link.cc                             Link.hh
   BufferedReader.cc                   BufferedReader.hh
   BufferedWriter.cc                   BufferedWriter.hh
-  RedisParser.cc                      RedisParser.hh
-  RedisRequest.cc                     RedisRequest.hh
-  StateMachine.cc                     StateMachine.hh
   Commands.cc                         Commands.hh
-  Formatter.cc                        Formatter.hh
+                                      Common.hh
+  Configuration.cc                    Configuration.hh
+  Connection.cc                       Connection.hh
   Dispatcher.cc                       Dispatcher.hh
+                                      EventFD.hh
+  Formatter.cc                        Formatter.hh
+  Link.cc                             Link.hh
   Poller.cc                           Poller.hh
-  Connection.cc                       Connection.hh
   QuarkDBNode.cc                      QuarkDBNode.hh
+  RedisParser.cc                      RedisParser.hh
+  RedisRequest.cc                     RedisRequest.hh
   Shard.cc                            Shard.hh
   ShardDirectory.cc                   ShardDirectory.hh
+  StateMachine.cc                     StateMachine.hh
+  Timekeeper.cc                       Timekeeper.hh
 
   auth/AuthenticationDispatcher.cc    auth/AuthenticationDispatcher.hh
 

src/Timekeeper.cc

+// ----------------------------------------------------------------------
+// File: Timekeeper.cc
+// Author: Georgios Bitzes - CERN
+// ----------------------------------------------------------------------
+
+/************************************************************************
+ * quarkdb - a redis-like highly available key-value store              *
+ * Copyright (C) 2018 CERN/Switzerland                                  *
+ *                                                                      *
+ * This program is free software: you can redistribute it and/or modify *
+ * it under the terms of the GNU General Public License as published by *
+ * the Free Software Foundation, either version 3 of the License, or    *
+ * (at your option) any later version.                                  *
+ *                                                                      *
+ * This program is distributed in the hope that it will be useful,      *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of       *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the        *
+ * GNU General Public License for more details.                         *
+ *                                                                      *
+ * You should have received a copy of the GNU General Public License    *
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.*
+ ************************************************************************/
+
+#include "Timekeeper.hh"
+using namespace quarkdb;
+
+Timekeeper::Timekeeper(ClockValue startup) : staticClock(startup) {
+  anchorPoint = std::chrono::steady_clock::now();
+}
+
+bool Timekeeper::synchronize(ClockValue observed) {
+  std::unique_lock<std::shared_mutex> lock(mtx);
+
+  if(observed > staticClock + getTimeSinceAnchor().count() ) {
+    // We have a timejump. Re-anchor, update static clock
+    anchorPoint = std::chrono::steady_clock::now();
+    staticClock = observed;
+    return true;
+  }
+  else {
+    // Nothing to do, the clock never goes back in time
+    return false;
+  }
+}
+
+ClockValue Timekeeper::getCurrentTime() {
+  std::shared_lock<std::shared_mutex> lock(mtx);
+  return staticClock + getTimeSinceAnchor().count();
+}
+
+std::chrono::milliseconds Timekeeper::getTimeSinceAnchor() {
+  return std::chrono::duration_cast<std::chrono::milliseconds>(
+    std::chrono::steady_clock::now() - anchorPoint
+  );
+}

src/Timekeeper.hh

+// ----------------------------------------------------------------------
+// File: Timekeeper.hh
+// Author: Georgios Bitzes - CERN
+// ----------------------------------------------------------------------
+
+/************************************************************************
+ * quarkdb - a redis-like highly available key-value store              *
+ * Copyright (C) 2018 CERN/Switzerland                                  *
+ *                                                                      *
+ * This program is free software: you can redistribute it and/or modify *
+ * it under the terms of the GNU General Public License as published by *
+ * the Free Software Foundation, either version 3 of the License, or    *
+ * (at your option) any later version.                                  *
+ *                                                                      *
+ * This program is distributed in the hope that it will be useful,      *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of       *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the        *
+ * GNU General Public License for more details.                         *
+ *                                                                      *
+ * You should have received a copy of the GNU General Public License    *
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.*
+ ************************************************************************/
+
+#ifndef QUARKDB_TIME_KEEPER_H
+#define QUARKDB_TIME_KEEPER_H
+
+#include <chrono>
+#include <shared_mutex>
+
+namespace quarkdb {
+
+using ClockValue = uint64_t;
+
+class Timekeeper {
+public:
+  //----------------------------------------------------------------------------
+  // Construct Timekeeper with the given initial ClockValue. Time starts
+  // rolling forward as soon as the object is constructed.
+  //----------------------------------------------------------------------------
+  Timekeeper(ClockValue startup);
+
+  //----------------------------------------------------------------------------
+  // The static clock has been updated to the given value. Most of the time,
+  // the given value should actually be less than what we have. If that's not
+  // the case, update.
+  //
+  // A Timekeeper will never go back in time!
+  //----------------------------------------------------------------------------
+  bool synchronize(ClockValue observed);
+
+  //----------------------------------------------------------------------------
+  // Get the current time in milliseconds.
+  //----------------------------------------------------------------------------
+  ClockValue getCurrentTime();
+
+private:
+  std::shared_mutex mtx;
+  ClockValue staticClock;
+  std::chrono::steady_clock::time_point anchorPoint;
+
+  //----------------------------------------------------------------------------
+  // Get time elapsed since last anchor point
+  //----------------------------------------------------------------------------
+  std::chrono::milliseconds getTimeSinceAnchor();
+};
+
+}
+
+#endif

test/CMakeLists.txt

@@ -83,6 +83,7 @@ add_executable(quarkdb-stress-tests
   stress/qclient.cc
   stress/replication.cc
   stress/resilvering.cc
+  stress/timekeeper.cc
   ${COMMON_TEST_SOURCES}
 )
 

test/stress/timekeeper.cc

+// ----------------------------------------------------------------------
+// File: timekeeper.cc
+// Author: Georgios Bitzes - CERN
+// ----------------------------------------------------------------------
+
+/************************************************************************
+ * quarkdb - a redis-like highly available key-value store              *
+ * Copyright (C) 2018 CERN/Switzerland                                  *
+ *                                                                      *
+ * This program is free software: you can redistribute it and/or modify *
+ * it under the terms of the GNU General Public License as published by *
+ * the Free Software Foundation, either version 3 of the License, or    *
+ * (at your option) any later version.                                  *
+ *                                                                      *
+ * This program is distributed in the hope that it will be useful,      *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of       *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the        *
+ * GNU General Public License for more details.                         *
+ *                                                                      *
+ * You should have received a copy of the GNU General Public License    *
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.*
+ ************************************************************************/
+
+#include "../test-reply-macros.hh"
+#include "../test-utils.hh"
+#include "Timekeeper.hh"
+#include <iostream>
+
+using namespace quarkdb;
+
+TEST(Timekeeper, BasicSanity) {
+  Timekeeper tk(ClockValue(123));
+  ASSERT_GE(tk.getCurrentTime(), ClockValue(123));
+  std::cerr << "Initialization: " << tk.getCurrentTime() << std::endl;
+
+  std::this_thread::sleep_for(std::chrono::seconds(1));
+  ASSERT_GE(tk.getCurrentTime(), ClockValue(1123));
+  std::cerr << "After 1 sec: " << tk.getCurrentTime() << std::endl;
+
+  // Timekeeper should not go back in time
+  ASSERT_FALSE(tk.synchronize(ClockValue(1000)));
+  ASSERT_GE(tk.getCurrentTime(), ClockValue(1123));
+  std::cerr << "After unsuccessful synchronization: " << tk.getCurrentTime() << std::endl;
+
+  // Timejump
+  ASSERT_TRUE(tk.synchronize(ClockValue(2000)));
+  ASSERT_GE(tk.getCurrentTime(), ClockValue(2000));
+  std::cerr << "After successful synchronization at 2000 ClockValue: " << tk.getCurrentTime() << std::endl;
+
+  // Ensure the clock doesn't go back, or something
+  ClockValue prevValue = tk.getCurrentTime();
+
+  for(size_t i = 0; i < 10; i++) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    ClockValue curVal = tk.getCurrentTime();
+    ASSERT_GE(curVal, prevValue);
+    prevValue = curVal;
+    std::cerr << "Tick: " << prevValue << std::endl;
+  }
+}