[Transfer Engine] Refresh RDMA metadata on HCA and GID change events - #2878
Conversation
![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 implements GID and topology refresh mechanisms in the RDMA transport engine to handle asynchronous InfiniBand events like GID and LID changes. The review feedback highlights a critical undefined behavior in WorkerPool::refreshPublishedLocalTopology() where moving a shared pointer while accessing its member in the same function call can cause a crash. Additionally, it suggests querying port attributes unconditionally to avoid publishing stale LIDs, and lowering the log level for unchanged GID events to prevent log clutter.
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.
| context_.engine().metadata_->addLocalSegment( | ||
| LOCAL_SEGMENT_ID, updated_desc->name, std::move(updated_desc)); | ||
| int ret = context_.engine().metadata_->updateLocalSegmentDesc(); | ||
| if (ret) { | ||
| LOG(WARNING) << "Failed to publish RDMA topology update for " | ||
| << context_.deviceName() << ", ret=" << ret; | ||
| } |
There was a problem hiding this comment.
In WorkerPool::refreshPublishedLocalTopology(), the call to addLocalSegment passes updated_desc->name as the second argument and std::move(updated_desc) as the third argument.
In C++, the order of evaluation of function arguments is unspecified. If the compiler evaluates the third argument (std::move(updated_desc)) before the second argument (updated_desc->name), updated_desc will be moved-from (leaving its internal pointer null) by the time updated_desc->name is evaluated. This will result in a null-pointer dereference and a crash at runtime.
To prevent this undefined behavior, copy the segment name to a local variable before calling addLocalSegment.
std::string segment_name = updated_desc->name;
context_.engine().metadata_->addLocalSegment(
LOCAL_SEGMENT_ID, segment_name, std::move(updated_desc));
int ret = context_.engine().metadata_->updateLocalSegmentDesc();
if (ret) {
LOG(WARNING) << "Failed to publish RDMA topology update for "
<< context_.deviceName() << ", ret=" << ret;
}| bool RdmaContext::refreshCurrentGid(std::string *previous_gid, | ||
| std::string *next_gid) { | ||
| std::lock_guard<std::mutex> reprobe_guard(gid_reprobe_lock_); | ||
| std::string current_gid_string; | ||
| int current_gid_index = -1; | ||
| int next_gid_index = -1; | ||
| uint16_t current_lid = 0; | ||
| ibv_context *current_context = nullptr; | ||
| uint8_t current_port = 0; | ||
| bool auto_gid_selection_enabled = false; | ||
| { | ||
| std::lock_guard<std::mutex> guard(gid_lock_); | ||
| if (!context_) { | ||
| return false; | ||
| } | ||
| current_gid_index = gid_index_; | ||
| current_gid_string = gidBytesToString(gid_.raw); | ||
| current_lid = lid_; | ||
| current_context = context_; | ||
| current_port = port_; | ||
| auto_gid_selection_enabled = auto_gid_selection_enabled_; | ||
| } | ||
|
|
||
| if (auto_gid_selection_enabled) { | ||
| ibv_port_attr port_attr; | ||
| if (ibv_query_port(current_context, current_port, &port_attr)) { | ||
| PLOG(WARNING) << "Failed to refresh port attributes on " | ||
| << device_name_ << "/" | ||
| << static_cast<int>(current_port); | ||
| return false; | ||
| } | ||
|
|
||
| std::vector<AutoGidCandidate> candidates; | ||
| candidates.reserve(port_attr.gid_tbl_len); | ||
| for (int i = 0; i < port_attr.gid_tbl_len; ++i) { | ||
| AutoGidCandidate candidate; | ||
| candidate.gid_index = i; | ||
|
|
||
| struct ibv_gid_entry gid_entry; | ||
| if (ibv_query_gid_ex(current_context, current_port, i, &gid_entry, | ||
| 0)) { | ||
| candidate.query_succeeded = false; | ||
| candidates.push_back(candidate); | ||
| continue; | ||
| } | ||
|
|
||
| const auto *gid_addr = | ||
| reinterpret_cast<const struct in6_addr *>(gid_entry.gid.raw); | ||
| std::string ndev = readGidNdev(device_name_, current_port, i); | ||
| candidate.gid = gidBytesToString(gid_entry.gid.raw); | ||
| candidate.gid_type = gid_entry.gid_type; | ||
| candidate.has_network_device = !ndev.empty(); | ||
| candidate.is_ipv4_mapped = ipv6_addr_v4mapped(gid_addr); | ||
| candidate.is_link_local_ipv6 = isLinkLocalIpv6(gid_addr); | ||
| candidate.is_overlay_network = | ||
| candidate.has_network_device && isOverlayNetwork(ndev); | ||
| candidate.is_overlay_ipv4 = | ||
| candidate.is_ipv4_mapped && isOverlayIPv4(gid_addr); | ||
| candidate.is_null_gid = isNullGid(&gid_entry.gid); | ||
| candidates.push_back(candidate); | ||
| } | ||
|
|
||
| auto selection = selectBestAutoGidCandidate(candidates); | ||
| if (!selection.has_value()) { | ||
| LOG(WARNING) << "No suitable GID found while refreshing " | ||
| << device_name_ << "/" | ||
| << static_cast<int>(current_port); | ||
| return false; | ||
| } | ||
| next_gid_index = selection->gid_index; | ||
| } else { | ||
| next_gid_index = current_gid_index; | ||
| } | ||
|
|
||
| ibv_gid new_gid = {}; | ||
| std::string next_gid_string; | ||
| if (ibv_query_gid(current_context, current_port, next_gid_index, | ||
| &new_gid)) { | ||
| return false; | ||
| } | ||
| if (isNullGid(&new_gid)) { | ||
| return false; | ||
| } | ||
| next_gid_string = gidBytesToString(new_gid.raw); | ||
|
|
||
| if (next_gid_index == current_gid_index && | ||
| next_gid_string == current_gid_string) { | ||
| if (next_gid) *next_gid = current_gid_string; | ||
| return false; | ||
| } | ||
|
|
||
| int publish_ret = engine_.refreshLocalDeviceDesc(device_name_, current_lid, | ||
| next_gid_string); | ||
| if (publish_ret) { | ||
| LOG(ERROR) << "Failed to refresh local device descriptor for " | ||
| << device_name_ << ": " << publish_ret; | ||
| return false; | ||
| } | ||
|
|
||
| { | ||
| std::lock_guard<std::mutex> guard(gid_lock_); | ||
| gid_ = new_gid; | ||
| gid_index_ = next_gid_index; | ||
| } | ||
| if (previous_gid) *previous_gid = current_gid_string; | ||
| if (next_gid) *next_gid = next_gid_string; | ||
|
|
||
| LOG(WARNING) << "Refreshed GID on " << device_name_ << "/" | ||
| << static_cast<int>(port_) << ": index " << current_gid_index | ||
| << " (" << current_gid_string << ") -> " << next_gid_index | ||
| << " (" << next_gid_string << ")"; | ||
| return true; | ||
| } |
There was a problem hiding this comment.
In RdmaContext::refreshCurrentGid(), the LID (current_lid) is read from the cached lid_ member variable. However, if a port event (like IBV_EVENT_LID_CHANGE) has occurred, the LID of the HCA port may have changed.
Currently, ibv_query_port is only called if auto_gid_selection_enabled is true. If it is false, or if we don't update lid_ with the queried value, we will publish a stale LID to the metadata server via refreshLocalDeviceDesc(). This can prevent peers from establishing connections to this node on InfiniBand networks.
To fix this, query the port attributes unconditionally to get the latest LID, use it for publishing, and update lid_ in the gid_lock_ block at the end of the function.
bool RdmaContext::refreshCurrentGid(std::string *previous_gid,
std::string *next_gid) {
std::lock_guard<std::mutex> reprobe_guard(gid_reprobe_lock_);
std::string current_gid_string;
int current_gid_index = -1;
int next_gid_index = -1;
uint16_t current_lid = 0;
ibv_context *current_context = nullptr;
uint8_t current_port = 0;
bool auto_gid_selection_enabled = false;
{
std::lock_guard<std::mutex> guard(gid_lock_);
if (!context_) {
return false;
}
current_gid_index = gid_index_;
current_gid_string = gidBytesToString(gid_.raw);
current_lid = lid_;
current_context = context_;
current_port = port_;
auto_gid_selection_enabled = auto_gid_selection_enabled_;
}
ibv_port_attr port_attr;
if (ibv_query_port(current_context, current_port, &port_attr)) {
PLOG(WARNING) << "Failed to query port attributes on "
<< device_name_ << "/"
<< static_cast<int>(current_port);
return false;
}
uint16_t latest_lid = port_attr.lid;
if (auto_gid_selection_enabled) {
std::vector<AutoGidCandidate> candidates;
candidates.reserve(port_attr.gid_tbl_len);
for (int i = 0; i < port_attr.gid_tbl_len; ++i) {
AutoGidCandidate candidate;
candidate.gid_index = i;
struct ibv_gid_entry gid_entry;
if (ibv_query_gid_ex(current_context, current_port, i, &gid_entry,
0)) {
candidate.query_succeeded = false;
candidates.push_back(candidate);
continue;
}
const auto *gid_addr =
reinterpret_cast<const struct in6_addr *>(gid_entry.gid.raw);
std::string ndev = readGidNdev(device_name_, current_port, i);
candidate.gid = gidBytesToString(gid_entry.gid.raw);
candidate.gid_type = gid_entry.gid_type;
candidate.has_network_device = !ndev.empty();
candidate.is_ipv4_mapped = ipv6_addr_v4mapped(gid_addr);
candidate.is_link_local_ipv6 = isLinkLocalIpv6(gid_addr);
candidate.is_overlay_network =
candidate.has_network_device && isOverlayNetwork(ndev);
candidate.is_overlay_ipv4 =
candidate.is_ipv4_mapped && isOverlayIPv4(gid_addr);
candidate.is_null_gid = isNullGid(&gid_entry.gid);
candidates.push_back(candidate);
}
auto selection = selectBestAutoGidCandidate(candidates);
if (!selection.has_value()) {
LOG(WARNING) << "No suitable GID found while refreshing "
<< device_name_ << "/"
<< static_cast<int>(current_port);
return false;
}
next_gid_index = selection->gid_index;
} else {
next_gid_index = current_gid_index;
}
ibv_gid new_gid = {};
std::string next_gid_string;
if (ibv_query_gid(current_context, current_port, next_gid_index,
&new_gid)) {
return false;
}
if (isNullGid(&new_gid)) {
return false;
}
next_gid_string = gidBytesToString(new_gid.raw);
bool lid_changed = (latest_lid != current_lid);
if (next_gid_index == current_gid_index &&
next_gid_string == current_gid_string &&
!lid_changed) {
if (next_gid) *next_gid = current_gid_string;
return false;
}
int publish_ret = engine_.refreshLocalDeviceDesc(device_name_, latest_lid,
next_gid_string);
if (publish_ret) {
LOG(ERROR) << "Failed to refresh local device descriptor for "
<< device_name_ << ": " << publish_ret;
return false;
}
{
std::lock_guard<std::mutex> guard(gid_lock_);
gid_ = new_gid;
gid_index_ = next_gid_index;
lid_ = latest_lid;
}
if (previous_gid) *previous_gid = current_gid_string;
if (next_gid) *next_gid = next_gid_string;
LOG(WARNING) << "Refreshed GID on " << device_name_ << "/"
<< static_cast<int>(port_) << ": index " << current_gid_index
<< " (" << current_gid_string << ") -> " << next_gid_index
<< " (" << next_gid_string << ")";
return true;
}| } else { | ||
| LOG(WARNING) << "Worker: received GID change event for " | ||
| << context_.deviceName() | ||
| << ", but current GID did not change"; | ||
| } |
There was a problem hiding this comment.
In WorkerPool::refreshPublishedLocalGid(), when a GID change event is received but the GID did not actually change, it logs a warning:
LOG(WARNING) << "Worker: received GID change event for ..., but current GID did not change";
In InfiniBand/RoCE environments, GID change events can be triggered frequently for other ports or GID indices that are not selected by this context. Logging this as a WARNING can clutter the logs with non-actionable warnings. It is highly recommended to change this to LOG(INFO) or VLOG(1).
} else {
LOG(INFO) << "Worker: received GID change event for "
<< context_.deviceName()
<< ", but current GID did not change";
}
|
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
he-yufeng
left a comment
he-yufeng
left a comment
There was a problem hiding this comment.
Nice — the auto-GID-vs-explicit-index split reads cleanly, and the async-event handling is correct: the GID_CHANGE branch acks and sets event_acked = true, and the trailing ack is guarded by if (!event_acked), so there's no double-ack.
One correctness concern on the failure path. refreshCurrentGid() returns false for two very different situations: a genuine no-op (the resolved GID equals the current one) and a hard failure (ibv_query_port / ibv_query_gid fails, no suitable candidate, null GID, or refreshLocalDeviceDesc fails to publish). refreshPublishedLocalGid() collapses both into changed == false, logs "current GID did not change", and doProcessContextEvents() then skips disconnectAllEndpoints(). So if a real IBV_EVENT_GID_CHANGE fires but we fail to query or publish the new GID, we log it as a benign no-change and leave every endpoint sitting on the old GID — which is exactly the case this event is meant to recover from, since after a GID change the previous GID may no longer be routable. The stale-GID connections would then fail silently until something else tears them down.
Could refreshCurrentGid distinguish "no change" from "failed to refresh" — e.g. a tri-state return, or an out-param/error code — so the failure path can log at ERROR and still disconnect (or trigger a retry) instead of being swallowed as "did not change"? Even just returning a distinct value on the publish/query failures would let the caller treat them differently.
Minor: the MakeNicPath(peer_segment_desc->name → nicPathServerName()) change in submitPostSend looks like a necessary companion to the republished topology names rather than part of the GID/HCA event handling — worth calling out in the description, and confirming the rail-failover path is exercised with the new accessor so the alt-path key still matches what's registered.
|
@he-yufeng Thanks, this is a good catch. I updated the GID refresh path to distinguish the three cases: UNCHANGED, CHANGED or FAILED. The IBV_EVENT_GID_CHANGE handler now only keeps endpoints when the result is explicitly UNCHANGED. For both CHANGED and FAILED, it disconnects all endpoints. The failure case is logged at ERROR, so we no longer swallow a real refresh/publish failure as a benign no-op or leave old-GID endpoints alive. Also agreed on the nicPathServerName() change. I’ll call that out separately in the PR description as a companion fix for alternate rail key construction in dual-NIC / rdma_server_name setups, so the alt-path key matches the registered endpoint/rail state. |
yuechen-sys
left a comment
yuechen-sys
left a comment
There was a problem hiding this comment.
I test it in two-node NIC flap experiment. This implementation is effective.
…vcache-ai#2878) * Update metadata for bad RNICs or GID changes * Code format change * Update GID change results
4 participants
Description
This PR improves RDMA Transfer Engine fault tolerance when a local HCA becomes unavailable or its GID changes at runtime.
Previously, a node could continue publishing metadata that allowed peers to select a failed HCA, or peers could keep rebuilding endpoints with stale GID information after a runtime GID change. This made recovery depend heavily on repeated endpoint failures.
This change updates the local published RDMA metadata when verbs async events indicate local path state changes:
IBV_EVENT_PORT_ACTIVE, mark the context active again and republish topology based on all current context states.IBV_EVENT_GID_CHANGE, refresh the local GID, update the publishedDeviceDesc.gid, and disconnect old endpoints so future handshakes rebuild with current metadata.devices[]stable and disable failed HCAs only through topology selection, preservingdevices[]/rkey[]index alignment.nicPathServerName()when constructing alternate peer NIC paths, fixing dual-NIC /rdma_server_namepath-key mismatches.Notes
This PR does not implement metadata cache TTL (#2795). Peers still need an existing force-refresh path or a future TTL-based refresh change to observe republished metadata automatically after their cache expires.
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?
Test commands:
# Example: bash scripts/run_ci_test.shTest results:
Checklist
./scripts/code_format.shpre-commit run --all-filesand all hooks passAI Assistance Disclosure