async_cifar10: real-sim parity analysis, bug fixes, and parity checker - #63
Merged
Conversation
…e + harness Adds the machinery to compare Felix vs OORT vs REFL under streaming data and show that OORT/REFL mis-select clients (stale statistical utility) while Felix tracks true utility via its eval-selector. Telemetry / ground truth: - Periodic global-model checkpoints (save_round_checkpoint on the shared aggregator base; wired into sync + async loops; gated by checkpoint.*). - New trainer_round fields: delta_weight_l2 (||update||), grad_norm_epoch1, task_to_perform -> relate update magnitude to amount of unlocked data. - scripts/analysis/oracle_misselection.py: offline oracle reconstructing each trainer's TRUE current utility (deterministic split+seed+visible prefix) and joining with logged believed utility to emit per-round mis-selection rate, utility regret, believed-minus-true gap, and selected-set true utility. Self-calibrates the streaming horizon from util_disparity telemetry. - analyze_run.py: comm-cost (model-equivalents, charging eval probes), new single-run plots (update/loss/utility vs visible fraction, mis-selection, comm-vs-accuracy), and compare_streaming() cross-baseline bundle incl. time-to-target. Experiment configs (n300, alpha=0.1, syn_0, agg_goal=10, 8 GPUs, horizon 10800s): canonical 6-arm YAML + per-node split (streaming/control) + 300-round pilot variants + n10 smoke. Pre-existing OORT selector fixes (blocked all sync OORT/REFL runs): - save_exploited_utility_history: len()==0 instead of truthiness on np array. - sample_by_util / sample_by_speed: cast np.str_ ids to str (dict-key safe). - increment_selected_count / stats loop: skip in-flight ids absent from ends. Sync aggregator evaluate() now honors evalEveryNRounds (was every round; the full test pass dominates per-round cost at n300 and made sync runs intractable). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Switch all 6-arm / node-split / pilot configs from simulated to real time_mode while simulated-mode issues are being resolved. In real mode the streaming clock is wall-clock (full_data_available_after_s is real seconds) and trainers sleep their modeled delays; the horizon may need re-calibration against the pilot's real per-round wall-clock, and the mis-selection oracle's visible-count reconstruction becomes approximate (documented in the YAML header). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…eddance FedDance is now a runnable async_cifar10 baseline (felix/oort/refl/feddance comparison). Resolves examples/async_cifar10/FEDDANCE_TODO.md. - baselines.yaml feddance: add aggr_num (default 10), switch tracking from ORACULAR to disabled (FedDance uses its own check-in predictor). Stack stays main_fedavg_agg.py (base syncfl), which drives on_update_received / on_round_completed (the Oort stack overrides _aggregate_weights and does not). - trainer/main.py: accumulate per-round local training accuracy (reset_local_accuracy + update_local_accuracy) so FedDance's A_m signal is real (harmless for other selectors). - main_fedavg_agg.py evaluate(): honor evalEveryNRounds (was every round). - syncfl/top_aggregator base fixes that unblock fedavg + feddance on the base stack: default FwdLLM-only data_id/iteration_per_data_id via getattr; guard PROP_ROUND_DURATION.total_seconds() against None; iterate diskcache Cache with list(self.cache) instead of .keys(). - Add 4-way quick pilot config (felix/oort/refl/feddance, 32 trainers, real mode, streaming) for ~30-min end-to-end validation. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
REFLOortSelector extends OortSelector, whose __init__ requires kwargs["aggr_num"]; the refl baseline and refl experiment arms never set it, so every refl run raised KeyError at selector init (caught by the 4-way pilot). Add aggr_num=10 default to the refl baseline and explicit aggr_num per refl arm (= agg_goal: 10 full / 5 pilot). Regenerated node-split files. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
REFLOortSelector.select() never called emit_selection, so EVENT_SELECTION was absent for REFL runs and the oracle/comparison had no REFL believed-utility / selected-set data. Add the same emit_selection call as oort/feddance. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- Canonical experiment now has 8 arms = {felix,oort,refl,feddance} x
{streaming,control}; renamed felix_oort_refl_streaming_alpha0.1.yaml ->
felix_oort_refl_feddance_alpha0.1.yaml.
- Replace the old node-split (node=streaming-only / node=control-only) with a
2-phase scheme: STREAMING_node1/2 then CONTROL_node1/2, each node running 2
arms (node1=felix+oort, node2=refl+feddance) so both nodes do streaming first,
then both do control.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- SELECTOR_STALENESS_AUDIT.md: documents each selector's input factors (OORT, REFL, FedDance, Felix) with believed-value source + staleness, true-value computation, and which are real staleness dimensions vs exact bookkeeping; lays out the self-relative counterfactual-replay framework. - analyze_run.py: im_staleness_by_round + compare overlays (compare_Im_rankcorr, compare_Im_gap) — per-baseline rank-correlation and normalized gap between BELIEVED and TRUE statistical utility over time, computed from existing oracle_utility.csv (no re-run needed). First staleness dimension; speed + A_m + the no-drift counterfactual replay are the next steps. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…idate So the offline audit can compare believed-vs-true per factor (not just I_m): - FedDance: log V_m, I_m, A_m, U_m and last_engaged_round per candidate in EVENT_SELECTION (it already computes PROP_V/I/A/U on ends). - Felix (async_oort): add last_train_round (PROP_LAST_SELECTED_ROUND) alongside last_eval_round so the audit can show whether believed I_m was refreshed by eval vs train -- the freshness mechanism under test. These are additive logging fields; no selection-behavior change. Must land before the full runs so their telemetry captures the factor bundle. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…ore logging - New flame/selector/scoring.py: pure, side-effect-free scoring formulas (oort (util+temporal)*system_util; feddance V*I*A*MAB; topk) shared by the live selectors and the offline counterfactual replay so they never drift. - OORT + AsyncOort(Felix): route the score combination through scoring.* and stash per-candidate components (believed_I, temporal, system_util), logged in EVENT_SELECTION.per_trainer for the audit. (Removed OORT's dead 95th-pct clip — it was overwritten before use; behavior unchanged.) - REFL inherits OORT's scorer; emits the stashed components too. All 127 selector tests pass (behavior-preserving). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- oort_utility_acc: also returns local top-1 accuracy from the forward pass; per-checkpoint loop stores it -> true A_m = accuracy slope across checkpoints (FedDance's accuracy-increment, true value). - Counterfactual replay per selection: re-rank candidates by each selector's own score with believed factors vs true factors substituted (I_m for all; A_m too for FedDance; speed ~identity here), using the logged score components + shared scoring fns. Emits self-relative cf_misselection_rate + cf_utility_regret (oracle_counterfactual.csv) and run-level means in oracle_summary.json. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…cf reporting - oracle: counterfactual regret now measured in each selector's OWN true-score units (true_topk maximizes true_score => regret >= 0), fixing negative values for multi-factor selectors (FedDance V*I*A). mis-selection rate unchanged. - analyze_run compare_streaming: add compare_cf_misselection / compare_cf_regret overlays (self-relative staleness penalty per baseline) from oracle_counterfactual.csv. Validated end-to-end on a 4-way n10 audit smoke: counterfactual populates for felix/oort/refl/feddance via the shared no-drift scorers + logged components + true I_m/A_m. (Smoke numbers are noise at n10/near-random; real signal needs the full runs.) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…selection/insights/system) plot_helpers: paper-style rcParams (fonttype 42), fixed per-selector colors, PDF output, config_stamp() from snapshot, and new figure types (scatter-with-diagonal, banded line, histogram, signed bar, heatmap, dual-axis); quiet font/mpl logging. analyze_run: reorganized into 5 per-run plot classes emitting PDFs: - performance: accuracy (vs round & sim-time), Δacc/eval, loss, accuracy-vs-data- unlocked, global weight-change norm (from checkpoints). - sanity: trainer runtime expected-vs-actual (+residual hist, early/late counts, overrun rate), utility believed-vs-true (+discrepancy over rounds), data-unlock band, selection/aggregation count consistency, stale-rejection (when logged). - selection: selected-vs-pool true utility, coverage+Gini, frequency, exploration factor, eval-vs-train, participation heatmap, availability composition. - insights: I_m staleness, mis-selection, counterfactual mis-selection/regret, data-unlock effects, streamed-vs-full utility. - system: comm-vs-accuracy, per-round train/eval comm, time breakdown, queue depth. telemetry: OORT stack now emits agg_round (it overrode _aggregate_weights and emitted none) so oort/refl get staleness/contributing/queue visibility; trainer logs current lr. Validated end-to-end on a felix n10 smoke (32 PDFs). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…verrides) Snapshots don't serialize trainer.config_overrides, so the stamp showed stream:off even when streaming was on. Fall back to inferring it from telemetry (a trainer_round with visible_samples < total_samples) so the config stamp is accurate on the real runs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…on & runtime sanity Per review feedback: - (i) trainer labels use last-3 digits (per-trainer bars / heatmap / breakdown); these plots already list only trainers (no aggregator). - (ii) histograms annotate each bar with its % of total. - (iii) smaller/crisper title & axis-label sizes (rcParams) so they fit the box. - (iv) config stamp moved to the TOP of each figure; also inferred from telemetry (visible<total) since snapshots omit config_overrides (was showing stream:off). - (iv-b) new selection plot: per-round avg speed & avg believed utility of the clients actually picked. - (v) aggregator-observed vs trainer-reported response time: new agg_observed_s field on agg_round (send->recv wall per contributing trainer) emitted by oort/asyncfl/syncfl stacks; new sanity plots runtime_agg_vs_trainer (scatter w/ diagonal) + runtime_overhead_hist (observed - reported) to catch aggregator processing/network overhead (the "trainer says 10s, aggregator sees 14s" case). Fix: OORT agg_round now emitted BEFORE optimizer.do (which consumes the cache) and reads cached TrainResult fields, so contributing/staleness/agg_observed populate. Analyzer joins per trainer by nearest round (robust to staleness). Validated on felix+oort n10 smokes (all plots render; OORT overhead plots populate). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
New optional aggregator hyperparameter max_runtime_s: when set, the aggregator stops the run once that many wall-seconds have elapsed since it started (checked in increment_round, so it applies to both sync and async stacks). Set to 12600 (3.5h) on all 8 arms + node files: streaming data fully unlocks at 3h, so this caps each arm shortly after without burning compute to the 1000-round budget. Validated: a run with max_runtime_s=60 stopped at round 3 (not 1000). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Channel._streamer_for_recv_fifo tracks one "active task" per end while
awaiting that end's message. The streamer is fire-and-forget and outlives
the recv_fifo caller, and the per-end End.get() had no timeout while the
active-set entry was only removed on successful delivery. So an end whose
trainer never sends (slow / dropped / unavailable) blocked forever and
stayed permanently in _active_recv_fifo_tasks. That end was then skipped
("already has active task") on every future receive, so its updates were
never consumed -> the aggregator stalled with a monotonically growing
active_tasks count (observed: 1 -> ~30, then no progress).
Fix:
- Thread the caller's timeout into _streamer_for_recv_fifo and wrap each
per-end get() in asyncio.wait_for, so a quiet end releases instead of
blocking forever. timeout=None preserves legacy blocking for sync callers.
- Move active-set cleanup into a finally (discard) so an end is always
released on delivery, timeout, error, or cancellation.
- Don't enqueue non-messages (timed-out ends) so they can't consume a
first_k slot ahead of a real update.
Simulator mode (_sim_recv_min, which probes recv_fifo per-end with a short
timeout) benefits from the same fix and is verified safe: CPython's
wait_for + Queue.get is cancellation-safe, so a late message is buffered on
the end and delivered on the next probe rather than lost.
Tests:
- tests/test_channel_recv_fifo.py: direct streamer-cleanup tests plus
full recv_fifo tests over a real loop mirroring the simulator probe
pattern (incl. late-message-not-lost). All pass on the fix; 8/9 fail on
the pre-fix code.
- tests/mode/test_async_sim_ordering.py: initialize _sim_pending_commit in
the fake aggregator (pre-existing test bug; suite was red).
Also: debug_stuck_run.py gains a streaming active-task leak analyzer and a
UTF-8 stdout fix so it no longer crashes when piped.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
ExampleConfig.aggregator_main now defaults to None (baseline-owned, with a runner-side fallback to "aggregator/pytorch/main.py"), and aggregator_main is resolved via ExperimentRunner._resolve_aggregator_main rather than being part of _resolve_example_paths. The tests still asserted the old defaults/shape. - test_experiment_config: aggregator_main default is now None. - test_runner_paths: assert aggregator_main is not in _resolve_example_paths, and exercise _resolve_aggregator_main (default, example override, baseline ownership, and the baseline+override conflict that must raise). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
felix_oort_refl_feddance_alpha0.1_STREAMING_node1.yaml runs felix then oort sequentially. Split into two standalone single-baseline YAMLs (execution config inlined, no shared anchor) so felix and oort can run on separate nodes in parallel instead of back-to-back on one node. Settings copied verbatim; num_trainers unchanged at 300. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
At high trainer-per-host concurrency (e.g. 300 trainers / 8 GPUs), the async runs showed large excess in the aggregator's send->recv lag beyond each trainer's modeled budget D. Decomposing the lag (new analyzer below) showed it is entirely trainer-side and NOT actual compute, the aggregator, or the network: ~2.3 s/sample of "GPU compute" on 1-sample rounds. The cost was defensive memory-hygiene + profiling that backfires under co-location. Trainer (examples/async_cifar10/trainer/pytorch/main.py): - MemoryProfiler is now gated off by default (memory_profiling_enabled hyperparameter). Its per-round heap walks (gc.collect + 3x gc.get_objects() with a per-object torch.is_tensor() check, x4 calls/round, two inside the GPU-timing window) dominated per-round wall time. - Removed per-round / per-epoch / per-50-batch torch.cuda.empty_cache() and the redundant gc.collect()s: under co-location empty_cache forces a CUDA sync and frees the allocator's blocks, so the next round re-allocates from the driver (serialized across processes) -- it hurts rather than helps. - grad-norm (epoch 1) and delta_weight_l2 are now telemetry-gated and use one fused on-device reduction + a single .item(), instead of a .item() per parameter (which forced ~12 GPU->CPU syncs each). - Added first-class per-phase timing (pre_train_s / post_train_s) to [TRAIN_CYCLE] logs and the trainer_round telemetry. Library (flame/mode/horizontal/syncfl/trainer.py): - update_local_accuracy() accumulated correct-count via .item() EVERY batch (a GPU sync per batch). Accumulate on-device; sync once in finalize_local_accuracy(). Analysis (examples/async_cifar10/scripts/analyze_timing_overrun.py): - New diagnostic: histogram of late responses by lateness bucket + CDF of deviation extent, each split into trainer-attributable vs aggregator-observed (plus transport), so future runs show where the excess originates. Also surfaces TIMING_OVERRUN_AGG / SEND_RECV_LAG_HIGH counts. Validated A/B (48 trainers / 2 GPUs, felix/async_oort): per-round trainer compute dropped 0.51s->0.03s (real) and 0.46s->0.01s (sim); aggregator-observed lag 0.99s->0.09s (real). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
… tests
Make simulated mode verifiably decision-equivalent to real mode (not just
faster). Adds the seeding prerequisite, fixes a parity-breaking RNG bug,
extracts a shared parity-check module, and lands tests at every level.
Code:
- Aggregator (syncfl/top_aggregator.py): optional deterministic seeding of the
process-global np.random/random (which the selector draws from) + torch, from
a `seed` hyperparameter, at internal_init. seed=None preserves legacy behavior.
Without this, selection can never match across runs/modes even with identical
state.
- FedBuff (selector/fedbuff.py): REMOVE the per-call `random.seed(time.time())`
in _handle_send_state. It reseeded the global RNG with wall-clock time on every
selection, making selection non-reproducible by construction, clobbering global
random state for all other consumers, and defeating any deterministic seed.
This was a primary source of real/sim selection divergence.
Shared checks:
- New scripts/parity_checks.py: canonical, dependency-free loaders/helpers +
parity/invariant functions (selection Jaccard, aggregation-sequence,
staleness, participation, sim_send_ts, GPU-budget, virtual-clock monotonicity,
agg_goal cycles). compare_parity.py now imports the shared loaders/helpers
(single source of truth for CLI + tests).
Tests:
- tests/sim/test_virtual_clock.py: buffer overwrite/clear, equal-ts advance.
- tests/mode/test_async_sim_ordering.py: real-path vs sim-path commit + staleness
equivalence under sorted arrival; sim recovers completion order and yields a
canonical staleness sequence under any arrival permutation.
- tests/selector/test_selection_determinism.py: FedBuff + Oort selection is a
pure function of (state, seed) — same seed reproduces, distinct seeds vary.
- tests/mode/test_parity_checks.py: unit tests for the parity functions on
synthetic telemetry (default suite, no MQTT/GPU).
- tests/mode/test_real_sim_e2e_parity.py: opt-in (env-gated) e2e parity assertion
over two run dirs, reusing parity_checks.
Configs: seeded parity pair (felix_n48_parity_seeded_{real,sim}) for the
faithful regime + a stress pair (felix_n48_stress_{real,sim}, 48/1-GPU, higher
c/agg_goal) to map where sim ordering degrades; n48 dataset split.
Determinism analysis + phased plan documented as Task 3 in
TELEMETRY_AND_SPEEDUP_PLAN.md.
252 passed, 7 e2e skipped.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Sync aggregator previously ignored time_mode in aggregation: it committed the
first_k updates by PHYSICAL ARRIVAL even in simulated mode, so the wrong k were
aggregated (arrival != sim-completion order without sleeps) and round duration
was wrong. This broke parity for every sync baseline (fedavg/oort/refl/feddance).
- syncfl/top_aggregator.py: new _sync_sim_recv_first_k. In simulated mode buffer
the selected ends' updates and commit the first_k with the SMALLEST
sim_completion_ts (the k that would finish first in real), advance the virtual
clock to the k-th smallest, and source PROP_ROUND_DURATION from
SIM_ROUND_DURATION (so OORT/REFL see the simulated speed). Real mode keeps the
arrival-ordered first_k path. Aggregation is order-independent (weighted
average), so parity needs only the committed set + round duration.
- tests/mode/test_sync_sim_ordering.py: commit-set == k-smallest-sct independent
of arrival permutation; vclock advances to k-th smallest; round-duration from
SIM_ROUND_DURATION; graceful when fewer than k respond.
- fedavg_n48_parity_seeded_{real,sim}.yaml: seeded sync parity pair for live
validation.
- Doc: cross-baseline coverage + a characterized OPEN issue — async (felix) sim
over-selects (in-flight grows toward N under no-sleep; staleness ~8.6 vs ~1.2
real). A targeted slot-accounting fix did not resolve it (buffer fills faster
than it drains); reverted rather than shipped unverified. Needs a focused pass
to bound the sim in-flight buffer to c. Sync path is unaffected.
257 passed, 7 e2e skipped.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…y tool random selector fixes (unblocks fedavg on the sync stack): - The active select() crashed with KeyError 'state' under sync FL: it read channel_props[KEY_CH_STATE] unconditionally, but the SEND/RECV channel state is the *buffered* concurrency pattern (FwdLLM / async stack) — stateless sync FL never sets it. Default the state to SEND when absent, with a comment on why it exists. fedavg/oort/refl/feddance(sync) + FwdLLM(buffered) both work now. - Removed a fully-duplicated dead select() method (silently overridden by the second definition) — the source of the confusion and ~95 lines of bloat. parity_checks: commit_sequence() + first_divergence() — a mode-agnostic LOGICAL ordering of committed updates (by round, agg_goal_count; no wall-clock) so real vs sim can be diffed for control/ordering bugs independent of timing. On the felix runs it localizes the first divergence to commit #1 (selection-level RNG desync), before staleness blows up. 260 passed, 7 e2e skipped. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…te time-split, acc-gain dual-axis Addresses several readability/clarity items in the run analyzer (works for all async_cifar10 baselines; reads existing telemetry): - participation heatmap (#12): discrete semantic colours instead of viridis — grey=not selected, light-blue=eval, light-green=train, light-red=unavailable, orange=selected-but-unavailable (availability forward-filled from avail_change), with a legend. New `heatmap(discrete=...)` path in plot_helpers. - response-lateness CDF (#10): the existing residual compares GPU time to budget ("early" just means GPU finished early — the trainer still SLEEPS to fill the budget). New CDF of lateness = max(gpu,D) - budget, which is 0 (on-time) or >0 (overrun), never negative — clarifies that trainers never respond early. - aggregate round-time split (#19/#5): stacked-area of mean pre/gpu/post/sleep per round across trainers — the system-level view of where round time goes (uses pre_train_s/post_train_s; visually confirms the per-round overhead fix). - accuracy-gain per eval (#4): now overlays overall accuracy on a secondary axis (new `signed_bar_line` helper) so delta and absolute trajectory show together. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
In simulated mode the sync aggregator loop spins without sleeps, so channel.ends() can transiently return None before trainers join + get selected — crashing on len(None) / enumerate(None). Real mode's pacing masked it. Guard all call sites in _aggregate_weights and _distribute_weights (skip+retry on None/empty; len(... or [])). Fixes the immediate sim crash for sync baselines (fedavg validated: distribute now works, no NoneType). NOTE: a separate open issue remains — the sync sim aggregation does not yet commit (SYNC_SIM_RECV=0); see TELEMETRY_AND_SPEEDUP_PLAN / handoff. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…xt steps) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…lots Aggregator parity fixes (sim mode now tracks real per baseline): - syncfl: distribute->SEND / aggregate->RECV channel state so the buffered (random) selector returns the in-flight set instead of stalling (fedavg/feddance). - asyncfl: bound sim in-flight to c (keep pending in selected_ends, reset buffered RECVD->NONE) fixing felix over-selection/staleness; guard async_oort-only trainer_eval_recv_ends so fedbuff's eval path doesn't crash. - oort stack: add sim-ordering (sim_send_ts stamp + sct-ordered commit + vclock) and cross-round straggler carry (persistent buffer) for oort/refl; oracular availability now uses the virtual clock in sim (was wall-clock). - Join barrier (min_trainers_to_start) so sim/real select from the same pool; configurable timeout for n300. - Baseline staleness semantics (oort reject / refl accept <=5) preserved in sim. Parity tooling: selection-parity check gated for stochastic selectors (DETERMINISTIC_SELECTORS) with participation-frequency parity enforced instead. All 6 baselines pass the e2e parity suite; seeded real/sim config pairs added. n300 overnight: OVERNIGHT/SIMULATED node configs (sim+real per baseline, both with the barrier); non-interactive batch runner + _sweep_stragglers cleanup between runs; overnight_run.sh (smoke+run) and compare_overnight.sh. Plots: comm in MB split by task+direction (train down/up, eval down) + message accounting; Lorenz/Gini fairness; late-fraction hist; P99; overhead CDF; compute time by task; speed/utility CDFs (expected vs actual); whole-run time-split bar. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
… script location
Auto-detect conda base via `conda info --base` / CONDA_EXE with fallbacks
(incl. /coc/scratch/$USER), instead of hardcoding ~/miniconda3; fail loudly if
not found. Derive the example/repo dirs from ${BASH_SOURCE} so the scripts work
regardless of where the repo or conda live on each node. FLAME_CONDA_ENV overrides
the env name (default dg_flame).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add WALL_SEND_TS (MessageType 38) stamped by the trainer at channel.send() so aggregators can split wall_lag_s into training_elapsed (GPU+setup+sleep) vs mqtt_lag (broker delivery). Both async and sync aggregators now emit [MQTT_DELIVERY_LAG] lines with the full decomposition for real-mode runs. New analysis plots: sanity/pre_train_s_cdf.pdf — GPU queue wait (large = contention) sanity/gpu_compute_cdf.pdf — on-device GPU time for train rounds system/mqtt_delivery_lag_cdf.pdf — pure broker delivery time CDF system/wall_lag_decomposition_cdf.pdf — wall/train_elapsed/mqtt together New debug_run.sh: configurable script to run just felix (node1) or refl (node2) at a specified wall-clock budget (default 1h, max_runtime_s applies as virtual-clock seconds in sim mode via the vclock fix). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
/tmp can be small on shared nodes; stdout/stderr from 300-trainer runs
can reach hundreds of MB. Both overnight_run.sh and debug_run.sh now
write to <example_dir>/run_logs/{overnight,debug}/ which lives on the
same filesystem as experiments/. The run_logs/ tree is gitignored.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
run_experiment stores all trainer/agg logs in experiments/run_*/ already; LOGDIR only holds the launcher stdout and generated YAML configs (both small), so /tmp is fine. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…safe - Replace [MQTT_DELIVERY_LAG] with [LAG_DECOMP] across asyncfl, syncfl, oort aggregators: 6 decomposition components (agg→trainer delivery, compute, post-compute wait, MQTT transit, queue/buffer wait, processing) - Add WALL_RECV_TS (39) and ROUND_COMPUTE_S (40) MessageTypes; trainer stamps both unconditionally in _send_weights (real and sim) - Remove [SEND_RECV_LAG_HIGH] fixed-threshold alerts; replace with [TIMING_OVERRUN_AGG] using budget_s + _NETWORK_SLACK_S (2s) threshold - Fix [TIMING_OVERRUN_AGG] sim check: was _virt_elapsed <= 0 (wrong), now _virt_elapsed > budget_s - Add _oort_sent_version_ts per-version send timestamps in oort for correct lag measurement on REFL stale updates (staleness ≤ 5) - Add wall-clock failsafe to increment_round(): sim runs hard-stop at max_runtime_s wall seconds even if vclock stalls (fixes REFL sim hang) - Update analyze_run.py: _parse_lag_decomp replaces _parse_mqtt_delivery_lags - debug_run.sh: default RUNTIME_S 3600 → 1800 (30-min runs) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…k fix Task 1 — per-phase trainer timing (CPU vs GPU split): - syncfl/trainer.py: add _phase_times dict + _phase() context manager; instrument mqtt_fetch_s, weights_to_ram_s, weights_to_gpu_s (fetch side) and weights_from_gpu_s, post_cpu_s, mqtt_send_s (send side); CUDA sync on GPU-bound phases for accurate timing. - trainer/pytorch/main.py: extend trainer_round telemetry extra with gpu_compute_s, sleep_s, and all _phase_times from the base trainer. - scripts/analyze_trainer_phases.py: new analyzer — reads trainer_round JSONL, averages CPU/GPU phases per round, emits stacked-bar plots (A/B) and per-trainer heatmap (C); --compare overlay for oort vs refl. Task 2 — automatic CPU core pinning per trainer: - spawner.py: TrainerSpawner discovers usable cores via sched_getaffinity(0); round-robin assigns one core per trainer; applies via preexec_fn sched_setaffinity; sets OMP/MKL/OPENBLAS/NUMEXPR_NUM_THREADS=1 to prevent thread oversubscription; --cpu_pinning on|off toggle (default on); logs trainer→(gpu,core) table after spawn_all. - trainer/pytorch/main.py: initialize() calls torch.set_num_threads(1) when OMP_NUM_THREADS=1 is inherited from spawner. Task 3 — wall-clock vs virtual-clock termination correctness: - syncfl/top_aggregator.py: increment_round reads new max_wall_runtime_s config; sim failsafe uses it (default 4×max_runtime_s) instead of reusing the virtual budget, so a sim run is not cut before vclock≈T; periodic [VCLOCK_PROGRESS] log every 30s shows vclock/wall/speedup. - scripts/debug_run.sh: --wall-runtime-s flag; make_debug_yaml writes max_wall_runtime_s into YAML (default 4×runtime_s). - scripts/compare_clock_parity.py: new script — per-round virtual-time deviation, sim speedup, failsafe-triggered check, and 2-panel plot. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…s analysis Framework changes: - syncfl/top_aggregator: rename 'speedup' -> 'sim_rate' in [VCLOCK_PROGRESS]; replace loose 4x failsafe with sim_wall_ceiling_s (default = max_runtime_s, 1x) - trainer/main.py: add [PLACEMENT] log (gpu, cpu_cores) at init - debug_run.sh: --sim-wall-ceiling-s flag; make_debug_yaml writes sim_wall_ceiling_s=runtime_s New analysis scripts: - plotters/_annot.py: shared annotate_percentiles(P50/P90/P99) + flush_percentile_table - analyze_pinning.py: CPU-core and GPU assignment balance from [PLACEMENT] logs - sanity_check_real_sim.py: T1-T7 battery (vclock monotone, accounting, sim_rate, wall_speedup, failsafe, selection balance, phase consistency); validated on 0606 runs - analyze_dynamics_timeline.py: AVL_TRAIN/UN_AVL/in_flight/idle per round + idle CDF Updated scripts: - analyze_trainer_phases, analyze_timing_overrun, analyze_send_recv_lag, compare_clock_parity: all now use shared annotate_percentiles; compare_clock_parity rewritten to clearly distinguish sim_rate (vclock/wall) vs wall_speedup (real/sim) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Implements the full §3 battery from real-sim_parity_checker_plan.md: §3.H Clock & throughput (the new enforced core): - K10 vclock_telemetry_present: FAIL-LOUD when sync sim omits vclock_now - K2 throughput_parity: rounds/virtual-s parity — catches 410 vs 673 rounds - K3 per_round_advance_parity: KS + mean on per-round Δvclock vs Δwall - K4 overlap_factor: localizes sim's missing inter-round overlap model - K8 terminal_state_parity: rounds + trainers at matched virtual budget V - U2 total_commits_parity: commit count at matched V - P3 trainer_speed_parity: control check isolating divergence to clock advance Also adds utility, convergence (C3 self-compare bug fixed), trainer_speed, inter_arrival_order, budget_not_cap, failsafe_ok checks. Structure: scripts/parity/checks.py — pure functions, stdlib only scripts/parity/report.py — section-grouped stdout + JSON + PNG scripts/parity/cli.py — argparse; single-pair and --batch mode scripts/parity_checks.py — re-export shim (backward compat for pytest) scripts/parity_check.py — stable CLI entry point All 34 unit tests pass; existing e2e tests unchanged. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…omments
Deletions:
- *_TODO.md, *_PLAN.md, *_AUDIT.md, *_HANDOFF.md planning docs (captured in git history)
- aggregator/agg_logs.{csv,txt} and *.pdf plots (generated artifacts)
- ava/david/seshu_first_task.md onboarding docs
- scripts/compare_parity.py, sanity_check_real_sim.py, compare_clock_parity.py
(superseded by scripts/parity/ package + parity_check.py entry point)
Updates:
- scripts/compare_overnight.sh: call parity_check.py --real/--sim DIR instead
of the deleted compare_parity.py with file-glob args; add --json-out
- aggregator/pytorch/main_{asyncfl,oort_sync}_agg.py: remove commented-out
config entries, stub-method comments, and narrating inline comments
- trainer/pytorch/main.py: remove per-phase taxonomy block comment headers
(field names self-descriptive); collapse initiate_heartbeat multi-comment
block to one line preserving the non-obvious dup_check_and_sleep reason
Added:
- experiments/parity_diagnosis_20260607.md: June 7 run parity findings
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
There was a problem hiding this comment.
Pull request overview
This PR tightens real-vs-sim parity for async (Felix) and sync (REFL/Oort/FedAvg/FedDance) CIFAR-10 runs by fixing simulation/streaming edge cases (recv_fifo task leakage, sim ordering/clock behavior, selection determinism), expanding telemetry needed for parity diagnostics, and adding a consolidated parity-check toolchain plus run scripts/configs to reproduce and validate parity end-to-end.
Changes:
- Fixes/guards for simulated-mode receive ordering, streaming recv_fifo task cleanup, and selection determinism (incl. RNG seeding and selector instrumentation for audit/replay).
- Adds a parity checker suite + stable CLI entry point, plus extensive parity/stress/streaming experiment YAMLs and operational scripts for overnight/debug runs.
- Extends telemetry schema and trainer/aggregator timing instrumentation to support deeper lag/overrun decomposition and offline audits.
Reviewed changes
Copilot reviewed 87 out of 94 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| lib/python/tests/test_channel_recv_fifo.py | New regression tests to ensure recv_fifo streamer tasks don’t leak and late messages aren’t lost. |
| lib/python/tests/sim/test_virtual_clock.py | Adds tests for VirtualClock noop-on-equal advance and SimReorderBuffer overwrite/clear behavior. |
| lib/python/tests/selector/test_selection_determinism.py | New tests asserting selector determinism under global RNG seeding (FedBuff/Oort). |
| lib/python/tests/mode/test_sync_sim_ordering.py | New tests validating sync simulated-mode commits smallest SIM_COMPLETION_TS first_k ordering. |
| lib/python/tests/mode/test_real_sim_e2e_parity.py | Opt-in end-to-end parity test harness comparing completed real vs sim run directories. |
| lib/python/tests/mode/test_async_sim_ordering.py | Extends async sim ordering tests; documents equivalence/divergence between real FIFO vs sim completion-order paths. |
| lib/python/tests/launch/test_runner_paths.py | Updates expectations for aggregator_main resolution and adds baseline-ownership conflict tests. |
| lib/python/tests/launch/test_experiment_config.py | Updates defaults: aggregator_main now baseline-owned and resolves later (defaults to None in config). |
| lib/python/flame/telemetry/events.py | Adds agg_observed_s and delta_weight_l2 fields + docstrings to telemetry event builders. |
| lib/python/flame/selector/scoring.py | New pure scoring helper module for selector score formulas shared by live selectors and offline audits. |
| lib/python/flame/selector/refl_oort.py | Emits selection-decision telemetry for REFL to support offline mis-selection/oracle analysis. |
| lib/python/flame/selector/random.py | Defaults channel state to SEND when absent; clarifies SEND/RECV semantics for buffered concurrency pattern. |
| lib/python/flame/selector/oort.py | Uses shared scoring helpers; fixes type/empty-check edge cases; emits per-trainer audit components. |
| lib/python/flame/selector/feddance.py | Emits per-candidate believed scoring factors to support offline staleness/mis-selection audits. |
| lib/python/flame/selector/fedbuff.py | Removes per-call random reseeding to preserve deterministic, aggregator-seeded selection behavior. |
| lib/python/flame/selector/async_oort.py | Adds per-trainer audit extras and uses shared score combiner for consistent offline replay. |
| lib/python/flame/mode/message.py | Adds new message types for wall send/recv timestamps and modeled compute duration. |
| lib/python/flame/mode/horizontal/syncfl/trainer.py | Adds phase timing instrumentation; stamps wall send/recv timestamps and compute markers; reduces per-batch GPU sync in accuracy accumulation. |
| lib/python/flame/mode/horizontal/asyncfl/top_aggregator.py | Fixes sim buffered-slot handling/over-selection, improves lag measurement using arrival timestamps, adds lag decomposition + sim vclock-based oracular checks, adds checkpoint tasklet. |
| lib/python/flame/launch/spawner.py | Adds optional CPU pinning with affinity discovery and per-trainer core assignment; adds CLI toggle. |
| lib/python/flame/launch/runner.py | Makes batch runner non-interactive in CI/non-tty; adds between-experiment straggler sweep and GPU settle wait. |
| lib/python/flame/channel.py | Threads timeout into recv_fifo streamer to prevent active-task leaks; avoids enqueueing timeouts; adds stronger cleanup via finally. |
| lib/python/examples/async_cifar10/trainer/pytorch/test_memory_profiler.py | Ensures enabled profiler path is exercised and disabled profiler truly no-ops. |
| lib/python/examples/async_cifar10/trainer/pytorch/memory_profiler.py | Adds enabled flag to avoid expensive heap walks by default; documents operational cost. |
| lib/python/examples/async_cifar10/seshu_first_task.md | Removes obsolete planning doc. |
| lib/python/examples/async_cifar10/scripts/plotters/_annot.py | Adds shared percentile annotation helper for plots. |
| lib/python/examples/async_cifar10/scripts/plotters/init.py | Package marker for plotter helpers. |
| lib/python/examples/async_cifar10/scripts/parity/cli.py | New parity checker CLI supporting single-pair and batch modes with JSON/plot outputs. |
| lib/python/examples/async_cifar10/scripts/parity/init.py | Declares parity package. |
| lib/python/examples/async_cifar10/scripts/parity_checks.py | Compatibility shim re-exporting parity checks from new scripts/parity package. |
| lib/python/examples/async_cifar10/scripts/parity_check.py | Stable CLI entry-point shim wrapping scripts.parity.cli. |
| lib/python/examples/async_cifar10/scripts/overnight_run.sh | New overnight runner script (sim then real) with robust conda activation and logging. |
| lib/python/examples/async_cifar10/scripts/debug_run.sh | New debug runner to generate filtered configs and override runtime for targeted comparisons. |
| lib/python/examples/async_cifar10/scripts/compare_overnight.sh | Post-run parity + cross-baseline comparison automation script. |
| lib/python/examples/async_cifar10/scripts/analyze_send_recv_lag.py | Improves lag statistics reporting and adds optional CDF plot output. |
| lib/python/examples/async_cifar10/scripts/analyze_pinning.py | New analyzer for CPU/GPU placement balance from trainer logs; optional plotting. |
| lib/python/examples/async_cifar10/FEDDANCE_TODO.md | Removes superseded FedDance TODO doc. |
| lib/python/examples/async_cifar10/expt_scripts_2026/refl_n48_parity_seeded_sim.yaml | New seeded REFL parity config (sim). |
| lib/python/examples/async_cifar10/expt_scripts_2026/refl_n48_parity_seeded_real.yaml | New seeded REFL parity config (real). |
| lib/python/examples/async_cifar10/expt_scripts_2026/oort_n48_parity_seeded_sim.yaml | New seeded Oort parity config (sim). |
| lib/python/examples/async_cifar10/expt_scripts_2026/oort_n48_parity_seeded_real.yaml | New seeded Oort parity config (real). |
| lib/python/examples/async_cifar10/expt_scripts_2026/oort_n300_alpha0.1_syn0_STREAMING.yaml | New standalone Oort streaming config (split per-node). |
| lib/python/examples/async_cifar10/expt_scripts_2026/misselection_smoke_n10.yaml | New smoke config to validate checkpoint + oracle mis-selection pipeline end-to-end. |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_oort_refl_feddance_quickpilot_n32.yaml | New short “quick pilot” config running all four selectors sequentially for pipeline validation. |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_oort_refl_feddance_alpha0.1_STREAMING_node2.yaml | New node2 streaming template (refl + feddance). |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_oort_refl_feddance_alpha0.1_STREAMING_node1.yaml | New node1 streaming template (felix + oort). |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_oort_refl_feddance_alpha0.1_SIMULATED_node2.yaml | New node2 simulated-time template (refl + feddance). |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_oort_refl_feddance_alpha0.1_SIMULATED_node1.yaml | New node1 simulated-time template (felix + oort). |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_oort_refl_feddance_alpha0.1_OVERNIGHT_node2.yaml | New overnight node2 template (sim then real), rounds cap raised to avoid early termination. |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_oort_refl_feddance_alpha0.1_OVERNIGHT_node1.yaml | New overnight node1 template (sim then real), rounds cap raised to avoid early termination. |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_oort_refl_feddance_alpha0.1_CONTROL_node2.yaml | New control (no streaming) node2 config templates. |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_oort_refl_feddance_alpha0.1_CONTROL_node1.yaml | New control (no streaming) node1 config templates. |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_n48_stress_sim.yaml | New stress config (sim) for heavy contention and sim-limit mapping. |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_n48_stress_real.yaml | New stress config (real) counterpart for parity comparisons. |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_n48_parity_seeded_sim.yaml | New seeded Felix parity config (sim) in deterministic regime. |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_n48_parity_seeded_real.yaml | New seeded Felix parity config (real) in deterministic regime. |
| lib/python/examples/async_cifar10/expt_scripts_2026/felix_n300_alpha0.1_syn0_STREAMING.yaml | New standalone Felix streaming config (split per-node). |
| lib/python/examples/async_cifar10/expt_scripts_2026/feddance_n48_parity_seeded_sim.yaml | New seeded FedDance parity config (sim). |
| lib/python/examples/async_cifar10/expt_scripts_2026/feddance_n48_parity_seeded_real.yaml | New seeded FedDance parity config (real). |
| lib/python/examples/async_cifar10/expt_scripts_2026/fedbuff_n48_parity_seeded_sim.yaml | New seeded FedBuff parity config (sim). |
| lib/python/examples/async_cifar10/expt_scripts_2026/fedbuff_n48_parity_seeded_real.yaml | New seeded FedBuff parity config (real). |
| lib/python/examples/async_cifar10/expt_scripts_2026/fedavg_n48_parity_seeded_sim.yaml | New seeded FedAvg parity config (sim) to validate sync sim-ordering behavior. |
| lib/python/examples/async_cifar10/expt_scripts_2026/fedavg_n48_parity_seeded_real.yaml | New seeded FedAvg parity config (real) counterpart. |
| lib/python/examples/async_cifar10/experiments/debug_stuck_run.py | Improves stuck-run debugging (UTF-8 stdout guard; adds streaming active-task leak detector). |
| lib/python/examples/async_cifar10/david_first_task.md | Removes obsolete planning doc. |
| lib/python/examples/async_cifar10/ava_first_task.md | Removes obsolete planning doc. |
| lib/python/examples/async_cifar10/aggregator/pytorch/main_oort_sync_agg.py | Gates eval cadence to evalEveryNRounds to reduce per-round cost; cleans comments/log noise. |
| lib/python/examples/async_cifar10/aggregator/pytorch/main_fedavg_agg.py | Gates eval cadence to evalEveryNRounds to reduce per-round cost. |
| lib/python/examples/async_cifar10/aggregator/pytorch/main_asyncfl_agg.py | Removes verbose prints and minor comment cleanups. |
| lib/python/examples/_metadata/baselines.yaml | Adds missing REFL aggr_num; clarifies FedDance baseline stack/availability and disables oracular tracking. |
| .gitignore | Ignores generated async_cifar10 run outputs and a few local artifacts. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| preserves the legacy blocking behavior used by synchronous callers. | ||
| """ | ||
|
|
||
| async def _get_inner(end_id) -> tuple[str, Any]: |
Comment on lines
+427
to
+432
| # Modeled round compute: max(real_gpu_time, training_delay_s). Stamped | ||
| # unconditionally (real + sim) so the aggregator can decompose the | ||
| # trainer-side lag into delivery + compute + post-wait in both modes. | ||
| _compute_s = getattr(self, "_sim_round_duration", None) | ||
| if _compute_s is not None: | ||
| msg[MessageType.ROUND_COMPUTE_S] = float(_compute_s) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Investigates and documents the root causes of real-vs-sim divergence in async (felix) and sync (REFL) FL runs: felix was running at c=10 instead of c=30, the round cap of 1000 caused early termination before the 3h budget, and the vclock advance model doesn't account for inter-round overlap or per-commit system overhead. Adds a rigorous parity check battery (scripts/parity/) covering availability, selection, training, utility, aggregation, and clock/throughput (34 checks, K1–K10 clock suite), with a single stable CLI entry point (parity_check.py) and --batch multi-baseline mode. Config fixes (c=30, rounds=20000) applied across all overnight/debug/streaming YAMLs, and stale artifacts, planning docs, and superseded scripts removed.