Add a SystemClock class to capture the time functions of an Env (face…

…book#7858)

Summary:
Introduces and uses a SystemClock class to RocksDB.  This class contains the time-related functions of an Env and these functions can be redirected from the Env to the SystemClock.

Many of the places that used an Env (Timer, PerfStepTimer, RepeatableThread, RateLimiter, WriteController) for time-related functions have been changed to use SystemClock instead.  There are likely more places that can be changed, but this is a start to show what can/should be done.  Over time it would be nice to migrate most (if not all) of the uses of the time functions from the Env to the SystemClock.

There are several Env classes that implement these functions.  Most of these have not been converted yet to SystemClock implementations; that will come in a subsequent PR.  It would be good to unify many of the Mock Timer implementations, so that they behave similarly and be tested similarly (some override Sleep, some use a MockSleep, etc).

Additionally, this change will allow new methods to be introduced to the SystemClock (like facebook#7101 WaitFor) in a consistent manner across a smaller number of classes.

Pull Request resolved: facebook#7858

Reviewed By: pdillinger

Differential Revision: D26006406

Pulled By: mrambacher

fbshipit-source-id: ed10a8abbdab7ff2e23d69d85bd25b3e7e899e90

CMakeLists.txt

@@ -642,6 +642,7 @@ set(SOURCES
         db/write_batch_base.cc
         db/write_controller.cc
         db/write_thread.cc
+        env/composite_env.cc
         env/env.cc
         env/env_chroot.cc
         env/env_encryption.cc

HISTORY.md

@@ -12,7 +12,7 @@
 
 ### Public API Change
 * Add a public API WriteBufferManager::dummy_entries_in_cache_usage() which reports the size of dummy entries stored in cache (passed to WriteBufferManager). Dummy entries are used to account for DataBlocks.
-
+* Add a SystemClock class that contains the time-related methods from Env.  The original methods in Env may be deprecated in a future release.  This class will allow easier testing, development, and expansion of time-related features.
 ## 6.16.0 (12/18/2020)
 ### Behavior Changes
 * Attempting to write a merge operand without explicitly configuring `merge_operator` now fails immediately, causing the DB to enter read-only mode. Previously, failure was deferred until the `merge_operator` was needed by a user read or a background operation.

TARGETS

@@ -206,6 +206,7 @@ cpp_library(
         "db/write_batch_base.cc",
         "db/write_controller.cc",
         "db/write_thread.cc",
+        "env/composite_env.cc",
         "env/env.cc",
         "env/env_chroot.cc",
         "env/env_encryption.cc",
@@ -510,6 +511,7 @@ cpp_library(
         "db/write_batch_base.cc",
         "db/write_controller.cc",
         "db/write_thread.cc",
+        "env/composite_env.cc",
         "env/env.cc",
         "env/env_chroot.cc",
         "env/env_encryption.cc",

cache/cache_bench.cc

@@ -13,13 +13,15 @@ int main() {
 
 #include <stdio.h>
 #include <sys/types.h>
+
 #include <cinttypes>
 #include <limits>
 
 #include "port/port.h"
 #include "rocksdb/cache.h"
 #include "rocksdb/db.h"
 #include "rocksdb/env.h"
+#include "rocksdb/system_clock.h"
 #include "util/coding.h"
 #include "util/gflags_compat.h"
 #include "util/hash.h"
@@ -210,6 +212,7 @@ class CacheBench {
 
   bool Run() {
     ROCKSDB_NAMESPACE::Env* env = ROCKSDB_NAMESPACE::Env::Default();
+    const auto& clock = env->GetSystemClock();
 
     PrintEnv();
     SharedState shared(this);
@@ -224,7 +227,7 @@ class CacheBench {
         shared.GetCondVar()->Wait();
       }
       // Record start time
-      uint64_t start_time = env->NowMicros();
+      uint64_t start_time = clock->NowMicros();
 
       // Start all threads
       shared.SetStart();
@@ -236,7 +239,7 @@ class CacheBench {
       }
 
       // Record end time
-      uint64_t end_time = env->NowMicros();
+      uint64_t end_time = clock->NowMicros();
       double elapsed = static_cast<double>(end_time - start_time) * 1e-6;
       uint32_t qps = static_cast<uint32_t>(
           static_cast<double>(FLAGS_threads * FLAGS_ops_per_thread) / elapsed);

db/blob/blob_file_builder.cc

@@ -49,7 +49,6 @@ BlobFileBuilder::BlobFileBuilder(
     std::vector<std::string>* blob_file_paths,
     std::vector<BlobFileAddition>* blob_file_additions)
     : file_number_generator_(std::move(file_number_generator)),
-      env_(env),
       fs_(fs),
       immutable_cf_options_(immutable_cf_options),
       min_blob_size_(mutable_cf_options->min_blob_size),
@@ -66,14 +65,15 @@ BlobFileBuilder::BlobFileBuilder(
       blob_count_(0),
       blob_bytes_(0) {
   assert(file_number_generator_);
-  assert(env_);
+  assert(env);
   assert(fs_);
   assert(immutable_cf_options_);
   assert(file_options_);
   assert(blob_file_paths_);
   assert(blob_file_paths_->empty());
   assert(blob_file_additions_);
   assert(blob_file_additions_->empty());
+  clock_ = env->GetSystemClock();
 }
 
 BlobFileBuilder::~BlobFileBuilder() = default;
@@ -181,14 +181,14 @@ Status BlobFileBuilder::OpenBlobFileIfNeeded() {
   Statistics* const statistics = immutable_cf_options_->statistics;
 
   std::unique_ptr<WritableFileWriter> file_writer(new WritableFileWriter(
-      std::move(file), blob_file_paths_->back(), *file_options_, env_,
+      std::move(file), blob_file_paths_->back(), *file_options_, clock_,
       nullptr /*IOTracer*/, statistics, immutable_cf_options_->listeners,
       immutable_cf_options_->file_checksum_gen_factory));
 
   constexpr bool do_flush = false;
 
   std::unique_ptr<BlobLogWriter> blob_log_writer(new BlobLogWriter(
-      std::move(file_writer), env_, statistics, blob_file_number,
+      std::move(file_writer), clock_, statistics, blob_file_number,
       immutable_cf_options_->use_fsync, do_flush));
 
   constexpr bool has_ttl = false;

db/blob/blob_file_builder.h

@@ -18,6 +18,7 @@ namespace ROCKSDB_NAMESPACE {
 
 class VersionSet;
 class FileSystem;
+class SystemClock;
 struct ImmutableCFOptions;
 struct MutableCFOptions;
 struct FileOptions;
@@ -69,8 +70,8 @@ class BlobFileBuilder {
   Status CloseBlobFileIfNeeded();
 
   std::function<uint64_t()> file_number_generator_;
-  Env* env_;
   FileSystem* fs_;
+  std::shared_ptr<SystemClock> clock_;
   const ImmutableCFOptions* immutable_cf_options_;
   uint64_t min_blob_size_;
   uint64_t blob_file_size_;

db/blob/blob_file_builder_test.cc

@@ -39,8 +39,10 @@ class TestFileNumberGenerator {
 
 class BlobFileBuilderTest : public testing::Test {
  protected:
-  BlobFileBuilderTest()
-      : mock_env_(Env::Default()), fs_(mock_env_.GetFileSystem().get()) {}
+  BlobFileBuilderTest() : mock_env_(Env::Default()) {
+    fs_ = mock_env_.GetFileSystem().get();
+    clock_ = mock_env_.GetSystemClock();
+  }
 
   void VerifyBlobFile(uint64_t blob_file_number,
                       const std::string& blob_file_path,
@@ -57,11 +59,10 @@ class BlobFileBuilderTest : public testing::Test {
         fs_->NewRandomAccessFile(blob_file_path, file_options_, &file, dbg));
 
     std::unique_ptr<RandomAccessFileReader> file_reader(
-        new RandomAccessFileReader(std::move(file), blob_file_path,
-                                   &mock_env_));
+        new RandomAccessFileReader(std::move(file), blob_file_path, clock_));
 
     constexpr Statistics* statistics = nullptr;
-    BlobLogSequentialReader blob_log_reader(std::move(file_reader), &mock_env_,
+    BlobLogSequentialReader blob_log_reader(std::move(file_reader), clock_,
                                             statistics);
 
     BlobLogHeader header;
@@ -109,6 +110,7 @@ class BlobFileBuilderTest : public testing::Test {
 
   MockEnv mock_env_;
   FileSystem* fs_;
+  std::shared_ptr<SystemClock> clock_;
   FileOptions file_options_;
 };
 

db/blob/blob_file_cache_test.cc

@@ -42,15 +42,15 @@ void WriteBlobFile(uint32_t column_family_id,
 
   std::unique_ptr<WritableFileWriter> file_writer(
       new WritableFileWriter(std::move(file), blob_file_path, FileOptions(),
-                             immutable_cf_options.env));
+                             immutable_cf_options.env->GetSystemClock()));
 
   constexpr Statistics* statistics = nullptr;
   constexpr bool use_fsync = false;
   constexpr bool do_flush = false;
 
-  BlobLogWriter blob_log_writer(std::move(file_writer),
-                                immutable_cf_options.env, statistics,
-                                blob_file_number, use_fsync, do_flush);
+  BlobLogWriter blob_log_writer(
+      std::move(file_writer), immutable_cf_options.env->GetSystemClock(),
+      statistics, blob_file_number, use_fsync, do_flush);
 
   constexpr bool has_ttl = false;
   constexpr ExpirationRange expiration_range;

db/blob/blob_file_reader.cc

@@ -117,10 +117,11 @@ Status BlobFileReader::OpenFile(
   }
 
   file_reader->reset(new RandomAccessFileReader(
-      std::move(file), blob_file_path, immutable_cf_options.env,
-      std::shared_ptr<IOTracer>(), immutable_cf_options.statistics,
-      BLOB_DB_BLOB_FILE_READ_MICROS, blob_file_read_hist,
-      immutable_cf_options.rate_limiter, immutable_cf_options.listeners));
+      std::move(file), blob_file_path,
+      immutable_cf_options.env->GetSystemClock(), std::shared_ptr<IOTracer>(),
+      immutable_cf_options.statistics, BLOB_DB_BLOB_FILE_READ_MICROS,
+      blob_file_read_hist, immutable_cf_options.rate_limiter,
+      immutable_cf_options.listeners));
 
   return Status::OK();
 }

db/blob/blob_file_reader_test.cc

@@ -50,15 +50,15 @@ void WriteBlobFile(const ImmutableCFOptions& immutable_cf_options,
 
   std::unique_ptr<WritableFileWriter> file_writer(
       new WritableFileWriter(std::move(file), blob_file_path, FileOptions(),
-                             immutable_cf_options.env));
+                             immutable_cf_options.env->GetSystemClock()));
 
   constexpr Statistics* statistics = nullptr;
   constexpr bool use_fsync = false;
   constexpr bool do_flush = false;
 
-  BlobLogWriter blob_log_writer(std::move(file_writer),
-                                immutable_cf_options.env, statistics,
-                                blob_file_number, use_fsync, do_flush);
+  BlobLogWriter blob_log_writer(
+      std::move(file_writer), immutable_cf_options.env->GetSystemClock(),
+      statistics, blob_file_number, use_fsync, do_flush);
 
   BlobLogHeader header(column_family_id, compression_type, has_ttl,
                        expiration_range_header);
@@ -260,15 +260,15 @@ TEST_F(BlobFileReaderTest, Malformed) {
 
     std::unique_ptr<WritableFileWriter> file_writer(
         new WritableFileWriter(std::move(file), blob_file_path, FileOptions(),
-                               immutable_cf_options.env));
+                               immutable_cf_options.env->GetSystemClock()));
 
     constexpr Statistics* statistics = nullptr;
     constexpr bool use_fsync = false;
     constexpr bool do_flush = false;
 
-    BlobLogWriter blob_log_writer(std::move(file_writer),
-                                  immutable_cf_options.env, statistics,
-                                  blob_file_number, use_fsync, do_flush);
+    BlobLogWriter blob_log_writer(
+        std::move(file_writer), immutable_cf_options.env->GetSystemClock(),
+        statistics, blob_file_number, use_fsync, do_flush);
 
     BlobLogHeader header(column_family_id, kNoCompression, has_ttl,
                          expiration_range);

db/blob/blob_log_sequential_reader.cc

@@ -13,10 +13,10 @@
 namespace ROCKSDB_NAMESPACE {
 
 BlobLogSequentialReader::BlobLogSequentialReader(
-    std::unique_ptr<RandomAccessFileReader>&& file_reader, Env* env,
-    Statistics* statistics)
+    std::unique_ptr<RandomAccessFileReader>&& file_reader,
+    const std::shared_ptr<SystemClock>& clock, Statistics* statistics)
     : file_(std::move(file_reader)),
-      env_(env),
+      clock_(clock),
       statistics_(statistics),
       next_byte_(0) {}
 
@@ -27,7 +27,7 @@ Status BlobLogSequentialReader::ReadSlice(uint64_t size, Slice* slice,
   assert(slice);
   assert(file_);
 
-  StopWatch read_sw(env_, statistics_, BLOB_DB_BLOB_FILE_READ_MICROS);
+  StopWatch read_sw(clock_, statistics_, BLOB_DB_BLOB_FILE_READ_MICROS);
   Status s = file_->Read(IOOptions(), next_byte_, static_cast<size_t>(size),
                          slice, buf, nullptr);
   next_byte_ += size;

db/blob/blob_log_sequential_reader.h

@@ -16,6 +16,7 @@ class RandomAccessFileReader;
 class Env;
 class Statistics;
 class Status;
+class SystemClock;
 
 /**
  * BlobLogSequentialReader is a general purpose log stream reader
@@ -35,7 +36,8 @@ class BlobLogSequentialReader {
 
   // Create a reader that will return log records from "*file_reader".
   BlobLogSequentialReader(std::unique_ptr<RandomAccessFileReader>&& file_reader,
-                          Env* env, Statistics* statistics);
+                          const std::shared_ptr<SystemClock>& clock,
+                          Statistics* statistics);
 
   // No copying allowed
   BlobLogSequentialReader(const BlobLogSequentialReader&) = delete;
@@ -63,7 +65,8 @@ class BlobLogSequentialReader {
   Status ReadSlice(uint64_t size, Slice* slice, char* buf);
 
   const std::unique_ptr<RandomAccessFileReader> file_;
-  Env* env_;
+  std::shared_ptr<SystemClock> clock_;
+
   Statistics* statistics_;
 
   Slice buffer_;

db/blob/blob_log_writer.cc

@@ -11,19 +11,19 @@
 #include "db/blob/blob_log_format.h"
 #include "file/writable_file_writer.h"
 #include "monitoring/statistics.h"
-#include "rocksdb/env.h"
+#include "rocksdb/system_clock.h"
 #include "test_util/sync_point.h"
 #include "util/coding.h"
 #include "util/stop_watch.h"
 
 namespace ROCKSDB_NAMESPACE {
 
 BlobLogWriter::BlobLogWriter(std::unique_ptr<WritableFileWriter>&& dest,
-                             Env* env, Statistics* statistics,
-                             uint64_t log_number, bool use_fs, bool do_flush,
-                             uint64_t boffset)
+                             const std::shared_ptr<SystemClock>& clock,
+                             Statistics* statistics, uint64_t log_number,
+                             bool use_fs, bool do_flush, uint64_t boffset)
     : dest_(std::move(dest)),
-      env_(env),
+      clock_(clock),
       statistics_(statistics),
       log_number_(log_number),
       block_offset_(boffset),
@@ -36,7 +36,7 @@ BlobLogWriter::~BlobLogWriter() = default;
 Status BlobLogWriter::Sync() {
   TEST_SYNC_POINT("BlobLogWriter::Sync");
 
-  StopWatch sync_sw(env_, statistics_, BLOB_DB_BLOB_FILE_SYNC_MICROS);
+  StopWatch sync_sw(clock_, statistics_, BLOB_DB_BLOB_FILE_SYNC_MICROS);
   Status s = dest_->Sync(use_fsync_);
   RecordTick(statistics_, BLOB_DB_BLOB_FILE_SYNCED);
   return s;
@@ -148,7 +148,7 @@ Status BlobLogWriter::EmitPhysicalRecord(const std::string& headerbuf,
                                          const Slice& key, const Slice& val,
                                          uint64_t* key_offset,
                                          uint64_t* blob_offset) {
-  StopWatch write_sw(env_, statistics_, BLOB_DB_BLOB_FILE_WRITE_MICROS);
+  StopWatch write_sw(clock_, statistics_, BLOB_DB_BLOB_FILE_WRITE_MICROS);
   Status s = dest_->Append(Slice(headerbuf));
   if (s.ok()) {
     s = dest_->Append(key);

db/blob/blob_log_writer.h

@@ -9,7 +9,6 @@
 #include <string>
 
 #include "db/blob/blob_log_format.h"
-#include "rocksdb/env.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/statistics.h"
 #include "rocksdb/status.h"
@@ -18,7 +17,7 @@
 namespace ROCKSDB_NAMESPACE {
 
 class WritableFileWriter;
-
+class SystemClock;
 /**
  * BlobLogWriter is the blob log stream writer. It provides an append-only
  * abstraction for writing blob data.
@@ -32,7 +31,8 @@ class BlobLogWriter {
   // Create a writer that will append data to "*dest".
   // "*dest" must be initially empty.
   // "*dest" must remain live while this BlobLogWriter is in use.
-  BlobLogWriter(std::unique_ptr<WritableFileWriter>&& dest, Env* env,
+  BlobLogWriter(std::unique_ptr<WritableFileWriter>&& dest,
+                const std::shared_ptr<SystemClock>& clock,
                 Statistics* statistics, uint64_t log_number, bool use_fsync,
                 bool do_flush, uint64_t boffset = 0);
   // No copying allowed
@@ -69,7 +69,7 @@ class BlobLogWriter {
 
  private:
   std::unique_ptr<WritableFileWriter> dest_;
-  Env* env_;
+  std::shared_ptr<SystemClock> clock_;
   Statistics* statistics_;
   uint64_t log_number_;
   uint64_t block_offset_;  // Current offset in block