Fix sim/real parity: unified max(gpu,D) timing model + GPU contention…

lib/python/examples/async_cifar10/README.md

@@ -148,8 +148,17 @@ tail -f eurosys26_expts/trainer_logs/log_trainer_*.log
 
 ## Next steps
 
-- First task for a new contributor: [`ava_first_task.md`](ava_first_task.md)
-  (split the trainer delay into compute + RTT, then make RTT time-varying).
+First-task guides for new contributors (each is self-contained; start from a
+passing smoke test run):
+
+| File | Task summary |
+|------|-------------|
+| [`ava_first_task.md`](ava_first_task.md) | Split trainer delay into compute + RTT; make RTT time-varying |
+| [`david_first_task.md`](david_first_task.md) | Add per-trainer lat/long from a mobility trace; emit location as telemetry each round |
+| [`seshu_first_task.md`](seshu_first_task.md) | Make `training_delay_s` sinusoidally time-varying (±20% of mean, 0.5 s floor) |
+
+All three tasks target **`time_mode: real`** (wall-clock mode). Simulation mode
+is still being validated and is not the target for these tasks.
 
 ## References
 

lib/python/examples/async_cifar10/TELEMETRY_AND_SPEEDUP_PLAN.md

@@ -40,9 +40,27 @@ Implemented and unit-verified (156 tests green incl. `tests/sim` + `tests/mode`)
 - Trainer: `real` sleeps `D`; `simulated` skips sleeps, reports `sim_completion_ts`/`D`, evaluates availability + streaming against sim-time (`_sim_now`), and skips (rather than busy-waits) when unavailable in sim mode.
 - Async aggregator: stamps `sim_send_ts` on distribute; `simulated` commits via `_sim_recv_min` (reorder buffer → ascending `sim_completion_ts`, advances `T_v`) and sources `PROP_ROUND_DURATION` from `D`; `real` keeps arrival-order FIFO. Bounded-recv guard retained.
 
+## Update (2026-06-01): unified timing model + GPU contention tracking
+
+**Unified `max(gpu, D)` round duration** ([trainer/pytorch/main.py](trainer/pytorch/main.py), [asyncfl/top_aggregator.py](../../flame/mode/horizontal/asyncfl/top_aggregator.py)):
+- Previously both modes used a fixed `D` for round duration. Now `sim_round_duration = max(gpu_time, D)` in all modes: no contention → `D` (correct device speed); GPU overrun → `gpu > D` (OORT correctly deprioritizes the contended trainer). `sim_completion_ts = sim_send_ts + max(gpu, D)` mirrors real `recv_ts − sent_ts = max(gpu, D)` exactly.
+- Real mode sleep: `time.sleep(max(0, D − gpu))` — absorbs remaining budget. No sleep on overrun.
+- Aggregator: `PROP_ROUND_DURATION = timedelta(SIM_ROUND_DURATION)` in sim mode; `recv_ts − sent_ts` in real mode. Both yield `max(gpu, D)`.
+
+**Budget tracking + overrun detection** (`MessageType.TRAINING_BUDGET_S = 37`):
+- Trainer sends `training_delay_s` (D) as `TRAINING_BUDGET_S` on every update (both modes).
+- Aggregator: `[TIMING_OVERRUN_AGG]` warning when virtual elapsed ≤ 0 (sim) or wall lag > budget (real). Trainer: `[TIMING_OVERRUN]` warning when `gpu > D` with advice to reduce trainers-per-GPU or add GPUs.
+
+**OORT crash fix** ([selector/async_oort.py](../../flame/selector/async_oort.py)):
+- Root cause: `timedelta(0)` round duration → `system_utility = 0` for all trainers → `np.random.choice(replace=False, p=[...])` fails when all probabilities are zero. Fixed at the source (overrun now yields `gpu > D > 0`, never zero duration) plus a defensive zero-probability filter in `sample_by_util`.
+
+**Compare-parity enhancements** ([scripts/compare_parity.py](scripts/compare_parity.py)):
+- Check 9: GPU contention analysis — per-trainer overrun fraction vs budget; FAIL >25%, WARN >10%; graceful on old runs without `training_budget_s`.
+- `--plot-out PATH`: 2-panel timing sanity plot — per-trainer mean GPU bar + budget marker (top), round-by-round deviation mean/max (bottom). Overrun zone highlighted in red.
+
 Remaining / known gaps:
 - **Sync aggregator**: init + `sim_send_ts` stamping done, but the `first_k`-by-`sim_completion_ts` selection and sim-sourced round duration are **not** implemented (sync still uses arrival order in sim mode). The primary target (felix) is async; sync (oort/refl/fedavg) sim-ordering is a follow-up.
-- **End-to-end two-run verification** (Q1: `real` vs `simulated` per-round parity + speedup) needs a GPU/MQTT run — not yet executed.
+- **End-to-end two-run verification** (Q1: `real` vs `simulated` per-round parity + speedup) needs a GPU/MQTT run — in progress (initial runs done; parity investigation ongoing).
 - **Availability in sim mode** uses the trainer's own trace evaluated at the stamped sim-time; correlated-trace / oracular-at-`T_v` selection nuances (and the client-notify-vs-oracular question) want runtime validation.
 - `examples/fwdllm` still hardcodes `speedup_factor=1.0` (separate example; left as-is).
 
@@ -72,7 +90,7 @@ This is more efficient than scaling sleeps: wall-clock is bounded by *real GPU t
 
 **Q1 — Two-run verification and the equivalence expectation.** Run the same experiment in `real` and `simulated` (fixed seed). Both read availability / data-visibility / speed / ordering off the *same logical times*, so they're expected to produce the **same per-(model-version) selection sets, staleness, and loss/accuracy** — "same per round/model-version, not per wall-clock-second" — with `simulated` wall-clock ≪ `real`. Equivalence is **exact in a deterministic scripted scenario** (well-separated `D`, controlled/mocked GPU time → arrival order == sim-completion order) and **within tolerance in a live smoke** (the only divergence source is real-GPU jitter reordering two trainers whose `D` are very close). Verify both: scripted parity test + smoke per-round parity & speedup.
 
-**Q2 — Trainer speed / system-utility source.** In `simulated` mode the wall-clock recv−send delta collapses to ~GPU time and misrepresents speed, so `PROP_ROUND_DURATION` is sourced from `sim_completion_ts − sim_send_ts` (= `D`). In `real` mode the real delta already ≈ `D` (the sleep dominates), so it stays authentic. Either way the value used for Oort's system-utility is the modeled duration. (The current `recv−send` path is *already* wrong under the old `speedup_factor>1`; removing speedup + this fix resolve it.)
+**Q2 — Trainer speed / system-utility source.** In `simulated` mode the wall-clock recv−send delta collapses to ~GPU time and misrepresents speed, so `PROP_ROUND_DURATION` is sourced from `SIM_ROUND_DURATION = max(gpu, D)`. In `real` mode the real delta `recv_ts − sent_ts = max(gpu, D)` (sleep absorbs remaining budget). Both modes deliver the same quantity to OORT: correct device speed when no contention, true overrun cost when GPU is shared. (The current `recv−send` path is *already* wrong under the old `speedup_factor>1`; removing speedup + this fix resolve it.)
 
 **Q3 — Dual timestamps (real + simulated) for lineage.** Keep both on every event record and in the aggregator's `_track_trainer_version_duration_s` (`real_ts` + `sim_ts`, send & recv), so lineage and wall-clock behavior stay reconstructable. Telemetry events gain `sim_ts`/`sim_completion_ts` alongside the existing real `ts`.
 

lib/python/examples/async_cifar10/david_first_task.md

@@ -0,0 +1,149 @@
+# First task: add time-varying trainer location (lat/long) from a mobility trace
+
+This builds on the async_cifar10 example. Follow [README](README.md) first and
+confirm you can run the felix smoke test end-to-end before starting here.
+
+The goal is to give each trainer a location identity that moves over time — like a
+real mobile device. The trainer should read a per-device lat/long trace, interpolate
+its current coordinates based on wall-clock time elapsed since it started, and emit
+location as structured telemetry each round.
+
+The smoke test to target is the **real-mode** felix smoke:
+```bash
+python -m flame.launch.run_experiment \
+    lib/python/examples/async_cifar10/expt_scripts_2026/felix_n10_alpha100_syn0_smoke.yaml
+```
+(Set `time_mode: real` in the YAML. Simulation mode is still being verified; do
+not target it for this task.)
+
+---
+
+## Background: how trainer metadata flows today
+
+1. **Registry** — `examples/_metadata/trainer_registry.yaml`. Each entry already
+   has a `mobiperf_device_id` (e.g. `device_001`). This is the logical handle
+   that links a trainer to a device profile.
+
+2. **Spawner injection** — `flame/launch/spawner.py` (~line 118-160) reads
+   `trainer_meta` from the registry and injects into each trainer's config:
+   `training_delay_s`, `trainer_indices_list`, all availability traces. You will
+   follow the same pattern to inject a location trace.
+
+3. **Schema** — `flame/config.py`, `Hyperparameters` class defines what fields the
+   trainer config carries. You will add a new field here.
+
+4. **Trainer** — `trainer/pytorch/main.py`. Reads from `config.hyperparameters` at
+   `__init__`. Emits structured telemetry via `telemetry.emit(ev, **fields)` in the
+   `train()` method (~line 721). You will add location reading and emission here.
+
+5. **Telemetry events** — `flame/telemetry/events.py` defines typed event builders
+   (e.g. `build_trainer_round`). You can add a field to `build_trainer_round` or
+   add it to the `extra={}` dict — pick whichever keeps the schema clean.
+
+---
+
+## Task A — Create a per-device location trace and wire it to the registry
+
+Each trainer needs a sequence of `(elapsed_s, lat, lon)` waypoints that represent
+where the device is at a given number of seconds after the experiment starts.
+
+Suggested subtasks:
+
+- [ ] **Design the trace format.** A simple YAML or JSON list per device works:
+  ```yaml
+  device_001:
+    - {elapsed_s: 0,    lat: 37.7749, lon: -122.4194}   # San Francisco
+    - {elapsed_s: 300,  lat: 37.3861, lon: -122.0839}   # Palo Alto
+    - {elapsed_s: 600,  lat: 37.3382, lon: -121.8863}   # San Jose
+  ```
+  Keep the data small (3-5 waypoints per device) for now. You can use any
+  geographically plausible coordinates — real-world city pairs work well.
+  For the 10 smoke-test trainers (device_001 … device_010), 10 entries suffice.
+
+- [ ] **Write a generator script.** Add a script under
+  `examples/_metadata/` (e.g. `generate_location_traces.py`) that produces a
+  `location_traces.yaml` file in the same directory. Use random or hand-chosen
+  routes; document what you chose and why. Check it in alongside the generated
+  file so it can be re-run if routes need changing.
+
+- [ ] **Add to registry.** No need to embed full trace data in
+  `trainer_registry.yaml`; the spawner already loads traces separately. Store the
+  mapping `trainer_id → device_id` (already there as `mobiperf_device_id`) and
+  let the spawner resolve it.
+
+---
+
+## Task B — Inject the location trace via the spawner
+
+Follow the same pattern as availability traces (~line 136-158 in `spawner.py`).
+
+- [ ] **`ExperimentMetadata`.** In `flame/launch/metadata.py` (or wherever
+  `get_mobiperf_trace` is implemented), add a `get_location_trace(trainer_id)`
+  method that loads the device's waypoint list from `location_traces.yaml`.
+
+- [ ] **`spawner.py`.** After the mobility trace injection block, add:
+  ```python
+  config["hyperparameters"]["location_trace"] = self.metadata.get_location_trace(trainer_id)
+  ```
+
+- [ ] **`flame/config.py`.** Add `location_trace: Optional[List[Dict]] = None`
+  to `Hyperparameters` (or `Optional[Any]` if the schema is flexible). Keep the
+  field optional so existing configs without it still load.
+
+---
+
+## Task C — Implement time-varying location in the trainer
+
+- [ ] **Read trace at `__init__`.** Parse `config.hyperparameters.location_trace`
+  into a list of `(elapsed_s, lat, lon)` tuples. Store as `self.location_trace`.
+  If absent or empty, set `self.location_trace = None`.
+
+- [ ] **Interpolate current location.** Write a small helper
+  `_current_location(elapsed_s)` that linearly interpolates `(lat, lon)` from the
+  waypoint list. Use the last waypoint once past the final entry (no extrapolation).
+  ```python
+  def _current_location(self, elapsed_s: float):
+      if not self.location_trace:
+          return None, None
+      # find bounding waypoints and interpolate
+      ...
+  ```
+
+- [ ] **Emit each round.** In `train()`, after computing `_cycle_start`, compute:
+  ```python
+  elapsed = time.time() - self._experiment_start_time
+  lat, lon = self._current_location(elapsed)
+  ```
+  Add `"lat": lat, "lon": lon` to the `extra={}` dict in the `build_trainer_round`
+  telemetry call (~line 741). Log it at DEBUG level too.
+
+- [ ] **`_experiment_start_time`.** Set `self._experiment_start_time = time.time()`
+  once in `__init__` (or on the first train call). Don't reset it per round.
+
+---
+
+## Task D — Verify
+
+- [ ] Run the real-mode smoke test (`time_mode: real`, 10 trainers). Confirm the
+  trainer JSONL telemetry files contain `lat`/`lon` fields in `trainer_round`
+  events.
+- [ ] Write a short script (or add to `compare_parity.py` as a helper) that reads
+  the trainer telemetry and plots each trainer's path (lat vs lon) over time. Two
+  trainers on the same device_id should trace identical paths.
+- [ ] Confirm `lat`/`lon` are `None` for trainers whose config has no
+  `location_trace` (graceful no-op).
+
+---
+
+## Notes & gotchas
+
+- The trainer already has `_sim_now()` which returns wall-clock-since-start in real
+  mode. You can reuse that instead of a separate `_experiment_start_time` if it
+  already tracks what you need.
+- The location feature should be **read-only** — it must not affect selection,
+  training, or aggregation. It is purely telemetry/metadata.
+- Keep trace files small (< 50 KB total). If you want more realistic movement,
+  sample from publicly available GPS datasets; just keep the format consistent.
+- The spawner runs inside the launcher process; the trainer runs as a subprocess.
+  Data passes via the generated config JSON — it must be JSON-serializable (list of
+  dicts, no tuples at the top level).