Add CloudScheduler class.

Class starts a thread and allows jobs to be executed in the future.

Add ability to do recurring jobs.  Move structs into CloudEnvImpl

Author: mrambacher

Makefile

@@ -447,6 +447,7 @@ TESTS = \
 	remote_compaction_test \
 	db_cloud_test \
 	cloud_manifest_test \
+	cloud_scheduler_test \
 	db_basic_test \
 	db_encryption_test \
 	db_test2 \
@@ -1685,6 +1686,9 @@ db_cloud_test: cloud/db_cloud_test.o $(LIBOBJECTS) $(TESTHARNESS)
 cloud_manifest_test: cloud/cloud_manifest_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_LINK)
 
+cloud_scheduler_test: cloud/cloud_scheduler_test.o $(LIBOBJECTS) $(TESTHARNESS)
+	$(AM_LINK)
+
 iostats_context_test: monitoring/iostats_context_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_V_CCLD)$(CXX) $^ $(EXEC_LDFLAGS) -o $@ $(LDFLAGS)
 

cloud/aws/aws_env.cc

@@ -12,6 +12,7 @@
 #include <set>
 
 #include "cloud/cloud_log_controller_impl.h"
+#include "cloud/cloud_scheduler.h"
 #include "cloud/cloud_storage_provider_impl.h"
 #include "cloud/filename.h"
 #include "port/port_posix.h"
@@ -164,107 +165,6 @@ Status AwsCloudAccessCredentials::GetCredentialsProvider(
 }
 
 #ifdef USE_AWS
-namespace detail {
-
-using ScheduledJob =
-    std::pair<std::chrono::steady_clock::time_point, std::function<void(void)>>;
-struct Comp {
-  bool operator()(const ScheduledJob& a, const ScheduledJob& b) const {
-    return a.first < b.first;
-  }
-};
-struct JobHandle {
-  std::multiset<ScheduledJob, Comp>::iterator itr;
-  JobHandle(std::multiset<ScheduledJob, Comp>::iterator i)
-      : itr(std::move(i)) {}
-};
-
-class JobExecutor {
- public:
-  std::shared_ptr<JobHandle> ScheduleJob(
-      std::chrono::steady_clock::time_point time,
-      std::function<void(void)> callback);
-  void CancelJob(JobHandle* handle);
-
-  JobExecutor();
-  ~JobExecutor();
-
- private:
-  void DoWork();
-
-  std::mutex mutex_;
-  // Notified when the earliest job to be scheduled has changed.
-  std::condition_variable jobs_changed_cv_;
-  std::multiset<ScheduledJob, Comp> scheduled_jobs_;
-  bool shutting_down_{false};
-
-  std::thread thread_;
-};
-
-JobExecutor::JobExecutor() {
-  thread_ = std::thread([this]() { DoWork(); });
-}
-
-JobExecutor::~JobExecutor() {
-  {
-    std::lock_guard<std::mutex> lk(mutex_);
-    shutting_down_ = true;
-    jobs_changed_cv_.notify_all();
-  }
-  if (thread_.joinable()) {
-    thread_.join();
-  }
-}
-
-std::shared_ptr<JobHandle> JobExecutor::ScheduleJob(
-    std::chrono::steady_clock::time_point time,
-    std::function<void(void)> callback) {
-  std::lock_guard<std::mutex> lk(mutex_);
-  auto itr = scheduled_jobs_.emplace(time, std::move(callback));
-  if (itr == scheduled_jobs_.begin()) {
-    jobs_changed_cv_.notify_all();
-  }
-  return std::make_shared<JobHandle>(itr);
-}
-
-void JobExecutor::CancelJob(JobHandle* handle) {
-  std::lock_guard<std::mutex> lk(mutex_);
-  if (scheduled_jobs_.begin() == handle->itr) {
-    jobs_changed_cv_.notify_all();
-  }
-  scheduled_jobs_.erase(handle->itr);
-}
-
-void JobExecutor::DoWork() {
-  while (true) {
-    std::unique_lock<std::mutex> lk(mutex_);
-    if (shutting_down_) {
-      break;
-    }
-    if (scheduled_jobs_.empty()) {
-      jobs_changed_cv_.wait(lk);
-      continue;
-    }
-    auto earliest_job = scheduled_jobs_.begin();
-    auto earliest_job_time = earliest_job->first;
-    if (earliest_job_time >= std::chrono::steady_clock::now()) {
-      jobs_changed_cv_.wait_until(lk, earliest_job_time);
-      continue;
-    }
-    // invoke the function
-    lk.unlock();
-    earliest_job->second();
-    lk.lock();
-    scheduled_jobs_.erase(earliest_job);
-  }
-}
-
-}  // namespace detail
-
-detail::JobExecutor* GetJobExecutor() {
-  static detail::JobExecutor executor;
-  return &executor;
-}
 
 //
 // The AWS credentials are specified to the constructor via
@@ -292,144 +192,8 @@ AwsEnv::AwsEnv(Env* underlying_env, const CloudEnvOptions& _cloud_env_options,
   base_env_ = underlying_env;
 }
 
-AwsEnv::~AwsEnv() {
-  {
-    std::lock_guard<std::mutex> lk(files_to_delete_mutex_);
-    using std::swap;
-    for (auto& e : files_to_delete_) {
-      GetJobExecutor()->CancelJob(e.second.get());
-    }
-    files_to_delete_.clear();
-  }
-
-  StopPurger();
-}
-
 void AwsEnv::Shutdown() { Aws::ShutdownAPI(Aws::SDKOptions()); }
 
-void AwsEnv::RemoveFileFromDeletionQueue(const std::string& filename) {
-  std::lock_guard<std::mutex> lk(files_to_delete_mutex_);
-  auto itr = files_to_delete_.find(filename);
-  if (itr != files_to_delete_.end()) {
-    GetJobExecutor()->CancelJob(itr->second.get());
-    files_to_delete_.erase(itr);
-  }
-}
-
-Status AwsEnv::DeleteFile(const std::string& logical_fname) {
-  auto fname = RemapFilename(logical_fname);
-  auto file_type = GetFileType(fname);
-  bool sstfile = (file_type == RocksDBFileType::kSstFile),
-       manifest = (file_type == RocksDBFileType::kManifestFile),
-       identity = (file_type == RocksDBFileType::kIdentityFile),
-       logfile = (file_type == RocksDBFileType::kLogFile);
-
-  if (manifest) {
-    // We don't delete manifest files. The reason for this is that even though
-    // RocksDB creates manifest with different names (like MANIFEST-00001,
-    // MANIFEST-00008) we actually map all of them to the same filename
-    // MANIFEST-[epoch].
-    // When RocksDB wants to roll the MANIFEST (let's say from 1 to 8) it does
-    // the following:
-    // 1. Create a new MANIFEST-8
-    // 2. Write everything into MANIFEST-8
-    // 3. Sync MANIFEST-8
-    // 4. Store "MANIFEST-8" in CURRENT file
-    // 5. Delete MANIFEST-1
-    //
-    // What RocksDB cloud does behind the scenes (the numbers match the list
-    // above):
-    // 1. Create manifest file MANIFEST-[epoch].tmp
-    // 2. Forward RocksDB writes to the file created in the first step
-    // 3. Atomic rename from MANIFEST-[epoch].tmp to MANIFEST-[epoch]. The old
-    // file with the same file name is overwritten.
-    // 4. Nothing. Whatever the contents of CURRENT file, we don't care, we
-    // always remap MANIFEST files to the correct with the latest epoch.
-    // 5. Also nothing. There is no file to delete, because we have overwritten
-    // it in the third step.
-    return Status::OK();
-  }
-
-  Status st;
-  // Delete from destination bucket and local dir
-  if (sstfile || manifest || identity) {
-    if (HasDestBucket()) {
-      // add the remote file deletion to the queue
-      st = DeleteCloudFileFromDest(basename(fname));
-    }
-    // delete from local, too. Ignore the result, though. The file might not be
-    // there locally.
-    base_env_->DeleteFile(fname);
-  } else if (logfile && !cloud_env_options.keep_local_log_files) {
-    // read from Kinesis
-    st = cloud_env_options.cloud_log_controller->status();
-    if (st.ok()) {
-      // Log a Delete record to kinesis stream
-      std::unique_ptr<CloudLogWritableFile> f(
-          cloud_env_options.cloud_log_controller->CreateWritableFile(
-              fname, EnvOptions()));
-      if (!f || !f->status().ok()) {
-        st = Status::IOError("[Kinesis] DeleteFile", fname.c_str());
-      } else {
-        st = f->LogDelete();
-      }
-    }
-  } else {
-    st = base_env_->DeleteFile(fname);
-  }
-  Log(InfoLogLevel::DEBUG_LEVEL, info_log_, "[s3] DeleteFile file %s %s",
-      fname.c_str(), st.ToString().c_str());
-  return st;
-}
-
-Status AwsEnv::CopyLocalFileToDest(const std::string& local_name,
-                                   const std::string& dest_name) {
-  RemoveFileFromDeletionQueue(basename(local_name));
-  return cloud_env_options.storage_provider->PutCloudObject(
-      local_name, GetDestBucketName(), dest_name);
-}
-
-Status AwsEnv::DeleteCloudFileFromDest(const std::string& fname) {
-  assert(HasDestBucket());
-  auto base = basename(fname);
-  // add the job to delete the file in 1 hour
-  auto doDeleteFile = [this, base]() {
-    {
-      std::lock_guard<std::mutex> lk(files_to_delete_mutex_);
-      auto itr = files_to_delete_.find(base);
-      if (itr == files_to_delete_.end()) {
-        // File was removed from files_to_delete_, do not delete!
-        return;
-      }
-      files_to_delete_.erase(itr);
-    }
-    auto path = GetDestObjectPath() + "/" + base;
-    // we are ready to delete the file!
-    auto st = cloud_env_options.storage_provider->DeleteCloudObject(
-        GetDestBucketName(), path);
-    if (!st.ok() && !st.IsNotFound()) {
-      Log(InfoLogLevel::ERROR_LEVEL, info_log_,
-          "[aws] DeleteFile DeletePathInS3 file %s error %s", path.c_str(),
-          st.ToString().c_str());
-    }
-  };
-  {
-    std::lock_guard<std::mutex> lk(files_to_delete_mutex_);
-    if (files_to_delete_.find(base) != files_to_delete_.end()) {
-      // already in the queue
-      return Status::OK();
-    }
-  }
-  {
-    std::lock_guard<std::mutex> lk(files_to_delete_mutex_);
-    auto handle = GetJobExecutor()->ScheduleJob(
-        std::chrono::steady_clock::now() + file_deletion_delay_,
-        std::move(doDeleteFile));
-    files_to_delete_.emplace(base, std::move(handle));
-  }
-  return Status::OK();
-}
-
 //
 // All db in a bucket are stored in path /.rockset/dbid/<dbid>
 // The value of the object is the pathname where the db resides.

cloud/aws/aws_env.h

@@ -21,11 +21,6 @@
 namespace ROCKSDB_NAMESPACE {
 
 class S3ReadableFile;
-
-namespace detail {
-struct JobHandle;
-}  // namespace detail
-
 //
 // The S3 environment for rocksdb. This class overrides all the
 // file/dir access methods and delegates all other methods to the
@@ -56,7 +51,7 @@ class AwsEnv : public CloudEnvImpl {
                           const std::shared_ptr<Logger>& info_log,
                           CloudEnv** cenv);
 
-  virtual ~AwsEnv();
+  virtual ~AwsEnv() {}
 
   const char* Name() const override { return "aws"; }
 
@@ -71,8 +66,6 @@ class AwsEnv : public CloudEnvImpl {
   // explicitly make the default region be us-west-2.
   static constexpr const char* default_region = "us-west-2";
 
-  virtual Status DeleteFile(const std::string& fname) override;
-
   virtual Status LockFile(const std::string& fname, FileLock** lock) override;
 
   virtual Status UnlockFile(FileLock* lock) override;
@@ -87,16 +80,6 @@ class AwsEnv : public CloudEnvImpl {
   Status GetDbidList(const std::string& bucket, DbidList* dblist) override;
   Status DeleteDbid(const std::string& bucket,
                     const std::string& dbid) override;
-  Status DeleteCloudFileFromDest(const std::string& fname) override;
-  Status CopyLocalFileToDest(const std::string& local_name,
-                             const std::string& cloud_name) override;
-
-  void RemoveFileFromDeletionQueue(const std::string& filename);
-
-  void TEST_SetFileDeletionDelay(std::chrono::seconds delay) {
-    std::lock_guard<std::mutex> lk(files_to_delete_mutex_);
-    file_deletion_delay_ = delay;
-  }
 
  private:
   //
@@ -109,10 +92,6 @@ class AwsEnv : public CloudEnvImpl {
   // The pathname that contains a list of all db's inside a bucket.
   static constexpr const char* dbid_registry_ = "/.rockset/dbid/";
 
-  std::mutex files_to_delete_mutex_;
-  std::chrono::seconds file_deletion_delay_ = std::chrono::hours(1);
-  std::unordered_map<std::string, std::shared_ptr<detail::JobHandle>>
-      files_to_delete_;
   Random64 rng_;
 };