Skip to content

fix(cpp): complete and adopt bounded lru cache #967

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Rayan-and-beyond wants to merge 2 commits into
base: develop
Choose a base branch
from
Open

fix(cpp): complete and adopt bounded lru cache #967

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
facc790
fix(cpp): complete and adopt bounded LRU cache
Rayan-and-beyond Sep 17, 2026
38d1a87
style(cpp): apply spotless formatting
Rayan-and-beyond Sep 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
123 changes: 74 additions & 49 deletions cpp/src/common/cache/lru_cache.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,11 @@
#define COMMON_CACHE_LRU_CACHE_H

#include <algorithm>
#include <limits>
#include <list>
#include <stdexcept>
#include <unordered_map>

#include "utils/errno_define.h"
#include <utility>

namespace common {

Expand All @@ -38,10 +39,20 @@ struct KeyValuePair {
};

/**
* The LRU Cache class templated by
* Key - key type
* Value - value type
* MapType - an associative container like std::unordered_map
* Least-recently-used cache.
*
* The cache is not internally synchronized. A caller that shares one cache
* across threads must serialize every operation, including the full lifetime
* of a pointer/reference returned by getPtr(), getRef(), or tryGetRef().
*
* maxSize is the normal retained-entry limit. elasticity allows the cache to
* grow temporarily to maxSize + elasticity; the insertion that would exceed
* that hard limit prunes least-recently-used entries back to maxSize. A
* maxSize of 0 means unbounded and elasticity is ignored.
*
* Pointers/references returned by non-copying lookups remain valid until that
* entry is updated, removed, or evicted, or until clear()/destruction. Any
* insertion can evict an entry when the cache is bounded.
*/
template <class Key, class Value,
class Map = std::unordered_map<
Expand All @@ -51,26 +62,23 @@ class Cache {
typedef KeyValuePair<Key, Value> node_type;
typedef std::list<KeyValuePair<Key, Value>> list_type;
typedef Map map_type;
/**
* the maxSize is the soft limit of entries and (maxSize + elasticity) is
* the hard limit the cache is allowed to grow till (maxSize + elasticity)
* and is pruned back to maxSize entries set maxSize = 0 for an unbounded
* cache (but in that case, you're better off using a std::unordered_map
* directly anyway! :)
*/

explicit Cache(size_t maxSize = 64, size_t elasticity = 10)
: maxSize_(maxSize), elasticity_(elasticity) {}
virtual ~Cache() = default;

size_t size() const { return cache_.size(); }
bool empty() const { return cache_.empty(); }

void clear() {
cache_.clear();
entries_.clear();
}

void insert(const Key& k, Value v) {
const auto iter = cache_.find(k);
if (iter != cache_.end()) {
iter->second->value = v;
iter->second->value = std::move(v);
entries_.splice(entries_.begin(), entries_, iter->second);
return;
}
Expand All @@ -79,39 +87,56 @@ class Cache {
cache_[k] = entries_.begin();
prune();
}
/**
for backward compatibility. redirects to tryGetCopy()
*/
bool tryGet(const Key& kIn, Value& vOut) { return tryGetCopy(kIn, vOut); }

bool tryGetCopy(const Key& kIn, Value& vOut) {
Value tmp;
if (!tryGetRef_nolock(kIn, tmp)) {
/** Backward-compatible copying lookup. */
bool tryGet(const Key& k, Value& vOut) { return tryGetCopy(k, vOut); }

bool tryGetCopy(const Key& k, Value& vOut) {
const Value* value = getPtr(k);
if (value == nullptr) {
return false;
}
vOut = tmp;
vOut = *value;
return true;
}

bool tryGetRef(const Key& kIn, Value& vOut) {
return tryGetRef_nolock(kIn, vOut);
}
/**
* The const reference returned here is only
* guaranteed to be valid till the next insert/delete
* in multi-threaded apps use getCopy() to be threadsafe
* Non-copying lookup. On success vOut points at the cached value and the
* entry is promoted to most-recently-used.
*/
const Value& getRef(const Key& k) { return get_nolock(k); }
bool tryGetRef(const Key& k, const Value*& vOut) {
vOut = getPtr(k);
return vOut != nullptr;
}

/**
added for backward compatibility
* Legacy overload retained for source compatibility. Despite its historic
* name it copies; new code should use the pointer overload or getPtr().
*/
bool tryGetRef(const Key& k, Value& vOut) { return tryGetCopy(k, vOut); }

/** Returns nullptr on miss and promotes a hit without copying the value. */
const Value* getPtr(const Key& k) {
const auto iter = cache_.find(k);
if (iter == cache_.end()) {
return nullptr;
}
entries_.splice(entries_.begin(), entries_, iter->second);
return &iter->second->value;
}

const Value& getRef(const Key& k) {
const Value* value = getPtr(k);
if (value == nullptr) {
throw std::out_of_range("LRU cache key not found");
}
return *value;
}

/** Backward-compatible copying lookup. */
Value get(const Key& k) { return getCopy(k); }
/**
* returns a copy of the stored object (if found)
* safe to use/recommended in multi-threaded apps
*/
Value getCopy(const Key& k) { return get_nolock(k); }

Value getCopy(const Key& k) { return getRef(k); }

bool remove(const Key& k) {
auto iter = cache_.find(k);
Expand All @@ -122,29 +147,30 @@ class Cache {
cache_.erase(iter);
return true;
}

bool contains(const Key& k) const { return cache_.find(k) != cache_.end(); }

size_t getMaxSize() const { return maxSize_; }
size_t getElasticity() const { return elasticity_; }
size_t getMaxAllowedSize() const { return maxSize_ + elasticity_; }
size_t getMaxAllowedSize() const {
if (maxSize_ == 0) {
return 0;
}
const size_t max = std::numeric_limits<size_t>::max();
if (elasticity_ > max - maxSize_) {
return max;
}
return maxSize_ + elasticity_;
}

template <typename F>
void cwalk(F& f) const {
std::for_each(entries_.begin(), entries_.end(), f);
}

protected:
bool tryGetRef_nolock(const Key& kIn, Value& vOut) {
const auto iter = cache_.find(kIn);
if (iter == cache_.end()) {
return false;
}
entries_.splice(entries_.begin(), entries_, iter->second);
vOut = iter->second->value;
return true;
}
size_t prune() {
size_t maxAllowed = maxSize_ + elasticity_;
if (maxSize_ == 0 || cache_.size() < maxAllowed) {
if (maxSize_ == 0 || cache_.size() <= getMaxAllowedSize()) {
return 0;
}
size_t count = 0;
Expand All @@ -157,7 +183,6 @@ class Cache {
}

private:
// Disallow copying.
Cache(const Cache&) = delete;
Cache& operator=(const Cache&) = delete;

Expand All @@ -168,4 +193,4 @@ class Cache {
};
} // namespace common

#endif // COMMON_CACHE_LRU_CACHE_H
#endif // COMMON_CACHE_LRU_CACHE_H
79 changes: 31 additions & 48 deletions cpp/src/file/tsfile_io_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,8 +56,10 @@ void TsFileIOReader::reset() {
}
read_file_ = nullptr;
tsfile_meta_page_arena_.destroy();
device_node_cache_.clear();
device_node_cache_pa_.destroy();
{
std::lock_guard<std::mutex> lk(device_node_cache_mu_);
device_node_cache_.clear();
}
tsfile_meta_ready_ = false;
}
}
Expand Down Expand Up @@ -623,22 +625,18 @@ std::string TsFileIOReader::device_node_cache_key(
int TsFileIOReader::get_cached_device_node(std::shared_ptr<IDeviceID> device_id,
common::PageArena& pa,
CachedDeviceNode& out) {
(void)pa;
std::string dev_name = device_node_cache_key(device_id);

{
std::lock_guard<std::mutex> lk(device_node_cache_mu_);
auto it = device_node_cache_.find(dev_name);
if (it != device_node_cache_.end()) {
out = it->second;
if (device_node_cache_.tryGetCopy(dev_name, out)) {
return E_OK;
}
}

// Read the device meta index outside the lock — load_device_index_entry()
// and the file read can block on I/O, and we don't want to serialize all
// concurrent first-time lookups behind one slow disk fetch. Two callers
// racing on the same missing device may both do the read; that's wasted
// work but not corruption — the second insert is dropped below.
// Read the device meta index outside the lock. Concurrent misses may do
// duplicate I/O, but the cache is re-checked before insertion below.
int ret = E_OK;
std::shared_ptr<IMetaIndexEntry> device_index_entry;
int64_t device_ie_end_offset = 0;
Expand All @@ -652,10 +650,6 @@ int TsFileIOReader::get_cached_device_node(std::shared_ptr<IDeviceID> device_id,
end_offset = device_ie_end_offset;
ASSERT(start_offset < end_offset);
const int64_t read_size_i64 = end_offset - start_offset;
// read_file_->read() takes int32_t; a meta index node larger than 2 GiB
// is implausible but explicitly reject it instead of silently truncating
// the read length and corrupting the parse. Distinguish the two cases:
// an inverted/empty range is corruption, an oversized one is an overflow.
if (read_size_i64 <= 0) {
return E_TSFILE_CORRUPTED;
}
Expand All @@ -665,14 +659,6 @@ int TsFileIOReader::get_cached_device_node(std::shared_ptr<IDeviceID> device_id,
const int32_t read_size = static_cast<int32_t>(read_size_i64);
int32_t ret_read_len = 0;

// Read into a heap-owned buffer outside the lock. The previous
// implementation allocated data_buf inside device_node_cache_pa_ before
// the read happened — every failed read or parse left that allocation
// pinned forever in the shared arena, and repeated disk errors on the
// same device let a long-lived reader grow it without bound. Using a
// unique_ptr here means the read buffer is released on every failure
// path, and only the small MetaIndexNode allocations inside the lock
// share the arena.
std::unique_ptr<char[]> data_buf(new (std::nothrow) char[read_size]);
if (data_buf == nullptr) {
return E_OOM;
Expand All @@ -684,37 +670,34 @@ int TsFileIOReader::get_cached_device_node(std::shared_ptr<IDeviceID> device_id,
return E_FILE_READ_ERR;
}

CachedDeviceNode cached;
// Give every cached node its own arena. The CachedDeviceNode keeps this
// arena alive for users that copied the entry, while LRU eviction releases
// the cache's ownership and therefore reclaims unused metadata pages.
CachedDeviceNode candidate;
candidate.arena = std::make_shared<common::PageArena>();
candidate.arena->init(512, common::MOD_TSFILE_READER);
void* m_idx_node_buf = candidate.arena->alloc(sizeof(MetaIndexNode));
if (IS_NULL(m_idx_node_buf)) {
return E_OOM;
}
auto* top_node_ptr =
new (m_idx_node_buf) MetaIndexNode(candidate.arena.get());
candidate.top_node = std::shared_ptr<MetaIndexNode>(
top_node_ptr, MetaIndexNode::self_deleter);
if (RET_FAIL(
candidate.top_node->deserialize_from(data_buf.get(), read_size))) {
return ret;
}
candidate.is_aligned = is_aligned_device(candidate.top_node);

{
// Allocations into device_node_cache_pa_ and the map insert must be
// serialized — PageArena is not thread-safe, and unordered_map's
// rehash invalidates concurrent lookups.
std::lock_guard<std::mutex> lk(device_node_cache_mu_);
// Re-check: another thread may have populated the entry while we
// were doing I/O.
auto it = device_node_cache_.find(dev_name);
if (it != device_node_cache_.end()) {
out = it->second;
if (device_node_cache_.tryGetCopy(dev_name, out)) {
return E_OK;
}

void* m_idx_node_buf =
device_node_cache_pa_.alloc(sizeof(MetaIndexNode));
if (IS_NULL(m_idx_node_buf)) {
return E_OOM;
}
auto* top_node_ptr =
new (m_idx_node_buf) MetaIndexNode(&device_node_cache_pa_);
auto top_node = std::shared_ptr<MetaIndexNode>(
top_node_ptr, MetaIndexNode::self_deleter);
if (RET_FAIL(top_node->deserialize_from(data_buf.get(), read_size))) {
return ret;
}
cached.top_node = top_node;
cached.is_aligned = is_aligned_device(top_node);
device_node_cache_.emplace(std::move(dev_name), cached);
device_node_cache_.insert(dev_name, candidate);
out = candidate;
}
out = cached;
return E_OK;
}

Expand Down
Loading