gian/ceph-mirror
1
0
Fork 0
mirror of https://github.com/ceph/ceph synced 2026-08-02 07:03:18 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Merge 021fdd3dc0 into eed82023c5

Browse Source
This commit is contained in:
Kefu Chai 2026-08-02 03:55:54 +00:00 committed by GitHub
commit cf412f3f55
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 247 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
src/kv/RocksDBStore.cc
View File

@ -1,6 +1,9 @@
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
// vim: ts=8 sw=2 sts=2 expandtab
#include <cstdarg>
#include <cstdio>
#include <ctime>
#include <filesystem>
#include <map>
#include <memory>
@ -11,6 +14,7 @@
#include <unordered_map>
#include <errno.h>
#include <unistd.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/stat.h>
@ -26,7 +30,6 @@
#include "rocksdb/merge_operator.h"
#include "common/version.h"
#include "rocksdb/util/stderr_logger.h"
#include "common/Clock.h" // for ceph_clock_now()
#include "common/perf_counters.h"
@ -71,6 +74,20 @@ using ceph::bufferlist;
using ceph::bufferptr;
using ceph::Formatter;
#if ROCKSDB_MAJOR >= 8
/*
* ConfigOptions defaults input_strings_escaped to true, where the pre-v8
* overloads taking the flag directly defaulted it to false, hence spelling
* it out.
*/
static rocksdb::ConfigOptions make_config_options()
{
rocksdb::ConfigOptions config_options;
config_options.input_strings_escaped = false;
return config_options;
}
#endif
static const char* sharding_def_dir = "sharding";
static const char* sharding_def_file = "sharding/def";
static const char* sharding_recreate = "sharding/recreate_columns";
@ -943,7 +960,12 @@ int RocksDBStore::update_column_family_options(const std::string& base_name,
<< " options=" << more_options << dendl;
return r;
}
#if ROCKSDB_MAJOR >= 8
status = rocksdb::GetColumnFamilyOptionsFromMap(
make_config_options(), *cf_opt, options_map, cf_opt);
#else
status = rocksdb::GetColumnFamilyOptionsFromMap(*cf_opt, options_map, cf_opt);
#endif
if (!status.ok()) {
dout(5) << __func__ << " invalid column family optionsp; column family="
<< base_name << " options=" << more_options << dendl;
@ -1008,7 +1030,12 @@ int RocksDBStore::apply_block_cache_options(const std::string& column_name,
}
rocksdb::BlockBasedTableOptions column_bbt_opts;
#if ROCKSDB_MAJOR >= 8
status = GetBlockBasedTableOptionsFromMap(
make_config_options(), bbt_opts, cache_options_map, &column_bbt_opts);
#else
status = GetBlockBasedTableOptionsFromMap(bbt_opts, cache_options_map, &column_bbt_opts);
#endif
if (!status.ok()) {
dout(5) << __func__ << " invalid block cache options; column=" << column_name
<< " options=" << block_cache_opt << dendl;
@ -2097,12 +2124,51 @@ KeyValueDB::BackupStats RocksDBStore::backup(const std::string &path)
return rv;
}
namespace {
// stderr logger for the backup engine's info_log
class StderrLogger : public rocksdb::Logger {
public:
explicit StderrLogger(
rocksdb::InfoLogLevel log_level = rocksdb::InfoLogLevel::INFO_LEVEL)
: rocksdb::Logger(log_level) {}
// keep the Logv(InfoLogLevel, ...) overload visible
using rocksdb::Logger::Logv;
void Logv(const char* format, va_list ap) override {
// rocksdb::Logger requires that nothing propagate out of here, as RocksDB
// is not exception-safe, so format into a fixed buffer instead of
// allocating, at the price of truncating a very long message. Prefix each
// line with the time and thread id, and emit it with a single fprintf(),
// so that lines from different threads do not interleave.
timeval now;
gettimeofday(&now, nullptr);
struct tm t;
localtime_r(&now.tv_sec, &t);
char buf[1024];
int prefix_len = snprintf(
buf, sizeof(buf), "%04d/%02d/%02d-%02d:%02d:%02d.%06d %llx ",
t.tm_year + 1900, t.tm_mon + 1, t.tm_mday,
t.tm_hour, t.tm_min, t.tm_sec, static_cast<int>(now.tv_usec),
static_cast<unsigned long long>(
rocksdb::Env::Default()->GetThreadID()));
if (prefix_len < 0) {
prefix_len = 0;
buf[0] = '\0';
}
if (static_cast<size_t>(prefix_len) < sizeof(buf)) {
vsnprintf(buf + prefix_len, sizeof(buf) - prefix_len, format, ap);
}
fprintf(stderr, "%s\n", buf);
}
};
}
bool RocksDBStore::restore_backup(CephContext *cct, const std::string &path,
const std::string &backup_location,
const std::optional<uint32_t> &version)
{
rocksdb::BackupEngineReadOnly* engine_ptr = nullptr;
rocksdb::StderrLogger logger = rocksdb::StderrLogger();
StderrLogger logger;
rocksdb::BackupEngineOptions engine_options = rocksdb::BackupEngineOptions(backup_location);
engine_options.info_log = &logger;

86
src/kv/rocksdb_cache/BinnedLRUCache.cc
View File

@ -359,11 +359,11 @@ rocksdb::Cache::Handle* BinnedLRUCacheShard::Lookup(const rocksdb::Slice& key, u
e->SetHit();
stats[l_hits]++;
}
return reinterpret_cast<rocksdb::Cache::Handle*>(e);
return e;
}
bool BinnedLRUCacheShard::Ref(rocksdb::Cache::Handle* h) {
BinnedLRUHandle* handle = reinterpret_cast<BinnedLRUHandle*>(h);
BinnedLRUHandle* handle = static_cast<BinnedLRUHandle*>(h);
std::lock_guard<std::mutex> l(mutex_);
if (handle->InCache() && handle->refs == 1) {
LRU_Remove(handle);
@ -383,7 +383,7 @@ bool BinnedLRUCacheShard::Release(rocksdb::Cache::Handle* handle, bool force_era
if (handle == nullptr) {
return false;
}
BinnedLRUHandle* e = reinterpret_cast<BinnedLRUHandle*>(handle);
BinnedLRUHandle* e = static_cast<BinnedLRUHandle*>(handle);
bool last_reference = false;
{
std::lock_guard<std::mutex> l(mutex_);
@ -485,7 +485,7 @@ rocksdb::Status BinnedLRUCacheShard::Insert(const rocksdb::Slice& key, uint32_t
if (handle == nullptr) {
LRU_Insert(e);
} else {
*handle = reinterpret_cast<rocksdb::Cache::Handle*>(e);
*handle = e;
}
s = rocksdb::Status::OK();
}
@ -498,6 +498,60 @@ rocksdb::Status BinnedLRUCacheShard::Insert(const rocksdb::Slice& key, uint32_t
return s;
}
#if CEPH_ROCKSDB_SINCE(8, 1)
rocksdb::Cache::Handle* BinnedLRUCacheShard::CreateStandalone(
const rocksdb::Slice& key, uint32_t hash, rocksdb::Cache::ObjectPtr value,
const rocksdb::Cache::CacheItemHelper* helper, size_t charge,
bool allow_uncharged) {
// Allocate up front, as Insert() does, so that a throwing allocation cannot
// leave the accounting below committed with no handle to release it. Never
// enters the hash table, so Lookup() cannot find it; the returned handle
// holds the only reference and Release() frees it.
auto e = new BinnedLRUHandle();
e->value = value;
e->helper = helper;
e->charge = charge;
e->key_length = key.size();
e->key_data = new char[e->key_length];
e->flags = 0;
e->hash = hash;
e->refs = 1;
e->next = e->prev = nullptr;
e->SetInCache(false);
std::copy_n(key.data(), e->key_length, e->key_data);
BinnedLRUHandle* deleted = nullptr;
bool ok = true;
{
std::lock_guard l(mutex_);
EvictFromLRU(charge, deleted);
if (usage_ - lru_usage_ + charge > capacity_ && strict_capacity_limit_) {
if (allow_uncharged) {
e->charge = 0; // hand one back anyway, with GetCharge() == 0
} else {
ok = false;
}
}
if (ok) {
usage_ += e->charge;
// Release() drops l_elems for any last reference, so count it here too
stats[l_elems]++;
// set so age_bin is never null; the charge is not binned, as the entry
// stays off the LRU list
e->age_bin = age_bins.front();
}
}
FreeDeleted(deleted); // outside the mutex, as Insert() does
if (!ok) {
delete[] e->key_data;
delete e;
return nullptr;
}
return e;
}
#endif
void BinnedLRUCacheShard::Erase(const rocksdb::Slice& key, uint32_t hash) {
BinnedLRUHandle* e;
bool last_reference = false;
@ -564,7 +618,7 @@ std::string BinnedLRUCacheShard::GetPrintableOptions() const {
const rocksdb::Cache::CacheItemHelper*
BinnedLRUCacheShard::GetCacheItemHelper(rocksdb::Cache::Handle* h) const
{
auto* handle = reinterpret_cast<BinnedLRUHandle*>(h);
auto* handle = static_cast<BinnedLRUHandle*>(h);
return handle->helper;
}
@ -705,15 +759,15 @@ const CacheShard* BinnedLRUCache::GetShard(int shard) const {
}
rocksdb::Cache::ObjectPtr BinnedLRUCache::Value(Handle* handle) {
return reinterpret_cast<const BinnedLRUHandle*>(handle)->value;
return static_cast<const BinnedLRUHandle*>(handle)->value;
}
size_t BinnedLRUCache::GetCharge(Handle* handle) const {
return reinterpret_cast<const BinnedLRUHandle*>(handle)->charge;
return static_cast<const BinnedLRUHandle*>(handle)->charge;
}
uint32_t BinnedLRUCache::GetHash(Handle* handle) const {
return reinterpret_cast<const BinnedLRUHandle*>(handle)->hash;
return static_cast<const BinnedLRUHandle*>(handle)->hash;
}
void BinnedLRUCache::DisownData() {
@ -726,9 +780,23 @@ void BinnedLRUCache::DisownData() {
const rocksdb::Cache::CacheItemHelper*
BinnedLRUCache::GetCacheItemHelper(Handle* handle) const
{
return reinterpret_cast<const BinnedLRUHandle*>(handle)->helper;
return static_cast<const BinnedLRUHandle*>(handle)->helper;
}
#if CEPH_ROCKSDB_SINCE(9, 7)
void BinnedLRUCache::ApplyToHandle(
rocksdb::Cache* /*cache*/, Handle* handle,
const std::function<void(const rocksdb::Slice& key,
rocksdb::Cache::ObjectPtr value, size_t charge,
const rocksdb::Cache::CacheItemHelper* helper)>&
callback)
{
// the handle need only be layout compatible, so "cache" is unused
auto h = static_cast<BinnedLRUHandle*>(handle);
callback(h->key(), h->value, h->charge, h->helper);
}
#endif
size_t BinnedLRUCache::TEST_GetLRUSize() {
size_t lru_size_of_all_shards = 0;
for (int i = 0; i < num_shards_; i++) {

23
src/kv/rocksdb_cache/BinnedLRUCache.h
View File

@ -56,7 +56,10 @@ std::shared_ptr<rocksdb::Cache> NewBinnedLRUCache(
bool strict_capacity_limit = false,
double high_pri_pool_ratio = 0.0);
struct BinnedLRUHandle {
// Derives from rocksdb::Cache::Handle, the opaque type RocksDB hands back to
// its callers, so that converting between the two needs no reinterpret_cast.
// The base is empty, so this does not change the layout.
struct BinnedLRUHandle : public rocksdb::Cache::Handle {
std::shared_ptr<uint64_t> age_bin;
rocksdb::Cache::ObjectPtr value;
const rocksdb::Cache::CacheItemHelper* helper;
@ -246,6 +249,13 @@ class alignas(CACHE_LINE_SIZE) BinnedLRUCacheShard : public CacheShard {
size_t charge,
rocksdb::Cache::Handle** handle,
rocksdb::Cache::Priority priority) override;
#if CEPH_ROCKSDB_SINCE(8, 1)
virtual rocksdb::Cache::Handle* CreateStandalone(
const rocksdb::Slice& key, uint32_t hash,
rocksdb::Cache::ObjectPtr value,
const rocksdb::Cache::CacheItemHelper* helper,
size_t charge, bool allow_uncharged) override;
#endif
virtual rocksdb::Cache::Handle* Lookup(const rocksdb::Slice& key, uint32_t hash) override;
virtual bool Ref(rocksdb::Cache::Handle* handle) override;
virtual bool Release(rocksdb::Cache::Handle* handle,
@ -390,6 +400,17 @@ class BinnedLRUCache : public ShardedCache {
virtual void DisownData() override;
virtual const rocksdb::Cache::CacheItemHelper* GetCacheItemHelper(
Handle* handle) const override;
#if CEPH_ROCKSDB_SINCE(9, 7)
// here rather than in ShardedCache: reporting the key needs the handle
// layout
virtual void ApplyToHandle(
rocksdb::Cache* cache, Handle* handle,
const std::function<void(const rocksdb::Slice& key,
rocksdb::Cache::ObjectPtr value,
size_t charge,
const rocksdb::Cache::CacheItemHelper* helper)>&
callback) override;
#endif
// Retrieves number of elements in LRU, for unit test purpose only
size_t TEST_GetLRUSize();
// Sets the high pri pool ratio

33
src/kv/rocksdb_cache/ShardedCache.cc
View File

@ -43,23 +43,52 @@ void ShardedCache::SetStrictCapacityLimit(bool strict_capacity_limit) {
strict_capacity_limit_ = strict_capacity_limit;
}
#if CEPH_ROCKSDB_SINCE(8, 7)
rocksdb::Status ShardedCache::Insert(const rocksdb::Slice& key,
rocksdb::Cache::ObjectPtr value,
const rocksdb::Cache::CacheItemHelper* helper,
size_t charge,
rocksdb::Cache::Handle** handle, Priority priority,
const rocksdb::Slice& /*compressed*/,
rocksdb::CompressionType /*type*/) {
#else
rocksdb::Status ShardedCache::Insert(const rocksdb::Slice& key,
rocksdb::Cache::ObjectPtr value,
const rocksdb::Cache::CacheItemHelper* helper,
size_t charge,
rocksdb::Cache::Handle** handle, Priority priority) {
#endif
// no compressed secondary cache, so compressed/type are ignored
uint32_t hash = HashSlice(key);
return GetShard(Shard(hash))
->Insert(key, hash, value, helper, charge, handle, priority);
}
#if CEPH_ROCKSDB_SINCE(8, 1)
rocksdb::Cache::Handle* ShardedCache::CreateStandalone(
const rocksdb::Slice& key, rocksdb::Cache::ObjectPtr value,
const rocksdb::Cache::CacheItemHelper* helper, size_t charge,
bool allow_uncharged) {
uint32_t hash = HashSlice(key);
return GetShard(Shard(hash))
->CreateStandalone(key, hash, value, helper, charge, allow_uncharged);
}
#endif
#if CEPH_ROCKSDB_SINCE(8, 1)
rocksdb::Cache::Handle* ShardedCache::Lookup(const rocksdb::Slice& key,
const rocksdb::Cache::CacheItemHelper* /*helper*/,
rocksdb::Cache::CreateContext* /*create_context*/,
Priority /*priority*/,
rocksdb::Statistics* /*stats*/) {
#else
rocksdb::Cache::Handle* ShardedCache::Lookup(const rocksdb::Slice& key,
const rocksdb::Cache::CacheItemHelper* /*helper*/,
rocksdb::Cache::CreateContext* /*create_context*/,
Priority /*priority*/, bool /*wait*/,
rocksdb::Statistics* /*stats*/) {
// Secondary cache is not supported by BinnedLRUCache, so the helper,
// create_context, priority, wait, and stats arguments are ignored here.
#endif
// no secondary cache, so everything but the key is ignored
uint32_t hash = HashSlice(key);
return GetShard(Shard(hash))->Lookup(key, hash);
}

49
src/kv/rocksdb_cache/ShardedCache.h
View File

@ -16,6 +16,10 @@
#include "rocksdb/version.h"
#include "rocksdb/cache.h"
#if __has_include("rocksdb/advanced_cache.h")
// since RocksDB v8, rocksdb/cache.h only forward-declares rocksdb::Cache
#include "rocksdb/advanced_cache.h"
#endif
#include "include/ceph_hash.h"
#include "common/PriorityCache.h"
//#include "hash.h"
@ -24,6 +28,14 @@
#define CACHE_LINE_SIZE 64 // XXX arch-specific define
#endif
// RocksDB reworked rocksdb::Cache one piece at a time, so each override has
// its own cut-off: v8.1 dropped "wait" from Lookup() and added
// CreateStandalone(), v8.7 gave Insert() its compressed/type parameters, and
// v9.7 added ApplyToHandle().
#define CEPH_ROCKSDB_SINCE(major, minor) \
(ROCKSDB_MAJOR > (major) || \
(ROCKSDB_MAJOR == (major) && ROCKSDB_MINOR >= (minor)))
namespace rocksdb_cache {
// Single cache shard interface.
@ -38,6 +50,14 @@ class CacheShard {
size_t charge,
rocksdb::Cache::Handle** handle,
rocksdb::Cache::Priority priority) = 0;
#if CEPH_ROCKSDB_SINCE(8, 1)
// backs rocksdb::Cache::CreateStandalone()
virtual rocksdb::Cache::Handle* CreateStandalone(
const rocksdb::Slice& key, uint32_t hash,
rocksdb::Cache::ObjectPtr value,
const rocksdb::Cache::CacheItemHelper* helper,
size_t charge, bool allow_uncharged) = 0;
#endif
virtual rocksdb::Cache::Handle* Lookup(const rocksdb::Slice& key, uint32_t hash) = 0;
virtual bool Ref(rocksdb::Cache::Handle* handle) = 0;
virtual bool Release(rocksdb::Cache::Handle* handle, bool force_erase = false) = 0;
@ -68,18 +88,47 @@ class ShardedCache : public rocksdb::Cache, public PriorityCache::PriCache {
// rocksdb::Cache
virtual const char* Name() const override = 0;
using rocksdb::Cache::Insert;
// these must match the base declarations exactly, or they stop overriding
// and ShardedCache stays abstract
#if CEPH_ROCKSDB_SINCE(8, 7)
virtual rocksdb::Status Insert(const rocksdb::Slice& key,
rocksdb::Cache::ObjectPtr value,
const rocksdb::Cache::CacheItemHelper* helper,
size_t charge,
rocksdb::Cache::Handle** handle,
Priority priority,
const rocksdb::Slice& compressed = rocksdb::Slice(),
rocksdb::CompressionType type =
rocksdb::CompressionType::kNoCompression) override;
#else
virtual rocksdb::Status Insert(const rocksdb::Slice& key,
rocksdb::Cache::ObjectPtr value,
const rocksdb::Cache::CacheItemHelper* helper,
size_t charge,
rocksdb::Cache::Handle** handle,
Priority priority) override;
#endif
#if CEPH_ROCKSDB_SINCE(8, 1)
// creates an entry Lookup() cannot find, used for charging memory
virtual rocksdb::Cache::Handle* CreateStandalone(
const rocksdb::Slice& key, rocksdb::Cache::ObjectPtr value,
const rocksdb::Cache::CacheItemHelper* helper,
size_t charge, bool allow_uncharged) override;
#endif
using rocksdb::Cache::Lookup;
#if CEPH_ROCKSDB_SINCE(8, 1)
virtual rocksdb::Cache::Handle* Lookup(const rocksdb::Slice& key,
const rocksdb::Cache::CacheItemHelper* helper,
rocksdb::Cache::CreateContext* create_context,
Priority priority,
rocksdb::Statistics* stats) override;
#else
virtual rocksdb::Cache::Handle* Lookup(const rocksdb::Slice& key,
const rocksdb::Cache::CacheItemHelper* helper,
rocksdb::Cache::CreateContext* create_context,
Priority priority, bool wait,
rocksdb::Statistics* stats) override;
#endif
virtual bool Ref(rocksdb::Cache::Handle* handle) override;
using rocksdb::Cache::Release;
virtual bool Release(rocksdb::Cache::Handle* handle, bool force_erase = false) override;