lucasfang commented on code in PR #209:
URL: https://github.com/apache/paimon-cpp/pull/209#discussion_r3811057124
##########
src/paimon/common/utils/read_ahead_cache.cpp:
##########
@@ -89,38 +152,66 @@ class ReadAheadCache::Impl {
std::vector<std::atomic<bool>> is_cached_;
std::vector<ByteRange> pending_ranges_;
bool is_initialized_ = false;
+ // Statistics of the Read() requests issued to the cache, aggregated over
+ // all streams sharing this cache.
+ std::atomic<uint64_t> read_count_{0};
+ std::atomic<uint64_t> read_bytes_{0};
+ std::atomic<uint64_t> hits_{0};
+ std::atomic<uint64_t> hit_bytes_{0};
+ std::atomic<uint64_t> misses_{0};
+ std::atomic<uint64_t> miss_bytes_{0};
+ // Prefetch IO statistics: how many requests and bytes were actually issued
+ // to the underlying stream.
+ std::atomic<uint64_t> io_count_{0};
+ std::atomic<uint64_t> io_bytes_{0};
};
-void ReadAheadCache::Impl::Cache(std::vector<ByteRange> ranges) {
- std::sort(ranges.begin(), ranges.end(),
- [](const ByteRange& a, const ByteRange& b) { return a.offset <
b.offset; });
- std::vector<RangeCacheEntry> new_entries = MakeCacheEntries(ranges);
- // Add new entries, themselves ordered by offset
- std::unique_lock<std::shared_mutex> lock(rw_mutex_);
- if (entries_.size() > 0) {
- size_t new_entries_size = 0;
- for (const auto& e : new_entries) {
- new_entries_size += e.range.length;
- }
-
- size_t total_size = 0;
- for (const auto& e : entries_) {
- total_size += e.range.length;
+void ReadAheadCache::Impl::Cache(std::vector<size_t> pending_indices) {
+ std::vector<RangeCacheEntry> new_entries;
+ std::vector<PendingFetch> fetches;
+ // Mark is_cached_, publish the promise-backed entries and only then
+ // dispatch the IOs. The mark and the publication happen atomically under
+ // the write lock: a reader racing the prefetch observes is_cached_=true
+ // only once the covering entries are already visible, so it waits on
+ // their futures instead of issuing a duplicate underlying read.
+ {
+ std::unique_lock<std::shared_mutex> lock(rw_mutex_);
+ for (size_t idx : pending_indices) {
+ if (is_cached_[idx].exchange(true)) {
+ continue;
+ }
+ const ByteRange& range = pending_ranges_[idx];
+ auto promise = std::make_shared<std::promise<Status>>();
+ auto future = promise->get_future();
+ auto buffer = std::make_shared<Bytes>(range.length,
memory_pool_.get());
+ fetches.push_back({range, buffer, promise});
+ new_entries.emplace_back(range, std::move(buffer),
std::move(future));
}
- size_t limit = config_.GetBufferSizeLimit();
- while (!entries_.empty() && total_size + new_entries_size > limit) {
- auto iter = entries_.begin();
- total_size -= entries_.front().range.length;
- entries_.erase(iter);
+ if (!new_entries.empty()) {
+ // Add new entries, themselves ordered by offset
+ size_t new_entries_size = 0;
+ for (const auto& e : new_entries) {
+ new_entries_size += e.range.length;
+ }
+
+ size_t total_size = 0;
+ for (const auto& e : entries_) {
+ total_size += e.range.length;
+ }
+ size_t limit = config_.GetBufferSizeLimit();
+ while (!entries_.empty() && total_size + new_entries_size > limit)
{
+ auto iter = entries_.begin();
+ total_size -= entries_.front().range.length;
+ entries_.erase(iter);
Review Comment:
Good point. Rather than tracking evicted requests separately, I've removed
the eviction mechanism entirely: the cache now retains every published entry
until ReleaseBuffers()/Reset(), since it only holds the prefetched ranges of a
single data file and the memory footprint is acceptable. With no eviction, an
in-flight fetch can never lose its entry, so the existing waits on entries_ in
ReleaseBuffers() and the destructor already cover every dispatched request — no
async read can outlive the stream or memory pool. buffer_size_limit is dropped
from CacheConfig accordingly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
