Struct ElChe 

pub struct ElChe { /* private fields */ }

Implementations

impl ElChe

pub fn new(world_size: usize, anchor: usize) -> Self

Create a new sync cadence.

world_size: number of devices (must be >= 2). anchor: initial batch count for the slow device per sync step.

The first step uses equal counts (anchor for every device). After report_timing, ratios adapt to measured throughput.

pub fn phase(&self) -> Phase

Current lifecycle phase.

pub fn anchor_rank(&self) -> Option<usize>

Currently elected slow-anchor rank (None until first calibration).

pub fn to_state(&self) -> ElCheState

Snapshot this ElChe’s trajectory state for checkpoint persistence.

Captures the fields a resume API needs to restore cadence behavior without re-calibrating from scratch: anchor, elected anchor rank, per-rank smoothed ms_per_batch (trust-window mean), phase, calibration count. User-set knobs (overhead_target, min_anchor, max_anchor, max_batch_diff) are NOT captured — they come from the user’s DdpRunConfig at controller construction on resume, so a re-bind to different knobs is supported.

Called by ClusterCoordinator when broadcasting ShutdownWithSave; the produced state is written to <save_path>.meta.json via CheckpointMeta::with_elche_state.

pub fn restore_from_state(&mut self, state: &ElCheState) -> Result<()>

Restore the dynamic trajectory fields from a previously-captured crate::distributed::ElCheState snapshot. Inverse of Self::to_state.

Restored: anchor, anchor_rank, phase, calibration_count, calibrated (true when any rank had a positive smoothed reading), and the per-rank trust window seeded with the saved smoothed_ms_per_batch. The user-set knobs (overhead_target, min_anchor, max_anchor, max_batch_diff) are left as-is on self — the caller already configured those from the user’s DdpRunConfig at construction time, and resume by design supports re-binding to different knobs.

state.world_size must match self.world_size; the saved smoothed_ms_per_batch length is the authoritative check. A mismatch surfaces loudly so callers don’t silently resume a 3-rank snapshot into a 2-rank cluster (a config-coherence bug).

Window seeding is lossy by design: the snapshot only carries the trust-window mean, not the raw samples. We seed each rank’s window with one sample equal to the mean. The first few post-resume report_timing calls re-populate raw samples and the smoothed signal converges back to actual conditions within TRUST_WINDOW_CAP calibrations.

pub fn with_overhead_target(self, target: f64) -> Self

Set the target per-window FIXED overhead (reduce + fill) as a fraction of the bottleneck rank’s window wall.

Default: 0.05 (5%). The anchor auto-tunes upward to keep this per-window overhead below the target — fewer, larger windows amortize the fixed cost. Lower values = fewer syncs = larger window = more gradient staleness (bounded by the convergence guard and the epoch cap). Clamped to [0.01, 0.50].

pub fn with_window_growth_applicable(self, applicable: bool) -> Self

Enable or disable window-pressure anchor growth for the active reduce policy.

Growth amortizes per-window fixed cost (reduce + fill) across a cadence window, so it is only meaningful when reduces are windowed (Cadence / Async). Under Sync the reduce fires every step, so there is no window to amortize; leaving growth on there would inflate the telemetry anchor and the checkpointed ElCheState.anchor, mis-seeding a later Cadence resume. The coordinator calls this with false in Sync mode and true otherwise. Default (unset): true.

pub fn with_max_anchor(self, max: usize) -> Self

Set the maximum anchor count (gradient staleness limit).

Default: 1000. Higher values allow fewer syncs but accumulate more batches of gradient before averaging. Set to 1 to sync after every slow-device batch (minimal accumulation, traditional DDP cadence). The overhead auto-tune typically settles well below this cap; the default exists primarily as a safety net against runaway growth.

pub fn set_max_total_batches(&mut self, max_total: usize)

Cap the total reduce window (sum(batch_counts)) at max_total batches — set by the coordinator to the epoch’s batch count so a reduce window can never grow past one dataset pass. The overhead auto-tune still grows the schedule to amortize sync cost, but recompute_batch_counts scales the per-rank counts down proportionally if their sum would exceed this bound. None (default) leaves the window unbounded.

pub fn with_min_anchor(self, min: usize) -> Self

Set the minimum anchor count (overhead auto-tune floor).

Default: equals the initial anchor (so the auto-tune cannot shrink below the start value). Set explicitly to force the auto-tune above its natural overhead-equilibrium setpoint, or together with Self::with_max_anchor (same value) to pin the anchor at a fixed cadence.

Note: only the overhead auto-tune’s shrink path honors this floor. The convergence guard’s NudgeDown, which routes through Self::nudge_anchor_down, bypasses it (treated as a stronger signal than overhead). For hard pinning, also disable the convergence guard at the crate::distributed::ddp_run::DdpRunConfig level.

pub fn with_max_batch_diff(self, max: usize) -> Self

Set the maximum batch difference between fastest and slowest worker.

When the fastest worker leads the slowest by more than this many batches, it is throttled (paused) until the gap closes. This prevents catastrophic divergence with large batches or extreme speed ratios.

• None (default): no limit, workers run freely.
• Some(0): strict lockstep, equivalent to synchronous DDP.
• Some(n): fast workers may lead by at most n batches.

pub fn max_batch_diff(&self) -> Option<usize>

Current max batch diff setting.

pub fn with_speed_ratio(self, slow_rank: usize, ratio: f64) -> Self

Set initial speed estimate before the first timing measurement.

slow_rank: which device is slowest (receives anchor batches). ratio: how many times faster the fastest device is (e.g., 3.0 means the fast GPU processes ~3x more batches per unit time).

Default (without this call): all devices start equal (anchor batches each). After the first report_timing, actual measurements replace this estimate, so even a wrong guess self-corrects in one step.

// RTX 5060 Ti (rank 0) is ~2.3x faster than GTX 1060 (rank 1)
let che = ElChe::new(2, 10).with_speed_ratio(1, 2.3);
// → rank 0: 23 batches, rank 1: 10 batches

pub fn with_initial_anchor(self, slow_rank: usize) -> Self

Pin the initial slow-anchor rank without committing to a speed ratio.

Used by the coordinator when the user supplies partition_ratios (smallest ratio = slow rank), or by with_device_indices after a spec-prior pick. The pin is “soft” — it only sets the cold-start anchor; once enough calibrations accumulate (Phase::Stable), elect_anchor may move the anchor based on measured timing.

pub fn with_device_indices(self, device_indices: &[i32]) -> Self

Auto-detect the cold-start anchor from device hardware specs.

Queries each CUDA device’s compute capability and total VRAM, scores them as sm_major*100 + sm_minor*10 + vram_gb, and picks the rank with the lowest score (slowest by spec). Skips silently if any device-property query fails (no CUDA, invalid index) or if an initial anchor was already pinned (e.g. via with_speed_ratio or with_initial_anchor) — explicit user knowledge outranks the prior.

device_indices must be ordered by rank: device_indices[r] is the CUDA device index for DDP rank r.

pub fn batches(&self, rank: usize) -> usize

Batch count for the given device rank in the current cadence step.

pub fn batch_counts(&self) -> &[usize]

Per-device batch counts (for Ddp::weighted_all_reduce_gradients).

pub fn apply_callback_slack(&mut self, slack_ms: &[f64])

Stage per-rank callback wall-time (ms) to absorb on the next recompute_batch_counts call. The coord sets this just before the last sync cycle of an epoch that fires a user callback on a known rank, so the firing rank’s quota for that cycle drops by ceil(slack_ms / smoothed_ms_per_batch) batches — leaving compute slack to run the callback without bloating the barrier wait.

Inputs:

• slack_ms: length-world_size vector. Index r is the callback budget for rank r (zero = no slack, the typical case for non-firing ranks).

Silently no-ops when slack_ms.len() != self.world_size to match the rest of the ElChe builder/setter shape (callers constructed off-by-one inputs would otherwise crash a running training cluster, not what we want; recompute-without-slack is a safe fallback).

The slack is consumed exactly once per recompute_batch_counts call: after the per-rank targets are computed with the slack subtracted, the pending vector is zeroed. The caller can re-set the vector before each recompute, or leave it zeroed for cycles where no callback fires.

pub fn pending_callback_slack_ms(&self) -> &[f64]

Read the currently-staged callback slack (ms per rank). Returns all-zero by default. Test/diagnostic accessor; production code goes through Self::apply_callback_slack.

pub fn set_window_fill_ms(&mut self, fill_ms: &[f64])

Stage the per-rank per-window FILL (ms) for the window-pressure controller, consumed once by the next propose_anchor. The fill is the window’s first-batch excess over the steady-state (marginal) rate — the amortizable per-window fixed cost the marginal-anchor allocation feed excludes. The coordinator computes it from the per-window timing and sets it before Self::report_timing; left unset (all-zero), window-pressure falls back to the reduce-overhead term alone.

Silently no-ops on a length mismatch (matches the rest of the builder/setter shape: a caller off-by-one must not crash a running cluster; growing on the reduce term alone is a safe fallback).

pub fn pending_window_fill_ms(&self) -> &[f64]

Read the currently-staged per-window fill (ms per rank). Returns all-zero by default. Test/diagnostic accessor.

pub fn growth_enabled(&self) -> bool

Whether window-pressure growth is currently armed (the latch). Test/ diagnostic accessor; the latch is driven by the guard verdict through Self::commit_proposed_anchor / Self::veto_proposed_growth / Self::discard_proposed_anchor.

pub fn total_batches(&self) -> usize

Total batches across all devices for this cadence step.

pub fn anchor(&self) -> usize

Current anchor batch count (slow device batches per step).

pub fn anchor_wall_ms(&self) -> f64

Target wall time (ms) for one sync interval.

Returns anchor * slowest_ms_per_batch, the intended wall-clock duration between AllReduce events. Both GPUs should accumulate this much compute time before syncing. Returns 0 if not yet calibrated (no timing data).

pub fn nudge_anchor_down(&mut self, factor: f64)

Reduce the anchor by factor (e.g. 0.5 = halve).

One-directional correction for parameter divergence: tightens sync cadence when replicas drift apart. Does NOT loosen; ElChe’s overhead auto-tune handles upward adjustment.

Bypasses min_anchor (clamped to 1) because divergence is a stronger signal than the overhead floor. The overhead auto-tune will recover the anchor upward once divergence subsides.

pub fn relax_anchor_up(&mut self)

Relax the anchor upward by 1 batch on stable convergence.

Symmetric upward path to Self::nudge_anchor_down: lets async-mode anchor drift toward max_anchor over time as long as the convergence guard reports Stable, amortizing AllReduce barrier cost over more local SGD steps. Pairs with the downward NudgeDown path so the control loop has both directions.

Honors the user-defined max_batch_diff cap when set: refuses to relax if the projected per-rank batch_counts spread at anchor + 1 would exceed max_batch_diff. With ratio R between fastest and slowest rank and cap M, anchor is bounded by M / (R - 1) — e.g. for ratio 3 and max_batch_diff = 100, anchor caps at 50 (yielding [50, 150], diff exactly 100).

No-op when already at max_anchor, or when no calibrated ms_per_batch exists yet (Probe phase).

pub fn commit_proposed_anchor(&mut self)

Commit any pending window-pressure grow proposal from the last report_timing. Called on ConvergenceAction::Stable — the guard saw no divergence concern, so the grow applies.

Also advances the growth-enable latch: each Stable verdict counts toward re-arming growth (GROWTH_REARM_STABLE consecutive clean verdicts), so growth that was latched off by a prior SuppressGrowth / NudgeDown only resumes once convergence is robustly clean again.

Pairs with Self::veto_proposed_growth and Self::discard_proposed_anchor to make the convergence guard authoritative over overhead_target. No-op on the anchor when no proposal is pending (Probe/Warmup phase, overhead below target, or no report_timing call between guard verdicts).

pub fn veto_proposed_growth(&mut self)

Drop the pending grow proposal and latch growth OFF. Called on ConvergenceAction::SuppressGrowth — the guard saw divergence trending up, so growth would make it worse. Growth stays disabled until GROWTH_REARM_STABLE consecutive Stable verdicts re-arm it (the margin to the cliff: don’t poke the boundary again until convergence is robustly clean).

pub fn discard_proposed_anchor(&mut self)

Drop any pending grow proposal and latch growth OFF. Called on ConvergenceAction::NudgeDown — the nudge (Self::nudge_anchor_down) shrinks the anchor directly; growth is disabled and re-arms only after GROWTH_REARM_STABLE consecutive Stable verdicts.

pub fn apply_verdict(&mut self, verdict: AnchorVerdict)

Apply a source-agnostic AnchorVerdict — the ONE verdict seam.

Every verdict producer (convergence guard, meta-controller, future detectors) funnels through here; ElChe never learns the source. The fine-grained anchor methods (Self::commit_proposed_anchor, Self::veto_proposed_growth, Self::discard_proposed_anchor, Self::nudge_anchor_down, Self::relax_anchor_up) remain available as building blocks, but orchestration code should speak verdicts.

NudgeDown both discards the pending grow proposal AND nudges: a proposal staged this cycle was computed from the pre-nudge anchor, so a later Stable verdict committing it would silently overwrite the nudge (the discard-then-nudge order is canonical; the two operations touch disjoint state, so producers that historically nudged first are unaffected).

pub fn is_calibrated(&self) -> bool

Whether at least one timing measurement has been reported.

pub fn has_speed_hint(&self) -> bool

Whether a speed hint was applied (batch_counts are non-uniform).

Used by the coordinator to decide if epoch 0 should use throughput-proportional partitions before calibration.

pub fn ms_per_batch(&self) -> Vec<f64>

Per-device smoothed milliseconds per batch (mean over trust window). Returns a fresh Vec rather than a slice because the smoothed values are computed from internal ring buffers; callers store or iterate the vec directly.

pub fn smoothed_ms_per_batch(&self, rank: usize) -> Option<f64>

Per-rank smoothed ms-per-batch as an Option<f64> — None when the trust window for that rank is empty (no positive reading has landed yet), Some(ms) when a calibrated value is available. Distinguishes “uncalibrated” from a legitimate 0.0 better than Self::ms_per_batch, which collapses both into 0.0. Used by ClusterCoordinator::resolve_fastest_role to pick the live rank with the lowest calibrated ms-per-batch (Fastest policy).

pub fn report_window(&mut self, report: &WindowReport)

Ingest one reduce window’s observations — the event-shaped feed.

Stages the window-pressure fill, selects the timing scale via WindowReport::select_feed (the mixed-scale inversion guard), and feeds Self::report_timing when the window carries any signal (an all-zero feed — e.g. a fully-idle window — reports nothing, so no spurious zero-ms sample poisons the trust windows). This is the coordinator’s one timing entry point; the lower-level Self::set_window_fill_ms + Self::report_timing pair remains for callers that assemble their own feed.

pub fn report_timing( &mut self, wall_ms: &[f64], actual_batches: &[usize], sync_ms: f64, )

Report timing after a cadence step completes.

wall_ms[rank]: wall-clock time for all batches on that device (ms). actual_batches[rank]: number of batches each rank actually processed since the last sync (i.e., steps_since_avg). In Cadence mode the fast GPU may process more batches than its intended batch_counts while waiting for the slow GPU to reach the trigger threshold. Using the intended count as divisor would inflate the fast GPU’s ms_per_batch, inverting the throughput ratio. sync_ms: AllReduce overhead for this step (ms).

Updates batch ratios based on measured throughput. If AllReduce overhead exceeds the target, anchor auto-tunes upward.

pub fn recent_batch_share(&self) -> Vec<f64>

Smoothed per-rank batch share as an observation of recent cadence.

Averages the last BATCH_COUNTS_WINDOW_CAP batch_counts snapshots (each captured at the end of report_timing) and normalizes to sum to 1.0. This is the metric source for per-epoch share reporting: it answers “what fraction of work did the balancer assign each rank, recently?” rather than “what fraction of samples did each rank happen to consume?” — those agree in steady state but diverge under progressive dispatch’s tail-balance equalization near epoch end.

Falls back to the current batch_counts ratio when no snapshots have been captured yet (no report_timing calls), and to equal shares if both are degenerate.

pub fn clamp_total(&self, max_total: usize) -> Vec<usize>

Clamp batch counts to a maximum total, preserving proportions.

Returns a new batch-count vector. Use near epoch boundaries to avoid consuming more batches than remain.