Use function objects as deleters in the block cache (#6545)

Summary:
As the first step of reintroducing eviction statistics for the block
cache, the patch switches from using simple function pointers as deleters
to function objects implementing an interface. This will enable using
deleters that have state, like a smart pointer to the statistics object
that is to be updated when an entry is removed from the cache. For now,
the patch adds a deleter template class `SimpleDeleter`, which simply
casts the `value` pointer to its original type and calls `delete` or
`delete[]` on it as appropriate. Note: to prevent object lifecycle
issues, deleters must outlive the cache entries referring to them;
`SimpleDeleter` ensures this by using the ("leaky") Meyers singleton
pattern.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/6545

Test Plan: `make asan_check`

Reviewed By: siying

Differential Revision: D20475823

Pulled By: ltamasi

fbshipit-source-id: fe354c33dd96d9bafc094605462352305449a22a

cache/cache_bench.cc

@@ -15,6 +15,7 @@ int main() {
 #include <sys/types.h>
 #include <cinttypes>
 
+#include "cache/simple_deleter.h"
 #include "port/port.h"
 #include "rocksdb/cache.h"
 #include "rocksdb/db.h"
@@ -49,9 +50,6 @@ namespace ROCKSDB_NAMESPACE {
 
 class CacheBench;
 namespace {
-void deleter(const Slice& /*key*/, void* value) {
-    delete reinterpret_cast<char *>(value);
-}
 
 // State shared by all concurrent executions of the same benchmark.
 class SharedState {
@@ -149,7 +147,8 @@ class CacheBench {
       // Cast uint64* to be char*, data would be copied to cache
       Slice key(reinterpret_cast<char*>(&rand_key), 8);
       // do insert
-      cache_->Insert(key, new char[10], 1, &deleter);
+      cache_->Insert(key, new char[10], 1,
+                     SimpleDeleter<char[]>::GetInstance());
     }
   }
 
@@ -227,7 +226,8 @@ class CacheBench {
       int32_t prob_op = thread->rnd.Uniform(100);
       if (prob_op >= 0 && prob_op < FLAGS_insert_percent) {
         // do insert
-        cache_->Insert(key, new char[10], 1, &deleter);
+        cache_->Insert(key, new char[10], 1,
+                       SimpleDeleter<char[]>::GetInstance());
       } else if (prob_op -= FLAGS_insert_percent &&
                  prob_op < FLAGS_lookup_percent) {
         // do lookup

cache/cache_test.cc

@@ -16,6 +16,7 @@
 #include <vector>
 #include "cache/clock_cache.h"
 #include "cache/lru_cache.h"
+#include "cache/simple_deleter.h"
 #include "test_util/testharness.h"
 #include "util/coding.h"
 #include "util/string_util.h"
@@ -40,21 +41,17 @@ static int DecodeValue(void* v) {
 const std::string kLRU = "lru";
 const std::string kClock = "clock";
 
-void dumbDeleter(const Slice& /*key*/, void* /*value*/) {}
-
-void eraseDeleter(const Slice& /*key*/, void* value) {
-  Cache* cache = reinterpret_cast<Cache*>(value);
-  cache->Erase("foo");
-}
-
 class CacheTest : public testing::TestWithParam<std::string> {
  public:
   static CacheTest* current_;
 
-  static void Deleter(const Slice& key, void* v) {
+  class Deleter : public Cache::Deleter {
+   public:
+    void operator()(const Slice& key, void* v) override {
       current_->deleted_keys_.push_back(DecodeKey(key));
       current_->deleted_values_.push_back(DecodeValue(v));
     }
+  };
 
   static const int kCacheSize = 1000;
   static const int kNumShardBits = 4;
@@ -64,6 +61,7 @@ class CacheTest : public testing::TestWithParam<std::string> {
 
   std::vector<int> deleted_keys_;
   std::vector<int> deleted_values_;
+  Deleter deleter_;
   std::shared_ptr<Cache> cache_;
   std::shared_ptr<Cache> cache2_;
 
@@ -117,8 +115,7 @@ class CacheTest : public testing::TestWithParam<std::string> {
 
   void Insert(std::shared_ptr<Cache> cache, int key, int value,
               int charge = 1) {
-    cache->Insert(EncodeKey(key), EncodeValue(value), charge,
-                  &CacheTest::Deleter);
+    cache->Insert(EncodeKey(key), EncodeValue(value), charge, &deleter_);
   }
 
   void Erase(std::shared_ptr<Cache> cache, int key) {
@@ -167,9 +164,9 @@ TEST_P(CacheTest, UsageTest) {
   for (int i = 1; i < 100; ++i) {
     std::string key(i, 'a');
     auto kv_size = key.size() + 5;
-    cache->Insert(key, reinterpret_cast<void*>(value), kv_size, dumbDeleter);
+    cache->Insert(key, reinterpret_cast<void*>(value), kv_size, nullptr);
     precise_cache->Insert(key, reinterpret_cast<void*>(value), kv_size,
-                          dumbDeleter);
+                          nullptr);
     usage += kv_size;
     ASSERT_EQ(usage, cache->GetUsage());
     ASSERT_LT(usage, precise_cache->GetUsage());
@@ -183,10 +180,9 @@ TEST_P(CacheTest, UsageTest) {
   // make sure the cache will be overloaded
   for (uint64_t i = 1; i < kCapacity; ++i) {
     auto key = ToString(i);
-    cache->Insert(key, reinterpret_cast<void*>(value), key.size() + 5,
-                  dumbDeleter);
+    cache->Insert(key, reinterpret_cast<void*>(value), key.size() + 5, nullptr);
     precise_cache->Insert(key, reinterpret_cast<void*>(value), key.size() + 5,
-                          dumbDeleter);
+                          nullptr);
   }
 
   // the usage should be close to the capacity
@@ -215,11 +211,11 @@ TEST_P(CacheTest, PinnedUsageTest) {
     auto kv_size = key.size() + 5;
     Cache::Handle* handle;
     Cache::Handle* handle_in_precise_cache;
-    cache->Insert(key, reinterpret_cast<void*>(value), kv_size, dumbDeleter,
+    cache->Insert(key, reinterpret_cast<void*>(value), kv_size, nullptr,
                   &handle);
     assert(handle);
-    precise_cache->Insert(key, reinterpret_cast<void*>(value), kv_size,
-                          dumbDeleter, &handle_in_precise_cache);
+    precise_cache->Insert(key, reinterpret_cast<void*>(value), kv_size, nullptr,
+                          &handle_in_precise_cache);
     assert(handle_in_precise_cache);
     pinned_usage += kv_size;
     ASSERT_EQ(pinned_usage, cache->GetPinnedUsage());
@@ -254,10 +250,9 @@ TEST_P(CacheTest, PinnedUsageTest) {
   // check that overloading the cache does not change the pinned usage
   for (uint64_t i = 1; i < 2 * kCapacity; ++i) {
     auto key = ToString(i);
-    cache->Insert(key, reinterpret_cast<void*>(value), key.size() + 5,
-                  dumbDeleter);
+    cache->Insert(key, reinterpret_cast<void*>(value), key.size() + 5, nullptr);
     precise_cache->Insert(key, reinterpret_cast<void*>(value), key.size() + 5,
-                          dumbDeleter);
+                          nullptr);
   }
   ASSERT_EQ(pinned_usage, cache->GetPinnedUsage());
   ASSERT_EQ(precise_cache_pinned_usage, precise_cache->GetPinnedUsage());
@@ -450,15 +445,25 @@ TEST_P(CacheTest, EvictionPolicyRef) {
 TEST_P(CacheTest, EvictEmptyCache) {
   // Insert item large than capacity to trigger eviction on empty cache.
   auto cache = NewCache(1, 0, false);
-  ASSERT_OK(cache->Insert("foo", nullptr, 10, dumbDeleter));
+  ASSERT_OK(cache->Insert("foo", nullptr, 10, nullptr));
 }
 
 TEST_P(CacheTest, EraseFromDeleter) {
   // Have deleter which will erase item from cache, which will re-enter
   // the cache at that point.
+  class EraseDeleter : public Cache::Deleter {
+   public:
+    void operator()(const Slice& /*key*/, void* value) override {
+      Cache* const cache = static_cast<Cache*>(value);
+      cache->Erase("foo");
+    }
+  };
+
+  EraseDeleter erase_deleter;
+
   std::shared_ptr<Cache> cache = NewCache(10, 0, false);
-  ASSERT_OK(cache->Insert("foo", nullptr, 1, dumbDeleter));
-  ASSERT_OK(cache->Insert("bar", cache.get(), 1, eraseDeleter));
+  ASSERT_OK(cache->Insert("foo", nullptr, 1, nullptr));
+  ASSERT_OK(cache->Insert("bar", cache.get(), 1, &erase_deleter));
   cache->Erase("bar");
   ASSERT_EQ(nullptr, cache->Lookup("foo"));
   ASSERT_EQ(nullptr, cache->Lookup("bar"));
@@ -527,17 +532,11 @@ class Value {
   size_t v_;
 };
 
-namespace {
-void deleter(const Slice& /*key*/, void* value) {
-  delete static_cast<Value *>(value);
-}
-}  // namespace
-
 TEST_P(CacheTest, ReleaseAndErase) {
   std::shared_ptr<Cache> cache = NewCache(5, 0, false);
   Cache::Handle* handle;
-  Status s = cache->Insert(EncodeKey(100), EncodeValue(100), 1,
-                           &CacheTest::Deleter, &handle);
+  Status s =
+      cache->Insert(EncodeKey(100), EncodeValue(100), 1, &deleter_, &handle);
   ASSERT_TRUE(s.ok());
   ASSERT_EQ(5U, cache->GetCapacity());
   ASSERT_EQ(1U, cache->GetUsage());
@@ -551,8 +550,8 @@ TEST_P(CacheTest, ReleaseAndErase) {
 TEST_P(CacheTest, ReleaseWithoutErase) {
   std::shared_ptr<Cache> cache = NewCache(5, 0, false);
   Cache::Handle* handle;
-  Status s = cache->Insert(EncodeKey(100), EncodeValue(100), 1,
-                           &CacheTest::Deleter, &handle);
+  Status s =
+      cache->Insert(EncodeKey(100), EncodeValue(100), 1, &deleter_, &handle);
   ASSERT_TRUE(s.ok());
   ASSERT_EQ(5U, cache->GetCapacity());
   ASSERT_EQ(1U, cache->GetUsage());
@@ -574,7 +573,8 @@ TEST_P(CacheTest, SetCapacity) {
   // Insert 5 entries, but not releasing.
   for (size_t i = 0; i < 5; i++) {
     std::string key = ToString(i+1);
-    Status s = cache->Insert(key, new Value(i + 1), 1, &deleter, &handles[i]);
+    Status s = cache->Insert(key, new Value(i + 1), 1,
+                             SimpleDeleter<Value>::GetInstance(), &handles[i]);
     ASSERT_TRUE(s.ok());
   }
   ASSERT_EQ(5U, cache->GetCapacity());
@@ -589,7 +589,8 @@ TEST_P(CacheTest, SetCapacity) {
   // and usage should be 7
   for (size_t i = 5; i < 10; i++) {
     std::string key = ToString(i+1);
-    Status s = cache->Insert(key, new Value(i + 1), 1, &deleter, &handles[i]);
+    Status s = cache->Insert(key, new Value(i + 1), 1,
+                             SimpleDeleter<Value>::GetInstance(), &handles[i]);
     ASSERT_TRUE(s.ok());
   }
   ASSERT_EQ(10U, cache->GetCapacity());
@@ -617,7 +618,8 @@ TEST_P(LRUCacheTest, SetStrictCapacityLimit) {
   Status s;
   for (size_t i = 0; i < 10; i++) {
     std::string key = ToString(i + 1);
-    s = cache->Insert(key, new Value(i + 1), 1, &deleter, &handles[i]);
+    s = cache->Insert(key, new Value(i + 1), 1,
+                      SimpleDeleter<Value>::GetInstance(), &handles[i]);
     ASSERT_OK(s);
     ASSERT_NE(nullptr, handles[i]);
   }
@@ -628,7 +630,8 @@ TEST_P(LRUCacheTest, SetStrictCapacityLimit) {
   Value* extra_value = new Value(0);
   cache->SetStrictCapacityLimit(true);
   Cache::Handle* handle;
-  s = cache->Insert(extra_key, extra_value, 1, &deleter, &handle);
+  s = cache->Insert(extra_key, extra_value, 1,
+                    SimpleDeleter<Value>::GetInstance(), &handle);
   ASSERT_TRUE(s.IsIncomplete());
   ASSERT_EQ(nullptr, handle);
   ASSERT_EQ(10, cache->GetUsage());
@@ -641,15 +644,18 @@ TEST_P(LRUCacheTest, SetStrictCapacityLimit) {
   std::shared_ptr<Cache> cache2 = NewCache(5, 0, true);
   for (size_t i = 0; i < 5; i++) {
     std::string key = ToString(i + 1);
-    s = cache2->Insert(key, new Value(i + 1), 1, &deleter, &handles[i]);
+    s = cache2->Insert(key, new Value(i + 1), 1,
+                       SimpleDeleter<Value>::GetInstance(), &handles[i]);
     ASSERT_OK(s);
     ASSERT_NE(nullptr, handles[i]);
   }
-  s = cache2->Insert(extra_key, extra_value, 1, &deleter, &handle);
+  s = cache2->Insert(extra_key, extra_value, 1,
+                     SimpleDeleter<Value>::GetInstance(), &handle);
   ASSERT_TRUE(s.IsIncomplete());
   ASSERT_EQ(nullptr, handle);
   // test insert without handle
-  s = cache2->Insert(extra_key, extra_value, 1, &deleter);
+  s = cache2->Insert(extra_key, extra_value, 1,
+                     SimpleDeleter<Value>::GetInstance());
   // AS if the key have been inserted into cache but get evicted immediately.
   ASSERT_OK(s);
   ASSERT_EQ(5, cache2->GetUsage());
@@ -671,7 +677,8 @@ TEST_P(CacheTest, OverCapacity) {
   // Insert n+1 entries, but not releasing.
   for (size_t i = 0; i < n + 1; i++) {
     std::string key = ToString(i+1);
-    Status s = cache->Insert(key, new Value(i + 1), 1, &deleter, &handles[i]);
+    Status s = cache->Insert(key, new Value(i + 1), 1,
+                             SimpleDeleter<Value>::GetInstance(), &handles[i]);
     ASSERT_TRUE(s.ok());
   }
 

cache/clock_cache.cc

@@ -175,11 +175,13 @@ namespace {
 
 // Cache entry meta data.
 struct CacheHandle {
+  using Deleter = Cache::Deleter;
+
   Slice key;
   uint32_t hash;
   void* value;
   size_t charge;
-  void (*deleter)(const Slice&, void* value);
+  Deleter* deleter;
 
   // Flags and counters associated with the cache handle:
   //   lowest bit: n-cache bit
@@ -193,8 +195,7 @@ struct CacheHandle {
 
   CacheHandle(const CacheHandle& a) { *this = a; }
 
-  CacheHandle(const Slice& k, void* v,
-              void (*del)(const Slice& key, void* value))
+  CacheHandle(const Slice& k, void* v, Deleter* del)
       : key(k), value(v), deleter(del) {}
 
   CacheHandle& operator=(const CacheHandle& a) {
@@ -269,8 +270,8 @@ class ClockCacheShard final : public CacheShard {
   void SetCapacity(size_t capacity) override;
   void SetStrictCapacityLimit(bool strict_capacity_limit) override;
   Status Insert(const Slice& key, uint32_t hash, void* value, size_t charge,
-                void (*deleter)(const Slice& key, void* value),
-                Cache::Handle** handle, Cache::Priority priority) override;
+                Deleter* deleter, Cache::Handle** handle,
+                Cache::Priority priority) override;
   Cache::Handle* Lookup(const Slice& key, uint32_t hash) override;
   // If the entry in in cache, increase reference count and return true.
   // Return false otherwise.
@@ -339,9 +340,8 @@ class ClockCacheShard final : public CacheShard {
   bool EvictFromCache(size_t charge, CleanupContext* context);
 
   CacheHandle* Insert(const Slice& key, uint32_t hash, void* value,
-                      size_t change,
-                      void (*deleter)(const Slice& key, void* value),
-                      bool hold_reference, CleanupContext* context);
+                      size_t charge, Deleter* deleter, bool hold_reference,
+                      CleanupContext* context);
 
   // Guards list_, head_, and recycle_. In addition, updating table_ also has
   // to hold the mutex, to avoid the cache being in inconsistent state.
@@ -561,9 +561,9 @@ void ClockCacheShard::SetStrictCapacityLimit(bool strict_capacity_limit) {
                                std::memory_order_relaxed);
 }
 
-CacheHandle* ClockCacheShard::Insert(
-    const Slice& key, uint32_t hash, void* value, size_t charge,
-    void (*deleter)(const Slice& key, void* value), bool hold_reference,
+CacheHandle* ClockCacheShard::Insert(const Slice& key, uint32_t hash,
+                                     void* value, size_t charge,
+                                     Deleter* deleter, bool hold_reference,
                                      CleanupContext* context) {
   size_t total_charge =
       CacheHandle::CalcTotalCharge(key, charge, metadata_charge_policy_);
@@ -610,8 +610,7 @@ CacheHandle* ClockCacheShard::Insert(
 }
 
 Status ClockCacheShard::Insert(const Slice& key, uint32_t hash, void* value,
-                               size_t charge,
-                               void (*deleter)(const Slice& key, void* value),
+                               size_t charge, Deleter* deleter,
                                Cache::Handle** out_handle,
                                Cache::Priority /*priority*/) {
   CleanupContext context;

cache/lru_cache.cc

@@ -337,8 +337,7 @@ bool LRUCacheShard::Release(Cache::Handle* handle, bool force_erase) {
 }
 
 Status LRUCacheShard::Insert(const Slice& key, uint32_t hash, void* value,
-                             size_t charge,
-                             void (*deleter)(const Slice& key, void* value),
+                             size_t charge, Deleter* deleter,
                              Cache::Handle** handle, Cache::Priority priority) {
   // Allocate the memory here outside of the mutex
   // If the cache is full, we'll have to release it

cache/lru_cache.h

@@ -48,8 +48,10 @@ namespace ROCKSDB_NAMESPACE {
 // (to move into state 3).
 
 struct LRUHandle {
+  using Deleter = Cache::Deleter;
+
   void* value;
-  void (*deleter)(const Slice&, void* value);
+  Deleter* deleter;
   LRUHandle* next_hash;
   LRUHandle* next;
   LRUHandle* prev;
@@ -209,9 +211,7 @@ class ALIGN_AS(CACHE_LINE_SIZE) LRUCacheShard final : public CacheShard {
 
   // Like Cache methods, but with an extra "hash" parameter.
   virtual Status Insert(const Slice& key, uint32_t hash, void* value,
-                        size_t charge,
-                        void (*deleter)(const Slice& key, void* value),
-                        Cache::Handle** handle,
+                        size_t charge, Deleter* deleter, Cache::Handle** handle,
                         Cache::Priority priority) override;
   virtual Cache::Handle* Lookup(const Slice& key, uint32_t hash) override;
   virtual bool Ref(Cache::Handle* handle) override;

cache/sharded_cache.cc

@@ -44,8 +44,8 @@ void ShardedCache::SetStrictCapacityLimit(bool strict_capacity_limit) {
 }
 
 Status ShardedCache::Insert(const Slice& key, void* value, size_t charge,
-                            void (*deleter)(const Slice& key, void* value),
-                            Handle** handle, Priority priority) {
+                            Deleter* deleter, Handle** handle,
+                            Priority priority) {
   uint32_t hash = HashSlice(key);
   return GetShard(Shard(hash))
       ->Insert(key, hash, value, charge, deleter, handle, priority);

cache/sharded_cache.h

@@ -21,13 +21,14 @@ namespace ROCKSDB_NAMESPACE {
 // Single cache shard interface.
 class CacheShard {
  public:
+  using Deleter = Cache::Deleter;
+
   CacheShard() = default;
   virtual ~CacheShard() = default;
 
   virtual Status Insert(const Slice& key, uint32_t hash, void* value,
-                        size_t charge,
-                        void (*deleter)(const Slice& key, void* value),
-                        Cache::Handle** handle, Cache::Priority priority) = 0;
+                        size_t charge, Deleter* deleter, Cache::Handle** handle,
+                        Cache::Priority priority) = 0;
   virtual Cache::Handle* Lookup(const Slice& key, uint32_t hash) = 0;
   virtual bool Ref(Cache::Handle* handle) = 0;
   virtual bool Release(Cache::Handle* handle, bool force_erase = false) = 0;
@@ -70,8 +71,8 @@ class ShardedCache : public Cache {
   virtual void SetStrictCapacityLimit(bool strict_capacity_limit) override;
 
   virtual Status Insert(const Slice& key, void* value, size_t charge,
-                        void (*deleter)(const Slice& key, void* value),
-                        Handle** handle, Priority priority) override;
+                        Deleter* deleter, Handle** handle,
+                        Priority priority) override;
   virtual Handle* Lookup(const Slice& key, Statistics* stats) override;
   virtual bool Ref(Handle* handle) override;
   virtual bool Release(Handle* handle, bool force_erase = false) override;

cache/simple_deleter.h

@@ -0,0 +1,47 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+
+#include "rocksdb/cache.h"
+#include "rocksdb/rocksdb_namespace.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+template <typename T>
+class SimpleDeleter : public Cache::Deleter {
+ public:
+  static SimpleDeleter* GetInstance() {
+    static auto deleter = new SimpleDeleter;
+    return deleter;
+  }
+
+  void operator()(const Slice& /* key */, void* value) override {
+    T* const t = static_cast<T*>(value);
+    delete t;
+  }
+
+ private:
+  SimpleDeleter() = default;
+};
+
+template <typename T>
+class SimpleDeleter<T[]> : public Cache::Deleter {
+ public:
+  static SimpleDeleter* GetInstance() {
+    static auto deleter = new SimpleDeleter;
+    return deleter;
+  }
+
+  void operator()(const Slice& /* key */, void* value) override {
+    T* const t = static_cast<T*>(value);
+    delete[] t;
+  }
+
+ private:
+  SimpleDeleter() = default;
+};
+
+}  // namespace ROCKSDB_NAMESPACE

db/db_basic_test.cc

@@ -2053,8 +2053,7 @@ class DBBasicTestWithParallelIO
     virtual const char* Name() const override { return "MyBlockCache"; }
 
     virtual Status Insert(const Slice& key, void* value, size_t charge,
-                          void (*deleter)(const Slice& key, void* value),
-                          Handle** handle = nullptr,
+                          Deleter* deleter, Handle** handle = nullptr,
                           Priority priority = Priority::LOW) override {
       num_inserts_++;
       return target_->Insert(key, value, charge, deleter, handle, priority);

db/db_block_cache_test.cc

@@ -443,9 +443,8 @@ class MockCache : public LRUCache {
                  false /*strict_capacity_limit*/, 0.0 /*high_pri_pool_ratio*/) {
   }
 
-  Status Insert(const Slice& key, void* value, size_t charge,
-                void (*deleter)(const Slice& key, void* value), Handle** handle,
-                Priority priority) override {
+  Status Insert(const Slice& key, void* value, size_t charge, Deleter* deleter,
+                Handle** handle, Priority priority) override {
     if (priority == Priority::LOW) {
       low_pri_insert_count++;
     } else {

db/table_cache.cc

@@ -9,6 +9,7 @@
 
 #include "db/table_cache.h"
 
+#include "cache/simple_deleter.h"
 #include "db/dbformat.h"
 #include "db/range_tombstone_fragmenter.h"
 #include "db/snapshot_impl.h"
@@ -33,12 +34,6 @@ namespace ROCKSDB_NAMESPACE {
 
 namespace {
 
-template <class T>
-static void DeleteEntry(const Slice& /*key*/, void* value) {
-  T* typed_value = reinterpret_cast<T*>(value);
-  delete typed_value;
-}
-
 static void UnrefEntry(void* arg1, void* arg2) {
   Cache* cache = reinterpret_cast<Cache*>(arg1);
   Cache::Handle* h = reinterpret_cast<Cache::Handle*>(arg2);
@@ -166,8 +161,8 @@ Status TableCache::FindTable(const FileOptions& file_options,
       // We do not cache error results so that if the error is transient,
       // or somebody repairs the file, we recover automatically.
     } else {
-      s = cache_->Insert(key, table_reader.get(), 1, &DeleteEntry<TableReader>,
-                         handle);
+      s = cache_->Insert(key, table_reader.get(), 1,
+                         SimpleDeleter<TableReader>::GetInstance(), handle);
       if (s.ok()) {
         // Release ownership of table reader.
         table_reader.release();
@@ -425,7 +420,7 @@ Status TableCache::Get(const ReadOptions& options,
         row_cache_key.Size() + row_cache_entry->size() + sizeof(std::string);
     void* row_ptr = new std::string(std::move(*row_cache_entry));
     ioptions_.row_cache->Insert(row_cache_key.GetUserKey(), row_ptr, charge,
-                                &DeleteEntry<std::string>);
+                                SimpleDeleter<std::string>::GetInstance());
   }
 #endif  // ROCKSDB_LITE
 
@@ -545,7 +540,7 @@ Status TableCache::MultiGet(const ReadOptions& options,
             row_cache_key.Size() + row_cache_entry.size() + sizeof(std::string);
         void* row_ptr = new std::string(std::move(row_cache_entry));
         ioptions_.row_cache->Insert(row_cache_key.GetUserKey(), row_ptr, charge,
-                                    &DeleteEntry<std::string>);
+                                    SimpleDeleter<std::string>::GetInstance());
       }
     }
   }

include/rocksdb/cache.h

@@ -132,6 +132,13 @@ extern std::shared_ptr<Cache> NewClockCache(
         kDefaultCacheMetadataChargePolicy);
 class Cache {
  public:
+  class Deleter {
+   public:
+    virtual ~Deleter() = default;
+
+    virtual void operator()(const Slice& key, void* value) = 0;
+  };
+
   // Depending on implementation, cache entries with high priority could be less
   // likely to get evicted than low priority entries.
   enum class Priority { HIGH, LOW };
@@ -168,10 +175,10 @@ class Cache {
   // insert. In case of error value will be cleanup.
   //
   // When the inserted entry is no longer needed, the key and
-  // value will be passed to "deleter".
+  // value will be passed to "deleter". It is the caller's responsibility to
+  // ensure that the deleter outlives the cache entries referring to it.
   virtual Status Insert(const Slice& key, void* value, size_t charge,
-                        void (*deleter)(const Slice& key, void* value),
-                        Handle** handle = nullptr,
+                        Deleter* deleter, Handle** handle = nullptr,
                         Priority priority = Priority::LOW) = 0;
 
   // If the cache has no mapping for "key", returns nullptr.

table/block_based/block_based_table_builder.cc

@@ -18,6 +18,7 @@
 #include <unordered_map>
 #include <utility>
 
+#include "cache/simple_deleter.h"
 #include "db/dbformat.h"
 #include "index_builder.h"
 
@@ -790,11 +791,6 @@ void BlockBasedTableBuilder::WriteRawBlock(const Slice& block_contents,
 
 Status BlockBasedTableBuilder::status() const { return rep_->status; }
 
-static void DeleteCachedBlockContents(const Slice& /*key*/, void* value) {
-  BlockContents* bc = reinterpret_cast<BlockContents*>(value);
-  delete bc;
-}
-
 //
 // Make a copy of the block contents and insert into compressed block cache
 //
@@ -829,7 +825,7 @@ Status BlockBasedTableBuilder::InsertBlockInCache(const Slice& block_contents,
     block_cache_compressed->Insert(
         key, block_contents_to_cache,
         block_contents_to_cache->ApproximateMemoryUsage(),
-        &DeleteCachedBlockContents);
+        SimpleDeleter<BlockContents>::GetInstance());
 
     // Invalidate OS cache.
     r->file->InvalidateCache(static_cast<size_t>(r->offset), size);

table/block_based/block_based_table_reader.cc

@@ -14,6 +14,8 @@
 #include <utility>
 #include <vector>
 
+#include "cache/simple_deleter.h"
+
 #include "db/dbformat.h"
 #include "db/pinned_iterators_manager.h"
 
@@ -179,13 +181,6 @@ Status ReadBlockFromFile(
   return s;
 }
 
-// Delete the entry resided in the cache.
-template <class Entry>
-void DeleteCachedEntry(const Slice& /*key*/, void* value) {
-  auto entry = reinterpret_cast<Entry*>(value);
-  delete entry;
-}
-
 // Release the cached entry and decrement its ref count.
 // Do not force erase
 void ReleaseCachedEntry(void* arg, void* h) {
@@ -1171,7 +1166,8 @@ Status BlockBasedTable::GetDataBlockFromCache(
       size_t charge = block_holder->ApproximateMemoryUsage();
       Cache::Handle* cache_handle = nullptr;
       s = block_cache->Insert(block_cache_key, block_holder.get(), charge,
-                              &DeleteCachedEntry<TBlocklike>, &cache_handle);
+                              SimpleDeleter<TBlocklike>::GetInstance(),
+                              &cache_handle);
       if (s.ok()) {
         assert(cache_handle != nullptr);
         block->SetCachedValue(block_holder.release(), block_cache,
@@ -1260,7 +1256,7 @@ Status BlockBasedTable::PutDataBlockToCache(
     s = block_cache_compressed->Insert(
         compressed_block_cache_key, block_cont_for_comp_cache,
         block_cont_for_comp_cache->ApproximateMemoryUsage(),
-        &DeleteCachedEntry<BlockContents>);
+        SimpleDeleter<BlockContents>::GetInstance());
     if (s.ok()) {
       // Avoid the following code to delete this cached block.
       RecordTick(statistics, BLOCK_CACHE_COMPRESSED_ADD);
@@ -1275,8 +1271,8 @@ Status BlockBasedTable::PutDataBlockToCache(
     size_t charge = block_holder->ApproximateMemoryUsage();
     Cache::Handle* cache_handle = nullptr;
     s = block_cache->Insert(block_cache_key, block_holder.get(), charge,
-                            &DeleteCachedEntry<TBlocklike>, &cache_handle,
-                            priority);
+                            SimpleDeleter<TBlocklike>::GetInstance(),
+                            &cache_handle, priority);
     if (s.ok()) {
       assert(cache_handle != nullptr);
       cached_block->SetCachedValue(block_holder.release(), block_cache,

table/block_based/partitioned_index_reader.cc

@@ -7,6 +7,8 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 #include "table/block_based/partitioned_index_reader.h"
+
+#include "cache/simple_deleter.h"
 #include "table/block_based/partitioned_index_iterator.h"
 
 namespace ROCKSDB_NAMESPACE {

utilities/simulator_cache/sim_cache.cc

@@ -168,19 +168,15 @@ class SimCacheImpl : public SimCache {
     cache_->SetStrictCapacityLimit(strict_capacity_limit);
   }
 
-  Status Insert(const Slice& key, void* value, size_t charge,
-                void (*deleter)(const Slice& key, void* value), Handle** handle,
-                Priority priority) override {
+  Status Insert(const Slice& key, void* value, size_t charge, Deleter* deleter,
+                Handle** handle, Priority priority) override {
     // The handle and value passed in are for real cache, so we pass nullptr
-    // to key_only_cache_ for both instead. Also, the deleter function pointer
-    // will be called by user to perform some external operation which should
-    // be applied only once. Thus key_only_cache accepts an empty function.
-    // *Lambda function without capture can be assgined to a function pointer
+    // to key_only_cache_ for both instead. Also, the deleter should be invoked
+    // only once (on the actual value), so we pass nullptr to key_only_cache for
+    // that one as well.
     Handle* h = key_only_cache_->Lookup(key);
     if (h == nullptr) {
-      key_only_cache_->Insert(key, nullptr, charge,
-                              [](const Slice& /*k*/, void* /*v*/) {}, nullptr,
-                              priority);
+      key_only_cache_->Insert(key, nullptr, charge, nullptr, nullptr, priority);
     } else {
       key_only_cache_->Release(h);
     }