Skip to content

Commit 6d9b526

Browse files
hx235meta-codesync[bot]
hx235
authored and
meta-codesync[bot]
committed
Add OpenAndCompact() to db_bench (facebook#14003)
Summary: **Context/Summary:** as titled. This can be used to benchmark OpenAndCompact() and OpenAndCompactionOptions::allow_resumption. See below for usage. Pull Request resolved: facebook#14003 Test Plan: 1. Simple OpenAndCompact() ``` openandcompact_allow_resumption=false ./db_bench --use_existing_db=true --db=$db --benchmarks=openandcompact --openandcompact_test_cancel_on_odd=false --openandcompact_cancel_after_millseconds=0 --openandcompact_allow_resumption=$openandcompact_allow_resumption --disable_auto_compactions=true --compression_type=none --secondary_path=$secondary_path ... DB path: [/dev/shm/test] Input files: 101 files, 10000 keys OpenAndCompact() API call : 39746440.000 micros/op 39.746 seconds/op OpenAndCompact status: OK Output: 358 files, average size: 69835396 bytes (66.60 MB) openandcompact : 39977603.000 micros/op 0 ops/sec 39.978 seconds 1 operations; ``` 2. OpenAndCompact() with cancellation (after the whole compaction essentially finishes) and resumption ``` openandcompact_allow_resumption=true ./db_bench --use_existing_db=true --db=$db --benchmarks=openandcompact[X2] --openandcompact_test_cancel_on_odd=false --openandcompact_cancel_after_millseconds=0 --openandcompact_allow_resumption=$openandcompact_allow_resumption --disable_auto_compactions=true --compression_type=none --secondary_path=$secondary_path .. DB path: [/dev/shm/test] Running benchmark for 2 times Input files: 101 files, 10000 keys OpenAndCompact() API call : 40095045.000 micros/op 40.095 seconds/op OpenAndCompact status: OK Output: 358 files, average size: 69835396 bytes (66.60 MB) openandcompact : 41471226.000 micros/op 0 ops/sec 41.471 seconds 1 operations; Input files: 101 files, 10000 keys OpenAndCompact() API call : 336588.000 micros/op 0.337 seconds/op // Resume OpenAndCompact status: OK Output: 358 files, average size: 69835396 bytes (66.60 MB) openandcompact : 573885.000 micros/op 1 ops/sec 0.574 seconds 1 operations; openandcompact [AVG 2 runs] : 0 (± 1) ops/sec openandcompact [AVG 2 runs] : 0 (± 1) ops/sec; 1132.236 ms/op openandcompact [MEDIAN 2 runs] : 0 ops/sec ``` 3. OpenAndCompact() with cancellation at a fixed point and resumption ``` openandcompact_allow_resumption=true ./db_bench --use_existing_db=true --db=$db --benchmarks=openandcompact[X2] --openandcompact_test_cancel_on_odd=true --openandcompact_cancel_after_millseconds=6000 --openandcompact_allow_resumption=$openandcompact_allow_resumption --disable_auto_compactions=true --compression_type=none --secondary_path=$secondary_path ... DB path: [/dev/shm/test] Running benchmark for 2 times --- Run 1 (odd - will cancel) --- Input files: 101 files, 10000 keys OpenAndCompact() API call : 6005787.000 micros/op 6.006 seconds/op // Cancel accordingly OpenAndCompact status: Result incomplete: Manual compaction paused openandcompact : 7255346.000 micros/op 0 ops/sec 7.255 seconds 1 operations; --- Run 2 (even - resume) --- Input files: 101 files, 10000 keys OpenAndCompact() API call : 33013725.000 micros/op 33.014 seconds/op // Resume OpenAndCompact status: OK Output: 358 files, average size: 69835396 bytes (66.60 MB) openandcompact : 33244026.000 micros/op 0 ops/sec 33.244 seconds 1 operations; openandcompact [AVG 2 runs] : 0 (± 0) ops/sec openandcompact [AVG 2 runs] : 0 (± 0) ops/sec; 11911.234 ms/op openandcompact [MEDIAN 2 runs] : 0 ops/sec ``` Reviewed By: jaykorean Differential Revision: D84839965 Pulled By: hx235 fbshipit-source-id: 21c4cd01be67da0a128e2de1c3aae93aa97662bd
1 parent f343f7e commit 6d9b526

1 file changed

Lines changed: 236 additions & 6 deletions

File tree

‎tools/db_bench_tool.cc‎

Lines changed: 236 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -159,6 +159,7 @@ DEFINE_string(
159159
"readrandomoperands,"
160160
"backup,"
161161
"restore,"
162+
"openandcompact,"
162163
"approximatememtablestats",
163164

164165
"Comma-separated list of operations to run in the specified"
@@ -230,6 +231,9 @@ DEFINE_string(
230231
"\tcompact1 -- compact L1 into L2\n"
231232
"\twaitforcompaction - pause until compaction is (probably) done\n"
232233
"\tflush - flush the memtable\n"
234+
"\topenandcompact -- Open DB and compact all files to bottommost level, "
235+
"writing output to separate directory without modifying source DB. "
236+
"Designed for remote compaction service testing\n"
233237
"\tstats -- Print DB stats\n"
234238
"\tresetstats -- Reset DB stats\n"
235239
"\tlevelstats -- Print the number of files and bytes per level\n"
@@ -1872,6 +1876,18 @@ DEFINE_bool(
18721876
.use_async_io,
18731877
"Sets MultiScanArgs::use_async_io");
18741878

1879+
DEFINE_bool(openandcompact_allow_resumption, false,
1880+
"Whether to keep existing progress and enable resume compaction in "
1881+
"OpenAndCompact benchmark");
1882+
1883+
DEFINE_bool(openandcompact_test_cancel_on_odd, false,
1884+
"During OpenAndCompact[Xn], odd runs gets cancelled "
1885+
"after specified `openandcompact_cancel_after_millseconds`");
1886+
1887+
DEFINE_uint32(openandcompact_cancel_after_millseconds, 1,
1888+
"Time to wait before cancelling compaction in odd runs when "
1889+
"openandcompact_test_cancel_on_odd is true");
1890+
18751891
namespace ROCKSDB_NAMESPACE {
18761892
namespace {
18771893
static Status CreateMemTableRepFactory(
@@ -2625,24 +2641,33 @@ class CombinedStats {
26252641
const char* name = bench_name.c_str();
26262642
int num_runs = static_cast<int>(throughput_ops_.size());
26272643

2644+
double avg_ops_per_sec = CalcAvg(throughput_ops_);
2645+
double avg_millis_per_op =
2646+
(avg_ops_per_sec > 0) ? (1000.0 / avg_ops_per_sec) : 0;
2647+
2648+
printf("\n");
2649+
26282650
if (throughput_mbs_.size() == throughput_ops_.size()) {
26292651
// \xC2\xB1 is +/- character in UTF-8
26302652
fprintf(stdout,
2631-
"%s [AVG %d runs] : %d (\xC2\xB1 %d) ops/sec; %6.1f (\xC2\xB1 "
2653+
"%s [AVG %d runs] : %d (\xC2\xB1 %d) ops/sec; %.3f ms/op; "
2654+
"%6.1f (\xC2\xB1 "
26322655
"%.1f) MB/sec\n"
26332656
"%s [MEDIAN %d runs] : %d ops/sec; %6.1f MB/sec\n",
26342657
name, num_runs, static_cast<int>(CalcAvg(throughput_ops_)),
26352658
static_cast<int>(CalcConfidence95(throughput_ops_)),
2636-
CalcAvg(throughput_mbs_), CalcConfidence95(throughput_mbs_), name,
2637-
num_runs, static_cast<int>(CalcMedian(throughput_ops_)),
2659+
avg_millis_per_op, CalcAvg(throughput_mbs_),
2660+
CalcConfidence95(throughput_mbs_), name, num_runs,
2661+
static_cast<int>(CalcMedian(throughput_ops_)),
26382662
CalcMedian(throughput_mbs_));
26392663
} else {
26402664
fprintf(stdout,
2641-
"%s [AVG %d runs] : %d (\xC2\xB1 %d) ops/sec\n"
2665+
"%s [AVG %d runs] : %d (\xC2\xB1 %d) ops/sec; %.3f ms/op\n"
26422666
"%s [MEDIAN %d runs] : %d ops/sec\n",
26432667
name, num_runs, static_cast<int>(CalcAvg(throughput_ops_)),
2644-
static_cast<int>(CalcConfidence95(throughput_ops_)), name,
2645-
num_runs, static_cast<int>(CalcMedian(throughput_ops_)));
2668+
static_cast<int>(CalcConfidence95(throughput_ops_)),
2669+
avg_millis_per_op, name, num_runs,
2670+
static_cast<int>(CalcMedian(throughput_ops_)));
26462671
}
26472672
}
26482673

@@ -2801,6 +2826,8 @@ class Duration {
28012826
uint64_t start_at_;
28022827
};
28032828

2829+
// Global run counter for cancel/resume-OpenAndCompact() testing
2830+
static std::atomic<int> openandcompact_run_counter{0};
28042831
class Benchmark {
28052832
private:
28062833
std::shared_ptr<Cache> cache_;
@@ -3853,6 +3880,9 @@ class Benchmark {
38533880
method = &Benchmark::Backup;
38543881
} else if (name == "restore") {
38553882
method = &Benchmark::Restore;
3883+
} else if (name == "openandcompact") {
3884+
fresh_db = false;
3885+
method = &Benchmark::OpenAndCompact;
38563886
} else if (!name.empty()) { // No error message for empty name
38573887
fprintf(stderr, "unknown benchmark '%s'\n", name.c_str());
38583888
ErrorExit();
@@ -5182,6 +5212,206 @@ class Benchmark {
51825212
DoWrite(thread, UNIQUE_RANDOM);
51835213
}
51845214

5215+
void OpenAndCompact(ThreadState* thread) {
5216+
if (thread->tid != 0) {
5217+
return;
5218+
}
5219+
5220+
int current_run = ++openandcompact_run_counter;
5221+
bool is_odd_run = (current_run % 2 == 1);
5222+
5223+
if (FLAGS_openandcompact_test_cancel_on_odd) {
5224+
const char* even_description = FLAGS_openandcompact_allow_resumption
5225+
? "even - resume"
5226+
: "even - normal";
5227+
fprintf(stdout, "\n--- Run %d (%s) ---\n", current_run,
5228+
is_odd_run ? "odd - will cancel" : even_description);
5229+
}
5230+
5231+
Status create_status =
5232+
db_.db->GetEnv()->CreateDirIfMissing(FLAGS_secondary_path);
5233+
if (!create_status.ok()) {
5234+
fprintf(stderr, "Failed to create secondary path: %s\n",
5235+
create_status.ToString().c_str());
5236+
return;
5237+
}
5238+
5239+
std::string options_file;
5240+
Status options_status =
5241+
GetLatestOptionsFileName(FLAGS_db, db_.db->GetEnv(), &options_file);
5242+
if (!options_status.ok()) {
5243+
fprintf(stderr, "FAILED: Cannot find OPTIONS file in %s: %s\n",
5244+
FLAGS_db.c_str(), options_status.ToString().c_str());
5245+
return;
5246+
}
5247+
5248+
uint64_t options_file_number;
5249+
FileType type;
5250+
if (!ParseFileName(options_file, &options_file_number, &type) ||
5251+
type != kOptionsFile) {
5252+
fprintf(stderr, "FAILED: Cannot parse OPTIONS file number from %s\n",
5253+
options_file.c_str());
5254+
return;
5255+
}
5256+
5257+
CompactionServiceInput compaction_input;
5258+
compaction_input.cf_name = kDefaultColumnFamilyName;
5259+
5260+
std::vector<std::string> input_file_names;
5261+
ColumnFamilyMetaData cf_meta;
5262+
db_.db->GetColumnFamilyMetaData(&cf_meta);
5263+
5264+
uint64_t total_input_keys = 0;
5265+
uint64_t total_input_files = 0;
5266+
5267+
// Collect files from all levels for full compaction
5268+
for (const auto& level : cf_meta.levels) {
5269+
for (const auto& file : level.files) {
5270+
input_file_names.push_back(file.name);
5271+
total_input_keys += file.num_entries;
5272+
total_input_files++;
5273+
}
5274+
}
5275+
5276+
// Set output level to configured bottom level (num_levels - 1)
5277+
compaction_input.output_level = FLAGS_num_levels - 1;
5278+
compaction_input.db_id = "db_bench_openandcompact";
5279+
compaction_input.options_file_number = options_file_number;
5280+
5281+
compaction_input.input_files = input_file_names;
5282+
5283+
std::string input_string;
5284+
Status serialize_status = compaction_input.Write(&input_string);
5285+
if (!serialize_status.ok()) {
5286+
fprintf(stderr, "FAILED: Cannot serialize compaction input: %s\n",
5287+
serialize_status.ToString().c_str());
5288+
return;
5289+
}
5290+
5291+
fprintf(stdout, "\nInput files: %" PRIu64 " files, %" PRIu64 " keys\n",
5292+
total_input_files, total_input_keys);
5293+
5294+
std::string output_directory =
5295+
FLAGS_secondary_path + "/openandcompact_" + std::to_string(thread->tid);
5296+
5297+
// Always clean up in odd run, depending on
5298+
// !FLAGS_openandcompact_allow_resumption in even run
5299+
bool should_cleanup = is_odd_run || !FLAGS_openandcompact_allow_resumption;
5300+
5301+
if (should_cleanup) {
5302+
std::vector<std::string> children;
5303+
Status list_status = FLAGS_env->GetChildren(output_directory, &children);
5304+
if (list_status.ok()) {
5305+
for (const auto& child : children) {
5306+
if (child != "." && child != "..") {
5307+
std::string child_path = output_directory + "/" + child;
5308+
Status del_status = FLAGS_env->DeleteFile(child_path);
5309+
if (!del_status.ok()) {
5310+
fprintf(stderr, "Warning: Failed to delete file %s: %s\n",
5311+
child_path.c_str(), del_status.ToString().c_str());
5312+
}
5313+
}
5314+
}
5315+
Status del_dir_status = FLAGS_env->DeleteDir(output_directory);
5316+
if (!del_dir_status.ok()) {
5317+
fprintf(stderr, "Warning: Failed to delete directory %s: %s\n",
5318+
output_directory.c_str(), del_dir_status.ToString().c_str());
5319+
}
5320+
}
5321+
}
5322+
5323+
Status create_output_status =
5324+
FLAGS_env->CreateDirIfMissing(output_directory);
5325+
if (!create_output_status.ok()) {
5326+
fprintf(stderr, "Failed to create output directory %s: %s\n",
5327+
output_directory.c_str(),
5328+
create_output_status.ToString().c_str());
5329+
return;
5330+
}
5331+
5332+
std::string result_string;
5333+
5334+
CompactionServiceOptionsOverride options_override;
5335+
options_override.env = FLAGS_env;
5336+
BlockBasedTableOptions table_options;
5337+
options_override.table_factory.reset(
5338+
NewBlockBasedTableFactory(table_options));
5339+
5340+
OpenAndCompactOptions options;
5341+
std::atomic<bool> should_cancel{false};
5342+
options.canceled = &should_cancel;
5343+
options.allow_resumption = FLAGS_openandcompact_allow_resumption;
5344+
5345+
Status s;
5346+
uint64_t start_time = FLAGS_env->NowMicros();
5347+
uint64_t end_time = start_time;
5348+
5349+
if (FLAGS_openandcompact_test_cancel_on_odd && is_odd_run) {
5350+
std::thread compaction_thread([&]() {
5351+
s = DB::OpenAndCompact(options, FLAGS_db, output_directory,
5352+
input_string, &result_string, options_override);
5353+
end_time = FLAGS_env->NowMicros();
5354+
});
5355+
5356+
std::thread cancellation_timer([&]() {
5357+
std::this_thread::sleep_for(std::chrono::milliseconds(
5358+
FLAGS_openandcompact_cancel_after_millseconds));
5359+
should_cancel.store(true);
5360+
});
5361+
5362+
compaction_thread.join();
5363+
cancellation_timer.join();
5364+
} else {
5365+
// Normal synchronous operation for even runs or when test_cancel_on_odd
5366+
// is false
5367+
s = DB::OpenAndCompact(options, FLAGS_db, output_directory, input_string,
5368+
&result_string, options_override);
5369+
end_time = FLAGS_env->NowMicros();
5370+
}
5371+
5372+
uint64_t latency_micros = end_time - start_time;
5373+
double latency_seconds = latency_micros / 1000000.0;
5374+
5375+
fprintf(stdout,
5376+
"OpenAndCompact() API call : %.3f micros/op %.3f seconds/op\n",
5377+
(double)latency_micros, latency_seconds);
5378+
5379+
fprintf(stdout, "OpenAndCompact status: %s\n", s.ToString().c_str());
5380+
5381+
if (FLAGS_openandcompact_test_cancel_on_odd && is_odd_run) {
5382+
if (!s.IsManualCompactionPaused()) {
5383+
fprintf(stdout, "Fail to cancel compaction");
5384+
}
5385+
return;
5386+
} else if (!s.ok()) {
5387+
fprintf(stderr, "OpenAndCompact failed: %s\n", s.ToString().c_str());
5388+
return;
5389+
}
5390+
5391+
CompactionServiceResult compaction_result;
5392+
Status parse_status =
5393+
CompactionServiceResult::Read(result_string, &compaction_result);
5394+
if (parse_status.ok()) {
5395+
uint64_t total_output_size = 0;
5396+
for (const auto& output_file : compaction_result.output_files) {
5397+
total_output_size += output_file.file_size;
5398+
}
5399+
5400+
uint64_t num_output_files = compaction_result.output_files.size();
5401+
uint64_t avg_output_file_size =
5402+
num_output_files > 0 ? total_output_size / num_output_files : 0;
5403+
5404+
fprintf(stdout,
5405+
"Output: %" PRIu64 " files, average size: %" PRIu64
5406+
" bytes (%.2f MB)\n",
5407+
num_output_files, avg_output_file_size,
5408+
avg_output_file_size / (1024.0 * 1024.0));
5409+
} else {
5410+
fprintf(stderr, "Failed to parse compaction result: %s\n",
5411+
parse_status.ToString().c_str());
5412+
}
5413+
}
5414+
51855415
class KeyGenerator {
51865416
public:
51875417
KeyGenerator(Random64* rand, WriteMode mode, uint64_t num,

0 commit comments

Comments
 (0)