This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 71eecdfc8b6 branch-4.2: [improvement](be) Skip stale S3 race reads and
serve peer reads without get_or_set #69620 (#69803)
71eecdfc8b6 is described below
commit 71eecdfc8b6851eceb365ad214ee0c1ed9f64db3
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 30 09:42:25 2026 +0800
branch-4.2: [improvement](be) Skip stale S3 race reads and serve peer reads
without get_or_set #69620 (#69803)
Cherry-picked from #69620
Co-authored-by: deardeng <[email protected]>
---
be/src/cloud/cloud_internal_service.cpp | 224 ++++++++-------
be/src/io/cache/block_file_cache.cpp | 17 +-
be/src/io/cache/cached_remote_file_reader.cpp | 15 +
.../cache/cached_remote_file_reader_peer_test.cpp | 303 +++++++++++++++++++++
4 files changed, 461 insertions(+), 98 deletions(-)
diff --git a/be/src/cloud/cloud_internal_service.cpp
b/be/src/cloud/cloud_internal_service.cpp
index 2a1974102b4..6f6fc852f61 100644
--- a/be/src/cloud/cloud_internal_service.cpp
+++ b/be/src/cloud/cloud_internal_service.cpp
@@ -446,6 +446,17 @@ void set_too_many_tasks_response(PFetchPeerDataResponse*
response, const std::st
response->mutable_status()->set_status_code(TStatusCode::TOO_MANY_TASKS);
}
+// NOT_FOUND means this peer cannot serve the block now; the client rotates
the candidate instead
+// of counting an RPC failure.
+Status reject_peer_block_not_found(PFetchPeerDataResponse* response, const
std::string& msg) {
+ g_file_cache_get_by_peer_failed_num << 1;
+ g_file_cache_get_by_peer_not_downloaded_block_num << 1;
+ LOG(WARNING) << msg;
+ response->mutable_status()->add_error_msgs(msg);
+ response->mutable_status()->set_status_code(TStatusCode::NOT_FOUND);
+ return Status::NotFound<false>(msg);
+}
+
bool try_reject_if_queue_timed_out(std::chrono::steady_clock::time_point
enqueue_ts,
PFetchPeerDataResponse* response) {
auto wait_us = elapsed_us(enqueue_ts);
@@ -680,6 +691,87 @@ Status trigger_peer_server_fill(io::FileBlockSPtr& fb,
int64_t fill_tablet_id,
file_size, offset, size, timeout_ms));
}
+// Serve one cache block of a peer request: wait for a DOWNLOADING block, fill
an EMPTY block when
+// `do_fill` is set, then append the block to `response`. The payload goes to
+// `response_attachment` when it is non-null, otherwise into the protobuf.
+Status serve_peer_cache_block(const PFetchPeerDataRequest* request, const
CacheBlockReqest& cb_req,
+ io::FileBlockSPtr& fb, const io::UInt128Wrapper&
hash,
+ size_t file_size, bool do_fill, butil::IOBuf*
response_attachment,
+ PFetchPeerDataResponse* response) {
+ auto fb_state = fb->state();
+ if (fb_state == io::FileBlock::State::DOWNLOADING) {
+ if (do_fill) {
+ // Only peer fill requests should wait longer here. Plain
peer-cache reads keep
+ // the short wait semantics so they can fail fast and let the
client race S3.
+ [[maybe_unused]] const bool completed =
+ wait_for_file_block_state(fb,
config::peer_server_cache_fill_timeout_ms);
+ fb_state = fb->state();
+ } else {
+ // Wait for in-progress download to complete using the normal
short timeout.
+ fb_state = fb->wait();
+ }
+ }
+ if (fb_state == io::FileBlock::State::EMPTY) {
+ if (!do_fill) {
+ return reject_peer_block_not_found(
+ response, fmt::format("cache block not downloaded, {}",
+
format_peer_cache_block_context(request, cb_req, fb, hash,
+
file_size, do_fill)));
+ }
+ // Server-side fill: request.path already carries the full remote path.
+ auto fill_st = trigger_peer_server_fill(
+ fb, get_rowset_meta_tablet_id_from_request(request),
+ get_peer_cache_filename(request->path()),
+ get_rowset_meta_resource_id_from_request(request),
request->path(),
+ request->has_file_size() ? request->file_size() : -1,
+ static_cast<int64_t>(fb->range().left),
static_cast<int64_t>(fb->range().size()),
+ config::peer_server_cache_fill_timeout_ms);
+ if (!fill_st.ok()) {
+ g_file_cache_get_by_peer_failed_num << 1;
+ g_file_cache_get_by_peer_not_downloaded_block_num << 1;
+ if (fill_st.is<ErrorCode::TOO_MANY_TASKS>()) {
+ // Server slot exhausted: healthy but overloaded. Client must
not rotate
+ // or evict — just fall back to S3 and retry same candidate
next time.
+
response->mutable_status()->add_error_msgs(std::string(fill_st.msg()));
+
response->mutable_status()->set_status_code(TStatusCode::TOO_MANY_TASKS);
+ } else if (fill_st.is<ErrorCode::NOT_FOUND>()) {
+ // Soft miss (fill incomplete, timeout, unexpected state) —
client rotates,
+ // not evicts.
+
response->mutable_status()->add_error_msgs(std::string(fill_st.msg()));
+
response->mutable_status()->set_status_code(TStatusCode::NOT_FOUND);
+ } else {
+ LOG(WARNING) << "cache block fill failed, status=" << fill_st
<< ", "
+ << format_peer_cache_block_context(request,
cb_req, fb, hash,
+ file_size,
do_fill);
+ set_error_response(response, "cache block not ready");
+ }
+ return fill_st;
+ }
+ fb_state = io::FileBlock::State::DOWNLOADED;
+ }
+ if (fb_state != io::FileBlock::State::DOWNLOADED) {
+ // A concurrent download was in progress (DOWNLOADING at request time)
but its
+ // wait() returned a non-DOWNLOADED state (e.g., timed-out while still
+ // DOWNLOADING, or some other non-EMPTY intermediate state). The
server is
+ // healthy; the block just isn't available yet.
+ return reject_peer_block_not_found(
+ response, fmt::format("cache block not ready after wait, {}",
+ format_peer_cache_block_context(request,
cb_req, fb, hash,
+
file_size, do_fill)));
+ }
+
+ g_file_cache_get_by_peer_blocks_num << 1;
+ doris::CacheBlockPB* out = response->add_datas();
+ // In attachment mode, metadata order must match attachment append order
because the
+ // client consumes attachment payload sequentially using resp.datas()
order.
+ Status read_status = read_file_block(fb, file_size, out,
response_attachment);
+ if (!read_status.ok()) {
+ set_error_response(response, "read cache file error");
+ return read_status;
+ }
+ return Status::OK();
+}
+
Status handle_peer_file_cache_block_request(const PFetchPeerDataRequest*
request,
PFetchPeerDataResponse* response,
brpc::Controller* cntl) {
@@ -722,6 +814,7 @@ Status handle_peer_file_cache_block_request(const
PFetchPeerDataRequest* request
const bool use_attachment =
cntl != nullptr && request->has_support_attachment() &&
request->support_attachment();
response->set_data_in_attachment(use_attachment);
+ butil::IOBuf* response_attachment = use_attachment ?
&cntl->response_attachment() : nullptr;
if (use_attachment) {
g_file_cache_get_by_peer_attachment_response_num << 1;
} else {
@@ -737,8 +830,8 @@ Status handle_peer_file_cache_block_request(const
PFetchPeerDataRequest* request
if (offset >= file_size) {
continue;
}
- // Clip tail requests before get_or_set so peer reads do not
synthesize EMPTY blocks past
- // EOF and then fail the whole RPC.
+ // Clip tail requests before the cache lookup so peer reads do not
synthesize EMPTY blocks
+ // past EOF and then fail the whole RPC.
size = std::min(size, file_size - offset);
if (size == 0) {
continue;
@@ -750,102 +843,41 @@ Status handle_peer_file_cache_block_request(const
PFetchPeerDataRequest* request
int sleep_ms = dp->param<int>("sleep_ms", 300);
bthread_usleep(sleep_ms * 1000);
});
- auto get_or_set_start = std::chrono::steady_clock::now();
- auto holder = cache->get_or_set(hash, offset, size, ctx);
- g_file_cache_get_by_peer_get_or_set_latency <<
elapsed_us(get_or_set_start);
- ++get_or_set_calls;
- get_or_set_blocks += holder.file_blocks.size();
-
- for (auto& fb : holder.file_blocks) {
- auto fb_state = fb->state();
- if (fb_state == io::FileBlock::State::DOWNLOADING) {
- if (do_fill) {
- // Only peer fill requests should wait longer here. Plain
peer-cache reads keep
- // the short wait semantics so they can fail fast and let
the client race S3.
- [[maybe_unused]] const bool completed =
wait_for_file_block_state(
- fb, config::peer_server_cache_fill_timeout_ms);
- fb_state = fb->state();
- } else {
- // Wait for in-progress download to complete using the
normal short timeout.
- fb_state = fb->wait();
- }
- }
- if (fb_state == io::FileBlock::State::EMPTY) {
- if (!do_fill) {
- const std::string msg =
- fmt::format("cache block not downloaded, {}",
-
format_peer_cache_block_context(request, cb_req, fb, hash,
-
file_size, do_fill));
- g_file_cache_get_by_peer_failed_num << 1;
- g_file_cache_get_by_peer_not_downloaded_block_num << 1;
- LOG(WARNING) << msg;
- // Use NOT_FOUND so the client can distinguish "block not
cached"
- // from an actual RPC/server error. On NOT_FOUND the
client rotates
- // the candidate to the end of its list (trying another CG
next time)
- // rather than incrementing the RPC-failure eviction
counter.
- response->mutable_status()->add_error_msgs(msg);
-
response->mutable_status()->set_status_code(TStatusCode::NOT_FOUND);
- return Status::NotFound<false>(msg);
- }
- // Server-side fill: request.path already carries the full
remote path.
- auto fill_st = trigger_peer_server_fill(
- fb, get_rowset_meta_tablet_id_from_request(request),
cache_key_path,
- get_rowset_meta_resource_id_from_request(request),
path,
- request->has_file_size() ? request->file_size() : -1,
- static_cast<int64_t>(fb->range().left),
- static_cast<int64_t>(fb->range().size()),
- config::peer_server_cache_fill_timeout_ms);
- if (!fill_st.ok()) {
- g_file_cache_get_by_peer_failed_num << 1;
- g_file_cache_get_by_peer_not_downloaded_block_num << 1;
- if (fill_st.is<ErrorCode::TOO_MANY_TASKS>()) {
- // Server slot exhausted: healthy but overloaded.
Client must not rotate
- // or evict — just fall back to S3 and retry same
candidate next time.
-
response->mutable_status()->add_error_msgs(std::string(fill_st.msg()));
-
response->mutable_status()->set_status_code(TStatusCode::TOO_MANY_TASKS);
- } else if (fill_st.is<ErrorCode::NOT_FOUND>()) {
- // Soft miss (fill incomplete, timeout, unexpected
state) — client rotates,
- // not evicts.
-
response->mutable_status()->add_error_msgs(std::string(fill_st.msg()));
-
response->mutable_status()->set_status_code(TStatusCode::NOT_FOUND);
- } else {
- LOG(WARNING) << "cache block fill failed, status=" <<
fill_st << ", "
- <<
format_peer_cache_block_context(request, cb_req, fb, hash,
-
file_size, do_fill);
- set_error_response(response, "cache block not ready");
- }
- return fill_st;
- }
- fb_state = io::FileBlock::State::DOWNLOADED;
- }
- if (fb_state != io::FileBlock::State::DOWNLOADED) {
- // A concurrent download was in progress (DOWNLOADING at
request time) but its
- // wait() returned a non-DOWNLOADED state (e.g., timed-out
while still
- // DOWNLOADING, or some other non-EMPTY intermediate state).
The server is
- // healthy; the block just isn't available yet. Return
NOT_FOUND so the client
- // rotates the candidate instead of evicting it.
- const std::string msg =
- fmt::format("cache block not ready after wait, {}",
- format_peer_cache_block_context(request,
cb_req, fb, hash,
- file_size,
do_fill));
- g_file_cache_get_by_peer_failed_num << 1;
- g_file_cache_get_by_peer_not_downloaded_block_num << 1;
- LOG(WARNING) << msg;
- response->mutable_status()->add_error_msgs(msg);
-
response->mutable_status()->set_status_code(TStatusCode::NOT_FOUND);
- return Status::NotFound<false>(msg);
+ if (do_fill) {
+ // Pull-through fill needs the EMPTY cell from get_or_set: the
fill download lands in
+ // that cell, and this handler reads it back once the download
finishes.
+ auto get_or_set_start = std::chrono::steady_clock::now();
+ auto holder = cache->get_or_set(hash, offset, size, ctx);
+ g_file_cache_get_by_peer_get_or_set_latency <<
elapsed_us(get_or_set_start);
+ ++get_or_set_calls;
+ get_or_set_blocks += holder.file_blocks.size();
+ for (auto& fb : holder.file_blocks) {
+ RETURN_IF_ERROR(serve_peer_cache_block(request, cb_req, fb,
hash, file_size,
+ do_fill,
response_attachment, response));
+ ++response_blocks;
}
+ continue;
+ }
- g_file_cache_get_by_peer_blocks_num << 1;
- doris::CacheBlockPB* out = response->add_datas();
- // In attachment mode, metadata order must match attachment append
order because the
- // client consumes attachment payload sequentially using
resp.datas() order.
- Status read_status = read_file_block(
- fb, file_size, out, use_attachment ?
&cntl->response_attachment() : nullptr);
- if (!read_status.ok()) {
- set_error_response(response, "read cache file error");
- return read_status;
- }
+ // A plain peer read only serves blocks that are already downloaded.
This lookup inserts no
+ // EMPTY cells, takes no downloader ownership, and accepts any cached
layout, including
+ // short cells left by a pull-through fill from a client with a
smaller cache block size.
+ // It also refreshes LRU state for the returned blocks.
+ io::FileBlocks blocks;
+ bool fully_covered = false;
+ RETURN_IF_ERROR(cache->get_downloaded_blocks_if_fully_covered(hash,
offset, size, ctx,
+ &blocks,
&fully_covered));
+ if (!fully_covered) {
+ return reject_peer_block_not_found(
+ response,
+ fmt::format(
+ "cache block not fully downloaded,
req_block=[offset={}, size={}], {}",
+ cb_req.block_offset(), cb_req.block_size(),
+ format_peer_request_context(request, hash,
file_size)));
+ }
+ for (auto& fb : blocks) {
+ RETURN_IF_ERROR(serve_peer_cache_block(request, cb_req, fb, hash,
file_size, do_fill,
+ response_attachment,
response));
++response_blocks;
}
}
diff --git a/be/src/io/cache/block_file_cache.cpp
b/be/src/io/cache/block_file_cache.cpp
index e1c7fa3aad1..abd1a277dd9 100644
--- a/be/src/io/cache/block_file_cache.cpp
+++ b/be/src/io/cache/block_file_cache.cpp
@@ -55,6 +55,7 @@
#include "io/cache/remote_scan_cache_write_limiter.h"
#include "runtime/runtime_profile.h"
#include "util/concurrency_stats.h"
+#include "util/defer_op.h"
#include "util/mem_info.h"
#include "util/stack_util.h"
#include "util/stopwatch.hpp"
@@ -812,6 +813,14 @@ Status
BlockFileCache::get_downloaded_blocks_if_fully_covered(const UInt128Wrapp
return Status::OK();
}
+ // Async touch defers LRU moves; queue them after the cache lock is
released, as get_or_set()
+ // does. Declared before the lock so it runs after the lock guard is
destroyed.
+ std::vector<FileBlockSPtr> need_update_lru_blocks;
+ Defer queue_lru_updates {[&]() {
+ for (auto& block : need_update_lru_blocks) {
+ add_need_update_lru_block(std::move(block));
+ }
+ }};
FileBlock::Range range(offset, offset + size - 1);
std::lock_guard cache_lock(_mutex);
auto it = _files.find(hash);
@@ -882,9 +891,13 @@ Status
BlockFileCache::get_downloaded_blocks_if_fully_covered(const UInt128Wrapp
if (!*fully_covered) {
return Status::OK();
}
+ const bool async_touch =
config::enable_file_cache_async_touch_on_get_or_set;
for (const auto* cell : covered_cells) {
- use_cell(*cell, blocks, need_to_move(cell->file_block->cache_type(),
context.cache_type),
- cache_lock);
+ const bool move_iter = need_to_move(cell->file_block->cache_type(),
context.cache_type);
+ use_cell(*cell, blocks, move_iter, cache_lock);
+ if (async_touch && move_iter) {
+ need_update_lru_blocks.push_back(cell->file_block);
+ }
}
return Status::OK();
}
diff --git a/be/src/io/cache/cached_remote_file_reader.cpp
b/be/src/io/cache/cached_remote_file_reader.cpp
index ef0c395b4c1..6a2abb5c847 100644
--- a/be/src/io/cache/cached_remote_file_reader.cpp
+++ b/be/src/io/cache/cached_remote_file_reader.cpp
@@ -104,6 +104,7 @@ bvar::PassiveStatus<int> g_active_peer_races_bvar(
bvar::Adder<uint64_t> g_peer_race_peer_win("peer_race_peer_win");
bvar::Adder<uint64_t> g_peer_race_s3_win("peer_race_s3_win");
bvar::Adder<uint64_t> g_peer_race_both_fail("peer_race_both_fail");
+bvar::Adder<uint64_t>
g_peer_race_s3_skipped_after_queue("peer_race_s3_skipped_after_queue");
bvar::Adder<uint64_t>
g_peer_cross_compute_group_read("peer_cross_compute_group_read");
bvar::Adder<uint64_t>
g_peer_same_compute_group_read("peer_same_compute_group_read");
bvar::Adder<uint64_t> g_peer_lazy_fetch_triggered("peer_lazy_fetch_triggered");
@@ -465,6 +466,20 @@ void launch_s3_race(std::shared_ptr<RaceState> race,
size_t empty_start, size_t
// peer wins; passing nullptr is safe because S3FileReader::read_at_impl
ignores it.
auto do_s3_read = [race, empty_start, span_size, remote_reader, owner]() {
(void)owner;
+
TEST_SYNC_POINT("CachedRemoteFileReader::_execute_winner_race::s3_task_start");
+ {
+ // The peer may win while this task waits in PeerRaceS3ThreadPool.
Skip the GET
+ // before allocating the span buffer or counting an S3 read.
+ std::unique_lock<bthread::Mutex> lk(race->mtx);
+ if (race->winner == 0) {
+ race->s3_done = true;
+ race->s3_status =
+ Status::InternalError<false>("skipped: peer won before
s3 task started");
+ race->cv.notify_all();
+ g_peer_race_s3_skipped_after_queue << 1;
+ return;
+ }
+ }
auto s3_buf = std::make_unique<char[]>(span_size);
size_t read_size = span_size;
s3_read_counter << 1;
diff --git a/be/test/io/cache/cached_remote_file_reader_peer_test.cpp
b/be/test/io/cache/cached_remote_file_reader_peer_test.cpp
index 07e83c93d96..d0a05908e52 100644
--- a/be/test/io/cache/cached_remote_file_reader_peer_test.cpp
+++ b/be/test/io/cache/cached_remote_file_reader_peer_test.cpp
@@ -28,6 +28,7 @@
#include <condition_variable>
#include <filesystem>
#include <fstream>
+#include <functional>
#include <mutex>
#include <string>
#include <thread>
@@ -53,6 +54,7 @@
namespace doris {
extern bvar::Adder<uint64_t> g_file_cache_get_by_peer_queue_timeout_num;
+extern bvar::Adder<uint64_t> g_file_cache_get_by_peer_get_or_set_calls;
Status test_handle_peer_file_cache_block_request(const PFetchPeerDataRequest*
request,
PFetchPeerDataResponse*
response,
@@ -64,6 +66,9 @@ bool
test_try_reject_if_queue_timed_out(std::chrono::steady_clock::time_point en
namespace doris::io {
extern bvar::Adder<uint64_t> peer_cache_reader_read_counter;
+extern bvar::Adder<uint64_t> s3_read_counter;
+extern bvar::Adder<uint64_t> g_peer_race_peer_win;
+extern bvar::Adder<uint64_t> g_peer_race_s3_skipped_after_queue;
namespace {
@@ -250,6 +255,9 @@ public:
: "";
}
+ if (before_response) {
+ before_response();
+ }
if (_options.fail_status) {
response->mutable_status()->set_status_code(TStatusCode::INTERNAL_ERROR);
return;
@@ -345,6 +353,8 @@ public:
std::atomic<int> request_block_count {0};
std::atomic<bool> support_attachment {false};
std::atomic<bool> request_fill {false};
+ // Runs on the RPC thread before the response is built; tests use it to
order the peer reply.
+ std::function<void()> before_response;
std::string get_last_fill_resource_id() const {
std::lock_guard lock(_request_mu);
return last_fill_resource_id;
@@ -994,6 +1004,166 @@ TEST_F(CachedRemoteFileReaderPeerTest,
peer_file_cache_handler_allows_cached_blo
create_peer_cache_block_request(file_path, 0, kPeerTestBlockSize,
content.size());
PFetchPeerDataResponse response;
brpc::Controller cntl;
+ const uint64_t get_or_set_before =
doris::g_file_cache_get_by_peer_get_or_set_calls.get_value();
+
+ auto st = doris::test_handle_peer_file_cache_block_request(&request,
&response, &cntl);
+
+ ASSERT_TRUE(st.ok()) << st;
+ EXPECT_EQ(response.status().status_code(), TStatusCode::OK);
+ ASSERT_EQ(response.datas_size(), 1);
+ EXPECT_EQ(response.datas(0).block_offset(), 0);
+ EXPECT_EQ(response.datas(0).data(), content.substr(0, kPeerTestBlockSize));
+ // A plain peer read is served by the read-only lookup and never reaches
get_or_set().
+ EXPECT_EQ(doris::g_file_cache_get_by_peer_get_or_set_calls.get_value() -
get_or_set_before, 0);
+}
+
+TEST_F(CachedRemoteFileReaderPeerTest,
peer_file_cache_handler_serves_fragmented_cached_layout) {
+ const std::string content = "abcdefghijklmnop";
+ const fs::path file_path =
+
create_peer_test_file("cached_remote_reader_peer_fragmented_layout.dat",
content);
+ Defer cleanup_file {[&]() {
+ std::error_code ec;
+ fs::remove(file_path, ec);
+ }};
+
+ const fs::path cache_path = caches_dir /
"cached_remote_reader_peer_fragmented_layout_cache";
+ Defer cleanup_cache {[&]() {
+ std::error_code ec;
+ fs::remove_all(cache_path, ec);
+ }};
+
+ clear_cached_remote_reader_factory();
+ BlockFileCache* cache = create_peer_test_cache(cache_path,
kPeerTestBlockSize);
+ // A pull-through fill from a client with a smaller cache block size
leaves short cells inside
+ // one of this cache's blocks. A block-sized plain peer read must still be
served.
+ populate_cache_block_with_content(cache, file_path, 0, kPeerTestBlockSize
/ 2, content);
+ populate_cache_block_with_content(cache, file_path, kPeerTestBlockSize / 2,
+ kPeerTestBlockSize / 2, content);
+
+ auto request =
+ create_peer_cache_block_request(file_path, 0, kPeerTestBlockSize,
content.size());
+ PFetchPeerDataResponse response;
+ brpc::Controller cntl;
+
+ auto st = doris::test_handle_peer_file_cache_block_request(&request,
&response, &cntl);
+
+ ASSERT_TRUE(st.ok()) << st;
+ EXPECT_EQ(response.status().status_code(), TStatusCode::OK);
+ ASSERT_EQ(response.datas_size(), 2);
+ EXPECT_EQ(response.datas(0).block_offset(), 0);
+ EXPECT_EQ(response.datas(0).data(), content.substr(0, kPeerTestBlockSize /
2));
+ EXPECT_EQ(response.datas(1).block_offset(), kPeerTestBlockSize / 2);
+ EXPECT_EQ(response.datas(1).data(),
+ content.substr(kPeerTestBlockSize / 2, kPeerTestBlockSize / 2));
+}
+
+TEST_F(CachedRemoteFileReaderPeerTest,
peer_file_cache_handler_queues_async_lru_touch) {
+ const bool old_async_touch =
config::enable_file_cache_async_touch_on_get_or_set;
+ const int64_t old_update_interval_ms =
+ config::file_cache_background_block_lru_update_interval_ms;
+ Defer restore_config {[&]() {
+ config::enable_file_cache_async_touch_on_get_or_set = old_async_touch;
+ config::file_cache_background_block_lru_update_interval_ms =
old_update_interval_ms;
+ }};
+ config::enable_file_cache_async_touch_on_get_or_set = true;
+ // Keep the background LRU worker from draining the queue before it is
inspected.
+ config::file_cache_background_block_lru_update_interval_ms = 60 * 60 *
1000;
+
+ const std::string content = "abcdefghijklmnop";
+ const fs::path file_path =
+
create_peer_test_file("cached_remote_reader_peer_async_lru_touch.dat", content);
+ Defer cleanup_file {[&]() {
+ std::error_code ec;
+ fs::remove(file_path, ec);
+ }};
+
+ const fs::path cache_path = caches_dir /
"cached_remote_reader_peer_async_lru_touch_cache";
+ Defer cleanup_cache {[&]() {
+ std::error_code ec;
+ fs::remove_all(cache_path, ec);
+ }};
+
+ clear_cached_remote_reader_factory();
+ BlockFileCache* cache = create_peer_test_cache(cache_path,
kPeerTestBlockSize);
+ populate_cache_block_with_content(cache, file_path, 0, kPeerTestBlockSize,
content);
+
+ auto request =
+ create_peer_cache_block_request(file_path, 0, kPeerTestBlockSize,
content.size());
+ PFetchPeerDataResponse response;
+ brpc::Controller cntl;
+ const size_t pending_before = cache->need_update_lru_blocks_size_unsafe();
+
+ auto st = doris::test_handle_peer_file_cache_block_request(&request,
&response, &cntl);
+
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(response.datas_size(), 1);
+ // Async touch defers the LRU move, so a served hit must be queued like a
get_or_set() hit.
+ EXPECT_EQ(cache->need_update_lru_blocks_size_unsafe(), pending_before + 1);
+}
+
+TEST_F(CachedRemoteFileReaderPeerTest,
peer_file_cache_handler_miss_returns_not_found) {
+ const std::string content = "abcdefghijklmnop";
+ const fs::path file_path =
create_peer_test_file("cached_remote_reader_peer_miss.dat", content);
+ Defer cleanup_file {[&]() {
+ std::error_code ec;
+ fs::remove(file_path, ec);
+ }};
+
+ const fs::path cache_path = caches_dir /
"cached_remote_reader_peer_miss_cache";
+ Defer cleanup_cache {[&]() {
+ std::error_code ec;
+ fs::remove_all(cache_path, ec);
+ }};
+
+ clear_cached_remote_reader_factory();
+ BlockFileCache* cache = create_peer_test_cache(cache_path,
kPeerTestBlockSize);
+ const auto hash = BlockFileCache::hash(file_path.filename().native());
+
+ auto request =
+ create_peer_cache_block_request(file_path, 0, kPeerTestBlockSize,
content.size());
+ PFetchPeerDataResponse response;
+ brpc::Controller cntl;
+ const uint64_t get_or_set_before =
doris::g_file_cache_get_by_peer_get_or_set_calls.get_value();
+
+ auto st = doris::test_handle_peer_file_cache_block_request(&request,
&response, &cntl);
+
+ ASSERT_TRUE(st.is<ErrorCode::NOT_FOUND>()) << st;
+ EXPECT_EQ(response.status().status_code(), TStatusCode::NOT_FOUND);
+ ASSERT_GE(response.status().error_msgs_size(), 1);
+ EXPECT_NE(response.status().error_msgs(0).find("cache block not fully
downloaded"),
+ std::string::npos);
+ EXPECT_EQ(response.datas_size(), 0);
+ EXPECT_EQ(doris::g_file_cache_get_by_peer_get_or_set_calls.get_value() -
get_or_set_before, 0);
+ EXPECT_FALSE(cache->_files.contains(hash));
+}
+
+TEST_F(CachedRemoteFileReaderPeerTest,
peer_file_cache_handler_serves_offset_inside_cached_block) {
+ const std::string content = "abcdefghijklmnop";
+ const fs::path file_path =
+
create_peer_test_file("cached_remote_reader_peer_offset_inside_block.dat",
content);
+ Defer cleanup_file {[&]() {
+ std::error_code ec;
+ fs::remove(file_path, ec);
+ }};
+
+ const fs::path cache_path = caches_dir /
"cached_remote_reader_peer_offset_inside_block_cache";
+ Defer cleanup_cache {[&]() {
+ std::error_code ec;
+ fs::remove_all(cache_path, ec);
+ }};
+
+ clear_cached_remote_reader_factory();
+ BlockFileCache* cache = create_peer_test_cache(cache_path,
kPeerTestBlockSize);
+ populate_cache_block_with_content(cache, file_path, 0, kPeerTestBlockSize,
content);
+
+ // A client with a smaller cache block size can request a range that
starts inside one of this
+ // cache's blocks. The server returns the covering block whole. This
checks the server only:
+ // the current client rejects a response block outside its requested
ranges, as with
+ // get_or_set().
+ auto request = create_peer_cache_block_request(file_path,
kPeerTestBlockSize / 2,
+ kPeerTestBlockSize / 2,
content.size());
+ PFetchPeerDataResponse response;
+ brpc::Controller cntl;
auto st = doris::test_handle_peer_file_cache_block_request(&request,
&response, &cntl);
@@ -2260,6 +2430,107 @@ protected:
.get_peer_candidates(kCrossTabletId);
}
+ // Let the peer win while the S3 task is held at `s3_sync_point`, wait for
the S3 task to
+ // finish, and report how many S3 reads and queue skips it recorded. With
+ // `peer_waits_for_s3_hold`, the peer replies only after S3 reaches the
sync point, so the
+ // peer cannot win before the S3 task starts.
+ void run_peer_win_with_s3_held_at(const std::string& name, const
std::string& s3_sync_point,
+ bool peer_waits_for_s3_hold, uint64_t*
s3_reads,
+ uint64_t* s3_skips) {
+ const std::string content = "abcdefghijklmnop";
+ const fs::path file_path = create_cross_cg_test_file(name + "_0.dat",
content);
+ Defer cleanup_file {[&]() {
+ std::error_code ec;
+ fs::remove(file_path, ec);
+ }};
+
+ const fs::path cache_path = caches_dir / (name + "_cache");
+ Defer cleanup_cache {[&]() {
+ std::error_code ec;
+ fs::remove_all(cache_path, ec);
+ }};
+
+ clear_cached_remote_reader_factory();
+ create_peer_test_cache(cache_path, kPeerTestBlockSize);
+
+ // Submit S3 immediately so only the S3 task itself can observe the
peer win.
+ const int32_t old_hedge_delay_ms = config::peer_race_hedge_delay_ms;
+ Defer restore_hedge_delay {
+ [&]() { config::peer_race_hedge_delay_ms = old_hedge_delay_ms;
}};
+ config::peer_race_hedge_delay_ms = 0;
+
+ std::mutex s3_hold_mutex;
+ std::condition_variable s3_hold_cv;
+ bool s3_held = false;
+ MockPeerCacheService service(content);
+ if (peer_waits_for_s3_hold) {
+ service.before_response = [&]() {
+ std::unique_lock lock(s3_hold_mutex);
+ s3_hold_cv.wait_for(lock, kPeerRpcWaitTimeout, [&] { return
s3_held; });
+ };
+ }
+ brpc::Server server;
+ auto addr = start_peer_test_server(&server, &service);
+ Defer stop_server {[&]() { stop_peer_test_server(&server); }};
+ register_cross_cg_candidate("127.0.0.1", addr.port);
+
+ const uint64_t peer_wins_before = g_peer_race_peer_win.get_value();
+ const uint64_t s3_reads_before = s3_read_counter.get_value();
+ const uint64_t s3_skips_before =
g_peer_race_s3_skipped_after_queue.get_value();
+
+ auto* sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+ SyncPoint::CallbackGuard s3_hold_guard;
+ sp->set_call_back(
+ s3_sync_point,
+ [&](auto&&) {
+ {
+ std::lock_guard lock(s3_hold_mutex);
+ s3_held = true;
+ }
+ s3_hold_cv.notify_all();
+ // collect_race_result counts the peer win only after
winner == 0 is set.
+ const auto deadline = std::chrono::steady_clock::now() +
kPeerRpcWaitTimeout;
+ while (g_peer_race_peer_win.get_value() ==
peer_wins_before &&
+ std::chrono::steady_clock::now() < deadline) {
+
std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ },
+ &s3_hold_guard);
+ Defer disable_sp {[&]() {
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+ }};
+
+ FileReaderSPtr local_reader;
+ ASSERT_TRUE(global_local_filesystem()->open_file(file_path.string(),
&local_reader).ok());
+
+ FileReaderOptions opts;
+ opts.cache_type = FileCachePolicy::FILE_BLOCK_CACHE;
+ opts.is_doris_table = true;
+ opts.mtime = 1;
+ opts.tablet_id = kCrossTabletId;
+ auto reader = std::make_shared<CachedRemoteFileReader>(local_reader,
opts);
+
+ std::string buffer(10, '#');
+ size_t bytes_read = 0;
+ IOContext io_ctx;
+ FileCacheStatistics cache_stats;
+ io_ctx.file_cache_stats = &cache_stats;
+
+ ASSERT_TRUE(
+ reader->read_at(1, Slice(buffer.data(), buffer.size()),
&bytes_read, &io_ctx).ok());
+ EXPECT_EQ(buffer, content.substr(1, 10));
+ EXPECT_EQ(bytes_read, 10);
+ EXPECT_EQ(cache_stats.num_peer_race_peer_win, 1);
+
+ // The losing S3 task outlives read_at(); wait for it before reading
its counters.
+ ExecEnv::GetInstance()->peer_race_s3_thread_pool()->wait();
+ *s3_reads = s3_read_counter.get_value() - s3_reads_before;
+ *s3_skips = g_peer_race_s3_skipped_after_queue.get_value() -
s3_skips_before;
+ }
+
private:
bool _old_enable_peer_s3_race = false;
int32_t _old_max_concurrent_peer_races = 64;
@@ -3418,6 +3689,38 @@ TEST_F(CrossCGWinnerRaceTest,
race_peer_stops_when_s3_wins) {
EXPECT_EQ(miss_service2.rpc_count.load(), 0) << "candidate 2 should be
skipped (S3 won)";
}
+// ----------------------------------------------------------------
+// Test: an S3 task that starts after the peer has won skips its GET.
+//
+// Holding S3 at s3_task_start models a task that waited in
PeerRaceS3ThreadPool while the peer
+// won. The task must return before counting or issuing the S3 read.
+// ----------------------------------------------------------------
+TEST_F(CrossCGWinnerRaceTest, s3_task_skips_read_when_peer_won_while_queued) {
+ uint64_t s3_reads = 0;
+ uint64_t s3_skips = 0;
+ run_peer_win_with_s3_held_at("s3_skip_queued",
+
"CachedRemoteFileReader::_execute_winner_race::s3_task_start",
+ /*peer_waits_for_s3_hold=*/false, &s3_reads,
&s3_skips);
+ EXPECT_EQ(s3_reads, 0);
+ EXPECT_EQ(s3_skips, 1);
+}
+
+// ----------------------------------------------------------------
+// Test: an S3 task that started before the peer won still reads.
+//
+// s3_before_read is after the queue check. The peer replies only once S3 is
held there, so the
+// task has passed the check before the peer wins and keeps the
read-then-discard behavior.
+// ----------------------------------------------------------------
+TEST_F(CrossCGWinnerRaceTest, s3_task_reads_when_started_before_peer_won) {
+ uint64_t s3_reads = 0;
+ uint64_t s3_skips = 0;
+ run_peer_win_with_s3_held_at("s3_read_started",
+
"CachedRemoteFileReader::_execute_winner_race::s3_before_read",
+ /*peer_waits_for_s3_hold=*/true, &s3_reads,
&s3_skips);
+ EXPECT_EQ(s3_reads, 1);
+ EXPECT_EQ(s3_skips, 0);
+}
+
// ----------------------------------------------------------------
// Test: cross-CG peer wins race and timer metrics are non-zero.
//
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
