Merge pull request #94 from rockset/DbStress

Fix DB Bench tool for AwsEnv

Author: hicder

cloud/aws/aws_env.h

@@ -21,8 +21,8 @@
 #include <unordered_map>
 
 namespace ROCKSDB_NAMESPACE {
-
 class S3ReadableFile;
+
 //
 // The S3 environment for rocksdb. This class overrides all the
 // file/dir access methods and delegates all other methods to the

cloud/aws/aws_file.h

@@ -4,6 +4,7 @@
 #ifdef USE_AWS
 
 #include <aws/core/Aws.h>
+
 #include "rocksdb/rocksdb_namespace.h"
 
 namespace ROCKSDB_NAMESPACE {

cloud/examples/clone_example.cc

@@ -44,10 +44,9 @@ Status CloneDB(const std::string& clone_name, const std::string& src_bucket,
 
   // Create new AWS env
   CloudEnv* cenv;
-  Status st = CloudEnv::NewAwsEnv(Env::Default(),
-                                  src_bucket, src_object_path, kRegion,
-                                  dest_bucket, dest_object_path, kRegion,
-                                  cloud_env_options, nullptr, &cenv);
+  Status st = CloudEnv::NewAwsEnv(Env::Default(), src_bucket, src_object_path,
+                                  kRegion, dest_bucket, dest_object_path,
+                                  kRegion, cloud_env_options, nullptr, &cenv);
   if (!st.ok()) {
     fprintf(stderr,
             "Unable to create an AWS environment with "
@@ -107,16 +106,14 @@ int main() {
   const std::string bucketName = bucketPrefix + kBucketSuffix;
 
   // Needed if using bucket prefix other than the default "rockset."
-  cloud_env_options.src_bucket.SetBucketName(kBucketSuffix,bucketPrefix);
-  cloud_env_options.dest_bucket.SetBucketName(kBucketSuffix,bucketPrefix);
-  
+  cloud_env_options.src_bucket.SetBucketName(kBucketSuffix, bucketPrefix);
+  cloud_env_options.dest_bucket.SetBucketName(kBucketSuffix, bucketPrefix);
+
   // Create a new AWS cloud env Status
   CloudEnv* cenv;
-  Status s =
-      CloudEnv::NewAwsEnv(Env::Default(),
-                          kBucketSuffix, kDBPath, kRegion,
-                          kBucketSuffix, kDBPath, kRegion,
-                          cloud_env_options, nullptr, &cenv);
+  Status s = CloudEnv::NewAwsEnv(Env::Default(), kBucketSuffix, kDBPath,
+                                 kRegion, kBucketSuffix, kDBPath, kRegion,
+                                 cloud_env_options, nullptr, &cenv);
   if (!s.ok()) {
     fprintf(stderr, "Unable to create cloud env in bucket %s. %s\n",
             bucketName.c_str(), s.ToString().c_str());

cloud/examples/cloud_dump.cc

@@ -44,19 +44,17 @@ int main() {
 
   // "rockset." is the default bucket prefix
   const std::string bucketPrefix = "rockset.";
-  cloud_env_options.src_bucket.SetBucketName(kBucketSuffix,bucketPrefix);
-  cloud_env_options.dest_bucket.SetBucketName(kBucketSuffix,bucketPrefix);
-  
+  cloud_env_options.src_bucket.SetBucketName(kBucketSuffix, bucketPrefix);
+  cloud_env_options.dest_bucket.SetBucketName(kBucketSuffix, bucketPrefix);
+
   // create a bucket name for debugging purposes
   const std::string bucketName = bucketPrefix + kBucketSuffix;
 
   // Create a new AWS cloud env Status
   CloudEnv* cenv;
-  Status s =
-      CloudEnv::NewAwsEnv(Env::Default(),
-                          kBucketSuffix, kDBPath, kRegion,
-                          kBucketSuffix, kDBPath, kRegion,
-                          cloud_env_options, nullptr, &cenv);
+  Status s = CloudEnv::NewAwsEnv(Env::Default(), kBucketSuffix, kDBPath,
+                                 kRegion, kBucketSuffix, kDBPath, kRegion,
+                                 cloud_env_options, nullptr, &cenv);
   if (!s.ok()) {
     fprintf(stderr, "Unable to create cloud env in bucket %s. %s\n",
             bucketName.c_str(), s.ToString().c_str());
@@ -95,7 +93,7 @@ int main() {
   // verify that the data is somewhat sane by manaully scanning for cfs
   std::vector<std::string> cf_names;
   s = ROCKSDB_NAMESPACE::DB::ListColumnFamilies(options, kDBPath, &cf_names);
-  for (std::string cf: cf_names) {
+  for (const std::string & cf : cf_names) {
     std::cout << " Found Column Family " << cf;
   }
   std::cout << " \n";

cloud/examples/cloud_durable_example.cc

@@ -47,19 +47,17 @@ int main() {
 
   // "rockset." is the default bucket prefix
   const std::string bucketPrefix = "rockset.";
-  cloud_env_options.src_bucket.SetBucketName(kBucketSuffix,bucketPrefix);
-  cloud_env_options.dest_bucket.SetBucketName(kBucketSuffix,bucketPrefix);
+  cloud_env_options.src_bucket.SetBucketName(kBucketSuffix, bucketPrefix);
+  cloud_env_options.dest_bucket.SetBucketName(kBucketSuffix, bucketPrefix);
 
   // create a bucket name for debugging purposes
   const std::string bucketName = bucketPrefix + kBucketSuffix;
-  
+
   // Create a new AWS cloud env Status
   CloudEnv* cenv;
-  Status s =
-      CloudEnv::NewAwsEnv(Env::Default(),
-                          kBucketSuffix, kDBPath, kRegion,
-                          kBucketSuffix, kDBPath, kRegion,
-                          cloud_env_options, nullptr, &cenv);
+  Status s = CloudEnv::NewAwsEnv(Env::Default(), kBucketSuffix, kDBPath,
+                                 kRegion, kBucketSuffix, kDBPath, kRegion,
+                                 cloud_env_options, nullptr, &cenv);
   if (!s.ok()) {
     fprintf(stderr, "Unable to create cloud env in bucket %s. %s\n",
             bucketName.c_str(), s.ToString().c_str());

cloud/remote_compaction_test.cc

@@ -4,10 +4,9 @@
 
 #ifdef USE_AWS
 
-#include <cinttypes>
-
 #include <algorithm>
 #include <chrono>
+#include <cinttypes>
 
 #include "cloud/aws/aws_env.h"
 #include "cloud/aws/aws_file.h"

db/db_impl/db_impl.h

@@ -838,8 +838,7 @@ class DBImpl : public DB {
   // Allow compaction with a set of explicitly specified snapshots
   virtual Status ExecuteRemoteCompactionRequest(
       const PluggableCompactionParam& inputParams,
-      PluggableCompactionResult* result,
-      bool sanitize) override;
+      PluggableCompactionResult* result, bool sanitize) override;
 
   // This registered service will be called to do a remote compaction
   virtual Status RegisterPluggableCompactionService(
@@ -2094,15 +2093,12 @@ class DBImpl : public DB {
   std::shared_ptr<PluggableCompactionService> remote_compaction_service_;
 
   Status doCompact(const CompactionOptions& compact_options,
-      ColumnFamilyData* cfd,
-      Version* version,
-      const std::vector<FilesInOneLevel>& input_file_names,
-      int output_level,
-      const std::vector<SequenceNumber>& existing_snapshots,
-      bool unitTests,
-      JobContext* job_context,
-      LogBuffer* log_buffer,
-      PluggableCompactionResult* result);
+                   ColumnFamilyData* cfd, Version* version,
+                   const std::vector<FilesInOneLevel>& input_file_names,
+                   int output_level,
+                   const std::vector<SequenceNumber>& existing_snapshots,
+                   bool unitTests, JobContext* job_context,
+                   LogBuffer* log_buffer, PluggableCompactionResult* result);
 
   bool wal_in_db_path_;
 };

db/db_test_util.cc

@@ -8,9 +8,10 @@
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 
 #include "db/db_test_util.h"
+
 #include "db/forward_iterator.h"
-#include "util/stderr_logger.h"
 #include "rocksdb/env_encryption.h"
+#include "util/stderr_logger.h"
 #ifdef USE_AWS
 #include "cloud/cloud_env_impl.h"
 #include "rocksdb/cloud/cloud_storage_provider.h"
@@ -85,7 +86,7 @@ DBTestBase::DBTestBase(const std::string path)
   // Randomize the test path so that multiple tests can run in parallel
   srand(static_cast<unsigned int>(time(nullptr)));
   std::string mypath = path + "_" + std::to_string(rand());
-  
+
   env_->NewLogger(test::TmpDir(env_) + "/rocksdb-cloud.log", &info_log_);
   info_log_->SetInfoLogLevel(InfoLogLevel::DEBUG_LEVEL);
   s3_env_ = CreateNewAwsEnv(mypath, env_);
@@ -187,45 +188,45 @@ bool DBTestBase::ShouldSkipOptions(int option_config, int skip_mask) {
 }
 
 bool DBTestBase::ShouldSkipAwsOptions(int option_config) {
-    // AWS Env doesn't work with DirectIO
-    return option_config == kDirectIO;
+  // AWS Env doesn't work with DirectIO
+  return option_config == kDirectIO;
 }
 
 // Switch to a fresh database with the next option configuration to
 // test.  Return false if there are no more configurations to test.
 bool DBTestBase::ChangeOptions(int skip_mask) {
   while (true) {
-   for (option_config_++; option_config_ < kEnd; option_config_++) {
-     if (ShouldSkipOptions(option_config_, skip_mask)) {
-       continue;
-     }
-     if (option_env_ == kAwsEnv && ShouldSkipAwsOptions(option_config_)) {
-         continue;
-     }
-     break;
-   }
-   if (option_config_ >= kEnd) {
+    for (option_config_++; option_config_ < kEnd; option_config_++) {
+      if (ShouldSkipOptions(option_config_, skip_mask)) {
+        continue;
+      }
+      if (option_env_ == kAwsEnv && ShouldSkipAwsOptions(option_config_)) {
+        continue;
+      }
+      break;
+    }
+    if (option_config_ >= kEnd) {
 #ifndef USE_AWS
-     // If not built for AWS, skip it
-     if (option_env_ + 1 == kAwsEnv) {
-       option_env_++;
-     }
+      // If not built for AWS, skip it
+      if (option_env_ + 1 == kAwsEnv) {
+        option_env_++;
+      }
 #endif
-     if (option_env_ + 1 >= kEndEnv) {
-       Destroy(last_options_);
-       return false;
-     } else {
-       option_env_++;
-       option_config_ = kDefault;
-       continue;
-     }
-   } else {
-     auto options = CurrentOptions();
-     options.create_if_missing = true;
-     DestroyAndReopen(options);
-     return true;
-   }
- }
+      if (option_env_ + 1 >= kEndEnv) {
+        Destroy(last_options_);
+        return false;
+      } else {
+        option_env_++;
+        option_config_ = kDefault;
+        continue;
+      }
+    } else {
+      auto options = CurrentOptions();
+      options.create_if_missing = true;
+      DestroyAndReopen(options);
+      return true;
+    }
+  }
 }
 
 // Switch between different compaction styles.
@@ -620,8 +621,8 @@ Options DBTestBase::GetOptions(
     case kAwsEnv: {
       assert(s3_env_);
       options.env = s3_env_;
-      options.recycle_log_file_num = 0; // do not reuse log files
-      options.allow_mmap_reads = false; // mmap is incompatible with S3
+      options.recycle_log_file_num = 0;  // do not reuse log files
+      options.allow_mmap_reads = false;  // mmap is incompatible with S3
       break;
     }
 #endif /* USE_AWS */
@@ -644,7 +645,7 @@ Options DBTestBase::GetOptions(
 }
 
 #ifdef USE_AWS
-  Env* DBTestBase::CreateNewAwsEnv(const std::string& prefix, Env *parent) {
+Env* DBTestBase::CreateNewAwsEnv(const std::string& prefix, Env* parent) {
   if (!prefix.empty()) {
     fprintf(stderr, "Creating new AWS env with prefix %s\n", prefix.c_str());
   }
@@ -747,7 +748,7 @@ void DBTestBase::Destroy(const Options& options, bool delete_cf_paths) {
   ASSERT_OK(DestroyDB(dbname_, options, column_families));
 #ifdef USE_AWS
   if (s3_env_) {
-    AwsEnv* aenv = static_cast<AwsEnv *>(s3_env_);
+    AwsEnv* aenv = static_cast<AwsEnv*>(s3_env_);
     Status st = aenv->GetCloudEnvOptions().storage_provider->EmptyBucket(
         aenv->GetSrcBucketName(), dbname_);
     ASSERT_TRUE(st.ok() || st.IsNotFound());

db/db_test_util.h

@@ -10,9 +10,9 @@
 #pragma once
 
 #include <fcntl.h>
-#include <cinttypes>
 
 #include <algorithm>
+#include <cinttypes>
 #include <map>
 #include <set>
 #include <string>
@@ -21,6 +21,7 @@
 #include <utility>
 #include <vector>
 
+#include "cloud/aws/aws_env.h"
 #include "db/db_impl/db_impl.h"
 #include "db/dbformat.h"
 #include "env/mock_env.h"
@@ -43,17 +44,14 @@
 #include "table/plain/plain_table_factory.h"
 #include "table/scoped_arena_iterator.h"
 #include "test_util/mock_time_env.h"
-#include "util/compression.h"
-#include "util/mutexlock.h"
-
 #include "test_util/sync_point.h"
 #include "test_util/testharness.h"
 #include "test_util/testutil.h"
+#include "util/compression.h"
+#include "util/mutexlock.h"
 #include "util/string_util.h"
 #include "utilities/merge_operators.h"
 
-#include "cloud/aws/aws_env.h"
-
 namespace ROCKSDB_NAMESPACE {
 
 namespace anon {
@@ -736,7 +734,7 @@ class DBTestBase : public testing::Test {
   };
 
 #ifdef USE_AWS
-  Env* CreateNewAwsEnv(const std::string& pathPrefix, Env *env);
+  Env* CreateNewAwsEnv(const std::string& pathPrefix, Env* env);
   std::shared_ptr<Logger> info_log_;
 #endif
 

db/flush_job.cc

@@ -387,8 +387,8 @@ Status FlushJob::WriteLevel0Table() {
           cfd_->ioptions()->compression_opts,
           mutable_cf_options_.paranoid_file_checks, cfd_->internal_stats(),
           TableFileCreationReason::kFlush, event_logger_, job_context_->job_id,
-          Env::IO_HIGH, &table_properties_, 0 /* level */,
-          creation_time, oldest_key_time, write_hint, current_time);
+          Env::IO_HIGH, &table_properties_, 0 /* level */, creation_time,
+          oldest_key_time, write_hint, current_time);
       LogFlush(db_options_.info_log);
     }
     ROCKS_LOG_INFO(db_options_.info_log,