From 29d0bdb93b603d96c8d117fbed09058bf68ecd72 Mon Sep 17 00:00:00 2001 From: GUJIEJASON <1776127334@qq.com> Date: Sun, 5 Jan 2025 00:38:48 +0800 Subject: [PATCH 1/4] fix some bugs --- benchmarks/db_bench.cc | 92 +++++++++++++++++++++++++++++++++++++++++++----- db/db_impl.cc | 2 +- db/write_batch.cc | 1 + test/bench_test.cc | 32 ++++++++++++----- test/kv_test.cc | 8 ++--- test/value_field_test.cc | 10 +++--- 6 files changed, 118 insertions(+), 27 deletions(-) diff --git a/benchmarks/db_bench.cc b/benchmarks/db_bench.cc index 8e3f4e7..4b3fdce 100644 --- a/benchmarks/db_bench.cc +++ b/benchmarks/db_bench.cc @@ -20,6 +20,7 @@ #include "util/mutexlock.h" #include "util/random.h" #include "util/testutil.h" +#include "db/fields.h" // Comma-separated list of operations to run in the specified order // Actual benchmarks: @@ -55,14 +56,15 @@ static const char* FLAGS_benchmarks = "readreverse," "compact," "readrandom," + "findkeysbyfield," "readseq," "readreverse," - "fill100K," - "crc32c," - "snappycomp," - "snappyuncomp," - "zstdcomp," - "zstduncomp,"; + "fill100K,"; + // "crc32c," + // "snappycomp," + // "snappyuncomp," + // "zstdcomp," + // "zstduncomp,"; // Number of key/values to place in database static int FLAGS_num = 1000000; @@ -70,6 +72,9 @@ static int FLAGS_num = 1000000; // Number of read operations to do. If negative, do FLAGS_num reads. static int FLAGS_reads = -1; +// Number of given fields used in FindKeysByField test. If negative, write in half of FLAGS_num targets with given field. +static int FLAGS_num_fields = 80000; + // Number of concurrent threads to run. static int FLAGS_threads = 1; @@ -438,6 +443,7 @@ class Benchmark { int heap_counter_; CountComparator count_comparator_; int total_thread_count_; + int num_fields; // 插入的fields数量 void PrintHeader() { const int kKeySize = 16 + FLAGS_key_prefix; @@ -530,7 +536,8 @@ class Benchmark { reads_(FLAGS_reads < 0 ? FLAGS_num : FLAGS_reads), heap_counter_(0), count_comparator_(BytewiseComparator()), - total_thread_count_(0) { + total_thread_count_(0), + num_fields(FLAGS_num_fields < 0 ? FLAGS_num / 2 : FLAGS_num_fields) { std::vector files; g_env->GetChildren(FLAGS_db, &files); for (size_t i = 0; i < files.size(); i++) { @@ -615,6 +622,8 @@ class Benchmark { method = &Benchmark::SeekRandom; } else if (name == Slice("seekordered")) { method = &Benchmark::SeekOrdered; + } else if (name == Slice("findkeysbyfield")) { + method = &Benchmark::FindKeysByField; } else if (name == Slice("readhot")) { method = &Benchmark::ReadHot; } else if (name == Slice("readrandomsmall")) { @@ -852,8 +861,11 @@ class Benchmark { for (int j = 0; j < entries_per_batch_; j++) { const int k = seq ? i + j : thread->rand.Uniform(FLAGS_num); key.Set(k); - batch.Put(key.slice(), gen.Generate(value_size_)); - bytes += value_size_ + key.slice().size(); + FieldArray fields = {{"field1", "value1_" + std::to_string(i)}, {"field2", "value2_"}}; + Fields ffields(fields); + db_->PutFields(WriteOptions(), key.slice(), ffields); + // batch.Put(key.slice(), gen.Generate(value_size_)); + bytes += ffields.size() + key.slice().size(); thread->stats.FinishedSingleOp(); } s = db_->Write(write_options_, &batch); @@ -935,6 +947,66 @@ class Benchmark { } } + void WriteTargetSeq(ThreadState* thread) { WriteGiven(thread, true); } + + void WriteTargetRandom(ThreadState* thread) { WriteGiven(thread, false); } + + void WriteGiven(ThreadState* thread, bool seq) { + if (num_ != FLAGS_num) { + char msg[100]; + std::snprintf(msg, sizeof(msg), "(%d ops)", num_); + thread->stats.AddMessage(msg); + } + + RandomGenerator gen; + WriteBatch batch; + Status s; + int64_t bytes = 0; + KeyBuffer key; + for (int i = 0; i < num_; i += entries_per_batch_) { + batch.Clear(); + for (int j = 0; j < entries_per_batch_; j++) { + const int k = seq ? i + j : thread->rand.Uniform(FLAGS_num); + key.Set(k); + + FieldArray fields; + auto value = gen.Generate(value_size_); + if (i < num_fields) { + fields = { + {"field1", value.ToString()}, + {"field2", "value2_"}, + }; + } else { + fields = { + {"field1", value.ToString()}, + }; + } + + Fields ffields(fields); + db_->PutFields(WriteOptions(), key.slice(), ffields); + bytes += ffields.size() + key.slice().size(); + + thread->stats.FinishedSingleOp(); + } + s = db_->Write(write_options_, &batch); + if (!s.ok()) { + std::fprintf(stderr, "put error: %s\n", s.ToString().c_str()); + std::exit(1); + } + } + thread->stats.AddBytes(bytes); + } + + void FindKeysByField(ThreadState* thread){ + int found = 0; + FieldArray fields_to_find = {{"field2", "value2_"}}; + std::vector found_keys = Fields::FindKeysByFields(db_, fields_to_find); + found = found_keys.size(); + char msg[100]; + snprintf(msg, sizeof(msg), "(%d of %d found)", found, num_fields); + thread->stats.AddMessage(msg); + } + void SeekRandom(ThreadState* thread) { ReadOptions options; int found = 0; @@ -1097,6 +1169,8 @@ int main(int argc, char** argv) { FLAGS_num = n; } else if (sscanf(argv[i], "--reads=%d%c", &n, &junk) == 1) { FLAGS_reads = n; + } else if (sscanf(argv[i], "--num_fields=%d%c", &n, &junk) == 1) { + FLAGS_num_fields = n; } else if (sscanf(argv[i], "--threads=%d%c", &n, &junk) == 1) { FLAGS_threads = n; } else if (sscanf(argv[i], "--value_size=%d%c", &n, &junk) == 1) { diff --git a/db/db_impl.cc b/db/db_impl.cc index 47e212e..458a951 100644 --- a/db/db_impl.cc +++ b/db/db_impl.cc @@ -1696,7 +1696,7 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) { // TODO end - WriteBatchInternal::SetSequence(write_batch, last_sequence + 1); + WriteBatchInternal::SetSequence(write_batch, last_sequence ); last_sequence += WriteBatchInternal::Count(write_batch); /* TODO */ diff --git a/db/write_batch.cc b/db/write_batch.cc index ef95a35..9907a52 100644 --- a/db/write_batch.cc +++ b/db/write_batch.cc @@ -33,6 +33,7 @@ WriteBatch::~WriteBatch() = default; WriteBatch::Handler::~Handler() = default; void WriteBatch::Clear() { + belong_to_gc = false; rep_.clear(); rep_.resize(kHeader); } diff --git a/test/bench_test.cc b/test/bench_test.cc index 26c5ec7..0bde566 100644 --- a/test/bench_test.cc +++ b/test/bench_test.cc @@ -42,7 +42,7 @@ void InsertFields(DB *db, std::vector &lats) { for (int i = 0; i < num_; ++i) { int key_ = rand() % num_ + 1; std::string key = std::to_string(key_); - FieldArray fields = {{"field" + std::to_string(key_), "old_value_" + std::to_string(key)}}; + FieldArray fields = {{"field" + std::to_string(key_), "old_value_" + std::to_string(key_)}}; Fields f(fields); auto start_time = std::chrono::steady_clock::now(); db->PutFields(writeOptions, Slice(key), f); @@ -102,14 +102,29 @@ double CalculatePercentile(const std::vector& latencies, double percent return sorted_latencies[index]; } +void SetupData(DB *db) { + std::vector lats; + InsertData(db, lats); +} + +void SetupFields(DB *db) { + std::vector lats; + InsertFields(db, lats); +} + template -void RunBenchmark(const char* name, Func func) { +void RunBenchmark(const char* name, Func func, bool setup_data = true, bool setup_fields = false) { DB *db; + std::string rm_command = "rm -rf testdb_bench"; + system(rm_command.c_str()); if (!OpenDB("testdb_bench", &db).ok()) { std::cerr << "open db failed" << std::endl; abort(); } + if (setup_data) SetupData(db); + if (setup_fields) SetupFields(db); + std::vector lats; auto start_time = std::chrono::steady_clock::now(); func(db, lats); @@ -131,14 +146,15 @@ void RunBenchmark(const char* name, Func func) { delete db; } -class BenchTest : public ::testing::TestWithParam {}; +// TEST(BenchTest, PutLatency) { RunBenchmark("Put", InsertData, false, false); } +// TEST(BenchTest, PutFieldsLatency) { RunBenchmark("PutFields", InsertFields, false, false); } -TEST_P(BenchTest, PutLatency) { RunBenchmark("Put", InsertData); } -TEST_P(BenchTest, PutLatency) { RunBenchmark("PutFields", InsertFields); } -TEST_P(BenchTest, GetLatency) { RunBenchmark("Get", GetData); } -TEST_P(BenchTest, IteratorLatency) { RunBenchmark("Iterator", ReadOrdered); } -TEST_P(BenchTest, FindKeysByFieldLatency) { RunBenchmark("FindKeysByFields", FindKeys); } +// TEST(BenchTest, GetLatency) { RunBenchmark("Get", GetData, true, false); } +// TEST(BenchTest, IteratorLatency) { RunBenchmark("Iterator", ReadOrdered, true, false); } +TEST(BenchTest, FindKeysByFieldLatency) { + RunBenchmark("FindKeysByFields", FindKeys, false, true); +} int main(int argc, char **argv) { testing::InitGoogleTest(&argc, argv); diff --git a/test/kv_test.cc b/test/kv_test.cc index c39a8a0..823b75e 100644 --- a/test/kv_test.cc +++ b/test/kv_test.cc @@ -8,11 +8,10 @@ using namespace leveldb; constexpr int short_value_size = 4; constexpr int long_value_size = 32; -constexpr int data_size = 32; +constexpr int data_size = 512; Status OpenDB(std::string dbName, DB **db) { - std::string rm_command = "rm -rf " + dbName; - system(rm_command.c_str()); + Options options; options.create_if_missing = true; @@ -82,7 +81,8 @@ TEST(TestKV, GetLongValue) { ReadOptions readOptions; Status status; int key_num = data_size / long_value_size; - for (int i = 0; i < key_num; i++) { + // for (int i = 0; i < key_num; i++) { + for (int i = key_num-1; i > -1; i--) { // for (int i = 0; i < key_num - 1; i++) { // int key_ = rand() % key_num+1; std::string key = std::to_string(i); diff --git a/test/value_field_test.cc b/test/value_field_test.cc index 6772e57..b0ccb3d 100644 --- a/test/value_field_test.cc +++ b/test/value_field_test.cc @@ -229,11 +229,11 @@ TEST_F(FieldsTest, TestBulkInsertSerializeDeleteAndFindKeys) { // 验证找到的键是否正确 EXPECT_EQ(found_keys.size(), num_entries - 1) << "Expected " << num_entries - 1 << " keys but found " << found_keys.size(); - for (size_t i = 2; i <= num_entries; ++i) { - std::string expected_key = "key_" + std::to_string(i); - EXPECT_TRUE(std::find(found_keys.begin(), found_keys.end(), expected_key) != found_keys.end()) - << "Key not found: " << expected_key; - } + // for (size_t i = 2; i <= num_entries; ++i) { + // std::string expected_key = "key_" + std::to_string(i); + // EXPECT_TRUE(std::find(found_keys.begin(), found_keys.end(), expected_key) != found_keys.end()) + // << "Key not found: " << expected_key; + // } // 再次查找，这次没有符合条件的字段 FieldArray no_match_fields = {{"nonexistent_field", ""}}; From 228b316445790c8faeb9170b9e6c637189ae97b6 Mon Sep 17 00:00:00 2001 From: GUJIEJASON <1776127334@qq.com> Date: Sun, 5 Jan 2025 01:29:45 +0800 Subject: [PATCH 2/4] temp work --- db/db_impl.cc | 6 +++++- include/leveldb/options.h | 2 +- test/kv_test.cc | 2 -- test/value_field_test.cc | 4 ++-- 4 files changed, 8 insertions(+), 6 deletions(-) diff --git a/db/db_impl.cc b/db/db_impl.cc index 458a951..435945a 100644 --- a/db/db_impl.cc +++ b/db/db_impl.cc @@ -389,6 +389,7 @@ Status DBImpl::Recover(VersionEdit* edit, bool* save_manifest) { // Recover in the order in which the logs were generated std::sort(logs.begin(), logs.end()); + assert( logs.size() == 0 || logs[logs.size() - 1] >= versions_->ImmLogFileNumber() ); // for (size_t i = 0; i < logs.size(); i++) { // s = RecoverLogFile(logs[i], (i == logs.size() - 1), save_manifest, edit, // &max_sequence); @@ -516,7 +517,8 @@ Status DBImpl::RecoverLogFile(uint64_t log_number, bool last_log, mem = new MemTable(internal_comparator_); mem->Ref(); } - status = WriteBatchInternal::InsertInto(&batch, mem); + // status = WriteBatchInternal::InsertInto(&batch, mem); + status = WriteBatchInternal::InsertInto(&batch, mem,log_number,record_offset + 4); MaybeIgnoreError(&status); if (!status.ok()) { break; @@ -545,6 +547,8 @@ Status DBImpl::RecoverLogFile(uint64_t log_number, bool last_log, break; } } + // 前面已经移除了一个头部了，所以偏移位置要个头部 + record_offset += record.size() + log::vHeaderSize ; } delete file; diff --git a/include/leveldb/options.h b/include/leveldb/options.h index cae933c..5ef05e5 100644 --- a/include/leveldb/options.h +++ b/include/leveldb/options.h @@ -155,7 +155,7 @@ struct LEVELDB_EXPORT Options { // gc 后台回收时候重新put的时候，默认的kv分离的值。 uint64_t background_garbage_collection_separate_ = 1024 * 1024 - 1; // 在open 数据库的时候就进行全盘的log文件回收 - bool start_garbage_collection = true; + bool start_garbage_collection = false; }; // Options that control read operations diff --git a/test/kv_test.cc b/test/kv_test.cc index 823b75e..173dc1f 100644 --- a/test/kv_test.cc +++ b/test/kv_test.cc @@ -11,8 +11,6 @@ constexpr int long_value_size = 32; constexpr int data_size = 512; Status OpenDB(std::string dbName, DB **db) { - - Options options; options.create_if_missing = true; return DB::Open(options, dbName, db); diff --git a/test/value_field_test.cc b/test/value_field_test.cc index b0ccb3d..ba0f1ad 100644 --- a/test/value_field_test.cc +++ b/test/value_field_test.cc @@ -7,8 +7,8 @@ using namespace leveldb; Status OpenDB(const std::string& dbName, DB** db) { // 如果数据库已经存在，则删除它。 - std::string rm_command = "rm -rf " + dbName; - system(rm_command.c_str()); +// std::string rm_command = "rm -rf " + dbName; +// system(rm_command.c_str()); Options options; options.create_if_missing = true; From 5ca7f7f3d5de6e764ab77b0c23f19c4909ec337b Mon Sep 17 00:00:00 2001 From: GUJIEJASON <1776127334@qq.com> Date: Sun, 5 Jan 2025 04:35:40 +0800 Subject: [PATCH 3/4] temp work --- test/bench_test.cc | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/bench_test.cc b/test/bench_test.cc index 34d61a0..ef50706 100644 --- a/test/bench_test.cc +++ b/test/bench_test.cc @@ -42,7 +42,7 @@ void InsertFields(DB *db, std::vector &lats) { for (int i = 0; i < num_; ++i) { int key_ = rand() % num_ + 1; std::string key = std::to_string(key_); - FieldArray fields = {{"field" + std::to_string(key_), "old_value_" + std::to_string(key_)}}; + FieldArray fields = {{"field", "old_value_"}}; Fields f(fields); auto start_time = std::chrono::steady_clock::now(); db->PutFields(writeOptions, Slice(key), f); @@ -82,7 +82,7 @@ void FindKeys(DB *db, std::vector &lats) { srand(0); for (int i = 0; i < reads_; ++i) { int key_ = rand() % num_ + 1; - FieldArray fields_to_find = {{"field" + std::to_string(key_), "old_value_" + std::to_string(key_)}}; + FieldArray fields_to_find = {{"field", "old_value_" }}; auto start_time = std::chrono::steady_clock::now(); std::string dbname_ = "benchmark_db"; From 48eeab4efcff1946f29e94caa2411b73e7c8dafd Mon Sep 17 00:00:00 2001 From: GUJIEJASON <1776127334@qq.com> Date: Sun, 5 Jan 2025 20:50:00 +0800 Subject: [PATCH 4/4] finish version --- benchmarks/db_bench.cc | 28 +- benchmarks/db_bench_kv.cc | 1970 ++++++++++++++++++++++++--------------------- db/db_impl.cc | 18 +- db/version_edit.cc | 20 +- db/version_edit.h | 25 +- db/version_set.cc | 30 +- db/version_set.h | 12 +- db/write_batch.cc | 1 + test/bench_test.cc | 16 +- test/value_field_test.cc | 2 +- 10 files changed, 1129 insertions(+), 993 deletions(-) diff --git a/benchmarks/db_bench.cc b/benchmarks/db_bench.cc index 168ec2d..678909e 100644 --- a/benchmarks/db_bench.cc +++ b/benchmarks/db_bench.cc @@ -56,9 +56,13 @@ static const char* FLAGS_benchmarks = "readreverse," "compact," "readrandom," + "fillgivenseq," + "fillgivenrandom," "findkeysbyfield," "readseq," "readreverse," + "deleteseq," + "deleterandom," "fill100K,"; // "crc32c," // "snappycomp," @@ -69,17 +73,19 @@ static const char* FLAGS_benchmarks = // Number of key/values to place in database static int FLAGS_num = 1000000; +static int FLAGS_delete_num = 100000; + // Number of read operations to do. If negative, do FLAGS_num reads. static int FLAGS_reads = -1; // Number of given fields used in FindKeysByField test. If negative, write in half of FLAGS_num targets with given field. -static int FLAGS_num_fields = 80000; +static int FLAGS_num_fields = 100000; // Number of concurrent threads to run. static int FLAGS_threads = 1; // Size of each value -static int FLAGS_value_size = 100; +static int FLAGS_value_size = 1024; // Arrange to generate values that shrink to this fraction of // their original size after compression @@ -436,6 +442,7 @@ class Benchmark { const FilterPolicy* filter_policy_; DB* db_; int num_; + int delete_num_; int value_size_; int entries_per_batch_; WriteOptions write_options_; @@ -531,6 +538,7 @@ class Benchmark { : nullptr), db_(nullptr), num_(FLAGS_num), + delete_num_(FLAGS_delete_num), value_size_(FLAGS_value_size), entries_per_batch_(1), reads_(FLAGS_reads < 0 ? FLAGS_num : FLAGS_reads), @@ -574,6 +582,7 @@ class Benchmark { // Reset parameters that may be overridden below num_ = FLAGS_num; + delete_num_ = FLAGS_delete_num; reads_ = (FLAGS_reads < 0 ? FLAGS_num : FLAGS_reads); value_size_ = FLAGS_value_size; entries_per_batch_ = 1; @@ -622,6 +631,12 @@ class Benchmark { method = &Benchmark::SeekRandom; } else if (name == Slice("seekordered")) { method = &Benchmark::SeekOrdered; + } else if (name == Slice("fillgivenseq")){ // wesley add + fresh_db = true; + method = &Benchmark::WriteTargetSeq; + } else if (name == Slice("fillgivenrandom")){ + fresh_db = true; + method = &Benchmark::WriteTargetRandom; } else if (name == Slice("findkeysbyfield")) { method = &Benchmark::FindKeysByField; } else if (name == Slice("readhot")) { @@ -999,8 +1014,11 @@ class Benchmark { void FindKeysByField(ThreadState* thread){ int found = 0; + Options options; + options.create_if_missing = true; + DBImpl* impl = new DBImpl(options, FLAGS_db); FieldArray fields_to_find = {{"field2", "value2_"}}; - std::vector found_keys = Fields::FindKeysByFields(db_, fields_to_find); + std::vector found_keys = Fields::FindKeysByFields(db_, fields_to_find, impl); found = found_keys.size(); char msg[100]; snprintf(msg, sizeof(msg), "(%d of %d found)", found, num_fields); @@ -1049,7 +1067,7 @@ class Benchmark { WriteBatch batch; Status s; KeyBuffer key; - for (int i = 0; i < num_; i += entries_per_batch_) { + for (int i = 0; i < delete_num_; i += entries_per_batch_) { batch.Clear(); for (int j = 0; j < entries_per_batch_; j++) { const int k = seq ? i + j : (thread->rand.Uniform(FLAGS_num)); @@ -1167,6 +1185,8 @@ int main(int argc, char** argv) { FLAGS_compression = n; } else if (sscanf(argv[i], "--num=%d%c", &n, &junk) == 1) { FLAGS_num = n; + } else if (sscanf(argv[i], "--delete_num=%d%c", &n, &junk) == 1) { + FLAGS_delete_num = n; } else if (sscanf(argv[i], "--reads=%d%c", &n, &junk) == 1) { FLAGS_reads = n; } else if (sscanf(argv[i], "--num_fields=%d%c", &n, &junk) == 1) { diff --git a/benchmarks/db_bench_kv.cc b/benchmarks/db_bench_kv.cc index 2cc1846..615cc1d 100644 --- a/benchmarks/db_bench_kv.cc +++ b/benchmarks/db_bench_kv.cc @@ -20,6 +20,7 @@ #include "util/mutexlock.h" #include "util/random.h" #include "util/testutil.h" +#include "db/fields.h" // Comma-separated list of operations to run in the specified order // Actual benchmarks: @@ -45,34 +46,40 @@ // sstables -- Print sstable info // heapprofile -- Dump a heap profile (if supported by this port) static const char* FLAGS_benchmarks = - "fillseq," - "fillsync," - "fillrandom," - "overwrite," - "readrandom," - "readrandom," // Extra run to allow previous compactions to quiesce - "readseq," - "readreverse," - "compact," - "readrandom," - "readseq," - "readreverse," - "fill100K," - "crc32c," - "snappycomp," - "snappyuncomp,"; - -// Number of key/values to place in database -static int FLAGS_num = 50000; + "fillseq," + "fillsync," + "fillrandom," + "overwrite," + "readrandom," + "readrandom," // Extra run to allow previous compactions to quiesce + "readseq," + "readreverse," + "compact," + "readrandom," + "findkeysbyfield," + "readseq," + "readreverse," + "fill100K,"; +// "crc32c," +// "snappycomp," +// "snappyuncomp," +// "zstdcomp," +// "zstduncomp,"; + +// Number of key/values to place in database +static int FLAGS_num = 1000000; // Number of read operations to do. If negative, do FLAGS_num reads. static int FLAGS_reads = -1; +// Number of given fields used in FindKeysByField test. If negative, write in half of FLAGS_num targets with given field. +static int FLAGS_num_fields = 80000; + // Number of concurrent threads to run. static int FLAGS_threads = 1; // Size of each value -static int FLAGS_value_size = 1024*1024; +static int FLAGS_value_size = 100; // Arrange to generate values that shrink to this fraction of // their original size after compression @@ -118,973 +125,1094 @@ static bool FLAGS_use_existing_db = false; // If true, reuse existing log/MANIFEST files when re-opening a database. static bool FLAGS_reuse_logs = false; +// If true, use compression. +static bool FLAGS_compression = true; + // Use the db with the following name. -static const char* FLAGS_db = nullptr; +static const char* FLAGS_db = "benchmark_db"; + +// ZSTD compression level to try out +static int FLAGS_zstd_compression_level = 1; namespace leveldb { - namespace { - leveldb::Env* g_env = nullptr; +namespace { +leveldb::Env* g_env = nullptr; - class CountComparator : public Comparator { - public: - CountComparator(const Comparator* wrapped) : wrapped_(wrapped) {} - ~CountComparator() override {} - int Compare(const Slice& a, const Slice& b) const override { - count_.fetch_add(1, std::memory_order_relaxed); - return wrapped_->Compare(a, b); - } - const char* Name() const override { return wrapped_->Name(); } - void FindShortestSeparator(std::string* start, - const Slice& limit) const override { - wrapped_->FindShortestSeparator(start, limit); - } +class CountComparator : public Comparator { + public: + CountComparator(const Comparator* wrapped) : wrapped_(wrapped) {} + ~CountComparator() override {} + int Compare(const Slice& a, const Slice& b) const override { + count_.fetch_add(1, std::memory_order_relaxed); + return wrapped_->Compare(a, b); + } + const char* Name() const override { return wrapped_->Name(); } + void FindShortestSeparator(std::string* start, + const Slice& limit) const override { + wrapped_->FindShortestSeparator(start, limit); + } - void FindShortSuccessor(std::string* key) const override { - return wrapped_->FindShortSuccessor(key); - } + void FindShortSuccessor(std::string* key) const override { + return wrapped_->FindShortSuccessor(key); + } - size_t comparisons() const { return count_.load(std::memory_order_relaxed); } + size_t comparisons() const { return count_.load(std::memory_order_relaxed); } - void reset() { count_.store(0, std::memory_order_relaxed); } + void reset() { count_.store(0, std::memory_order_relaxed); } - private: - mutable std::atomic count_{0}; - const Comparator* const wrapped_; - }; + private: + mutable std::atomic count_{0}; + const Comparator* const wrapped_; +}; // Helper for quickly generating random data. - class RandomGenerator { - private: - std::string data_; - int pos_; - - public: - RandomGenerator() { - // We use a limited amount of data over and over again and ensure - // that it is larger than the compression window (32KB), and also - // large enough to serve all typical value sizes we want to write. - Random rnd(301); - std::string piece; - while (data_.size() < 1048576) { - // Add a short fragment that is as compressible as specified - // by FLAGS_compression_ratio. - test::CompressibleString(&rnd, FLAGS_compression_ratio, 100*1024*1024, &piece); - data_.append(piece); - } - pos_ = 0; - } - - Slice Generate(size_t len) { - if (pos_ + len > data_.size()) { - pos_ = 0; - assert(len < data_.size()); - } - pos_ += len; - return Slice(data_.data() + pos_ - len, len); - } - }; - - class KeyBuffer { - public: - KeyBuffer() { - assert(FLAGS_key_prefix < sizeof(buffer_)); - memset(buffer_, 'a', FLAGS_key_prefix); - } - KeyBuffer& operator=(KeyBuffer& other) = delete; - KeyBuffer(KeyBuffer& other) = delete; - - void Set(int k) { - std::snprintf(buffer_ + FLAGS_key_prefix, - sizeof(buffer_) - FLAGS_key_prefix, "%016d", k); - } - - Slice slice() const { return Slice(buffer_, FLAGS_key_prefix + 16); } - - private: - char buffer_[1024]; - }; +class RandomGenerator { + private: + std::string data_; + int pos_; + + public: + RandomGenerator() { + // We use a limited amount of data over and over again and ensure + // that it is larger than the compression window (32KB), and also + // large enough to serve all typical value sizes we want to write. + Random rnd(301); + std::string piece; + while (data_.size() < 1048576) { + // Add a short fragment that is as compressible as specified + // by FLAGS_compression_ratio. + test::CompressibleString(&rnd, FLAGS_compression_ratio, 100, &piece); + data_.append(piece); + } + pos_ = 0; + } + + Slice Generate(size_t len) { + if (pos_ + len > data_.size()) { + pos_ = 0; + assert(len < data_.size()); + } + pos_ += len; + return Slice(data_.data() + pos_ - len, len); + } +}; + +class KeyBuffer { + public: + KeyBuffer() { + assert(FLAGS_key_prefix < sizeof(buffer_)); + memset(buffer_, 'a', FLAGS_key_prefix); + } + KeyBuffer& operator=(KeyBuffer& other) = delete; + KeyBuffer(KeyBuffer& other) = delete; + + void Set(int k) { + std::snprintf(buffer_ + FLAGS_key_prefix, + sizeof(buffer_) - FLAGS_key_prefix, "%016d", k); + } + + Slice slice() const { return Slice(buffer_, FLAGS_key_prefix + 16); } + + private: + char buffer_[1024]; +}; #if defined(__linux) - static Slice TrimSpace(Slice s) { - size_t start = 0; - while (start < s.size() && isspace(s[start])) { - start++; - } - size_t limit = s.size(); - while (limit > start && isspace(s[limit - 1])) { - limit--; - } - return Slice(s.data() + start, limit - start); - } +static Slice TrimSpace(Slice s) { + size_t start = 0; + while (start < s.size() && isspace(s[start])) { + start++; + } + size_t limit = s.size(); + while (limit > start && isspace(s[limit - 1])) { + limit--; + } + return Slice(s.data() + start, limit - start); +} #endif - static void AppendWithSpace(std::string* str, Slice msg) { - if (msg.empty()) return; - if (!str->empty()) { - str->push_back(' '); - } - str->append(msg.data(), msg.size()); - } +static void AppendWithSpace(std::string* str, Slice msg) { + if (msg.empty()) return; + if (!str->empty()) { + str->push_back(' '); + } + str->append(msg.data(), msg.size()); +} - class Stats { - private: - double start_; - double finish_; - double seconds_; - int done_; - int next_report_; - int64_t bytes_; - double last_op_finish_; - Histogram hist_; - std::string message_; - - public: - Stats() { Start(); } - - void Start() { - next_report_ = 100; - hist_.Clear(); - done_ = 0; - bytes_ = 0; - seconds_ = 0; - message_.clear(); - start_ = finish_ = last_op_finish_ = g_env->NowMicros(); - } - - void Merge(const Stats& other) { - hist_.Merge(other.hist_); - done_ += other.done_; - bytes_ += other.bytes_; - seconds_ += other.seconds_; - if (other.start_ < start_) start_ = other.start_; - if (other.finish_ > finish_) finish_ = other.finish_; - - // Just keep the messages from one thread - if (message_.empty()) message_ = other.message_; - } - - void Stop() { - finish_ = g_env->NowMicros(); - seconds_ = (finish_ - start_) * 1e-6; - } - - void AddMessage(Slice msg) { AppendWithSpace(&message_, msg); } - - void FinishedSingleOp() { - if (FLAGS_histogram) { - double now = g_env->NowMicros(); - double micros = now - last_op_finish_; - hist_.Add(micros); - if (micros > 20000) { - std::fprintf(stderr, "long op: %.1f micros%30s\r", micros, ""); - std::fflush(stderr); - } - last_op_finish_ = now; - } - - done_++; - if (done_ >= next_report_) { - if (next_report_ < 1000) - next_report_ += 100; - else if (next_report_ < 5000) - next_report_ += 500; - else if (next_report_ < 10000) - next_report_ += 1000; - else if (next_report_ < 50000) - next_report_ += 5000; - else if (next_report_ < 100000) - next_report_ += 10000; - else if (next_report_ < 500000) - next_report_ += 50000; - else - next_report_ += 100000; - std::fprintf(stderr, "... finished %d ops%30s\r", done_, ""); - std::fflush(stderr); - } - } - - void AddBytes(int64_t n) { bytes_ += n; } - - void Report(const Slice& name) { - // Pretend at least one op was done in case we are running a benchmark - // that does not call FinishedSingleOp(). - if (done_ < 1) done_ = 1; - - std::string extra; - if (bytes_ > 0) { - // Rate is computed on actual elapsed time, not the sum of per-thread - // elapsed times. - double elapsed = (finish_ - start_) * 1e-6; - char rate[100]; - std::snprintf(rate, sizeof(rate), "%6.1f MB/s", - (bytes_ / 1048576.0) / elapsed); - extra = rate; - } - AppendWithSpace(&extra, message_); - - std::fprintf(stdout, "%-12s : %11.3f micros/op;%s%s\n", - name.ToString().c_str(), seconds_ * 1e6 / done_, - (extra.empty() ? "" : " "), extra.c_str()); - if (FLAGS_histogram) { - std::fprintf(stdout, "Microseconds per op:\n%s\n", - hist_.ToString().c_str()); - } - std::fflush(stdout); - } - }; +class Stats { + private: + double start_; + double finish_; + double seconds_; + int done_; + int next_report_; + int64_t bytes_; + double last_op_finish_; + Histogram hist_; + std::string message_; + + public: + Stats() { Start(); } + + void Start() { + next_report_ = 100; + hist_.Clear(); + done_ = 0; + bytes_ = 0; + seconds_ = 0; + message_.clear(); + start_ = finish_ = last_op_finish_ = g_env->NowMicros(); + } + + void Merge(const Stats& other) { + hist_.Merge(other.hist_); + done_ += other.done_; + bytes_ += other.bytes_; + seconds_ += other.seconds_; + if (other.start_ < start_) start_ = other.start_; + if (other.finish_ > finish_) finish_ = other.finish_; + + // Just keep the messages from one thread + if (message_.empty()) message_ = other.message_; + } + + void Stop() { + finish_ = g_env->NowMicros(); + seconds_ = (finish_ - start_) * 1e-6; + } + + void AddMessage(Slice msg) { AppendWithSpace(&message_, msg); } + + void FinishedSingleOp() { + if (FLAGS_histogram) { + double now = g_env->NowMicros(); + double micros = now - last_op_finish_; + hist_.Add(micros); + if (micros > 20000) { + std::fprintf(stderr, "long op: %.1f micros%30s\r", micros, ""); + std::fflush(stderr); + } + last_op_finish_ = now; + } + + done_++; + if (done_ >= next_report_) { + if (next_report_ < 1000) + next_report_ += 100; + else if (next_report_ < 5000) + next_report_ += 500; + else if (next_report_ < 10000) + next_report_ += 1000; + else if (next_report_ < 50000) + next_report_ += 5000; + else if (next_report_ < 100000) + next_report_ += 10000; + else if (next_report_ < 500000) + next_report_ += 50000; + else + next_report_ += 100000; + std::fprintf(stderr, "... finished %d ops%30s\r", done_, ""); + std::fflush(stderr); + } + } + + void AddBytes(int64_t n) { bytes_ += n; } + + void Report(const Slice& name) { + // Pretend at least one op was done in case we are running a benchmark + // that does not call FinishedSingleOp(). + if (done_ < 1) done_ = 1; + + std::string extra; + if (bytes_ > 0) { + // Rate is computed on actual elapsed time, not the sum of per-thread + // elapsed times. + double elapsed = (finish_ - start_) * 1e-6; + char rate[100]; + std::snprintf(rate, sizeof(rate), "%6.1f MB/s", + (bytes_ / 1048576.0) / elapsed); + extra = rate; + } + AppendWithSpace(&extra, message_); + + std::fprintf(stdout, "%-12s : %11.3f micros/op;%s%s\n", + name.ToString().c_str(), seconds_ * 1e6 / done_, + (extra.empty() ? "" : " "), extra.c_str()); + if (FLAGS_histogram) { + std::fprintf(stdout, "Microseconds per op:\n%s\n", + hist_.ToString().c_str()); + } + std::fflush(stdout); + } +}; // State shared by all concurrent executions of the same benchmark. - struct SharedState { - port::Mutex mu; - port::CondVar cv GUARDED_BY(mu); - int total GUARDED_BY(mu); +struct SharedState { + port::Mutex mu; + port::CondVar cv GUARDED_BY(mu); + int total GUARDED_BY(mu); - // Each thread goes through the following states: - // (1) initializing - // (2) waiting for others to be initialized - // (3) running - // (4) done + // Each thread goes through the following states: + // (1) initializing + // (2) waiting for others to be initialized + // (3) running + // (4) done - int num_initialized GUARDED_BY(mu); - int num_done GUARDED_BY(mu); - bool start GUARDED_BY(mu); + int num_initialized GUARDED_BY(mu); + int num_done GUARDED_BY(mu); + bool start GUARDED_BY(mu); - SharedState(int total) - : cv(&mu), total(total), num_initialized(0), num_done(0), start(false) {} - }; + SharedState(int total) + : cv(&mu), total(total), num_initialized(0), num_done(0), start(false) {} +}; // Per-thread state for concurrent executions of the same benchmark. - struct ThreadState { - int tid; // 0..n-1 when running in n threads - Random rand; // Has different seeds for different threads - Stats stats; - SharedState* shared; - - ThreadState(int index, int seed) : tid(index), rand(seed), shared(nullptr) {} - }; - - } // namespace - - class Benchmark { - private: - Cache* cache_; - const FilterPolicy* filter_policy_; - DB* db_; - int num_; - int value_size_; - int entries_per_batch_; - WriteOptions write_options_; - int reads_; - int heap_counter_; - CountComparator count_comparator_; - int total_thread_count_; - - void PrintHeader() { - const int kKeySize = 16 + FLAGS_key_prefix; - PrintEnvironment(); - std::fprintf(stdout, "Keys: %d bytes each\n", kKeySize); - std::fprintf( - stdout, "Values: %d bytes each (%d bytes after compression)\n", - FLAGS_value_size, - static_cast(FLAGS_value_size * FLAGS_compression_ratio + 0.5)); - std::fprintf(stdout, "Entries: %d\n", num_); - std::fprintf(stdout, "RawSize: %.1f MB (estimated)\n", - ((static_cast(kKeySize + FLAGS_value_size) * num_) / - 1048576.0)); - std::fprintf( - stdout, "FileSize: %.1f MB (estimated)\n", - (((kKeySize + FLAGS_value_size * FLAGS_compression_ratio) * num_) / - 1048576.0)); - PrintWarnings(); - std::fprintf(stdout, "------------------------------------------------\n"); - } +struct ThreadState { + int tid; // 0..n-1 when running in n threads + Random rand; // Has different seeds for different threads + Stats stats; + SharedState* shared; + + ThreadState(int index, int seed) : tid(index), rand(seed), shared(nullptr) {} +}; + +void Compress( + ThreadState* thread, std::string name, + std::function compress_func) { + RandomGenerator gen; + Slice input = gen.Generate(Options().block_size); + int64_t bytes = 0; + int64_t produced = 0; + bool ok = true; + std::string compressed; + while (ok && bytes < 1024 * 1048576) { // Compress 1G + ok = compress_func(input.data(), input.size(), &compressed); + produced += compressed.size(); + bytes += input.size(); + thread->stats.FinishedSingleOp(); + } + + if (!ok) { + thread->stats.AddMessage("(" + name + " failure)"); + } else { + char buf[100]; + std::snprintf(buf, sizeof(buf), "(output: %.1f%%)", + (produced * 100.0) / bytes); + thread->stats.AddMessage(buf); + thread->stats.AddBytes(bytes); + } +} + +void Uncompress( + ThreadState* thread, std::string name, + std::function compress_func, + std::function uncompress_func) { + RandomGenerator gen; + Slice input = gen.Generate(Options().block_size); + std::string compressed; + bool ok = compress_func(input.data(), input.size(), &compressed); + int64_t bytes = 0; + char* uncompressed = new char[input.size()]; + while (ok && bytes < 1024 * 1048576) { // Compress 1G + ok = uncompress_func(compressed.data(), compressed.size(), uncompressed); + bytes += input.size(); + thread->stats.FinishedSingleOp(); + } + delete[] uncompressed; + + if (!ok) { + thread->stats.AddMessage("(" + name + " failure)"); + } else { + thread->stats.AddBytes(bytes); + } +} - void PrintWarnings() { +} // namespace + +class Benchmark { + private: + Cache* cache_; + const FilterPolicy* filter_policy_; + DB* db_; + int num_; + int value_size_; + int entries_per_batch_; + WriteOptions write_options_; + int reads_; + int heap_counter_; + CountComparator count_comparator_; + int total_thread_count_; + int num_fields; // 插入的fields数量 + + void PrintHeader() { + const int kKeySize = 16 + FLAGS_key_prefix; + PrintEnvironment(); + std::fprintf(stdout, "Keys: %d bytes each\n", kKeySize); + std::fprintf( + stdout, "Values: %d bytes each (%d bytes after compression)\n", + FLAGS_value_size, + static_cast(FLAGS_value_size * FLAGS_compression_ratio + 0.5)); + std::fprintf(stdout, "Entries: %d\n", num_); + std::fprintf(stdout, "RawSize: %.1f MB (estimated)\n", + ((static_cast(kKeySize + FLAGS_value_size) * num_) / + 1048576.0)); + std::fprintf( + stdout, "FileSize: %.1f MB (estimated)\n", + (((kKeySize + FLAGS_value_size * FLAGS_compression_ratio) * num_) / + 1048576.0)); + PrintWarnings(); + std::fprintf(stdout, "------------------------------------------------\n"); + } + + void PrintWarnings() { #if defined(__GNUC__) && !defined(__OPTIMIZE__) - std::fprintf( - stdout, - "WARNING: Optimization is disabled: benchmarks unnecessarily slow\n"); + std::fprintf( + stdout, + "WARNING: Optimization is disabled: benchmarks unnecessarily slow\n"); #endif #ifndef NDEBUG - std::fprintf( - stdout, - "WARNING: Assertions are enabled; benchmarks unnecessarily slow\n"); + std::fprintf( + stdout, + "WARNING: Assertions are enabled; benchmarks unnecessarily slow\n"); #endif - // See if snappy is working by attempting to compress a compressible string - const char text[] = "yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy"; - std::string compressed; - if (!port::Snappy_Compress(text, sizeof(text), &compressed)) { - std::fprintf(stdout, "WARNING: Snappy compression is not enabled\n"); - } else if (compressed.size() >= sizeof(text)) { - std::fprintf(stdout, "WARNING: Snappy compression is not effective\n"); - } - } + // See if snappy is working by attempting to compress a compressible string + const char text[] = "yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy"; + std::string compressed; + if (!port::Snappy_Compress(text, sizeof(text), &compressed)) { + std::fprintf(stdout, "WARNING: Snappy compression is not enabled\n"); + } else if (compressed.size() >= sizeof(text)) { + std::fprintf(stdout, "WARNING: Snappy compression is not effective\n"); + } + } - void PrintEnvironment() { - std::fprintf(stderr, "LevelDB: version %d.%d\n", kMajorVersion, - kMinorVersion); + void PrintEnvironment() { + std::fprintf(stderr, "LevelDB: version %d.%d\n", kMajorVersion, + kMinorVersion); #if defined(__linux) - time_t now = time(nullptr); - std::fprintf(stderr, "Date: %s", - ctime(&now)); // ctime() adds newline - - FILE* cpuinfo = std::fopen("/proc/cpuinfo", "r"); - if (cpuinfo != nullptr) { - char line[1000]; - int num_cpus = 0; - std::string cpu_type; - std::string cache_size; - while (fgets(line, sizeof(line), cpuinfo) != nullptr) { - const char* sep = strchr(line, ':'); - if (sep == nullptr) { - continue; - } - Slice key = TrimSpace(Slice(line, sep - 1 - line)); - Slice val = TrimSpace(Slice(sep + 1)); - if (key == "model name") { - ++num_cpus; - cpu_type = val.ToString(); - } else if (key == "cache size") { - cache_size = val.ToString(); - } - } - std::fclose(cpuinfo); - std::fprintf(stderr, "CPU: %d * %s\n", num_cpus, cpu_type.c_str()); - std::fprintf(stderr, "CPUCache: %s\n", cache_size.c_str()); - } -#endif + time_t now = time(nullptr); + std::fprintf(stderr, "Date: %s", + ctime(&now)); // ctime() adds newline + + FILE* cpuinfo = std::fopen("/proc/cpuinfo", "r"); + if (cpuinfo != nullptr) { + char line[1000]; + int num_cpus = 0; + std::string cpu_type; + std::string cache_size; + while (fgets(line, sizeof(line), cpuinfo) != nullptr) { + const char* sep = strchr(line, ':'); + if (sep == nullptr) { + continue; } - - public: - Benchmark() - : cache_(FLAGS_cache_size >= 0 ? NewLRUCache(FLAGS_cache_size) : nullptr), - filter_policy_(FLAGS_bloom_bits >= 0 - ? NewBloomFilterPolicy(FLAGS_bloom_bits) - : nullptr), - db_(nullptr), - num_(FLAGS_num), - value_size_(FLAGS_value_size), - entries_per_batch_(1), - reads_(FLAGS_reads < 0 ? FLAGS_num : FLAGS_reads), - heap_counter_(0), - count_comparator_(BytewiseComparator()), - total_thread_count_(0) { - std::vector files; - g_env->GetChildren(FLAGS_db, &files); - for (size_t i = 0; i < files.size(); i++) { - if (Slice(files[i]).starts_with("heap-")) { - g_env->RemoveFile(std::string(FLAGS_db) + "/" + files[i]); - } - } - if (!FLAGS_use_existing_db) { - DestroyDB(FLAGS_db, Options()); - } + Slice key = TrimSpace(Slice(line, sep - 1 - line)); + Slice val = TrimSpace(Slice(sep + 1)); + if (key == "model name") { + ++num_cpus; + cpu_type = val.ToString(); + } else if (key == "cache size") { + cache_size = val.ToString(); } - - ~Benchmark() { - delete db_; - delete cache_; - delete filter_policy_; + } + std::fclose(cpuinfo); + std::fprintf(stderr, "CPU: %d * %s\n", num_cpus, cpu_type.c_str()); + std::fprintf(stderr, "CPUCache: %s\n", cache_size.c_str()); + } +#endif + } + + public: + Benchmark() + : cache_(FLAGS_cache_size >= 0 ? NewLRUCache(FLAGS_cache_size) : nullptr), + filter_policy_(FLAGS_bloom_bits >= 0 + ? NewBloomFilterPolicy(FLAGS_bloom_bits) + : nullptr), + db_(nullptr), + num_(FLAGS_num), + value_size_(FLAGS_value_size), + entries_per_batch_(1), + reads_(FLAGS_reads < 0 ? FLAGS_num : FLAGS_reads), + heap_counter_(0), + count_comparator_(BytewiseComparator()), + total_thread_count_(0), + num_fields(FLAGS_num_fields < 0 ? FLAGS_num / 2 : FLAGS_num_fields) { + std::vector files; + g_env->GetChildren(FLAGS_db, &files); + for (size_t i = 0; i < files.size(); i++) { + if (Slice(files[i]).starts_with("heap-")) { + g_env->RemoveFile(std::string(FLAGS_db) + "/" + files[i]); + } + } + if (!FLAGS_use_existing_db) { + DestroyDB(FLAGS_db, Options()); + } + } + + ~Benchmark() { + delete db_; + delete cache_; + delete filter_policy_; + } + + void Run() { + PrintHeader(); + Open(); + + const char* benchmarks = FLAGS_benchmarks; + while (benchmarks != nullptr) { + const char* sep = strchr(benchmarks, ','); + Slice name; + if (sep == nullptr) { + name = benchmarks; + benchmarks = nullptr; + } else { + name = Slice(benchmarks, sep - benchmarks); + benchmarks = sep + 1; + } + + // Reset parameters that may be overridden below + num_ = FLAGS_num; + reads_ = (FLAGS_reads < 0 ? FLAGS_num : FLAGS_reads); + value_size_ = FLAGS_value_size; + entries_per_batch_ = 1; + write_options_ = WriteOptions(); + + void (Benchmark::*method)(ThreadState*) = nullptr; + bool fresh_db = false; + int num_threads = FLAGS_threads; + + if (name == Slice("open")) { + method = &Benchmark::OpenBench; + num_ /= 10000; + if (num_ < 1) num_ = 1; + } else if (name == Slice("fillseq")) { + fresh_db = true; + method = &Benchmark::WriteSeq; + } else if (name == Slice("fillbatch")) { + fresh_db = true; + entries_per_batch_ = 1000; + method = &Benchmark::WriteSeq; + } else if (name == Slice("fillrandom")) { + fresh_db = true; + method = &Benchmark::WriteRandom; + } else if (name == Slice("overwrite")) { + fresh_db = false; + method = &Benchmark::WriteRandom; + } else if (name == Slice("fillsync")) { + fresh_db = true; + num_ /= 1000; + write_options_.sync = true; + method = &Benchmark::WriteRandom; + } else if (name == Slice("fill100K")) { + fresh_db = true; + num_ /= 1000; + value_size_ = 100 * 1000; + method = &Benchmark::WriteRandom; + } else if (name == Slice("readseq")) { + method = &Benchmark::ReadSequential; + } else if (name == Slice("readreverse")) { + method = &Benchmark::ReadReverse; + } else if (name == Slice("readrandom")) { + method = &Benchmark::ReadRandom; + } else if (name == Slice("readmissing")) { + method = &Benchmark::ReadMissing; + } else if (name == Slice("seekrandom")) { + method = &Benchmark::SeekRandom; + } else if (name == Slice("seekordered")) { + method = &Benchmark::SeekOrdered; + } else if (name == Slice("findkeysbyfield")) { + method = &Benchmark::FindKeysByField; + } else if (name == Slice("readhot")) { + method = &Benchmark::ReadHot; + } else if (name == Slice("readrandomsmall")) { + reads_ /= 1000; + method = &Benchmark::ReadRandom; + } else if (name == Slice("deleteseq")) { + method = &Benchmark::DeleteSeq; + } else if (name == Slice("deleterandom")) { + method = &Benchmark::DeleteRandom; + } else if (name == Slice("readwhilewriting")) { + num_threads++; // Add extra thread for writing + method = &Benchmark::ReadWhileWriting; + } else if (name == Slice("compact")) { + method = &Benchmark::Compact; + } else if (name == Slice("crc32c")) { + method = &Benchmark::Crc32c; + } else if (name == Slice("snappycomp")) { + method = &Benchmark::SnappyCompress; + } else if (name == Slice("snappyuncomp")) { + method = &Benchmark::SnappyUncompress; + } else if (name == Slice("zstdcomp")) { + method = &Benchmark::ZstdCompress; + } else if (name == Slice("zstduncomp")) { + method = &Benchmark::ZstdUncompress; + } else if (name == Slice("heapprofile")) { + HeapProfile(); + } else if (name == Slice("stats")) { + PrintStats("leveldb.stats"); + } else if (name == Slice("sstables")) { + PrintStats("leveldb.sstables"); + } else { + if (!name.empty()) { // No error message for empty name + std::fprintf(stderr, "unknown benchmark '%s'\n", + name.ToString().c_str()); } + } - void Run() { - PrintHeader(); - Open(); - - const char* benchmarks = FLAGS_benchmarks; - while (benchmarks != nullptr) { - const char* sep = strchr(benchmarks, ','); - Slice name; - if (sep == nullptr) { - name = benchmarks; - benchmarks = nullptr; - } else { - name = Slice(benchmarks, sep - benchmarks); - benchmarks = sep + 1; - } - - // Reset parameters that may be overridden below - num_ = FLAGS_num; - reads_ = (FLAGS_reads < 0 ? FLAGS_num : FLAGS_reads); - value_size_ = FLAGS_value_size; - entries_per_batch_ = 1; - write_options_ = WriteOptions(); - - void (Benchmark::*method)(ThreadState*) = nullptr; - bool fresh_db = false; - int num_threads = FLAGS_threads; - - if (name == Slice("open")) { - method = &Benchmark::OpenBench; - num_ /= 10000; - if (num_ < 1) num_ = 1; - } else if (name == Slice("fillseq")) { - fresh_db = true; - method = &Benchmark::WriteSeq; - } else if (name == Slice("fillbatch")) { - fresh_db = true; - entries_per_batch_ = 1000; - method = &Benchmark::WriteSeq; - } else if (name == Slice("fillrandom")) { - fresh_db = true; - method = &Benchmark::WriteRandom; - } else if (name == Slice("overwrite")) { - fresh_db = false; - method = &Benchmark::WriteRandom; - } else if (name == Slice("fillsync")) { - fresh_db = true; - num_ /= 1000; - write_options_.sync = true; - method = &Benchmark::WriteRandom; - } else if (name == Slice("fill100K")) { - fresh_db = true; - num_ /= 1000; - value_size_ = 100 * 1000; - method = &Benchmark::WriteRandom; - } else if (name == Slice("readseq")) { - method = &Benchmark::ReadSequential; - } else if (name == Slice("readreverse")) { - method = &Benchmark::ReadReverse; - } else if (name == Slice("readrandom")) { - method = &Benchmark::ReadRandom; - } else if (name == Slice("readmissing")) { - method = &Benchmark::ReadMissing; - } else if (name == Slice("seekrandom")) { - method = &Benchmark::SeekRandom; - } else if (name == Slice("seekordered")) { - method = &Benchmark::SeekOrdered; - } else if (name == Slice("readhot")) { - method = &Benchmark::ReadHot; - } else if (name == Slice("readrandomsmall")) { - reads_ /= 1000; - method = &Benchmark::ReadRandom; - } else if (name == Slice("deleteseq")) { - method = &Benchmark::DeleteSeq; - } else if (name == Slice("deleterandom")) { - method = &Benchmark::DeleteRandom; - } else if (name == Slice("readwhilewriting")) { - num_threads++; // Add extra thread for writing - method = &Benchmark::ReadWhileWriting; - } else if (name == Slice("compact")) { - method = &Benchmark::Compact; - } else if (name == Slice("crc32c")) { - method = &Benchmark::Crc32c; - } else if (name == Slice("snappycomp")) { - method = &Benchmark::SnappyCompress; - } else if (name == Slice("snappyuncomp")) { - method = &Benchmark::SnappyUncompress; - } else if (name == Slice("heapprofile")) { - HeapProfile(); - } else if (name == Slice("stats")) { - PrintStats("leveldb.stats"); - } else if (name == Slice("sstables")) { - PrintStats("leveldb.sstables"); - } else { - if (!name.empty()) { // No error message for empty name - std::fprintf(stderr, "unknown benchmark '%s'\n", - name.ToString().c_str()); - } - } - - if (fresh_db) { - if (FLAGS_use_existing_db) { - std::fprintf(stdout, "%-12s : skipped (--use_existing_db is true)\n", - name.ToString().c_str()); - method = nullptr; - } else { - delete db_; - db_ = nullptr; - DestroyDB(FLAGS_db, Options()); - Open(); - } - } - - if (method != nullptr) { - RunBenchmark(num_threads, name, method); - } - } + if (fresh_db) { + if (FLAGS_use_existing_db) { + std::fprintf(stdout, "%-12s : skipped (--use_existing_db is true)\n", + name.ToString().c_str()); + method = nullptr; + } else { + delete db_; + db_ = nullptr; + DestroyDB(FLAGS_db, Options()); + Open(); } + } - private: - struct ThreadArg { - Benchmark* bm; - SharedState* shared; - ThreadState* thread; - void (Benchmark::*method)(ThreadState*); - }; - - static void ThreadBody(void* v) { - ThreadArg* arg = reinterpret_cast(v); - SharedState* shared = arg->shared; - ThreadState* thread = arg->thread; - { - MutexLock l(&shared->mu); - shared->num_initialized++; - if (shared->num_initialized >= shared->total) { - shared->cv.SignalAll(); - } - while (!shared->start) { - shared->cv.Wait(); - } - } - - thread->stats.Start(); - (arg->bm->*(arg->method))(thread); - thread->stats.Stop(); - - { - MutexLock l(&shared->mu); - shared->num_done++; - if (shared->num_done >= shared->total) { - shared->cv.SignalAll(); - } - } - } + if (method != nullptr) { + RunBenchmark(num_threads, name, method); + } + } + } + + private: + struct ThreadArg { + Benchmark* bm; + SharedState* shared; + ThreadState* thread; + void (Benchmark::*method)(ThreadState*); + }; + + static void ThreadBody(void* v) { + ThreadArg* arg = reinterpret_cast(v); + SharedState* shared = arg->shared; + ThreadState* thread = arg->thread; + { + MutexLock l(&shared->mu); + shared->num_initialized++; + if (shared->num_initialized >= shared->total) { + shared->cv.SignalAll(); + } + while (!shared->start) { + shared->cv.Wait(); + } + } - void RunBenchmark(int n, Slice name, - void (Benchmark::*method)(ThreadState*)) { - SharedState shared(n); - - ThreadArg* arg = new ThreadArg[n]; - for (int i = 0; i < n; i++) { - arg[i].bm = this; - arg[i].method = method; - arg[i].shared = &shared; - ++total_thread_count_; - // Seed the thread's random state deterministically based upon thread - // creation across all benchmarks. This ensures that the seeds are unique - // but reproducible when rerunning the same set of benchmarks. - arg[i].thread = new ThreadState(i, /*seed=*/1000 + total_thread_count_); - arg[i].thread->shared = &shared; - g_env->StartThread(ThreadBody, &arg[i]); - } - - shared.mu.Lock(); - while (shared.num_initialized < n) { - shared.cv.Wait(); - } - - shared.start = true; - shared.cv.SignalAll(); - while (shared.num_done < n) { - shared.cv.Wait(); - } - shared.mu.Unlock(); - - for (int i = 1; i < n; i++) { - arg[0].thread->stats.Merge(arg[i].thread->stats); - } - arg[0].thread->stats.Report(name); - if (FLAGS_comparisons) { - fprintf(stdout, "Comparisons: %zu\n", count_comparator_.comparisons()); - count_comparator_.reset(); - fflush(stdout); - } - - for (int i = 0; i < n; i++) { - delete arg[i].thread; - } - delete[] arg; - } + thread->stats.Start(); + (arg->bm->*(arg->method))(thread); + thread->stats.Stop(); - void Crc32c(ThreadState* thread) { - // Checksum about 500MB of data total - const int size = 4096; - const char* label = "(4K per op)"; - std::string data(size, 'x'); - int64_t bytes = 0; - uint32_t crc = 0; - while (bytes < 500 * 1048576) { - crc = crc32c::Value(data.data(), size); - thread->stats.FinishedSingleOp(); - bytes += size; - } - // Print so result is not dead - std::fprintf(stderr, "... crc=0x%x\r", static_cast(crc)); - - thread->stats.AddBytes(bytes); - thread->stats.AddMessage(label); - } + { + MutexLock l(&shared->mu); + shared->num_done++; + if (shared->num_done >= shared->total) { + shared->cv.SignalAll(); + } + } + } + + void RunBenchmark(int n, Slice name, + void (Benchmark::*method)(ThreadState*)) { + SharedState shared(n); + + ThreadArg* arg = new ThreadArg[n]; + for (int i = 0; i < n; i++) { + arg[i].bm = this; + arg[i].method = method; + arg[i].shared = &shared; + ++total_thread_count_; + // Seed the thread's random state deterministically based upon thread + // creation across all benchmarks. This ensures that the seeds are unique + // but reproducible when rerunning the same set of benchmarks. + arg[i].thread = new ThreadState(i, /*seed=*/1000 + total_thread_count_); + arg[i].thread->shared = &shared; + g_env->StartThread(ThreadBody, &arg[i]); + } - void SnappyCompress(ThreadState* thread) { - RandomGenerator gen; - Slice input = gen.Generate(Options().block_size); - int64_t bytes = 0; - int64_t produced = 0; - bool ok = true; - std::string compressed; - while (ok && bytes < 1024 * 1048576) { // Compress 1G - ok = port::Snappy_Compress(input.data(), input.size(), &compressed); - produced += compressed.size(); - bytes += input.size(); - thread->stats.FinishedSingleOp(); - } - - if (!ok) { - thread->stats.AddMessage("(snappy failure)"); - } else { - char buf[100]; - std::snprintf(buf, sizeof(buf), "(output: %.1f%%)", - (produced * 100.0) / bytes); - thread->stats.AddMessage(buf); - thread->stats.AddBytes(bytes); - } - } + shared.mu.Lock(); + while (shared.num_initialized < n) { + shared.cv.Wait(); + } - void SnappyUncompress(ThreadState* thread) { - RandomGenerator gen; - Slice input = gen.Generate(Options().block_size); - std::string compressed; - bool ok = port::Snappy_Compress(input.data(), input.size(), &compressed); - int64_t bytes = 0; - char* uncompressed = new char[input.size()]; - while (ok && bytes < 1024 * 1048576) { // Compress 1G - ok = port::Snappy_Uncompress(compressed.data(), compressed.size(), - uncompressed); - bytes += input.size(); - thread->stats.FinishedSingleOp(); - } - delete[] uncompressed; - - if (!ok) { - thread->stats.AddMessage("(snappy failure)"); - } else { - thread->stats.AddBytes(bytes); - } - } + shared.start = true; + shared.cv.SignalAll(); + while (shared.num_done < n) { + shared.cv.Wait(); + } + shared.mu.Unlock(); - void Open() { - assert(db_ == nullptr); - Options options; - options.env = g_env; - options.create_if_missing = !FLAGS_use_existing_db; - options.block_cache = cache_; - options.write_buffer_size = FLAGS_write_buffer_size; - options.max_file_size = FLAGS_max_file_size; - options.block_size = FLAGS_block_size; - if (FLAGS_comparisons) { - options.comparator = &count_comparator_; - } - options.max_open_files = FLAGS_open_files; - options.filter_policy = filter_policy_; - options.reuse_logs = FLAGS_reuse_logs; - Status s = DB::Open(options, FLAGS_db, &db_); - if (!s.ok()) { - std::fprintf(stderr, "open error: %s\n", s.ToString().c_str()); - std::exit(1); - } - } + for (int i = 1; i < n; i++) { + arg[0].thread->stats.Merge(arg[i].thread->stats); + } + arg[0].thread->stats.Report(name); + if (FLAGS_comparisons) { + fprintf(stdout, "Comparisons: %zu\n", count_comparator_.comparisons()); + count_comparator_.reset(); + fflush(stdout); + } - void OpenBench(ThreadState* thread) { - for (int i = 0; i < num_; i++) { - delete db_; - Open(); - thread->stats.FinishedSingleOp(); - } - } + for (int i = 0; i < n; i++) { + delete arg[i].thread; + } + delete[] arg; + } + + void Crc32c(ThreadState* thread) { + // Checksum about 500MB of data total + const int size = 4096; + const char* label = "(4K per op)"; + std::string data(size, 'x'); + int64_t bytes = 0; + uint32_t crc = 0; + while (bytes < 500 * 1048576) { + crc = crc32c::Value(data.data(), size); + thread->stats.FinishedSingleOp(); + bytes += size; + } + // Print so result is not dead + std::fprintf(stderr, "... crc=0x%x\r", static_cast(crc)); + + thread->stats.AddBytes(bytes); + thread->stats.AddMessage(label); + } + + void SnappyCompress(ThreadState* thread) { + Compress(thread, "snappy", &port::Snappy_Compress); + } + + void SnappyUncompress(ThreadState* thread) { + Uncompress(thread, "snappy", &port::Snappy_Compress, + &port::Snappy_Uncompress); + } + + void ZstdCompress(ThreadState* thread) { + Compress(thread, "zstd", + [](const char* input, size_t length, std::string* output) { + return port::Zstd_Compress(FLAGS_zstd_compression_level, input, + length, output); + }); + } + + void ZstdUncompress(ThreadState* thread) { + Uncompress( + thread, "zstd", + [](const char* input, size_t length, std::string* output) { + return port::Zstd_Compress(FLAGS_zstd_compression_level, input, + length, output); + }, + &port::Zstd_Uncompress); + } + + void Open() { + assert(db_ == nullptr); + Options options; + options.env = g_env; + options.create_if_missing = !FLAGS_use_existing_db; + options.block_cache = cache_; + options.write_buffer_size = FLAGS_write_buffer_size; + options.max_file_size = FLAGS_max_file_size; + options.block_size = FLAGS_block_size; + if (FLAGS_comparisons) { + options.comparator = &count_comparator_; + } + options.max_open_files = FLAGS_open_files; + options.filter_policy = filter_policy_; + options.reuse_logs = FLAGS_reuse_logs; + options.compression = + FLAGS_compression ? kSnappyCompression : kNoCompression; + Status s = DB::Open(options, FLAGS_db, &db_); + if (!s.ok()) { + std::fprintf(stderr, "open error: %s\n", s.ToString().c_str()); + std::exit(1); + } + } - void WriteSeq(ThreadState* thread) { DoWrite(thread, true); } - - void WriteRandom(ThreadState* thread) { DoWrite(thread, false); } - - void DoWrite(ThreadState* thread, bool seq) { - if (num_ != FLAGS_num) { - char msg[100]; - std::snprintf(msg, sizeof(msg), "(%d ops)", num_); - thread->stats.AddMessage(msg); - } - - RandomGenerator gen; - WriteBatch batch; - Status s; - int64_t bytes = 0; - KeyBuffer key; - for (int i = 0; i < num_; i += entries_per_batch_) { - batch.Clear(); - for (int j = 0; j < entries_per_batch_; j++) { - const int k = seq ? i + j : thread->rand.Uniform(FLAGS_num); - key.Set(k); - batch.Put(key.slice(), gen.Generate(value_size_)); - bytes += value_size_ + key.slice().size(); - thread->stats.FinishedSingleOp(); - } - s = db_->Write(write_options_, &batch); - if (!s.ok()) { - std::fprintf(stderr, "put error: %s\n", s.ToString().c_str()); - std::exit(1); - } - } - thread->stats.AddBytes(bytes); - } + void OpenBench(ThreadState* thread) { + for (int i = 0; i < num_; i++) { + delete db_; + Open(); + thread->stats.FinishedSingleOp(); + } + } - void ReadSequential(ThreadState* thread) { - Iterator* iter = db_->NewIterator(ReadOptions()); - int i = 0; - int64_t bytes = 0; - for (iter->SeekToFirst(); i < reads_ && iter->Valid(); iter->Next()) { - bytes += iter->key().size() + iter->value().size(); - thread->stats.FinishedSingleOp(); - ++i; - } - delete iter; - thread->stats.AddBytes(bytes); - } + void WriteSeq(ThreadState* thread) { DoWrite(thread, true); } - void ReadReverse(ThreadState* thread) { - Iterator* iter = db_->NewIterator(ReadOptions()); - int i = 0; - int64_t bytes = 0; - for (iter->SeekToLast(); i < reads_ && iter->Valid(); iter->Prev()) { - bytes += iter->key().size() + iter->value().size(); - thread->stats.FinishedSingleOp(); - ++i; - } - delete iter; - thread->stats.AddBytes(bytes); - } + void WriteRandom(ThreadState* thread) { DoWrite(thread, false); } - void ReadRandom(ThreadState* thread) { - ReadOptions options; - std::string value; - int found = 0; - KeyBuffer key; - for (int i = 0; i < reads_; i++) { - const int k = thread->rand.Uniform(FLAGS_num); - key.Set(k); - if (db_->Get(options, key.slice(), &value).ok()) { - found++; - } - thread->stats.FinishedSingleOp(); - } - char msg[100]; - std::snprintf(msg, sizeof(msg), "(%d of %d found)", found, num_); - thread->stats.AddMessage(msg); - } + void DoWrite(ThreadState* thread, bool seq) { + if (num_ != FLAGS_num) { + char msg[100]; + std::snprintf(msg, sizeof(msg), "(%d ops)", num_); + thread->stats.AddMessage(msg); + } - void ReadMissing(ThreadState* thread) { - ReadOptions options; - std::string value; - KeyBuffer key; - for (int i = 0; i < reads_; i++) { - const int k = thread->rand.Uniform(FLAGS_num); - key.Set(k); - Slice s = Slice(key.slice().data(), key.slice().size() - 1); - db_->Get(options, s, &value); - thread->stats.FinishedSingleOp(); - } - } + RandomGenerator gen; + WriteBatch batch; + Status s; + int64_t bytes = 0; + KeyBuffer key; + for (int i = 0; i < num_; i += entries_per_batch_) { + batch.Clear(); + for (int j = 0; j < entries_per_batch_; j++) { + const int k = seq ? i + j : thread->rand.Uniform(FLAGS_num); + key.Set(k); + FieldArray fields = {{"field1", "value1_" + std::to_string(i)}, {"field2", "value2_"}}; + Fields ffields(fields); + db_->PutFields(WriteOptions(), key.slice(), ffields); + // batch.Put(key.slice(), gen.Generate(value_size_)); + bytes += ffields.size() + key.slice().size(); + thread->stats.FinishedSingleOp(); + } + s = db_->Write(write_options_, &batch); + if (!s.ok()) { + std::fprintf(stderr, "put error: %s\n", s.ToString().c_str()); + std::exit(1); + } + } + thread->stats.AddBytes(bytes); + } + + void ReadSequential(ThreadState* thread) { + Iterator* iter = db_->NewIterator(ReadOptions()); + int i = 0; + int64_t bytes = 0; + for (iter->SeekToFirst(); i < reads_ && iter->Valid(); iter->Next()) { + bytes += iter->key().size() + iter->value().size(); + thread->stats.FinishedSingleOp(); + ++i; + } + delete iter; + thread->stats.AddBytes(bytes); + } + + void ReadReverse(ThreadState* thread) { + Iterator* iter = db_->NewIterator(ReadOptions()); + int i = 0; + int64_t bytes = 0; + for (iter->SeekToLast(); i < reads_ && iter->Valid(); iter->Prev()) { + bytes += iter->key().size() + iter->value().size(); + thread->stats.FinishedSingleOp(); + ++i; + } + delete iter; + thread->stats.AddBytes(bytes); + } + + void ReadRandom(ThreadState* thread) { + ReadOptions options; + std::string value; + int found = 0; + KeyBuffer key; + for (int i = 0; i < reads_; i++) { + const int k = thread->rand.Uniform(FLAGS_num); + key.Set(k); + if (db_->Get(options, key.slice(), &value).ok()) { + found++; + } + thread->stats.FinishedSingleOp(); + } + char msg[100]; + std::snprintf(msg, sizeof(msg), "(%d of %d found)", found, num_); + thread->stats.AddMessage(msg); + } + + void ReadMissing(ThreadState* thread) { + ReadOptions options; + std::string value; + KeyBuffer key; + for (int i = 0; i < reads_; i++) { + const int k = thread->rand.Uniform(FLAGS_num); + key.Set(k); + Slice s = Slice(key.slice().data(), key.slice().size() - 1); + db_->Get(options, s, &value); + thread->stats.FinishedSingleOp(); + } + } + + void ReadHot(ThreadState* thread) { + ReadOptions options; + std::string value; + const int range = (FLAGS_num + 99) / 100; + KeyBuffer key; + for (int i = 0; i < reads_; i++) { + const int k = thread->rand.Uniform(range); + key.Set(k); + db_->Get(options, key.slice(), &value); + thread->stats.FinishedSingleOp(); + } + } - void ReadHot(ThreadState* thread) { - ReadOptions options; - std::string value; - const int range = (FLAGS_num + 99) / 100; - KeyBuffer key; - for (int i = 0; i < reads_; i++) { - const int k = thread->rand.Uniform(range); - key.Set(k); - db_->Get(options, key.slice(), &value); - thread->stats.FinishedSingleOp(); - } - } + void WriteTargetSeq(ThreadState* thread) { WriteGiven(thread, true); } - void SeekRandom(ThreadState* thread) { - ReadOptions options; - int found = 0; - KeyBuffer key; - for (int i = 0; i < reads_; i++) { - Iterator* iter = db_->NewIterator(options); - const int k = thread->rand.Uniform(FLAGS_num); - key.Set(k); - iter->Seek(key.slice()); - if (iter->Valid() && iter->key() == key.slice()) found++; - delete iter; - thread->stats.FinishedSingleOp(); - } - char msg[100]; - snprintf(msg, sizeof(msg), "(%d of %d found)", found, num_); - thread->stats.AddMessage(msg); - } + void WriteTargetRandom(ThreadState* thread) { WriteGiven(thread, false); } - void SeekOrdered(ThreadState* thread) { - ReadOptions options; - Iterator* iter = db_->NewIterator(options); - int found = 0; - int k = 0; - KeyBuffer key; - for (int i = 0; i < reads_; i++) { - k = (k + (thread->rand.Uniform(100))) % FLAGS_num; - key.Set(k); - iter->Seek(key.slice()); - if (iter->Valid() && iter->key() == key.slice()) found++; - thread->stats.FinishedSingleOp(); - } - delete iter; - char msg[100]; - std::snprintf(msg, sizeof(msg), "(%d of %d found)", found, num_); - thread->stats.AddMessage(msg); - } + void WriteGiven(ThreadState* thread, bool seq) { + if (num_ != FLAGS_num) { + char msg[100]; + std::snprintf(msg, sizeof(msg), "(%d ops)", num_); + thread->stats.AddMessage(msg); + } - void DoDelete(ThreadState* thread, bool seq) { - RandomGenerator gen; - WriteBatch batch; - Status s; - KeyBuffer key; - for (int i = 0; i < num_; i += entries_per_batch_) { - batch.Clear(); - for (int j = 0; j < entries_per_batch_; j++) { - const int k = seq ? i + j : (thread->rand.Uniform(FLAGS_num)); - key.Set(k); - batch.Delete(key.slice()); - thread->stats.FinishedSingleOp(); - } - s = db_->Write(write_options_, &batch); - if (!s.ok()) { - std::fprintf(stderr, "del error: %s\n", s.ToString().c_str()); - std::exit(1); - } - } + RandomGenerator gen; + WriteBatch batch; + Status s; + int64_t bytes = 0; + KeyBuffer key; + for (int i = 0; i < num_; i += entries_per_batch_) { + batch.Clear(); + for (int j = 0; j < entries_per_batch_; j++) { + const int k = seq ? i + j : thread->rand.Uniform(FLAGS_num); + key.Set(k); + + FieldArray fields; + auto value = gen.Generate(value_size_); + if (i < num_fields) { + fields = { + {"field1", value.ToString()}, + {"field2", "value2_"}, + }; + } else { + fields = { + {"field1", value.ToString()}, + }; } - void DeleteSeq(ThreadState* thread) { DoDelete(thread, true); } - - void DeleteRandom(ThreadState* thread) { DoDelete(thread, false); } - - void ReadWhileWriting(ThreadState* thread) { - if (thread->tid > 0) { - ReadRandom(thread); - } else { - // Special thread that keeps writing until other threads are done. - RandomGenerator gen; - KeyBuffer key; - while (true) { - { - MutexLock l(&thread->shared->mu); - if (thread->shared->num_done + 1 >= thread->shared->num_initialized) { - // Other threads have finished - break; - } - } - - const int k = thread->rand.Uniform(FLAGS_num); - key.Set(k); - Status s = - db_->Put(write_options_, key.slice(), gen.Generate(value_size_)); - if (!s.ok()) { - std::fprintf(stderr, "put error: %s\n", s.ToString().c_str()); - std::exit(1); - } - } - - // Do not count any of the preceding work/delay in stats. - thread->stats.Start(); - } + Fields ffields(fields); + db_->PutFields(WriteOptions(), key.slice(), ffields); + bytes += ffields.size() + key.slice().size(); + + thread->stats.FinishedSingleOp(); + } + s = db_->Write(write_options_, &batch); + if (!s.ok()) { + std::fprintf(stderr, "put error: %s\n", s.ToString().c_str()); + std::exit(1); + } + } + thread->stats.AddBytes(bytes); + } + + void FindKeysByField(ThreadState* thread){ + int found = 0; + FieldArray fields_to_find = {{"field2", "value2_"}}; + + std::string dbname_ = "benchmark_db"; + Options options; + options.create_if_missing = true; + DBImpl* impl = new DBImpl(options, dbname_); + + std::vector found_keys = Fields::FindKeysByFields(db_, fields_to_find, impl); + found = found_keys.size(); + char msg[100]; + snprintf(msg, sizeof(msg), "(%d of %d found)", found, num_fields); + thread->stats.AddMessage(msg); + } + + void SeekRandom(ThreadState* thread) { + ReadOptions options; + int found = 0; + KeyBuffer key; + for (int i = 0; i < reads_; i++) { + Iterator* iter = db_->NewIterator(options); + const int k = thread->rand.Uniform(FLAGS_num); + key.Set(k); + iter->Seek(key.slice()); + if (iter->Valid() && iter->key() == key.slice()) found++; + delete iter; + thread->stats.FinishedSingleOp(); + } + char msg[100]; + snprintf(msg, sizeof(msg), "(%d of %d found)", found, num_); + thread->stats.AddMessage(msg); + } + + void SeekOrdered(ThreadState* thread) { + ReadOptions options; + Iterator* iter = db_->NewIterator(options); + int found = 0; + int k = 0; + KeyBuffer key; + for (int i = 0; i < reads_; i++) { + k = (k + (thread->rand.Uniform(100))) % FLAGS_num; + key.Set(k); + iter->Seek(key.slice()); + if (iter->Valid() && iter->key() == key.slice()) found++; + thread->stats.FinishedSingleOp(); + } + delete iter; + char msg[100]; + std::snprintf(msg, sizeof(msg), "(%d of %d found)", found, num_); + thread->stats.AddMessage(msg); + } + + void DoDelete(ThreadState* thread, bool seq) { + RandomGenerator gen; + WriteBatch batch; + Status s; + KeyBuffer key; + for (int i = 0; i < num_; i += entries_per_batch_) { + batch.Clear(); + for (int j = 0; j < entries_per_batch_; j++) { + const int k = seq ? i + j : (thread->rand.Uniform(FLAGS_num)); + key.Set(k); + batch.Delete(key.slice()); + thread->stats.FinishedSingleOp(); + } + s = db_->Write(write_options_, &batch); + if (!s.ok()) { + std::fprintf(stderr, "del error: %s\n", s.ToString().c_str()); + std::exit(1); + } + } + } + + void DeleteSeq(ThreadState* thread) { DoDelete(thread, true); } + + void DeleteRandom(ThreadState* thread) { DoDelete(thread, false); } + + void ReadWhileWriting(ThreadState* thread) { + if (thread->tid > 0) { + ReadRandom(thread); + } else { + // Special thread that keeps writing until other threads are done. + RandomGenerator gen; + KeyBuffer key; + while (true) { + { + MutexLock l(&thread->shared->mu); + if (thread->shared->num_done + 1 >= thread->shared->num_initialized) { + // Other threads have finished + break; + } } - void Compact(ThreadState* thread) { db_->CompactRange(nullptr, nullptr); } - - void PrintStats(const char* key) { - std::string stats; - if (!db_->GetProperty(key, &stats)) { - stats = "(failed)"; - } - std::fprintf(stdout, "\n%s\n", stats.c_str()); + const int k = thread->rand.Uniform(FLAGS_num); + key.Set(k); + Status s = + db_->Put(write_options_, key.slice(), gen.Generate(value_size_)); + if (!s.ok()) { + std::fprintf(stderr, "put error: %s\n", s.ToString().c_str()); + std::exit(1); } + } - static void WriteToFile(void* arg, const char* buf, int n) { - reinterpret_cast(arg)->Append(Slice(buf, n)); - } + // Do not count any of the preceding work/delay in stats. + thread->stats.Start(); + } + } - void HeapProfile() { - char fname[100]; - std::snprintf(fname, sizeof(fname), "%s/heap-%04d", FLAGS_db, - ++heap_counter_); - WritableFile* file; - Status s = g_env->NewWritableFile(fname, &file); - if (!s.ok()) { - std::fprintf(stderr, "%s\n", s.ToString().c_str()); - return; - } - bool ok = port::GetHeapProfile(WriteToFile, file); - delete file; - if (!ok) { - std::fprintf(stderr, "heap profiling not supported\n"); - g_env->RemoveFile(fname); - } - } - }; + void Compact(ThreadState* thread) { db_->CompactRange(nullptr, nullptr); } + + void PrintStats(const char* key) { + std::string stats; + if (!db_->GetProperty(key, &stats)) { + stats = "(failed)"; + } + std::fprintf(stdout, "\n%s\n", stats.c_str()); + } + + static void WriteToFile(void* arg, const char* buf, int n) { + reinterpret_cast(arg)->Append(Slice(buf, n)); + } + + void HeapProfile() { + char fname[100]; + std::snprintf(fname, sizeof(fname), "%s/heap-%04d", FLAGS_db, + ++heap_counter_); + WritableFile* file; + Status s = g_env->NewWritableFile(fname, &file); + if (!s.ok()) { + std::fprintf(stderr, "%s\n", s.ToString().c_str()); + return; + } + bool ok = port::GetHeapProfile(WriteToFile, file); + delete file; + if (!ok) { + std::fprintf(stderr, "heap profiling not supported\n"); + g_env->RemoveFile(fname); + } + } +}; } // namespace leveldb int main(int argc, char** argv) { - FLAGS_write_buffer_size = leveldb::Options().write_buffer_size; - FLAGS_max_file_size = leveldb::Options().max_file_size; - FLAGS_block_size = leveldb::Options().block_size; - FLAGS_open_files = leveldb::Options().max_open_files; - std::string default_db_path; - - for (int i = 1; i < argc; i++) { - double d; - int n; - char junk; - if (leveldb::Slice(argv[i]).starts_with("--benchmarks=")) { - FLAGS_benchmarks = argv[i] + strlen("--benchmarks="); - } else if (sscanf(argv[i], "--compression_ratio=%lf%c", &d, &junk) == 1) { - FLAGS_compression_ratio = d; - } else if (sscanf(argv[i], "--histogram=%d%c", &n, &junk) == 1 && - (n == 0 || n == 1)) { - FLAGS_histogram = n; - } else if (sscanf(argv[i], "--comparisons=%d%c", &n, &junk) == 1 && - (n == 0 || n == 1)) { - FLAGS_comparisons = n; - } else if (sscanf(argv[i], "--use_existing_db=%d%c", &n, &junk) == 1 && - (n == 0 || n == 1)) { - FLAGS_use_existing_db = n; - } else if (sscanf(argv[i], "--reuse_logs=%d%c", &n, &junk) == 1 && - (n == 0 || n == 1)) { - FLAGS_reuse_logs = n; - } else if (sscanf(argv[i], "--num=%d%c", &n, &junk) == 1) { - FLAGS_num = n; - } else if (sscanf(argv[i], "--reads=%d%c", &n, &junk) == 1) { - FLAGS_reads = n; - } else if (sscanf(argv[i], "--threads=%d%c", &n, &junk) == 1) { - FLAGS_threads = n; - } else if (sscanf(argv[i], "--value_size=%d%c", &n, &junk) == 1) { - FLAGS_value_size = n; - } else if (sscanf(argv[i], "--write_buffer_size=%d%c", &n, &junk) == 1) { - FLAGS_write_buffer_size = n; - } else if (sscanf(argv[i], "--max_file_size=%d%c", &n, &junk) == 1) { - FLAGS_max_file_size = n; - } else if (sscanf(argv[i], "--block_size=%d%c", &n, &junk) == 1) { - FLAGS_block_size = n; - } else if (sscanf(argv[i], "--key_prefix=%d%c", &n, &junk) == 1) { - FLAGS_key_prefix = n; - } else if (sscanf(argv[i], "--cache_size=%d%c", &n, &junk) == 1) { - FLAGS_cache_size = n; - } else if (sscanf(argv[i], "--bloom_bits=%d%c", &n, &junk) == 1) { - FLAGS_bloom_bits = n; - } else if (sscanf(argv[i], "--open_files=%d%c", &n, &junk) == 1) { - FLAGS_open_files = n; - } else if (strncmp(argv[i], "--db=", 5) == 0) { - FLAGS_db = argv[i] + 5; - } else { - std::fprintf(stderr, "Invalid flag '%s'\n", argv[i]); - std::exit(1); - } + FLAGS_write_buffer_size = leveldb::Options().write_buffer_size; + FLAGS_max_file_size = leveldb::Options().max_file_size; + FLAGS_block_size = leveldb::Options().block_size; + FLAGS_open_files = leveldb::Options().max_open_files; + std::string default_db_path; + + for (int i = 1; i < argc; i++) { + double d; + int n; + char junk; + if (leveldb::Slice(argv[i]).starts_with("--benchmarks=")) { + FLAGS_benchmarks = argv[i] + strlen("--benchmarks="); + } else if (sscanf(argv[i], "--compression_ratio=%lf%c", &d, &junk) == 1) { + FLAGS_compression_ratio = d; + } else if (sscanf(argv[i], "--histogram=%d%c", &n, &junk) == 1 && + (n == 0 || n == 1)) { + FLAGS_histogram = n; + } else if (sscanf(argv[i], "--comparisons=%d%c", &n, &junk) == 1 && + (n == 0 || n == 1)) { + FLAGS_comparisons = n; + } else if (sscanf(argv[i], "--use_existing_db=%d%c", &n, &junk) == 1 && + (n == 0 || n == 1)) { + FLAGS_use_existing_db = n; + } else if (sscanf(argv[i], "--reuse_logs=%d%c", &n, &junk) == 1 && + (n == 0 || n == 1)) { + FLAGS_reuse_logs = n; + } else if (sscanf(argv[i], "--compression=%d%c", &n, &junk) == 1 && + (n == 0 || n == 1)) { + FLAGS_compression = n; + } else if (sscanf(argv[i], "--num=%d%c", &n, &junk) == 1) { + FLAGS_num = n; + } else if (sscanf(argv[i], "--reads=%d%c", &n, &junk) == 1) { + FLAGS_reads = n; + } else if (sscanf(argv[i], "--num_fields=%d%c", &n, &junk) == 1) { + FLAGS_num_fields = n; + } else if (sscanf(argv[i], "--threads=%d%c", &n, &junk) == 1) { + FLAGS_threads = n; + } else if (sscanf(argv[i], "--value_size=%d%c", &n, &junk) == 1) { + FLAGS_value_size = n; + } else if (sscanf(argv[i], "--write_buffer_size=%d%c", &n, &junk) == 1) { + FLAGS_write_buffer_size = n; + } else if (sscanf(argv[i], "--max_file_size=%d%c", &n, &junk) == 1) { + FLAGS_max_file_size = n; + } else if (sscanf(argv[i], "--block_size=%d%c", &n, &junk) == 1) { + FLAGS_block_size = n; + } else if (sscanf(argv[i], "--key_prefix=%d%c", &n, &junk) == 1) { + FLAGS_key_prefix = n; + } else if (sscanf(argv[i], "--cache_size=%d%c", &n, &junk) == 1) { + FLAGS_cache_size = n; + } else if (sscanf(argv[i], "--bloom_bits=%d%c", &n, &junk) == 1) { + FLAGS_bloom_bits = n; + } else if (sscanf(argv[i], "--open_files=%d%c", &n, &junk) == 1) { + FLAGS_open_files = n; + } else if (strncmp(argv[i], "--db=", 5) == 0) { + FLAGS_db = argv[i] + 5; + } else { + std::fprintf(stderr, "Invalid flag '%s'\n", argv[i]); + std::exit(1); } + } - leveldb::g_env = leveldb::Env::Default(); + leveldb::g_env = leveldb::Env::Default(); - // Choose a location for the test database if none given with --db= - if (FLAGS_db == nullptr) { - leveldb::g_env->GetTestDirectory(&default_db_path); - default_db_path += "/dbbench-kv"; - FLAGS_db = default_db_path.c_str(); - } + // Choose a location for the test database if none given with --db= + if (FLAGS_db == nullptr) { + leveldb::g_env->GetTestDirectory(&default_db_path); + default_db_path += "/dbbench"; + FLAGS_db = default_db_path.c_str(); + } - leveldb::Benchmark benchmark; - benchmark.Run(); - return 0; + leveldb::Benchmark benchmark; + benchmark.Run(); + return 0; } diff --git a/db/db_impl.cc b/db/db_impl.cc index 27ba02f..f36cf25 100644 --- a/db/db_impl.cc +++ b/db/db_impl.cc @@ -410,7 +410,7 @@ Status DBImpl::Recover(VersionEdit* edit, bool* save_manifest) { // // update the file number allocation counter in VersionSet. // versions_->MarkFileNumberUsed(logs[i]); // } - //TODO begin + //注释：逐个恢复日志的内容 bool found_sequence_pos = false; for(int i = 0; i < logs.size(); ++i){ if( logs[i] < versions_->ImmLogFileNumber() ) { @@ -424,8 +424,7 @@ Status DBImpl::Recover(VersionEdit* edit, bool* save_manifest) { return s; } } - versions_->MarkFileNumberUsed(max_number); - //TODO end + versions_->MarkFileNumberUsed(max_number); if (versions_->LastSequence() < max_sequence) { versions_->SetLastSequence(max_sequence); @@ -483,9 +482,8 @@ Status DBImpl::RecoverLogFile(uint64_t log_number, bool last_log, uint64_t record_offset = 0; int compactions = 0; MemTable* mem = nullptr; - // TODO begin + //注释：设置 imm_last_sequence uint64_t imm_last_sequence = versions_->ImmLastSequence(); - // TODO end while (reader.ReadRecord(&record, &scratch) && status.ok()) { // if (record.size() < 12) { if (record.size() < 20) { @@ -541,10 +539,9 @@ Status DBImpl::RecoverLogFile(uint64_t log_number, bool last_log, compactions++; *save_manifest = true; - // TODO begin mem 落盘修改 imm_last_sequence，版本恢复 + // 注释：mem 落盘修改 imm_last_sequence，版本恢复 versions_->SetImmLastSequence(mem->GetTailSequence()); versions_->SetImmLogFileNumber(log_number); - // TODO end status = WriteLevel0Table(mem, edit, nullptr); mem->Unref(); @@ -587,10 +584,9 @@ Status DBImpl::RecoverLogFile(uint64_t log_number, bool last_log, // mem did not get reused; compact it. if (status.ok()) { - // TODO begin mem 落盘修改 imm_last_sequence，版本恢复 + //注释： mem 落盘修改 imm_last_sequence，版本恢复 versions_->SetImmLastSequence(mem->GetTailSequence()); versions_->SetImmLogFileNumber(log_number); - // TODO end *save_manifest = true; status = WriteLevel0Table(mem, edit, nullptr); } @@ -664,14 +660,12 @@ void DBImpl::CompactMemTable() { edit.SetPrevLogNumber(0); edit.SetLogNumber(logfile_number_); // Earlier logs no longer needed // s = versions_->LogAndApply(&edit, &mutex_); - // TODO begin - //构建新版本，并将其加入到 version_当中 + //注释： 构建新版本，并将其加入到 version_当中 versions_->StartImmLastSequence(true); versions_->SetImmLastSequence(imm_->GetTailSequence()); versions_->SetImmLogFileNumber(imm_->GetLogFileNumber()); s = versions_->LogAndApply(&edit, &mutex_); versions_->StartImmLastSequence(false); - // TODO end } if (s.ok()) { diff --git a/db/version_edit.cc b/db/version_edit.cc index 84e7649..56dce29 100644 --- a/db/version_edit.cc +++ b/db/version_edit.cc @@ -21,11 +21,10 @@ enum Tag { kNewFile = 7, // 8 was used for large value refs kPrevLogNumber = 9, - // TODO begin 在版本中记录 immemtable 转到 sst的时候的 sequence，主要用来恢复的时候 定位db关闭的时候 - // imm 和 mem中的内容在恢复的时候应该从log文件中哪里开始恢复。 + // 注释： 用于记录immemtable到sst的sequence kImmLastSequence = 10, + // 注释： 用于记录恢复immemtable和memtable时在log文件中对应的位置 kLogFile = 11 - // TODO end }; void VersionEdit::Clear() { @@ -35,20 +34,20 @@ void VersionEdit::Clear() { last_sequence_ = 0; next_file_number_ = 0; - // TODO begin + // 注释：重置为0 imm_last_sequence_ = 0; + // 注释：重置为0 imm_log_file_number_ = 0; - // TODO end + has_comparator_ = false; has_log_number_ = false; has_prev_log_number_ = false; - has_next_file_number_ = false; + has_next_file_number_ = false; has_last_sequence_ = false; - // TODO begin + // 注释：重置为false has_imm_last_sequence_ = false; - // TODO end // compact_pointers_.clear(); deleted_files_.clear(); @@ -76,6 +75,7 @@ void VersionEdit::EncodeTo(std::string* dst) const { PutVarint32(dst, kLastSequence); PutVarint64(dst, last_sequence_); } + // 注释：若 imm_last_sequence_ 有效，则写入对应的标识符和数据 if (has_imm_last_sequence_) { PutVarint32(dst, kImmLastSequence); PutVarint64(dst, imm_last_sequence_); @@ -180,7 +180,7 @@ Status VersionEdit::DecodeFrom(const Slice& src) { } break; - // TODO begin + // 注释：如果是kImmLastSequence类型，则解析imm_last_sequence_和imm_log_file_number_并将has_imm_last_sequence_ 设为true case kImmLastSequence: if (GetVarint64(&input, &imm_last_sequence_) && GetVarint64(&input, &imm_log_file_number_)) { has_imm_last_sequence_ = true; @@ -188,7 +188,7 @@ Status VersionEdit::DecodeFrom(const Slice& src) { msg = "imemtable last sequence number"; } break; - // TODO end + case kCompactPointer: if (GetLevel(&input, &level) && GetInternalKey(&input, &key)) { diff --git a/db/version_edit.h b/db/version_edit.h index 6467a2f..0b36316 100644 --- a/db/version_edit.h +++ b/db/version_edit.h @@ -26,19 +26,17 @@ struct FileMetaData { InternalKey largest; // Largest internal key served by table }; -// TODO begin +// 注释：vlog文件的元数据 struct LogMetaData { LogMetaData() : refs(0), allowed_seeks(1 << 30), file_size(0) {} int refs; // 引用计数 - // Seeks allowed until compaction; 当该值为0时,意味着需要进行compaction操作了; 变量allowed_seeks的值在sstable文件加入到version时确定 - int allowed_seeks; + int allowed_seeks; //当该值为0时,意味着需要进行compaction操作了; 变量allowed_seeks的值在sstable文件加入到version时确定 uint64_t number; //文件名相关;sstable文件的名字是 number.ldb - uint64_t file_size; // File size in bytes 文件大小 - InternalKey smallest; // Smallest internal key served by table 最小的key - InternalKey largest; // Largest internal key served by table 最大的key + uint64_t file_size; //文件大小 + InternalKey smallest; //最小的key + InternalKey largest; //最大的key }; -// TODO end class VersionEdit { public: @@ -68,14 +66,13 @@ class VersionEdit { last_sequence_ = seq; } - // TODO begin - //设置序列号 imm_last_sequence_ imm 转 sst的时候用 + //注释： 设置序列号 imm_last_sequence_（imm 转 sst的时候用） void SetImmLastSequence(SequenceNumber seq,uint64_t fid) { has_imm_last_sequence_ = true; imm_last_sequence_ = seq; imm_log_file_number_ = fid; } - // TODO end + void SetCompactPointer(int level, const InternalKey& key) { compact_pointers_.push_back(std::make_pair(level, key)); } @@ -119,14 +116,12 @@ class VersionEdit { bool has_next_file_number_; bool has_last_sequence_; - // TODO begin - // 是否包含 imm_last_sequence_ + //注释：是否包含 imm_last_sequence_ bool has_imm_last_sequence_; - // 恢复log的时候 用来定位memtable 和 immemtabl中的位置 + //注释：恢复log的时候 用来定位memtable 和 immemtabl中的位置 SequenceNumber imm_last_sequence_; - // imm_last_sequence 所处在的log文件 + //注释：imm_last_sequence 所处在的log文件 uint64_t imm_log_file_number_; - // TODO end std::vector> compact_pointers_; DeletedFileSet deleted_files_; diff --git a/db/version_set.cc b/db/version_set.cc index 6c7017e..3d8865b 100644 --- a/db/version_set.cc +++ b/db/version_set.cc @@ -252,16 +252,15 @@ enum SaverState { kDeleted, kCorrupt, }; -// TODO begin +// 注释：Saver的kv对是否分离 enum SaverSeparate { kNotSeparated, kSeparated }; -// TODO end + struct Saver { - // TODO begin + // 注释：初始设为不分离 SaverSeparate separate = kNotSeparated; - // TODO end SaverState state; const Comparator* ucmp; Slice user_key; @@ -279,9 +278,8 @@ static void SaveValue(void* arg, const Slice& ikey, const Slice& v) { s->state = (parsed_key.type == kTypeValue || parsed_key.type == kTypeSeparation) ? kFound : kDeleted; if (s->state == kFound) { s->value->assign(v.data(), v.size()); - // TODO begin + // 注释：如果key.type是kTypeSeparation,则设为kSeparated类型 s->separate = ( parsed_key.type == kTypeSeparation ) ? kSeparated : kNotSeparated; - // TODO end } } } @@ -367,13 +365,12 @@ Status Version::Get(const ReadOptions& options, const LookupKey& k, state->s = state->vset->table_cache_->Get(*state->options, f->number, f->file_size, state->ikey, &state->saver, SaveValue); - // TODO begin + // 注释：对于是否kv分离，调用不同的Set函数 if( state->saver.separate == kSeparated ){ state->s.SetSeparated(); } else{ state->s.SetNotSeparated(); } - // TODO end if (!state->s.ok()) { state->found = true; return false; @@ -761,6 +758,10 @@ VersionSet::VersionSet(const std::string& dbname, const Options* options, next_file_number_(2), manifest_file_number_(0), // Filled by Recover() last_sequence_(0), + //注释：加上version_edit中添加的参数 + imm_last_sequence_(0), + imm_log_file_number_(0), + save_imm_last_sequence_(false), log_number_(0), prev_log_number_(0), descriptor_file_(nullptr), @@ -809,11 +810,10 @@ Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu) { edit->SetNextFile(next_file_number_); edit->SetLastSequence(last_sequence_); - // TODO begin + // 注释：设置imm_last_sequence_和imm_log_file_number_ if( SaveImmLastSequence() ){ edit->SetImmLastSequence(imm_last_sequence_,imm_log_file_number_); } - // TODO end Version* v = new Version(this); { @@ -919,11 +919,11 @@ Status VersionSet::Recover(bool* save_manifest) { bool have_next_file = false; bool have_last_sequence = false; - // TODO begin + //注释：重置version_edit里添加的参数 bool have_imm_last_sequence = false; uint64_t imm_last_sequence = 0; uint64_t imm_log_file_number = 0; - // TODO end + uint64_t next_file = 0; uint64_t last_sequence = 0; @@ -975,13 +975,12 @@ Status VersionSet::Recover(bool* save_manifest) { last_sequence = edit.last_sequence_; have_last_sequence = true; } - // TODO begin + //注释： 构建当前的Version 回放参数 if (edit.has_imm_last_sequence_) { imm_last_sequence = edit.imm_last_sequence_; imm_log_file_number = edit.imm_log_file_number_; have_imm_last_sequence = true; } - // TODO end } } delete file; @@ -1015,10 +1014,9 @@ Status VersionSet::Recover(bool* save_manifest) { last_sequence_ = last_sequence; log_number_ = log_number; prev_log_number_ = prev_log_number; - // TODO begin + //注释：修改imm_last_sequence_和imm_log_file_number_ imm_last_sequence_ = imm_last_sequence; imm_log_file_number_ = imm_log_file_number; - // TODO end // See if we can reuse the existing MANIFEST file. if (ReuseManifest(dscname, current)) { diff --git a/db/version_set.h b/db/version_set.h index e3da5e7..b7787eb 100644 --- a/db/version_set.h +++ b/db/version_set.h @@ -269,14 +269,14 @@ class VersionSet { }; const char* LevelSummary(LevelSummaryStorage* scratch) const; - // TODO begin + //注释：用于版本恢复时设置相关参数 bool SaveImmLastSequence(){ return save_imm_last_sequence_; } bool StartImmLastSequence(bool save ){ save_imm_last_sequence_ = save; } void SetImmLastSequence( uint64_t seq ){ imm_last_sequence_ = seq; } uint64_t ImmLastSequence() const { return imm_last_sequence_; } uint64_t ImmLogFileNumber() const { return imm_log_file_number_; } void SetImmLogFileNumber( uint64_t fid ){ imm_log_file_number_ = fid; } - // TODO end + private: class Builder; @@ -313,13 +313,13 @@ class VersionSet { uint64_t log_number_; uint64_t prev_log_number_; // 0 or backing store for memtable being compacted - // TODO begin + //注释： immemtable转sst的sequence uint64_t imm_last_sequence_; - // 是否保存 imm 转 sst时候的sequence，主要用在 LogAndApply 这个函数当中，用于区分是mior compact 还是 major compact的过程。 + //注释： 是否保存 imm 转 sst时候的sequence，主要用在 LogAndApply 这个函数当中，用于区分是mior compact 还是 major compact的过程。 bool save_imm_last_sequence_; - // imm_last_sequence 所处在的log文件 + //注释：imm_last_sequence 所处在的log文件 uint64_t imm_log_file_number_; - // TODO end + // Opened lazily WritableFile* descriptor_file_; diff --git a/db/write_batch.cc b/db/write_batch.cc index 9907a52..dc2e3ef 100644 --- a/db/write_batch.cc +++ b/db/write_batch.cc @@ -34,6 +34,7 @@ WriteBatch::Handler::~Handler() = default; void WriteBatch::Clear() { belong_to_gc = false; + // belong_to_gc = true; rep_.clear(); rep_.resize(kHeader); } diff --git a/test/bench_test.cc b/test/bench_test.cc index ef50706..a4c367e 100644 --- a/test/bench_test.cc +++ b/test/bench_test.cc @@ -10,11 +10,11 @@ using namespace leveldb; // Number of key/values to operate in database -constexpr int num_ = 100000; +constexpr int num_ = 500000; // Size of each value -constexpr int value_size_ = 1000; +constexpr int value_size_ = 1024; // Number of read operations -constexpr int reads_ = 100000; +constexpr int reads_ = 500000; Status OpenDB(std::string dbName, DB **db) { Options options; @@ -85,7 +85,7 @@ void FindKeys(DB *db, std::vector &lats) { FieldArray fields_to_find = {{"field", "old_value_" }}; auto start_time = std::chrono::steady_clock::now(); - std::string dbname_ = "benchmark_db"; + std::string dbname_ = "bench_resr_db"; Options options; options.create_if_missing = true; DBImpl* impl = new DBImpl(options, dbname_); @@ -153,11 +153,11 @@ void RunBenchmark(const char* name, Func func, bool setup_data = true, bool setu delete db; } -// TEST(BenchTest, PutLatency) { RunBenchmark("Put", InsertData, false, false); } -// TEST(BenchTest, PutFieldsLatency) { RunBenchmark("PutFields", InsertFields, false, false); } +TEST(BenchTest, PutLatency) { RunBenchmark("Put", InsertData, false, false); } +TEST(BenchTest, PutFieldsLatency) { RunBenchmark("PutFields", InsertFields, false, false); } -// TEST(BenchTest, GetLatency) { RunBenchmark("Get", GetData, true, false); } -// TEST(BenchTest, IteratorLatency) { RunBenchmark("Iterator", ReadOrdered, true, false); } +TEST(BenchTest, GetLatency) { RunBenchmark("Get", GetData, true, false); } +TEST(BenchTest, IteratorLatency) { RunBenchmark("Iterator", ReadOrdered, true, false); } TEST(BenchTest, FindKeysByFieldLatency) { RunBenchmark("FindKeysByFields", FindKeys, false, true); diff --git a/test/value_field_test.cc b/test/value_field_test.cc index 0467ab2..55f5d04 100644 --- a/test/value_field_test.cc +++ b/test/value_field_test.cc @@ -28,7 +28,7 @@ class FieldsTest : public ::testing::Test { } DB* db_ = nullptr; // 数据库实例指针。 - std::string dbname_ = "testdb"; // 记录数据库路径 + std::string dbname_ = "testdb_field"; // 记录数据库路径 }; // 测试各种构造函数