From d76919d6a27d6f19fe4eace9e74dde3ace1450c9 Mon Sep 17 00:00:00 2001
From: jgrusewski <jgrusewski@localhost>
Date: Thu, 28 May 2026 00:28:25 +0200
Subject: [PATCH] perf(rl): gate diag staging to every 10th step
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 crates/ml-alpha/examples/alpha_rl_train.rs | 54 +++++++++++++---------
 1 file changed, 31 insertions(+), 23 deletions(-)

diff --git a/crates/ml-alpha/examples/alpha_rl_train.rs b/crates/ml-alpha/examples/alpha_rl_train.rs
index c769b998a..379e9f7f6 100644
--- a/crates/ml-alpha/examples/alpha_rl_train.rs
+++ b/crates/ml-alpha/examples/alpha_rl_train.rs
@@ -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