gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(rl): gate diag staging to every 10th step

Browse Source 
At mega-graph speeds (329 sps), the diag sync_and_swap blocks for
100ms+ because the DtoD copies haven't finished in the 3ms step time.
Gate sync+snapshot to every 10th step (or log/checkpoint boundaries).
The DiagFrame still sends every step using stale staging data — the
background writer drops most frames anyway via try_send(1).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-05-28 00:28:25 +02:00
parent 9c1b70edec
commit d76919d6a2
1 changed files with 31 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
crates/ml-alpha/examples/alpha_rl_train.rs
View File

@@ -1021,30 +1021,38 @@ fn main() -> Result<()> {
// so the diag copies have had the entire training step to finish —
// the event sync is instant (~0 us). On step 0 this is a no-op
// (event sync on a freshly-created event returns immediately).
diag_staging.sync_and_swap().context("diag sync_and_swap")?;
// Diag staging: only sync+snapshot every 10 steps or on log/checkpoint
// boundaries. At mega-graph speeds (300+ sps), the per-step diag
// sync_and_swap blocks for 100ms+ (diag DtoD copies haven't finished
// in 3ms). Skipping 9/10 steps keeps the GPU saturated.
let diag_this_step = step % 10 == 0
|| step % cli.log_every == 0
|| (cli.checkpoint_every > 0 && step % cli.checkpoint_every == 0)
|| step == start_step
|| step + 1 == cli.n_steps;
// Launch async DtoD copies of ALL diag buffers into the current
// staging buffer. Non-blocking on the training stream — the copies
// run on diag_staging's separate stream.
diag_staging
.snapshot_async(
trainer.isv_dev_ptr,
trainer.rewards_d.raw_ptr(),
trainer.dones_d.raw_ptr(),
trainer.actions_d.raw_ptr(),
trainer.raw_rewards_d.raw_ptr(),
trainer.trade_duration_emit_d.raw_ptr(),
trainer.outcome_ema_d.raw_ptr(),
trainer.prev_position_lots_d.raw_ptr(),
trainer.pyramid_units_count_d.raw_ptr(),
trainer.unit_entry_price_d.raw_ptr(),
trainer.unit_entry_step_d.raw_ptr(),
trainer.unit_lots_d.raw_ptr(),
trainer.unit_trail_distance_d.raw_ptr(),
trainer.close_unit_index_d.raw_ptr(),
trainer.frd_logits_d.raw_ptr(),
)
.context("diag snapshot_async")?;
if diag_this_step {
diag_staging.sync_and_swap().context("diag sync_and_swap")?;
diag_staging
.snapshot_async(
trainer.isv_dev_ptr,
trainer.rewards_d.raw_ptr(),
trainer.dones_d.raw_ptr(),
trainer.actions_d.raw_ptr(),
trainer.raw_rewards_d.raw_ptr(),
trainer.trade_duration_emit_d.raw_ptr(),
trainer.outcome_ema_d.raw_ptr(),
trainer.prev_position_lots_d.raw_ptr(),
trainer.pyramid_units_count_d.raw_ptr(),
trainer.unit_entry_price_d.raw_ptr(),
trainer.unit_entry_step_d.raw_ptr(),
trainer.unit_lots_d.raw_ptr(),
trainer.unit_trail_distance_d.raw_ptr(),
trainer.close_unit_index_d.raw_ptr(),
trainer.frd_logits_d.raw_ptr(),
)
.context("diag snapshot_async")?;
}
// ── Snapshot diagnostic data into owned DiagFrame (~0.1ms). ──
// Reads from the DiagStaging double-buffer (previous step's