SteNicholas commented on code in PR #209:
URL: https://github.com/apache/paimon-cpp/pull/209#discussion_r3810802936
##########
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:
Keep evicted async reads tracked until completion. This can erase an entry
whose asynchronous read is still in flight. The callback keeps the buffer
alive, but `ReleaseBuffers()` and the destructor only wait for futures that
remain in `entries_`; a later close/destruction can therefore release the
stream or memory pool while an evicted request is still writing, and `Bytes`
only retains a raw `MemoryPool*`. Please track every dispatched request
independently until completion and wait for all of them during release, or
avoid evicting in-flight entries.
--
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]
