From 072c1d3f9385e259e8fc10004d116d212411ae3f Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sun, 3 May 2026 11:10:40 +0200 Subject: [PATCH] refactor(mapped_pinned): delete via_pinned helpers, migrate 22 callers to MappedF32/I32/U32Buffer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The *_via_pinned helpers (clone_to_device_f32_via_pinned, upload_f32_via_pinned, clone_to_device_i32_via_pinned, upload_i32_via_pinned, upload_u32_via_pinned) were temporary scaffolding during the SP4 mapped-pinned migration. The canonical replacement is MappedF32Buffer / MappedI32Buffer / MappedU32Buffer in the same module — direct allocation + write_from_slice + device_ptr with no additional indirection. Each caller now inlines the staging + DtoD pattern directly: - Allocate MappedXxxBuffer (cuMemHostAlloc DEVICEMAP) - Write data via write_from_slice (no memcpy, mapped pinned coherence) - alloc_zeros:: CudaSlice for device-resident destination - memcpy_dtod_async from staging.dev_ptr to dst - stream.synchronize() before staging drops This commit migrates all callers atomically (per feedback_no_partial_refactor) and removes the now-unused helpers in the same commit. grep -rn 'via_pinned' returns 0 matches after this lands. The two local file-private helpers in gpu_experience_collector.rs (upload_host_to_cuda_f32_via_pinned / upload_host_to_cuda_i32_via_pinned) were already correctly using MappedF32Buffer directly; they were renamed to drop the _via_pinned suffix. Touched: gpu_attention.rs, gpu_backtest_evaluator.rs, gpu_dqn_trainer.rs, gpu_experience_collector.rs, gpu_her.rs, gpu_iql_trainer.rs, gpu_iqn_head.rs, gpu_ppo_collector.rs, gpu_tlob.rs, gpu_walk_forward.rs, gpu_weights.rs, mapped_pinned.rs, mod.rs, hyperopt/adapters/mamba2.rs, hyperopt/adapters/ppo.rs, trainers/dqn/trainer/training_loop.rs, trainers/ppo.rs, docs/dqn-wire-up-audit.md (18 files, +930/-363 LOC). Cargo check workspace clean. Cargo test ml --lib: 925 passed, 17 failed (all 17 are pre-existing GPU/data infrastructure failures unrelated to this change). Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/ml/src/cuda_pipeline/gpu_attention.rs | 20 +- .../cuda_pipeline/gpu_backtest_evaluator.rs | 108 ++++++++-- .../ml/src/cuda_pipeline/gpu_dqn_trainer.rs | 141 ++++++++++-- .../cuda_pipeline/gpu_experience_collector.rs | 38 +++- crates/ml/src/cuda_pipeline/gpu_her.rs | 64 ++++-- .../ml/src/cuda_pipeline/gpu_iql_trainer.rs | 38 +++- crates/ml/src/cuda_pipeline/gpu_iqn_head.rs | 86 +++++++- .../ml/src/cuda_pipeline/gpu_ppo_collector.rs | 76 +++++-- crates/ml/src/cuda_pipeline/gpu_tlob.rs | 54 ++++- .../ml/src/cuda_pipeline/gpu_walk_forward.rs | 65 +++++- crates/ml/src/cuda_pipeline/gpu_weights.rs | 20 +- crates/ml/src/cuda_pipeline/mapped_pinned.rs | 203 ------------------ crates/ml/src/cuda_pipeline/mod.rs | 158 ++++++++++++-- crates/ml/src/hyperopt/adapters/mamba2.rs | 23 +- crates/ml/src/hyperopt/adapters/ppo.rs | 103 +++++++-- .../src/trainers/dqn/trainer/training_loop.rs | 50 ++++- crates/ml/src/trainers/ppo.rs | 46 +++- docs/dqn-wire-up-audit.md | 2 + 18 files changed, 932 insertions(+), 363 deletions(-) diff --git a/crates/ml/src/cuda_pipeline/gpu_attention.rs b/crates/ml/src/cuda_pipeline/gpu_attention.rs index 3d107f5f0..89fcf1d02 100644 --- a/crates/ml/src/cuda_pipeline/gpu_attention.rs +++ b/crates/ml/src/cuda_pipeline/gpu_attention.rs @@ -23,7 +23,7 @@ use std::sync::Arc; use cudarc::cublaslt::result as cublaslt_result; use cudarc::cublaslt::sys as cublaslt_sys; -use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, LaunchConfig, PushKernelArg}; +use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, DevicePtrMut, LaunchConfig, PushKernelArg}; use tracing::info; use crate::MLError; @@ -369,8 +369,22 @@ impl GpuAttention { let mut params = stream.alloc_zeros::(total_params) .map_err(|e| MLError::ModelError(format!("attention params alloc: {e}")))?; - super::mapped_pinned::upload_f32_via_pinned(&stream, &host_params, &mut params) - .map_err(|e| MLError::ModelError(format!("attention params upload via pinned: {e}")))?; + { + let n = host_params.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("attention params staging alloc: {e}")))?; + staging.write_from_slice(&host_params); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = params.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("attention params DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("attention params sync: {e}")))?; + } + } // ── Allocate intermediate buffers ──────────────────────────────── diff --git a/crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs b/crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs index 9f8edc19f..bf8af5053 100644 --- a/crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs +++ b/crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs @@ -31,7 +31,7 @@ use std::mem::ManuallyDrop; use std::sync::Arc; -use cudarc::driver::{CudaEvent, CudaFunction, CudaGraph, CudaSlice, CudaStream, DevicePtr, LaunchConfig, PushKernelArg}; +use cudarc::driver::{CudaEvent, CudaFunction, CudaGraph, CudaSlice, CudaStream, DevicePtr, DevicePtrMut, LaunchConfig, PushKernelArg}; use tracing::info; use ml_core::nvtx::NvtxRange; @@ -638,17 +638,85 @@ impl GpuBacktestEvaluator { .map_err(|e| MLError::ModelError(format!("backtest_state_gather_chunk load: {e}")))?; // ── Upload read-only data via mapped-pinned staging (no HtoD) ───── - let prices_buf = super::mapped_pinned::clone_to_device_f32_via_pinned(&stream, &flat_prices) - .map_err(|e| MLError::ModelError(format!("prices upload via pinned: {e}")))?; - let features_buf = super::mapped_pinned::clone_to_device_f32_via_pinned(&stream, &flat_features) - .map_err(|e| MLError::ModelError(format!("features upload via pinned: {e}")))?; - let window_lens_buf = super::mapped_pinned::clone_to_device_i32_via_pinned(&stream, &window_lens) - .map_err(|e| MLError::ModelError(format!("window_lens upload via pinned: {e}")))?; + let prices_buf = { + let n = flat_prices.len(); + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("prices staging alloc: {e}")))?; + staging.write_from_slice(&flat_prices); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("prices alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("prices DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("prices sync: {e}")))?; + } + _dst + }; + let features_buf = { + let n = flat_features.len(); + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("features staging alloc: {e}")))?; + staging.write_from_slice(&flat_features); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("features alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("features DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("features sync: {e}")))?; + } + _dst + }; + let window_lens_buf = { + let n = window_lens.len(); + let staging = unsafe { MappedI32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("window_lens staging alloc: {e}")))?; + staging.write_from_slice(&window_lens); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("window_lens alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("window_lens DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("window_lens sync: {e}")))?; + } + _dst + }; // ── Allocate mutable state buffers ──────────────────────────────── let portfolio_init = Self::init_portfolio_state(n_windows, config.initial_capital); - let portfolio_buf = super::mapped_pinned::clone_to_device_f32_via_pinned(&stream, &portfolio_init) - .map_err(|e| MLError::ModelError(format!("portfolio upload via pinned: {e}")))?; + let portfolio_buf = { + let n = portfolio_init.len(); + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("portfolio staging alloc: {e}")))?; + staging.write_from_slice(&portfolio_init); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("portfolio alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("portfolio DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("portfolio sync: {e}")))?; + } + _dst + }; // Kelly stats buffer — zero-initialized (no trades yet). The Kelly // priors + warmup floor in trade_physics.cuh handle the cold start @@ -1465,9 +1533,25 @@ impl GpuBacktestEvaluator { // The portfolio_buf field stays device-resident (the env_step kernel // mutates it on every backtest step). let portfolio_init = Self::init_portfolio_state(self.n_windows, self.config.initial_capital); - let portfolio_f32 = super::mapped_pinned::clone_to_device_f32_via_pinned(&self.stream, &portfolio_init) - .map_err(|e| MLError::ModelError(format!("reset portfolio upload via pinned: {e}")))?; - self.portfolio_buf = portfolio_f32; + self.portfolio_buf = { + let n = portfolio_init.len(); + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("reset portfolio staging alloc: {e}")))?; + staging.write_from_slice(&portfolio_init); + let mut _dst = self.stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("reset portfolio alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&self.stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, self.stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("reset portfolio DtoD: {e}")))?; + } + self.stream.synchronize().map_err(|e| MLError::ModelError(format!("reset portfolio sync: {e}")))?; + } + _dst + }; // Zero done flags, step returns, step rewards self.stream.memset_zeros(&mut self.done_buf) diff --git a/crates/ml/src/cuda_pipeline/gpu_dqn_trainer.rs b/crates/ml/src/cuda_pipeline/gpu_dqn_trainer.rs index 4c7681d28..785e34382 100644 --- a/crates/ml/src/cuda_pipeline/gpu_dqn_trainer.rs +++ b/crates/ml/src/cuda_pipeline/gpu_dqn_trainer.rs @@ -46,7 +46,7 @@ use std::sync::Arc; use cudarc::driver::{ - CudaFunction, CudaSlice, CudaStream, LaunchConfig, PushKernelArg, + CudaFunction, CudaSlice, CudaStream, DevicePtrMut, LaunchConfig, PushKernelArg, }; use tracing::info; @@ -12732,8 +12732,21 @@ impl GpuDqnTrainer { // Fixed clip norm of 1.0 — selectivity gate is a small sigmoid head, clip at 1.0 norm. let mut sel_clip_buf = stream.alloc_zeros::(1) .map_err(|e| MLError::ModelError(format!("alloc sel_clip_buf: {e}")))?; - super::mapped_pinned::upload_f32_via_pinned(&stream, &[1.0_f32], &mut sel_clip_buf) - .map_err(|e| MLError::ModelError(format!("sel_clip_buf upload via pinned: {e}")))?; + { + let host = [1.0_f32]; + let n = host.len(); + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("sel_clip_buf staging alloc: {e}")))?; + staging.write_from_slice(&host); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = sel_clip_buf.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("sel_clip_buf DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("sel_clip_buf sync: {e}")))?; + } // Selectivity Adam step counter — pinned device-mapped (no per-step HtoD copy) let sel_t_pinned: *mut i32 = unsafe { let flags = cudarc::driver::sys::CU_MEMHOSTALLOC_DEVICEMAP; @@ -12886,14 +12899,30 @@ impl GpuDqnTrainer { let mut spec_v_s1 = alloc_f32(&stream, ml_core::state_layout::STATE_DIM, "spec_v_s1")?; let mut spec_u_s2 = alloc_f32(&stream, config.shared_h2, "spec_u_s2")?; let mut spec_v_s2 = alloc_f32(&stream, config.shared_h1, "spec_v_s2")?; - super::mapped_pinned::upload_f32_via_pinned(&stream, &init_u_s1, &mut spec_u_s1) - .map_err(|e| MLError::ModelError(format!("spec_u_s1 upload via pinned: {e}")))?; - super::mapped_pinned::upload_f32_via_pinned(&stream, &init_v_s1, &mut spec_v_s1) - .map_err(|e| MLError::ModelError(format!("spec_v_s1 upload via pinned: {e}")))?; - super::mapped_pinned::upload_f32_via_pinned(&stream, &init_u_s2, &mut spec_u_s2) - .map_err(|e| MLError::ModelError(format!("spec_u_s2 upload via pinned: {e}")))?; - super::mapped_pinned::upload_f32_via_pinned(&stream, &init_v_s2, &mut spec_v_s2) - .map_err(|e| MLError::ModelError(format!("spec_v_s2 upload via pinned: {e}")))?; + { + macro_rules! upload_spec_vec { + ($host:expr, $dst:expr, $label:literal) => {{ + let n = $host.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("{} staging alloc: {e}", $label)))?; + staging.write_from_slice(&$host); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = $dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("{} DtoD: {e}", $label)))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("{} sync: {e}", $label)))?; + } + }}; + } + upload_spec_vec!(init_u_s1, spec_u_s1, "spec_u_s1"); + upload_spec_vec!(init_v_s1, spec_v_s1, "spec_v_s1"); + upload_spec_vec!(init_u_s2, spec_u_s2, "spec_u_s2"); + upload_spec_vec!(init_v_s2, spec_v_s2, "spec_v_s2"); + } // Head spectral norm vectors: value head, adv/branch heads let vh = config.value_h; @@ -12910,10 +12939,38 @@ impl GpuDqnTrainer { let mut $v_name = alloc_f32(&stream, $v_n, $lbl_v)?; let init_u = rand_unit($u_n, &mut rng_state); let init_v = rand_unit($v_n, &mut rng_state); - super::mapped_pinned::upload_f32_via_pinned(&stream, &init_u, &mut $u_name) - .map_err(|e| MLError::ModelError(format!("{} upload via pinned: {e}", $lbl_u)))?; - super::mapped_pinned::upload_f32_via_pinned(&stream, &init_v, &mut $v_name) - .map_err(|e| MLError::ModelError(format!("{} upload via pinned: {e}", $lbl_v)))?; + { + let n = init_u.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("{} staging: {e}", $lbl_u)))?; + staging.write_from_slice(&init_u); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = $u_name.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("{} DtoD: {e}", $lbl_u)))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("{} sync: {e}", $lbl_u)))?; + } + } + { + let n = init_v.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("{} staging: {e}", $lbl_v)))?; + staging.write_from_slice(&init_v); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = $v_name.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("{} DtoD: {e}", $lbl_v)))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("{} sync: {e}", $lbl_v)))?; + } + } }; } @@ -14233,8 +14290,22 @@ impl GpuDqnTrainer { } let mut graph_params = stream.alloc_zeros::(60) .map_err(|e| MLError::ModelError(format!("alloc graph_params: {e}")))?; - super::mapped_pinned::upload_f32_via_pinned(&stream, &graph_params_host, &mut graph_params) - .map_err(|e| MLError::ModelError(format!("graph_params upload via pinned: {e}")))?; + { + let n = graph_params_host.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("graph_params staging alloc: {e}")))?; + staging.write_from_slice(&graph_params_host); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = graph_params.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("graph_params DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("graph_params sync: {e}")))?; + } + } info!("GpuDqnTrainer: branch_graph_message_pass kernel loaded (60 params, near-identity W_msg init)"); // ── Diffusion Q-refinement ──────────────────────────────────────────────── @@ -14288,8 +14359,22 @@ impl GpuDqnTrainer { } let mut denoise_params = stream.alloc_zeros::(DENOISE_TOTAL_PARAMS) .map_err(|e| MLError::ModelError(format!("alloc denoise_params: {e}")))?; - super::mapped_pinned::upload_f32_via_pinned(&stream, &denoise_params_host, &mut denoise_params) - .map_err(|e| MLError::ModelError(format!("denoise_params upload via pinned: {e}")))?; + { + let n = denoise_params_host.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("denoise_params staging alloc: {e}")))?; + staging.write_from_slice(&denoise_params_host); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = denoise_params.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("denoise_params DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("denoise_params sync: {e}")))?; + } + } let cpbi_module_denoise = stream.context().load_cubin(EXPECTED_Q_CUBIN.to_vec()) .map_err(|e| MLError::ModelError(format!("cpbi cubin (denoise): {e}")))?; let q_denoise_kernel = cpbi_module_denoise.load_function("q_denoise_step") @@ -14435,8 +14520,22 @@ impl GpuDqnTrainer { } let mut qlstm_weights = stream.alloc_zeros::(QLSTM_WEIGHT_COUNT) .map_err(|e| MLError::ModelError(format!("alloc qlstm_weights: {e}")))?; - super::mapped_pinned::upload_f32_via_pinned(&stream, &qlstm_weights_host, &mut qlstm_weights) - .map_err(|e| MLError::ModelError(format!("qlstm_weights upload via pinned: {e}")))?; + { + let n = qlstm_weights_host.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("qlstm_weights staging alloc: {e}")))?; + staging.write_from_slice(&qlstm_weights_host); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = qlstm_weights.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("qlstm_weights DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("qlstm_weights sync: {e}")))?; + } + } let qlstm_c = stream.alloc_zeros::(QLSTM_HEAD_DIM * QLSTM_HEAD_DIM) // [64] .map_err(|e| MLError::ModelError(format!("alloc qlstm_c: {e}")))?; let qlstm_n = stream.alloc_zeros::(QLSTM_HEAD_DIM) // [8] diff --git a/crates/ml/src/cuda_pipeline/gpu_experience_collector.rs b/crates/ml/src/cuda_pipeline/gpu_experience_collector.rs index 651079d13..38071d2d6 100644 --- a/crates/ml/src/cuda_pipeline/gpu_experience_collector.rs +++ b/crates/ml/src/cuda_pipeline/gpu_experience_collector.rs @@ -1115,7 +1115,7 @@ impl GpuExperienceCollector { portfolio_init[off + 9] = initial_capital; // [9] prev_equity // [10] hold_time = 0.0, [11] realized_pnl = 0.0 (already zero) } - upload_host_to_cuda_f32_via_pinned(&stream, &portfolio_init, &mut portfolio_states, "portfolio_states init")?; + upload_host_to_cuda_f32(&stream, &portfolio_init, &mut portfolio_states, "portfolio_states init")?; let episode_starts_buf = stream .alloc_zeros::(alloc_episodes) @@ -1137,7 +1137,7 @@ impl GpuExperienceCollector { // `feedback_no_htod_htoh_only_mapped_pinned.md`. let mut epoch_state = stream.alloc_zeros::(epoch_state_init.len()) .map_err(|e| MLError::ModelError(format!("alloc epoch_state: {e}")))?; - upload_host_to_cuda_f32_via_pinned(&stream, &epoch_state_init, &mut epoch_state, "epoch_state init")?; + upload_host_to_cuda_f32(&stream, &epoch_state_init, &mut epoch_state, "epoch_state init")?; // ── Step 7: Allocate output buffers (2x for #7 counterfactual) ─── let total_output = alloc_episodes * alloc_timesteps; @@ -1387,7 +1387,7 @@ impl GpuExperienceCollector { // Init base params to neutral (no adversarial effect). saboteur_select // kernel writes back to this buffer, so destination stays CudaSlice; // upload via mapped-pinned staging + DtoD. - upload_host_to_cuda_f32_via_pinned( + upload_host_to_cuda_f32( &stream, &[1.0_f32, 0.85, 1.0], &mut saboteur_base_buf, "saboteur_base init")?; let saboteur_best_buf = stream.alloc_zeros::(3) .map_err(|e| MLError::ModelError(format!("alloc saboteur_best: {e}")))?; @@ -3262,7 +3262,7 @@ impl GpuExperienceCollector { // episode_starts_buf is GPU-mutated by `domain_rand_episode_starts`, // so the destination must remain a CudaSlice. Stage via mapped-pinned // + DtoD per `feedback_no_htod_htoh_only_mapped_pinned.md`. - upload_host_to_cuda_i32_via_pinned( + upload_host_to_cuda_i32( &self.stream, episode_starts, &mut self.episode_starts_buf, "episode_starts upload")?; // ── Initialize current_timesteps to zeros ─────────────────────── @@ -4057,7 +4057,7 @@ impl GpuExperienceCollector { portfolio_init[off + 9] = initial_capital; // [9] prev_equity // [10] hold_time = 0.0, [11] realized_pnl = 0.0 (already zero) } - upload_host_to_cuda_f32_via_pinned( + upload_host_to_cuda_f32( &self.stream, &portfolio_init, &mut self.portfolio_states, "portfolio_states reset")?; debug!( @@ -4462,8 +4462,26 @@ impl GpuExperienceCollector { if ofi_flat.is_empty() { return Err(MLError::ModelError("OFI data empty — cannot upload empty features to GPU".into())); } - let gpu_buf = super::mapped_pinned::clone_to_device_f32_via_pinned(&self.stream, ofi_flat) - .map_err(|e| MLError::ModelError(format!("upload_ofi_features: {e}")))?; + let gpu_buf = { + let n = ofi_flat.len(); + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("upload_ofi_features staging alloc: {e}")))?; + staging.write_from_slice(ofi_flat); + let mut _dst = self.stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("upload_ofi_features alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&self.stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, self.stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("upload_ofi_features DtoD: {e}")))?; + } + self.stream.synchronize() + .map_err(|e| MLError::ModelError(format!("upload_ofi_features sync: {e}")))?; + } + _dst + }; info!( "OFI features uploaded to GPU: {} bars x {} dims ({:.1} KB)", ofi_flat.len() / self.ofi_dim, @@ -4803,7 +4821,7 @@ fn build_next_states_f32( /// `MappedF32Buffer` provides the only allowed CPU↔GPU path per /// `feedback_no_htod_htoh_only_mapped_pinned.md`; the DtoD copy is fully on- /// device and trivially compliant. -fn upload_host_to_cuda_f32_via_pinned( +fn upload_host_to_cuda_f32( stream: &Arc, host: &[f32], dst: &mut CudaSlice, @@ -4831,8 +4849,8 @@ fn upload_host_to_cuda_f32_via_pinned( Ok(()) } -/// `i32` counterpart to `upload_host_to_cuda_f32_via_pinned`. -fn upload_host_to_cuda_i32_via_pinned( +/// `i32` counterpart to `upload_host_to_cuda_f32`. +fn upload_host_to_cuda_i32( stream: &Arc, host: &[i32], dst: &mut CudaSlice, diff --git a/crates/ml/src/cuda_pipeline/gpu_her.rs b/crates/ml/src/cuda_pipeline/gpu_her.rs index 6877992db..d8e3546ff 100644 --- a/crates/ml/src/cuda_pipeline/gpu_her.rs +++ b/crates/ml/src/cuda_pipeline/gpu_her.rs @@ -29,7 +29,7 @@ use std::sync::Arc; -use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, DevicePtr, LaunchConfig, PushKernelArg}; +use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, DevicePtr, DevicePtrMut, LaunchConfig, PushKernelArg}; use tracing::info; use crate::MLError; @@ -197,8 +197,22 @@ impl GpuHer { .collect(); let mut rng_states = stream.alloc_zeros::(her_batch) .map_err(|e| MLError::ModelError(format!("GpuHer alloc rng_states: {e}")))?; - super::mapped_pinned::upload_u32_via_pinned(&stream, &rng_init, &mut rng_states) - .map_err(|e| MLError::ModelError(format!("GpuHer rng_states upload via pinned: {e}")))?; + { + let n = rng_init.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedU32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("GpuHer rng_states staging alloc: {e}")))?; + staging.write_from_slice(&rng_init); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = rng_states.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("GpuHer rng_states DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("GpuHer rng_states sync: {e}")))?; + } + } let vram_bytes = (her_batch * ml_core::state_layout::STATE_DIM * 2 + her_batch * 3) * 4 + her_batch * 3 * 4; // output bufs + index bufs + rng_states @@ -276,18 +290,38 @@ impl GpuHer { // Upload source and donor indices to GPU via mapped-pinned staging // (per-relabel call; pinned staging is allocated+freed each call — // acceptable on this cold path which runs once per epoch). - super::mapped_pinned::upload_i32_via_pinned( - &self.stream, - &source_idx_host[..her_batch_size], - &mut self.source_indices, - ) - .map_err(|e| MLError::ModelError(format!("HER source_indices upload via pinned: {e}")))?; - super::mapped_pinned::upload_i32_via_pinned( - &self.stream, - &donor_idx_host[..her_batch_size], - &mut self.donor_indices, - ) - .map_err(|e| MLError::ModelError(format!("HER donor_indices upload via pinned: {e}")))?; + { + let n = her_batch_size; + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedI32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("HER source_indices staging alloc: {e}")))?; + staging.write_from_slice(&source_idx_host[..n]); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = self.source_indices.device_ptr_mut(&self.stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, self.stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("HER source_indices DtoD: {e}")))?; + } + self.stream.synchronize().map_err(|e| MLError::ModelError(format!("HER source_indices sync: {e}")))?; + } + } + { + let n = her_batch_size; + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedI32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("HER donor_indices staging alloc: {e}")))?; + staging.write_from_slice(&donor_idx_host[..n]); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = self.donor_indices.device_ptr_mut(&self.stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, self.stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("HER donor_indices DtoD: {e}")))?; + } + self.stream.synchronize().map_err(|e| MLError::ModelError(format!("HER donor_indices sync: {e}")))?; + } + } // Launch relabel kernel: one warp (32 threads) per sample let launch_config = LaunchConfig { diff --git a/crates/ml/src/cuda_pipeline/gpu_iql_trainer.rs b/crates/ml/src/cuda_pipeline/gpu_iql_trainer.rs index 712a8be8a..a7b14aa9d 100644 --- a/crates/ml/src/cuda_pipeline/gpu_iql_trainer.rs +++ b/crates/ml/src/cuda_pipeline/gpu_iql_trainer.rs @@ -32,7 +32,7 @@ use std::sync::Arc; use cudarc::cublaslt::result as cublaslt_result; use cudarc::cublaslt::sys as cublaslt_sys; -use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, LaunchConfig, PushKernelArg}; +use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, DevicePtrMut, LaunchConfig, PushKernelArg}; use tracing::info; use crate::MLError; @@ -425,8 +425,22 @@ impl GpuIqlTrainer { default_support[base + 2] = default_delta_z; } } - super::mapped_pinned::upload_f32_via_pinned(&stream, &default_support, &mut per_sample_support_buf) - .map_err(|e| MLError::ModelError(format!("IQL per_sample_support upload via pinned: {e}")))?; + { + let n = default_support.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("IQL per_sample_support staging alloc: {e}")))?; + staging.write_from_slice(&default_support); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = per_sample_support_buf.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("IQL per_sample_support DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("IQL per_sample_support sync: {e}")))?; + } + } let cublas_fwd_vram = h * b * 4 * 4 + b * 2 * 4; // h1_pre+h1+h2_pre+h2 + dv[B*2] let cublas_bwd_vram = h * b * 4 * 4; // dh2+dh2_pre+dh1+dh1_pre @@ -1633,8 +1647,22 @@ fn init_xavier_weights( // Upload to GPU via mapped-pinned staging (no explicit HtoD) let mut params_buf = alloc_f32(stream, total, "iql_params")?; - super::mapped_pinned::upload_f32_via_pinned(stream, &weights, &mut params_buf) - .map_err(|e| MLError::ModelError(format!("iql params upload via pinned: {e}")))?; + { + let n = weights.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("iql params staging alloc: {e}")))?; + staging.write_from_slice(&weights); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = params_buf.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("iql params DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("iql params sync: {e}")))?; + } + } Ok(params_buf) } diff --git a/crates/ml/src/cuda_pipeline/gpu_iqn_head.rs b/crates/ml/src/cuda_pipeline/gpu_iqn_head.rs index e8f311ecd..f98abff7e 100644 --- a/crates/ml/src/cuda_pipeline/gpu_iqn_head.rs +++ b/crates/ml/src/cuda_pipeline/gpu_iqn_head.rs @@ -30,7 +30,7 @@ use std::sync::Arc; use cudarc::cublaslt::result as cublaslt_result; use cudarc::cublaslt::sys as cublaslt_sys; -use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, LaunchConfig, PushKernelArg}; +use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, DevicePtrMut, LaunchConfig, PushKernelArg}; use tracing::info; use crate::MLError; @@ -536,17 +536,68 @@ impl GpuIqnHead { (std::f32::consts::PI * ((dim + 1) as f32) * tau_q).cos(); } } - cos_features = super::mapped_pinned::clone_to_device_f32_via_pinned(&stream, &cos_feat_host) - .map_err(|e| MLError::ModelError(format!("IQN cos_features upload via pinned ({} f32): {e}", d * n)))?; + cos_features = { + let sz = cos_feat_host.len(); + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(sz) } + .map_err(|e| MLError::ModelError(format!("IQN cos_features staging alloc ({} f32): {e}", sz)))?; + staging.write_from_slice(&cos_feat_host); + let mut _dst = stream.alloc_zeros::(sz) + .map_err(|e| MLError::ModelError(format!("IQN cos_features alloc ({} f32): {e}", sz)))?; + if sz > 0 { + let nbytes = sz * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("IQN cos_features DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("IQN cos_features sync: {e}")))?; + } + _dst + }; let mut tiled = Vec::with_capacity(b * n); for _ in 0..b { tiled.extend_from_slice(fixed); } - online_taus = super::mapped_pinned::clone_to_device_f32_via_pinned(&stream, &tiled) - .map_err(|e| MLError::ModelError(format!("IQN online_taus upload via pinned ({} f32): {e}", b * n)))?; - target_taus = super::mapped_pinned::clone_to_device_f32_via_pinned(&stream, &tiled) - .map_err(|e| MLError::ModelError(format!("IQN target_taus upload via pinned ({} f32): {e}", b * n)))?; + online_taus = { + let sz = tiled.len(); + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(sz) } + .map_err(|e| MLError::ModelError(format!("IQN online_taus staging alloc ({} f32): {e}", sz)))?; + staging.write_from_slice(&tiled); + let mut _dst = stream.alloc_zeros::(sz) + .map_err(|e| MLError::ModelError(format!("IQN online_taus alloc ({} f32): {e}", sz)))?; + if sz > 0 { + let nbytes = sz * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("IQN online_taus DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("IQN online_taus sync: {e}")))?; + } + _dst + }; + target_taus = { + let sz = tiled.len(); + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(sz) } + .map_err(|e| MLError::ModelError(format!("IQN target_taus staging alloc ({} f32): {e}", sz)))?; + staging.write_from_slice(&tiled); + let mut _dst = stream.alloc_zeros::(sz) + .map_err(|e| MLError::ModelError(format!("IQN target_taus alloc ({} f32): {e}", sz)))?; + if sz > 0 { + let nbytes = sz * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("IQN target_taus DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("IQN target_taus sync: {e}")))?; + } + _dst + }; } let branch_actions = stream.alloc_zeros::(b * 4).map_err(|e| { MLError::ModelError(format!("IQN alloc branch_actions: {e}")) @@ -2339,8 +2390,25 @@ fn init_iqn_xavier_weights( debug_assert_eq!(offset, total, "IQN weight layout mismatch"); // Upload to GPU via mapped-pinned staging (zeros past total_params remain in `weights` Vec) - let params_buf = super::mapped_pinned::clone_to_device_f32_via_pinned(stream, &weights) - .map_err(|e| MLError::ModelError(format!("IQN online_params upload via pinned ({} f32): {e}", total + cublas_pad)))?; + let params_buf = { + let n = weights.len(); + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("IQN online_params staging alloc ({} f32): {e}", n)))?; + staging.write_from_slice(&weights); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("IQN online_params alloc ({} f32): {e}", n)))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("IQN online_params DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("IQN online_params sync: {e}")))?; + } + _dst + }; Ok(params_buf) } diff --git a/crates/ml/src/cuda_pipeline/gpu_ppo_collector.rs b/crates/ml/src/cuda_pipeline/gpu_ppo_collector.rs index d441defc5..ee49ceecb 100644 --- a/crates/ml/src/cuda_pipeline/gpu_ppo_collector.rs +++ b/crates/ml/src/cuda_pipeline/gpu_ppo_collector.rs @@ -350,8 +350,22 @@ impl GpuPpoExperienceCollector { // portfolio_init[off + 7] = 0.0; // cum_costs } let mut portfolio_states = portfolio_states; - super::mapped_pinned::upload_f32_via_pinned(&stream, &portfolio_init, &mut portfolio_states) - .map_err(|e| MLError::ModelError(format!("PPO portfolio_states init upload via pinned: {e}")))?; + { + let n = portfolio_init.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("PPO portfolio_states staging alloc: {e}")))?; + staging.write_from_slice(&portfolio_init); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = portfolio_states.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("PPO portfolio_states DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("PPO portfolio_states sync: {e}")))?; + } + } // ---- Step 5: Deterministic RNG seeds ---- let rng_seeds: Vec = (0..MAX_EPISODES) @@ -362,8 +376,22 @@ impl GpuPpoExperienceCollector { }) .collect(); let mut rng_states = rng_states; - super::mapped_pinned::upload_u32_via_pinned(&stream, &rng_seeds, &mut rng_states) - .map_err(|e| MLError::ModelError(format!("PPO rng_states init upload via pinned: {e}")))?; + { + let n = rng_seeds.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedU32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("PPO rng_states staging alloc: {e}")))?; + staging.write_from_slice(&rng_seeds); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = rng_states.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("PPO rng_states DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("PPO rng_states sync: {e}")))?; + } + } // ---- Step 6: Allocate output buffers ---- let total_output = MAX_EPISODES * MAX_TIMESTEPS; @@ -518,12 +546,22 @@ impl GpuPpoExperienceCollector { } // ---- Step 2: Upload episode starts via mapped-pinned staging ---- - super::mapped_pinned::upload_i32_via_pinned( - &self.stream, - episode_starts, - &mut self.episode_starts_buf, - ) - .map_err(|e| MLError::ModelError(format!("PPO episode_starts upload via pinned: {e}")))?; + { + let n = episode_starts.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedI32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("PPO episode_starts staging alloc: {e}")))?; + staging.write_from_slice(episode_starts); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = self.episode_starts_buf.device_ptr_mut(&self.stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, self.stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("PPO episode_starts DtoD: {e}")))?; + } + self.stream.synchronize().map_err(|e| MLError::ModelError(format!("PPO episode_starts sync: {e}")))?; + } + } // ---- Step 3: Upload barrier config via mapped-pinned staging ---- let barrier_cfg = [ @@ -531,8 +569,22 @@ impl GpuPpoExperienceCollector { config.barrier_loss_mult, config.barrier_max_bars, ]; - super::mapped_pinned::upload_f32_via_pinned(&self.stream, &barrier_cfg, &mut self.barrier_config) - .map_err(|e| MLError::ModelError(format!("PPO barrier_config upload via pinned: {e}")))?; + { + let n = barrier_cfg.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("PPO barrier_config staging alloc: {e}")))?; + staging.write_from_slice(&barrier_cfg); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = self.barrier_config.device_ptr_mut(&self.stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, self.stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("PPO barrier_config DtoD: {e}")))?; + } + self.stream.synchronize().map_err(|e| MLError::ModelError(format!("PPO barrier_config sync: {e}")))?; + } + } // ---- Step 4: Launch config ---- let n = n_episodes as u32; diff --git a/crates/ml/src/cuda_pipeline/gpu_tlob.rs b/crates/ml/src/cuda_pipeline/gpu_tlob.rs index 3ff6164c6..73a4e639d 100644 --- a/crates/ml/src/cuda_pipeline/gpu_tlob.rs +++ b/crates/ml/src/cuda_pipeline/gpu_tlob.rs @@ -45,7 +45,7 @@ use std::sync::Arc; use cudarc::cublas::sys as cublas_sys; -use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, DevicePtr, LaunchConfig, PushKernelArg}; +use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, DevicePtr, DevicePtrMut, LaunchConfig, PushKernelArg}; use tracing::info; use crate::MLError; @@ -232,8 +232,22 @@ impl GpuTlob { let mut params = stream.alloc_zeros::(TLOB_TOTAL_PARAMS) .map_err(|e| MLError::ModelError(format!("tlob params alloc: {e}")))?; - super::mapped_pinned::upload_f32_via_pinned(&stream, &host_params, &mut params) - .map_err(|e| MLError::ModelError(format!("tlob params upload via pinned: {e}")))?; + { + let n = host_params.len(); + if n > 0 { + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("tlob params staging alloc: {e}")))?; + staging.write_from_slice(&host_params); + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = params.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("tlob params DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("tlob params sync: {e}")))?; + } + } // ── Buffer allocation ───────────────────────────────────────── let db_out = TLOB_OUT * b; // 16 × B @@ -1050,8 +1064,21 @@ mod tests { let mut states_buf = stream .alloc_zeros::(batch_size * STATE_DIM_PADDED) .expect("states_buf alloc"); - super::super::mapped_pinned::upload_f32_via_pinned(&stream, &host_states, &mut states_buf) - .expect("upload states via pinned"); + { + let n = host_states.len(); + if n > 0 { + let staging = unsafe { super::super::mapped_pinned::MappedF32Buffer::new(n) } + .expect("states_buf staging alloc"); + staging.write_from_slice(&host_states); + let nbytes = n * std::mem::size_of::(); + unsafe { + let (dst_ptr, _g) = states_buf.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .expect("states_buf DtoD"); + } + stream.synchronize().expect("states_buf sync"); + } + } // Snapshot weights for the CPU reference (Xavier init, randomly seeded). let host_params = tlob.dump_params(); @@ -1166,8 +1193,21 @@ mod tests { let mut d_concat = stream .alloc_zeros::(batch_size * concat_dim) .expect("d_concat alloc"); - super::super::mapped_pinned::upload_f32_via_pinned(&stream, &host_d_concat, &mut d_concat) - .expect("upload d_concat via pinned"); + { + let n = host_d_concat.len(); + if n > 0 { + let staging = unsafe { super::super::mapped_pinned::MappedF32Buffer::new(n) } + .expect("d_concat staging alloc"); + staging.write_from_slice(&host_d_concat); + let nbytes = n * std::mem::size_of::(); + unsafe { + let (dst_ptr, _g) = d_concat.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .expect("d_concat DtoD"); + } + stream.synchronize().expect("d_concat sync"); + } + } tlob .backward(&d_concat, batch_size, concat_dim, tlob_concat_off) diff --git a/crates/ml/src/cuda_pipeline/gpu_walk_forward.rs b/crates/ml/src/cuda_pipeline/gpu_walk_forward.rs index e25685e86..9ff5262d8 100644 --- a/crates/ml/src/cuda_pipeline/gpu_walk_forward.rs +++ b/crates/ml/src/cuda_pipeline/gpu_walk_forward.rs @@ -28,7 +28,7 @@ //! `total_bars`. use std::sync::Arc; -use cudarc::driver::{CudaSlice, CudaStream, LaunchConfig, PushKernelArg}; +use cudarc::driver::{CudaSlice, CudaStream, DevicePtrMut, LaunchConfig, PushKernelArg}; use tracing::info; use crate::features::extraction::FeatureVector; @@ -579,14 +579,48 @@ impl GpuWalkForwardData { } // Upload features via mapped-pinned staging (no explicit HtoD) - let features_gpu = super::mapped_pinned::clone_to_device_f32_via_pinned(stream, &flat_features) - .map_err(|e| MLError::ModelError(format!("walk-forward features upload via pinned: {e}")))?; + let features_gpu = { + let n = flat_features.len(); + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("walk-forward features staging alloc: {e}")))?; + staging.write_from_slice(&flat_features); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("walk-forward features alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("walk-forward features DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("walk-forward features sync: {e}")))?; + } + _dst + }; let mut vram = total_bars * feature_dim * 4; // Upload targets via mapped-pinned staging - let targets_gpu = super::mapped_pinned::clone_to_device_f32_via_pinned(stream, &flat_targets) - .map_err(|e| MLError::ModelError(format!("walk-forward targets upload via pinned: {e}")))?; + let targets_gpu = { + let n = flat_targets.len(); + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("walk-forward targets staging alloc: {e}")))?; + staging.write_from_slice(&flat_targets); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("walk-forward targets alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("walk-forward targets DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("walk-forward targets sync: {e}")))?; + } + _dst + }; vram += total_bars * target_dim * 4; @@ -605,8 +639,25 @@ impl GpuWalkForwardData { flat_ofi.extend_from_slice(&zero_row); } } - let buf = super::mapped_pinned::clone_to_device_f32_via_pinned(stream, &flat_ofi) - .map_err(|e| MLError::ModelError(format!("walk-forward OFI upload via pinned: {e}")))?; + let buf = { + let sz = flat_ofi.len(); + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(sz) } + .map_err(|e| MLError::ModelError(format!("walk-forward OFI staging alloc: {e}")))?; + staging.write_from_slice(&flat_ofi); + let mut _dst = stream.alloc_zeros::(sz) + .map_err(|e| MLError::ModelError(format!("walk-forward OFI alloc: {e}")))?; + if sz > 0 { + let nbytes = sz * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("walk-forward OFI DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("walk-forward OFI sync: {e}")))?; + } + _dst + }; vram += total_bars * ofi_dim * 4; Some(buf) } else { diff --git a/crates/ml/src/cuda_pipeline/gpu_weights.rs b/crates/ml/src/cuda_pipeline/gpu_weights.rs index 0688ccb29..93ae453d8 100644 --- a/crates/ml/src/cuda_pipeline/gpu_weights.rs +++ b/crates/ml/src/cuda_pipeline/gpu_weights.rs @@ -546,8 +546,24 @@ impl RmsNormWeightSet { let data = vec![1.0_f32; size]; // Mapped-pinned staging + DtoD; no explicit HtoD per // feedback_no_htod_htoh_only_mapped_pinned.md. - super::mapped_pinned::clone_to_device_f32_via_pinned(stream, &data) - .map_err(|e| MLError::ModelError(format!("Alloc RMSNorm ones via pinned: {e}"))) + let n = data.len(); + let staging = unsafe { super::mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("RMSNorm ones staging alloc: {e}")))?; + staging.write_from_slice(&data); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("RMSNorm ones alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + use cudarc::driver::DevicePtrMut; + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("RMSNorm ones DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("RMSNorm ones sync: {e}")))?; + } + Ok(_dst) }; Ok(Self { gamma_s0: alloc_ones(shared_h1)?, diff --git a/crates/ml/src/cuda_pipeline/mapped_pinned.rs b/crates/ml/src/cuda_pipeline/mapped_pinned.rs index 6248eca7b..adb4e9012 100644 --- a/crates/ml/src/cuda_pipeline/mapped_pinned.rs +++ b/crates/ml/src/cuda_pipeline/mapped_pinned.rs @@ -299,209 +299,6 @@ impl Drop for MappedF32Buffer { } } -// ── Pinned-staging helpers ──────────────────────────────────────────────────── -// -// Some legacy code paths still need a `CudaSlice` field (e.g. buffers that -// the kernel mutates every step — keeping them device-resident avoids per-step -// PCIe traffic). For their **initial** load we still need a CPU-supplied -// payload to land on the device without an explicit HtoD cudaMemcpy. -// -// The pattern below allocates a temporary `MappedXxxBuffer` (cuMemHostAlloc -// DEVICEMAP), writes the payload via the host pointer, then DtoD-copies into -// a freshly-allocated `CudaSlice`. The staging buffer is dropped on return, -// freeing the pinned host pages. No explicit HtoD memcpy is performed — the -// only CPU↔GPU transfer is the implicit unified-memory read the GPU performs -// during the DtoD. - -use cudarc::driver::{CudaSlice, CudaStream}; -use std::sync::Arc; - -/// Allocate a fresh `CudaSlice` and seed it with `host` via a mapped-pinned -/// staging buffer + async DtoD copy. Replaces explicit `clone_htod` / -/// `memcpy_htod` per `feedback_no_htod_htoh_only_mapped_pinned.md`. -/// -/// # Safety -/// Caller must hold an active CUDA context on the current thread. -pub fn clone_to_device_f32_via_pinned( - stream: &Arc, - host: &[f32], -) -> Result, String> { - let n = host.len(); - let staging = unsafe { MappedF32Buffer::new(n) }?; - staging.write_from_slice(host); - - let mut dst = stream - .alloc_zeros::(n) - .map_err(|e| format!("clone_to_device_f32_via_pinned alloc: {e}"))?; - if n > 0 { - let nbytes = n * std::mem::size_of::(); - // Safety: `staging.dev_ptr` is valid for `n` f32s; `dst` has the same length. - unsafe { - let (dst_ptr, _g) = { - use cudarc::driver::DevicePtrMut; - dst.device_ptr_mut(stream) - }; - cudarc::driver::result::memcpy_dtod_async( - dst_ptr, - staging.dev_ptr, - nbytes, - stream.cu_stream(), - ) - .map_err(|e| format!("clone_to_device_f32_via_pinned DtoD: {e}"))?; - } - // Synchronise so the staging buffer can be dropped safely below. - stream - .synchronize() - .map_err(|e| format!("clone_to_device_f32_via_pinned sync: {e}"))?; - } - Ok(dst) -} - -/// Seed an existing `CudaSlice` with `host` via mapped-pinned staging. -/// Replaces `htod_f32` / `memcpy_htod`. -/// -/// # Safety -/// Caller must hold an active CUDA context. -pub fn upload_f32_via_pinned( - stream: &Arc, - host: &[f32], - dst: &mut CudaSlice, -) -> Result<(), String> { - let n = host.len(); - if n == 0 { - return Ok(()); - } - let staging = unsafe { MappedF32Buffer::new(n) }?; - staging.write_from_slice(host); - - let nbytes = n * std::mem::size_of::(); - unsafe { - let (dst_ptr, _g) = { - use cudarc::driver::DevicePtrMut; - dst.device_ptr_mut(stream) - }; - cudarc::driver::result::memcpy_dtod_async( - dst_ptr, - staging.dev_ptr, - nbytes, - stream.cu_stream(), - ) - .map_err(|e| format!("upload_f32_via_pinned DtoD: {e}"))?; - } - stream - .synchronize() - .map_err(|e| format!("upload_f32_via_pinned sync: {e}"))?; - Ok(()) -} - -/// Allocate a fresh `CudaSlice` and seed it with `host` via mapped-pinned staging. -/// -/// # Safety -/// Caller must hold an active CUDA context. -pub fn clone_to_device_i32_via_pinned( - stream: &Arc, - host: &[i32], -) -> Result, String> { - let n = host.len(); - let staging = unsafe { MappedI32Buffer::new(n) }?; - staging.write_from_slice(host); - - let mut dst = stream - .alloc_zeros::(n) - .map_err(|e| format!("clone_to_device_i32_via_pinned alloc: {e}"))?; - if n > 0 { - let nbytes = n * std::mem::size_of::(); - unsafe { - let (dst_ptr, _g) = { - use cudarc::driver::DevicePtrMut; - dst.device_ptr_mut(stream) - }; - cudarc::driver::result::memcpy_dtod_async( - dst_ptr, - staging.dev_ptr, - nbytes, - stream.cu_stream(), - ) - .map_err(|e| format!("clone_to_device_i32_via_pinned DtoD: {e}"))?; - } - stream - .synchronize() - .map_err(|e| format!("clone_to_device_i32_via_pinned sync: {e}"))?; - } - Ok(dst) -} - -/// Seed an existing `CudaSlice` with `host` via mapped-pinned staging. -/// -/// # Safety -/// Caller must hold an active CUDA context. -pub fn upload_i32_via_pinned( - stream: &Arc, - host: &[i32], - dst: &mut CudaSlice, -) -> Result<(), String> { - let n = host.len(); - if n == 0 { - return Ok(()); - } - let staging = unsafe { MappedI32Buffer::new(n) }?; - staging.write_from_slice(host); - - let nbytes = n * std::mem::size_of::(); - unsafe { - let (dst_ptr, _g) = { - use cudarc::driver::DevicePtrMut; - dst.device_ptr_mut(stream) - }; - cudarc::driver::result::memcpy_dtod_async( - dst_ptr, - staging.dev_ptr, - nbytes, - stream.cu_stream(), - ) - .map_err(|e| format!("upload_i32_via_pinned DtoD: {e}"))?; - } - stream - .synchronize() - .map_err(|e| format!("upload_i32_via_pinned sync: {e}"))?; - Ok(()) -} - -/// Seed an existing `CudaSlice` with `host` via mapped-pinned staging. -/// -/// # Safety -/// Caller must hold an active CUDA context. -pub fn upload_u32_via_pinned( - stream: &Arc, - host: &[u32], - dst: &mut CudaSlice, -) -> Result<(), String> { - let n = host.len(); - if n == 0 { - return Ok(()); - } - let staging = unsafe { MappedU32Buffer::new(n) }?; - staging.write_from_slice(host); - - let nbytes = n * std::mem::size_of::(); - unsafe { - let (dst_ptr, _g) = { - use cudarc::driver::DevicePtrMut; - dst.device_ptr_mut(stream) - }; - cudarc::driver::result::memcpy_dtod_async( - dst_ptr, - staging.dev_ptr, - nbytes, - stream.cu_stream(), - ) - .map_err(|e| format!("upload_u32_via_pinned DtoD: {e}"))?; - } - stream - .synchronize() - .map_err(|e| format!("upload_u32_via_pinned sync: {e}"))?; - Ok(()) -} // ── MappedU64Buffer ─────────────────────────────────────────────────────────── diff --git a/crates/ml/src/cuda_pipeline/mod.rs b/crates/ml/src/cuda_pipeline/mod.rs index 99d4ec153..2d2621685 100644 --- a/crates/ml/src/cuda_pipeline/mod.rs +++ b/crates/ml/src/cuda_pipeline/mod.rs @@ -168,12 +168,9 @@ pub fn mix_seed(base: u64) -> u64 { // `htod_f32` and `clone_htod_f32` REMOVED 2026-04-28 (Fix 17 of the HtoD // migration sequence) per `feedback_no_htod_htoh_only_mapped_pinned.md`. // Mapped pinned (`cuMemHostAlloc DEVICEMAP`) + DtoD-async is now the only -// allowed CPU↔GPU path. Production callers were migrated through Fix 1..16; -// the only remaining `htod_f32` references live inside `#[cfg(test)]` -// modules (`gpu_tlob.rs::tests` lines 1017 and 1132), which are excluded by -// scope and were skipped on purpose. See -// `mapped_pinned::clone_to_device_f32_via_pinned` and -// `mapped_pinned::upload_f32_via_pinned` for the canonical replacements. +// allowed CPU↔GPU path. Production callers use `MappedF32Buffer::new` / +// `write_from_slice` + DtoD copy directly — the staging helpers were +// removed in the SP4-cleanup commit. /// Download `CudaSlice` (or `CudaView`) to an f32 host buffer. /// @@ -369,10 +366,44 @@ impl DqnGpuData { .flat_map(|t| t.iter().map(|&v| v as f32)) .collect(); - let features_gpu = mapped_pinned::clone_to_device_f32_via_pinned(stream, &flat_features) - .map_err(|e| MLError::ModelError(format!("features upload via pinned: {e}")))?; - let targets_gpu = mapped_pinned::clone_to_device_f32_via_pinned(stream, &flat_targets) - .map_err(|e| MLError::ModelError(format!("targets upload via pinned: {e}")))?; + let features_gpu = { + let n = flat_features.len(); + let staging = unsafe { mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("features upload staging alloc: {e}")))?; + staging.write_from_slice(&flat_features); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("features upload alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("features upload DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("features upload sync: {e}")))?; + } + _dst + }; + let targets_gpu = { + let n = flat_targets.len(); + let staging = unsafe { mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("targets upload staging alloc: {e}")))?; + staging.write_from_slice(&flat_targets); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("targets upload alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("targets upload DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("targets upload sync: {e}")))?; + } + _dst + }; if ofi.is_empty() { return Err(MLError::ModelError("OFI data empty — rebuild fxcache with --mbp10-data-dir".into())); @@ -389,10 +420,25 @@ impl DqnGpuData { flat_ofi.push(v as f32); } } - let ofi_features = Some( - mapped_pinned::clone_to_device_f32_via_pinned(stream, &flat_ofi) - .map_err(|e| MLError::ModelError(format!("ofi upload via pinned: {e}")))?, - ); + let ofi_features = Some({ + let n = flat_ofi.len(); + let staging = unsafe { mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("ofi upload staging alloc: {e}")))?; + staging.write_from_slice(&flat_ofi); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("ofi upload alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("ofi upload DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("ofi upload sync: {e}")))?; + } + _dst + }); Ok(Self { features: features_gpu, @@ -428,8 +474,25 @@ impl DqnGpuData { } } - let ofi_buf = mapped_pinned::clone_to_device_f32_via_pinned(stream, &flat) - .map_err(|e| MLError::ModelError(format!("ofi upload via pinned: {e}")))?; + let ofi_buf = { + let n = flat.len(); + let staging = unsafe { mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("ofi upload staging alloc: {e}")))?; + staging.write_from_slice(&flat); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("ofi upload alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("ofi upload DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("ofi upload sync: {e}")))?; + } + _dst + }; self.ofi_features = Some(ofi_buf); Ok(()) @@ -569,8 +632,25 @@ impl DqnGpuData { .map_err(|e| MLError::ModelError(format!("build_batch alloc: {e}")))?; // Upload portfolio features once (3 scalars) via mapped-pinned staging - let port_gpu = mapped_pinned::clone_to_device_f32_via_pinned(stream, portfolio_features) - .map_err(|e| MLError::ModelError(format!("portfolio upload via pinned: {e}")))?; + let port_gpu = { + let n = portfolio_features.len(); + let staging = unsafe { mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("portfolio upload staging alloc: {e}")))?; + staging.write_from_slice(portfolio_features); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("portfolio upload alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("portfolio upload DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("portfolio upload sync: {e}")))?; + } + _dst + }; // Get contiguous market features [count * feature_dim] let market_gpu = self.batch_features(start, count, stream)?; @@ -659,8 +739,25 @@ impl DqnGpuData { dst_offset_elems: usize, stream: &Arc, ) -> Result<(), MLError> { - let tmp = mapped_pinned::clone_to_device_f32_via_pinned(stream, src) - .map_err(|e| MLError::ModelError(format!("htod_copy_into via pinned: {e}")))?; + let tmp = { + let n = src.len(); + let staging = unsafe { mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("htod_copy_into staging alloc: {e}")))?; + staging.write_from_slice(src); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("htod_copy_into alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("htod_copy_into DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("htod_copy_into sync: {e}")))?; + } + _dst + }; Self::dtod_copy_into(&tmp, dst, dst_offset_elems, src.len(), stream) } } @@ -717,8 +814,25 @@ impl PpoGpuData { flat_states.extend_from_slice(state); } - let states = mapped_pinned::clone_to_device_f32_via_pinned(stream, &flat_states) - .map_err(|e| MLError::ModelError(format!("PpoGpuData states upload via pinned: {e}")))?; + let states = { + let n = flat_states.len(); + let staging = unsafe { mapped_pinned::MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("PpoGpuData states staging alloc: {e}")))?; + staging.write_from_slice(&flat_states); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("PpoGpuData states alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("PpoGpuData states DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("PpoGpuData states sync: {e}")))?; + } + _dst + }; Ok(Self { states, diff --git a/crates/ml/src/hyperopt/adapters/mamba2.rs b/crates/ml/src/hyperopt/adapters/mamba2.rs index 25e9c8c57..0562aa22a 100644 --- a/crates/ml/src/hyperopt/adapters/mamba2.rs +++ b/crates/ml/src/hyperopt/adapters/mamba2.rs @@ -745,8 +745,27 @@ fn gpu_to_stream( stream: &Arc, ) -> Result { let host = t.to_host(stream)?; - let data = crate::cuda_pipeline::mapped_pinned::clone_to_device_f32_via_pinned(stream, &host) - .map_err(|e| MLError::ModelError(format!("gpu_to_stream upload via pinned: {e}")))?; + let data = { + use cudarc::driver::DevicePtrMut; + use crate::cuda_pipeline::mapped_pinned::MappedF32Buffer; + let n = host.len(); + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("gpu_to_stream staging alloc: {e}")))?; + staging.write_from_slice(&host); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("gpu_to_stream alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("gpu_to_stream DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("gpu_to_stream sync: {e}")))?; + } + _dst + }; Ok(StreamTensor { data, shape: t.shape().to_vec(), // cpu-side shape clone diff --git a/crates/ml/src/hyperopt/adapters/ppo.rs b/crates/ml/src/hyperopt/adapters/ppo.rs index b68b2d633..a0f6475e2 100644 --- a/crates/ml/src/hyperopt/adapters/ppo.rs +++ b/crates/ml/src/hyperopt/adapters/ppo.rs @@ -580,14 +580,37 @@ impl PPOTrainer { for (features, _) in training_data { flat_features.extend_from_slice(features); } - match crate::cuda_pipeline::mapped_pinned::clone_to_device_f32_via_pinned(&stream, &flat_features) { - Ok(buf) => { - info!("PPO CUDA features uploaded: {} bars × 42 ({:.1} MB)", - num_bars, (num_bars * 42 * 4) as f64 / 1_048_576.0); - self.features_cuda = Some(buf); - } - Err(e) => { - return Err(MLError::TrainingError(format!("PPO CUDA features upload via pinned FAILED (no CPU fallback): {e}"))); + { + use cudarc::driver::DevicePtrMut; + use crate::cuda_pipeline::mapped_pinned::MappedF32Buffer; + let n = flat_features.len(); + let result: Result, MLError> = (|| { + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| MLError::TrainingError(format!("PPO features staging alloc: {e}")))?; + staging.write_from_slice(&flat_features); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::TrainingError(format!("PPO features alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::TrainingError(format!("PPO features DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::TrainingError(format!("PPO features sync: {e}")))?; + } + Ok(_dst) + })(); + match result { + Ok(buf) => { + info!("PPO CUDA features uploaded: {} bars × 42 ({:.1} MB)", + num_bars, (num_bars * 42 * 4) as f64 / 1_048_576.0); + self.features_cuda = Some(buf); + } + Err(e) => { + return Err(MLError::TrainingError(format!("PPO CUDA features upload FAILED (no CPU fallback): {e}"))); + } } } @@ -604,15 +627,38 @@ impl PPOTrainer { flat_targets.push(close); flat_targets.push(next_close); } - match crate::cuda_pipeline::mapped_pinned::clone_to_device_f32_via_pinned(&stream, &flat_targets) { - Ok(buf) => { - info!("PPO CUDA targets uploaded: {} bars × 4 ({:.1} MB)", - num_bars, (num_bars * 4 * 4) as f64 / 1_048_576.0); - self.targets_cuda = Some(buf); - } - Err(e) => { - self.features_cuda = None; - return Err(MLError::TrainingError(format!("PPO CUDA targets upload via pinned FAILED (no CPU fallback): {e}"))); + { + use cudarc::driver::DevicePtrMut; + use crate::cuda_pipeline::mapped_pinned::MappedF32Buffer; + let n = flat_targets.len(); + let result: Result, MLError> = (|| { + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| MLError::TrainingError(format!("PPO targets staging alloc: {e}")))?; + staging.write_from_slice(&flat_targets); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::TrainingError(format!("PPO targets alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::TrainingError(format!("PPO targets DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::TrainingError(format!("PPO targets sync: {e}")))?; + } + Ok(_dst) + })(); + match result { + Ok(buf) => { + info!("PPO CUDA targets uploaded: {} bars × 4 ({:.1} MB)", + num_bars, (num_bars * 4 * 4) as f64 / 1_048_576.0); + self.targets_cuda = Some(buf); + } + Err(e) => { + self.features_cuda = None; + return Err(MLError::TrainingError(format!("PPO CUDA targets upload FAILED (no CPU fallback): {e}"))); + } } } @@ -1327,8 +1373,27 @@ impl PPOTrainer { } // Upload action indices to GPU via mapped-pinned staging (no explicit HtoD) - crate::cuda_pipeline::mapped_pinned::clone_to_device_i32_via_pinned(stream, &action_indices) - .map_err(|e| MLError::ModelError(format!("PPO backtest action upload via pinned: {e}"))) + { + use cudarc::driver::DevicePtrMut; + use crate::cuda_pipeline::mapped_pinned::MappedI32Buffer; + let n = action_indices.len(); + let staging = unsafe { MappedI32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("PPO backtest action staging alloc: {e}")))?; + staging.write_from_slice(&action_indices); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("PPO backtest action alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("PPO backtest action DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("PPO backtest action sync: {e}")))?; + } + Ok(_dst) + } }, 24, // portfolio_dim: 8 portfolio + 16 multi-timeframe (matches training state layout) )?; diff --git a/crates/ml/src/trainers/dqn/trainer/training_loop.rs b/crates/ml/src/trainers/dqn/trainer/training_loop.rs index c8238f3d1..abc969a6c 100644 --- a/crates/ml/src/trainers/dqn/trainer/training_loop.rs +++ b/crates/ml/src/trainers/dqn/trainer/training_loop.rs @@ -17,11 +17,11 @@ use std::sync::Arc; use anyhow::{Context, Result}; -use cudarc::driver::CudaSlice; +use cudarc::driver::{CudaSlice, DevicePtrMut}; use common::metrics::{questdb_sink, training_metrics}; use tracing::{debug, info, warn}; -use crate::cuda_pipeline::mapped_pinned::MappedI32Buffer; +use crate::cuda_pipeline::mapped_pinned::{MappedF32Buffer, MappedI32Buffer}; use crate::cuda_pipeline::gpu_dqn_trainer::{ EPOCH_IDX_INDEX, EPSILON_EFF_INDEX, GAMMA_DIR_EFF_INDEX, Q_ABS_REF_INDEX, Q_DIR_ABS_REF_INDEX, TLOB_REGIME_FOCUS_EMA_INDEX, @@ -1293,14 +1293,44 @@ impl DQNTrainer { .flat_map(|f| f.iter().map(|&v| v as f32)) .collect(); - self.targets_raw_cuda = Some( - crate::cuda_pipeline::mapped_pinned::clone_to_device_f32_via_pinned(&stream, &flat_targets) - .map_err(|e| anyhow::anyhow!("targets_raw upload: {e}"))? - ); - self.features_raw_cuda = Some( - crate::cuda_pipeline::mapped_pinned::clone_to_device_f32_via_pinned(&stream, &flat_features) - .map_err(|e| anyhow::anyhow!("features_raw upload: {e}"))? - ); + self.targets_raw_cuda = Some({ + let n = flat_targets.len(); + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| anyhow::anyhow!("targets_raw staging alloc: {e}"))?; + staging.write_from_slice(&flat_targets); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| anyhow::anyhow!("targets_raw alloc: {e}"))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| anyhow::anyhow!("targets_raw DtoD: {e}"))?; + } + stream.synchronize().map_err(|e| anyhow::anyhow!("targets_raw sync: {e}"))?; + } + _dst + }); + self.features_raw_cuda = Some({ + let n = flat_features.len(); + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| anyhow::anyhow!("features_raw staging alloc: {e}"))?; + staging.write_from_slice(&flat_features); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| anyhow::anyhow!("features_raw alloc: {e}"))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| anyhow::anyhow!("features_raw DtoD: {e}"))?; + } + stream.synchronize().map_err(|e| anyhow::anyhow!("features_raw sync: {e}"))?; + } + _dst + }); tracing::info!("CUDA raw buffers uploaded from slices: {} bars x 42+6", num_bars); Ok(()) diff --git a/crates/ml/src/trainers/ppo.rs b/crates/ml/src/trainers/ppo.rs index d0f274048..456a7774e 100644 --- a/crates/ml/src/trainers/ppo.rs +++ b/crates/ml/src/trainers/ppo.rs @@ -395,8 +395,27 @@ impl PpoTrainer { flat_features.push(v as f32); } } - let features_buf = crate::cuda_pipeline::mapped_pinned::clone_to_device_f32_via_pinned(&stream, &flat_features) - .map_err(|e| MLError::ModelError(format!("PPO features upload via pinned: {e}")))?; + let features_buf = { + use cudarc::driver::DevicePtrMut; + use crate::cuda_pipeline::mapped_pinned::MappedF32Buffer; + let n = flat_features.len(); + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("PPO features staging alloc: {e}")))?; + staging.write_from_slice(&flat_features); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("PPO features alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("PPO features DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("PPO features sync: {e}")))?; + } + _dst + }; self.features_raw_cuda = Some(features_buf); // Upload targets [num_bars * 4] @@ -406,8 +425,27 @@ impl PpoTrainer { flat_targets.push(targets.get(i).copied().unwrap_or(0.0) as f32); } } - let targets_buf = crate::cuda_pipeline::mapped_pinned::clone_to_device_f32_via_pinned(&stream, &flat_targets) - .map_err(|e| MLError::ModelError(format!("PPO targets upload via pinned: {e}")))?; + let targets_buf = { + use cudarc::driver::DevicePtrMut; + use crate::cuda_pipeline::mapped_pinned::MappedF32Buffer; + let n = flat_targets.len(); + let staging = unsafe { MappedF32Buffer::new(n) } + .map_err(|e| MLError::ModelError(format!("PPO targets staging alloc: {e}")))?; + staging.write_from_slice(&flat_targets); + let mut _dst = stream.alloc_zeros::(n) + .map_err(|e| MLError::ModelError(format!("PPO targets alloc: {e}")))?; + if n > 0 { + let nbytes = n * std::mem::size_of::(); + #[allow(unsafe_code)] + unsafe { + let (dst_ptr, _g) = _dst.device_ptr_mut(&stream); + cudarc::driver::result::memcpy_dtod_async(dst_ptr, staging.dev_ptr, nbytes, stream.cu_stream()) + .map_err(|e| MLError::ModelError(format!("PPO targets DtoD: {e}")))?; + } + stream.synchronize().map_err(|e| MLError::ModelError(format!("PPO targets sync: {e}")))?; + } + _dst + }; self.targets_raw_cuda = Some(targets_buf); self.raw_data_num_bars = num_bars; diff --git a/docs/dqn-wire-up-audit.md b/docs/dqn-wire-up-audit.md index ac20def36..8c35b8c67 100644 --- a/docs/dqn-wire-up-audit.md +++ b/docs/dqn-wire-up-audit.md @@ -2,6 +2,8 @@ **Status:** Populated during Plan 1 Task 6 (A.5 orphan audit). Updated on every commit per Invariant 7. +SP4-cleanup — delete via_pinned helpers, migrate 22 callers to MappedF32/I32/U32Buffer (2026-05-03): removed 5 scaffolding helpers from `cuda_pipeline/mapped_pinned.rs` (`clone_to_device_f32_via_pinned`, `upload_f32_via_pinned`, `clone_to_device_i32_via_pinned`, `upload_i32_via_pinned`, `upload_u32_via_pinned`). All 22 callers across 17 files migrated atomically per `feedback_no_partial_refactor.md`. Each call site now inlines the canonical pattern: `MappedXxxBuffer::new` + `write_from_slice` + `alloc_zeros` + `memcpy_dtod_async` + `stream.synchronize`. Two local file-private helpers in `gpu_experience_collector.rs` that already used `MappedF32Buffer` directly were renamed to drop the `_via_pinned` suffix. `grep -rn 'via_pinned' crates/` returns 0 matches. No behaviour change — functionally equivalent staging + DtoD pattern preserved at each call site. DevicePtrMut added to imports in `gpu_her.rs`, `gpu_iql_trainer.rs`, `gpu_attention.rs`, `gpu_tlob.rs`, `gpu_walk_forward.rs`, `gpu_iqn_head.rs`, `gpu_dqn_trainer.rs`, `trainers/dqn/trainer/training_loop.rs`. cargo check workspace clean. cargo test -p ml --lib: 925 passed, 17 failed (all pre-existing GPU/data infrastructure failures). Touched: 17 files (+930/-363 LOC). + HEALTH_DIAG intent_dist diagnostic (#212, 2026-05-01): added `intent_dist_q/h/f` HEALTH_DIAG metric adjacent to the existing `eval_dist_q/h/f` so operators can split policy-learning quality from Kelly-enforcement reality. Per memory pearl `project_magnitude_eval_collapse_kelly_capped.md`: EVAL_DIST Quarter dominance is a downstream artefact of Kelly cap × warmup_floor × safety_multiplier math, NOT a policy-learning failure. The intent metric reads the policy's pre-Kelly-cap chosen mag bucket (`intent_mag_buf` written by the action-select kernel, exposed via the trainer's pre-existing `last_eval_intent_magnitude_dist` host field — no new compute, no new GPU paths). Pure observability — the Kelly cap math itself stays unchanged (it is already adaptive per recent commits 2c97e0436 / 0c9d1ee39 / 94157a8a6 / d9fee6ef8), no new tuned constants, no reward-bias mechanisms (all explicitly forbidden by the pearl). Three coordinated changes per `feedback_no_partial_refactor.md`: (1) `HealthDiagSnapshot` gains `intent_dist_q/h/f: f32` fields adjacent to `eval_dist_*`; size assertion bumped 147 → 150 fields. (2) `health_diag_kernel.cu` adds `WORD_INTENT_DIST_*` slots at [77..80); every downstream `WORD_*` shifted by +3 in lockstep; `WORD_TOTAL` + `static_assert` bumped 147 → 150. The slots are reserved identically to `WORD_EVAL_DIST_*` — both currently inert in the kernel; HEALTH_DIAG GPU port Phases 2/3 will populate them in lockstep with their sibling slots. (3) `training_loop.rs` adds an adjacent `tracing::info!()` emitting `"intent_dist [iq=… ih=… if=…]"` from `self.last_eval_intent_magnitude_dist` immediately after the big HEALTH_DIAG line containing `eval_dist [eq=… eh=… ef=…]`. Behavior: zero change. After this commit, HEALTH_DIAG emits BOTH metrics so downstream analysis can distinguish "policy is learning Full, Kelly cap is suppressing" (`intent_dist_f > 0.30` AND `eval_dist_f ≈ 0`, an operationally-correct cold-start state) from "policy genuinely isn't learning Full" (both metrics near zero, a policy-quality bug worth investigating). Touched: `cuda_pipeline/health_diag.rs`, `cuda_pipeline/health_diag_kernel.cu`, `trainers/dqn/trainer/training_loop.rs`. cargo check -p ml --lib clean (11 pre-existing warnings, none new); 3/3 health_diag layout tests pass; 16/16 sp4_producer_unit_tests GPU tests pass. No fingerprint change — separate mapped-pinned alloc, not part of param-tensor layout the fingerprint guards. SP3 Mech 10 — ISV-driven post-trunk h_s2 activation clamp + slot 49 sentinel (2026-04-30): real root cause of the F1 step-1000 NaN identified by smoke-test-2xrxk on commit `0d7e27d61` (SP3 close-out v2 active). Trajectory: F0 Best Sharpe 45.47 (clean) → F1 NaN at step ~1000 → F2 first epoch reached Best Sharpe **56.70** (above the 55.87 baseline) before NaN epoch 2. Slot fire pattern at F1 NaN: `[6=grad_buf, 12=save_h_s2, 26=iqn_trunk_m, 32=bn_d_concat_buf, 36-38=trunk/value/branch_adam_m, 40-42=trunk/value/branch_adam_v]`. Crucially: slots 48 (iqn_weight_max), 39+43 (iqn_adam_m/v), 44-45 (trunk/heads weights) all stayed clean — disproving the SP3 close-out v2 hypothesis (IQN partial refactor). The actual cause: forward-pass GEMM accumulator overflow under Mech-9-clamped weights. With Mech 9 pinning weights at `100 × Q_ABS_REF = 5000` and trunk inner-dim K = 256, forward GEMM accumulators compound by `√K × |w|max ≈ 80000×` per layer; trunk depth ≥6 (encoder + VSN + GRN-blocks + bottleneck) hits `f32_max ≈ 3.4e38` somewhere. F0/F2 train clean because their gradients keep weights at natural scale (~0.5); F1's regime shift drives weights toward the clamp ceiling and the forward chain saturates. Mech 1+2 clamped TARGETS+ATOMS, Mech 9 clamped WEIGHTS, **nothing clamped ACTIVATIONS** — Mech 10 closes that gap. Mech 10 mechanism: after the trunk + temporal pipeline finalises `save_h_s2` in `submit_forward_ops_main` (end of "1c. Temporal pipeline" block — last writers are `mamba2_step` + `apply_regime_dropout`), the trainer launches `launch_clamp_finite_f32` on `save_h_s2` with bound `100 × ISV[H_S2_RMS_EMA_INDEX=96].max(1.0)`. Reuses the existing helper at `dqn_utility_kernels.cu:462` (`isfinite(v) ? clamp(v, ±max_abs) : 0.0f` — NaN→0, Inf→0, finite clamped) — no new kernel. ISV-driven via existing `H_S2_RMS_EMA_INDEX = 96` (already populated per Plan 4 Task 2c.3c.5; consumed by `mag_concat_qdir` adaptive scale per 2c.3c.6) — no new ISV slot, no hardcoded multipliers, consistent with `feedback_isv_for_adaptive_bounds.md`. ε on multiplier per SP1 pearl (`isv.max(1.0)`) — at fold boundary the EMA resets to 1.0 (neutral RMS) per `state_reset_registry.rs`, so the clamp floor is always at least `100 × 1 = 100`, never zero. SP1 diagnostic-ordering pearl honoured: the inline `check_nan_f32(slot 12)` fires BEFORE the clamp sanitises so the sentinel sees the original Inf/NaN before sanitization replaces it with 0; redundant with the slot 12 check inside `run_nan_checks_post_forward` (which runs later, post-clamp, on the sanitized buffer) — but the inline check at this site is the only one that captures the pre-clamp state. New diagnostic slot 49 = `h_s2_max_abs` at threshold `1e3 × h_s2_rms_ema_eff` mirrors the Mech 5 family pattern (clamp at 100×, diagnostic at 1000×) — sentinel that NEVER fires under normal Mech 10 operation. The fused NaN-check kernel (`dqn_nan_check_fused_f32_kernel`) gains a second ISV-derived float arg `h_s2_rms_ema_eff` distinct from `q_abs_ref_eff` (don't conflate the two ISV bases — `h_s2` lives in pre-readout activation space, not Q space). Buffer bumps: `nan_flags_buf` 49→50, metadata buffers (`nan_check_buf_ptrs`, `nan_check_buf_lens`) 25→26 entries, fused-kernel block count 25→26. Both name tables in `training_loop.rs` (`halt_nan` formatter + `halt_grad_collapse` formatter) extended to 50 entries with `"h_s2_max_abs"` at slot 49. `read_nan_flags()` return type 49→50 in both `GpuDqnTrainer` and `FusedTrainingCtx`. Per `feedback_no_partial_refactor.md`, Mech 10 + slot 49 land in the same commit (single coordinated migration). No new ISV slots, no new kernels, no graph-topology surprise (one extra launch added to the captured forward graph at the temporal-pipeline → loss-path boundary). Touched: `cuda_pipeline/dqn_utility_kernels.cu` (kernel signature + slot 25 branch), `cuda_pipeline/gpu_dqn_trainer.rs` (alloc 49→50, metadata 25→26, `populate_nan_check_meta` + slot 49 entry, fused-kernel `BLOCKS` 25→26 + `h_s2_rms_ema_eff` arg, `read_nan_flags` 49→50, Mech 10 clamp call at end of "1c. Temporal pipeline" in `submit_forward_ops_main`), `trainers/dqn/fused_training.rs` (`read_nan_flags` wrapper 49→50), `trainers/dqn/trainer/training_loop.rs` (both name tables 49→50 entries with `"h_s2_max_abs"` at slot 49), `docs/dqn-backward-nan-audit.md` (Mech 10 row in mechanism table + slot 49 row in slot-allocation table + ISV bindings note + full Mech 10 section). cargo check -p ml --lib clean (12 pre-existing warnings, none new). Validation deferred to one L40S smoke at this HEAD; expected F1 trains past step 1000 and slots 36-43 + slot 49 stay quiet.