rok commented on code in PR #49855:
URL: https://github.com/apache/arrow/pull/49855#discussion_r3645391354
##########
cpp/src/arrow/io/caching.cc:
##########
@@ -167,115 +175,151 @@ 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) {
+ std::unique_lock<std::mutex> guard(entry_mutex);
+ const auto first_kept =
+ std::remove_if(entries.begin(), entries.end(),
+ [end_offset](const RangeCacheEntry& entry) {
+ return entry.range.offset + entry.range.length <=
end_offset;
+ });
+ const auto n_evicted = static_cast<int64_t>(entries.end() - first_kept);
+ entries.erase(first_kept, entries.end());
+ return n_evicted;
+ }
Review Comment:
Thinking out loud: `std::remove_if` scans all remaining entries and shifts
survivors on every row-group eviction, so R evictions over E entries cost
O(R×E). A deque can pop_front() each expired entry once in O(1), making all
evictions O(E) total. Do you think something like proposal below could make
material difference?
```diff
diff --git a/cpp/src/arrow/io/caching.cc b/cpp/src/arrow/io/caching.cc
--- a/cpp/src/arrow/io/caching.cc
+++ b/cpp/src/arrow/io/caching.cc
@@ -19,6 +19,7 @@
#include <algorithm>
#include <atomic>
#include <cmath>
+#include <deque>
#include <mutex>
#include <utility>
#include <vector>
@@ -156,7 +157,7 @@ struct ReadRangeCache::Impl {
// below, so both the eager and lazy variants are safe to call concurrently
// from multiple threads.
- std::vector<RangeCacheEntry> entries;
+ std::deque<RangeCacheEntry> entries;
std::mutex entry_mutex;
@@ -192,12 +193,15 @@ struct ReadRangeCache::Impl {
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::deque<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);
+ for (auto& entry : new_entries) {
+ entries.push_back(std::move(entry));
+ }
}
}
@@ -269,14 +273,15 @@ struct ReadRangeCache::Impl {
// (coalesced with a range a later consumer still needs) is kept.
int64_t EvictEntriesBefore(int64_t end_offset) {
std::unique_lock<std::mutex> guard(entry_mutex);
- const auto first_kept =
- std::remove_if(entries.begin(), entries.end(),
- [end_offset](const RangeCacheEntry& entry) {
- return entry.range.offset + entry.range.length <=
end_offset;
- });
- const auto n_evicted = static_cast<int64_t>(entries.end() - first_kept);
- entries.erase(first_kept, entries.end());
+ int64_t n_evicted = 0;
+ while (!entries.empty()) {
+ const auto& range = entries.front().range;
+ if (range.offset + range.length > end_offset) {
+ break;
+ }
+ entries.pop_front();
+ ++n_evicted;
+ }
return n_evicted;
}
```
--
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]
