Parallel checkpointing stress test (#15037)

Summary:
Adds db_stress / crash-test coverage for the parallel `CheckpointEngine` so it
is exercised alongside existing features (atomic flush, sync fault injection,
cf_consistency, backups).

- Opens one shared `CheckpointEngine` per db_stress process, configured by
  `--checkpoint_engine_max_background_operations` and
  `--checkpoint_engine_use_link_file_when_available`.
- `--parallel_checkpoint_one_in=N`: each checkpoint independently uses the
  parallel engine with probability 1/N, otherwise the legacy serial
  `Checkpoint` API, so serial and parallel paths run concurrently across threads.
- `db_crashtest.py` randomizes all three per run.

This coverage surfaced the missing destination fsync in the parallel copy path
fixed by the parent diff.

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

Reviewed By: xingbowang

Differential Revision: D114414578

Pulled By: rban1

fbshipit-source-id: 48e7920a62b9b663a1c6054fc60b5765667bcbe5

db_stress_tool/db_stress_common.h

@@ -228,6 +228,9 @@ DECLARE_uint64(sst_file_manager_bytes_per_truncate);
 DECLARE_int32(backup_one_in);
 DECLARE_uint64(backup_max_size);
 DECLARE_int32(checkpoint_one_in);
+DECLARE_int32(checkpoint_engine_max_background_operations);
+DECLARE_bool(checkpoint_engine_use_link_file_when_available);
+DECLARE_int32(parallel_checkpoint_one_in);
 DECLARE_int32(ingest_external_file_one_in);
 DECLARE_int32(ingest_external_file_width);
 DECLARE_int32(ingest_external_file_prepare_commit_one_in);

db_stress_tool/db_stress_gflags.cc

@@ -861,6 +861,19 @@ DEFINE_int32(checkpoint_one_in, 0,
              "every N operations on average.  0 indicates CreateCheckpoint() "
              "is disabled.");
 
+DEFINE_int32(checkpoint_engine_max_background_operations, 1,
+             "Background-thread parallelism of the shared CheckpointEngine "
+             "(always opened). Values <= 0 are treated as 1.");
+
+DEFINE_bool(checkpoint_engine_use_link_file_when_available, true,
+            "Whether the parallel CheckpointEngine may hard-link data files "
+            "instead of copying them. False forces the parallel copy path.");
+
+DEFINE_int32(parallel_checkpoint_one_in, 0,
+             "Each checkpoint uses the parallel CheckpointEngine with "
+             "probability 1/N, else the legacy serial API (0 = always serial, "
+             "1 = always parallel).");
+
 DEFINE_int32(ingest_external_file_one_in, 0,
              "If non-zero, then IngestExternalFile() will be called once for "
              "every N operations on average.  0 indicates IngestExternalFile() "

db_stress_tool/db_stress_test_base.cc

@@ -462,6 +462,18 @@ StressTest::StressTest(int db_index, const std::string& db_path,
             s.ToString().c_str());
     exit(1);
   }
+
+  CheckpointEngineOptions engine_options;
+  engine_options.max_background_operations =
+      FLAGS_checkpoint_engine_max_background_operations;
+  engine_options.use_link_file_when_available =
+      FLAGS_checkpoint_engine_use_link_file_when_available;
+  s = CheckpointEngine::Open(engine_options, &checkpoint_engine_);
+  if (!s.ok()) {
+    fprintf(stderr, "Error opening CheckpointEngine: %s\n",
+            s.ToString().c_str());
+    exit(1);
+  }
 }
 
 void StressTest::CleanUp() {
@@ -3606,36 +3618,45 @@ Status StressTest::TestCheckpoint(ThreadState* thread,
   if (db_fault_injection_fs_) {
     db_fault_injection_fs_->EnableAllThreadLocalErrorInjection();
   }
+  const bool use_parallel_engine =
+      thread->rand.OneInOpt(FLAGS_parallel_checkpoint_one_in);
   Checkpoint* checkpoint = nullptr;
-  Status s = Checkpoint::Create(db_, &checkpoint);
-  if (s.ok()) {
-    s = checkpoint->CreateCheckpoint(checkpoint_dir);
-    if (!s.ok() && !IsErrorInjectedAndRetryable(s)) {
-      fprintf(stderr, "Fail to create checkpoint to %s\n",
-              checkpoint_dir.c_str());
-      std::vector<std::string> files;
+  Status s;
+  if (use_parallel_engine) {
+    const IOStatus io_s =
+        checkpoint_engine_->CreateCheckpoint(db_, checkpoint_dir);
+    s = io_s;
+  } else {
+    s = Checkpoint::Create(db_, &checkpoint);
+    if (s.ok()) {
+      s = checkpoint->CreateCheckpoint(checkpoint_dir);
+    }
+  }
+  if (!s.ok() && !IsErrorInjectedAndRetryable(s)) {
+    fprintf(stderr, "Fail to create checkpoint to %s\n",
+            checkpoint_dir.c_str());
+    std::vector<std::string> files;
 
-      // Temporarily disable error injection to print debugging information
-      if (db_fault_injection_fs_) {
-        db_fault_injection_fs_->DisableThreadLocalErrorInjection(
-            FaultInjectionIOType::kMetadataRead);
-      }
+    // Temporarily disable error injection to print debugging information
+    if (db_fault_injection_fs_) {
+      db_fault_injection_fs_->DisableThreadLocalErrorInjection(
+          FaultInjectionIOType::kMetadataRead);
+    }
 
-      Status my_s = GetDbEnv()->GetChildren(checkpoint_dir, &files);
+    Status my_s = GetDbEnv()->GetChildren(checkpoint_dir, &files);
 
-      // Enable back disable error injection disabled for printing debugging
-      // information
-      if (db_fault_injection_fs_) {
-        db_fault_injection_fs_->EnableThreadLocalErrorInjection(
-            FaultInjectionIOType::kMetadataRead);
-      }
-      if (!my_s.ok()) {
-        fprintf(stderr, "Fail to GetChildren under %s due to %s\n",
-                checkpoint_dir.c_str(), my_s.ToString().c_str());
-      } else {
-        for (const auto& f : files) {
-          fprintf(stderr, " %s\n", f.c_str());
-        }
+    // Enable back disable error injection disabled for printing debugging
+    // information
+    if (db_fault_injection_fs_) {
+      db_fault_injection_fs_->EnableThreadLocalErrorInjection(
+          FaultInjectionIOType::kMetadataRead);
+    }
+    if (!my_s.ok()) {
+      fprintf(stderr, "Fail to GetChildren under %s due to %s\n",
+              checkpoint_dir.c_str(), my_s.ToString().c_str());
+    } else {
+      for (const auto& f : files) {
+        fprintf(stderr, " %s\n", f.c_str());
       }
     }
   }

db_stress_tool/db_stress_test_base.h

@@ -16,6 +16,7 @@
 #include "env/composite_env_wrapper.h"
 #include "rocksdb/experimental.h"
 #include "rocksdb/user_defined_index.h"
+#include "rocksdb/utilities/checkpoint.h"
 #include "utilities/fault_injection_fs.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -529,6 +530,8 @@ class StressTest {
   std::vector<ColumnFamilyHandle*> secondary_cfhs_;
   bool is_db_stopped_;
 
+  std::unique_ptr<CheckpointEngine> checkpoint_engine_;
+
   // MANIFEST verification state for reopen
   ManifestVerifyMode manifest_verify_mode_;
   uint64_t manifest_file_number_before_reopen_;

tools/db_crashtest.py

@@ -155,6 +155,9 @@ default_params = {
     "charge_table_reader": lambda: random.choice([0, 1]),
     "charge_file_metadata": lambda: random.choice([0, 1]),
     "checkpoint_one_in": lambda: random.choice([0, 0, 10000, 1000000]),
+    "checkpoint_engine_max_background_operations": lambda: random.choice([4, 8]),
+    "checkpoint_engine_use_link_file_when_available": lambda: random.randint(0, 1),
+    "parallel_checkpoint_one_in": lambda: random.choice([0, 1, 2]),
     "compression_type": lambda: random.choice(
         ["none", "snappy", "zlib", "lz4", "lz4hc", "xpress", "zstd"]
     ),