Fix false negative in TieredSecondaryCache nvm cache lookup (#12134)

Summary:
There is a bug in the `TieredSecondaryCache` that can result in a false negative. This can happen when a MultiGet does a cache lookup that gets a hit in the `TieredSecondaryCache` local nvm cache tier, and the result is available before MultiGet calls `WaitAll` (i.e the nvm cache `SecondaryCacheResultHandle` `IsReady` returns true).

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

Test Plan: Add a new unit test in tiered_secondary_cache_test

Reviewed By: akankshamahajan15

Differential Revision: D52023309

Pulled By: anand1976

fbshipit-source-id: e5ae681226a0f12753fecb2f6acc7e5f254ae72b

cache/tiered_secondary_cache.cc

@@ -109,10 +109,8 @@ void TieredSecondaryCache::WaitAll(
   }
   nvm_sec_cache_->WaitAll(nvm_handles);
   for (auto handle : my_handles) {
-    assert(handle->IsReady());
-    auto nvm_handle = handle->inner_handle();
-    handle->SetSize(nvm_handle->Size());
-    handle->SetValue(nvm_handle->Value());
+    assert(handle->inner_handle()->IsReady());
+    handle->Complete();
   }
 }
 

cache/tiered_secondary_cache.h

@@ -79,7 +79,10 @@ class TieredSecondaryCache : public SecondaryCacheWrapper {
     ~ResultHandle() override {}
 
     bool IsReady() override {
-      return !inner_handle_ || inner_handle_->IsReady();
+      if (inner_handle_ && inner_handle_->IsReady()) {
+        Complete();
+      }
+      return ready_;
     }
 
     void Wait() override {
@@ -92,10 +95,10 @@ class TieredSecondaryCache : public SecondaryCacheWrapper {
     Cache::ObjectPtr Value() override { return value_; }
 
     void Complete() {
-      assert(IsReady());
       size_ = inner_handle_->Size();
       value_ = inner_handle_->Value();
       inner_handle_.reset();
+      ready_ = true;
     }
 
     void SetInnerHandle(std::unique_ptr<SecondaryCacheResultHandle>&& handle) {
@@ -115,6 +118,7 @@ class TieredSecondaryCache : public SecondaryCacheWrapper {
     CreateContext ctx_;
     size_t size_;
     Cache::ObjectPtr value_;
+    bool ready_ = false;
   };
 
   static void NoopDelete(Cache::ObjectPtr /*obj*/,

cache/tiered_secondary_cache_test.cc

@@ -15,10 +15,11 @@ namespace ROCKSDB_NAMESPACE {
 
 class TestSecondaryCache : public SecondaryCache {
  public:
-  explicit TestSecondaryCache(size_t capacity)
+  explicit TestSecondaryCache(size_t capacity, bool ready_before_wait)
       : cache_(NewLRUCache(capacity, 0, false, 0.5 /* high_pri_pool_ratio */,
                            nullptr, kDefaultToAdaptiveMutex,
                            kDontChargeCacheMetadata)),
+        ready_before_wait_(ready_before_wait),
         num_insert_saved_(0),
         num_hits_(0),
         num_misses_(0) {}
@@ -88,7 +89,8 @@ class TestSecondaryCache : public SecondaryCache {
                             /*alloc*/ nullptr, &value, &charge);
       if (s.ok()) {
         secondary_handle.reset(new TestSecondaryCacheResultHandle(
-            cache_.get(), handle, value, charge, /*ready=*/wait));
+            cache_.get(), handle, value, charge,
+            /*ready=*/wait || ready_before_wait_));
         kept_in_sec_cache = true;
       } else {
         cache_.Release(handle);
@@ -168,6 +170,7 @@ class TestSecondaryCache : public SecondaryCache {
       BasicTypedSharedCacheInterface<char[], CacheEntryRole::kMisc>;
   using TypedHandle = SharedCache::TypedHandle;
   SharedCache cache_;
+  bool ready_before_wait_;
   uint32_t num_insert_saved_;
   uint32_t num_hits_;
   uint32_t num_misses_;
@@ -179,11 +182,10 @@ class DBTieredSecondaryCacheTest : public DBTestBase {
   DBTieredSecondaryCacheTest()
       : DBTestBase("db_tiered_secondary_cache_test", /*env_do_fsync=*/true) {}
 
-  std::shared_ptr<Cache> NewCache(size_t pri_capacity,
-                                  size_t compressed_capacity,
-                                  size_t nvm_capacity,
-                                  TieredAdmissionPolicy adm_policy =
-                                      TieredAdmissionPolicy::kAdmPolicyAuto) {
+  std::shared_ptr<Cache> NewCache(
+      size_t pri_capacity, size_t compressed_capacity, size_t nvm_capacity,
+      TieredAdmissionPolicy adm_policy = TieredAdmissionPolicy::kAdmPolicyAuto,
+      bool ready_before_wait = false) {
     LRUCacheOptions lru_opts;
     TieredCacheOptions opts;
     lru_opts.capacity = 0;
@@ -194,10 +196,11 @@ class DBTieredSecondaryCacheTest : public DBTestBase {
     opts.comp_cache_opts.capacity = 0;
     opts.comp_cache_opts.num_shard_bits = 0;
     opts.total_capacity = pri_capacity + compressed_capacity;
-    opts.compressed_secondary_ratio =
+    opts.compressed_secondary_ratio = compressed_secondary_ratio_ =
         (double)compressed_capacity / opts.total_capacity;
     if (nvm_capacity > 0) {
-      nvm_sec_cache_.reset(new TestSecondaryCache(nvm_capacity));
+      nvm_sec_cache_.reset(
+          new TestSecondaryCache(nvm_capacity, ready_before_wait));
       opts.nvm_sec_cache = nvm_sec_cache_;
     }
     opts.adm_policy = adm_policy;
@@ -207,6 +210,12 @@ class DBTieredSecondaryCacheTest : public DBTestBase {
     return cache_;
   }
 
+  void ClearPrimaryCache() {
+    ASSERT_EQ(UpdateTieredCache(cache_, -1, 1.0), Status::OK());
+    ASSERT_EQ(UpdateTieredCache(cache_, -1, compressed_secondary_ratio_),
+              Status::OK());
+  }
+
   TestSecondaryCache* nvm_sec_cache() { return nvm_sec_cache_.get(); }
 
   CompressedSecondaryCache* compressed_secondary_cache() {
@@ -218,6 +227,7 @@ class DBTieredSecondaryCacheTest : public DBTestBase {
  private:
   std::shared_ptr<Cache> cache_;
   std::shared_ptr<TestSecondaryCache> nvm_sec_cache_;
+  double compressed_secondary_ratio_;
 };
 
 // In this test, the block size is set to 4096. Each value is 1007 bytes, so
@@ -582,6 +592,116 @@ TEST_F(DBTieredSecondaryCacheTest, WaitAllTest) {
   Destroy(options);
 }
 
+TEST_F(DBTieredSecondaryCacheTest, ReadyBeforeWaitAllTest) {
+  if (!LZ4_Supported()) {
+    ROCKSDB_GTEST_SKIP("This test requires LZ4 support.");
+    return;
+  }
+
+  BlockBasedTableOptions table_options;
+  table_options.block_cache = NewCache(250 * 1024, 20 * 1024, 256 * 1024,
+                                       TieredAdmissionPolicy::kAdmPolicyAuto,
+                                       /*ready_before_wait=*/true);
+  table_options.block_size = 4 * 1024;
+  table_options.cache_index_and_filter_blocks = false;
+  Options options = GetDefaultOptions();
+  options.create_if_missing = true;
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+  options.statistics = CreateDBStatistics();
+
+  options.paranoid_file_checks = false;
+  DestroyAndReopen(options);
+  Random rnd(301);
+  const int N = 256;
+  for (int i = 0; i < N; i++) {
+    std::string p_v;
+    test::CompressibleString(&rnd, 0.5, 1007, &p_v);
+    ASSERT_OK(Put(Key(i), p_v));
+  }
+
+  ASSERT_OK(Flush());
+
+  std::vector<std::string> keys;
+  std::vector<std::string> values;
+
+  keys.push_back(Key(0));
+  keys.push_back(Key(4));
+  keys.push_back(Key(8));
+  values = MultiGet(keys, /*snapshot=*/nullptr, /*async=*/true);
+  ASSERT_EQ(values.size(), keys.size());
+  for (auto value : values) {
+    ASSERT_EQ(1007, value.size());
+  }
+  ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 3u);
+  ASSERT_EQ(nvm_sec_cache()->num_misses(), 3u);
+  ASSERT_EQ(nvm_sec_cache()->num_hits(), 0u);
+  ASSERT_EQ(options.statistics->getTickerCount(BLOCK_CACHE_MISS), 3u);
+
+  keys.clear();
+  values.clear();
+  keys.push_back(Key(12));
+  keys.push_back(Key(16));
+  keys.push_back(Key(20));
+  values = MultiGet(keys, /*snapshot=*/nullptr, /*async=*/true);
+  ASSERT_EQ(values.size(), keys.size());
+  for (auto value : values) {
+    ASSERT_EQ(1007, value.size());
+  }
+  ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 6u);
+  ASSERT_EQ(nvm_sec_cache()->num_misses(), 6u);
+  ASSERT_EQ(nvm_sec_cache()->num_hits(), 0u);
+  ASSERT_EQ(options.statistics->getTickerCount(BLOCK_CACHE_MISS), 6u);
+
+  keys.clear();
+  values.clear();
+  keys.push_back(Key(0));
+  keys.push_back(Key(4));
+  keys.push_back(Key(8));
+  values = MultiGet(keys, /*snapshot=*/nullptr, /*async=*/true);
+  ASSERT_EQ(values.size(), keys.size());
+  for (auto value : values) {
+    ASSERT_EQ(1007, value.size());
+  }
+  ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 6u);
+  ASSERT_EQ(nvm_sec_cache()->num_misses(), 6u);
+  ASSERT_EQ(nvm_sec_cache()->num_hits(), 3u);
+  ASSERT_EQ(options.statistics->getTickerCount(BLOCK_CACHE_MISS), 6u);
+
+  ClearPrimaryCache();
+
+  keys.clear();
+  values.clear();
+  keys.push_back(Key(0));
+  keys.push_back(Key(32));
+  keys.push_back(Key(36));
+  values = MultiGet(keys, /*snapshot=*/nullptr, /*async=*/true);
+  ASSERT_EQ(values.size(), keys.size());
+  for (auto value : values) {
+    ASSERT_EQ(1007, value.size());
+  }
+  ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 8u);
+  ASSERT_EQ(nvm_sec_cache()->num_misses(), 8u);
+  ASSERT_EQ(nvm_sec_cache()->num_hits(), 4u);
+  ASSERT_EQ(options.statistics->getTickerCount(BLOCK_CACHE_MISS), 8u);
+
+  keys.clear();
+  values.clear();
+  keys.push_back(Key(0));
+  keys.push_back(Key(32));
+  keys.push_back(Key(36));
+  values = MultiGet(keys, /*snapshot=*/nullptr, /*async=*/true);
+  ASSERT_EQ(values.size(), keys.size());
+  for (auto value : values) {
+    ASSERT_EQ(1007, value.size());
+  }
+  ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 8u);
+  ASSERT_EQ(nvm_sec_cache()->num_misses(), 8u);
+  ASSERT_EQ(nvm_sec_cache()->num_hits(), 4u);
+  ASSERT_EQ(options.statistics->getTickerCount(BLOCK_CACHE_MISS), 8u);
+
+  Destroy(options);
+}
+
 // This test is for iteration. It iterates through a set of keys in two
 // passes. First pass loads the compressed blocks into the nvm tier, and
 // the second pass should hit all of those blocks.

unreleased_history/bug_fixes/tiered_cache_low_latency_false_negative.md

@@ -0,0 +1 @@
+A lookup by MultiGet in a TieredCache that goes to the local flash cache and finishes with very low latency, i.e before the subsequent call to WaitAll, is ignored, resulting in a false negative and a memory leak.