From ce1e13519bc6c1011f8c87d843edb8eed2e98347 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sat, 23 May 2026 14:14:55 +0200 Subject: [PATCH] =?UTF-8?q?fix(rl):=20mapped-pinned=20for=20all=20R7d/R8?= =?UTF-8?q?=20CPU=E2=86=94GPU=20paths=20+=20diag=20JSONL=20+=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two concerns in one commit since they're entangled: ## 1. feedback_no_htod_htoh_only_mapped_pinned violations R7d (PER push/sample) + R8 (CLI binary) + the new per-step diag dump shipped with 8 raw `stream.memcpy_htod` / `stream.memcpy_dtoh` calls. The rule is explicit: "mapped-pinned only for CPU↔GPU; tests not exempt." A raw `stream.memcpy_*` on a regular `&[T]` / `&mut [T]` is NOT mapped-pinned — the source/dest slice isn't page-locked, so the CUDA driver does an internal blocking HtoD/DtoH that stalls the stream. Refactored all 8 violations to use the mapped-pinned + DtoD pattern (cuMemHostAlloc DEVICEMAP — host writes via `host_ptr`, kernel reads `dev_ptr`, DtoD between them via `cudarc::driver::result::memcpy_dtod_async`). New shared helpers in `trainer/integrated.rs`: * `read_slice_i32_d` — DtoH for `i32` device buffers via `MappedI32Buffer` staging. Counterpart to the existing `read_slice_d` (f32 version). * `write_slice_f32_d` — CPU→GPU upload for `f32` via `MappedF32Buffer.write_from_slice` + DtoD into destination. * `write_slice_i32_d` — CPU→GPU upload for `i32` via `MappedI32Buffer.host_slice_mut().copy_from_slice` + DtoD. `pub fn` wrappers (`read_slice_*_d_pub`) expose the f32/i32 helpers to the CLI binary so the per-step diag DtoH uses the same canonical pattern. Call-site refactors: * `push_to_replay`: 4× `stream.memcpy_dtoh` → `read_slice_*_d`. * `sample_and_gather`: 3× `stream.memcpy_htod` → `write_slice_*_d`. * `step_with_lobsim` pre-PER ISV refresh: raw `memcpy_dtoh` → `read_slice_d` (424 floats per step). * `step_with_lobsim` post-Q PER priority TD readback: raw `memcpy_dtoh` → `read_slice_d` (b_size floats per step). * `step_synthetic` ISV mirror refresh: raw `memcpy_dtoh` → `read_slice_d` (pre-existing pre-R9 violation; fixed in the same commit since it's the same pattern in the same file). * Init-time (one-shot) `prng_state` upload: raw `memcpy_htod` → inline mapped-pinned DtoD (custom because cast through i32 for the u32 buffer). * Init-time (one-shot) `atom_supports` upload: raw `memcpy_htod` → `write_slice_f32_d`. * `examples/alpha_rl_train.rs` per-step diag DtoH (3 calls) → `read_slice_*_d_pub`. ## 2. Pre-commit guard gap — diff-aware HtoD/DtoH check The existing GPU hot-path guard (`scripts/gpu-hotpath-guard.sh`) EXPLICITLY skips memcpy_htod/dtoh on the assumption that such calls only appear in `cuda_pipeline/` (where mapped-pinned is the convention). That assumption was falsified by R7d/R8 — the guard shipped 8 violations green. Added `check_no_raw_htod_dtoh` to `scripts/pre-commit-hook.sh` (the real file behind the `.git/hooks/pre-commit` symlink). The check is DIFF-AWARE: it greps only the `+` lines of `git diff --cached -U0`, so pre-existing violations elsewhere (143 sites across the codebase) don't block commits touching unrelated files. NEW additions of `\.memcpy_(htod|dtoh)\(` are flagged with a clear error pointing at the mapped-pinned alternative. Suppress per-line with `// gpu-ok: ` (same convention as the existing guards). Pre-existing violations in `ml-alpha/src/aux_heads.rs`, `mamba2_block.rs`, `cfc/`, `data/`, etc. are a separate cleanup — not blocked by this commit's check because the diff-aware filter ignores anything that was already on `HEAD~1`. ## Verified gates (post-fix, local sm_86) G1 isv_bootstrap ✅ unchanged G3 controllers_emit ✅ unchanged G4 target_soft_update ✅ unchanged G6 r7d_per_wiring ✅ unchanged (PER round-trips all mapped-pinned now) R3 ema/advantage (3 tests) ✅ unchanged R4 action kernels (3 tests) ✅ unchanged end integrated_trainer_smoke ✅ unchanged Mapped-pinned is semantically equivalent to raw memcpy_htod/dtoh — just routed through page-locked staging so the driver doesn't have to do its own internal pinning. Behaviour identical; the cost shifts from "driver hidden HtoD per call" to "mapped-pinned alloc + DtoD per call." For the smoke (b_size=1, 1000 steps), the cost difference is in the microseconds. ## Per-step diag JSONL (separate concern, same commit) Added `--diag-jsonl ` flag to `alpha_rl_train.rs` (default: `/diag.jsonl`). After each `step_with_lobsim`, writes one JSON record capturing: * step number, elapsed wall time * all 5 head losses + λs * all 7 RL controller outputs (γ τ ε coef n_roll per_α scale) * all 5 per-head learning rates (lr_bce/q/pi/v/aux) * all 7 EMA inputs the controllers consume * replay buffer length * per-step reward stats (sum, max, min, abs_max) * per-step done count * per-step action histogram (9 action classes) Critical for cluster smoke debugging — the prior CLI only flushed an `eprintln` progress line every N steps (default 100), making in-flight controller drift / replay stagnation / reward explosion invisible until they produced a NaN abort. The JSONL is line- buffered + flushed every `log_every` steps so `tail -f` shows progress live. The stderr progress line is also beefed up to include γ / ε / per_α / reward_scale / dones / rew_sum at each tick so a casual `argo logs` inspection sees the controller behaviour without parsing JSONL. ## Why R9 cluster submission needs this Without the diag dump, an R9 1000-step smoke is "blind" — only the final summary tells us what happened. With the dump, post-hoc analysis can answer: * Did the controllers adapt or stay at bootstrap? * Did the reward scale stabilise or saturate? * Did the PER buffer fill? * Was the action histogram dominated by any one action? * Where did the per-head losses converge to? Co-Authored-By: Claude Opus 4.7 --- crates/ml-alpha/examples/alpha_rl_train.rs | 157 +++++++++++++- crates/ml-alpha/src/trainer/integrated.rs | 225 +++++++++++++++++---- scripts/pre-commit-hook.sh | 57 ++++++ 3 files changed, 393 insertions(+), 46 deletions(-) diff --git a/crates/ml-alpha/examples/alpha_rl_train.rs b/crates/ml-alpha/examples/alpha_rl_train.rs index c75058eb6..86e8d6149 100644 --- a/crates/ml-alpha/examples/alpha_rl_train.rs +++ b/crates/ml-alpha/examples/alpha_rl_train.rs @@ -37,13 +37,23 @@ use ml_alpha::data::loader::{ DEFAULT_OUTCOME_LABEL_COST_ES, }; use ml_alpha::heads::HORIZONS; +use ml_alpha::rl::isv_slots::{ + RL_ADVANTAGE_VAR_RATIO_EMA_INDEX, RL_ENTROPY_COEF_INDEX, RL_ENTROPY_OBSERVED_EMA_INDEX, + RL_GAMMA_INDEX, RL_KL_PI_EMA_INDEX, RL_LR_AUX_INDEX, RL_LR_BCE_INDEX, RL_LR_PI_INDEX, + RL_LR_Q_INDEX, RL_LR_V_INDEX, RL_MEAN_ABS_PNL_EMA_INDEX, RL_MEAN_TRADE_DURATION_EMA_INDEX, + RL_N_ROLLOUT_STEPS_INDEX, RL_PER_ALPHA_INDEX, RL_PPO_CLIP_INDEX, RL_Q_DIVERGENCE_EMA_INDEX, + RL_REWARD_SCALE_INDEX, RL_TARGET_TAU_INDEX, RL_TD_KURTOSIS_EMA_INDEX, +}; use ml_alpha::trainer::integrated::{ - IntegratedStepStats, IntegratedTrainer, IntegratedTrainerConfig, + read_slice_d_pub, read_slice_i32_d_pub, IntegratedStepStats, IntegratedTrainer, + IntegratedTrainerConfig, }; use ml_alpha::trainer::perception::PerceptionTrainerConfig; use ml_backtesting::sim::LobSimCuda; use ml_core::device::MlDevice; use serde::Serialize; +use serde_json::json; +use std::io::{BufWriter, Write}; use std::path::PathBuf; use std::process; @@ -118,6 +128,30 @@ struct Cli { /// visibility only. #[arg(long, default_value_t = 100)] log_every: usize, + + /// Per-step diagnostic JSONL path. One record per step capturing: + /// * step number, wall time + /// * loss components (l_bce/q/pi/v/aux/total) + λs (5) + /// * all 7 RL controller outputs (γ τ ε coef n_roll per_α scale) + /// * all 5 per-head learning rates (lr_bce/q/pi/v/aux) + /// * all 7 EMA inputs the controllers consume + /// * replay buffer length + /// * per-step reward + done aggregates (DtoH b_size floats per stat) + /// * per-step action histogram (counts per action class) + /// + /// Default: `/diag.jsonl`. Pass `--diag-jsonl /dev/null` (or + /// any unwritable path that errors at open) to disable. The dump + /// happens AFTER each `step_with_lobsim` so it sees the freshly- + /// adapted ISV values. Writes are line-buffered + flushed every + /// `log_every` steps so a `tail -f` shows in-flight progress. + /// + /// Critically: this provides the per-step diagnostic visibility + /// the cluster smoke needs to detect anomalies early. Without it, + /// the smoke is "blind" — only the final summary is observable, + /// and a controller drifting off-anchor mid-run is invisible until + /// it produces a non-finite loss (G8 NaN abort). + #[arg(long)] + diag_jsonl: Option, } fn parse_instrument_mode(s: &str) -> Result { @@ -261,6 +295,23 @@ fn main() -> Result<()> { }; let mut last_stats: Option = None; + // ── Per-step JSONL diag writer. ────────────────────────────────── + // Default path: `/diag.jsonl`. The smoke run's failure-mode + // catalogue (NaN abort, controller drift, replay stagnation, reward + // explosion) is detected via post-hoc inspection of this file — + // without it, the run is blind beyond the eprintln every-N-step + // ticks. + let diag_path = cli + .diag_jsonl + .clone() + .unwrap_or_else(|| cli.out.join("diag.jsonl")); + let diag_file = std::fs::File::create(&diag_path) + .with_context(|| format!("create diag jsonl {}", diag_path.display()))?; + let mut diag = BufWriter::new(diag_file); + eprintln!("per-step diag JSONL: {}", diag_path.display()); + + // Per-step scratch (replaced each step via mapped-pinned helpers). + // ── Training loop. ─────────────────────────────────────────────── let t_start = std::time::Instant::now(); for step in 0..cli.n_steps { @@ -304,23 +355,125 @@ fn main() -> Result<()> { process::exit(2); } + // ── Per-step diag dump (mapped-pinned reads). ──────────────── + // Per `feedback_no_htod_htoh_only_mapped_pinned`: raw + // `stream.memcpy_dtoh` on a regular `&mut [T]` is forbidden; + // the only permitted CPU↔GPU path is mapped-pinned staging + // (cuMemHostAlloc DEVICEMAP → host writes via host_ptr + + // kernel reads dev_ptr + DtoD between them). The + // `read_slice_*_d_pub` helpers in `trainer::integrated` + // encapsulate the pattern. + let dev_stream = dev.cuda_stream().context("cuda_stream for diag")?; + let actions_host = read_slice_i32_d_pub(dev_stream, &trainer.actions_d, cli.n_backtests) + .context("diag: read actions_d")?; + let rewards_host = read_slice_d_pub(dev_stream, &trainer.rewards_d, cli.n_backtests) + .context("diag: read rewards_d")?; + let dones_host = read_slice_d_pub(dev_stream, &trainer.dones_d, cli.n_backtests) + .context("diag: read dones_d")?; + + let mut act_hist = [0u32; 9]; + for &a in &actions_host { + if (0..9).contains(&a) { + act_hist[a as usize] += 1; + } + } + let reward_sum: f32 = rewards_host.iter().sum(); + let reward_max = rewards_host.iter().cloned().fold(f32::NEG_INFINITY, f32::max); + let reward_min = rewards_host.iter().cloned().fold(f32::INFINITY, f32::min); + let reward_abs_max = rewards_host.iter().map(|r| r.abs()).fold(0.0f32, f32::max); + let done_count: u32 = dones_host.iter().map(|&d| if d > 0.5 { 1 } else { 0 }).sum(); + + let isv = &trainer.isv_host; + let record = json!({ + "step": step, + "elapsed_s": t_start.elapsed().as_secs_f32(), + "loss": { + "bce": stats.l_bce, + "q": stats.l_q, + "pi": stats.l_pi, + "v": stats.l_v, + "aux": stats.l_aux, + "total": stats.l_total, + }, + "lambdas": { + "bce": stats.lambdas.bce, + "q": stats.lambdas.q, + "pi": stats.lambdas.pi, + "v": stats.lambdas.v, + "aux": stats.lambdas.aux, + }, + // 7 R5 controller outputs (the adaptive knobs). + "isv_out": { + "gamma": isv[RL_GAMMA_INDEX], + "target_tau": isv[RL_TARGET_TAU_INDEX], + "ppo_clip_eps": isv[RL_PPO_CLIP_INDEX], + "entropy_coef": isv[RL_ENTROPY_COEF_INDEX], + "n_rollout_steps": isv[RL_N_ROLLOUT_STEPS_INDEX], + "per_alpha": isv[RL_PER_ALPHA_INDEX], + "reward_scale": isv[RL_REWARD_SCALE_INDEX], + }, + // 5 per-head learning rates (rl_lr_controller emits). + "isv_lr": { + "bce": isv[RL_LR_BCE_INDEX], + "q": isv[RL_LR_Q_INDEX], + "pi": isv[RL_LR_PI_INDEX], + "v": isv[RL_LR_V_INDEX], + "aux": isv[RL_LR_AUX_INDEX], + }, + // 7 EMA inputs the controllers consume — knowing these + // makes controller behaviour debuggable (was target wrong? + // was input wrong? was alpha wrong?). + "isv_ema_in": { + "mean_trade_duration": isv[RL_MEAN_TRADE_DURATION_EMA_INDEX], + "q_divergence": isv[RL_Q_DIVERGENCE_EMA_INDEX], + "kl_pi": isv[RL_KL_PI_EMA_INDEX], + "entropy_observed": isv[RL_ENTROPY_OBSERVED_EMA_INDEX], + "advantage_var_ratio": isv[RL_ADVANTAGE_VAR_RATIO_EMA_INDEX], + "td_kurtosis": isv[RL_TD_KURTOSIS_EMA_INDEX], + "mean_abs_pnl": isv[RL_MEAN_ABS_PNL_EMA_INDEX], + }, + "replay_len": trainer.replay.len(), + "rewards": { + "sum": reward_sum, + "max": reward_max, + "min": reward_min, + "abs_max": reward_abs_max, + }, + "done_count": done_count, + // Action histogram: indices match `Action` enum + // (0=ShortLarge, 1=ShortSmall, 2=Hold, 3=FlatFromLong, + // 4=FlatFromShort, 5=LongSmall, 6=LongLarge, + // 7=TrailTighten, 8=TrailLoosen). + "action_hist": act_hist.to_vec(), + }); + writeln!(diag, "{}", record).context("diag: writeln jsonl")?; + if step % cli.log_every == 0 || step == cli.n_steps - 1 { + diag.flush().context("diag: flush")?; eprintln!( "step {:>6}/{}: l_q={:.4} l_pi={:.4} l_v={:.4} l_total={:.4} \ - replay={} elapsed={:.1}s", + γ={:.4} ε={:.4} per_α={:.4} scale={:.4} \ + replay={} dones={} rew_sum={:.3} elapsed={:.1}s", step, cli.n_steps, stats.l_q, stats.l_pi, stats.l_v, stats.l_total, + isv[RL_GAMMA_INDEX], + isv[RL_PPO_CLIP_INDEX], + isv[RL_PER_ALPHA_INDEX], + isv[RL_REWARD_SCALE_INDEX], trainer.replay.len(), + done_count, + reward_sum, t_start.elapsed().as_secs_f32() ); } last_stats = Some(stats); summary.n_steps_completed = step + 1; } + diag.flush().context("diag: final flush")?; // ── Compose + emit summary. ────────────────────────────────────── if let Some(s) = last_stats { diff --git a/crates/ml-alpha/src/trainer/integrated.rs b/crates/ml-alpha/src/trainer/integrated.rs index 5a9f4a40d..6794584fe 100644 --- a/crates/ml-alpha/src/trainer/integrated.rs +++ b/crates/ml-alpha/src/trainer/integrated.rs @@ -85,7 +85,7 @@ use ml_core::device::MlDevice; use crate::cfc::snap_features::Mbp10RawInput; use crate::heads::HIDDEN_DIM; -use crate::pinned_mem::MappedF32Buffer; +use crate::pinned_mem::{MappedF32Buffer, MappedI32Buffer}; use rand::{Rng, SeedableRng}; use crate::rl::common::{N_ACTIONS, Q_N_ATOMS, Q_V_MAX, Q_V_MIN}; @@ -690,12 +690,34 @@ impl IntegratedTrainer { let prng_seeds: Vec = (0..b_size) .map(|_| prng_host_rng.gen::().max(1)) .collect(); - let mut prng_state_d = stream + let prng_state_d = stream .alloc_zeros::(b_size) .context("alloc prng_state_d")?; - stream - .memcpy_htod(&prng_seeds, &mut prng_state_d) - .context("memcpy_htod prng_state_d")?; + // Mapped-pinned upload via inline DtoD per + // `feedback_no_htod_htoh_only_mapped_pinned`. One-shot at + // construction. Cast through i32 for the typed mapped-pinned + // buffer (u32 and i32 share width); the kernel reads as u32 — + // wraparound is well-defined. + { + let mut staging = unsafe { MappedI32Buffer::new(b_size) } + .map_err(|e| anyhow::anyhow!("prng_state staging: {e}"))?; + let dst = staging.host_slice_mut(); + for (i, &v) in prng_seeds.iter().enumerate() { + dst[i] = v as i32; + } + unsafe { + let s = stream.cu_stream(); + let (dev_dst, _g) = prng_state_d.device_ptr(&stream); + cudarc::driver::result::memcpy_dtod_async( + dev_dst, + staging.dev_ptr, + b_size * std::mem::size_of::(), + s, + ) + .context("prng_state DtoD upload")?; + } + stream.synchronize().context("prng_state sync")?; + } // Atom supports — structural constant (atom span [-1, +1]). // Single allocation + upload at init; read every step by the @@ -707,9 +729,10 @@ impl IntegratedTrainer { let mut atom_supports_d = stream .alloc_zeros::(Q_N_ATOMS) .context("alloc atom_supports_d")?; - stream - .memcpy_htod(&atom_supports_host, &mut atom_supports_d) - .context("memcpy_htod atom_supports_d")?; + // Mapped-pinned upload per + // `feedback_no_htod_htoh_only_mapped_pinned`. One-shot. + write_slice_f32_d(&stream, &atom_supports_host, &mut atom_supports_d) + .context("atom_supports upload")?; // Phase R6 per-step buffers. prev_* are sentinel-zero at init // so the first call's delta is `current - 0 = current`. For a @@ -1396,9 +1419,13 @@ impl IntegratedTrainer { // (sentinel-zero → 1e-3); subsequent calls Wiener-α blend. self.launch_rl_lr_controller() .context("rl_lr_controller launch")?; - self.stream - .memcpy_dtoh(&self.isv_d, self.isv_host.as_mut_slice()) - .context("isv dtoh")?; + // Mapped-pinned ISV mirror refresh per + // `feedback_no_htod_htoh_only_mapped_pinned` — was a raw + // `stream.memcpy_dtoh` (forbidden) pre-R9 audit. + let isv_host_fresh = + read_slice_d(&self.stream, &self.isv_d, self.isv_host.len()) + .context("step_synthetic: read isv")?; + self.isv_host.copy_from_slice(&isv_host_fresh); let lambdas = read_loss_lambdas_from_isv(&self.isv_host); // Mutate each per-head Adam's lr field from the ISV mirror. The // controller has just emitted into ISV[412..417], so this read @@ -2236,9 +2263,14 @@ impl IntegratedTrainer { // is later (Step 2 of step_synthetic) — too late for this // pre-step_synthetic call. The cost is one DtoH of the full // ISV slice (424 floats), bounded. - self.stream - .memcpy_dtoh(&self.isv_d, self.isv_host.as_mut_slice()) - .context("step_with_lobsim: isv dtoh (pre-PER)")?; + // Mapped-pinned staging for ISV mirror refresh per + // `feedback_no_htod_htoh_only_mapped_pinned`. The full ISV + // slice is 424 floats; one mapped-pinned alloc + DtoD per + // step is bounded. + let isv_host_fresh = + read_slice_d(&self.stream, &self.isv_d, self.isv_host.len()) + .context("step_with_lobsim: read isv (pre-PER)")?; + self.isv_host.copy_from_slice(&isv_host_fresh); self.push_to_replay(b_size) .context("step_with_lobsim: push_to_replay")?; @@ -2269,10 +2301,9 @@ impl IntegratedTrainer { // step's sample picks the high-TD transitions per the // canonical Schaul 2015 PER recipe. if !per_indices.is_empty() { - let mut td_per_sample_host = vec![0.0f32; b_size]; - self.stream - .memcpy_dtoh(&self.td_per_sample_d, td_per_sample_host.as_mut_slice()) - .context("step_with_lobsim: dtoh td_per_sample_d")?; + let td_per_sample_host = + read_slice_d(&self.stream, &self.td_per_sample_d, b_size) + .context("step_with_lobsim: read td_per_sample_d")?; self.replay .update_priorities(&per_indices, &td_per_sample_host); } @@ -2303,23 +2334,20 @@ impl IntegratedTrainer { fn push_to_replay(&mut self, b_size: usize) -> Result<()> { use crate::rl::common::{N_ACTIONS, Transition}; - // DtoH per-batch metadata. 4 × b_size floats/ints total. - let mut actions_host = vec![0i32; b_size]; - let mut rewards_host = vec![0.0f32; b_size]; - let mut dones_host = vec![0.0f32; b_size]; - let mut log_pi_old_host = vec![0.0f32; b_size]; - self.stream - .memcpy_dtoh(&self.actions_d, actions_host.as_mut_slice()) - .context("push_to_replay: dtoh actions_d")?; - self.stream - .memcpy_dtoh(&self.rewards_d, rewards_host.as_mut_slice()) - .context("push_to_replay: dtoh rewards_d")?; - self.stream - .memcpy_dtoh(&self.dones_d, dones_host.as_mut_slice()) - .context("push_to_replay: dtoh dones_d")?; - self.stream - .memcpy_dtoh(&self.log_pi_old_d, log_pi_old_host.as_mut_slice()) - .context("push_to_replay: dtoh log_pi_old_d")?; + // Per-batch metadata via mapped-pinned staging (the only + // permitted CPU↔GPU path per + // `feedback_no_htod_htoh_only_mapped_pinned`). Each helper + // allocates a mapped-pinned buffer, issues a DtoD copy from + // the device source, syncs, and returns a host Vec read from + // the page-mapped host_ptr. + let actions_host = read_slice_i32_d(&self.stream, &self.actions_d, b_size) + .context("push_to_replay: read actions_d")?; + let rewards_host = read_slice_d(&self.stream, &self.rewards_d, b_size) + .context("push_to_replay: read rewards_d")?; + let dones_host = read_slice_d(&self.stream, &self.dones_d, b_size) + .context("push_to_replay: read dones_d")?; + let log_pi_old_host = read_slice_d(&self.stream, &self.log_pi_old_d, b_size) + .context("push_to_replay: read log_pi_old_d")?; // For each batch index, alloc per-transition device buffers // and DtoD-copy the per-batch slice of h_t / h_tp1 in. @@ -2409,15 +2437,14 @@ impl IntegratedTrainer { rewards_host[i] = t.reward; dones_host[i] = if t.done { 1.0 } else { 0.0 }; } - self.stream - .memcpy_htod(&actions_host, &mut self.sampled_actions_d) - .context("sample_and_gather: htod sampled_actions_d")?; - self.stream - .memcpy_htod(&rewards_host, &mut self.sampled_rewards_d) - .context("sample_and_gather: htod sampled_rewards_d")?; - self.stream - .memcpy_htod(&dones_host, &mut self.sampled_dones_d) - .context("sample_and_gather: htod sampled_dones_d")?; + // Mapped-pinned staging — the only permitted CPU→GPU path + // per `feedback_no_htod_htoh_only_mapped_pinned`. + write_slice_i32_d(&self.stream, &actions_host, &mut self.sampled_actions_d) + .context("sample_and_gather: write sampled_actions_d")?; + write_slice_f32_d(&self.stream, &rewards_host, &mut self.sampled_rewards_d) + .context("sample_and_gather: write sampled_rewards_d")?; + write_slice_f32_d(&self.stream, &dones_host, &mut self.sampled_dones_d) + .context("sample_and_gather: write sampled_dones_d")?; // Per-batch DtoD: per-transition h_t / h_tp1 device buffers → // contiguous `sampled_h_t_d` / `sampled_h_tp1_d` at slot i. @@ -2678,6 +2705,116 @@ fn accumulate_grad_h( // remaining `read_*_d` helpers are still load-bearing for the per-step // loss-scalar readback inside step_synthetic's Q + V backward chain. +/// Read `n` `i32`s from a device buffer into a host `Vec` via +/// mapped-pinned staging + DtoD + stream sync. Same pattern as +/// `read_slice_d` (f32 version) — the only permitted CPU↔GPU path +/// per `feedback_no_htod_htoh_only_mapped_pinned`. Used by R7d's +/// `push_to_replay` to read the per-step actions device buffer for +/// PER metadata staging. +pub fn read_slice_i32_d_pub( + stream: &Arc, + src: &CudaSlice, + n: usize, +) -> Result> { + read_slice_i32_d(stream, src, n) +} + +pub fn read_slice_d_pub( + stream: &Arc, + src: &CudaSlice, + n: usize, +) -> Result> { + read_slice_d(stream, src, n) +} + +fn read_slice_i32_d( + stream: &Arc, + src: &CudaSlice, + n: usize, +) -> Result> { + debug_assert!(src.len() >= n); + if n == 0 { + return Ok(Vec::new()); + } + let mut staging = unsafe { MappedI32Buffer::new(n) } + .map_err(|e| anyhow::anyhow!("read_slice_i32_d staging: {e}"))?; + unsafe { + let (src_ptr, _g) = src.device_ptr(stream); + cudarc::driver::result::memcpy_dtod_async( + staging.dev_ptr, + src_ptr, + n * std::mem::size_of::(), + stream.cu_stream(), + ) + .context("read_slice_i32_d DtoD")?; + } + stream.synchronize().context("read_slice_i32_d sync")?; + Ok(staging.host_slice_mut().to_vec()) +} + +/// Write a host `&[f32]` slice into a device buffer via mapped-pinned +/// staging + DtoD. The only permitted CPU→GPU path per +/// `feedback_no_htod_htoh_only_mapped_pinned` — `stream.memcpy_htod` +/// on a regular `&[T]` is forbidden (the source slice is not +/// page-locked, so the driver does an internal HtoD blocking copy). +/// Mapped-pinned staging keeps the host write zero-copy on the +/// device side. +fn write_slice_f32_d( + stream: &Arc, + src: &[f32], + dst: &mut CudaSlice, +) -> Result<()> { + debug_assert!(dst.len() >= src.len()); + let n = src.len(); + if n == 0 { + return Ok(()); + } + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| anyhow::anyhow!("write_slice_f32_d staging: {e}"))?; + staging.write_from_slice(src); + unsafe { + let (dst_ptr, _g) = dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async( + dst_ptr, + staging.dev_ptr, + n * std::mem::size_of::(), + stream.cu_stream(), + ) + .context("write_slice_f32_d DtoD")?; + } + stream.synchronize().context("write_slice_f32_d sync")?; + Ok(()) +} + +/// Write a host `&[i32]` slice into a device buffer via mapped-pinned +/// staging + DtoD. See `write_slice_f32_d` for rationale. +fn write_slice_i32_d( + stream: &Arc, + src: &[i32], + dst: &mut CudaSlice, +) -> Result<()> { + debug_assert!(dst.len() >= src.len()); + let n = src.len(); + if n == 0 { + return Ok(()); + } + let mut staging = unsafe { MappedI32Buffer::new(n) } + .map_err(|e| anyhow::anyhow!("write_slice_i32_d staging: {e}"))?; + staging.host_slice_mut()[..n].copy_from_slice(src); + unsafe { + let (dst_ptr, _g) = dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async( + dst_ptr, + staging.dev_ptr, + n * std::mem::size_of::(), + stream.cu_stream(), + ) + .context("write_slice_i32_d DtoD")?; + } + stream.synchronize().context("write_slice_i32_d sync")?; + Ok(()) +} + fn read_scalar_d(stream: &Arc, src: &CudaSlice) -> Result { debug_assert!(src.len() >= 1); let staging = unsafe { MappedF32Buffer::new(1) } diff --git a/scripts/pre-commit-hook.sh b/scripts/pre-commit-hook.sh index f60dccf5f..9c4032eef 100755 --- a/scripts/pre-commit-hook.sh +++ b/scripts/pre-commit-hook.sh @@ -227,10 +227,67 @@ check_no_isv_migrations() { fi } +# Diff-aware guard: any NEW `stream.memcpy_htod(` or `stream.memcpy_dtoh(` +# line added in this commit is a violation of +# `feedback_no_htod_htoh_only_mapped_pinned`. The only permitted CPU↔GPU +# path is mapped-pinned staging (cuMemHostAlloc DEVICEMAP — host writes +# via `host_ptr`, kernels read `dev_ptr`, DtoD copy between them via +# `cudarc::driver::result::memcpy_dtod_async`). +# +# The existing gpu-hotpath-guard.sh explicitly skips memcpy_htod/dtoh +# (see scripts/gpu-hotpath-guard.sh:103-105) on the assumption that +# such calls only appear in cuda_pipeline/ where mapped-pinned is the +# convention. That assumption was falsified in 2026-05-23 when the R7d +# PER push/sample + R8 CLI diag landed 8 raw `stream.memcpy_htod/dtoh` +# calls in ml-alpha/src/trainer/ and ml-alpha/examples/ — the guard +# was deliberately blind to them and they shipped. +# +# This diff-aware check fires on NEW + lines only, so pre-existing +# violations elsewhere don't block commits touching unrelated files. +# Suppress per-line with `// gpu-ok: `. +check_no_raw_htod_dtoh() { + local staged + staged=$(git diff --cached --name-only --diff-filter=ACM | grep '\.rs$' || true) + if [ -z "$staged" ]; then return 0; fi + + local bad="" + while IFS= read -r f; do + [ -z "$f" ] && continue + local hits + hits=$(git diff --cached -U0 -- "$f" 2>/dev/null \ + | grep -E '^\+[^+]' \ + | grep -E '\.memcpy_(htod|dtoh)\(' \ + | grep -v 'gpu-ok:' \ + || true) + if [ -n "$hits" ]; then + bad+="${bad:+$'\n'}$f:"$'\n'"$hits" + fi + done <<< "$staged" + + if [ -n "$bad" ]; then + echo "❌ feedback_no_htod_htoh_only_mapped_pinned violation: raw" + echo " stream.memcpy_htod/dtoh on a regular &[T]/&mut [T] is" + echo " forbidden. The source/dest slice is not page-locked, so the" + echo " CUDA driver does an internal blocking HtoD/DtoH copy." + echo " Use the mapped-pinned pattern instead: MappedF32Buffer /" + echo " MappedI32Buffer (host_ptr write + dev_ptr DtoD via" + echo " cudarc::driver::result::memcpy_dtod_async). See the" + echo " read_slice_d / write_slice_*_d helpers in" + echo " crates/ml-alpha/src/trainer/integrated.rs for the canonical" + echo " reference implementation." + echo "" + echo " Lines flagged (NEW additions only; pre-existing violations" + echo " are a separate cleanup tracked outside this guard):" + echo "$bad" | sed 's/^/ /' + return 1 + fi +} + check_audit_doc_updates || exit 1 check_no_todo_fixme || exit 1 check_no_isv_migrations || exit 1 check_no_dtod_via_pinned || exit 1 +check_no_raw_htod_dtoh || exit 1 check_sp18_consumer_audit || exit 1 echo "✅ All pre-commit checks passed!"