[Bugfix][TransferEngine] Fix RDMA handshake thundering herd#2793
[Bugfix][TransferEngine] Fix RDMA handshake thundering herd#2793dqxlzlz wants to merge 1 commit into
Conversation
When many nodes connect simultaneously during large-scale startup, the target's RPC server is overwhelmed with concurrent handshake requests, causing handshake latency to exceed the QP hardware retry timeout (~470ms). This leads to permanent 'transport retry counter exceeded' errors and zero throughput. Add a two-layer defense: 1. ConnectionLimiter: a per-transport counting semaphore limiting concurrent outbound handshakes. Configurable via MC_MAX_CONCURRENT_HANDSHAKES (default 16). 2. Exponential backoff on handshake failure (100ms -> 5s cap) in the worker pool. Validated on a 4-node cluster (32 NICs, 240 concurrent handshakes): handshake setup time dropped from ~96s to <1s. Co-authored-by: Tony Lu <tonylu@linux.alibaba.com>
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a ConnectionLimiter semaphore and an exponential handshake backoff mechanism to mitigate thundering-herd connection failures during large-scale startups. The review feedback highlights three key issues: a critical bug where slices in handshake backoff are immediately failed and cause CPU busy-looping, a potential deadlock risk if sendHandshake() throws an exception before releasing its handshake slot, and a need for defensive initialization of ConnectionLimiter to prevent deadlocks when configured with non-positive values.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| // Check handshake backoff before attempting connection | ||
| if (isInHandshakeBackoff(entry.first)) { | ||
| for (auto &slice : entry.second) | ||
| failed_slice_list.push_back(slice); | ||
| entry.second.clear(); | ||
| continue; | ||
| } |
There was a problem hiding this comment.
Critical Bug: Handshake Backoff Causes Immediate Transfer Failure & Busy-Looping
In the current implementation, when an endpoint is in handshake backoff, its slices are added to failed_slice_list and cleared from entry.second.
However, this triggers a severe issue:
- Slices in
failed_slice_listhave their retry count (slice->rdma.retry_cnt) incremented viashouldRetrySlice()and are immediately redispatched back to the queue. - Because the worker thread busy-loops very quickly, these slices will be picked up again almost instantly, find the endpoint still in backoff, increment their retry count again, and redispatch.
- Within microseconds, the slices will exhaust their maximum retries (
max_retry_cnt) and fail permanently, completely defeating the purpose of the backoff. - This also causes 100% CPU busy-looping on the worker thread during the backoff period.
Solution
We should collect these slices in a local list, redispatch them directly without incrementing their retry count, and introduce a small sleep (e.g., 10ms) to prevent high CPU usage.
// Check handshake backoff before attempting connection
if (isInHandshakeBackoff(entry.first)) {
SliceList backoff_slice_list;
for (auto &slice : entry.second)
backoff_slice_list.push_back(slice);
entry.second.clear();
redispatch(backoff_slice_list, thread_id);
std::this_thread::sleep_for(std::chrono::milliseconds(10));
continue;
}| context_.engine().acquireHandshakeSlot(); | ||
| int rc = context_.engine().sendHandshake(peer_server_name, local_desc, | ||
| peer_desc); | ||
| context_.engine().releaseHandshakeSlot(); |
There was a problem hiding this comment.
Exception Safety & Deadlock Risk
Directly calling acquireHandshakeSlot() and releaseHandshakeSlot() around sendHandshake() is risky. If sendHandshake() (or any internal metadata/RPC operation) throws an exception, or if an early return is added in the future, releaseHandshakeSlot() will be bypassed. This will permanently leak a slot in the ConnectionLimiter semaphore, eventually causing all future handshakes to block indefinitely (deadlock).
We should use a simple RAII guard to ensure the handshake slot is always released safely.
struct HandshakeSlotGuard {
RdmaTransport& engine;
explicit HandshakeSlotGuard(RdmaTransport& eng) : engine(eng) {
engine.acquireHandshakeSlot();
}
~HandshakeSlotGuard() {
engine.releaseHandshakeSlot();
}
};
HandshakeSlotGuard guard(context_.engine());
int rc = context_.engine().sendHandshake(peer_server_name, local_desc,
peer_desc);| explicit ConnectionLimiter(int max_concurrent) | ||
| : max_concurrent_(max_concurrent), current_(0) {} |
There was a problem hiding this comment.
Defensive Programming: Guard Against Invalid/Zero Max Concurrent Slots
If max_concurrent is configured to 0 or a negative value (e.g., via programmatic configuration or unexpected environment parsing), current_ < max_concurrent_ will always be false, causing acquire() to block permanently and deadlock the system.
We should defensively ensure max_concurrent_ is at least 1.
| explicit ConnectionLimiter(int max_concurrent) | |
| : max_concurrent_(max_concurrent), current_(0) {} | |
| explicit ConnectionLimiter(int max_concurrent) | |
| : max_concurrent_(max_concurrent > 0 ? max_concurrent : 1), current_(0) {} |
There was a problem hiding this comment.
Pull request overview
Mitigates RDMA handshake “thundering herd” failures in the Transfer Engine by adding per-transport admission control for outbound handshakes and introducing retry backoff behavior in the worker pool to avoid overwhelming the target handshake RPC server.
Changes:
- Add a per-transport
ConnectionLimiter(counting semaphore) configured viaMC_MAX_CONCURRENT_HANDSHAKES(default 16). - Introduce handshake failure tracking and exponential backoff state in
WorkerPoolwhen connection setup fails. - Add unit tests and CMake wiring for the new limiter + config parsing.
Reviewed changes
Copilot reviewed 10 out of 10 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| mooncake-transfer-engine/tests/connection_limiter_test.cpp | Adds behavioral tests for ConnectionLimiter under contention and blocking semantics. |
| mooncake-transfer-engine/tests/config_test.cpp | Adds env-var parsing tests for MC_MAX_CONCURRENT_HANDSHAKES. |
| mooncake-transfer-engine/tests/CMakeLists.txt | Builds and registers connection_limiter_test. |
| mooncake-transfer-engine/src/transport/rdma_transport/worker_pool.cpp | Adds handshake backoff logic and rail state tracking for connection setup failures. |
| mooncake-transfer-engine/src/transport/rdma_transport/rdma_transport.cpp | Instantiates the per-transport limiter from global config. |
| mooncake-transfer-engine/src/transport/rdma_transport/rdma_endpoint.cpp | Wraps outbound handshake RPC with limiter acquire/release calls. |
| mooncake-transfer-engine/src/config.cpp | Parses MC_MAX_CONCURRENT_HANDSHAKES into GlobalConfig. |
| mooncake-transfer-engine/include/transport/rdma_transport/worker_pool.h | Adds handshake backoff fields and constants to WorkerPool state. |
| mooncake-transfer-engine/include/transport/rdma_transport/rdma_transport.h | Introduces ConnectionLimiter and transport methods to acquire/release handshake slots. |
| mooncake-transfer-engine/include/config.h | Adds GlobalConfig::max_concurrent_handshakes with default 16. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| if (isInHandshakeBackoff(entry.first)) { | ||
| for (auto &slice : entry.second) | ||
| failed_slice_list.push_back(slice); | ||
| entry.second.clear(); | ||
| continue; | ||
| } |
| state.handshake_failures = | ||
| std::min(state.handshake_failures + 1, kHandshakeBackoffMaxFailures); |
| explicit ConnectionLimiter(int max_concurrent) | ||
| : max_concurrent_(max_concurrent), current_(0) {} |
| void release() { | ||
| std::unique_lock<std::mutex> lock(mu_); | ||
| --current_; | ||
| cv_.notify_one(); | ||
| } |
| // Acquire a handshake slot to limit concurrent connection storms. | ||
| context_.engine().acquireHandshakeSlot(); | ||
| int rc = context_.engine().sendHandshake(peer_server_name, local_desc, | ||
| peer_desc); | ||
| context_.engine().releaseHandshakeSlot(); |
| EXPECT_EQ(config.max_concurrent_handshakes, 8); | ||
| } | ||
|
|
||
| TEST_F(PkeyIndexEnvTest, MaxConcurrentHandshakesNonNumericKeepsDefault) { |
|
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
| if (isInHandshakeBackoff(entry.first)) { | ||
| for (auto &slice : entry.second) | ||
| failed_slice_list.push_back(slice); | ||
| entry.second.clear(); | ||
| continue; | ||
| } |
There was a problem hiding this comment.
The current backoff branch may incorrectly consume the slice retry budget and cause busy-waiting.
| context_.engine().acquireHandshakeSlot(); | ||
| int rc = context_.engine().sendHandshake(peer_server_name, local_desc, | ||
| peer_desc); | ||
| context_.engine().releaseHandshakeSlot(); |
Description
The problem we hit in production
In a large-scale PD-disaggregated deployment, when many nodes start up and connect
at the same time, some QPs get stuck in a permanently degraded state: they report
transport retry counter exceededand their throughput drops to ~0 and neverrecovers.
We reproduced this on a real cluster. With 240 concurrent handshakes hitting a
single target (unpatched build):
This confirms the target's handshake RPC server gets overwhelmed by the connection
storm.
Why this happens (root cause)
The handshake path has no admission control: every node fires all of its
handshake RPCs at once. When 40+ nodes x 8 NICs start simultaneously, hundreds of
handshake requests pile up in the target's RPC queue. Meanwhile the RDMA QP has a
fixed hardware retry window (
kTimeout=14~67ms xkRetryCount=7~= 470ms).The mismatch is fatal: the RPC queue latency (tens of seconds) far exceeds the QP's
470ms retry window. Before the handshake response arrives, the QP's hardware retry
counter is exhausted, the QP transitions to
ERROR, and it never recovers -throughput stays at 0.
What this PR does
A two-layer defense:
outbound handshakes so a single instance can't flood the target. Configurable via
the
MC_MAX_CONCURRENT_HANDSHAKESenvironment variable (default: 16).exponentially (100ms -> 200ms -> ... -> 5s cap) in the worker pool, giving the
overwhelmed target room to drain its queue and recover.
After the fix, the same 240-concurrent test brings handshake setup time back down
from ~96s to < 1s, keeping it well under the 470ms hardware retry window.
Module
mooncake-transfer-engine)mooncake-store)mooncake-ep)mooncake-pg)mooncake-integration)mooncake-p2p-store)mooncake-wheel)mooncake-common)mooncake-rl)Type of Change
How Has This Been Tested?
Unit tests
Added
connection_limiter_test.cpp(behavioral tests for the semaphore) andextended
config_test.cpp(env-var parsing forMC_MAX_CONCURRENT_HANDSHAKES).ConnectionLimiternever exceeds the configured max concurrent slots under32-thread contention.
acquire()blocks when full and unblocks onrelease().MC_MAX_CONCURRENT_HANDSHAKESparsing: default 16, valid override applied,zero/negative/non-numeric values rejected.
All unit tests pass.
Cluster stress test (thundering herd)
Tested on a 4-node cluster, 8x mlx5_bond NICs per node (32 NICs total), with a
target node under 240 concurrent handshake connections from 3 initiator nodes.
Test commands:
Observed results (directly measured):
Note: the cluster repro directly reproduces the handshake queueing delay
(~96s >> the 470ms QP retry window) and the resulting throughput degradation.
The permanent QP
ERRORtransition is the downstream production symptomexplained in the root-cause section; it was not separately captured as a log
line in this fresh-connection benchmark.
Checklist
./scripts/code_format.shpre-commit run --all-filesand all hooks passAI Assistance Disclosure
An AI coding assistant helped implement the fix and orchestrate the multi-node
stress tests across the 4-node cluster. The human submitter has reviewed every
changed line and can defend the change end-to-end.