Streaming data, more telemetry, init simulation implementation - #61
Merged
Conversation
Remove the orphaned main_oort_trainer.py: the launcher always spawns trainer/pytorch/main.py for every baseline, and main.py's base class already provides the oort methods, so it serves oort/refl too. Add an opt-in data-streaming mode to main.py: instead of exposing the full partition at init, retain the full pool plus a one-time shuffle (seeded by trainer_id) and reveal a growing prefix over time. Controlled by a data_streaming config block (enabled, full_data_available_after_s); unlock is linear in sim time (scaled by speedup_factor), floored at 1 sample, and stays at 100% after X. The loader is rebuilt per selection in train()/evaluate(); disabled by default in trainer_base.yaml. Document the feature for future example migrations in MIGRATING_TO_LAUNCHER.md (§3 + §5) instead of a standalone doc. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
…wiring and async availability robustness Telemetry (generic, reused via base classes; no-op unless FLAME_TELEMETRY_DIR set): - flame/telemetry: JSONL TelemetryWriter + typed event schema (selection, agg_eval, agg_round, trainer_round, util_disparity, avail_change) - selector AbstractSelector.emit_selection() centralizes the selection schema (availability composition, utility/speed, chosen set) for cross-selector comparison; wired into oort/async_oort/fedbuff/feddance - aggregator loss/acc via base update_metrics; staleness/agg-goal/in-flight in sync+async _aggregate_weights - trainer per-round timing/availability + opt-in streamed-vs-full counterfactual statistical-utility (util_counterfactual config) - scripts/analysis/analyze_run.py + plot_helpers.py: post-run plot bundle with a --compare mode; launcher auto-runs it and exports FLAME_TELEMETRY_DIR per process - tests/telemetry: writer + schema + streaming-disparity tests Launcher / async fixes (required to run speedup/availability experiments): - spawner now passes --speedup_factor/--battery_threshold to the trainer (were ignored, so speedup_factor always fell back to 1.0) - recv_fifo gains an optional timeout; async aggregate receive is now bounded (RECV_TIMEOUT_WAIT_S) so a quiet/ghost in-flight trainer can't hang the loop; skip cycle when no live ends remain - async_oort drops departed ends from selected_ends to free concurrency - UTF-8 launcher stdio so status glyphs don't crash on latin-1 locales See examples/async_cifar10/TELEMETRY_AND_SPEEDUP_PLAN.md for the Task 2 follow-ups (virtual-clock decoupling, in-flight reconciliation, decorrelated availability). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Core simulated-mode bug fixes (aggregator): - run_async: cancel future on timeout so stale _get_message_inner coroutines don't steal messages from _rx_queue (fix #1) - _sim_recv_min: buffer messages keyed by actual sender (metadata[0]) not the probed end, preventing wrong-end commits (fix #2) - _aggregate_weights: bypass recv_ends==None guard when _sim_buffer has entries; probing marks ends RECVD so all 10 end up RECVD after 3 calls, stranding buffered messages forever (fix #3) - Guard PROP_ROUND_DURATION.total_seconds() against None crash Simulator fidelity fix (trainer): - sim_round_duration = real_gpu_time_s + training_delay_s (was just D). Without GPU time, all trainers report identical D, making sim_completion_ts equal for every trainer selected in the same asyncfl-loop iteration → deterministic OORT over-selection of the same top-5 every round. Adding GPU jitter gives OORT real speed heterogeneity, replicating real-mode arrival diversity. Speedup improvements: - SIM_RECV_FILL_TIMEOUT_S: 2.0s → 0.5s (probe granularity, not training wait; shorter = less wasted time per still-computing trainer) - Inter-send sleep: 0.5s → 0.2s in simulated mode (broker still protected; trainers cycle faster in sim so lower interval is safe) - evalEveryNRounds config (default 10): skip aggregator evaluation on non-eval rounds, saves ~2s/round Experiment runner fix: - runner.py: wait for aggregator.process.wait() before calling wait_all(). Previously wait_all fired immediately after spawn, killing each trainer after 30s regardless of training progress (caused runs to stall after ~44 rounds with 10 trainers) New tooling: - scripts/compare_parity.py: 6-dimensional real-vs-sim comparator (selection Jaccard, utility KS-stat, aggregation sequence, staleness, participation counts, convergence curves) - scripts/analyze_send_recv_lag.py: post-run SEND_RECV_LAG parser with per-trainer p50/p95/max and threshold warnings - expt_scripts_2026/felix_n10_parity_real_vs_sim.yaml: sequential real+sim parity experiment (same config, only time_mode differs) - SEND_RECV_LAG instrumentation in _aggregate_weights for wall-clock send→receive lag logging per trainer per round Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Root cause: in simulated mode, all N trainers respond immediately so all
N messages are delivered in the first asyncfl-loop iteration of each
round. Only agg_goal (5) are committed; the remaining N-agg_goal
trainers have their messages buffered in _sim_buffer but are never freed
from OORT's all_selected.
The deadlock: a trainer can only leave all_selected by being committed
(cleanup_recvd_ends processes ordered_updates_recv_ends). But it can only
be committed if OORT selects it. But OORT excludes all_selected from
filtered_ends. So: locked in all_selected → never selectable → never
committed → never freed. Permanent.
Traced via telemetry: from round 3 onward, filtered_ends was always
exactly {370,373,375,376,379} — the same 5 trainers that happened to
commit first in rounds 1-2. Trainers {371,372,374,377,378} (including
trainer 378 with the HIGHEST stat-utility) were completely locked out
for 98 of 100 rounds.
Fix: after cleanup_recvd_ends() at round end, release all trainers
remaining in all_selected (selected but not committed). Their buffered
messages remain valid in _sim_buffer as legitimate stale updates for
future rounds. State is reset to NONE so OORT can re-probe them.
Effect: with all trainers eligible each round, OORT's temporal
uncertainty term (sqrt(0.1*log(V)/last_selected)) accumulates for
unselected trainers and rotates selection correctly. At round 50, a
trainer not selected since round 1 has temporal bonus ~0.625 which
outweighs the ~0.1 bonus of a trainer selected last round — matching
the rotation behavior seen in real-mode runs.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…luding old refl yamls
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.
No description provided.