Skip to content

[TENT] Half-Open rail state machine + admit/isAvailable split #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Colors-111 wants to merge 2 commits into from
Closed

[TENT] Half-Open rail state machine + admit/isAvailable split #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
062fab6
[TENT] Half-Open rail state machine + admit/isAvailable split
Sep 15, 2026
f2f05c8
code format
Sep 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 42 additions & 3 deletions mooncake-transfer-engine/tent/include/tent/transport/rdma/rail_monitor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@ class RailMonitor {
"transports/rdma/rail_error_window_secs";
static constexpr const char *kCfgCooldownSecs =
"transports/rdma/rail_cooldown_secs";
static constexpr const char *kCfgProbeIntervalSecs =
"transports/rdma/rail_probe_interval_secs";

public:
RailMonitor() = default;
Expand All @@ -67,12 +69,28 @@ class RailMonitor {

bool ready() { return ready_; }

bool available(int local_nic, int remote_nic);
// Pure predicate: true iff the rail is Closed/healthy and directly usable.
// Does NOT mutate state and does NOT call updateBestMapping, so it is safe
// to call from updateBestMapping (no recursion) and from any query path.
bool isAvailable(int local_nic, int remote_nic) const;

// Mutating admit: the ONLY path that arms a probe/trial. Returns true when
// a transfer may use this rail -- Closed (no-op), or an exploratory probe /
// Half-Open trial. Sets probe_in_flight for paused rails so a second probe
// cannot race the first while it is on the wire. Callers that select a rail
// but fail to post must call cancelProbe() to clear the flag.
bool admit(int local_nic, int remote_nic);

void markFailed(int local_nic, int remote_nic);

void markRecovered(int local_nic, int remote_nic);

// Clear an armed probe/trial when a slice selected the rail via admit() but
// never reached the wire (e.g. endpoint creation failed, HW rejected the
// post). No-op when no probe is in flight, so callers may invoke it
// unconditionally on the re-queue path.
void cancelProbe(int local_nic, int remote_nic);

int findBestRemoteDevice(int local_nic, int remote_numa);

const Topology *local() const { return local_.get(); }
Expand Down Expand Up @@ -103,8 +121,28 @@ class RailMonitor {
std::chrono::seconds cooldown{0};
std::chrono::steady_clock::time_point last_error{};
std::chrono::steady_clock::time_point resume_time{};

// Derived: a rail is paused iff a resume_time has been armed.
// Last time admit() armed a probe/trial. Throttles the probe rate
// during Open and defers the first probe by one probe_interval_ after a
// pause.
std::chrono::steady_clock::time_point last_probe_time{};
// Half-Open: the cooldown expired but recovery is not yet proven.
// admit() arms exactly one trial; a trial success closes the rail
// (markRecovered), a trial failure escalates the cooldown and re-arms
// (markFailed). This replaces the old "expiry fully reopens" path,
// which slammed a still-dead peer with every slice and re-triggered the
// storm at 30s/60s/120s.
bool half_open = false;
// A probe/trial slice is selected/posted and its completion has not yet
// resolved the rail. admit() refuses to arm a second while this is set,
// so at most one probe/trial is on the wire per rail (single-threaded
// worker ownership makes a bool sufficient). Cleared by markFailed /
// markRecovered on completion, and by cancelProbe() on a pre-wire
// failure that never reaches the wire.
bool probe_in_flight = false;

// Derived: a rail is paused iff a resume_time has been armed. The
// Half-Open sub-state also has resume_time armed so paused() stays
// true.
bool paused() const {
return resume_time != std::chrono::steady_clock::time_point{};
}
Expand All @@ -117,6 +155,7 @@ class RailMonitor {
int error_threshold_ = 3;
std::chrono::seconds error_window_{10};
std::chrono::seconds cooldown_{30};
std::chrono::seconds probe_interval_{1};
};

} // namespace tent
Expand Down
147 changes: 121 additions & 26 deletions mooncake-transfer-engine/tent/src/transport/rdma/rail_monitor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,12 +77,15 @@ Status RailMonitor::load(std::shared_ptr<const Topology> local,
conf->get(kCfgErrorWindowSecs, (int)error_window_.count()));
cooldown_ = std::chrono::seconds(
conf->get(kCfgCooldownSecs, (int)cooldown_.count()));
probe_interval_ = std::chrono::seconds(
conf->get(kCfgProbeIntervalSecs, (int)probe_interval_.count()));
// Config is identical on every COW snapshot refresh. Log once per
// monitor so PD/e2e does not reprint the banner per slice/worker.
if (first_load) {
LOG(INFO) << "RailMonitor: error_threshold=" << error_threshold_
<< " error_window=" << error_window_.count() << "s"
<< " cooldown=" << cooldown_.count() << "s";
<< " cooldown=" << cooldown_.count() << "s"
<< " probe_interval=" << probe_interval_.count() << "s";
}
}
if (same_layout) return Status::OK();
Expand All @@ -94,26 +97,75 @@ Status RailMonitor::load(std::shared_ptr<const Topology> local,
return loadDefault();
}

bool RailMonitor::available(int local_nic, int remote_nic) {
bool RailMonitor::isAvailable(int local_nic, int remote_nic) const {
auto it = rail_states_.find(std::make_pair(local_nic, remote_nic));
if (it == rail_states_.end()) return false;
// True only for Closed/healthy rails. Open (cooldown running),
// expired-Open, and Half-Open all have resume_time armed -> paused() ->
// false. They must go through admit() (which may arm a probe/trial), not be
// used directly. This is a pure predicate: it mutates nothing and never
// calls updateBestMapping, so it is safe to call from updateBestMapping
// without the recursion the old available() had.
return !it->second.paused();
}

bool RailMonitor::admit(int local_nic, int remote_nic) {
auto it = rail_states_.find(std::make_pair(local_nic, remote_nic));
if (it == rail_states_.end()) return false;
auto& st = it->second;
// Closed: healthy and directly usable. No probe, no flag -- a normal
// transfer whose completion is the fast-path markRecovered no-op.
if (!st.paused()) return true;
// Half-Open: one trial is already admitted and unresolved. Block more
// traffic until the trial resolves (markRecovered closes, markFailed
// re-arms, cancelProbe reverts). resume_time stays armed so paused()
// remains true and the rail is not mistaken for Closed.
if (st.half_open) return false;
auto now = std::chrono::steady_clock::now();
if (now < st.resume_time) return false;
// Cooldown expired. Clear the pause so traffic resumes. Half-Open
// (admit one trial on expiry, escalate only on trial failure) is a
// follow-up: it needs available() split into a non-mutating predicate
// plus an admit() with in-flight tracking, so probes don't race the
// fallback/updateBestMapping callers that also call available().
st.resume_time = {};
st.error_count = 0;
updateBestMapping();
LOG(INFO) << "Rail recovered: local_nic=" << local_nic
<< " remote_nic=" << remote_nic
<< " (cooldown expired, cooldown_retained=" << st.cooldown.count()
<< "s)";
return true;
// Open: the cooldown timer is still running. Admit ONE exploratory probe
// every probe_interval_ so a successful probe closes the rail early (defect
// B, transient faults); a failed probe is a no-op for markFailed (was
// paused, defect A). probe_in_flight gates a second probe from racing the
// first while it is still on the wire.
if (now < st.resume_time) {
if (!st.probe_in_flight &&
(probe_interval_.count() == 0 ||
now - st.last_probe_time >= probe_interval_)) {
st.probe_in_flight = true;
st.last_probe_time = now;
return true;
}
return false;
}
// Expired-Open: the cooldown timer fired but no probe proved the rail
// healthy (it is still paused). Transition to Half-Open and admit exactly
// ONE trial; escalation happens only if the trial FAILS (markFailed), not
// because the clock fired -- elapsed time does not prove the path healthy.
// If an Open-phase probe is still in flight at expiry, wait for it.
if (!st.probe_in_flight) {
st.half_open = true;
st.probe_in_flight = true;
st.last_probe_time = now;
LOG(INFO) << "Rail half-open: local_nic=" << local_nic
<< " remote_nic=" << remote_nic
<< " (cooldown=" << st.cooldown.count()
<< "s retained, trial admitted)";
return true;
}
return false;
}

void RailMonitor::cancelProbe(int local_nic, int remote_nic) {
auto it = rail_states_.find(std::make_pair(local_nic, remote_nic));
if (it == rail_states_.end()) return;
auto& st = it->second;
if (st.probe_in_flight) {
st.probe_in_flight = false;
// Revert a Half-Open trial to expired-Open so the next admit() re-arms
// a fresh trial. A plain Open probe just clears its flag.
st.half_open = false;
// Leave resume_time, cooldown, last_probe_time untouched.
}
}

void RailMonitor::markFailed(int local_nic, int remote_nic) {
Expand All @@ -128,6 +180,32 @@ void RailMonitor::markFailed(int local_nic, int remote_nic) {
}
st.last_error = now;

// A probe/trial completion failed. Clear the in-flight flag and route by
// sub-state: a Half-Open trial failure escalates and re-arms (the ONLY
// escalation point for a paused rail); an Open-phase probe failure is a
// no-op (was paused, defect A). Either way this is not a fresh burst.
if (st.probe_in_flight) {
st.probe_in_flight = false;
if (st.half_open) {
st.half_open = false;
st.cooldown *= 2;
if (st.cooldown > kMaxCooldown) st.cooldown = kMaxCooldown;
st.error_count = 0; // fresh cycle
st.resume_time = now + st.cooldown;
st.last_probe_time = now; // defer next probe by one interval
LOG(INFO) << "Rail half-open trial failed; re-paused: local_nic="
<< local_nic << " remote_nic=" << remote_nic
<< " (cooldown escalated to " << st.cooldown.count()
<< "s)";
updateBestMapping();
}
// Open-phase probe failed: leave resume_time and cooldown untouched.
// The pause runs its course; admit() will arm the next probe after
// probe_interval_. Escalating here would let a sustained outage
// extend the pause indefinitely -- the original defect.
return;
}

const bool was_paused = st.paused();

if (st.error_count >= error_threshold_) {
Expand Down Expand Up @@ -157,22 +235,38 @@ void RailMonitor::markFailed(int local_nic, int remote_nic) {
<< error_window_.count()
<< "s, cooldown=" << st.cooldown.count() << "s)";
st.resume_time = now + st.cooldown;
// Defer the first probe by one probe_interval_: without this,
// last_probe_time defaults to epoch and admit() would arm a probe
// the instant the pause arms, defeating the throttle.
st.last_probe_time = now;
updateBestMapping();
}
// Already paused: leave resume_time and cooldown untouched. Re-arming
// or escalating here would let a sustained outage extend the pause
// indefinitely -- the original defect. The pause runs its course and
// available() reopens on cooldown expiry.
// Open-phase probing and Half-Open (admit one trial on expiry, escalate
// on trial failure) are follow-ups: they need available() split into a
// non-mutating predicate plus an admit() with in-flight tracking.
// Already paused (Open, no probe in flight): leave resume_time and
// cooldown untouched. Re-arming or escalating here would let a
// sustained outage extend the pause indefinitely -- the original
// defect. The pause runs its course; admit() probes recovery.
}
}

void RailMonitor::markRecovered(int local_nic, int remote_nic) {
auto it = rail_states_.find(std::make_pair(local_nic, remote_nic));
if (it == rail_states_.end()) return;
auto& st = it->second;
// A probe/trial succeeded: clear the in-flight flag and close the rail,
// resetting all backoff memory. This is the early-recovery path for both
// an Open-phase probe and a Half-Open trial.
if (st.probe_in_flight) {
st.probe_in_flight = false;
st.half_open = false;
st.error_count = 0;
st.resume_time = {};
st.cooldown = std::chrono::seconds(0);
st.last_probe_time = {};
LOG(INFO) << "Rail recovered: local_nic=" << local_nic
<< " remote_nic=" << remote_nic << " (probe/trial succeeded)";
updateBestMapping();
return;
}
// Fast path: a healthy rail stays healthy. 99%+ of completions land
// here, so we must not touch best_mapping_ or write any field.
if (!st.paused() && st.error_count == 0 && st.cooldown.count() == 0) return;
Expand All @@ -183,6 +277,7 @@ void RailMonitor::markRecovered(int local_nic, int remote_nic) {
st.error_count = 0;
st.resume_time = {};
st.cooldown = std::chrono::seconds(0);
st.last_probe_time = {};
if (was_paused) {
LOG(INFO) << "Rail recovered: local_nic=" << local_nic
<< " remote_nic=" << remote_nic
Expand Down Expand Up @@ -439,18 +534,18 @@ void RailMonitor::updateBestMapping() {
remote_nic =
remote_devices[remote_numa][i % remote_cnt];
}
if (!available(local_nic, remote_nic)) {
if (!isAvailable(local_nic, remote_nic)) {
bool found = false;
for (int cand : remote_devices[remote_numa]) {
if (available(local_nic, cand)) {
if (isAvailable(local_nic, cand)) {
remote_nic = cand;
found = true;
break;
}
}
if (!found) {
for (int cand = 0; cand < remote_nic_count; ++cand) {
if (available(local_nic, cand)) {
if (isAvailable(local_nic, cand)) {
remote_nic = cand;
break;
}
Expand Down
25 changes: 19 additions & 6 deletions mooncake-transfer-engine/tent/src/transport/rdma/workers.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -772,6 +772,12 @@ void Workers::asyncPostSend() {
discountFromOwner(worker, slice);
} else {
// The re-submit moves the count to the lane it lands on.
// No endpoint was obtained, so this slice never reached the
// wire on the rail admit() selected: clear any probe/trial
// it armed so the rail is not stranded.
if (auto* rail = slice->rail_monitor)
rail->cancelProbe(slice->source_dev_id,
slice->target_dev_id);
submitFromTick(worker, slice);
}
}
Expand Down Expand Up @@ -820,6 +826,11 @@ void Workers::asyncPostSend() {
updateSliceStatus(slice, FAILED);
discountFromOwner(worker, slice);
} else {
// Rejected by hardware before reaching the wire: clear any
// probe/trial admit() armed on the selected rail.
if (auto* rail = slice->rail_monitor)
rail->cancelProbe(slice->source_dev_id,
slice->target_dev_id);
submitFromTick(worker, slice);
}
}
Expand Down Expand Up @@ -1566,7 +1577,7 @@ Status Workers::selectOptimalDevice(RouteHint& source, RouteHint& target,
}

if (gdr_excluded ||
!rail.available(slice->source_dev_id, slice->target_dev_id)) {
!rail.admit(slice->source_dev_id, slice->target_dev_id)) {
VLOG(1) << "Optimal device pair not available: source_dev_id "
<< slice->source_dev_id << ", target_dev_id "
<< slice->target_dev_id;
Expand Down Expand Up @@ -1674,13 +1685,15 @@ Status Workers::selectFallbackDevice(RouteHint& source, RouteHint& target,
if (strictLocalNuma() &&
source.topo->isCrossNuma(*source.topo_entry, sdev))
continue;
bool reachable = same_machine ? (sdev == tdev) // loopback is safe
: rail_mon->available(sdev, tdev);

// Skip NICs that cannot GPUDirect-DMA to the source/target GPU.
if (reachable && gdr_learned &&
// Checked BEFORE admit() so a probe/trial is never armed on a rail
// that GDR will reject -- arming then discarding would orphan the
// in-flight flag and strand the rail.
if (!same_machine && gdr_learned &&
gdrPairExcluded(source, target, sdev, tdev, src_gpu, dst_gpu))
reachable = false;
continue;
bool reachable = same_machine ? (sdev == tdev) // loopback is safe
: rail_mon->admit(sdev, tdev);

if (reachable) {
// A retry gets here after the failure path returned the slice's
Expand Down
Loading
Loading