diff --git a/db_stress_tool/db_stress_common.h b/db_stress_tool/db_stress_common.h index 1dd140304a..bfe3f1df2b 100644 --- a/db_stress_tool/db_stress_common.h +++ b/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); diff --git a/db_stress_tool/db_stress_gflags.cc b/db_stress_tool/db_stress_gflags.cc index 30beb27cbe..9978980054 100644 --- a/db_stress_tool/db_stress_gflags.cc +++ b/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() " diff --git a/db_stress_tool/db_stress_test_base.cc b/db_stress_tool/db_stress_test_base.cc index 3948371680..5464be22aa 100644 --- a/db_stress_tool/db_stress_test_base.cc +++ b/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 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 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()); } } } diff --git a/db_stress_tool/db_stress_test_base.h b/db_stress_tool/db_stress_test_base.h index 2351bf3213..e3360ca387 100644 --- a/db_stress_tool/db_stress_test_base.h +++ b/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 secondary_cfhs_; bool is_db_stopped_; + std::unique_ptr checkpoint_engine_; + // MANIFEST verification state for reopen ManifestVerifyMode manifest_verify_mode_; uint64_t manifest_file_number_before_reopen_; diff --git a/tools/db_crashtest.py b/tools/db_crashtest.py index acfbd0e3a6..1765e00a7b 100644 --- a/tools/db_crashtest.py +++ b/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"] ),