Create Cache as a Customizable Object

HISTORY.md

@@ -153,6 +153,7 @@
   * `rocksdb_comparator_with_ts_create` to create timestamp aware comparator
   * Put, Get, Delete, SingleDelete, MultiGet APIs has corresponding timestamp aware APIs with suffix `with_ts`
   * And Add C API's for Transaction, SstFileWriter, Compaction as mentioned [here](https://github.com/facebook/rocksdb/wiki/User-defined-Timestamp-(Experimental))
+* Removed the memory_allocator_ private variable from the Cache implementation.  Custom implementations must provide their own storage and implementation of the memory_allocator API.
 * The contract for implementations of Comparator::IsSameLengthImmediateSuccessor has been updated to work around a design bug in `auto_prefix_mode`.
 * The API documentation for `auto_prefix_mode` now notes some corner cases in which it returns different results than `total_order_seek`, due to design bugs that are not easily fixed. Users using built-in comparators and keys at least the size of a fixed prefix length are not affected.
 * Obsoleted the NUM_DATA_BLOCKS_READ_PER_LEVEL stat and introduced the NUM_LEVEL_READ_PER_MULTIGET and MULTIGET_COROUTINE_COUNT stats
@@ -163,6 +164,7 @@
 * Add blob garbage collection parameters `blob_garbage_collection_policy` and `blob_garbage_collection_age_cutoff` to both force-enable and force-disable GC, as well as selectively override age cutoff when using CompactRange.
 * Add an extra sanity check in `GetSortedWalFiles()` (also used by `GetLiveFilesStorageInfo()`, `BackupEngine`, and `Checkpoint`) to reduce risk of successfully created backup or checkpoint failing to open because of missing WAL file.
 * Add a new column family option `blob_file_starting_level` to enable writing blob files during flushes and compactions starting from the specified LSM tree level.
+* Allow Cache instances to be created from custom implementations registered with the ObjectRegistry.
 * Add support for timestamped snapshots (#9879)
 * Provide support for AbortIO in posix to cancel submitted asynchronous requests using io_uring.
 * Add support for rate-limiting batched `MultiGet()` APIs

cache/cache.cc

@@ -9,14 +9,71 @@
 
 #include "rocksdb/cache.h"
 
+#include "cache/clock_cache.h"
+#include "cache/fast_lru_cache.h"
 #include "cache/lru_cache.h"
+#include "rocksdb/configurable.h"
 #include "rocksdb/secondary_cache.h"
 #include "rocksdb/utilities/customizable_util.h"
 #include "rocksdb/utilities/options_type.h"
 #include "util/string_util.h"
 
 namespace ROCKSDB_NAMESPACE {
+namespace {
 #ifndef ROCKSDB_LITE
+int RegisterBuiltinCache(ObjectLibrary& library, const std::string& /*arg*/) {
+  library.AddFactory<Cache>(
+      lru_cache::LRUCache::kClassName(),
+      [](const std::string& /*uri*/, std::unique_ptr<Cache>* guard,
+         std::string* /* errmsg */) {
+        *guard = std::make_unique<lru_cache::LRUCache>();
+        return guard->get();
+      });
+  library.AddFactory<Cache>(
+      fast_lru_cache::LRUCache::kClassName(),
+      [](const std::string& /*uri*/, std::unique_ptr<Cache>* guard,
+         std::string* /* errmsg */) {
+        *guard = std::make_unique<fast_lru_cache::LRUCache>();
+        return guard->get();
+      });
+  library.AddFactory<Cache>(
+      clock_cache::ClockCache::kClassName(),
+      [](const std::string& /*uri*/, std::unique_ptr<Cache>* guard,
+         std::string* /*errmsg*/) {
+        *guard = std::make_unique<clock_cache::ClockCache>();
+        return guard->get();
+      });
+  //** Register AsIndividualId for the moment to pass the tests
+  // If the Cache is made to create as a ManagedObject, these factories
+  // may not be necessary as the ManagedObject code should handle it.
+  library.AddFactory<Cache>(
+      ObjectLibrary::PatternEntry::AsIndividualId(
+          lru_cache::LRUCache::kClassName()),
+      [](const std::string& /*uri*/, std::unique_ptr<Cache>* guard,
+         std::string* /* errmsg */) {
+        *guard = std::make_unique<lru_cache::LRUCache>();
+        return guard->get();
+      });
+  library.AddFactory<Cache>(
+      ObjectLibrary::PatternEntry::AsIndividualId(
+          fast_lru_cache::LRUCache::kClassName()),
+      [](const std::string& /*uri*/, std::unique_ptr<Cache>* guard,
+         std::string* /* errmsg */) {
+        *guard = std::make_unique<fast_lru_cache::LRUCache>();
+        return guard->get();
+      });
+  library.AddFactory<Cache>(
+      ObjectLibrary::PatternEntry::AsIndividualId(
+          clock_cache::ClockCache::kClassName()),
+      [](const std::string& /*uri*/, std::unique_ptr<Cache>* guard,
+         std::string* /*errmsg*/) {
+        *guard = std::make_unique<clock_cache::ClockCache>();
+        return guard->get();
+      });
+  size_t num_types;
+  return static_cast<int>(library.GetFactoryCount(&num_types));
+}
+
 static std::unordered_map<std::string, OptionTypeInfo>
     lru_cache_options_type_info = {
         {"capacity",
@@ -60,6 +117,7 @@ static std::unordered_map<std::string, OptionTypeInfo>
           OptionTypeFlags::kMutable}},
 };
 #endif  // ROCKSDB_LITE
+}  // namespace
 
 Status SecondaryCache::CreateFromString(
     const ConfigOptions& config_options, const std::string& value,
@@ -98,23 +156,27 @@ Status SecondaryCache::CreateFromString(
 Status Cache::CreateFromString(const ConfigOptions& config_options,
                                const std::string& value,
                                std::shared_ptr<Cache>* result) {
+#ifndef ROCKSDB_LITE
+  static std::once_flag once;
+  std::call_once(
+      once, [&]() { RegisterBuiltinCache(*(ObjectLibrary::Default()), ""); });
+#endif  // ROCKSDB_LITE
   Status status;
   std::shared_ptr<Cache> cache;
-  if (value.find('=') == std::string::npos) {
-    cache = NewLRUCache(ParseSizeT(value));
-  } else {
-#ifndef ROCKSDB_LITE
-    LRUCacheOptions cache_opts;
-    status = OptionTypeInfo::ParseStruct(config_options, "",
-                                         &lru_cache_options_type_info, "",
-                                         value, &cache_opts);
-    if (status.ok()) {
-      cache = NewLRUCache(cache_opts);
+  if (!value.empty()) {
+    std::string id;
+    std::unordered_map<std::string, std::string> opt_map;
+    status = Configurable::GetOptionsMap(value, LRUCache::kClassName(), &id,
+                                         &opt_map);
+    if (!status.ok()) {
+      return status;
+    } else if (opt_map.empty() && !id.empty() && isdigit(id.at(0))) {
+      // If there are no name=value options and the id is a digit, assume
+      // it is an old-style LRUCache created by capacity only
+      cache = NewLRUCache(ParseSizeT(id));
+    } else {
+      status = NewSharedObject<Cache>(config_options, id, opt_map, &cache);
     }
-#else
-    (void)config_options;
-    status = Status::NotSupported("Cannot load cache in LITE mode ", value);
-#endif  //! ROCKSDB_LITE
   }
   if (status.ok()) {
     result->swap(cache);

cache/cache_bench_tool.cc

@@ -84,8 +84,8 @@ DEFINE_string(secondary_cache_uri, "",
               "Full URI for creating a custom secondary cache object");
 static class std::shared_ptr<ROCKSDB_NAMESPACE::SecondaryCache> secondary_cache;
 #endif  // ROCKSDB_LITE
-
-DEFINE_string(cache_type, "lru_cache", "Type of block cache.");
+DEFINE_string(cache_uri, "lru_cache", "URI for creating a cache");
+DEFINE_bool(use_clock_cache, false, "");
 
 // ## BEGIN stress_cache_key sub-tool options ##
 // See class StressCacheKey below.
@@ -291,26 +291,29 @@ class CacheBench {
       if (max_key > (static_cast<uint64_t>(1) << max_log_)) max_log_++;
     }
 
-    if (FLAGS_cache_type == "clock_cache") {
+    ConfigOptions config_options;
+    config_options.ignore_unknown_options = false;
+    config_options.ignore_unsupported_options = false;
+    if (FLAGS_cache_uri == "clock_cache") {
       cache_ = ExperimentalNewClockCache(
           FLAGS_cache_size, FLAGS_value_bytes, FLAGS_num_shard_bits,
           false /*strict_capacity_limit*/, kDefaultCacheMetadataChargePolicy);
       if (!cache_) {
         fprintf(stderr, "Clock cache not supported.\n");
         exit(1);
       }
-    } else if (FLAGS_cache_type == "fast_lru_cache") {
+    } else if (FLAGS_cache_uri == "fast_lru_cache") {
       cache_ = NewFastLRUCache(
           FLAGS_cache_size, FLAGS_value_bytes, FLAGS_num_shard_bits,
           false /*strict_capacity_limit*/, kDefaultCacheMetadataChargePolicy);
-    } else if (FLAGS_cache_type == "lru_cache") {
+    } else if (FLAGS_cache_uri == "lru_cache") {
       LRUCacheOptions opts(FLAGS_cache_size, FLAGS_num_shard_bits,
                            false /* strict_capacity_limit */,
                            0.5 /* high_pri_pool_ratio */);
 #ifndef ROCKSDB_LITE
       if (!FLAGS_secondary_cache_uri.empty()) {
         Status s = SecondaryCache::CreateFromString(
-            ConfigOptions(), FLAGS_secondary_cache_uri, &secondary_cache);
+            config_options, FLAGS_secondary_cache_uri, &secondary_cache);
         if (secondary_cache == nullptr) {
           fprintf(
               stderr,
@@ -324,8 +327,13 @@ class CacheBench {
 
       cache_ = NewLRUCache(opts);
     } else {
-      fprintf(stderr, "Cache type not supported.");
-      exit(1);
+      Status s =
+          Cache::CreateFromString(config_options, FLAGS_cache_uri, &cache_);
+      if (!s.ok() || !cache_) {
+        fprintf(stderr, "No cache registered matching string: %s status=%s\n",
+                FLAGS_cache_uri.c_str(), s.ToString().c_str());
+        exit(1);
+      }
     }
   }
 
@@ -621,9 +629,12 @@ class CacheBench {
     printf("DMutex impl name    : %s\n", DMutex::kName());
     printf("Number of threads   : %u\n", FLAGS_threads);
     printf("Ops per thread      : %" PRIu64 "\n", FLAGS_ops_per_thread);
-    printf("Cache size          : %s\n",
-           BytesToHumanString(FLAGS_cache_size).c_str());
-    printf("Num shard bits      : %u\n", FLAGS_num_shard_bits);
+    printf("Cache               : %s\n", cache_->Name());
+    std::string cache_opts;
+    Status s = cache_->GetOptionString(ConfigOptions(), &cache_opts);
+    if (s.ok()) {
+      printf("Cache Options       : %s\n", cache_opts.c_str());
+    }
     printf("Max key             : %" PRIu64 "\n", max_key_);
     printf("Resident ratio      : %g\n", FLAGS_resident_ratio);
     printf("Skew degree         : %u\n", FLAGS_skew);

cache/cache_test.cc

@@ -19,6 +19,7 @@
 #include "cache/fast_lru_cache.h"
 #include "cache/lru_cache.h"
 #include "port/stack_trace.h"
+#include "rocksdb/convenience.h"
 #include "test_util/testharness.h"
 #include "util/coding.h"
 #include "util/string_util.h"
@@ -990,9 +991,119 @@ TEST_P(CacheTest, GetChargeAndDeleter) {
   cache_->Release(h1);
 }
 
-std::shared_ptr<Cache> (*new_clock_cache_func)(size_t, size_t, int, bool,
-                                               CacheMetadataChargePolicy) =
-    ExperimentalNewClockCache;
+class CreateCacheTest : public testing::Test {
+ public:
+  CreateCacheTest() {
+    config_options_.ignore_unsupported_options = false;
+    config_options_.invoke_prepare_options = false;
+  }
+
+ protected:
+  ConfigOptions config_options_;
+};
+
+TEST_F(CreateCacheTest, LRUFromSize) {
+  std::shared_ptr<Cache> cache;
+  Options options;
+  ASSERT_OK(Cache::CreateFromString(config_options_, "1M", &cache));
+  ASSERT_NE(cache, nullptr);
+  ASSERT_STREQ(cache->Name(), lru_cache::LRUCache::kClassName());
+  ASSERT_EQ(cache->GetCapacity(), 1024U * 1024U);
+  ASSERT_OK(cache->ValidateOptions(options, options));
+}
+
+TEST_F(CreateCacheTest, LRUFromOptions) {
+  std::shared_ptr<Cache> cache, copy;
+  Options options;
+  Status s =
+      Cache::CreateFromString(config_options_,
+                              "capacity=1M;num_shard_bits=5;strict_capacity_"
+                              "limit=true; metadata_charge_policy=kDontCharge",
+                              &cache);
+#ifndef ROCKSDB_LITE
+  ASSERT_OK(s);
+  ASSERT_NE(cache, nullptr);
+  ASSERT_STREQ(cache->Name(), lru_cache::LRUCache::kClassName());
+  ASSERT_EQ(cache->GetCapacity(), 1024U * 1024U);
+  ASSERT_NOK(cache->ValidateOptions(options, options));  // Not prepared
+  ASSERT_OK(cache->PrepareOptions(config_options_));     // Prepare it
+  ASSERT_OK(cache->ValidateOptions(options, options));   // Prepared, all good
+  config_options_.invoke_prepare_options = true;
+  ASSERT_OK(Cache::CreateFromString(config_options_,
+                                    cache->ToString(config_options_), &copy));
+  ASSERT_NE(copy, nullptr);
+  ASSERT_STREQ(copy->Name(), lru_cache::LRUCache::kClassName());
+  std::string mismatch;
+  ASSERT_TRUE(cache->AreEquivalent(config_options_, copy.get(), &mismatch));
+  ASSERT_OK(copy->ValidateOptions(options, options));
+#else
+  ASSERT_NOK(s);
+#endif  // ROCKSDB_LITE
+}
+
+TEST_F(CreateCacheTest, ClockCache) {
+  std::shared_ptr<Cache> cache, copy;
+  Options options;
+  std::string id = clock_cache::ClockCache::kClassName();
+  Status s = Cache::CreateFromString(config_options_, id, &cache);
+#ifndef ROCKSDB_LITE
+  ASSERT_OK(s);
+  ASSERT_NE(cache, nullptr);
+  ASSERT_EQ(id, cache->Name());
+  ASSERT_NOK(cache->ValidateOptions(options, options));
+  cache->SetCapacity(1024U * 1024U);
+  ASSERT_EQ(cache->GetCapacity(), 1024U * 1024U);
+
+  ASSERT_OK(cache->PrepareOptions(config_options_));
+  ASSERT_OK(cache->ValidateOptions(options, options));
+
+  // cache->SetCapacity(1024U * 1024U); Currently ClockCache does not support
+  // SetCapacity
+  ASSERT_EQ(cache->GetCapacity(), 1024U * 1024U);
+
+  config_options_.invoke_prepare_options = true;
+  ASSERT_OK(Cache::CreateFromString(config_options_,
+                                    cache->ToString(config_options_), &copy));
+  ASSERT_NE(copy, nullptr);
+  ASSERT_STREQ(copy->Name(), clock_cache::ClockCache::kClassName());
+  std::string mismatch;
+  ASSERT_TRUE(cache->AreEquivalent(config_options_, copy.get(), &mismatch));
+  ASSERT_OK(copy->ValidateOptions(options, options));
+#else
+  ASSERT_NOK(s);
+#endif  // ROCKSDB_LITE
+}
+
+#ifndef ROCKSDB_LITE
+TEST_F(CreateCacheTest, LRUFromBadOptions) {
+  std::shared_ptr<Cache> cache;
+
+  ASSERT_NOK(Cache::CreateFromString(
+      config_options_, "capacityXX=1M; metadata_charge_policy=kDontCharge",
+      &cache));
+  ASSERT_NOK(Cache::CreateFromString(config_options_,
+                                     "metadata_charge_policy=kXX", &cache));
+  config_options_.invoke_prepare_options = true;
+  ASSERT_NOK(
+      Cache::CreateFromString(config_options_, "num_shard_bits=21", &cache));
+}
+
+TEST_F(CreateCacheTest, ClockCacheFromBadOptions) {
+  std::shared_ptr<Cache> cache;
+  std::string id = clock_cache::ClockCache::kClassName();
+  ASSERT_NOK(Cache::CreateFromString(
+      config_options_,
+      "capacityXX=1M; metadata_charge_policy=kDontCharge;id=" + id, &cache));
+  ASSERT_NOK(Cache::CreateFromString(
+      config_options_, "metadata_charge_policy=kXX; id=" + id, &cache));
+
+  config_options_.invoke_prepare_options = true;
+  ASSERT_NOK(Cache::CreateFromString(config_options_,
+                                     "num_shard_bits=21;id=" + id, &cache));
+}
+
+#endif  // ROCKSDB_LITE
+
 INSTANTIATE_TEST_CASE_P(CacheTestInstance, CacheTest,
                         testing::Values(kLRU, kClock, kFast));
 INSTANTIATE_TEST_CASE_P(CacheTestInstance, LRUCacheTest,