This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new a42928c5caf [fix](be) Avoid repeated peer connect failures (#67464)
a42928c5caf is described below
commit a42928c5caf1e064dbc4f6b7450430d49a1549e7
Author: deardeng <[email protected]>
AuthorDate: Tue Sep 8 16:02:14 2026 +0800
[fix](be) Avoid repeated peer connect failures (#67464)
Problem:
Peer reads create an uncached BRPC client for every request. When a peer
is unreachable, readers across tablets repeatedly connect to the same
address with the default 2-second connect timeout and 10 retries.
Impact:
An unavailable cache peer can repeatedly delay reads before
remote-storage fallback and generate unnecessary connection attempts and
warning logs.
Fix:
- Add a BE-wide, address-level circuit breaker shared by peer readers.
- Open the circuit after a configurable number of consecutive connection
or RPC failures (3 by default) for a configurable cooldown (30 seconds
by default).
- Reject reads while the circuit is open, allow one probe after
cooldown, and clear the failure state after a successful RPC.
- Make uncached BRPC connect timeout and retry count configurable at the
call site, and use a 200 ms timeout with no retries for peer reads.
Test:
Stop the peer test server and verify that four reads issue only three
RPC attempts, with the final read rejected by the circuit breaker.
---
be/src/cloud/config.cpp | 4 ++
be/src/cloud/config.h | 3 ++
be/src/io/cache/peer_file_cache_reader.cpp | 63 ++++++++++++++++++++--
be/src/util/brpc_client_cache.h | 7 +--
.../cache/cached_remote_file_reader_peer_test.cpp | 23 ++++++++
5 files changed, 93 insertions(+), 7 deletions(-)
diff --git a/be/src/cloud/config.cpp b/be/src/cloud/config.cpp
index 9dcf4daa627..b8475ef888d 100644
--- a/be/src/cloud/config.cpp
+++ b/be/src/cloud/config.cpp
@@ -166,6 +166,10 @@
DEFINE_mInt64(file_cache_warmup_download_rate_limit_bytes_per_second, "104857600
DEFINE_mInt64(peer_candidate_cleanup_interval_s, "3600"); // cleanup interval,
1 hour
DEFINE_mInt64(peer_candidate_expiry_s, "3600"); // candidate expiry,
1 hour
DEFINE_mInt32(peer_rpc_failure_eviction_threshold, "3"); // consecutive
failures to evict
+// Consecutive connection or RPC failures to one peer address before opening
its circuit.
+DEFINE_mInt32(cache_peer_read_failure_threshold, "3");
+// Seconds to reject reads to a peer address before allowing one recovery
probe.
+DEFINE_mInt32(cache_peer_read_circuit_open_seconds, "30");
DEFINE_mInt32(peer_all_miss_cooldown_threshold,
"5"); // consecutive all-miss races to trigger cooldown
DEFINE_mInt64(peer_all_miss_cooldown_duration_s, "300"); // cooldown duration,
5 minutes
diff --git a/be/src/cloud/config.h b/be/src/cloud/config.h
index 14f5cab309d..9bc1b11bdc6 100644
--- a/be/src/cloud/config.h
+++ b/be/src/cloud/config.h
@@ -209,6 +209,9 @@
DECLARE_mInt64(file_cache_warmup_download_rate_limit_bytes_per_second);
DECLARE_mInt64(peer_candidate_cleanup_interval_s);
DECLARE_mInt64(peer_candidate_expiry_s);
DECLARE_mInt32(peer_rpc_failure_eviction_threshold);
+// Address-level circuit breaker shared by peer reads across tablets.
+DECLARE_mInt32(cache_peer_read_failure_threshold);
+DECLARE_mInt32(cache_peer_read_circuit_open_seconds);
DECLARE_mInt32(peer_all_miss_cooldown_threshold);
DECLARE_mInt64(peer_all_miss_cooldown_duration_s);
diff --git a/be/src/io/cache/peer_file_cache_reader.cpp
b/be/src/io/cache/peer_file_cache_reader.cpp
index 76cffab8d44..e72d084304d 100644
--- a/be/src/io/cache/peer_file_cache_reader.cpp
+++ b/be/src/io/cache/peer_file_cache_reader.cpp
@@ -17,6 +17,7 @@
#include "io/cache/peer_file_cache_reader.h"
#include <brpc/controller.h>
+#include <bthread/mutex.h>
#include <butil/iobuf.h>
#include <bvar/latency_recorder.h>
#include <bvar/reducer.h>
@@ -25,8 +26,12 @@
#include <glog/logging.h>
#include <algorithm>
+#include <chrono>
+#include <mutex>
+#include <unordered_map>
#include <utility>
+#include "cloud/config.h"
#include "common/compiler_util.h" // IWYU pragma: keep
#include "common/metrics/doris_metrics.h"
#include "runtime/exec_env.h"
@@ -45,6 +50,51 @@ namespace doris::io {
namespace {
+struct PeerConnectionHealth {
+ int32_t consecutive_failures = 0;
+ std::chrono::steady_clock::time_point circuit_open_until;
+ bool probe_in_flight = false;
+};
+
+// Entries are removed after a successful connection. Add expiry cleanup if
permanently
+// unavailable peer addresses accumulate.
+bthread::Mutex peer_connection_health_mutex;
+std::unordered_map<std::string, PeerConnectionHealth> peer_connection_health;
+
+// Every true result must be paired with exactly one
finish_peer_connection_attempt() call.
+// Use Defer at the call site so every return path completes the attempt.
+bool peer_connection_circuit_allows(const std::string& address) {
+ std::unique_lock<bthread::Mutex> lock(peer_connection_health_mutex);
+ auto it = peer_connection_health.find(address);
+ if (it == peer_connection_health.end() ||
+ it->second.circuit_open_until == std::chrono::steady_clock::time_point
{}) {
+ return true;
+ }
+ auto& health = it->second;
+ if (std::chrono::steady_clock::now() < health.circuit_open_until ||
health.probe_in_flight) {
+ return false;
+ }
+ health.probe_in_flight = true;
+ return true;
+}
+
+// Completes an attempt admitted by peer_connection_circuit_allows().
+void finish_peer_connection_attempt(const std::string& address, bool
succeeded) {
+ std::unique_lock<bthread::Mutex> lock(peer_connection_health_mutex);
+ if (succeeded) {
+ peer_connection_health.erase(address);
+ return;
+ }
+ auto& health = peer_connection_health[address];
+ health.probe_in_flight = false;
+ ++health.consecutive_failures;
+ if (health.consecutive_failures >= std::max(1,
config::cache_peer_read_failure_threshold)) {
+ health.circuit_open_until =
+ std::chrono::steady_clock::now() +
+ std::chrono::seconds(std::max(0,
config::cache_peer_read_circuit_open_seconds));
+ }
+}
+
struct ExpectedPeerFetch {
std::vector<FileBlock::Range> expected_ranges;
std::vector<FileBlock::Range> pending_ranges;
@@ -212,15 +262,19 @@ Status PeerFileCacheReader::fetch_blocks(const
std::vector<FileBlockSPtr>& block
}
}
std::string brpc_addr = get_host_port(realhost, port);
- Status st = Status::OK();
+ if (!peer_connection_circuit_allows(brpc_addr)) {
+ return Status::RpcError<false>("Peer connection circuit is open for
{}", brpc_addr);
+ }
+ bool transport_succeeded = false;
+ Defer finish_connection_attempt {
+ [&] { finish_peer_connection_attempt(brpc_addr,
transport_succeeded); }};
std::shared_ptr<PBackendService_Stub> brpc_stub =
ExecEnv::GetInstance()->brpc_internal_client_cache()->get_new_client_no_cache(
- brpc_addr);
+ brpc_addr, "", "", "", 200, 0);
if (!brpc_stub) {
peer_cache_reader_failed_counter << 1;
LOG(WARNING) << "failed to get brpc stub " << brpc_addr;
- st = Status::RpcError<false>("Address {} is wrong", brpc_addr);
- return st;
+ return Status::RpcError<false>("Address {} is wrong", brpc_addr);
}
size_t filled = 0;
@@ -246,6 +300,7 @@ Status PeerFileCacheReader::fetch_blocks(const
std::vector<FileBlockSPtr>& block
if (cntl.Failed()) {
return Status::RpcError<false>(cntl.ErrorText());
}
+ transport_succeeded = true;
if (resp.has_status()) {
Status st2 = Status::create<false>(resp.status());
LOG_EVERY_N(WARNING, 1000) << "peer cache read failed, status=" <<
st2.msg();
diff --git a/be/src/util/brpc_client_cache.h b/be/src/util/brpc_client_cache.h
index 510fbb4afa7..9818b5247e5 100644
--- a/be/src/util/brpc_client_cache.h
+++ b/be/src/util/brpc_client_cache.h
@@ -253,7 +253,8 @@ public:
std::shared_ptr<T> get_new_client_no_cache(const std::string& host_port,
const std::string& protocol =
"",
const std::string&
connection_type = "",
- const std::string&
connection_group = "") {
+ const std::string&
connection_group = "",
+ int connect_timeout_ms = 2000,
int max_retry = 10) {
brpc::ChannelOptions options;
Status status =
doris::client::configure_brpc_channel_options(&options);
if (!status.ok()) {
@@ -276,9 +277,9 @@ public:
}
// Add random connection id to connection_group to make sure use new
socket
options.connection_group +=
std::to_string(_connection_id.fetch_add(1));
- options.connect_timeout_ms = 2000;
+ options.connect_timeout_ms = connect_timeout_ms;
options.timeout_ms = 2000;
- options.max_retry = 10;
+ options.max_retry = max_retry;
std::unique_ptr<FailureDetectChannel> channel(new
FailureDetectChannel());
int ret_code = 0;
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 0e8b79e4d1b..879f1607963 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
@@ -64,6 +64,8 @@ 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;
+
namespace {
constexpr size_t kPeerTestBlockSize = 4;
@@ -3491,4 +3493,25 @@ TEST_F(CrossCGWinnerRaceTest,
cross_cg_peer_wins_race_records_timer_metrics) {
<< "cross_cg_peer_io_timer should be non-zero after cross-CG peer
win";
}
+TEST_F(CachedRemoteFileReaderPeerTest,
peer_connection_failures_open_address_circuit) {
+ MockPeerCacheService service("abcd");
+ brpc::Server server;
+ const auto addr = start_peer_test_server(&server, &service);
+ stop_peer_test_server(&server);
+
+ const std::vector<FileBlockSPtr> blocks {
+ create_manual_peer_test_block("peer_connection_failure", 0,
kPeerTestBlockSize)};
+ const uint64_t rpc_count_before =
peer_cache_reader_read_counter.get_value();
+
+ for (int i = 0; i < 4; ++i) {
+ PeerFileCacheReader peer_reader(Path("peer_connection_failure"), true,
"127.0.0.1",
+ addr.port);
+ PeerFetchResult result;
+ const auto st = peer_reader.fetch_blocks(blocks, &result,
kPeerTestBlockSize, nullptr);
+ EXPECT_TRUE(st.is<ErrorCode::THRIFT_RPC_ERROR>()) << st;
+ }
+
+ EXPECT_EQ(peer_cache_reader_read_counter.get_value() - rpc_count_before,
3);
+}
+
} // namespace doris::io
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
