diff --git a/CMakeLists.txt b/CMakeLists.txt index 1a53fcba30..b475a2224f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -632,6 +632,7 @@ set(SOURCES cache/secondary_cache.cc cache/secondary_cache_adapter.cc cache/sharded_cache.cc + cache/tiered_secondary_cache.cc db/arena_wrapped_db_iter.cc db/blob/blob_contents.cc db/blob/blob_fetcher.cc @@ -1263,6 +1264,7 @@ if(WITH_TESTS) cache/cache_test.cc cache/compressed_secondary_cache_test.cc cache/lru_cache_test.cc + cache/tiered_secondary_cache_test.cc db/blob/blob_counting_iterator_test.cc db/blob/blob_file_addition_test.cc db/blob/blob_file_builder_test.cc diff --git a/Makefile b/Makefile index 08ad7e48ab..71c96f2846 100644 --- a/Makefile +++ b/Makefile @@ -1885,6 +1885,9 @@ compressed_secondary_cache_test: $(OBJ_DIR)/cache/compressed_secondary_cache_tes lru_cache_test: $(OBJ_DIR)/cache/lru_cache_test.o $(TEST_LIBRARY) $(LIBRARY) $(AM_LINK) +tiered_secondary_cache_test: $(OBJ_DIR)/cache/tiered_secondary_cache_test.o $(TEST_LIBRARY) $(LIBRARY) + $(AM_LINK) + range_del_aggregator_test: $(OBJ_DIR)/db/range_del_aggregator_test.o $(TEST_LIBRARY) $(LIBRARY) $(AM_LINK) diff --git a/TARGETS b/TARGETS index ad0da61008..86d62e2c53 100644 --- a/TARGETS +++ b/TARGETS @@ -21,6 +21,7 @@ cpp_library_wrapper(name="rocksdb_lib", srcs=[ "cache/secondary_cache.cc", "cache/secondary_cache_adapter.cc", "cache/sharded_cache.cc", + "cache/tiered_secondary_cache.cc", "db/arena_wrapped_db_iter.cc", "db/blob/blob_contents.cc", "db/blob/blob_fetcher.cc", @@ -5475,6 +5476,12 @@ cpp_unittest_wrapper(name="tiered_compaction_test", extra_compiler_flags=[]) +cpp_unittest_wrapper(name="tiered_secondary_cache_test", + srcs=["cache/tiered_secondary_cache_test.cc"], + deps=[":rocksdb_test_lib"], + extra_compiler_flags=[]) + + cpp_unittest_wrapper(name="timer_queue_test", srcs=["util/timer_queue_test.cc"], deps=[":rocksdb_test_lib"], diff --git a/cache/cache.cc b/cache/cache.cc index a65f5ec4f8..3dbea128e1 100644 --- a/cache/cache.cc +++ b/cache/cache.cc @@ -66,6 +66,41 @@ static std::unordered_map OptionTypeFlags::kMutable}}, }; +namespace { +static void NoopDelete(Cache::ObjectPtr /*obj*/, + MemoryAllocator* /*allocator*/) { + assert(false); +} + +static size_t SliceSize(Cache::ObjectPtr obj) { + return static_cast(obj)->size(); +} + +static Status SliceSaveTo(Cache::ObjectPtr from_obj, size_t from_offset, + size_t length, char* out) { + const Slice& slice = *static_cast(from_obj); + std::memcpy(out, slice.data() + from_offset, length); + return Status::OK(); +} + +static Status NoopCreate(const Slice& /*data*/, CompressionType /*type*/, + CacheTier /*source*/, Cache::CreateContext* /*ctx*/, + MemoryAllocator* /*allocator*/, + Cache::ObjectPtr* /*out_obj*/, + size_t* /*out_charge*/) { + assert(false); + return Status::NotSupported(); +} + +static Cache::CacheItemHelper kBasicCacheItemHelper(CacheEntryRole::kMisc, + &NoopDelete); +} // namespace + +const Cache::CacheItemHelper kSliceCacheItemHelper{ + CacheEntryRole::kMisc, &NoopDelete, &SliceSize, + &SliceSaveTo, &NoopCreate, &kBasicCacheItemHelper, +}; + Status SecondaryCache::CreateFromString( const ConfigOptions& config_options, const std::string& value, std::shared_ptr* result) { diff --git a/cache/cache_bench_tool.cc b/cache/cache_bench_tool.cc index e33e361626..b758353485 100644 --- a/cache/cache_bench_tool.cc +++ b/cache/cache_bench_tool.cc @@ -290,7 +290,8 @@ Status SaveToFn(Cache::ObjectPtr from_obj, size_t /*from_offset*/, return Status::OK(); } -Status CreateFn(const Slice& data, Cache::CreateContext* /*context*/, +Status CreateFn(const Slice& data, CompressionType /*type*/, + CacheTier /*source*/, Cache::CreateContext* /*context*/, MemoryAllocator* /*allocator*/, Cache::ObjectPtr* out_obj, size_t* out_charge) { *out_obj = new char[data.size()]; diff --git a/cache/cache_helpers.cc b/cache/cache_helpers.cc index 22597bf6da..bceb6f3c0a 100644 --- a/cache/cache_helpers.cc +++ b/cache/cache_helpers.cc @@ -25,7 +25,8 @@ Status WarmInCache(Cache* cache, const Slice& key, const Slice& saved, assert(helper->create_cb); Cache::ObjectPtr value; size_t charge; - Status st = helper->create_cb(saved, create_context, + Status st = helper->create_cb(saved, CompressionType::kNoCompression, + CacheTier::kVolatileTier, create_context, cache->memory_allocator(), &value, &charge); if (st.ok()) { st = diff --git a/cache/charged_cache.cc b/cache/charged_cache.cc index e44288ecd6..6a21bacfbc 100644 --- a/cache/charged_cache.cc +++ b/cache/charged_cache.cc @@ -19,8 +19,10 @@ ChargedCache::ChargedCache(std::shared_ptr cache, Status ChargedCache::Insert(const Slice& key, ObjectPtr obj, const CacheItemHelper* helper, size_t charge, - Handle** handle, Priority priority) { - Status s = target_->Insert(key, obj, helper, charge, handle, priority); + Handle** handle, Priority priority, + const Slice& compressed_val, CompressionType type) { + Status s = target_->Insert(key, obj, helper, charge, handle, priority, + compressed_val, type); if (s.ok()) { // Insert may cause the cache entry eviction if the cache is full. So we // directly call the reservation manager to update the total memory used diff --git a/cache/charged_cache.h b/cache/charged_cache.h index f2eacb9edd..a59c178abe 100644 --- a/cache/charged_cache.h +++ b/cache/charged_cache.h @@ -22,9 +22,11 @@ class ChargedCache : public CacheWrapper { ChargedCache(std::shared_ptr cache, std::shared_ptr block_cache); - Status Insert(const Slice& key, ObjectPtr obj, const CacheItemHelper* helper, - size_t charge, Handle** handle = nullptr, - Priority priority = Priority::LOW) override; + Status Insert( + const Slice& key, ObjectPtr obj, const CacheItemHelper* helper, + size_t charge, Handle** handle = nullptr, + Priority priority = Priority::LOW, const Slice& compressed_val = Slice(), + CompressionType type = CompressionType::kNoCompression) override; Cache::Handle* Lookup(const Slice& key, const CacheItemHelper* helper, CreateContext* create_context, diff --git a/cache/compressed_secondary_cache.cc b/cache/compressed_secondary_cache.cc index 4a35b8d1e6..3a35945e2f 100644 --- a/cache/compressed_secondary_cache.cc +++ b/cache/compressed_secondary_cache.cc @@ -11,6 +11,7 @@ #include "memory/memory_allocator_impl.h" #include "monitoring/perf_context_imp.h" +#include "util/coding.h" #include "util/compression.h" #include "util/string_util.h" @@ -54,40 +55,65 @@ std::unique_ptr CompressedSecondaryCache::Lookup( CacheAllocationPtr* ptr{nullptr}; CacheAllocationPtr merged_value; size_t handle_value_charge{0}; + const char* data_ptr = nullptr; + CacheTier source = CacheTier::kVolatileCompressedTier; + CompressionType type = cache_options_.compression_type; if (cache_options_.enable_custom_split_merge) { CacheValueChunk* value_chunk_ptr = reinterpret_cast(handle_value); merged_value = MergeChunksIntoValue(value_chunk_ptr, handle_value_charge); ptr = &merged_value; + data_ptr = ptr->get(); } else { + uint32_t type_32 = static_cast(type); + uint32_t source_32 = static_cast(source); ptr = reinterpret_cast(handle_value); handle_value_charge = cache_->GetCharge(lru_handle); + data_ptr = ptr->get(); + data_ptr = GetVarint32Ptr(data_ptr, data_ptr + 1, + static_cast(&type_32)); + type = static_cast(type_32); + data_ptr = GetVarint32Ptr(data_ptr, data_ptr + 1, + static_cast(&source_32)); + source = static_cast(source_32); + handle_value_charge -= (data_ptr - ptr->get()); } MemoryAllocator* allocator = cache_options_.memory_allocator.get(); Status s; Cache::ObjectPtr value{nullptr}; size_t charge{0}; - if (cache_options_.compression_type == kNoCompression || - cache_options_.do_not_compress_roles.Contains(helper->role)) { - s = helper->create_cb(Slice(ptr->get(), handle_value_charge), - create_context, allocator, &value, &charge); - } else { - UncompressionContext uncompression_context(cache_options_.compression_type); - UncompressionInfo uncompression_info(uncompression_context, - UncompressionDict::GetEmptyDict(), - cache_options_.compression_type); + if (source == CacheTier::kVolatileCompressedTier) { + if (cache_options_.compression_type == kNoCompression || + cache_options_.do_not_compress_roles.Contains(helper->role)) { + s = helper->create_cb(Slice(data_ptr, handle_value_charge), + kNoCompression, CacheTier::kVolatileTier, + create_context, allocator, &value, &charge); + } else { + UncompressionContext uncompression_context( + cache_options_.compression_type); + UncompressionInfo uncompression_info(uncompression_context, + UncompressionDict::GetEmptyDict(), + cache_options_.compression_type); - size_t uncompressed_size{0}; - CacheAllocationPtr uncompressed = UncompressData( - uncompression_info, (char*)ptr->get(), handle_value_charge, - &uncompressed_size, cache_options_.compress_format_version, allocator); + size_t uncompressed_size{0}; + CacheAllocationPtr uncompressed = + UncompressData(uncompression_info, (char*)data_ptr, + handle_value_charge, &uncompressed_size, + cache_options_.compress_format_version, allocator); - if (!uncompressed) { - cache_->Release(lru_handle, /*erase_if_last_ref=*/true); - return nullptr; + if (!uncompressed) { + cache_->Release(lru_handle, /*erase_if_last_ref=*/true); + return nullptr; + } + s = helper->create_cb(Slice(uncompressed.get(), uncompressed_size), + kNoCompression, CacheTier::kVolatileTier, + create_context, allocator, &value, &charge); } - s = helper->create_cb(Slice(uncompressed.get(), uncompressed_size), + } else { + // The item was not compressed by us. Let the helper create_cb + // uncompress it + s = helper->create_cb(Slice(data_ptr, handle_value_charge), type, source, create_context, allocator, &value, &charge); } @@ -112,45 +138,56 @@ std::unique_ptr CompressedSecondaryCache::Lookup( return handle; } -Status CompressedSecondaryCache::Insert(const Slice& key, - Cache::ObjectPtr value, - const Cache::CacheItemHelper* helper, - bool force_insert) { - if (value == nullptr) { - return Status::InvalidArgument(); +bool CompressedSecondaryCache::MaybeInsertDummy(const Slice& key) { + auto internal_helper = GetHelper(cache_options_.enable_custom_split_merge); + Cache::Handle* lru_handle = cache_->Lookup(key); + if (lru_handle == nullptr) { + PERF_COUNTER_ADD(compressed_sec_cache_insert_dummy_count, 1); + // Insert a dummy handle if the handle is evicted for the first time. + cache_->Insert(key, /*obj=*/nullptr, internal_helper, /*charge=*/0) + .PermitUncheckedError(); + return true; + } else { + cache_->Release(lru_handle, /*erase_if_last_ref=*/false); } - if (disable_cache_) { + return false; +} + +Status CompressedSecondaryCache::InsertInternal( + const Slice& key, Cache::ObjectPtr value, + const Cache::CacheItemHelper* helper, CompressionType type, + CacheTier source) { + if (source != CacheTier::kVolatileCompressedTier && + cache_options_.enable_custom_split_merge) { + // We don't support custom split/merge for the tiered case return Status::OK(); } auto internal_helper = GetHelper(cache_options_.enable_custom_split_merge); - if (!force_insert) { - Cache::Handle* lru_handle = cache_->Lookup(key); - if (lru_handle == nullptr) { - PERF_COUNTER_ADD(compressed_sec_cache_insert_dummy_count, 1); - // Insert a dummy handle if the handle is evicted for the first time. - return cache_->Insert(key, /*obj=*/nullptr, internal_helper, - /*charge=*/0); - } else { - cache_->Release(lru_handle, /*erase_if_last_ref=*/false); - } - } + char header[10]; + char* payload = header; + payload = EncodeVarint32(payload, static_cast(type)); + payload = EncodeVarint32(payload, static_cast(source)); - size_t size = (*helper->size_cb)(value); + size_t header_size = payload - header; + size_t data_size = (*helper->size_cb)(value); + size_t total_size = data_size + header_size; CacheAllocationPtr ptr = - AllocateBlock(size, cache_options_.memory_allocator.get()); + AllocateBlock(total_size, cache_options_.memory_allocator.get()); + char* data_ptr = ptr.get() + header_size; - Status s = (*helper->saveto_cb)(value, 0, size, ptr.get()); + Status s = (*helper->saveto_cb)(value, 0, data_size, data_ptr); if (!s.ok()) { return s; } - Slice val(ptr.get(), size); + Slice val(data_ptr, data_size); std::string compressed_val; if (cache_options_.compression_type != kNoCompression && + type == kNoCompression && !cache_options_.do_not_compress_roles.Contains(helper->role)) { - PERF_COUNTER_ADD(compressed_sec_cache_uncompressed_bytes, size); + PERF_COUNTER_ADD(compressed_sec_cache_uncompressed_bytes, data_size); CompressionOptions compression_opts; CompressionContext compression_context(cache_options_.compression_type, compression_opts); @@ -168,12 +205,14 @@ Status CompressedSecondaryCache::Insert(const Slice& key, } val = Slice(compressed_val); - size = compressed_val.size(); - PERF_COUNTER_ADD(compressed_sec_cache_compressed_bytes, size); + data_size = compressed_val.size(); + total_size = header_size + data_size; + PERF_COUNTER_ADD(compressed_sec_cache_compressed_bytes, data_size); if (!cache_options_.enable_custom_split_merge) { - ptr = AllocateBlock(size, cache_options_.memory_allocator.get()); - memcpy(ptr.get(), compressed_val.data(), size); + ptr = AllocateBlock(total_size, cache_options_.memory_allocator.get()); + data_ptr = ptr.get() + header_size; + memcpy(data_ptr, compressed_val.data(), data_size); } } @@ -184,11 +223,45 @@ Status CompressedSecondaryCache::Insert(const Slice& key, SplitValueIntoChunks(val, cache_options_.compression_type, charge); return cache_->Insert(key, value_chunks_head, internal_helper, charge); } else { + std::memcpy(ptr.get(), header, header_size); CacheAllocationPtr* buf = new CacheAllocationPtr(std::move(ptr)); - return cache_->Insert(key, buf, internal_helper, size); + return cache_->Insert(key, buf, internal_helper, total_size); } } +Status CompressedSecondaryCache::Insert(const Slice& key, + Cache::ObjectPtr value, + const Cache::CacheItemHelper* helper, + bool force_insert) { + if (value == nullptr) { + return Status::InvalidArgument(); + } + + if (!force_insert && MaybeInsertDummy(key)) { + return Status::OK(); + } + + return InsertInternal(key, value, helper, kNoCompression, + CacheTier::kVolatileCompressedTier); +} + +Status CompressedSecondaryCache::InsertSaved( + const Slice& key, const Slice& saved, CompressionType type = kNoCompression, + CacheTier source = CacheTier::kVolatileTier) { + if (type == kNoCompression) { + return Status::OK(); + } + + auto slice_helper = &kSliceCacheItemHelper; + if (MaybeInsertDummy(key)) { + return Status::OK(); + } + + return InsertInternal( + key, static_cast(const_cast(&saved)), + slice_helper, type, source); +} + void CompressedSecondaryCache::Erase(const Slice& key) { cache_->Erase(key); } Status CompressedSecondaryCache::SetCapacity(size_t capacity) { diff --git a/cache/compressed_secondary_cache.h b/cache/compressed_secondary_cache.h index 773a25bf9c..32e6fd0df9 100644 --- a/cache/compressed_secondary_cache.h +++ b/cache/compressed_secondary_cache.h @@ -80,6 +80,9 @@ class CompressedSecondaryCache : public SecondaryCache { const Cache::CacheItemHelper* helper, bool force_insert) override; + Status InsertSaved(const Slice& key, const Slice& saved, CompressionType type, + CacheTier source) override; + std::unique_ptr Lookup( const Slice& key, const Cache::CacheItemHelper* helper, Cache::CreateContext* create_context, bool /*wait*/, bool advise_erase, @@ -130,6 +133,12 @@ class CompressedSecondaryCache : public SecondaryCache { CacheAllocationPtr MergeChunksIntoValue(const void* chunks_head, size_t& charge); + bool MaybeInsertDummy(const Slice& key); + + Status InsertInternal(const Slice& key, Cache::ObjectPtr value, + const Cache::CacheItemHelper* helper, + CompressionType type, CacheTier source); + // TODO: clean up to use cleaner interfaces in typed_cache.h const Cache::CacheItemHelper* GetHelper(bool enable_custom_split_merge) const; std::shared_ptr cache_; diff --git a/cache/compressed_secondary_cache_test.cc b/cache/compressed_secondary_cache_test.cc index 54727a2fc5..eb31738f70 100644 --- a/cache/compressed_secondary_cache_test.cc +++ b/cache/compressed_secondary_cache_test.cc @@ -992,7 +992,7 @@ class CompressedSecCacheTestWithTiered /*_capacity=*/70 << 20, /*_estimated_entry_charge=*/256 << 10, /*_num_shard_bits=*/0); - TieredVolatileCacheOptions opts; + TieredCacheOptions opts; lru_opts.capacity = 70 << 20; lru_opts.num_shard_bits = 0; lru_opts.high_pri_pool_ratio = 0; @@ -1006,7 +1006,7 @@ class CompressedSecCacheTestWithTiered ; opts.comp_cache_opts.capacity = 30 << 20; opts.comp_cache_opts.num_shard_bits = 0; - cache_ = NewTieredVolatileCache(opts); + cache_ = NewTieredCache(opts); cache_res_mgr_ = std::make_shared>( cache_); diff --git a/cache/lru_cache_test.cc b/cache/lru_cache_test.cc index 047f5b80b1..27fd5cc854 100644 --- a/cache/lru_cache_test.cc +++ b/cache/lru_cache_test.cc @@ -983,13 +983,14 @@ class TestSecondaryCache : public SecondaryCache { using ResultMap = std::unordered_map; - explicit TestSecondaryCache(size_t capacity) + explicit TestSecondaryCache(size_t capacity, bool insert_saved = false) : cache_(NewLRUCache(capacity, 0, false, 0.5 /* high_pri_pool_ratio */, nullptr, kDefaultToAdaptiveMutex, kDontChargeCacheMetadata)), num_inserts_(0), num_lookups_(0), - inject_failure_(false) {} + inject_failure_(false), + insert_saved_(insert_saved) {} const char* Name() const override { return "TestSecondaryCache"; } @@ -1020,6 +1021,17 @@ class TestSecondaryCache : public SecondaryCache { return cache_.Insert(key, buf, size); } + Status InsertSaved(const Slice& key, const Slice& saved, + CompressionType /*type*/ = kNoCompression, + CacheTier /*source*/ = CacheTier::kVolatileTier) override { + if (insert_saved_) { + return Insert(key, const_cast(&saved), &kSliceCacheItemHelper, + /*force_insert=*/true); + } else { + return Status::OK(); + } + } + std::unique_ptr Lookup( const Slice& key, const Cache::CacheItemHelper* helper, Cache::CreateContext* create_context, bool /*wait*/, @@ -1048,7 +1060,8 @@ class TestSecondaryCache : public SecondaryCache { char* ptr = cache_.Value(handle); size_t size = DecodeFixed64(ptr); ptr += sizeof(uint64_t); - s = helper->create_cb(Slice(ptr, size), create_context, + s = helper->create_cb(Slice(ptr, size), kNoCompression, + CacheTier::kVolatileTier, create_context, /*alloc*/ nullptr, &value, &charge); } if (s.ok()) { @@ -1137,6 +1150,7 @@ class TestSecondaryCache : public SecondaryCache { uint32_t num_inserts_; uint32_t num_lookups_; bool inject_failure_; + bool insert_saved_; std::string ckey_prefix_; ResultMap result_map_; }; @@ -1167,7 +1181,7 @@ INSTANTIATE_TEST_CASE_P(DBSecondaryCacheTest, DBSecondaryCacheTest, TEST_P(BasicSecondaryCacheTest, BasicTest) { std::shared_ptr secondary_cache = - std::make_shared(4096); + std::make_shared(4096, true); std::shared_ptr cache = NewCache(1024 /* capacity */, 0 /* num_shard_bits */, false /* strict_capacity_limit */, secondary_cache); @@ -1224,7 +1238,7 @@ TEST_P(BasicSecondaryCacheTest, BasicTest) { TEST_P(BasicSecondaryCacheTest, StatsTest) { std::shared_ptr secondary_cache = - std::make_shared(4096); + std::make_shared(4096, true); std::shared_ptr cache = NewCache(1024 /* capacity */, 0 /* num_shard_bits */, false /* strict_capacity_limit */, secondary_cache); @@ -1278,7 +1292,7 @@ TEST_P(BasicSecondaryCacheTest, StatsTest) { TEST_P(BasicSecondaryCacheTest, BasicFailTest) { std::shared_ptr secondary_cache = - std::make_shared(2048); + std::make_shared(2048, true); std::shared_ptr cache = NewCache(1024 /* capacity */, 0 /* num_shard_bits */, false /* strict_capacity_limit */, secondary_cache); @@ -1320,7 +1334,7 @@ TEST_P(BasicSecondaryCacheTest, BasicFailTest) { TEST_P(BasicSecondaryCacheTest, SaveFailTest) { std::shared_ptr secondary_cache = - std::make_shared(2048); + std::make_shared(2048, true); std::shared_ptr cache = NewCache(1024 /* capacity */, 0 /* num_shard_bits */, false /* strict_capacity_limit */, secondary_cache); @@ -1361,7 +1375,7 @@ TEST_P(BasicSecondaryCacheTest, SaveFailTest) { TEST_P(BasicSecondaryCacheTest, CreateFailTest) { std::shared_ptr secondary_cache = - std::make_shared(2048); + std::make_shared(2048, true); std::shared_ptr cache = NewCache(1024 /* capacity */, 0 /* num_shard_bits */, false /* strict_capacity_limit */, secondary_cache); @@ -1402,7 +1416,7 @@ TEST_P(BasicSecondaryCacheTest, CreateFailTest) { TEST_P(BasicSecondaryCacheTest, FullCapacityTest) { for (bool strict_capacity_limit : {false, true}) { std::shared_ptr secondary_cache = - std::make_shared(2048); + std::make_shared(2048, true); std::shared_ptr cache = NewCache(1024 /* capacity */, 0 /* num_shard_bits */, strict_capacity_limit, secondary_cache); @@ -2021,8 +2035,9 @@ class CacheWithStats : public CacheWrapper { Status Insert(const Slice& key, Cache::ObjectPtr value, const CacheItemHelper* helper, size_t charge, - Handle** handle = nullptr, - Priority priority = Priority::LOW) override { + Handle** handle = nullptr, Priority priority = Priority::LOW, + const Slice& /*compressed*/ = Slice(), + CompressionType /*type*/ = kNoCompression) override { insert_count_++; return target_->Insert(key, value, helper, charge, handle, priority); } @@ -2115,7 +2130,7 @@ TEST_P(DBSecondaryCacheTest, LRUCacheDumpLoadBasic) { // we have a new cache it is empty, then, before we do the Get, we do the // dumpload std::shared_ptr secondary_cache = - std::make_shared(2048 * 1024); + std::make_shared(2048 * 1024, true); // This time with secondary cache base_cache = NewCache(1024 * 1024 /* capacity */, 0 /* num_shard_bits */, false /* strict_capacity_limit */, secondary_cache); @@ -2271,7 +2286,7 @@ TEST_P(DBSecondaryCacheTest, LRUCacheDumpLoadWithFilter) { // we have a new cache it is empty, then, before we do the Get, we do the // dumpload std::shared_ptr secondary_cache = - std::make_shared(2048 * 1024); + std::make_shared(2048 * 1024, true); // This time with secondary_cache base_cache = NewCache(1024 * 1024 /* capacity */, 0 /* num_shard_bits */, false /* strict_capacity_limit */, secondary_cache); diff --git a/cache/secondary_cache.cc b/cache/secondary_cache.cc index 5fecc0a6e7..4439869f19 100644 --- a/cache/secondary_cache.cc +++ b/cache/secondary_cache.cc @@ -9,37 +9,4 @@ namespace ROCKSDB_NAMESPACE { -namespace { - -void NoopDelete(Cache::ObjectPtr, MemoryAllocator*) {} - -size_t SliceSize(Cache::ObjectPtr obj) { - return static_cast(obj)->size(); -} - -Status SliceSaveTo(Cache::ObjectPtr from_obj, size_t from_offset, size_t length, - char* out) { - const Slice& slice = *static_cast(from_obj); - std::memcpy(out, slice.data() + from_offset, length); - return Status::OK(); -} - -Status FailCreate(const Slice&, Cache::CreateContext*, MemoryAllocator*, - Cache::ObjectPtr*, size_t*) { - return Status::NotSupported("Only for dumping data into SecondaryCache"); -} - -} // namespace - -Status SecondaryCache::InsertSaved(const Slice& key, const Slice& saved) { - static Cache::CacheItemHelper helper_no_secondary{CacheEntryRole::kMisc, - &NoopDelete}; - static Cache::CacheItemHelper helper{ - CacheEntryRole::kMisc, &NoopDelete, &SliceSize, - &SliceSaveTo, &FailCreate, &helper_no_secondary}; - // NOTE: depends on Insert() being synchronous, not keeping pointer `&saved` - return Insert(key, const_cast(&saved), &helper, - /*force_insert=*/true); -} - } // namespace ROCKSDB_NAMESPACE diff --git a/cache/secondary_cache_adapter.cc b/cache/secondary_cache_adapter.cc index 7f5968bb7b..dba68e8d2f 100644 --- a/cache/secondary_cache_adapter.cc +++ b/cache/secondary_cache_adapter.cc @@ -5,6 +5,7 @@ #include "cache/secondary_cache_adapter.h" +#include "cache/tiered_secondary_cache.h" #include "monitoring/perf_context_imp.h" #include "util/cast_util.h" @@ -111,7 +112,7 @@ CacheWithSecondaryAdapter::~CacheWithSecondaryAdapter() { size_t sec_capacity = 0; Status s = secondary_cache_->GetCapacity(sec_capacity); assert(s.ok()); - assert(pri_cache_res_->GetTotalReservedCacheSize() == sec_capacity); + assert(pri_cache_res_->GetTotalMemoryUsed() == sec_capacity); } #endif // NDEBUG } @@ -119,7 +120,8 @@ CacheWithSecondaryAdapter::~CacheWithSecondaryAdapter() { bool CacheWithSecondaryAdapter::EvictionHandler(const Slice& key, Handle* handle, bool was_hit) { auto helper = GetCacheItemHelper(handle); - if (helper->IsSecondaryCacheCompatible()) { + if (helper->IsSecondaryCacheCompatible() && + adm_policy_ != TieredAdmissionPolicy::kAdmPolicyThreeQueue) { auto obj = target_->Value(handle); // Ignore dummy entry if (obj != kDummyObj) { @@ -225,7 +227,9 @@ Cache::Handle* CacheWithSecondaryAdapter::Promote( Status CacheWithSecondaryAdapter::Insert(const Slice& key, ObjectPtr value, const CacheItemHelper* helper, size_t charge, Handle** handle, - Priority priority) { + Priority priority, + const Slice& compressed_value, + CompressionType type) { Status s = target_->Insert(key, value, helper, charge, handle, priority); if (s.ok() && value == nullptr && distribute_cache_res_) { size_t sec_charge = static_cast(charge * (sec_cache_res_ratio_)); @@ -234,6 +238,12 @@ Status CacheWithSecondaryAdapter::Insert(const Slice& key, ObjectPtr value, s = pri_cache_res_->UpdateCacheReservation(sec_charge, /*increase=*/false); assert(s.ok()); } + // Warm up the secondary cache with the compressed block. The secondary + // cache may choose to ignore it based on the admission policy. + if (value != nullptr && !compressed_value.empty()) { + Status status = secondary_cache_->InsertSaved(key, compressed_value, type); + assert(status.ok()); + } return s; } @@ -411,8 +421,7 @@ const char* CacheWithSecondaryAdapter::Name() const { return target_->Name(); } -std::shared_ptr NewTieredVolatileCache( - TieredVolatileCacheOptions& opts) { +std::shared_ptr NewTieredCache(TieredCacheOptions& opts) { if (!opts.cache_opts) { return nullptr; } @@ -440,6 +449,17 @@ std::shared_ptr NewTieredVolatileCache( std::shared_ptr sec_cache; sec_cache = NewCompressedSecondaryCache(opts.comp_cache_opts); + if (opts.nvm_sec_cache) { + if (opts.adm_policy == TieredAdmissionPolicy::kAdmPolicyThreeQueue || + opts.adm_policy == TieredAdmissionPolicy::kAdmPolicyAuto) { + sec_cache = std::make_shared( + sec_cache, opts.nvm_sec_cache, + TieredAdmissionPolicy::kAdmPolicyThreeQueue); + } else { + return nullptr; + } + } + return std::make_shared( cache, sec_cache, opts.adm_policy, /*distribute_cache_res=*/true); } diff --git a/cache/secondary_cache_adapter.h b/cache/secondary_cache_adapter.h index 4ef048de58..c9b846df60 100644 --- a/cache/secondary_cache_adapter.h +++ b/cache/secondary_cache_adapter.h @@ -20,10 +20,12 @@ class CacheWithSecondaryAdapter : public CacheWrapper { ~CacheWithSecondaryAdapter() override; - Status Insert(const Slice& key, ObjectPtr value, - const CacheItemHelper* helper, size_t charge, - Handle** handle = nullptr, - Priority priority = Priority::LOW) override; + Status Insert( + const Slice& key, ObjectPtr value, const CacheItemHelper* helper, + size_t charge, Handle** handle = nullptr, + Priority priority = Priority::LOW, + const Slice& compressed_value = Slice(), + CompressionType type = CompressionType::kNoCompression) override; Handle* Lookup(const Slice& key, const CacheItemHelper* helper, CreateContext* create_context, diff --git a/cache/sharded_cache.h b/cache/sharded_cache.h index 5c42194d88..39042137f8 100644 --- a/cache/sharded_cache.h +++ b/cache/sharded_cache.h @@ -170,9 +170,12 @@ class ShardedCache : public ShardedCacheBase { [s_c_l](CacheShard* cs) { cs->SetStrictCapacityLimit(s_c_l); }); } - Status Insert(const Slice& key, ObjectPtr obj, const CacheItemHelper* helper, - size_t charge, Handle** handle = nullptr, - Priority priority = Priority::LOW) override { + Status Insert( + const Slice& key, ObjectPtr obj, const CacheItemHelper* helper, + size_t charge, Handle** handle = nullptr, + Priority priority = Priority::LOW, + const Slice& /*compressed_value*/ = Slice(), + CompressionType /*type*/ = CompressionType::kNoCompression) override { assert(helper); HashVal hash = CacheShard::ComputeHash(key, hash_seed_); auto h_out = reinterpret_cast(handle); diff --git a/cache/tiered_secondary_cache.cc b/cache/tiered_secondary_cache.cc new file mode 100644 index 0000000000..493e695722 --- /dev/null +++ b/cache/tiered_secondary_cache.cc @@ -0,0 +1,119 @@ +// Copyright (c) Meta Platforms, Inc. and affiliates. +// 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). + +#include "cache/tiered_secondary_cache.h" + +namespace ROCKSDB_NAMESPACE { + +// Creation callback for use in the lookup path. It calls the upper layer +// create_cb to create the object, and optionally calls the compressed +// secondary cache InsertSaved to save the compressed block. If +// advise_erase is set, it means the primary cache wants the block to be +// erased in the secondary cache, so we skip calling InsertSaved. +// +// For the time being, we assume that all blocks in the nvm tier belong to +// the primary block cache (i.e CacheTier::kVolatileTier). That can be changed +// if we implement demotion from the compressed secondary cache to the nvm +// cache in the future. +Status TieredSecondaryCache::MaybeInsertAndCreate( + const Slice& data, CompressionType type, CacheTier source, + Cache::CreateContext* ctx, MemoryAllocator* allocator, + Cache::ObjectPtr* out_obj, size_t* out_charge) { + TieredSecondaryCache::CreateContext* context = + static_cast(ctx); + assert(source == CacheTier::kVolatileTier); + if (!context->advise_erase && type != kNoCompression) { + // Attempt to insert into compressed secondary cache + // TODO: Don't hardcode the source + context->comp_sec_cache->InsertSaved(*context->key, data, type, source) + .PermitUncheckedError(); + } + // Primary cache will accept the object, so call its helper to create + // the object + return context->helper->create_cb(data, type, source, context->inner_ctx, + allocator, out_obj, out_charge); +} + +// The lookup first looks up in the compressed secondary cache. If its a miss, +// then the nvm cache lookup is called. The cache item helper and create +// context are wrapped in order to intercept the creation callback to make +// the decision on promoting to the compressed secondary cache. +std::unique_ptr TieredSecondaryCache::Lookup( + const Slice& key, const Cache::CacheItemHelper* helper, + Cache::CreateContext* create_context, bool wait, bool advise_erase, + bool& kept_in_sec_cache) { + bool dummy = false; + std::unique_ptr result = + target()->Lookup(key, helper, create_context, wait, advise_erase, + /*kept_in_sec_cache=*/dummy); + // We never want the item to spill back into the secondary cache + kept_in_sec_cache = true; + if (result) { + assert(result->IsReady()); + return result; + } + + // If wait is true, then we can be a bit more efficient and avoid a memory + // allocation for the CReateContext. + const Cache::CacheItemHelper* outer_helper = + TieredSecondaryCache::GetHelper(); + if (wait) { + TieredSecondaryCache::CreateContext ctx; + ctx.key = &key; + ctx.advise_erase = advise_erase; + ctx.helper = helper; + ctx.inner_ctx = create_context; + ctx.comp_sec_cache = target(); + + return nvm_sec_cache_->Lookup(key, outer_helper, &ctx, wait, advise_erase, + kept_in_sec_cache); + } + + // If wait is false, i.e its an async lookup, we have to allocate a result + // handle for tracking purposes. Embed the CreateContext inside the handle + // so we need only allocate memory once instead of twice. + std::unique_ptr handle(new ResultHandle()); + handle->ctx()->key = &key; + handle->ctx()->advise_erase = advise_erase; + handle->ctx()->helper = helper; + handle->ctx()->inner_ctx = create_context; + handle->ctx()->comp_sec_cache = target(); + handle->SetInnerHandle(nvm_sec_cache_->Lookup( + key, outer_helper, handle->ctx(), wait, advise_erase, kept_in_sec_cache)); + if (!handle->inner_handle()) { + handle.reset(); + } else { + result.reset(handle.release()); + } + + return result; +} + +// Call the nvm cache WaitAll to complete the lookups +void TieredSecondaryCache::WaitAll( + std::vector handles) { + std::vector nvm_handles; + std::vector my_handles; + nvm_handles.reserve(handles.size()); + for (auto handle : handles) { + // The handle could belong to the compressed secondary cache. Skip if + // that's the case. + if (handle->IsReady()) { + continue; + } + ResultHandle* hdl = static_cast(handle); + nvm_handles.push_back(hdl->inner_handle()); + my_handles.push_back(hdl); + } + 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()); + } +} + +} // namespace ROCKSDB_NAMESPACE diff --git a/cache/tiered_secondary_cache.h b/cache/tiered_secondary_cache.h new file mode 100644 index 0000000000..6e05364367 --- /dev/null +++ b/cache/tiered_secondary_cache.h @@ -0,0 +1,155 @@ +// Copyright (c) Meta Platforms, Inc. and affiliates. +// 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/secondary_cache.h" + +namespace ROCKSDB_NAMESPACE { + +// A SecondaryCache that implements stacking of a compressed secondary cache +// and a non-volatile (local flash) cache. It implements an admission +// policy of warming the bottommost tier (local flash) with compressed +// blocks from the SST on misses, and on hits in the bottommost tier, +// promoting to the compressed and/or primary block cache. The admission +// policies of the primary block cache and compressed secondary cache remain +// unchanged - promote on second access. There is no demotion ofablocks +// evicted from a tier. They are just discarded. +// +// In order to properly handle compressed blocks directly read from SSTs, and +// to allow writeback of blocks compressed by the compressed secondary +// cache in the future, we make use of the compression type and source +// cache tier arguments in InsertSaved. +class TieredSecondaryCache : public SecondaryCacheWrapper { + public: + TieredSecondaryCache(std::shared_ptr comp_sec_cache, + std::shared_ptr nvm_sec_cache, + TieredAdmissionPolicy adm_policy) + : SecondaryCacheWrapper(comp_sec_cache), nvm_sec_cache_(nvm_sec_cache) { +#ifndef NDEBUG + assert(adm_policy == TieredAdmissionPolicy::kAdmPolicyThreeQueue); +#else + (void)adm_policy; +#endif + } + + ~TieredSecondaryCache() override {} + + const char* Name() const override { return "TieredSecondaryCache"; } + + // This is a no-op as we currently don't allow demotion (i.e + // insertion by the upper layer) of evicted blocks. + virtual Status Insert(const Slice& /*key*/, Cache::ObjectPtr /*obj*/, + const Cache::CacheItemHelper* /*helper*/, + bool /*force_insert*/) override { + return Status::OK(); + } + + // Warm up the nvm tier directly + virtual Status InsertSaved( + const Slice& key, const Slice& saved, + CompressionType type = CompressionType::kNoCompression, + CacheTier source = CacheTier::kVolatileTier) override { + return nvm_sec_cache_->InsertSaved(key, saved, type, source); + } + + virtual std::unique_ptr Lookup( + const Slice& key, const Cache::CacheItemHelper* helper, + Cache::CreateContext* create_context, bool wait, bool advise_erase, + bool& kept_in_sec_cache) override; + + virtual void WaitAll( + std::vector handles) override; + + private: + struct CreateContext : public Cache::CreateContext { + const Slice* key; + bool advise_erase; + const Cache::CacheItemHelper* helper; + Cache::CreateContext* inner_ctx; + std::shared_ptr inner_handle; + SecondaryCache* comp_sec_cache; + }; + + class ResultHandle : public SecondaryCacheResultHandle { + public: + ~ResultHandle() override {} + + bool IsReady() override { + return !inner_handle_ || inner_handle_->IsReady(); + } + + void Wait() override { + inner_handle_->Wait(); + Complete(); + } + + size_t Size() override { return size_; } + + Cache::ObjectPtr Value() override { return value_; } + + void Complete() { + assert(IsReady()); + size_ = inner_handle_->Size(); + value_ = inner_handle_->Value(); + inner_handle_.reset(); + } + + void SetInnerHandle(std::unique_ptr&& handle) { + inner_handle_ = std::move(handle); + } + + void SetSize(size_t size) { size_ = size; } + + void SetValue(Cache::ObjectPtr val) { value_ = val; } + + CreateContext* ctx() { return &ctx_; } + + SecondaryCacheResultHandle* inner_handle() { return inner_handle_.get(); } + + private: + std::unique_ptr inner_handle_; + CreateContext ctx_; + size_t size_; + Cache::ObjectPtr value_; + }; + + static void NoopDelete(Cache::ObjectPtr /*obj*/, + MemoryAllocator* /*allocator*/) { + assert(false); + } + static size_t ZeroSize(Cache::ObjectPtr /*obj*/) { + assert(false); + return 0; + } + static Status NoopSaveTo(Cache::ObjectPtr /*from_obj*/, + size_t /*from_offset*/, size_t /*length*/, + char* /*out_buf*/) { + assert(false); + return Status::OK(); + } + static Status MaybeInsertAndCreate(const Slice& data, CompressionType type, + CacheTier source, + Cache::CreateContext* ctx, + MemoryAllocator* allocator, + Cache::ObjectPtr* out_obj, + size_t* out_charge); + + static const Cache::CacheItemHelper* GetHelper() { + const static Cache::CacheItemHelper basic_helper(CacheEntryRole::kMisc, + &NoopDelete); + const static Cache::CacheItemHelper maybe_insert_and_create_helper{ + CacheEntryRole::kMisc, &NoopDelete, &ZeroSize, + &NoopSaveTo, &MaybeInsertAndCreate, &basic_helper, + }; + return &maybe_insert_and_create_helper; + } + + std::shared_ptr comp_sec_cache_; + std::shared_ptr nvm_sec_cache_; +}; + +} // namespace ROCKSDB_NAMESPACE diff --git a/cache/tiered_secondary_cache_test.cc b/cache/tiered_secondary_cache_test.cc new file mode 100644 index 0000000000..67b0e39553 --- /dev/null +++ b/cache/tiered_secondary_cache_test.cc @@ -0,0 +1,642 @@ +// Copyright (c) Meta Platforms, Inc. and affiliates. +// 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). +// +#include "db/db_test_util.h" +#include "rocksdb/cache.h" +#include "rocksdb/secondary_cache.h" +#include "typed_cache.h" +#include "util/random.h" + +namespace ROCKSDB_NAMESPACE { + +class TestSecondaryCache : public SecondaryCache { + public: + explicit TestSecondaryCache(size_t capacity) + : cache_(NewLRUCache(capacity, 0, false, 0.5 /* high_pri_pool_ratio */, + nullptr, kDefaultToAdaptiveMutex, + kDontChargeCacheMetadata)), + num_insert_saved_(0), + num_hits_(0), + num_misses_(0) {} + + const char* Name() const override { return "TestSecondaryCache"; } + + Status Insert(const Slice& /*key*/, Cache::ObjectPtr /*value*/, + const Cache::CacheItemHelper* /*helper*/, + bool /*force_insert*/) override { + assert(false); + return Status::NotSupported(); + } + + Status InsertSaved(const Slice& key, const Slice& saved, + CompressionType type = kNoCompression, + CacheTier source = CacheTier::kVolatileTier) override { + CheckCacheKeyCommonPrefix(key); + size_t size; + char* buf; + Status s; + + num_insert_saved_++; + size = saved.size(); + buf = new char[size + sizeof(uint64_t) + 2 * sizeof(uint16_t)]; + EncodeFixed64(buf, size); + buf += sizeof(uint64_t); + EncodeFixed16(buf, type); + buf += sizeof(uint16_t); + EncodeFixed16(buf, (uint16_t)source); + buf += sizeof(uint16_t); + memcpy(buf, saved.data(), size); + buf -= sizeof(uint64_t) + 2 * sizeof(uint16_t); + if (!s.ok()) { + delete[] buf; + return s; + } + return cache_.Insert(key, buf, size); + } + + std::unique_ptr Lookup( + const Slice& key, const Cache::CacheItemHelper* helper, + Cache::CreateContext* create_context, bool wait, bool /*advise_erase*/, + bool& kept_in_sec_cache) override { + std::string key_str = key.ToString(); + TEST_SYNC_POINT_CALLBACK("TestSecondaryCache::Lookup", &key_str); + + std::unique_ptr secondary_handle; + kept_in_sec_cache = false; + + TypedHandle* handle = cache_.Lookup(key); + if (handle) { + num_hits_++; + Cache::ObjectPtr value = nullptr; + size_t charge = 0; + Status s; + char* ptr = cache_.Value(handle); + CompressionType type; + CacheTier source; + size_t size = DecodeFixed64(ptr); + ptr += sizeof(uint64_t); + type = static_cast(DecodeFixed16(ptr)); + ptr += sizeof(uint16_t); + source = static_cast(DecodeFixed16(ptr)); + assert(source == CacheTier::kVolatileTier); + ptr += sizeof(uint16_t); + s = helper->create_cb(Slice(ptr, size), type, source, create_context, + /*alloc*/ nullptr, &value, &charge); + if (s.ok()) { + secondary_handle.reset(new TestSecondaryCacheResultHandle( + cache_.get(), handle, value, charge, /*ready=*/wait)); + kept_in_sec_cache = true; + } else { + cache_.Release(handle); + } + } else { + num_misses_++; + } + return secondary_handle; + } + + bool SupportForceErase() const override { return false; } + + void Erase(const Slice& /*key*/) override {} + + void WaitAll(std::vector handles) override { + for (SecondaryCacheResultHandle* handle : handles) { + TestSecondaryCacheResultHandle* sec_handle = + static_cast(handle); + EXPECT_FALSE(sec_handle->IsReady()); + sec_handle->SetReady(); + } + } + + std::string GetPrintableOptions() const override { return ""; } + + uint32_t num_insert_saved() { return num_insert_saved_; } + + uint32_t num_hits() { return num_hits_; } + + uint32_t num_misses() { return num_misses_; } + + void CheckCacheKeyCommonPrefix(const Slice& key) { + Slice current_prefix(key.data(), OffsetableCacheKey::kCommonPrefixSize); + if (ckey_prefix_.empty()) { + ckey_prefix_ = current_prefix.ToString(); + } else { + EXPECT_EQ(ckey_prefix_, current_prefix.ToString()); + } + } + + private: + class TestSecondaryCacheResultHandle : public SecondaryCacheResultHandle { + public: + TestSecondaryCacheResultHandle(Cache* cache, Cache::Handle* handle, + Cache::ObjectPtr value, size_t size, + bool ready) + : cache_(cache), + handle_(handle), + value_(value), + size_(size), + is_ready_(ready) {} + + ~TestSecondaryCacheResultHandle() override { cache_->Release(handle_); } + + bool IsReady() override { return is_ready_; } + + void Wait() override {} + + Cache::ObjectPtr Value() override { + assert(is_ready_); + return value_; + } + + size_t Size() override { return Value() ? size_ : 0; } + + void SetReady() { is_ready_ = true; } + + private: + Cache* cache_; + Cache::Handle* handle_; + Cache::ObjectPtr value_; + size_t size_; + bool is_ready_; + }; + + using SharedCache = + BasicTypedSharedCacheInterface; + using TypedHandle = SharedCache::TypedHandle; + SharedCache cache_; + uint32_t num_insert_saved_; + uint32_t num_hits_; + uint32_t num_misses_; + std::string ckey_prefix_; +}; + +class DBTieredSecondaryCacheTest : public DBTestBase { + public: + DBTieredSecondaryCacheTest() + : DBTestBase("db_tiered_secondary_cache_test", /*env_do_fsync=*/true) {} + + std::shared_ptr NewCache(size_t pri_capacity, + size_t compressed_capacity, + size_t nvm_capacity) { + LRUCacheOptions lru_opts; + TieredCacheOptions opts; + lru_opts.capacity = pri_capacity; + lru_opts.num_shard_bits = 0; + lru_opts.high_pri_pool_ratio = 0; + opts.cache_opts = &lru_opts; + opts.cache_type = PrimaryCacheType::kCacheTypeLRU; + opts.adm_policy = TieredAdmissionPolicy::kAdmPolicyThreeQueue; + opts.comp_cache_opts.capacity = compressed_capacity; + opts.comp_cache_opts.num_shard_bits = 0; + nvm_sec_cache_.reset(new TestSecondaryCache(nvm_capacity)); + opts.nvm_sec_cache = nvm_sec_cache_; + cache_ = NewTieredCache(opts); + assert(cache_ != nullptr); + +#if 0 + CacheWithSecondaryAdapter* adapter_cache_ = + static_cast(cache_.get()); + TieredSecondaryCache* tiered_cache_ = + static_cast( + adapter_cache_->TEST_GetSecondaryCache()); +#endif + + return cache_; + } + + TestSecondaryCache* nvm_sec_cache() { return nvm_sec_cache_.get(); } + + private: + std::shared_ptr cache_; + std::shared_ptr nvm_sec_cache_; +}; + +// In this test, the block size is set to 4096. Each value is 1007 bytes, so +// each data block contains exactly 4 KV pairs. Metadata blocks are not +// cached, so we can accurately estimate the cache usage. +TEST_F(DBTieredSecondaryCacheTest, BasicTest) { + if (!LZ4_Supported()) { + ROCKSDB_GTEST_SKIP("This test requires LZ4 support."); + return; + } + + BlockBasedTableOptions table_options; + // We want a block cache of size 5KB, and a compressed secondary cache of + // size 5KB. However, we specify a block cache size of 256KB here in order + // to take into account the cache reservation in the block cache on + // behalf of the compressed cache. The unit of cache reservation is 256KB. + // The effective block cache capacity will be calculated as 256 + 5 = 261KB, + // and 256KB will be reserved for the compressed cache, leaving 5KB for + // the primary block cache. We only have to worry about this here because + // the cache size is so small. + table_options.block_cache = NewCache(256 * 1024, 5 * 1024, 256 * 1024); + 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)); + + // Disable paranoid_file_checks so that flush will not read back the newly + // written file + 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()); + + // The first 2 Gets, for keys 0 and 5, will load the corresponding data + // blocks as they will be cache misses. The nvm secondary cache will be + // warmed up with the compressed blocks + std::string v = Get(Key(0)); + ASSERT_EQ(1007, v.size()); + ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 1u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 1u); + + v = Get(Key(5)); + ASSERT_EQ(1007, v.size()); + ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 2u); + + // At this point, the nvm cache is warmed up with the data blocks for 0 + // and 5. The next Get will lookup the block in nvm and will be a hit. + // It will be created as a standalone entry in memory, and a placeholder + // will be inserted in the primary and compressed caches. + v = Get(Key(0)); + ASSERT_EQ(1007, v.size()); + ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_hits(), 1u); + + // For this Get, the primary and compressed only have placeholders for + // the required data block. So we will lookup the nvm cache and find the + // block there. This time, the block will be promoted to the primary + // block cache. No promotion to the compressed secondary cache happens, + // and it will retain the placeholder. + v = Get(Key(0)); + ASSERT_EQ(1007, v.size()); + ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_hits(), 2u); + + // This Get will find the data block in the primary cache. + v = Get(Key(0)); + ASSERT_EQ(1007, v.size()); + ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_hits(), 2u); + + // We repeat the sequence for key 5. This will end up evicting the block + // for 0 from the in-memory cache. + v = Get(Key(5)); + ASSERT_EQ(1007, v.size()); + ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_hits(), 3u); + + v = Get(Key(5)); + ASSERT_EQ(1007, v.size()); + ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_hits(), 4u); + + v = Get(Key(5)); + ASSERT_EQ(1007, v.size()); + ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_hits(), 4u); + + // This Get for key 0 will find the data block in nvm. Since the compressed + // cache still has the placeholder, the block (compressed) will be + // admitted. It is theh inserted into the primary as a standalone entry. + v = Get(Key(0)); + ASSERT_EQ(1007, v.size()); + ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_hits(), 5u); + + // This Get for key 0 will find the data block in the compressed secondary + // cache. + v = Get(Key(0)); + ASSERT_EQ(1007, v.size()); + ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 2u); + ASSERT_EQ(nvm_sec_cache()->num_hits(), 5u); + + Destroy(options); +} + +// This test is very similar to BasicTest, except it calls MultiGet rather +// than Get, in order to exercise the async lookup and WaitAll path. +TEST_F(DBTieredSecondaryCacheTest, BasicMultiGetTest) { + if (!LZ4_Supported()) { + ROCKSDB_GTEST_SKIP("This test requires LZ4 support."); + return; + } + + BlockBasedTableOptions table_options; + table_options.block_cache = NewCache(260 * 1024, 10 * 1024, 256 * 1024); + 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.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 keys; + std::vector 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); + + 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); + + 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); + + 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(), 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(), 6u); + + 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(), 9u); + + 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(), 12u); + + 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(), 12u); + + Destroy(options); +} + +TEST_F(DBTieredSecondaryCacheTest, WaitAllTest) { + 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); + 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.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 keys; + std::vector 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); + + 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); + + // Insert placeholders for 4 in primary and compressed + std::string val = Get(Key(4)); + + // Force placeholder 4 out of primary + keys.clear(); + values.clear(); + keys.push_back(Key(24)); + keys.push_back(Key(28)); + 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(), 10u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 10u); + ASSERT_EQ(nvm_sec_cache()->num_hits(), 1u); + + // Now read 4 again. This will create a placeholder in primary, and insert + // in compressed secondary since it already has a placeholder + val = Get(Key(4)); + + // Now read 0, 4 and 8. While 4 is already in the compressed secondary + // cache, 0 and 8 will be read asynchronously from the nvm tier. The + // WaitAll will be called for all 3 blocks. + 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(), 10u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 10u); + ASSERT_EQ(nvm_sec_cache()->num_hits(), 4u); + + 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. +TEST_F(DBTieredSecondaryCacheTest, IterateTest) { + if (!LZ4_Supported()) { + ROCKSDB_GTEST_SKIP("This test requires LZ4 support."); + return; + } + + BlockBasedTableOptions table_options; + table_options.block_cache = NewCache(250 * 1024, 10 * 1024, 256 * 1024); + 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.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()); + + ReadOptions ro; + ro.readahead_size = 256 * 1024; + auto iter = dbfull()->NewIterator(ro); + iter->SeekToFirst(); + for (int i = 0; i < 31; ++i) { + ASSERT_EQ(Key(i), iter->key().ToString()); + ASSERT_EQ(1007, iter->value().size()); + iter->Next(); + } + ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 8u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 8u); + ASSERT_EQ(nvm_sec_cache()->num_hits(), 0u); + delete iter; + + iter = dbfull()->NewIterator(ro); + iter->SeekToFirst(); + for (int i = 0; i < 31; ++i) { + ASSERT_EQ(Key(i), iter->key().ToString()); + ASSERT_EQ(1007, iter->value().size()); + iter->Next(); + } + ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 8u); + ASSERT_EQ(nvm_sec_cache()->num_misses(), 8u); + ASSERT_EQ(nvm_sec_cache()->num_hits(), 8u); + delete iter; + + Destroy(options); +} + +} // namespace ROCKSDB_NAMESPACE + +int main(int argc, char** argv) { + ROCKSDB_NAMESPACE::port::InstallStackTraceHandler(); + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/cache/typed_cache.h b/cache/typed_cache.h index e42aa4c260..125bfa0f50 100644 --- a/cache/typed_cache.h +++ b/cache/typed_cache.h @@ -234,15 +234,19 @@ class FullTypedCacheHelperFns : public BasicTypedCacheHelperFns { return Status::OK(); } - static Status Create(const Slice& data, CreateContext* context, + static Status Create(const Slice& data, CompressionType type, + CacheTier source, CreateContext* context, MemoryAllocator* allocator, ObjectPtr* out_obj, size_t* out_charge) { std::unique_ptr value = nullptr; + if (source != CacheTier::kVolatileTier) { + return Status::InvalidArgument(); + } if constexpr (sizeof(TCreateContext) > 0) { TCreateContext* tcontext = static_cast(context); - tcontext->Create(&value, out_charge, data, allocator); + tcontext->Create(&value, out_charge, data, type, allocator); } else { - TCreateContext::Create(&value, out_charge, data, allocator); + TCreateContext::Create(&value, out_charge, data, type, allocator); } *out_obj = UpCastValue(value.release()); return Status::OK(); @@ -301,13 +305,15 @@ class FullTypedCacheInterface inline Status InsertFull( const Slice& key, TValuePtr value, size_t charge, TypedHandle** handle = nullptr, Priority priority = Priority::LOW, - CacheTier lowest_used_cache_tier = CacheTier::kNonVolatileBlockTier) { + CacheTier lowest_used_cache_tier = CacheTier::kNonVolatileBlockTier, + const Slice& compressed = Slice(), + CompressionType type = CompressionType::kNoCompression) { auto untyped_handle = reinterpret_cast(handle); - auto helper = lowest_used_cache_tier == CacheTier::kNonVolatileBlockTier + auto helper = lowest_used_cache_tier > CacheTier::kVolatileTier ? GetFullHelper() : GetBasicHelper(); return this->cache_->Insert(key, UpCastValue(value), helper, charge, - untyped_handle, priority); + untyped_handle, priority, compressed, type); } // Like SecondaryCache::InsertSaved, with SecondaryCache compatibility @@ -319,9 +325,9 @@ class FullTypedCacheInterface size_t* out_charge = nullptr) { ObjectPtr value; size_t charge; - Status st = GetFullHelper()->create_cb(data, create_context, - this->cache_->memory_allocator(), - &value, &charge); + Status st = GetFullHelper()->create_cb( + data, kNoCompression, CacheTier::kVolatileTier, create_context, + this->cache_->memory_allocator(), &value, &charge); if (out_charge) { *out_charge = charge; } @@ -340,7 +346,7 @@ class FullTypedCacheInterface const Slice& key, TCreateContext* create_context = nullptr, Priority priority = Priority::LOW, Statistics* stats = nullptr, CacheTier lowest_used_cache_tier = CacheTier::kNonVolatileBlockTier) { - if (lowest_used_cache_tier == CacheTier::kNonVolatileBlockTier) { + if (lowest_used_cache_tier > CacheTier::kVolatileTier) { return reinterpret_cast(this->cache_->Lookup( key, GetFullHelper(), create_context, priority, stats)); } else { @@ -352,7 +358,7 @@ class FullTypedCacheInterface inline void StartAsyncLookupFull( TypedAsyncLookupHandle& async_handle, CacheTier lowest_used_cache_tier = CacheTier::kNonVolatileBlockTier) { - if (lowest_used_cache_tier == CacheTier::kNonVolatileBlockTier) { + if (lowest_used_cache_tier > CacheTier::kVolatileTier) { async_handle.helper = GetFullHelper(); this->cache_->StartAsyncLookup(async_handle); } else { diff --git a/db/blob/blob_contents.h b/db/blob/blob_contents.h index 15a672a0ae..40b94d51f9 100644 --- a/db/blob/blob_contents.h +++ b/db/blob/blob_contents.h @@ -46,7 +46,8 @@ class BlobContents { class BlobContentsCreator : public Cache::CreateContext { public: static void Create(std::unique_ptr* out, size_t* out_charge, - const Slice& contents, MemoryAllocator* alloc) { + const Slice& contents, CompressionType /*type*/, + MemoryAllocator* alloc) { auto raw = new BlobContents(AllocateAndCopyBlock(contents, alloc), contents.size()); out->reset(raw); diff --git a/db/blob/blob_file_reader.cc b/db/blob/blob_file_reader.cc index bdab3ae68e..0c30efbc11 100644 --- a/db/blob/blob_file_reader.cc +++ b/db/blob/blob_file_reader.cc @@ -585,7 +585,8 @@ Status BlobFileReader::UncompressBlobIfNeeded( assert(result); if (compression_type == kNoCompression) { - BlobContentsCreator::Create(result, nullptr, value_slice, allocator); + BlobContentsCreator::Create(result, nullptr, value_slice, kNoCompression, + allocator); return Status::OK(); } diff --git a/db/db_basic_test.cc b/db/db_basic_test.cc index 1cb78e62b7..3d57941499 100644 --- a/db/db_basic_test.cc +++ b/db/db_basic_test.cc @@ -3654,10 +3654,12 @@ class DBBasicTestMultiGet : public DBTestBase { Status Insert(const Slice& key, Cache::ObjectPtr value, const CacheItemHelper* helper, size_t charge, - Handle** handle = nullptr, - Priority priority = Priority::LOW) override { + Handle** handle = nullptr, Priority priority = Priority::LOW, + const Slice& compressed = Slice(), + CompressionType type = kNoCompression) override { num_inserts_++; - return target_->Insert(key, value, helper, charge, handle, priority); + return target_->Insert(key, value, helper, charge, handle, priority, + compressed, type); } Handle* Lookup(const Slice& key, const CacheItemHelper* helper, diff --git a/db/db_block_cache_test.cc b/db/db_block_cache_test.cc index 512a481499..3335626b6e 100644 --- a/db/db_block_cache_test.cc +++ b/db/db_block_cache_test.cc @@ -294,7 +294,9 @@ class ReadOnlyCacheWrapper : public CacheWrapper { Status Insert(const Slice& /*key*/, Cache::ObjectPtr /*value*/, const CacheItemHelper* /*helper*/, size_t /*charge*/, - Handle** /*handle*/, Priority /*priority*/) override { + Handle** /*handle*/, Priority /*priority*/, + const Slice& /*compressed*/, + CompressionType /*type*/) override { return Status::NotSupported(); } }; @@ -628,13 +630,15 @@ class MockCache : public LRUCache { Status Insert(const Slice& key, Cache::ObjectPtr value, const Cache::CacheItemHelper* helper, size_t charge, - Handle** handle, Priority priority) override { + Handle** handle, Priority priority, const Slice& compressed, + CompressionType type) override { if (priority == Priority::LOW) { low_pri_insert_count++; } else { high_pri_insert_count++; } - return LRUCache::Insert(key, value, helper, charge, handle, priority); + return LRUCache::Insert(key, value, helper, charge, handle, priority, + compressed, type); } }; diff --git a/db/db_test.cc b/db/db_test.cc index 714f426081..c59951d78f 100644 --- a/db/db_test.cc +++ b/db/db_test.cc @@ -7031,8 +7031,9 @@ TEST_F(DBTest, RowCache) { using CacheWrapper::CacheWrapper; const char* Name() const override { return "FailInsertionCache"; } Status Insert(const Slice&, Cache::ObjectPtr, const CacheItemHelper*, - size_t, Handle** = nullptr, - Priority = Priority::LOW) override { + size_t, Handle** = nullptr, Priority = Priority::LOW, + const Slice& /*compressed*/ = Slice(), + CompressionType /*type*/ = kNoCompression) override { return Status::MemoryLimit(); } }; diff --git a/db/db_test_util.cc b/db/db_test_util.cc index 01d934bd15..189002270c 100644 --- a/db/db_test_util.cc +++ b/db/db_test_util.cc @@ -1716,12 +1716,12 @@ TargetCacheChargeTrackingCache::TargetCacheChargeTrackingCache( cache_charge_increments_sum_(0) {} template -Status TargetCacheChargeTrackingCache::Insert(const Slice& key, - ObjectPtr value, - const CacheItemHelper* helper, - size_t charge, Handle** handle, - Priority priority) { - Status s = target_->Insert(key, value, helper, charge, handle, priority); +Status TargetCacheChargeTrackingCache::Insert( + const Slice& key, ObjectPtr value, const CacheItemHelper* helper, + size_t charge, Handle** handle, Priority priority, const Slice& compressed, + CompressionType type) { + Status s = target_->Insert(key, value, helper, charge, handle, priority, + compressed, type); if (helper == kCrmHelper) { if (last_peak_tracked_) { cache_charge_peak_ = 0; diff --git a/db/db_test_util.h b/db/db_test_util.h index 52e856cb34..1e1225812a 100644 --- a/db/db_test_util.h +++ b/db/db_test_util.h @@ -936,8 +936,9 @@ class TargetCacheChargeTrackingCache : public CacheWrapper { Status Insert(const Slice& key, ObjectPtr value, const CacheItemHelper* helper, size_t charge, - Handle** handle = nullptr, - Priority priority = Priority::LOW) override; + Handle** handle = nullptr, Priority priority = Priority::LOW, + const Slice& compressed = Slice(), + CompressionType type = kNoCompression) override; using Cache::Release; bool Release(Handle* handle, bool erase_if_last_ref = false) override; diff --git a/include/rocksdb/advanced_cache.h b/include/rocksdb/advanced_cache.h index ec4a5b296a..a5a19d3a0f 100644 --- a/include/rocksdb/advanced_cache.h +++ b/include/rocksdb/advanced_cache.h @@ -13,7 +13,9 @@ #include #include "rocksdb/cache.h" +#include "rocksdb/compression_type.h" #include "rocksdb/memory_allocator.h" +#include "rocksdb/options.h" #include "rocksdb/slice.h" #include "rocksdb/status.h" @@ -109,13 +111,18 @@ class Cache { // pointer into static data). using DeleterFn = void (*)(ObjectPtr obj, MemoryAllocator* allocator); - // The CreateCallback is takes in a buffer from the NVM cache and constructs - // an object using it. The callback doesn't have ownership of the buffer and + // The CreateCallback is takes in a buffer from the secondary cache and + // constructs an object using it. The buffer could be compressed or + // uncompressed, as indicated by the type argument. If compressed, + // the callback is responsible for uncompressing it using information + // from the context, such as compression dictionary. + // The callback doesn't have ownership of the buffer and // should copy the contents into its own buffer. The CreateContext* is // provided by Lookup and may be used to follow DB- or CF-specific settings. // In case of some error, non-OK is returned and the caller should ignore // any result in out_obj. (The implementation must clean up after itself.) - using CreateCallback = Status (*)(const Slice& data, CreateContext* context, + using CreateCallback = Status (*)(const Slice& data, CompressionType type, + CacheTier source, CreateContext* context, MemoryAllocator* allocator, ObjectPtr* out_obj, size_t* out_charge); @@ -242,12 +249,19 @@ class Cache { // the item is only inserted into the primary cache. It may // defer the insertion to the secondary cache as it sees fit. // + // Along with the object pointer, the caller may pass a Slice pointing to + // the compressed serialized data of the object. If compressed is + // non-empty, then the caller must pass the type indicating the compression + // algorithm used. The cache may, optionally, also insert the compressed + // block into one or more cache tiers. + // // When the inserted entry is no longer needed, it will be destroyed using // helper->del_cb (if non-nullptr). - virtual Status Insert(const Slice& key, ObjectPtr obj, - const CacheItemHelper* helper, size_t charge, - Handle** handle = nullptr, - Priority priority = Priority::LOW) = 0; + virtual Status Insert( + const Slice& key, ObjectPtr obj, const CacheItemHelper* helper, + size_t charge, Handle** handle = nullptr, + Priority priority = Priority::LOW, const Slice& compressed = Slice(), + CompressionType type = CompressionType::kNoCompression) = 0; // Similar to Insert, but used for creating cache entries that cannot // be found with Lookup, such as for memory charging purposes. The @@ -536,11 +550,14 @@ class CacheWrapper : public Cache { // Only function that derived class must provide // const char* Name() const override { ... } - Status Insert(const Slice& key, ObjectPtr value, - const CacheItemHelper* helper, size_t charge, - Handle** handle = nullptr, - Priority priority = Priority::LOW) override { - return target_->Insert(key, value, helper, charge, handle, priority); + Status Insert( + const Slice& key, ObjectPtr value, const CacheItemHelper* helper, + size_t charge, Handle** handle = nullptr, + Priority priority = Priority::LOW, + const Slice& compressed_value = Slice(), + CompressionType type = CompressionType::kNoCompression) override { + return target_->Insert(key, value, helper, charge, handle, priority, + compressed_value, type); } Handle* CreateStandalone(const Slice& key, ObjectPtr obj, diff --git a/include/rocksdb/advanced_options.h b/include/rocksdb/advanced_options.h index 4f481a8ee7..794d70be55 100644 --- a/include/rocksdb/advanced_options.h +++ b/include/rocksdb/advanced_options.h @@ -275,7 +275,8 @@ struct CompactionOptionsFIFO { // In the future, we may add more caching layers. enum class CacheTier : uint8_t { kVolatileTier = 0, - kNonVolatileBlockTier = 0x01, + kVolatileCompressedTier = 0x01, + kNonVolatileBlockTier = 0x02, }; enum UpdateStatus { // Return status For inplace update callback diff --git a/include/rocksdb/cache.h b/include/rocksdb/cache.h index a85595e4f1..061ebbb365 100644 --- a/include/rocksdb/cache.h +++ b/include/rocksdb/cache.h @@ -481,22 +481,33 @@ enum TieredAdmissionPolicy { // Same as kAdmPolicyPlaceholder, but also if an entry in the primary cache // was a hit, then force insert it into the compressed secondary cache kAdmPolicyAllowCacheHits, + // An admission policy for three cache tiers - primary uncompressed, + // compressed secondary, and a compressed local flash (non-volatile) cache. + // Each tier is managed as an independent queue. + kAdmPolicyThreeQueue, kAdmPolicyMax, }; +// EXPERIMENTAL +// The following feature is experimental, and the API is subject to change +// // A 2-tier cache with a primary block cache, and a compressed secondary // cache. The returned cache instance will internally allocate a primary // uncompressed cache of the specified type, and a compressed secondary // cache. Any cache memory reservations, such as WriteBufferManager // allocations costed to the block cache, will be distributed // proportionally across both the primary and secondary. -struct TieredVolatileCacheOptions { +struct TieredCacheOptions { ShardedCacheOptions* cache_opts; PrimaryCacheType cache_type; TieredAdmissionPolicy adm_policy; CompressedSecondaryCacheOptions comp_cache_opts; + // An optional secondary cache that will serve as the persistent cache + // tier. If present, compressed blocks will be written to this + // secondary cache. + std::shared_ptr nvm_sec_cache; }; -extern std::shared_ptr NewTieredVolatileCache( - TieredVolatileCacheOptions& cache_opts); +// EXPERIMENTAL +extern std::shared_ptr NewTieredCache(TieredCacheOptions& cache_opts); } // namespace ROCKSDB_NAMESPACE diff --git a/include/rocksdb/secondary_cache.h b/include/rocksdb/secondary_cache.h index 292c0ffe04..49792ca67a 100644 --- a/include/rocksdb/secondary_cache.h +++ b/include/rocksdb/secondary_cache.h @@ -11,6 +11,7 @@ #include "rocksdb/advanced_cache.h" #include "rocksdb/customizable.h" +#include "rocksdb/options.h" #include "rocksdb/slice.h" #include "rocksdb/statistics.h" #include "rocksdb/status.h" @@ -83,15 +84,19 @@ class SecondaryCache : public Customizable { bool force_insert) = 0; // Insert a value from its saved/persistable data (typically uncompressed - // block), as if generated by SaveToCallback/SizeCallback. This can be used - // in "warming up" the cache from some auxiliary source, and like Insert() - // may or may not write it to cache depending on the admission control - // policy, even if the return status is success. + // block), as if generated by SaveToCallback/SizeCallback. The data can be + // compressed, in which case the type argument should specify the + // compression algorithm used. Additionally, the source argument should + // be set to the appropriate tier that will be responsible for + // uncompressing the data. // - // The default implementation only assumes the entry helper's create_cb is - // called at Lookup() time and not Insert() time, so should work for all - // foreseeable implementations. - virtual Status InsertSaved(const Slice& key, const Slice& saved); + // This method can be used in "warming up" the cache from some auxiliary + // source, and like Insert() may or may not write it to cache depending on + // the admission control policy, even if the return status is success. + virtual Status InsertSaved( + const Slice& key, const Slice& saved, + CompressionType type = CompressionType::kNoCompression, + CacheTier source = CacheTier::kVolatileTier) = 0; // Lookup the data for the given key in this cache. The create_cb // will be used to create the object. The handle returned may not be @@ -148,4 +153,70 @@ class SecondaryCache : public Customizable { virtual Status Inflate(size_t /*increase*/) { return Status::NotSupported(); } }; +// A wrapper around a SecondaryCache object. A derived class may selectively +// override methods to implement a different behavior. +class SecondaryCacheWrapper : public SecondaryCache { + public: + explicit SecondaryCacheWrapper(std::shared_ptr target) + : target_(std::move(target)) {} + + virtual Status Insert(const Slice& key, Cache::ObjectPtr obj, + const Cache::CacheItemHelper* helper, + bool force_insert) override { + return target()->Insert(key, obj, helper, force_insert); + } + + virtual Status InsertSaved( + const Slice& key, const Slice& saved, + CompressionType type = CompressionType::kNoCompression, + CacheTier source = CacheTier::kVolatileTier) override { + return target()->InsertSaved(key, saved, type, source); + } + + virtual std::unique_ptr Lookup( + const Slice& key, const Cache::CacheItemHelper* helper, + Cache::CreateContext* create_context, bool wait, bool advise_erase, + bool& kept_in_sec_cache) override { + return target()->Lookup(key, helper, create_context, wait, advise_erase, + kept_in_sec_cache); + } + + virtual bool SupportForceErase() const override { + return target()->SupportForceErase(); + } + + virtual void Erase(const Slice& key) override { target()->Erase(key); } + + virtual void WaitAll( + std::vector handles) override { + target()->WaitAll(handles); + } + + virtual Status SetCapacity(size_t capacity) override { + return target()->SetCapacity(capacity); + } + + virtual Status GetCapacity(size_t& capacity) override { + return target()->GetCapacity(capacity); + } + + virtual Status Deflate(size_t decrease) override { + return target()->Deflate(decrease); + } + + virtual Status Inflate(size_t increase) override { + return target()->Inflate(increase); + } + + protected: + SecondaryCache* target() const { return target_.get(); } + + private: + std::shared_ptr target_; +}; + +// Useful for cache entries that just need to be copied into a +// secondary cache, such as compressed blocks +extern const Cache::CacheItemHelper kSliceCacheItemHelper; + } // namespace ROCKSDB_NAMESPACE diff --git a/options/customizable_test.cc b/options/customizable_test.cc index 125a5aabe0..0e614ed160 100644 --- a/options/customizable_test.cc +++ b/options/customizable_test.cc @@ -1234,6 +1234,10 @@ class TestSecondaryCache : public SecondaryCache { bool /*force_insert*/) override { return Status::NotSupported(); } + Status InsertSaved(const Slice& /*key*/, const Slice& /*saved*/, + CompressionType /*type*/, CacheTier /*source*/) override { + return Status::OK(); + } std::unique_ptr Lookup( const Slice& /*key*/, const Cache::CacheItemHelper* /*helper*/, Cache::CreateContext* /*create_context*/, bool /*wait*/, diff --git a/src.mk b/src.mk index 2992f5c4a6..edc8bdef5e 100644 --- a/src.mk +++ b/src.mk @@ -12,6 +12,7 @@ LIB_SOURCES = \ cache/secondary_cache.cc \ cache/secondary_cache_adapter.cc \ cache/sharded_cache.cc \ + cache/tiered_secondary_cache.cc \ db/arena_wrapped_db_iter.cc \ db/blob/blob_contents.cc \ db/blob/blob_fetcher.cc \ @@ -432,8 +433,9 @@ BENCH_MAIN_SOURCES = \ TEST_MAIN_SOURCES = \ cache/cache_test.cc \ cache/cache_reservation_manager_test.cc \ - cache/lru_cache_test.cc \ cache/compressed_secondary_cache_test.cc \ + cache/lru_cache_test.cc \ + cache/tiered_secondary_cache_test.cc \ db/blob/blob_counting_iterator_test.cc \ db/blob/blob_file_addition_test.cc \ db/blob/blob_file_builder_test.cc \ diff --git a/table/block_based/block_based_table_builder.cc b/table/block_based/block_based_table_builder.cc index 051f9d87b4..cc4f17413b 100644 --- a/table/block_based/block_based_table_builder.cc +++ b/table/block_based/block_based_table_builder.cc @@ -488,7 +488,7 @@ struct BlockBasedTableBuilder::Rep { flush_block_policy( table_options.flush_block_policy_factory->NewFlushBlockPolicy( table_options, data_block)), - create_context(&table_options, ioptions.stats, + create_context(&table_options, &ioptions, ioptions.stats, compression_type == kZSTD || compression_type == kZSTDNotFinalCompression, tbo.moptions.block_protection_bytes_per_key, diff --git a/table/block_based/block_based_table_reader.cc b/table/block_based/block_based_table_reader.cc index 528b87bb95..c53aa0fb9b 100644 --- a/table/block_based/block_based_table_reader.cc +++ b/table/block_based/block_based_table_reader.cc @@ -683,7 +683,7 @@ Status BlockBasedTable::Open( rep->table_properties->compression_name == CompressionTypeToString(kZSTDNotFinalCompression)); rep->create_context = BlockCreateContext( - &rep->table_options, rep->ioptions.stats, + &rep->table_options, &rep->ioptions, rep->ioptions.stats, blocks_definitely_zstd_compressed, block_protection_bytes_per_key, rep->internal_comparator.user_comparator(), rep->index_value_is_full, rep->index_has_first_key); @@ -1303,8 +1303,8 @@ Cache::Priority BlockBasedTable::GetCachePriority() const { template WithBlocklikeCheck BlockBasedTable::GetDataBlockFromCache( const Slice& cache_key, BlockCacheInterface block_cache, - CachableEntry* out_parsed_block, - GetContext* get_context) const { + CachableEntry* out_parsed_block, GetContext* get_context, + const UncompressionDict* dict) const { assert(out_parsed_block); assert(out_parsed_block->IsEmpty()); @@ -1313,10 +1313,12 @@ WithBlocklikeCheck BlockBasedTable::GetDataBlockFromCache( // Lookup uncompressed cache first if (block_cache) { + BlockCreateContext create_ctx = rep_->create_context; + create_ctx.dict = dict; assert(!cache_key.empty()); auto cache_handle = block_cache.LookupFull( - cache_key, &rep_->create_context, GetCachePriority(), - statistics, rep_->ioptions.lowest_used_cache_tier); + cache_key, &create_ctx, GetCachePriority(), statistics, + rep_->ioptions.lowest_used_cache_tier); // Avoid updating metrics here if the handle is not complete yet. This // happens with MultiGet and secondary cache. So update the metrics only @@ -1343,8 +1345,9 @@ WithBlocklikeCheck BlockBasedTable::GetDataBlockFromCache( template WithBlocklikeCheck BlockBasedTable::PutDataBlockToCache( const Slice& cache_key, BlockCacheInterface block_cache, - CachableEntry* out_parsed_block, BlockContents&& block_contents, - CompressionType block_comp_type, + CachableEntry* out_parsed_block, + BlockContents&& uncompressed_block_contents, + BlockContents&& compressed_block_contents, CompressionType block_comp_type, const UncompressionDict& uncompression_dict, MemoryAllocator* memory_allocator, GetContext* get_context) const { const ImmutableOptions& ioptions = rep_->ioptions; @@ -1356,23 +1359,22 @@ WithBlocklikeCheck BlockBasedTable::PutDataBlockToCache( Statistics* statistics = ioptions.stats; std::unique_ptr block_holder; - if (block_comp_type != kNoCompression) { + if (block_comp_type != kNoCompression && + uncompressed_block_contents.data.empty()) { + assert(compressed_block_contents.data.data()); // Retrieve the uncompressed contents into a new buffer - BlockContents uncompressed_block_contents; UncompressionContext context(block_comp_type); UncompressionInfo info(context, uncompression_dict, block_comp_type); - s = UncompressBlockData(info, block_contents.data.data(), - block_contents.data.size(), + s = UncompressBlockData(info, compressed_block_contents.data.data(), + compressed_block_contents.data.size(), &uncompressed_block_contents, format_version, ioptions, memory_allocator); if (!s.ok()) { return s; } - rep_->create_context.Create(&block_holder, - std::move(uncompressed_block_contents)); - } else { - rep_->create_context.Create(&block_holder, std::move(block_contents)); } + rep_->create_context.Create(&block_holder, + std::move(uncompressed_block_contents)); // insert into uncompressed block cache if (block_cache && block_holder->own_bytes()) { @@ -1380,7 +1382,8 @@ WithBlocklikeCheck BlockBasedTable::PutDataBlockToCache( BlockCacheTypedHandle* cache_handle = nullptr; s = block_cache.InsertFull(cache_key, block_holder.get(), charge, &cache_handle, GetCachePriority(), - rep_->ioptions.lowest_used_cache_tier); + rep_->ioptions.lowest_used_cache_tier, + compressed_block_contents.data, block_comp_type); if (s.ok()) { assert(cache_handle != nullptr); @@ -1500,7 +1503,7 @@ BlockBasedTable::MaybeReadBlockAndLoadToCache( if (!contents) { if (use_block_cache_for_lookup) { s = GetDataBlockFromCache(key, block_cache, out_parsed_block, - get_context); + get_context, &uncompression_dict); // Value could still be null at this point, so check the cache handle // and update the read pattern for prefetching if (out_parsed_block->GetValue() || @@ -1531,14 +1534,26 @@ BlockBasedTable::MaybeReadBlockAndLoadToCache( TBlocklike::kBlockType != BlockType::kFilter && TBlocklike::kBlockType != BlockType::kCompressionDictionary && rep_->blocks_maybe_compressed; + // This flag, if true, tells BlockFetcher to return the uncompressed + // block when ReadBlockContents() is called. const bool do_uncompress = maybe_compressed; CompressionType contents_comp_type; // Maybe serialized or uncompressed BlockContents tmp_contents; + BlockContents uncomp_contents; + BlockContents comp_contents; if (!contents) { Histograms histogram = for_compaction ? READ_BLOCK_COMPACTION_MICROS : READ_BLOCK_GET_MICROS; StopWatch sw(rep_->ioptions.clock, statistics, histogram); + // Setting do_uncompress to false may cause an extra mempcy in the + // following cases - + // 1. Compression is enabled, but block is not actually compressed + // 2. Compressed block is in the prefetch buffer + // 3. Direct IO + // + // It would also cause a memory allocation to be used rather than + // stack if the compressed block size is < 5KB BlockFetcher block_fetcher( rep_->file.get(), prefetch_buffer, rep_->footer, ro, handle, &tmp_contents, rep_->ioptions, do_uncompress, maybe_compressed, @@ -1559,7 +1574,6 @@ BlockBasedTable::MaybeReadBlockAndLoadToCache( } contents_comp_type = block_fetcher.get_compression_type(); - contents = &tmp_contents; if (get_context) { switch (TBlocklike::kBlockType) { case BlockType::kIndex: @@ -1573,17 +1587,43 @@ BlockBasedTable::MaybeReadBlockAndLoadToCache( break; } } + if (s.ok()) { + if (do_uncompress && contents_comp_type != kNoCompression) { + comp_contents = BlockContents(block_fetcher.GetCompressedBlock()); + uncomp_contents = std::move(tmp_contents); + } else if (contents_comp_type != kNoCompression) { + // do_uncompress must be false, so output of BlockFetcher is + // compressed + comp_contents = std::move(tmp_contents); + } else { + uncomp_contents = std::move(tmp_contents); + } + + // If filling cache is allowed and a cache is configured, try to put + // the block to the cache. Do this here while block_fetcher is in + // scope, since comp_contents will be a reference to the compressed + // block in block_fetcher + s = PutDataBlockToCache( + key, block_cache, out_parsed_block, std::move(uncomp_contents), + std::move(comp_contents), contents_comp_type, uncompression_dict, + GetMemoryAllocator(rep_->table_options), get_context); + } } else { contents_comp_type = GetBlockCompressionType(*contents); - } + if (contents_comp_type != kNoCompression) { + comp_contents = std::move(*contents); + } else { + uncomp_contents = std::move(*contents); + } - if (s.ok()) { - // If filling cache is allowed and a cache is configured, try to put the - // block to the cache. - s = PutDataBlockToCache( - key, block_cache, out_parsed_block, std::move(*contents), - contents_comp_type, uncompression_dict, - GetMemoryAllocator(rep_->table_options), get_context); + if (s.ok()) { + // If filling cache is allowed and a cache is configured, try to put + // the block to the cache. + s = PutDataBlockToCache( + key, block_cache, out_parsed_block, std::move(uncomp_contents), + std::move(comp_contents), contents_comp_type, uncompression_dict, + GetMemoryAllocator(rep_->table_options), get_context); + } } } } diff --git a/table/block_based/block_based_table_reader.h b/table/block_based/block_based_table_reader.h index 6162c5889b..32c5f68523 100644 --- a/table/block_based/block_based_table_reader.h +++ b/table/block_based/block_based_table_reader.h @@ -410,7 +410,8 @@ class BlockBasedTable : public TableReader { template WithBlocklikeCheck GetDataBlockFromCache( const Slice& cache_key, BlockCacheInterface block_cache, - CachableEntry* block, GetContext* get_context) const; + CachableEntry* block, GetContext* get_context, + const UncompressionDict* dict) const; // Put a maybe compressed block to the corresponding block caches. // This method will perform decompression against block_contents if needed @@ -425,7 +426,9 @@ class BlockBasedTable : public TableReader { template WithBlocklikeCheck PutDataBlockToCache( const Slice& cache_key, BlockCacheInterface block_cache, - CachableEntry* cached_block, BlockContents&& block_contents, + CachableEntry* cached_block, + BlockContents&& uncompressed_block_contents, + BlockContents&& compressed_block_contents, CompressionType block_comp_type, const UncompressionDict& uncompression_dict, MemoryAllocator* memory_allocator, GetContext* get_context) const; diff --git a/table/block_based/block_based_table_reader_sync_and_async.h b/table/block_based/block_based_table_reader_sync_and_async.h index 8ee594db64..e7621909cc 100644 --- a/table/block_based/block_based_table_reader_sync_and_async.h +++ b/table/block_based/block_based_table_reader_sync_and_async.h @@ -402,6 +402,7 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::MultiGet) BCI block_cache{rep_->table_options.block_cache.get()}; std::array async_handles; + BlockCreateContext create_ctx = rep_->create_context; std::array cache_keys; size_t cache_lookup_count = 0; @@ -448,6 +449,9 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::MultiGet) sst_file_range.SkipKey(miter); continue; } + create_ctx.dict = uncompression_dict.GetValue() + ? uncompression_dict.GetValue() + : &UncompressionDict::GetEmptyDict(); if (v.handle.offset() == prev_offset) { // This key can reuse the previous block (later on). @@ -475,7 +479,7 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::MultiGet) GetCacheKey(rep_->base_cache_key, v.handle); async_handle.key = cache_keys[cache_lookup_count].AsSlice(); // NB: StartAsyncLookupFull populates async_handle.helper - async_handle.create_context = &rep_->create_context; + async_handle.create_context = &create_ctx; async_handle.priority = GetCachePriority(); async_handle.stats = rep_->ioptions.statistics.get(); diff --git a/table/block_based/block_cache.cc b/table/block_based/block_cache.cc index a252899d24..08f5d2158d 100644 --- a/table/block_based/block_cache.cc +++ b/table/block_based/block_cache.cc @@ -5,6 +5,8 @@ #include "table/block_based/block_cache.h" +#include "table/block_based/block_based_table_reader.h" + namespace ROCKSDB_NAMESPACE { void BlockCreateContext::Create(std::unique_ptr* parsed_out, @@ -96,7 +98,7 @@ const std::array CacheTier::kVolatileTier) { return kCacheItemFullHelperForBlockType[static_cast(block_type)]; } else { return kCacheItemBasicHelperForBlockType[static_cast(block_type)]; diff --git a/table/block_based/block_cache.h b/table/block_based/block_cache.h index 00eaface37..06ba50566e 100644 --- a/table/block_based/block_cache.h +++ b/table/block_based/block_cache.h @@ -70,24 +70,28 @@ class Block_kMetaIndex : public Block { struct BlockCreateContext : public Cache::CreateContext { BlockCreateContext() {} BlockCreateContext(const BlockBasedTableOptions* _table_options, - Statistics* _statistics, bool _using_zstd, - uint8_t _protection_bytes_per_key, + const ImmutableOptions* _ioptions, Statistics* _statistics, + bool _using_zstd, uint8_t _protection_bytes_per_key, const Comparator* _raw_ucmp, bool _index_value_is_full = false, bool _index_has_first_key = false) : table_options(_table_options), + ioptions(_ioptions), statistics(_statistics), + raw_ucmp(_raw_ucmp), using_zstd(_using_zstd), protection_bytes_per_key(_protection_bytes_per_key), - raw_ucmp(_raw_ucmp), index_value_is_full(_index_value_is_full), index_has_first_key(_index_has_first_key) {} const BlockBasedTableOptions* table_options = nullptr; + const ImmutableOptions* ioptions = nullptr; Statistics* statistics = nullptr; + const Comparator* raw_ucmp = nullptr; + const UncompressionDict* dict = nullptr; + uint32_t format_version; bool using_zstd = false; uint8_t protection_bytes_per_key = 0; - const Comparator* raw_ucmp = nullptr; bool index_value_is_full; bool index_has_first_key; @@ -95,9 +99,24 @@ struct BlockCreateContext : public Cache::CreateContext { template inline void Create(std::unique_ptr* parsed_out, size_t* charge_out, const Slice& data, - MemoryAllocator* alloc) { - Create(parsed_out, - BlockContents(AllocateAndCopyBlock(data, alloc), data.size())); + CompressionType type, MemoryAllocator* alloc) { + BlockContents uncompressed_block_contents; + if (type != CompressionType::kNoCompression) { + assert(dict != nullptr); + UncompressionContext context(type); + UncompressionInfo info(context, *dict, type); + Status s = UncompressBlockData( + info, data.data(), data.size(), &uncompressed_block_contents, + table_options->format_version, *ioptions, alloc); + if (!s.ok()) { + parsed_out->reset(); + return; + } + } else { + uncompressed_block_contents = + BlockContents(AllocateAndCopyBlock(data, alloc), data.size()); + } + Create(parsed_out, std::move(uncompressed_block_contents)); *charge_out = parsed_out->get()->ApproximateMemoryUsage(); } diff --git a/table/block_based/block_test.cc b/table/block_based/block_test.cc index 3264371c19..9082a08e9f 100644 --- a/table/block_based/block_test.cc +++ b/table/block_based/block_test.cc @@ -848,9 +848,12 @@ TEST_F(BlockPerKVChecksumTest, EmptyBlock) { Options options = Options(); BlockBasedTableOptions tbo; uint8_t protection_bytes_per_key = 8; - BlockCreateContext create_context{ - &tbo, nullptr /* statistics */, false /* using_zstd */, - protection_bytes_per_key, options.comparator}; + BlockCreateContext create_context{&tbo, + nullptr, + nullptr /* statistics */, + false /* using_zstd */, + protection_bytes_per_key, + options.comparator}; create_context.Create(&data_block, std::move(contents)); std::unique_ptr biter{data_block->NewDataIterator( options.comparator, kDisableGlobalSequenceNumber)}; @@ -885,9 +888,12 @@ TEST_F(BlockPerKVChecksumTest, InitializeProtectionInfo) { Options options = Options(); BlockBasedTableOptions tbo; uint8_t protection_bytes_per_key = 8; - BlockCreateContext create_context{ - &tbo, nullptr /* statistics */, false /* using_zstd */, - protection_bytes_per_key, options.comparator}; + BlockCreateContext create_context{&tbo, + nullptr /* ioptions */, + nullptr /* statistics */, + false /* using_zstd */, + protection_bytes_per_key, + options.comparator}; { std::string invalid_content = "1"; @@ -949,14 +955,19 @@ TEST_F(BlockPerKVChecksumTest, ApproximateMemory) { uint8_t protection_bytes_per_key = 8; BlockCreateContext with_checksum_create_context{ &tbo, + nullptr /* ioptions */, nullptr /* statistics */, false /* using_zstd */, protection_bytes_per_key, options.comparator, true /* index_value_is_full */}; - BlockCreateContext create_context{ - &tbo, nullptr /* statistics */, false /* using_zstd */, - 0, options.comparator, true /* index_value_is_full */}; + BlockCreateContext create_context{&tbo, + nullptr /* ioptions */, + nullptr /* statistics */, + false /* using_zstd */, + 0, + options.comparator, + true /* index_value_is_full */}; { std::unique_ptr data_block; @@ -1045,8 +1056,11 @@ class DataBlockKVChecksumTest std::vector &keys, std::vector &values, int num_record) { BlockBasedTableOptions tbo; - BlockCreateContext create_context{&tbo, nullptr /* statistics */, - false /* using_zstd */, GetChecksumLen(), + BlockCreateContext create_context{&tbo, + nullptr /* statistics */, + nullptr /* ioptions */, + false /* using_zstd */, + GetChecksumLen(), Options().comparator}; builder_ = std::make_unique( static_cast(GetRestartInterval()), @@ -1172,6 +1186,7 @@ class IndexBlockKVChecksumTest uint8_t protection_bytes_per_key = GetChecksumLen(); BlockCreateContext create_context{ &tbo, + nullptr /* ioptions */, nullptr /* statistics */, false /* _using_zstd */, protection_bytes_per_key, @@ -1312,9 +1327,12 @@ class MetaIndexBlockKVChecksumTest Options options = Options(); BlockBasedTableOptions tbo; uint8_t protection_bytes_per_key = GetChecksumLen(); - BlockCreateContext create_context{ - &tbo, nullptr /* statistics */, false /* using_zstd */, - protection_bytes_per_key, options.comparator}; + BlockCreateContext create_context{&tbo, + nullptr /* ioptions */, + nullptr /* statistics */, + false /* using_zstd */, + protection_bytes_per_key, + options.comparator}; builder_ = std::make_unique(static_cast(GetRestartInterval())); // add a bunch of records to a block @@ -1344,9 +1362,12 @@ TEST_P(MetaIndexBlockKVChecksumTest, ChecksumConstructionAndVerification) { Options options = Options(); BlockBasedTableOptions tbo; uint8_t protection_bytes_per_key = GetChecksumLen(); - BlockCreateContext create_context{ - &tbo, nullptr /* statistics */, false /* using_zstd */, - protection_bytes_per_key, options.comparator}; + BlockCreateContext create_context{&tbo, + nullptr /* ioptions */, + nullptr /* statistics */, + false /* using_zstd */, + protection_bytes_per_key, + options.comparator}; std::vector num_restart_intervals = {1, 16}; for (const auto num_restart_interval : num_restart_intervals) { const int kNumRecords = num_restart_interval * GetRestartInterval(); @@ -1680,4 +1701,4 @@ int main(int argc, char **argv) { ROCKSDB_NAMESPACE::port::InstallStackTraceHandler(); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); -} \ No newline at end of file +} diff --git a/table/block_fetcher.cc b/table/block_fetcher.cc index 34d3e23e9a..257a1a42ea 100644 --- a/table/block_fetcher.cc +++ b/table/block_fetcher.cc @@ -336,9 +336,11 @@ IOStatus BlockFetcher::ReadBlockContents() { #ifndef NDEBUG num_heap_buf_memcpy_++; #endif - compression_type_ = kNoCompression; + // Save the compressed block without trailer + slice_ = Slice(slice_.data(), block_size_); } else { GetBlockContents(); + slice_ = Slice(); } InsertUncompressedBlockToPersistentCacheIfNeeded(); @@ -387,7 +389,6 @@ IOStatus BlockFetcher::ReadAsyncBlockContents() { #ifndef NDEBUG num_heap_buf_memcpy_++; #endif - compression_type_ = kNoCompression; } else { GetBlockContents(); } diff --git a/table/block_fetcher.h b/table/block_fetcher.h index da6c352d0a..e5a51e3eb2 100644 --- a/table/block_fetcher.h +++ b/table/block_fetcher.h @@ -79,6 +79,10 @@ class BlockFetcher { inline size_t GetBlockSizeWithTrailer() const { return block_size_with_trailer_; } + inline Slice& GetCompressedBlock() { + assert(compression_type_ != kNoCompression); + return slice_; + } #ifndef NDEBUG int TEST_GetNumStackBufMemcpy() const { return num_stack_buf_memcpy_; } diff --git a/table/block_fetcher_test.cc b/table/block_fetcher_test.cc index 18109811d5..d738fa3df8 100644 --- a/table/block_fetcher_test.cc +++ b/table/block_fetcher_test.cc @@ -299,7 +299,7 @@ class BlockFetcherTest : public testing::Test { MemoryAllocator* heap_buf_allocator, MemoryAllocator* compressed_buf_allocator, BlockContents* contents, MemcpyStats* stats, - CompressionType* compresstion_type) { + CompressionType* compression_type) { ImmutableOptions ioptions(options_); ReadOptions roptions; PersistentCacheOptions persistent_cache_options; @@ -318,7 +318,11 @@ class BlockFetcherTest : public testing::Test { stats->num_compressed_buf_memcpy = fetcher->TEST_GetNumCompressedBufMemcpy(); - *compresstion_type = fetcher->get_compression_type(); + if (do_uncompress) { + *compression_type = kNoCompression; + } else { + *compression_type = fetcher->get_compression_type(); + } } // NOTE: expected_compression_type is the expected compression diff --git a/test_util/secondary_cache_test_util.cc b/test_util/secondary_cache_test_util.cc index 1c62dc4ad7..6f0bd38494 100644 --- a/test_util/secondary_cache_test_util.cc +++ b/test_util/secondary_cache_test_util.cc @@ -37,7 +37,8 @@ Status SaveToCallbackFail(Cache::ObjectPtr /*obj*/, size_t /*offset*/, return Status::NotSupported(); } -Status CreateCallback(const Slice& data, Cache::CreateContext* context, +Status CreateCallback(const Slice& data, CompressionType /*type*/, + CacheTier /*source*/, Cache::CreateContext* context, MemoryAllocator* /*allocator*/, Cache::ObjectPtr* out_obj, size_t* out_charge) { auto t = static_cast(context); diff --git a/tools/db_bench_tool.cc b/tools/db_bench_tool.cc index f6662c09fc..340a8a3a1e 100644 --- a/tools/db_bench_tool.cc +++ b/tools/db_bench_tool.cc @@ -3060,12 +3060,12 @@ class Benchmark { FLAGS_cache_numshardbits); opts.hash_seed = GetCacheHashSeed(); if (use_tiered_cache) { - TieredVolatileCacheOptions tiered_opts; + TieredCacheOptions tiered_opts; opts.capacity += secondary_cache_opts.capacity; tiered_opts.cache_type = PrimaryCacheType::kCacheTypeHCC; tiered_opts.cache_opts = &opts; tiered_opts.comp_cache_opts = secondary_cache_opts; - return NewTieredVolatileCache(tiered_opts); + return NewTieredCache(tiered_opts); } else { return opts.MakeSharedCache(); } @@ -3093,12 +3093,12 @@ class Benchmark { } if (use_tiered_cache) { - TieredVolatileCacheOptions tiered_opts; + TieredCacheOptions tiered_opts; opts.capacity += secondary_cache_opts.capacity; tiered_opts.cache_type = PrimaryCacheType::kCacheTypeLRU; tiered_opts.cache_opts = &opts; tiered_opts.comp_cache_opts = secondary_cache_opts; - return NewTieredVolatileCache(tiered_opts); + return NewTieredCache(tiered_opts); } else { return opts.MakeSharedCache(); } diff --git a/unreleased_history/new_features/secondary_cache_stacking.md b/unreleased_history/new_features/secondary_cache_stacking.md new file mode 100644 index 0000000000..05e1bb2420 --- /dev/null +++ b/unreleased_history/new_features/secondary_cache_stacking.md @@ -0,0 +1 @@ +RocksDB now allows the block cache to be stacked on top of a compressed secondary cache and a non-volatile secondary cache, thus creating a three-tier cache. To set it up, use the `NewTieredCache()` API in rocksdb/cache.h.. diff --git a/unreleased_history/public_api_changes/new_tiered_cache.md b/unreleased_history/public_api_changes/new_tiered_cache.md new file mode 100644 index 0000000000..935e6a7ad4 --- /dev/null +++ b/unreleased_history/public_api_changes/new_tiered_cache.md @@ -0,0 +1 @@ +The `NewTieredVolatileCache()` API in rocksdb/cache.h has been renamed to `NewTieredCache()`. diff --git a/utilities/fault_injection_secondary_cache.h b/utilities/fault_injection_secondary_cache.h index 60488dcfb0..dd73ac1563 100644 --- a/utilities/fault_injection_secondary_cache.h +++ b/utilities/fault_injection_secondary_cache.h @@ -35,6 +35,11 @@ class FaultInjectionSecondaryCache : public SecondaryCache { const Cache::CacheItemHelper* helper, bool force_insert) override; + Status InsertSaved(const Slice& /*key*/, const Slice& /*saved*/, + CompressionType /*type*/, CacheTier /*source*/) override { + return Status::OK(); + } + std::unique_ptr Lookup( const Slice& key, const Cache::CacheItemHelper* helper, Cache::CreateContext* create_context, bool wait, bool advise_erase, diff --git a/utilities/simulator_cache/sim_cache.cc b/utilities/simulator_cache/sim_cache.cc index d58c3b34f1..ff9d52dca9 100644 --- a/utilities/simulator_cache/sim_cache.cc +++ b/utilities/simulator_cache/sim_cache.cc @@ -169,7 +169,8 @@ class SimCacheImpl : public SimCache { Status Insert(const Slice& key, Cache::ObjectPtr value, const CacheItemHelper* helper, size_t charge, Handle** handle, - Priority priority) override { + Priority priority, const Slice& compressed = {}, + CompressionType type = kNoCompression) 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 @@ -178,8 +179,9 @@ class SimCacheImpl : public SimCache { Handle* h = key_only_cache_->Lookup(key); if (h == nullptr) { // TODO: Check for error here? - auto s = key_only_cache_->Insert(key, nullptr, &kNoopCacheItemHelper, - charge, nullptr, priority); + auto s = + key_only_cache_->Insert(key, nullptr, &kNoopCacheItemHelper, charge, + nullptr, priority, compressed, type); s.PermitUncheckedError(); } else { key_only_cache_->Release(h); @@ -189,7 +191,8 @@ class SimCacheImpl : public SimCache { if (!target_) { return Status::OK(); } - return target_->Insert(key, value, helper, charge, handle, priority); + return target_->Insert(key, value, helper, charge, handle, priority, + compressed, type); } Handle* Lookup(const Slice& key, const CacheItemHelper* helper,