Merge pull request #118 from rockset/AsanUnity

Make Cloud work for LITE and Unity tests

Author: dhruba

cloud/aws/aws_env.cc

@@ -1,5 +1,6 @@
 //  Copyright (c) 2016-present, Rockset, Inc.  All rights reserved.
 //
+#ifndef ROCKSDB_LITE
 #include "cloud/aws/aws_env.h"
 
 #include <chrono>
@@ -18,7 +19,7 @@
 #include "rocksdb/cloud/cloud_storage_provider.h"
 #include "rocksdb/env.h"
 #include "rocksdb/status.h"
-#include "util/stderr_logger.h"
+#include "rocksdb/utilities/options_type.h"
 #include "util/string_util.h"
 
 #ifdef USE_AWS
@@ -41,16 +42,6 @@ static const std::unordered_map<std::string, AwsAccessType> AwsAccessTypeMap = {
     {"anonymous", AwsAccessType::kAnonymous},
 };
 
-template <typename T>
-bool ParseEnum(const std::unordered_map<std::string, T>& type_map,
-               const std::string& type, T* value) {
-  auto iter = type_map.find(type);
-  if (iter != type_map.end()) {
-    *value = iter->second;
-    return true;
-  }
-  return false;
-}
 
 AwsAccessType AwsCloudAccessCredentials::GetAccessType() const {
   if (type != AwsAccessType::kUndefined) {
@@ -177,8 +168,7 @@ Status AwsCloudAccessCredentials::GetCredentialsProvider(
 //
 AwsEnv::AwsEnv(Env* underlying_env, const CloudEnvOptions& _cloud_env_options,
                const std::shared_ptr<Logger>& info_log)
-    : CloudEnvImpl(_cloud_env_options, underlying_env, info_log),
-      rng_(time(nullptr)) {
+    : CloudEnvImpl(_cloud_env_options, underlying_env, info_log) {
   Aws::InitAPI(Aws::SDKOptions());
   if (cloud_env_options.src_bucket.GetRegion().empty() ||
       cloud_env_options.dest_bucket.GetRegion().empty()) {
@@ -370,3 +360,4 @@ std::string AwsEnv::GetWALCacheDir() {
 
 #endif  // USE_AWS
 }  // namespace ROCKSDB_NAMESPACE
+#endif // ROCKSDB_LITE

cloud/aws/aws_env.h

@@ -9,8 +9,6 @@
 #include <iostream>
 
 #include "cloud/cloud_env_impl.h"
-#include "port/sys_time.h"
-#include "util/random.h"
 
 #ifdef USE_AWS
 
@@ -93,8 +91,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/";
-
-  Random64 rng_;
 };
 
 }  // namespace ROCKSDB_NAMESPACE

cloud/aws/aws_s3.cc

@@ -4,6 +4,7 @@
 // A directory maps to an an zero-size object in an S3 bucket
 // A sst file maps to an object in that S3 bucket.
 //
+#ifndef ROCKSDB_LITE
 #ifdef USE_AWS
 #include <aws/core/Aws.h>
 #include <aws/core/utils/Outcome.h>
@@ -903,7 +904,7 @@ Status S3StorageProvider::DoPutCloudObject(const std::string& local_file,
 }
 
 #endif /* USE_AWS */
-
+  
 Status CloudStorageProviderImpl::CreateS3Provider(
     std::shared_ptr<CloudStorageProvider>* provider) {
 #ifndef USE_AWS
@@ -916,3 +917,4 @@ Status CloudStorageProviderImpl::CreateS3Provider(
 #endif /* USE_AWS */
 }
 }  // namespace ROCKSDB_NAMESPACE
+#endif // ROCKSDB_LITE

cloud/cloud_scheduler.cc

@@ -46,6 +46,7 @@ class CloudSchedulerImpl : public CloudScheduler {
                             std::function<void(void*)> callback,
                             void* arg) override;
   bool CancelJob(long handle) override;
+  bool IsScheduled(long handle) override;
 
  private:
   void DoWork();
@@ -70,16 +71,25 @@ class LocalCloudScheduler : public CloudScheduler {
  public:
   LocalCloudScheduler(const std::shared_ptr<CloudScheduler>& scheduler,
                       long local_id)
-      : scheduler_(scheduler), next_local_id_(local_id) {}
+      : scheduler_(scheduler),
+        next_local_id_(local_id),
+        shutting_down_(false) {}
   ~LocalCloudScheduler() override {
-    for (const auto job : jobs_) {
+    {
+      std::lock_guard<std::mutex> lk(job_mutex_);
+      shutting_down_ = true;
+    }
+    for (const auto& job : jobs_) {
       scheduler_->CancelJob(job.second);
     }
     jobs_.clear();
   }
 
   long ScheduleJob(std::chrono::microseconds when,
                    std::function<void(void*)> callback, void* arg) override {
+    if (shutting_down_) {
+      return -1;
+    }
     std::lock_guard<std::mutex> lk(job_mutex_);
     long local_id = next_local_id_++;
     auto job = [this, local_id, callback](void* a) {
@@ -95,12 +105,37 @@ class LocalCloudScheduler : public CloudScheduler {
                             std::chrono::microseconds frequency,
                             std::function<void(void*)> callback,
                             void* arg) override {
+    if (shutting_down_) {
+      return -1;
+    }
     auto job = scheduler_->ScheduleRecurringJob(when, frequency, callback, arg);
     std::lock_guard<std::mutex> lk(job_mutex_);
     long local_id = next_local_id_++;
     jobs_[local_id] = job;
     return local_id;
   }
+
+  bool IsScheduled(long handle) override {
+    if (shutting_down_) {
+      return false;
+    } else {
+      std::lock_guard<std::mutex> lk(job_mutex_);
+      const auto& it = jobs_.find(handle);
+      if (it == jobs_.end()) {
+        // We do not have the job in our queue.  Return false
+        return false;
+      } else if (scheduler_->IsScheduled(it->second)) {
+        // The job is still scheduled.  Return false
+        return true;
+      } else {
+        // We have the job in our queue but it has already
+        // completed.  Erase from our queue and return false
+        jobs_.erase(it);
+        return false;
+      }
+    }
+  }
+
   // Cancels the job referred to by handle if it is active and associated with
   // this scheduler
   bool CancelJob(long handle) override {
@@ -123,6 +158,7 @@ class LocalCloudScheduler : public CloudScheduler {
   std::mutex job_mutex_;
   std::shared_ptr<CloudScheduler> scheduler_;
   long next_local_id_;
+  bool shutting_down_;
   std::unordered_map<long, long> jobs_;
 };
 
@@ -187,6 +223,22 @@ long CloudSchedulerImpl::ScheduleRecurringJob(
   return id;
 }
 
+bool CloudSchedulerImpl::IsScheduled(long id) {
+  if (id < 0) {
+    return false;
+  } else {
+    std::unique_lock<std::mutex> lk(mutex_);
+    if (!scheduled_jobs_.empty()) {
+      for (const auto& job : scheduled_jobs_) {
+        if (job.id == id) {
+          return true;
+        }
+      }
+    }
+    return false;
+  }
+}
+
 bool CloudSchedulerImpl::CancelJob(long id) {
   if (id < 0) {
     return false;

cloud/cloud_scheduler.h

@@ -37,6 +37,9 @@ class CloudScheduler {
   // was canceled, false otherwise.
   virtual bool CancelJob(long handle) = 0;
 
+  // Returns true if the job represented by handle has not completed,
+  // meaning that it is either currently schedule or actively running
+  virtual bool IsScheduled(long handle) = 0;
   // Returns a new instance of a cloud scheduler.  The caller is responsible
   // for freeing the scheduler when it is no longer required.
   static std::shared_ptr<CloudScheduler> Get();

cloud/cloud_scheduler_test.cc

@@ -1,5 +1,6 @@
 // Copyright (c) 2017 Rockset
 
+#ifndef ROCKSDB_LITE
 #include "cloud/cloud_scheduler.h"
 
 #include <gtest/gtest.h>
@@ -22,6 +23,21 @@ class CloudSchedulerTest : public testing::Test {
   ~CloudSchedulerTest() {}
 
   std::shared_ptr<CloudScheduler> scheduler_;
+  void WaitForJobs(const std::vector<long> &jobs, uint32_t delay) {
+    bool running = true;
+    while (running) {
+      running = false;
+      for (const auto &job : jobs) {
+        if (scheduler_->IsScheduled(job)) {
+          running = true;
+          break;
+        }
+      }
+      if (running) {
+        usleep(delay);
+      }
+    }
+  }
 };
 
 // This test tests basic scheduling function. There are 2 jobs, job2 is
@@ -33,10 +49,13 @@ TEST_F(CloudSchedulerTest, TestSchedule) {
   auto job1 = [&p1](void *) { p1 = std::chrono::steady_clock::now(); };
   auto job2 = [&p2](void *) { p2 = std::chrono::steady_clock::now(); };
 
-  scheduler_->ScheduleJob(std::chrono::milliseconds(300), job1, nullptr);
-  scheduler_->ScheduleJob(std::chrono::milliseconds(100), job2, nullptr);
-
-  std::this_thread::sleep_for(std::chrono::milliseconds(400));
+  auto h1 =
+      scheduler_->ScheduleJob(std::chrono::milliseconds(300), job1, nullptr);
+  auto h2 =
+      scheduler_->ScheduleJob(std::chrono::milliseconds(100), job2, nullptr);
+  while (scheduler_->IsScheduled(h1) && scheduler_->IsScheduled(h2)) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(400));
+  }
   ASSERT_LT(p2, p1);
 }
 
@@ -50,16 +69,16 @@ TEST_F(CloudSchedulerTest, TestCancel) {
   auto handle2 =
       scheduler_->ScheduleJob(std::chrono::microseconds(200), doJob, &job2);
   ASSERT_TRUE(scheduler_->CancelJob(handle2));
-  usleep(300);
+  WaitForJobs({handle1, handle2}, 300);
   ASSERT_EQ(job1, 2);
   ASSERT_EQ(job2, 0);
   ASSERT_FALSE(scheduler_->CancelJob(handle1));
   ASSERT_FALSE(scheduler_->CancelJob(handle2));
 }
 
 TEST_F(CloudSchedulerTest, TestRecurring) {
-  int job1 = 1;
-  int job2 = 1;
+  std::atomic<int> job1(1);
+  std::atomic<int> job2(1);
   auto doJob1 = [&job1](void *) { job1++; };
   auto doJob2 = [&job2](void *) { job2++; };
 
@@ -69,22 +88,24 @@ TEST_F(CloudSchedulerTest, TestRecurring) {
   scheduler_->ScheduleRecurringJob(std::chrono::microseconds(120),
                                    std::chrono::microseconds(100), doJob2,
                                    nullptr);
-  usleep(700);
-  ASSERT_GE(job2, 4);
-  ASSERT_GT(job1, job2);
+  while (job2 <= 4) {
+    usleep(100);
+  }
+  ASSERT_GE(job2.load(), 4);
+  ASSERT_GT(job1.load(), job2);
   ASSERT_TRUE(scheduler_->CancelJob(handle1));
-  auto old1 = job1;
-  auto old2 = job2;
+  auto old1 = job1.load();
+  auto old2 = job2.load();
   usleep(200);
-  ASSERT_EQ(job1, old1);
-  ASSERT_GT(job2, old2);
+  ASSERT_EQ(job1.load(), old1);
+  ASSERT_GT(job2.load(), old2);
 }
 
 TEST_F(CloudSchedulerTest, TestMultipleSchedulers) {
   auto scheduler2 = CloudScheduler::Get();
 
-  int job1 = 1;
-  int job2 = 1;
+  std::atomic<int> job1(1);
+  std::atomic<int> job2(1);
   auto doJob1 = [&job1](void *) { job1++; };
   auto doJob2 = [&job2](void *) { job2++; };
 
@@ -107,8 +128,8 @@ TEST_F(CloudSchedulerTest, TestMultipleSchedulers) {
                                    std::chrono::microseconds(20), doJob2,
                                    nullptr);
   scheduler2.reset();
-  auto old1 = job1;
-  auto old2 = job2;
+  auto old1 = job1.load();
+  auto old2 = job2.load();
   usleep(200);
   ASSERT_EQ(job2, old2);
   ASSERT_GT(job1, old1);
@@ -159,3 +180,12 @@ int main(int argc, char **argv) {
   ::testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();
 }
+#else
+#include <stdio.h>
+
+int main(int /*argc*/, char** /*argv*/) {
+  fprintf(stderr, "SKIPPED as CloudSchedulerTest is not supported in ROCKSDB_LITE\n");
+  return 0;
+}
+
+#endif  // ROCKSDB_LITE

cloud/cloud_storage_provider.cc

@@ -20,7 +20,7 @@
 #include "util/string_util.h"
 
 namespace ROCKSDB_NAMESPACE {
-
+#ifndef ROCKSDB_LITE
 /******************** Readablefile ******************/
 CloudStorageReadableFileImpl::CloudStorageReadableFileImpl(
     const std::shared_ptr<Logger>& info_log, const std::string& bucket,
@@ -361,4 +361,5 @@ Status CloudStorageProviderImpl::PutCloudObject(
   return DoPutCloudObject(local_file, bucket_name, object_path, fsize);
 }
 
+#endif //ROCKSDB_LITE
 }  // namespace ROCKSDB_NAMESPACE

cloud/db_cloud_test.cc

@@ -64,8 +64,9 @@ class CloudTest : public testing::Test {
 
     CloudEnv* aenv;
     // create a dummy aws env
-    CloudEnv::NewAwsEnv(base_env_, cloud_env_options_, options_.info_log,
-                        &aenv);
+    ASSERT_OK(CloudEnv::NewAwsEnv(base_env_, cloud_env_options_,
+                                  options_.info_log, &aenv));
+    ASSERT_NE(aenv, nullptr);
     aenv_.reset(aenv);
     // delete all pre-existing contents from the bucket
     Status st = aenv_->GetCloudEnvOptions().storage_provider->EmptyBucket(
@@ -1559,7 +1560,7 @@ int main(int, char**) {
 
 #include <stdio.h>
 
-int main(int argc, char** argv) {
+int main(int, char** ) {
   fprintf(stderr, "SKIPPED as DBCloud is not supported in ROCKSDB_LITE\n");
   return 0;
 }

cloud/remote_compaction_test.cc

@@ -517,7 +517,7 @@ int main(int, char**) {
 
 #include <stdio.h>
 
-int main(int argc, char** argv) {
+int main(int, char**) {
   fprintf(stderr, "SKIPPED as DBCloud is not supported in ROCKSDB_LITE\n");
   return 0;
 }

db/db_impl/db_impl_remote_compaction.cc

@@ -151,13 +151,15 @@ Status DBImpl::doCompact(const CompactionOptions& compact_options,
   // remove this compaction job from the list of compactions
   c->ReleaseCompactionFiles(s);
 
+#ifndef ROCKSDB_LITE
   // Make sure SstFileManager does its bookkeeping
   auto sfm = static_cast<SstFileManagerImpl*>(
       immutable_db_options_.sst_file_manager.get());
   if (sfm) {
     sfm->OnCompactionCompletion(c.get());
   }
-
+#endif // ROCKSDB_LITE
+  
   // remove our iterator
   ReleaseFileNumberFromPendingOutputs(pending_outputs_inserted_elem);