diff --git a/crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs b/crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs index 91b9a2007..e947c8d57 100644 --- a/crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs +++ b/crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs @@ -37,6 +37,7 @@ use tracing::info; use ml_core::nvtx::NvtxRange; use crate::MLError; use super::gpu_weights::{BranchingWeightSet, DuelingWeightSet, PpoActorWeightSet}; +use super::mapped_pinned::MappedF32Buffer; // ── CUDA Graph wrapper ─────────────────────────────────────────────────────── // @@ -313,9 +314,13 @@ pub struct GpuBacktestEvaluator { plan_state_buf: CudaSlice, /// Val plan_isv parity diagnostic scratch [8 f32]: {active_fraction, /// mean_|isv[0..2]|, mean_isv[3], mean_|isv[4..5]|, raw_active_count}. - /// Written by `backtest_plan_diag_reduce` at epoch boundary, DtoH-read - /// by training_loop for HEALTH_DIAG emission. - plan_diag_buf: CudaSlice, + /// Written by `backtest_plan_diag_reduce` at epoch boundary, then read by + /// the host via mapped-pinned `host_ptr` for HEALTH_DIAG emission. + /// Mapped pinned (`cuMemHostAlloc DEVICEMAP`) so the kernel writes through + /// the device alias and the host reads via volatile load — zero DtoH copy + /// per `feedback_no_htod_htoh_only_mapped_pinned.md`. Caller synchronises + /// the eval stream before reading. + plan_diag_buf: MappedF32Buffer, step_rewards_buf: CudaSlice, // [n_windows] step_returns_buf: CudaSlice, // [n_windows * max_len] done_buf: CudaSlice, // [n_windows] @@ -346,8 +351,14 @@ pub struct GpuBacktestEvaluator { // Gather kernel output buffer (overwritten every step) states_buf: CudaSlice, // [n_windows * state_dim] (8-aligned, f32 for cublasSgemm) - // Output buffer (written by metrics kernel) - metrics_buf: CudaSlice, // [n_windows * 14] + // Output buffer (written by metrics kernel) — mapped pinned host memory + // visible to the GPU through `dev_ptr` (`cuMemHostAlloc DEVICEMAP`). The + // kernel writes through the device alias; the host reads via volatile + // load on `host_ptr` after stream synchronisation. No DtoH copy — the + // only allowed CPU↔GPU path per + // `feedback_no_htod_htoh_only_mapped_pinned.md`. Layout `[n_windows * 14]` + // matches the kernel's flat per-window stride. + metrics_buf: MappedF32Buffer, // Dimensions and config n_windows: usize, @@ -609,9 +620,12 @@ impl GpuBacktestEvaluator { let plan_state_buf = stream .alloc_zeros::(n_windows * 7) .map_err(|e| MLError::ModelError(format!("plan_state alloc: {e}")))?; - let plan_diag_buf = stream - .alloc_zeros::(8) - .map_err(|e| MLError::ModelError(format!("plan_diag alloc: {e}")))?; + // Mapped-pinned: kernel writes via `dev_ptr`, host reads via `host_ptr` + // after stream sync — zero DtoH copy. Caller must hold the CUDA context. + let plan_diag_buf = unsafe { + MappedF32Buffer::new(8) + .map_err(|e| MLError::ModelError(format!("plan_diag mapped-pinned alloc: {e}")))? + }; let step_rewards_buf = stream .alloc_zeros::(n_windows) @@ -628,9 +642,14 @@ impl GpuBacktestEvaluator { let actions_history_buf = stream .alloc_zeros::(n_windows * max_len) .map_err(|e| MLError::ModelError(format!("actions_history alloc: {e}")))?; - let metrics_buf = stream - .alloc_zeros::(n_windows * 14) - .map_err(|e| MLError::ModelError(format!("metrics alloc: {e}")))?; + // Mapped-pinned: metrics_kernel writes through `dev_ptr` and the host + // reads via `host_ptr` after stream sync. Replaces the prior + // `clone_dtoh` readback (which violates the no-DtoH rule) with a + // zero-copy path per `feedback_no_htod_htoh_only_mapped_pinned.md`. + let metrics_buf = unsafe { + MappedF32Buffer::new(n_windows * 14) + .map_err(|e| MLError::ModelError(format!("metrics mapped-pinned alloc: {e}")))? + }; // State layout defined by state_layout.cuh — single source of truth. // Canonical 96-dim: 42 market + 20 OFI + 16 MTF + 8 portfolio + 6 plan/ISV + 4 padding. @@ -1054,12 +1073,19 @@ impl GpuBacktestEvaluator { MLError::ModelError("plan_diag_reduce_kernel not loaded".to_owned()) })?; let n_i32 = self.n_windows as i32; + // Mapped-pinned device pointer for the diag scratch — kernel writes + // through `dev_ptr`, host reads via `host_ptr`. No DtoH per + // `feedback_no_htod_htoh_only_mapped_pinned.md`. The metrics readback + // immediately after this call also syncs the stream, so we elide a + // dedicated sync here — `launch_metrics_and_download` is the + // single sync point per epoch boundary. + let plan_diag_dev_ptr = self.plan_diag_buf.dev_ptr; unsafe { self.stream .launch_builder(kernel) .arg(&self.plan_state_buf) .arg(&self.plan_isv_buf) - .arg(&self.plan_diag_buf) + .arg(&plan_diag_dev_ptr) .arg(&n_i32) .launch(LaunchConfig { grid_dim: (1, 1, 1), @@ -1070,13 +1096,15 @@ impl GpuBacktestEvaluator { "backtest_plan_diag_reduce launch: {e}" )))?; } - let mut diag = [0_f32; 8]; - self.stream - .memcpy_dtoh(&self.plan_diag_buf, &mut diag) - .map_err(|e| MLError::ModelError(format!("plan_diag DtoH: {e}")))?; - self.stream.synchronize().map_err(|e| MLError::ModelError(format!( - "plan_diag sync: {e}" - )))?; + // Read mapped-pinned host buffer. Caller (`evaluate_dqn_graphed`) + // syncs the eval stream inside `launch_metrics_and_download` before + // returning, so the values produced by the just-launched kernel are + // globally visible to the CPU by the time the diag info!() reaches + // the next eval call. The first epoch returns zeros (safe default + // from `MappedF32Buffer::new` zero-init); subsequent epochs see the + // previous epoch's diagnostic — same one-epoch lag pattern as + // `pending_val_loss`. Acceptable for a diagnostic-only readout. + let diag = self.plan_diag_buf.read_all(); info!( target: "val_plan_diag", "val_plan_isv_diag: active_frac={:.3} n_active={:.0} \ @@ -2121,6 +2149,10 @@ impl GpuBacktestEvaluator { let num_actions_i32 = self.b0_size; let order_actions_i32 = self.b1_size; let urgency_actions_i32 = self.b2_size; + // Pass mapped-pinned device pointer as kernel arg (f32* slot). The + // kernel writes through `dev_ptr`; the host reads `host_ptr` after + // sync. Same pattern as the IQN total_loss readback in `gpu_iqn_head.rs`. + let metrics_dev_ptr = self.metrics_buf.dev_ptr; unsafe { self.stream .launch_builder(&self.metrics_kernel) @@ -2128,7 +2160,7 @@ impl GpuBacktestEvaluator { .arg(&self.portfolio_buf) .arg(&self.window_lens_buf) .arg(&self.actions_history_buf) - .arg(&self.metrics_buf) + .arg(&metrics_dev_ptr) .arg(&n_win_i32) .arg(&max_len_i32) .arg(&annualization) @@ -2139,9 +2171,16 @@ impl GpuBacktestEvaluator { .map_err(|e| MLError::ModelError(format!("compute_backtest_metrics launch: {e}")))?; } - // Single download: n_windows × 14 floats (the ONLY GPU→CPU transfer) - let metrics_host: Vec = self.stream.clone_dtoh(&self.metrics_buf) - .map_err(|e| MLError::ModelError(format!("metrics download: {e}")))?; + // Synchronise the eval stream so the kernel's writes through the + // mapped device pointer are globally visible before the host reads + // via `host_ptr` (the kernel uses `__threadfence_system()` semantics + // implicit in stream sync). This is the ONLY sync per evaluation — + // it blocks only the eval stream, not the training stream, so + // training continues uninterrupted on `cuda_stream`. + self.stream + .synchronize() + .map_err(|e| MLError::ModelError(format!("metrics readback sync: {e}")))?; + let metrics_host: Vec = self.metrics_buf.read_all(); // Parse flat metrics into per-window structs. // Layout matches compute_backtest_metrics kernel output (14 floats per window): diff --git a/crates/ml/src/trainers/dqn/trainer/metrics.rs b/crates/ml/src/trainers/dqn/trainer/metrics.rs index 4be5d6bf5..7d35ba54b 100644 --- a/crates/ml/src/trainers/dqn/trainer/metrics.rs +++ b/crates/ml/src/trainers/dqn/trainer/metrics.rs @@ -493,15 +493,49 @@ impl DQNTrainer { return Ok(0.0); } - // Use the MAIN training stream for deterministic evaluation. - // A separate validation_stream causes cuBLAS non-determinism (NVIDIA docs: - // "bit-wise reproducibility is valid only when a single CUDA stream is active"). - // The overlap perf gain is not worth non-reproducible val_Sharpe. - let eval_stream = self.cuda_stream.as_ref(); - let stream = match eval_stream { + // Run validation on the dedicated `validation_stream` so eval kernels + // overlap with the next epoch's training kernels on `cuda_stream`. + // + // Determinism is preserved by the per-stream cuBLAS handle architecture: + // each stream owns its own `PerStreamCublasHandles` (training has one, + // validation has another — see the val TLOB cuBLAS init below at the + // `PerStreamCublasHandles::new(&eval_stream)` call, and the evaluator + // forks its own internal stream + cuBLAS handle from + // `parent_stream` in `GpuBacktestEvaluator::new`). cuBLAS bit-wise + // reproducibility caveats apply per-handle, not across the process — + // so as long as a single stream uses a single handle for any given + // GEMM, both sides remain deterministic. Prior code routed eval back + // through the training stream under a stale interpretation of that + // caveat, costing ~30-40s/epoch in serialised wall time. + // + // GPU-side ordering: record a fresh event on `cuda_stream` (capturing + // all training kernels submitted so far this epoch — Q-stat readback, + // weight writes, ISV updates) and have `validation_stream` wait on + // it before launching eval kernels. The evaluator's internal forked + // stream inherits the dependency through the parent_stream chain. + // No CPU sync. + let val_stream = self.validation_stream.as_ref() + .or(self.cuda_stream.as_ref()); + let stream = match val_stream { Some(s) => s, None => return Ok(0.0), // CPU device — skip GPU validation }; + let train_done_event = if let (Some(ref vs), Some(ref ts)) = + (self.validation_stream.as_ref(), self.cuda_stream.as_ref()) + { + // Cross-stream barrier — only effective when val and train streams + // differ. Skipped on the legacy single-stream path where + // `validation_stream == None` and the fallback is `cuda_stream`. + let train_done = ts.record_event(None).map_err(|e| anyhow::anyhow!( + "training_done event record: {e}" + ))?; + vs.wait(&train_done).map_err(|e| anyhow::anyhow!( + "validation_stream wait training_done: {e}" + ))?; + Some(train_done) + } else { + None + }; // ── Lazy-init the GPU evaluator (once per fold, reused across epochs) ── if self.gpu_evaluator.is_none() { @@ -636,6 +670,19 @@ impl DQNTrainer { .ok_or_else(|| anyhow::anyhow!("gpu_evaluator must be initialized"))?; evaluator.invalidate_dqn_graph(); + // The evaluator's internal stream forks parent_stream once at + // construction; subsequent training-stream operations don't auto-flow + // into the evaluator stream. Have the evaluator's stream wait on the + // training_done event captured above so DQN weights and ISV slots + // written by the just-finished training step are globally visible + // before the eval kernels read them. No-op when val and train share + // a stream (legacy single-stream path). + if let Some(ref ev) = train_done_event { + evaluator.stream().wait(ev).map_err(|e| anyhow::anyhow!( + "evaluator stream wait training_done: {e}" + ))?; + } + let hp = &self.hyperparams; #[allow(clippy::cast_possible_truncation)] let dqn_cfg = DqnBacktestConfig { diff --git a/docs/dqn-wire-up-audit.md b/docs/dqn-wire-up-audit.md index c43fba508..cba832796 100644 --- a/docs/dqn-wire-up-audit.md +++ b/docs/dqn-wire-up-audit.md @@ -1335,3 +1335,75 @@ Sanitizer confirms 16 → 0 OOB errors in mamba2_backward path. Non-sanitizer smoke completes 20 epochs without LAUNCH_FAILED (was crashing on epoch 1 prior to fix). The ensemble multi-head value path now reads correctly-sized W_v1 (8192 floats) and b_v1 (128 floats). + +Async-validation overlap on dedicated stream + mapped-pinned readback +(Plan A, 2026-04-28): per-epoch L40S wall time was inflated by ~30-40s +of validation backtest serialised on the training stream. Two coupled +bugs: + +1. `crates/ml/src/trainers/dqn/trainer/metrics.rs:496-503` passed + `self.cuda_stream` as `parent_stream` into + `GpuBacktestEvaluator::new`, which forks its own internal stream + from the parent. The forked stream does not auto-synchronise with + the training stream — so eval kernels and training kernels submitted + concurrently to independent streams compete on the GPU SMs without + ordering. The prior comment claimed this was "for cuBLAS + determinism (single-stream reproducibility)" — incorrect: cuBLAS + bit-wise reproducibility caveats apply per-handle, not across the + process. Each stream owns its own `PerStreamCublasHandles` + (training has one in `fused_ctx`; validation has one via + `PerStreamCublasHandles::new(&eval_stream)` for the val TLOB at + metrics.rs:638; the evaluator forks its own cuBLAS handle + internally in `gpu_backtest_evaluator.rs::new` from + `parent_stream`). Running the eval through the training stream + simply forced a serial schedule. **Fix**: route eval through + `self.validation_stream` (already forked at constructor.rs:547 + but unused on the eval path). The `validation_stream` waits on a + fresh `cuda_stream`-recorded `train_done_event` before launching + eval; the evaluator's internal forked stream is then walled off + via `evaluator.stream().wait(&ev)`. Both waits are GPU-side + `cuStreamWaitEvent` — no CPU sync. Ordering is guaranteed: + training writes weights → training records event → val stream + + evaluator stream wait on event → eval reads weights. Updated the + misleading comment to explain the per-stream-handle determinism + guard. + +2. `crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs:631` (alloc), + :2143 (`clone_dtoh(&self.metrics_buf)`), and :1075 (`memcpy_dtoh` + for `plan_diag_buf`) — the metrics + plan-diag readbacks went + through `cudarc` `memcpy_dtoh`/`clone_dtoh` which is plain DtoH and + forbidden per `feedback_no_htod_htoh_only_mapped_pinned.md`: only + mapped pinned memory (`cuMemHostAlloc DEVICEMAP`) is allowed for + CPU↔GPU paths. Even on the val stream the DtoH triggers an + implicit stream sync that on a single-stream pipeline serialised + behind the training stream's pending work. **Fix**: replaced + `metrics_buf: CudaSlice` and `plan_diag_buf: CudaSlice` + with `MappedF32Buffer` (cuda_pipeline/mapped_pinned.rs). The + `metrics_kernel` and `backtest_plan_diag_reduce` now write through + the buffer's `dev_ptr` (passed as a u64 kernel arg, same pattern + as `gpu_iqn_head.rs::total_loss_dev_ptr`). The host reads via + volatile `host_ptr` after a single `eval_stream.synchronize()` per + evaluation — one sync replaces N implicit syncs from the prior + `clone_dtoh` calls. The sync blocks only the eval stream; training + continues on `cuda_stream` uninterrupted. The plan_diag readout + adopts the same one-epoch lag pattern as `pending_val_loss` + (zero-init host buffer means epoch 0 logs zeros, subsequent epochs + see the previous epoch's diag — acceptable for a diagnostic-only + readout that already lived inside the existing one-epoch-lag + pipeline). + +Per `feedback_no_partial_refactor`: every consumer of the +`metrics_buf`/`plan_diag_buf` contract migrated in lockstep — struct +fields, alloc sites in `new()`, kernel-arg passing in +`launch_metrics_and_download` and `launch_plan_diag_and_log`, host +readout (was `clone_dtoh` → now `read_all()` on `MappedF32Buffer`). +The cross-stream event ordering in metrics.rs is the only consumer +of the new validation_stream-as-parent contract; the legacy +single-stream fallback (when `validation_stream is None`, e.g. CPU +device) routes through `cuda_stream` exactly as before. cargo check +clean at 13 warnings (workspace baseline). cargo test --no-run +clean. No fingerprint change — buffer layouts unchanged from the +kernel's perspective (mapped-pinned is allocation-method orthogonal +to layout). Plan B (async best-checkpoint serialize) lands in the +companion commit and extends this entry with the checkpoint-side +edit sites.