github-actions[bot] commented on code in PR #69620:
URL: https://github.com/apache/doris/pull/69620#discussion_r4375086382
##########
be/src/cloud/cloud_internal_service.cpp:
##########
@@ -765,103 +861,46 @@ 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_block(cb_req, fb));
}
+ 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 looks up cached blocks, so probe() serves it
without inserting
+ // EMPTY cells or taking downloader ownership. probe() asserts
block-aligned offsets, and
+ // this offset comes from another BE whose cache block size may differ.
+ if (offset % cache->max_file_block_size() != 0) {
+ return not_found(fmt::format(
+ "unaligned peer cache block request, req_block=[offset={},
size={}], "
+ "max_file_block_size={}, {}",
+ cb_req.block_offset(), cb_req.block_size(),
cache->max_file_block_size(),
+ format_peer_request_context(request, hash, file_size)));
+ }
+ auto probe_start = std::chrono::steady_clock::now();
+ auto probe_result = cache->probe(hash, offset, size, ctx);
Review Comment:
[P1] Handle fragmented cached layouts before calling probe
The offset guard does not establish probe()'s cached-layout preconditions.
With a 4 MiB server cache block size and a file larger than 4 MiB, a 1 MiB
client's pull-through fill calls the retained get_or_set(hash, 0, 1 MiB),
publishing a short [0, 1 MiB) cell. An ordinary peer request for [0, 4 MiB),
during or after that fill, passes the modulo check, but probe() encounters this
cell and fails DORIS_CHECK(file_block->range().right >= expected_range.right)
before checking its state. The fill downloader's aligned get_or_set preserves
existing short cells, so this can also persist after fill completes. The failed
check escapes the peer-fetch worker without an exception boundary and
terminates the BE; the previous get_or_set path could serve the split layout.
Use a non-mutating lookup that tolerates existing fragmented cells, or reject
incompatible layouts with NOT_FOUND, and cover a smaller-block fill followed by
an aligned plain read.
##########
be/test/io/cache/cached_remote_file_reader_peer_test.cpp:
##########
@@ -3501,6 +3671,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",
+ &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, so the held task has already
passed it and keeps
+// the existing 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",
Review Comment:
[P2] Order peer completion after the S3 task passes its queue check
Holding s3_before_read only delays S3 after the new winner check; it does
not ensure the task reaches that point before the peer wins.
_execute_winner_race starts the peer bthread before submitting S3, and this
fixture's pool starts with zero worker threads, so the mock peer can publish
winner == 0 before the worker runs. The task then correctly skips before
invoking this callback, producing s3_reads == 0 and s3_skips == 1, contrary to
the assertions below. This makes the new already-started test fail under a
valid scheduling order. Gate the peer response with an explicit handshake until
S3 reaches s3_before_read, then allow the peer to win.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
