SteNicholas commented on code in PR #209:
URL: https://github.com/apache/paimon-cpp/pull/209#discussion_r3809798645
##########
src/paimon/common/utils/read_ahead_cache.cpp:
##########
@@ -193,31 +266,96 @@ void ReadAheadCache::Impl::Reset() {
is_cached_.clear();
pending_ranges_.clear();
is_initialized_ = false;
+ // The read/io counters are deliberately kept: a reader closed at EOF must
+ // still be able to report them through CollectMetrics().
+}
+
+void ReadAheadCache::Impl::CollectMetrics(std::shared_ptr<Metrics>* metrics)
const {
+ if (metrics == nullptr || !*metrics) {
+ return;
+ }
+ auto& m = *metrics;
+ m->SetCounter(ReadAheadCacheMetrics::READ_COUNT,
read_count_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::READ_BYTES,
read_bytes_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::READ_HITS,
hits_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::READ_HIT_BYTES,
+ hit_bytes_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::READ_MISSES,
misses_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::READ_MISS_BYTES,
+ miss_bytes_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::IO_COUNT,
io_count_.load(std::memory_order_relaxed));
+ m->SetCounter(ReadAheadCacheMetrics::IO_BYTES,
io_bytes_.load(std::memory_order_relaxed));
+}
+
+void ReadAheadCache::Impl::Warmup() {
+ // Init() only registers the pending ranges; without this the first fetch
+ // starts when the first Read() arrives, racing the reader's own miss
fetch.
+ if (!pending_ranges_.empty()) {
+ PreBuffer(pending_ranges_.front().offset);
+ }
}
-Result<ByteSlice> ReadAheadCache::Impl::Read(const ByteRange& range) {
+std::vector<RangeCacheEntry> ReadAheadCache::Impl::FindCoveringEntries(const
ByteRange& range) {
+ std::vector<RangeCacheEntry> covering;
+ std::shared_lock<std::shared_mutex> lock(rw_mutex_);
+ // Find the entry holding the start of the range: the first entry whose
+ // end is beyond range.offset (entries are disjoint and sorted by offset).
+ auto it = std::lower_bound(entries_.begin(), entries_.end(), range.offset,
+ [](const RangeCacheEntry& e, uint64_t offset) {
+ return e.range.offset + e.range.length <=
offset;
+ });
+ if (it == entries_.end() || it->range.offset > range.offset) {
+ return covering;
+ }
+ if (it->range.Contains(range)) {
+ covering.push_back(*it);
+ return covering;
+ }
+ // The request spans several adjacent entries (a column chunk larger than
+ // one coalesced range): collect the contiguous run and check it covers
+ // the whole request. Entries exist from the moment their fetch is
+ // SUBMITTED, so a reader racing the prefetch waits for the in-flight
+ // fetch instead of issuing a second one for the same bytes.
+ uint64_t covered_end = it->range.offset + it->range.length;
+ covering.push_back(*it);
+ auto next = std::next(it);
+ while (covered_end < range.offset + range.length && next != entries_.end()
&&
+ next->range.offset == covered_end) {
+ covered_end = next->range.offset + next->range.length;
+ covering.push_back(*next);
+ ++next;
+ }
+ if (covered_end < range.offset + range.length) {
+ covering.clear();
+ }
+ return covering;
+}
+
+Result<bool> ReadAheadCache::Impl::Read(const ByteRange& range, char* dest) {
if (range.length == 0) {
- return ByteSlice{std::make_shared<Bytes>(0, memory_pool_.get()), 0, 0};
+ return true;
}
+ read_count_.fetch_add(1, std::memory_order_relaxed);
+ read_bytes_.fetch_add(range.length, std::memory_order_relaxed);
PreBuffer(range.offset);
Review Comment:
Publish in-flight entries before readers can miss.`PreBuffer` marks each
range cached before `Cache` is called, but `Cache` does not publish its entries
until `MakeCacheEntries` has dispatched or completed every `ReadAsync`. A
concurrent reader can therefore see `is_cached=true` while
`FindCoveringEntries` still finds nothing, then issue a duplicate underlying
read. Please publish the promise-backed entries under the lock before starting
I/O, or otherwise expose the pending state to readers.
--
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]
