Encapsulate path access in StressTest via accessors (#14756)

Summary:
Pull Request resolved: https://github.com/facebook/rocksdb/pull/14756

Pure mechanical refactor: replace all direct FLAGS_db / FLAGS_expected_values_dir / FLAGS_secondaries_base reads with accessor methods on StressTest. No new flags, no parameters, no behavior change. Prepares for multi-DB stress test where each StressTest instance has its own DB.

Changes:
- GetDbPath(), GetExpectedValuesDir(), GetSecondariesBase() accessors return the corresponding FLAGS values directly
- Replace ~15 FLAGS_db references with GetDbPath() in db_stress_test_base.cc
- Move SharedState constructor from .h to .cc (needs full StressTest type for GetExpectedValuesDir())
- Move DbStressListener constructor from .h to .cc (same reason)
- Replace FLAGS_db / FLAGS_expected_values_dir in db_stress_driver.cc with accessor calls
- NO changes to db_crashtest.py or db_stress_gflags.cc

Reviewed By: anand1976

Differential Revision: D104959943

fbshipit-source-id: d7ef6a39d4c2ed467b2960417629c09f3988faf5

db_stress_tool/db_stress_driver.cc

@@ -68,9 +68,11 @@ bool RunStressTestImpl(SharedState* shared) {
   StressTest* stress = shared->GetStressTest();
 
   if (shared->ShouldVerifyAtBeginning() && FLAGS_preserve_unverified_changes) {
-    Status s = InitUnverifiedSubdir(FLAGS_db);
-    if (s.ok() && !FLAGS_expected_values_dir.empty()) {
-      s = InitUnverifiedSubdir(FLAGS_expected_values_dir);
+    const auto& db_path = stress->GetDbPath();
+    const auto& ev_dir = stress->GetExpectedValuesDir();
+    Status s = InitUnverifiedSubdir(db_path);
+    if (s.ok() && !ev_dir.empty()) {
+      s = InitUnverifiedSubdir(ev_dir);
     }
     if (!s.ok()) {
       fprintf(stderr, "Failed to setup unverified state dir: %s\n",
@@ -159,9 +161,10 @@ bool RunStressTestImpl(SharedState* shared) {
         fprintf(stderr, "Crash-recovery verification failed :(\n");
       } else {
         fprintf(stdout, "Crash-recovery verification passed :)\n");
-        Status s = DestroyUnverifiedSubdir(FLAGS_db);
-        if (s.ok() && !FLAGS_expected_values_dir.empty()) {
-          s = DestroyUnverifiedSubdir(FLAGS_expected_values_dir);
+        const auto& ev_dir = stress->GetExpectedValuesDir();
+        Status s = DestroyUnverifiedSubdir(stress->GetDbPath());
+        if (s.ok() && !ev_dir.empty()) {
+          s = DestroyUnverifiedSubdir(ev_dir);
         }
         if (!s.ok()) {
           fprintf(stderr, "Failed to cleanup unverified state dir: %s\n",
@@ -174,7 +177,7 @@ bool RunStressTestImpl(SharedState* shared) {
     if (!FLAGS_verification_only) {
       // This is after the verification step to avoid making all those `Get()`s
       // and `MultiGet()`s contend on the DB-wide trace mutex.
-      if (!FLAGS_expected_values_dir.empty()) {
+      if (!stress->GetExpectedValuesDir().empty()) {
         stress->TrackExpectedState(shared);
       }
 

db_stress_tool/db_stress_listener.cc

@@ -7,6 +7,7 @@
 
 #include <cstdint>
 
+#include "db_stress_tool/db_stress_test_base.h"
 #include "file/file_util.h"
 #include "rocksdb/file_system.h"
 #include "util/coding_lean.h"
@@ -15,6 +16,19 @@ namespace ROCKSDB_NAMESPACE {
 
 #ifdef GFLAGS
 
+DbStressListener::DbStressListener(
+    const std::string& db_name, const std::vector<DbPath>& db_paths,
+    const std::vector<ColumnFamilyDescriptor>& column_families,
+    SharedState* shared)
+    : db_name_(db_name),
+      db_paths_(db_paths),
+      column_families_(column_families),
+      num_pending_file_creations_(0),
+      unique_ids_(shared->GetStressTest()->GetExpectedValuesDir().empty()
+                      ? db_name
+                      : shared->GetStressTest()->GetExpectedValuesDir()),
+      shared_(shared) {}
+
 UniqueIdVerifier::UniqueIdVerifier(const std::string& dir)
     : path_(dir + "/.unique_ids") {
   // We expect such a small number of files generated during this test

db_stress_tool/db_stress_listener.h

@@ -58,15 +58,7 @@ class DbStressListener : public EventListener {
   DbStressListener(const std::string& db_name,
                    const std::vector<DbPath>& db_paths,
                    const std::vector<ColumnFamilyDescriptor>& column_families,
-                   SharedState* shared)
-      : db_name_(db_name),
-        db_paths_(db_paths),
-        column_families_(column_families),
-        num_pending_file_creations_(0),
-        unique_ids_(FLAGS_expected_values_dir.empty()
-                        ? db_name
-                        : FLAGS_expected_values_dir),
-        shared_(shared) {}
+                   SharedState* shared);
 
   const char* Name() const override { return kClassName(); }
   static const char* kClassName() { return "DBStressListener"; }

db_stress_tool/db_stress_shared_state.cc

@@ -11,7 +11,109 @@
 #ifdef GFLAGS
 #include "db_stress_tool/db_stress_shared_state.h"
 
+#include "db_stress_tool/db_stress_test_base.h"
+
 namespace ROCKSDB_NAMESPACE {
 thread_local bool SharedState::ignore_read_error;
+
+SharedState::SharedState(Env* /*env*/, StressTest* stress_test)
+    : cv_(&mu_),
+      seed_(static_cast<uint32_t>(FLAGS_seed)),
+      max_key_(FLAGS_max_key),
+      log2_keys_per_lock_(static_cast<uint32_t>(FLAGS_log2_keys_per_lock)),
+      num_threads_(0),
+      num_initialized_(0),
+      num_populated_(0),
+      vote_reopen_(0),
+      num_done_(0),
+      start_(false),
+      start_verify_(false),
+      num_bg_threads_(0),
+      should_stop_bg_thread_(false),
+      bg_thread_finished_(0),
+      stress_test_(stress_test),
+      verification_failure_(false),
+      should_stop_test_(false),
+      no_overwrite_ids_(GenerateNoOverwriteIds()),
+      expected_state_manager_(nullptr),
+      printing_verification_results_(false),
+      start_timestamp_(Env::Default()->NowNanos()) {
+  Status status;
+  // TODO: We should introduce a way to explicitly disable verification
+  // during shutdown. When that is disabled and FLAGS_expected_values_dir
+  // is empty (disabling verification at startup), we can skip tracking
+  // expected state. Only then should we permit bypassing the below feature
+  // compatibility checks.
+  const auto& expected_values_dir = stress_test_->GetExpectedValuesDir();
+  if (!expected_values_dir.empty()) {
+    if (!std::atomic<uint32_t>{}.is_lock_free() ||
+        !std::atomic<uint64_t>{}.is_lock_free()) {
+      std::ostringstream status_s;
+      status_s << "Cannot use --expected_values_dir on platforms without "
+                  "lock-free "
+               << (!std::atomic<uint32_t>{}.is_lock_free()
+                       ? "std::atomic<uint32_t>"
+                       : "std::atomic<uint64_t>");
+      status = Status::InvalidArgument(status_s.str());
+    }
+
+    if (status.ok() && FLAGS_clear_column_family_one_in > 0) {
+      status = Status::InvalidArgument(
+          "Cannot use --expected_values_dir on when "
+          "--clear_column_family_one_in is greater than zero.");
+    }
+  }
+  if (status.ok()) {
+    if (expected_values_dir.empty()) {
+      expected_state_manager_.reset(
+          new AnonExpectedStateManager(FLAGS_max_key, FLAGS_column_families));
+    } else {
+      expected_state_manager_.reset(new FileExpectedStateManager(
+          FLAGS_max_key, FLAGS_column_families, expected_values_dir));
+    }
+    status = expected_state_manager_->Open();
+  }
+  if (!status.ok()) {
+    fprintf(stderr, "Failed setting up expected state with error: %s\n",
+            status.ToString().c_str());
+    exit(1);  // NOLINT(concurrency-mt-unsafe)
+  }
+
+  if (FLAGS_test_batches_snapshots) {
+    fprintf(stdout, "No lock creation because test_batches_snapshots set\n");
+    return;
+  }
+
+  long num_locks = static_cast<long>(max_key_ >> log2_keys_per_lock_);
+  if (max_key_ & ((1 << log2_keys_per_lock_) - 1)) {
+    num_locks++;
+  }
+  fprintf(stdout, "Creating %ld locks\n", num_locks * FLAGS_column_families);
+  key_locks_.resize(FLAGS_column_families);
+
+  for (int i = 0; i < FLAGS_column_families; ++i) {
+    key_locks_[i].reset(new port::Mutex[num_locks]);
+  }
+  if (FLAGS_read_fault_one_in || FLAGS_metadata_read_fault_one_in) {
+#ifdef NDEBUG
+    // Unsupported in release mode because it relies on
+    // `IGNORE_STATUS_IF_ERROR` to distinguish faults not expected to lead to
+    // failure.
+    fprintf(stderr,
+            "Cannot set nonzero value for --read_fault_one_in in "
+            "release mode.");
+    exit(1);  // NOLINT(concurrency-mt-unsafe)
+#else         // NDEBUG
+    SyncPoint::GetInstance()->SetCallBack("FaultInjectionIgnoreError",
+                                          IgnoreReadErrorCallback);
+    SyncPoint::GetInstance()->EnableProcessing();
+#endif        // NDEBUG
+  }
+}
+
+bool SharedState::ShouldVerifyAtBeginning() const {
+  return !stress_test_->GetExpectedValuesDir().empty();
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 #endif  // GFLAGS

db_stress_tool/db_stress_shared_state.h

@@ -78,99 +78,7 @@ class SharedState {
   // for those calls
   static thread_local bool ignore_read_error;
 
-  SharedState(Env* /*env*/, StressTest* stress_test)
-      : cv_(&mu_),
-        seed_(static_cast<uint32_t>(FLAGS_seed)),
-        max_key_(FLAGS_max_key),
-        log2_keys_per_lock_(static_cast<uint32_t>(FLAGS_log2_keys_per_lock)),
-        num_threads_(0),
-        num_initialized_(0),
-        num_populated_(0),
-        vote_reopen_(0),
-        num_done_(0),
-        start_(false),
-        start_verify_(false),
-        num_bg_threads_(0),
-        should_stop_bg_thread_(false),
-        bg_thread_finished_(0),
-        stress_test_(stress_test),
-        verification_failure_(false),
-        should_stop_test_(false),
-        no_overwrite_ids_(GenerateNoOverwriteIds()),
-        expected_state_manager_(nullptr),
-        printing_verification_results_(false),
-        start_timestamp_(Env::Default()->NowNanos()) {
-    Status status;
-    // TODO: We should introduce a way to explicitly disable verification
-    // during shutdown. When that is disabled and FLAGS_expected_values_dir
-    // is empty (disabling verification at startup), we can skip tracking
-    // expected state. Only then should we permit bypassing the below feature
-    // compatibility checks.
-    if (!FLAGS_expected_values_dir.empty()) {
-      if (!std::atomic<uint32_t>{}.is_lock_free() ||
-          !std::atomic<uint64_t>{}.is_lock_free()) {
-        std::ostringstream status_s;
-        status_s << "Cannot use --expected_values_dir on platforms without "
-                    "lock-free "
-                 << (!std::atomic<uint32_t>{}.is_lock_free()
-                         ? "std::atomic<uint32_t>"
-                         : "std::atomic<uint64_t>");
-        status = Status::InvalidArgument(status_s.str());
-      }
-
-      if (status.ok() && FLAGS_clear_column_family_one_in > 0) {
-        status = Status::InvalidArgument(
-            "Cannot use --expected_values_dir on when "
-            "--clear_column_family_one_in is greater than zero.");
-      }
-    }
-    if (status.ok()) {
-      if (FLAGS_expected_values_dir.empty()) {
-        expected_state_manager_.reset(
-            new AnonExpectedStateManager(FLAGS_max_key, FLAGS_column_families));
-      } else {
-        expected_state_manager_.reset(new FileExpectedStateManager(
-            FLAGS_max_key, FLAGS_column_families, FLAGS_expected_values_dir));
-      }
-      status = expected_state_manager_->Open();
-    }
-    if (!status.ok()) {
-      fprintf(stderr, "Failed setting up expected state with error: %s\n",
-              status.ToString().c_str());
-      exit(1);
-    }
-
-    if (FLAGS_test_batches_snapshots) {
-      fprintf(stdout, "No lock creation because test_batches_snapshots set\n");
-      return;
-    }
-
-    long num_locks = static_cast<long>(max_key_ >> log2_keys_per_lock_);
-    if (max_key_ & ((1 << log2_keys_per_lock_) - 1)) {
-      num_locks++;
-    }
-    fprintf(stdout, "Creating %ld locks\n", num_locks * FLAGS_column_families);
-    key_locks_.resize(FLAGS_column_families);
-
-    for (int i = 0; i < FLAGS_column_families; ++i) {
-      key_locks_[i].reset(new port::Mutex[num_locks]);
-    }
-    if (FLAGS_read_fault_one_in || FLAGS_metadata_read_fault_one_in) {
-#ifdef NDEBUG
-      // Unsupported in release mode because it relies on
-      // `IGNORE_STATUS_IF_ERROR` to distinguish faults not expected to lead to
-      // failure.
-      fprintf(stderr,
-              "Cannot set nonzero value for --read_fault_one_in in "
-              "release mode.");
-      exit(1);
-#else   // NDEBUG
-      SyncPoint::GetInstance()->SetCallBack("FaultInjectionIgnoreError",
-                                            IgnoreReadErrorCallback);
-      SyncPoint::GetInstance()->EnableProcessing();
-#endif  // NDEBUG
-    }
-  }
+  SharedState(Env* env, StressTest* stress_test);
 
   ~SharedState() {
 #ifndef NDEBUG
@@ -430,9 +338,7 @@ class SharedState {
     return bg_thread_finished_ == num_bg_threads_;
   }
 
-  bool ShouldVerifyAtBeginning() const {
-    return !FLAGS_expected_values_dir.empty();
-  }
+  bool ShouldVerifyAtBeginning() const;
 
   bool PrintingVerificationResults() {
     bool tmp = false;

db_stress_tool/db_stress_test_base.cc

@@ -65,6 +65,16 @@ std::shared_ptr<const FilterPolicy> CreateFilterPolicy() {
 
 }  // namespace
 
+const std::string& StressTest::GetDbPath() const { return FLAGS_db; }
+
+const std::string& StressTest::GetExpectedValuesDir() const {
+  return FLAGS_expected_values_dir;
+}
+
+const std::string& StressTest::GetSecondariesBase() const {
+  return FLAGS_secondaries_base;
+}
+
 StressTest::StressTest()
     : cache_(NewCache(FLAGS_cache_size, FLAGS_cache_numshardbits)),
       filter_policy_(CreateFilterPolicy()),
@@ -81,7 +91,7 @@ StressTest::StressTest()
       manifest_file_number_before_reopen_(0),
       manifest_file_size_before_reopen_(0) {
   if (FLAGS_destroy_db_initially) {
-    const Status s = DbStressDestroyDb(FLAGS_db);
+    const Status s = DbStressDestroyDb(GetDbPath());
     if (!s.ok()) {
       fprintf(stderr, "Cannot destroy original db: %s\n", s.ToString().c_str());
       exit(1);
@@ -1360,7 +1370,7 @@ void StressTest::OperateDb(ThreadState* thread) {
         uint64_t total_size = 0;
         if (FLAGS_backup_max_size > 0) {
           std::vector<FileAttributes> files;
-          db_stress_env->GetChildrenFileAttributes(FLAGS_db, &files);
+          db_stress_env->GetChildrenFileAttributes(GetDbPath(), &files);
           for (auto& file : files) {
             total_size += file.size_bytes;
           }
@@ -2483,9 +2493,9 @@ Status StressTest::TestBackupRestore(
   }
 
   const std::string backup_dir =
-      FLAGS_db + "/.backup" + std::to_string(thread->tid);
+      GetDbPath() + "/.backup" + std::to_string(thread->tid);
   const std::string restore_dir =
-      FLAGS_db + "/.restore" + std::to_string(thread->tid);
+      GetDbPath() + "/.restore" + std::to_string(thread->tid);
   BackupEngineOptions backup_opts(backup_dir);
   // For debugging, get info_log from live options
   backup_opts.info_log = db_->GetDBOptions().info_log.get();
@@ -2949,7 +2959,7 @@ Status StressTest::TestCheckpoint(ThreadState* thread,
   }
 
   std::string checkpoint_dir =
-      FLAGS_db + "/.checkpoint" + std::to_string(thread->tid);
+      GetDbPath() + "/.checkpoint" + std::to_string(thread->tid);
   Options tmp_opts(options_);
   tmp_opts.listeners.clear();
   tmp_opts.env = db_stress_env;
@@ -3810,6 +3820,7 @@ void StressTest::Open(SharedState* shared, bool reopen) {
   assert(db_ == nullptr);
   assert(txn_db_ == nullptr);
   assert(optimistic_txn_db_ == nullptr);
+  const auto& db_path = GetDbPath();
   if (FLAGS_use_trie_index) {
     udi_factory_ = std::make_shared<trie_index::TrieIndexFactory>();
   }
@@ -3919,13 +3930,13 @@ void StressTest::Open(SharedState* shared, bool reopen) {
     fprintf(stdout, "Integrated BlobDB: blob cache disabled\n");
   }
 
-  fprintf(stdout, "DB path: [%s]\n", FLAGS_db.c_str());
+  fprintf(stdout, "DB path: [%s]\n", db_path.c_str());
 
   Status s;
 
   if (FLAGS_ttl == -1) {
     std::vector<std::string> existing_column_families;
-    s = DB::ListColumnFamilies(DBOptions(options_), FLAGS_db,
+    s = DB::ListColumnFamilies(DBOptions(options_), db_path,
                                &existing_column_families);  // ignore errors
     if (!s.ok()) {
       // DB doesn't exist
@@ -3974,7 +3985,7 @@ void StressTest::Open(SharedState* shared, bool reopen) {
 
     options_.listeners.clear();
     options_.listeners.emplace_back(new DbStressListener(
-        FLAGS_db, options_.db_paths, cf_descriptors, shared));
+        db_path, options_.db_paths, cf_descriptors, shared));
     RegisterAdditionalListeners();
 
     if (!FLAGS_listener_uri.empty()) {
@@ -4007,8 +4018,7 @@ void StressTest::Open(SharedState* shared, bool reopen) {
       if ((inject_sync_fault || inject_open_meta_read_error ||
            inject_open_meta_write_error || inject_open_read_error ||
            inject_open_write_error) &&
-          fault_fs_guard
-              ->FileExists(FLAGS_db + "/CURRENT", IOOptions(), nullptr)
+          fault_fs_guard->FileExists(db_path + "/CURRENT", IOOptions(), nullptr)
               .ok()) {
         if (inject_sync_fault || inject_open_write_error) {
           fault_fs_guard->SetFilesystemDirectWritable(false);
@@ -4052,7 +4062,7 @@ void StressTest::Open(SharedState* shared, bool reopen) {
           blob_db_options.enable_garbage_collection = FLAGS_blob_db_enable_gc;
 
           blob_db::BlobDB* blob_db = nullptr;
-          s = blob_db::BlobDB::Open(options_, blob_db_options, FLAGS_db,
+          s = blob_db::BlobDB::Open(options_, blob_db_options, db_path,
                                     cf_descriptors, &column_families_,
                                     &blob_db);
           if (s.ok()) {
@@ -4061,11 +4071,11 @@ void StressTest::Open(SharedState* shared, bool reopen) {
           }
         } else {
           if (db_preload_finished_.load() && FLAGS_read_only) {
-            s = DB::OpenForReadOnly(DBOptions(options_), FLAGS_db,
+            s = DB::OpenForReadOnly(DBOptions(options_), db_path,
                                     cf_descriptors, &column_families_,
                                     &db_owner_);
           } else {
-            s = DB::Open(DBOptions(options_), FLAGS_db, cf_descriptors,
+            s = DB::Open(DBOptions(options_), db_path, cf_descriptors,
                          &column_families_, &db_owner_);
           }
           if (s.ok()) {
@@ -4135,7 +4145,7 @@ void StressTest::Open(SharedState* shared, bool reopen) {
           optimistic_txn_db_options.shared_lock_buckets = nullptr;
         }
         s = OptimisticTransactionDB::Open(
-            options_, optimistic_txn_db_options, FLAGS_db, cf_descriptors,
+            options_, optimistic_txn_db_options, db_path, cf_descriptors,
             &column_families_, &optimistic_txn_db_);
         if (!s.ok()) {
           fprintf(stderr, "Error in opening the OptimisticTransactionDB [%s]\n",
@@ -4173,7 +4183,7 @@ void StressTest::Open(SharedState* shared, bool reopen) {
         txn_db_options.use_per_key_point_lock_mgr =
             FLAGS_use_per_key_point_lock_mgr;
         PrepareTxnDbOptions(shared, txn_db_options);
-        s = TransactionDB::Open(options_, txn_db_options, FLAGS_db,
+        s = TransactionDB::Open(options_, txn_db_options, db_path,
                                 cf_descriptors, &column_families_, &txn_db_);
         if (!s.ok()) {
           fprintf(stderr, "Error in opening the TransactionDB [%s]\n",
@@ -4213,16 +4223,16 @@ void StressTest::Open(SharedState* shared, bool reopen) {
       // TODO(yanqin) support max_open_files != -1 for secondary instance.
       tmp_opts.max_open_files = -1;
       tmp_opts.env = db_stress_env;
-      const std::string& secondary_path = FLAGS_secondaries_base;
-      s = DB::OpenAsSecondary(tmp_opts, FLAGS_db, secondary_path,
-                              cf_descriptors, &secondary_cfhs_, &secondary_db_);
+      const std::string& secondary_path = GetSecondariesBase();
+      s = DB::OpenAsSecondary(tmp_opts, db_path, secondary_path, cf_descriptors,
+                              &secondary_cfhs_, &secondary_db_);
       assert(s.ok());
       assert(secondary_cfhs_.size() ==
              static_cast<size_t>(FLAGS_column_families));
     }
   } else {
     DBWithTTL* db_with_ttl;
-    s = DBWithTTL::Open(options_, FLAGS_db, &db_with_ttl, FLAGS_ttl);
+    s = DBWithTTL::Open(options_, db_path, &db_with_ttl, FLAGS_ttl);
     db_owner_.reset(db_with_ttl);
     db_ = db_with_ttl;
   }

db_stress_tool/db_stress_test_base.h

@@ -44,6 +44,10 @@ class StressTest {
 
   virtual ~StressTest() {}
 
+  const std::string& GetDbPath() const;
+  const std::string& GetExpectedValuesDir() const;
+  const std::string& GetSecondariesBase() const;
+
   std::shared_ptr<Cache> NewCache(size_t capacity, int32_t num_shard_bits);
 
   static std::vector<std::string> GetBlobCompressionTags();

db_stress_tool/no_batched_ops_stress.cc

@@ -2325,11 +2325,11 @@ class NonBatchedOpsStressTest : public StressTest {
         FLAGS_test_ingest_standalone_range_deletion_one_in);
     std::vector<std::string> external_files;
     const std::string sst_filename =
-        FLAGS_db + "/." + std::to_string(thread->tid) + ".sst";
+        GetDbPath() + "/." + std::to_string(thread->tid) + ".sst";
     external_files.push_back(sst_filename);
     std::string standalone_rangedel_filename;
     if (test_standalone_range_deletion) {
-      standalone_rangedel_filename = FLAGS_db + "/." +
+      standalone_rangedel_filename = GetDbPath() + "/." +
                                      std::to_string(thread->tid) +
                                      "_standalone_rangedel.sst";
       external_files.push_back(standalone_rangedel_filename);