From 12a8be1d440de8ef390e045e4eacd4a8b3edd46b Mon Sep 17 00:00:00 2001 From: Matthew Von-Maszewski Date: Thu, 21 Jan 2021 13:10:37 -0800 Subject: [PATCH] =?UTF-8?q?MergeHelper::FilterMerge()=20calling=20ElapsedN?= =?UTF-8?q?anosSafe()=20upon=20exit=20even=20=E2=80=A6=20(#7867)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: …when unused. Causes many calls to clock_gettime, impacting performance. Was looking for something else via Linux "perf" command when I spotted heavy usage of clock_gettime during a compaction. Our product heavily uses the rocksdb::Options::merge_operator. MergeHelper::FilterMerge() properly tests if timing is enabled/disabled upon entry, but not on exit. This patch fixes the exit. Note: the entry test also verifies if "nullptr!=stats_". This test is redundant to code within ShouldReportDetailedTime(). Therefore I omitted it in my change. merge_test.cc updated with test that shows failure before merge_helper.cc change ... and fix after change. Pull Request resolved: https://github.com/facebook/rocksdb/pull/7867 Reviewed By: jay-zhuang Differential Revision: D25960175 Pulled By: ajkr fbshipit-source-id: 56e66d7eb6ae5eae89c8e0d5a262bd2905a226b6 --- db/merge_helper.cc | 5 +++-- db/merge_test.cc | 52 +++++++++++++++++++++++++++++----------------- 2 files changed, 36 insertions(+), 21 deletions(-) diff --git a/db/merge_helper.cc b/db/merge_helper.cc index ed2646ea1..8497d23b0 100644 --- a/db/merge_helper.cc +++ b/db/merge_helper.cc @@ -183,7 +183,6 @@ Status MergeHelper::MergeUntil(InternalIterator* iter, assert(IsValueType(ikey.type)); if (ikey.type != kTypeMerge) { - // hit a put/delete/single delete // => merge the put value or a nullptr with operands_ // => store result in operands_.back() (and update keys_.back()) @@ -414,7 +413,9 @@ CompactionFilter::Decision MergeHelper::FilterMerge(const Slice& user_key, kValueTypeForSeek); } } - total_filter_time_ += filter_timer_.ElapsedNanosSafe(); + if (stats_ != nullptr && ShouldReportDetailedTime(env_, stats_)) { + total_filter_time_ += filter_timer_.ElapsedNanosSafe(); + } return ret; } diff --git a/db/merge_test.cc b/db/merge_test.cc index de3b0dba3..2cca0735e 100644 --- a/db/merge_test.cc +++ b/db/merge_test.cc @@ -49,12 +49,8 @@ class CountMergeOperator : public AssociativeMergeOperator { return true; } - return mergeOperator_->PartialMerge( - key, - *existing_value, - value, - new_value, - logger); + return mergeOperator_->PartialMerge(key, *existing_value, value, new_value, + logger); } bool PartialMergeMulti(const Slice& key, @@ -73,6 +69,29 @@ class CountMergeOperator : public AssociativeMergeOperator { std::shared_ptr mergeOperator_; }; +class EnvMergeTest : public EnvWrapper { + public: + EnvMergeTest() : EnvWrapper(Env::Default()) {} + // ~EnvMergeTest() override {} + + uint64_t NowNanos() override { + ++now_nanos_count_; + return target()->NowNanos(); + } + + static uint64_t now_nanos_count_; + + static std::unique_ptr singleton_; + + static EnvMergeTest* GetInstance() { + if (nullptr == singleton_) singleton_.reset(new EnvMergeTest); + return singleton_.get(); + } +}; + +uint64_t EnvMergeTest::now_nanos_count_{0}; +std::unique_ptr EnvMergeTest::singleton_; + std::shared_ptr OpenDb(const std::string& dbname, const bool ttl = false, const size_t max_successive_merges = 0) { DB* db; @@ -80,6 +99,7 @@ std::shared_ptr OpenDb(const std::string& dbname, const bool ttl = false, options.create_if_missing = true; options.merge_operator = std::make_shared(); options.max_successive_merges = max_successive_merges; + options.env = EnvMergeTest::GetInstance(); EXPECT_OK(DestroyDB(dbname, Options())); Status s; // DBWithTTL is not supported in ROCKSDB_LITE @@ -106,7 +126,6 @@ std::shared_ptr OpenDb(const std::string& dbname, const bool ttl = false, // set, add, get and remove // This is a quick implementation without a Merge operation. class Counters { - protected: std::shared_ptr db_; @@ -190,7 +209,6 @@ class Counters { return get(key, &base) && set(key, base + value); } - // convenience functions for testing void assert_set(const std::string& key, uint64_t value) { assert(set(key, value)); @@ -202,27 +220,25 @@ class Counters { uint64_t value = default_; int result = get(key, &value); assert(result); - if (result == 0) exit(1); // Disable unused variable warning. + if (result == 0) exit(1); // Disable unused variable warning. return value; } void assert_add(const std::string& key, uint64_t value) { int result = add(key, value); assert(result); - if (result == 0) exit(1); // Disable unused variable warning. + if (result == 0) exit(1); // Disable unused variable warning. } }; // Implement 'add' directly with the new Merge operation class MergeBasedCounters : public Counters { private: - WriteOptions merge_option_; // for merge + WriteOptions merge_option_; // for merge public: explicit MergeBasedCounters(std::shared_ptr db, uint64_t defaultCount = 0) - : Counters(db, defaultCount), - merge_option_() { - } + : Counters(db, defaultCount), merge_option_() {} // mapped to a rocksdb Merge operation bool add(const std::string& key, uint64_t value) override { @@ -243,14 +259,13 @@ class MergeBasedCounters : public Counters { void dumpDb(DB* db) { auto it = std::unique_ptr(db->NewIterator(ReadOptions())); for (it->SeekToFirst(); it->Valid(); it->Next()) { - //uint64_t value = DecodeFixed64(it->value().data()); - //std::cout << it->key().ToString() << ": " << value << std::endl; + // uint64_t value = DecodeFixed64(it->value().data()); + // std::cout << it->key().ToString() << ": " << value << std::endl; } assert(it->status().ok()); // Check for any errors found during the scan } void testCounters(Counters& counters, DB* db, bool test_compaction) { - FlushOptions o; o.wait = true; @@ -392,7 +407,6 @@ void testCountersWithFlushAndCompaction(Counters& counters, DB* db) { void testSuccessiveMerge(Counters& counters, size_t max_num_merges, size_t num_merges) { - counters.assert_remove("z"); uint64_t sum = 0; @@ -449,6 +463,7 @@ void testPartialMerge(Counters* counters, DB* db, size_t max_merge, ASSERT_OK(db->CompactRange(CompactRangeOptions(), nullptr, nullptr)); ASSERT_EQ(tmp_sum, counters->assert_get("c")); ASSERT_EQ(num_partial_merge_calls, 0U); + ASSERT_EQ(EnvMergeTest::now_nanos_count_, 0U); } void testSingleBatchSuccessiveMerge(DB* db, size_t max_num_merges, @@ -486,7 +501,6 @@ void testSingleBatchSuccessiveMerge(DB* db, size_t max_num_merges, } void runTest(const std::string& dbname, const bool use_ttl = false) { - { auto db = OpenDb(dbname, use_ttl);