justinli500 commented on code in PR #49855:
URL: https://github.com/apache/arrow/pull/49855#discussion_r3593685577
##########
cpp/src/arrow/io/caching.cc:
##########
@@ -167,115 +175,154 @@ struct ReadRangeCache::Impl {
std::vector<RangeCacheEntry> new_entries;
new_entries.reserve(ranges.size());
for (const auto& range : ranges) {
- new_entries.emplace_back(range, file->ReadAsync(ctx, range.offset,
range.length,
-
/*allow_short_read=*/false));
+ new_entries.emplace_back(range, file->ReadAsync(ctx, range.offset,
range.length));
}
return new_entries;
}
- // Add the given ranges to the cache, coalescing them where possible
- virtual Status Cache(std::vector<ReadRange> ranges) {
+ // -- Public entry points (acquire entry_mutex, then delegate). --
+
+ // Add the given ranges to the cache, coalescing them where possible.
+ Status Cache(std::vector<ReadRange> ranges) {
ARROW_ASSIGN_OR_RAISE(
ranges, internal::CoalesceReadRanges(std::move(ranges),
options.hole_size_limit,
options.range_size_limit));
- std::vector<RangeCacheEntry> new_entries = MakeCacheEntries(ranges);
- // Add new entries, themselves ordered by offset
- if (entries.size() > 0) {
- std::vector<RangeCacheEntry> merged(entries.size() + new_entries.size());
- std::merge(entries.begin(), entries.end(), new_entries.begin(),
new_entries.end(),
- merged.begin());
- entries = std::move(merged);
- } else {
- entries = std::move(new_entries);
+ Status st;
+ {
+ std::unique_lock<std::mutex> guard(entry_mutex);
+ std::vector<RangeCacheEntry> new_entries = MakeCacheEntries(ranges);
+ // Add new entries, themselves ordered by offset
+ if (entries.size() > 0) {
+ std::vector<RangeCacheEntry> merged(entries.size() +
new_entries.size());
+ std::merge(entries.begin(), entries.end(), new_entries.begin(),
new_entries.end(),
+ merged.begin());
+ entries = std::move(merged);
+ } else {
+ entries = std::move(new_entries);
+ }
}
- // Prefetch immediately, regardless of executor availability, if possible
- auto st = file->WillNeed(ranges);
+ // Prefetch immediately, regardless of executor availability, if possible.
+ // Do this outside the lock: WillNeed() may block on an mmap advise / I/O
+ // hint and we don't want to serialize concurrent Reads on it.
+ st = file->WillNeed(ranges);
// As this is optimisation only, I/O failures should not be treated as
fatal
if (st.IsIOError()) {
return Status::OK();
}
return st;
}
- // Read the given range from the cache, blocking if needed. Cannot read a
range
- // that spans cache entries.
- virtual Result<std::shared_ptr<Buffer>> Read(ReadRange range) {
+ // Read the given range from the cache, blocking if needed. Cannot read a
+ // range that spans cache entries.
+ Result<std::shared_ptr<Buffer>> Read(ReadRange range) {
if (range.length == 0) {
static const uint8_t byte = 0;
return std::make_shared<Buffer>(&byte, 0);
}
- const auto it = std::lower_bound(
- entries.begin(), entries.end(), range,
- [](const RangeCacheEntry& entry, const ReadRange& range) {
- return entry.range.offset + entry.range.length < range.offset +
range.length;
- });
- if (it != entries.end() && it->range.Contains(range)) {
- auto fut = MaybeRead(&*it);
- ARROW_ASSIGN_OR_RAISE(auto buf, fut.result());
+ Future<std::shared_ptr<Buffer>> fut;
+ int64_t slice_offset = 0;
+ {
+ std::unique_lock<std::mutex> guard(entry_mutex);
+ const auto it = std::lower_bound(
+ entries.begin(), entries.end(), range,
+ [](const RangeCacheEntry& entry, const ReadRange& range) {
+ return entry.range.offset + entry.range.length < range.offset +
range.length;
+ });
+ if (it == entries.end() || !it->range.Contains(range)) {
+ return Status::Invalid("ReadRangeCache did not find matching cache
entry");
+ }
+ fut = MaybeRead(&*it);
+ slice_offset = range.offset - it->range.offset;
if (options.lazy && options.prefetch_limit > 0) {
int64_t num_prefetched = 0;
for (auto next_it = it + 1;
next_it != entries.end() && num_prefetched <
options.prefetch_limit;
++next_it) {
if (!next_it->future.is_valid()) {
next_it->future =
- file->ReadAsync(ctx, next_it->range.offset,
next_it->range.length,
- /*allow_short_read=*/false);
+ file->ReadAsync(ctx, next_it->range.offset,
next_it->range.length);
}
++num_prefetched;
}
}
- return SliceBuffer(std::move(buf), range.offset - it->range.offset,
range.length);
}
- return Status::Invalid("ReadRangeCache did not find matching cache entry");
+ // Drop the lock before blocking on the I/O future so other threads can
+ // still do lookups while a previously queued read is in flight.
+ ARROW_ASSIGN_OR_RAISE(auto buf, fut.result());
+ return SliceBuffer(std::move(buf), slice_offset, range.length);
}
- virtual Future<> Wait() {
+ Future<> Wait() {
std::vector<Future<>> futures;
- for (auto& entry : entries) {
- futures.emplace_back(MaybeRead(&entry));
+ {
+ std::unique_lock<std::mutex> guard(entry_mutex);
+ futures.reserve(entries.size());
+ for (auto& entry : entries) {
+ futures.emplace_back(MaybeRead(&entry));
+ }
}
return AllComplete(futures);
}
+ // `entries` is sorted by offset; an entry that extends past `end_offset`
+ // (coalesced with a range a later consumer still needs) is kept.
+ int64_t EvictEntriesBefore(int64_t end_offset) {
+ int64_t n_evicted = 0;
+ std::unique_lock<std::mutex> guard(entry_mutex);
+ auto it = entries.begin();
+ while (it != entries.end() && it->range.offset < end_offset) {
+ if (it->range.offset + it->range.length <= end_offset) {
+ it = entries.erase(it);
+ ++n_evicted;
+ } else {
+ ++it;
+ }
+ }
Review Comment:
Looks great, went with your version here. This should keep the entries
ordered and also avoid repeated vector erases.
--
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]
