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

refactor(cuda): features/targets_raw_cuda + 10 cold-path inits → MappedF32Buffer + Bug 2 diag

Browse Source 
Two related blocks landing together (Bug 2 diagnostic forces the structural
conversion of touched files; pre-commit guard at 5275932f4 enforces zero
*_via_pinned helpers per `feedback_no_hiding`).

== Bug 2 instrumentation (training_loop.rs after collect_experiences_gpu) ==

One-shot AtomicBool-gated DIAG_BUG2 dump of first 5 samples × 5 cols of
both gpu_batch.states and gpu_batch.next_states + col-0 mean/std/mean_abs
across the full batch. Prints once at first rollout — pre-graph-capture so
no hot-path impact. Interpretation key embedded:
  mean_abs ~0.001 = normalized log_return ✓
  mean_abs ~5000  = raw price ✗ (Bug 2 confirmed)

Resolves the smoke-vs-production state[0] divergence question that pure
static code reading couldn't pin: production aux_label_scale=5300 traces
back through gather kernel to "feat[0] of next_states_buf" but the fxcache
shows feat[0] stddev=1.0 (z-normalized log_return). Either next_states_buf
is populated from a different source than fxcache feat[0], or some kernel
mutates state[0] post-gather. The diagnostic prints both states (post-gather)
and next_states (post-shift) to disambiguate.

== Structural conversion: targets_raw_cuda + features_raw_cuda + 10 sites ==

Field types (DQNTrainer in trainer/mod.rs:621/624):
  Option<CudaSlice<f32>> → Option<MappedF32Buffer>

init_gpu_raw_buffers_from_slices (training_loop.rs): clone_to_device_f32_via_pinned
calls (lines 1308/1312) replaced with `MappedF32Buffer::new + write_from_slice`.

Consumer signatures (3 functions across 2 files):
  - gpu_experience_collector::collect_experiences_gpu(market_features_buf,
    targets_buf): &CudaSlice<f32> → &MappedF32Buffer (each)
  - gpu_experience_collector::launch_timestep_loop: same
  - gpu_experience_collector::compute_difficulty_scores(targets): same
  - decision_transformer::build_dt_trajectories(features_gpu, targets_gpu): same

Launch sites: `.arg(buf)` → `.arg(&buf.dev_ptr)` at all 5 launch_builder
invocations in gpu_experience_collector.rs and the 2 in build_dt_trajectories.

== Cold-path init conversions (forced by guard touching shared files) ==

gpu_dqn_trainer.rs (9 sites → 4 buffer-migration groups):
  - spec_u_s1/v_s1/u_s2/v_s2 (4 buffers, explicit)
  - spec_u/v macro pairs (alloc_spec_pair! body, expands to ~22 buffers)
  - graph_params (cross-branch graph message-passing, 60 floats)
  - denoise_params (diffusion Q-refinement MLP, 1800 floats)
  - qlstm_weights (xLSTM mLSTM-cell, 528 floats)

gpu_experience_collector.rs (1 site):
  - upload_ofi_features → ofi_gpu field type Option<CudaSlice<f32>> → MappedF32Buffer

Inherited from prior worktree-agent attempts (compile clean, included here):
  - sel_clip_buf in gpu_dqn_trainer.rs
  - RmsNormWeightSet γ buffers in gpu_weights.rs

PPO trainer (trainers/ppo.rs) deferred — not on eval-collapse hot path.

== Validation ==

cargo check -p ml --offline: clean
pre-commit hook (check_no_dtod_via_pinned + Invariant 7 + GPU hot-path guard): pass

Bug 2 diagnostic will fire on next L40S run and print state[0] stats for
first batch. If mean_abs ~0.001 → state[0] is correctly normalized and Bug 2
is elsewhere (maybe label_scale_ema initialization). If mean_abs ~5000 →
state[0] really is raw price; the rollout state-builder is the bug.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-05-02 12:57:12 +02:00
parent 5a5dd0fed1
commit cba9f25ed9
7 changed files with 318 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
crates/ml/src/cuda_pipeline/decision_transformer.rs
View File

@@ -1043,8 +1043,10 @@ impl DecisionTransformer {
/// 3. Pack `[rtg, state, action]` tokens into trajectory tensor via GPU kernel
///
/// # Arguments
/// * `features_gpu` - GPU buffer `[num_bars, feat_dim]` (42-dim feature vectors)
/// * `targets_gpu` - GPU buffer `[num_bars, 4]` (OHLC per bar)
/// * `features_gpu` - mapped-pinned buffer `[num_bars, feat_dim]` (42-dim
/// feature vectors). Kernel reads via `dev_ptr`; host writes happen at
/// fxcache load.
/// * `targets_gpu` - mapped-pinned buffer `[num_bars, 4]` (OHLC per bar).
/// * `num_bars` - Number of bars in the dataset
/// * `gamma` - Discount factor for return-to-go
///
@@ -1055,8 +1057,8 @@ impl DecisionTransformer {
/// - `num_complete_batches` is how many full batches of `batch_size` episodes are available
pub fn build_dt_trajectories(
&mut self,
features_gpu: &CudaSlice<f32>,
targets_gpu: &CudaSlice<f32>,
features_gpu: &MappedF32Buffer,
targets_gpu: &MappedF32Buffer,
num_bars: usize,
gamma: f64,
) -> Result<(CudaSlice<f32>, CudaSlice<i32>, usize), MLError> {
@@ -1103,7 +1105,7 @@ impl DecisionTransformer {
let mut actions_all_gpu = stream.alloc_zeros::<i32>(num_bars)
.map_err(|e| MLError::ModelError(format!("DT actions alloc: {e}")))?;
let targets_ptr = raw_ptr_f32(targets_gpu, stream);
let targets_ptr = targets_gpu.dev_ptr;
let rewards_ptr = raw_ptr_f32_mut(&mut rewards_gpu, stream);
let actions_all_ptr = {
let (ptr, guard) = actions_all_gpu.device_ptr_mut(stream);
@@ -1228,7 +1230,7 @@ impl DecisionTransformer {
.map_err(|e| MLError::ModelError(format!("DT target_actions alloc: {e}")))?;
{
let features_ptr = raw_ptr_f32(features_gpu, stream);
let features_ptr = features_gpu.dev_ptr;
let rtg_ptr = raw_ptr_f32(&rtg_gpu, stream);
let actions_all_ptr_rd = {
let (ptr, guard) = actions_all_gpu.device_ptr(stream);

183
crates/ml/src/cuda_pipeline/gpu_dqn_trainer.rs
View File

@@ -2739,22 +2739,27 @@ pub struct GpuDqnTrainer {
clip_grad_kernel: CudaFunction,
pad_states_kernel: CudaFunction,
/// Spectral norm left singular vectors: [out_dim] per weight matrix (W_s1, W_s2)
spec_u_s1: CudaSlice<f32>,
spec_v_s1: CudaSlice<f32>,
spec_u_s2: CudaSlice<f32>,
spec_v_s2: CudaSlice<f32>,
// Head spectral norm vectors (power iteration)
spec_u_v1: CudaSlice<f32>, spec_v_v1: CudaSlice<f32>,
spec_u_v2: CudaSlice<f32>, spec_v_v2: CudaSlice<f32>,
spec_u_a1: CudaSlice<f32>, spec_v_a1: CudaSlice<f32>,
spec_u_a2: CudaSlice<f32>, spec_v_a2: CudaSlice<f32>,
spec_u_bo1: CudaSlice<f32>, spec_v_bo1: CudaSlice<f32>,
spec_u_bo2: CudaSlice<f32>, spec_v_bo2: CudaSlice<f32>,
spec_u_bu1: CudaSlice<f32>, spec_v_bu1: CudaSlice<f32>,
spec_u_bu2: CudaSlice<f32>, spec_v_bu2: CudaSlice<f32>,
spec_u_bg1: CudaSlice<f32>, spec_v_bg1: CudaSlice<f32>,
spec_u_bg2: CudaSlice<f32>, spec_v_bg2: CudaSlice<f32>,
spec_u_bn: CudaSlice<f32>, spec_v_bn: CudaSlice<f32>,
/// Mapped pinned (cuMemHostAlloc DEVICEMAP) — host-init at construction, kernel
/// reads via dev_ptr. Avoids DtoD-via-pinned per
/// feedback_no_htod_htoh_only_mapped_pinned. Stored in `spectral_norm_descriptors`
/// as CUdeviceptr (= dev_ptr) so the batched power-iteration kernel reads them
/// indirectly through the descriptor table.
spec_u_s1: super::mapped_pinned::MappedF32Buffer,
spec_v_s1: super::mapped_pinned::MappedF32Buffer,
spec_u_s2: super::mapped_pinned::MappedF32Buffer,
spec_v_s2: super::mapped_pinned::MappedF32Buffer,
// Head spectral norm vectors (power iteration) — same mapped-pinned semantics as the trunk pair above.
spec_u_v1: super::mapped_pinned::MappedF32Buffer, spec_v_v1: super::mapped_pinned::MappedF32Buffer,
spec_u_v2: super::mapped_pinned::MappedF32Buffer, spec_v_v2: super::mapped_pinned::MappedF32Buffer,
spec_u_a1: super::mapped_pinned::MappedF32Buffer, spec_v_a1: super::mapped_pinned::MappedF32Buffer,
spec_u_a2: super::mapped_pinned::MappedF32Buffer, spec_v_a2: super::mapped_pinned::MappedF32Buffer,
spec_u_bo1: super::mapped_pinned::MappedF32Buffer, spec_v_bo1: super::mapped_pinned::MappedF32Buffer,
spec_u_bo2: super::mapped_pinned::MappedF32Buffer, spec_v_bo2: super::mapped_pinned::MappedF32Buffer,
spec_u_bu1: super::mapped_pinned::MappedF32Buffer, spec_v_bu1: super::mapped_pinned::MappedF32Buffer,
spec_u_bu2: super::mapped_pinned::MappedF32Buffer, spec_v_bu2: super::mapped_pinned::MappedF32Buffer,
spec_u_bg1: super::mapped_pinned::MappedF32Buffer, spec_v_bg1: super::mapped_pinned::MappedF32Buffer,
spec_u_bg2: super::mapped_pinned::MappedF32Buffer, spec_v_bg2: super::mapped_pinned::MappedF32Buffer,
spec_u_bn: super::mapped_pinned::MappedF32Buffer, spec_v_bn: super::mapped_pinned::MappedF32Buffer,
/// IQN trunk gradient scratch [trunk_params] (f32, receives backward dW output).
iqn_trunk_m: CudaSlice<f32>,
/// IQN trunk gradient Adam second moment [trunk_params].
@@ -2844,7 +2849,10 @@ pub struct GpuDqnTrainer {
/// [1] partial sums scratch for sel grad_norm reduction (SH2+1 ≤ 256, so 1 block).
sel_norm_partials: CudaSlice<f32>, // [1]
/// [1] max grad norm for sel Adam clipping (fixed 1.0).
sel_clip_buf: CudaSlice<f32>, // [1]
/// Mapped pinned (cuMemHostAlloc DEVICEMAP) — host-init at construction;
/// kernel reads via dev_ptr. Avoids DtoD-via-pinned per
/// feedback_no_htod_htoh_only_mapped_pinned.
sel_clip_buf: super::mapped_pinned::MappedF32Buffer, // [1]
/// Selectivity Adam step counter — pinned device-mapped (no per-step HtoD copy).
sel_t_pinned: *mut i32,
sel_t_dev_ptr: u64,
@@ -4460,13 +4468,19 @@ pub struct GpuDqnTrainer {
/// Fixed at near-identity initialization — not trained. The 4 edges encode
/// structural dependencies (dir→mag, etc.) via domain knowledge; training
/// would require a graph-level loss (deferred to future work).
graph_params: CudaSlice<f32>,
/// Mapped pinned (cuMemHostAlloc DEVICEMAP) — host-init at construction,
/// kernel reads via dev_ptr. Avoids DtoD-via-pinned per
/// feedback_no_htod_htoh_only_mapped_pinned.
graph_params: super::mapped_pinned::MappedF32Buffer,
/// branch_graph_message_pass kernel handle.
graph_msg_kernel: CudaFunction,
// ── Diffusion Q-refinement (2-step denoiser conditioned on Var[Q]) ─
/// Flat params for 2 denoising steps: 2 × (W1[24,24]+b1[24]+W2[12,24]+b2[12]) = 2 × 900 = 1800.
denoise_params: CudaSlice<f32>,
/// Mapped pinned (cuMemHostAlloc DEVICEMAP) — host-init at construction,
/// then trained in place by the aux Adam path (kernel writes via dev_ptr).
/// Avoids DtoD-via-pinned per feedback_no_htod_htoh_only_mapped_pinned.
denoise_params: super::mapped_pinned::MappedF32Buffer,
/// Adam first moment for denoise_params [1800].
denoise_adam_m: CudaSlice<f32>,
/// Adam second moment for denoise_params [1800].
@@ -4557,7 +4571,10 @@ pub struct GpuDqnTrainer {
// ── xLSTM temporal context (mLSTM cell) ────────────────────────────
/// [528] 6 weight matrices × [11, 8]: W_k, W_v, W_q, W_i, W_f, W_o
qlstm_weights: CudaSlice<f32>,
/// Mapped pinned (cuMemHostAlloc DEVICEMAP) — host-init at construction,
/// kernel reads via dev_ptr and trains in place. Avoids DtoD-via-pinned
/// per feedback_no_htod_htoh_only_mapped_pinned.
qlstm_weights: super::mapped_pinned::MappedF32Buffer,
/// [64] persistent 8×8 matrix memory (device state across steps)
qlstm_c: CudaSlice<f32>,
/// [8] persistent normalizer vector (device state across steps)
@@ -6087,7 +6104,7 @@ impl GpuDqnTrainer {
let b = batch_size as i32;
let blocks = ((batch_size as u32 + 255) / 256).max(1);
let q_ptr = self.q_coord_buf.raw_ptr();
let params_ptr = self.graph_params.raw_ptr();
let params_ptr = self.graph_params.dev_ptr;
unsafe {
self.stream
.launch_builder(&self.graph_msg_kernel)
@@ -6115,7 +6132,7 @@ impl GpuDqnTrainer {
let b = batch_size as i32;
let blocks = ((batch_size as u32 + 255) / 256).max(1);
let var_ptr = self.q_var_buf_trainer.raw_ptr();
let params_ptr = self.denoise_params.raw_ptr();
let params_ptr = self.denoise_params.dev_ptr;
// q_var_buf_trainer is sized [B, total_actions] to match the kernel's
// per-action variance write stride in compute_expected_q. The denoiser
// MLP only consumes the first D=12 slots per sample, but the launch
@@ -14262,10 +14279,11 @@ impl GpuDqnTrainer {
let sel_norm_partials = stream.alloc_zeros::<f32>(1)
.map_err(|e| MLError::ModelError(format!("alloc sel_norm_partials: {e}")))?;
// 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::<f32>(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}")))?;
// Mapped pinned: host-init via host_ptr; aux Adam-update kernel reads dev_ptr.
// No DtoD copy per feedback_no_htod_htoh_only_mapped_pinned.
let sel_clip_buf = unsafe { super::mapped_pinned::MappedF32Buffer::new(1) }
.map_err(|e| MLError::ModelError(format!("sel_clip_buf alloc (1 f32): {e}")))?;
sel_clip_buf.write_from_slice(&[1.0_f32]);
// 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;
@@ -14414,18 +14432,21 @@ impl GpuDqnTrainer {
let aux_norm_scratch = alloc_f32(&stream, 1, "aux_norm_scratch")?;
let attn_bw_scratch = alloc_f32(&stream, b * config.shared_h2, "attn_bw_scratch")?;
let iqn_trunk_t_buf = alloc_i32(&stream, 1, "iqn_trunk_t_buf")?;
let mut spec_u_s1 = alloc_f32(&stream, config.shared_h1, "spec_u_s1")?;
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}")))?;
// Mapped pinned: host-init via host_ptr; batched power-iteration kernel
// reads via dev_ptr (indirected through `spectral_norm_descriptors`).
// No DtoD copy per feedback_no_htod_htoh_only_mapped_pinned.
let spec_u_s1 = unsafe { super::mapped_pinned::MappedF32Buffer::new(config.shared_h1) }
.map_err(|e| MLError::ModelError(format!("spec_u_s1 alloc ({} f32): {e}", config.shared_h1)))?;
let spec_v_s1 = unsafe { super::mapped_pinned::MappedF32Buffer::new(ml_core::state_layout::STATE_DIM) }
.map_err(|e| MLError::ModelError(format!("spec_v_s1 alloc ({} f32): {e}", ml_core::state_layout::STATE_DIM)))?;
let spec_u_s2 = unsafe { super::mapped_pinned::MappedF32Buffer::new(config.shared_h2) }
.map_err(|e| MLError::ModelError(format!("spec_u_s2 alloc ({} f32): {e}", config.shared_h2)))?;
let spec_v_s2 = unsafe { super::mapped_pinned::MappedF32Buffer::new(config.shared_h1) }
.map_err(|e| MLError::ModelError(format!("spec_v_s2 alloc ({} f32): {e}", config.shared_h1)))?;
spec_u_s1.write_from_slice(&init_u_s1);
spec_v_s1.write_from_slice(&init_v_s1);
spec_u_s2.write_from_slice(&init_u_s2);
spec_v_s2.write_from_slice(&init_v_s2);
// Head spectral norm vectors: value head, adv/branch heads
let vh = config.value_h;
@@ -14436,16 +14457,20 @@ impl GpuDqnTrainer {
let b1 = config.branch_1_size;
let b2 = config.branch_2_size;
// Mapped-pinned spectral-norm pair: host-init via host_ptr; the batched
// power-iteration kernel reads dev_ptr (indirected through
// `spectral_norm_descriptors`). No DtoD copy per
// feedback_no_htod_htoh_only_mapped_pinned.
macro_rules! alloc_spec_pair {
($u_name:ident, $v_name:ident, $u_n:expr, $v_n:expr, $lbl_u:literal, $lbl_v:literal) => {
let mut $u_name = alloc_f32(&stream, $u_n, $lbl_u)?;
let mut $v_name = alloc_f32(&stream, $v_n, $lbl_v)?;
let $u_name = unsafe { super::mapped_pinned::MappedF32Buffer::new($u_n) }
.map_err(|e| MLError::ModelError(format!("{} alloc ({} f32): {e}", $lbl_u, $u_n)))?;
let $v_name = unsafe { super::mapped_pinned::MappedF32Buffer::new($v_n) }
.map_err(|e| MLError::ModelError(format!("{} alloc ({} f32): {e}", $lbl_v, $v_n)))?;
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)))?;
$u_name.write_from_slice(&init_u);
$v_name.write_from_slice(&init_v);
};
}
@@ -14502,33 +14527,36 @@ impl GpuDqnTrainer {
// normed — Task 2c.3c will extend coverage.
let bn_dim_u64 = config.bottleneck_dim.max(1) as u64;
let md_u64 = config.market_dim.max(1) as u64;
// Mapped-pinned spec_u/spec_v: dev_ptr is the CUdeviceptr (u64) the
// batched power-iteration kernel reads. Same width as `raw_ptr()` on
// CudaSlice — purely a structural rename, no semantic change.
let host_desc: [u64; 78] = [
// [0] w_a_h_s1 [shared_h1, state_dim] — GRN h_s1 Linear_a (was W_s1)
w_ptrs[0], spec_u_s1.raw_ptr(), spec_v_s1.raw_ptr(), sh1_u64, sd_u64, 0,
w_ptrs[0], spec_u_s1.dev_ptr, spec_v_s1.dev_ptr, sh1_u64, sd_u64, 0,
// [1] w_a_h_s2 [shared_h2, shared_h1] — GRN h_s2 Linear_a (was W_s2)
w_ptrs[7], spec_u_s2.raw_ptr(), spec_v_s2.raw_ptr(), sh2_u64, sh1_u64, 0,
w_ptrs[7], spec_u_s2.dev_ptr, spec_v_s2.dev_ptr, sh2_u64, sh1_u64, 0,
// [2] W_v1 [value_h, shared_h2] — GOFF 13 (was 4)
w_ptrs[13], spec_u_v1.raw_ptr(), spec_v_v1.raw_ptr(), vh_u64, sh2_u64, 0,
w_ptrs[13], spec_u_v1.dev_ptr, spec_v_v1.dev_ptr, vh_u64, sh2_u64, 0,
// [3] W_v2 [num_atoms, value_h] — GOFF 15 (was 6)
w_ptrs[15], spec_u_v2.raw_ptr(), spec_v_v2.raw_ptr(), na_u64, vh_u64, 0,
w_ptrs[15], spec_u_v2.dev_ptr, spec_v_v2.dev_ptr, na_u64, vh_u64, 0,
// [4] W_a1 [adv_h, shared_h2] — GOFF 17 (was 8)
w_ptrs[17], spec_u_a1.raw_ptr(), spec_v_a1.raw_ptr(), ah_u64, sh2_u64, 0,
w_ptrs[17], spec_u_a1.dev_ptr, spec_v_a1.dev_ptr, ah_u64, sh2_u64, 0,
// [5] W_a2 [b0*num_atoms, adv_h] — GOFF 19 (was 10)
w_ptrs[19], spec_u_a2.raw_ptr(), spec_v_a2.raw_ptr(), b0_u64 * na_u64, ah_u64, 0,
w_ptrs[19], spec_u_a2.dev_ptr, spec_v_a2.dev_ptr, b0_u64 * na_u64, ah_u64, 0,
// [6] W_bo1 [adv_h, shared_h2] — GOFF 21 (was 12)
w_ptrs[21], spec_u_bo1.raw_ptr(), spec_v_bo1.raw_ptr(), ah_u64, sh2_u64, 0,
w_ptrs[21], spec_u_bo1.dev_ptr, spec_v_bo1.dev_ptr, ah_u64, sh2_u64, 0,
// [7] W_bo2 [b1*num_atoms, adv_h] — GOFF 23 (was 14)
w_ptrs[23], spec_u_bo2.raw_ptr(), spec_v_bo2.raw_ptr(), b1_u64 * na_u64, ah_u64, 0,
w_ptrs[23], spec_u_bo2.dev_ptr, spec_v_bo2.dev_ptr, b1_u64 * na_u64, ah_u64, 0,
// [8] W_bu1 [adv_h, shared_h2] — GOFF 25 (was 16)
w_ptrs[25], spec_u_bu1.raw_ptr(), spec_v_bu1.raw_ptr(), ah_u64, sh2_u64, 0,
w_ptrs[25], spec_u_bu1.dev_ptr, spec_v_bu1.dev_ptr, ah_u64, sh2_u64, 0,
// [9] W_bu2 [b2*num_atoms, adv_h] — GOFF 27 (was 18)
w_ptrs[27], spec_u_bu2.raw_ptr(), spec_v_bu2.raw_ptr(), b2_u64 * na_u64, ah_u64, 0,
w_ptrs[27], spec_u_bu2.dev_ptr, spec_v_bu2.dev_ptr, b2_u64 * na_u64, ah_u64, 0,
// [10] W_bg1 [adv_h, shared_h2] — GOFF 29 (was 20)
w_ptrs[29], spec_u_bg1.raw_ptr(), spec_v_bg1.raw_ptr(), ah_u64, sh2_u64, 0,
w_ptrs[29], spec_u_bg1.dev_ptr, spec_v_bg1.dev_ptr, ah_u64, sh2_u64, 0,
// [11] W_bg2 [b3*num_atoms, adv_h] — GOFF 31 (was 22)
w_ptrs[31], spec_u_bg2.raw_ptr(), spec_v_bg2.raw_ptr(), b3_u64 * na_u64, ah_u64, 0,
w_ptrs[31], spec_u_bg2.dev_ptr, spec_v_bg2.dev_ptr, b3_u64 * na_u64, ah_u64, 0,
// [12] W_bn [bottleneck_dim, market_dim] — GOFF 33 (was 24)
w_ptrs[33], spec_u_bn.raw_ptr(), spec_v_bn.raw_ptr(), bn_dim_u64, md_u64, 0,
w_ptrs[33], spec_u_bn.dev_ptr, spec_v_bn.dev_ptr, bn_dim_u64, md_u64, 0,
];
upload_via_mapped_u64(&stream, 78, &host_desc, "spectral_norm_descriptors")?
};
@@ -15781,10 +15809,11 @@ impl GpuDqnTrainer {
graph_params_host[e * 15 + 6 + 4] = 0.1; // row 1, col 1
graph_params_host[e * 15 + 6 + 8] = 0.1; // row 2, col 2
}
let mut graph_params = stream.alloc_zeros::<f32>(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}")))?;
// Mapped pinned: host-init via host_ptr; branch_graph_message_pass kernel
// reads dev_ptr. No DtoD copy per feedback_no_htod_htoh_only_mapped_pinned.
let graph_params = unsafe { super::mapped_pinned::MappedF32Buffer::new(60) }
.map_err(|e| MLError::ModelError(format!("graph_params alloc (60 f32): {e}")))?;
graph_params.write_from_slice(&graph_params_host);
info!("GpuDqnTrainer: branch_graph_message_pass kernel loaded (60 params, near-identity W_msg init)");
// ── Diffusion Q-refinement ────────────────────────────────────────────────
@@ -15836,10 +15865,12 @@ impl GpuDqnTrainer {
off += 12;
}
}
let mut denoise_params = stream.alloc_zeros::<f32>(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}")))?;
// Mapped pinned: host-init (Xavier) via host_ptr; q_denoise_step kernel
// reads dev_ptr (and aux Adam writes back). No DtoD copy per
// feedback_no_htod_htoh_only_mapped_pinned.
let denoise_params = unsafe { super::mapped_pinned::MappedF32Buffer::new(DENOISE_TOTAL_PARAMS) }
.map_err(|e| MLError::ModelError(format!("denoise_params alloc ({DENOISE_TOTAL_PARAMS} f32): {e}")))?;
denoise_params.write_from_slice(&denoise_params_host);
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")
@@ -15983,10 +16014,12 @@ impl GpuDqnTrainer {
*w = u * 2.0 * qlstm_xavier_scale;
}
}
let mut qlstm_weights = stream.alloc_zeros::<f32>(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}")))?;
// Mapped pinned: host-init (Xavier uniform) via host_ptr; xLSTM kernels
// (qlstm_step + qlstm_train_step) read dev_ptr and SGD-update in place.
// No DtoD copy per feedback_no_htod_htoh_only_mapped_pinned.
let qlstm_weights = unsafe { super::mapped_pinned::MappedF32Buffer::new(QLSTM_WEIGHT_COUNT) }
.map_err(|e| MLError::ModelError(format!("qlstm_weights alloc ({QLSTM_WEIGHT_COUNT} f32): {e}")))?;
qlstm_weights.write_from_slice(&qlstm_weights_host);
let qlstm_c = stream.alloc_zeros::<f32>(QLSTM_HEAD_DIM * QLSTM_HEAD_DIM) // [64]
.map_err(|e| MLError::ModelError(format!("alloc qlstm_c: {e}")))?;
let qlstm_n = stream.alloc_zeros::<f32>(QLSTM_HEAD_DIM) // [8]
@@ -20142,7 +20175,7 @@ impl GpuDqnTrainer {
let c_ptr = self.qlstm_c.raw_ptr();
let n_ptr = self.qlstm_n.raw_ptr();
let ctx_ptr = self.qlstm_context.raw_ptr();
let w_ptr = self.qlstm_weights.raw_ptr();
let w_ptr = self.qlstm_weights.dev_ptr;
let id = 11_i32;
let hd = 8_i32;
unsafe {
@@ -20170,7 +20203,7 @@ impl GpuDqnTrainer {
/// Then updates prev_q_mean from current q_stats[3] (q_mean slot).
/// Takes &self: raw pointers (prev_q_mean_pinned) are mutated via unsafe.
pub(crate) fn launch_qlstm_train_step(&self) -> Result<(), MLError> {
let w_ptr = self.qlstm_weights.raw_ptr();
let w_ptr = self.qlstm_weights.dev_ptr;
let ctx_ptr = self.qlstm_context.raw_ptr();
let c_ptr = self.qlstm_c.raw_ptr();
let n_ptr = self.qlstm_n.raw_ptr();
@@ -20285,7 +20318,7 @@ impl GpuDqnTrainer {
let blocks = ((1800_u32 + 255) / 256).max(1);
let q_refined_ptr = self.q_coord_buf.raw_ptr();
let q_target_ptr = self.denoise_target_q_buf.raw_ptr();
let params_ptr = self.denoise_params.raw_ptr();
let params_ptr = self.denoise_params.dev_ptr;
let d_params_ptr = self.denoise_grad.raw_ptr();
let q_input_ptr = self.denoise_q_input_buf.raw_ptr();
let var_ptr = self.q_var_buf_trainer.raw_ptr();
@@ -20347,7 +20380,7 @@ impl GpuDqnTrainer {
let alpha: f32 = 1.0;
let beta: f32 = 0.0;
let params_ptr = self.denoise_params.raw_ptr();
let params_ptr = self.denoise_params.dev_ptr;
let d_params_ptr = self.denoise_grad.raw_ptr();
let q_input_ptr = self.denoise_q_input_buf.raw_ptr();
let var_ptr = self.q_var_buf_trainer.raw_ptr();
@@ -20930,7 +20963,7 @@ impl GpuDqnTrainer {
let beta2: f32 = 0.999;
let eps: f32 = 1e-8;
let weight_decay: f32 = 0.0;
let params_ptr = self.denoise_params.raw_ptr();
let params_ptr = self.denoise_params.dev_ptr;
let m_ptr = self.denoise_adam_m.raw_ptr();
let v_ptr = self.denoise_adam_v.raw_ptr();
let clip_ptr = self.denoise_clip_buf.raw_ptr();
@@ -25422,7 +25455,7 @@ impl GpuDqnTrainer {
let params_ptr = self.sel_params.raw_ptr();
let m_ptr = self.sel_adam_m.raw_ptr();
let v_ptr = self.sel_adam_v.raw_ptr();
let clip_ptr = self.sel_clip_buf.raw_ptr();
let clip_ptr = self.sel_clip_buf.dev_ptr;
let t_ptr = self.sel_t_dev_ptr;
// Reuse main mask buffer as dummy — weight_decay=0.0 nullifies all mask values.
let wd_mask_ptr = self.weight_decay_mask.raw_ptr();

53
crates/ml/src/cuda_pipeline/gpu_experience_collector.rs
View File

@@ -684,8 +684,11 @@ pub struct GpuExperienceCollector {
/// Expert demo override probability (0.0 = disabled, decays over epochs).
expert_ratio: f32,
// OFI features pre-uploaded to GPU [total_bars * OFI_DIM] — unconditional
ofi_gpu: CudaSlice<f32>,
// OFI features pre-uploaded to GPU [total_bars * OFI_DIM] — unconditional.
// Mapped pinned (cuMemHostAlloc DEVICEMAP) — host writes once via host_ptr in
// `upload_ofi_features` before the first collect, kernel reads via dev_ptr.
// Avoids DtoD-via-pinned per feedback_no_htod_htoh_only_mapped_pinned.
ofi_gpu: super::mapped_pinned::MappedF32Buffer,
ofi_dim: usize,
// GPU-resident curiosity forward model trainer (None if curiosity disabled)
@@ -1326,9 +1329,12 @@ impl GpuExperienceCollector {
let pinned_rewards = unsafe { PinnedHostBuf::<f32>::new(total_output)? };
let pinned_dones = unsafe { PinnedHostBuf::<i32>::new(total_output)? };
// OFI buffer — starts as 1-element zero, replaced by upload_ofi_features before first collect
let ofi_init = stream.alloc_zeros::<f32>(1)
.map_err(|e| MLError::ModelError(format!("alloc OFI init: {e}")))?;
// OFI buffer — starts as 1-element zero placeholder. `upload_ofi_features`
// replaces this with the real allocation (`total_bars * OFI_DIM`) before the
// first collect. Mapped pinned: host-init zero via host_ptr; kernel reads
// dev_ptr. No DtoD copy per feedback_no_htod_htoh_only_mapped_pinned.
let ofi_init = unsafe { super::mapped_pinned::MappedF32Buffer::new(1) }
.map_err(|e| MLError::ModelError(format!("ofi_gpu init alloc (1 f32): {e}")))?;
// ── Step 10: Log buffer sizes ───────────────────────────────────
let per_timestep_bytes = n * state_dim * 4 // batch_states
@@ -3032,8 +3038,8 @@ impl GpuExperienceCollector {
/// these buffers to ops that run on a different CUDA stream.
pub fn collect_experiences_gpu(
&mut self,
market_features_buf: &CudaSlice<f32>,
targets_buf: &CudaSlice<f32>,
market_features_buf: &super::mapped_pinned::MappedF32Buffer,
targets_buf: &super::mapped_pinned::MappedF32Buffer,
episode_starts: &[i32],
config: &ExperienceCollectorConfig,
) -> Result<GpuExperienceBatch, MLError> {
@@ -3185,7 +3191,7 @@ impl GpuExperienceCollector {
unsafe {
self.stream
.launch_builder(&self.hindsight_relabel_kernel)
.arg(targets_buf) // const float* targets
.arg(&targets_buf.dev_ptr) // const float* targets (mapped pinned dev_ptr)
.arg(&actions) // const int* actions
.arg(&bar_indices_gpu_ptr) // const int* bar_indices (mapped pinned dev_ptr)
.arg(&rewards) // float* rewards
@@ -3245,8 +3251,8 @@ impl GpuExperienceCollector {
/// 5. `experience_env_step` — portfolio simulation, reward, done, advance timestep
fn launch_timestep_loop(
&mut self,
market_features_buf: &CudaSlice<f32>,
targets_buf: &CudaSlice<f32>,
market_features_buf: &super::mapped_pinned::MappedF32Buffer,
targets_buf: &super::mapped_pinned::MappedF32Buffer,
episode_starts: &[i32],
config: &ExperienceCollectorConfig,
) -> Result<(usize, usize), MLError> {
@@ -3394,12 +3400,12 @@ impl GpuExperienceCollector {
// Plan params, ISV signals, and OFI features pointers
let plan_ptr = self.plan_params_dev_ptr;
let isv_ptr = self.isv_signals_dev_ptr;
let ofi_ptr = self.ofi_gpu.raw_ptr();
let ofi_ptr = self.ofi_gpu.dev_ptr;
let ofi_dim_i32 = self.ofi_dim as i32;
self.stream
.launch_builder(&self.state_gather_kernel)
.arg(market_features_buf)
.arg(&market_features_buf.dev_ptr) // mapped pinned dev_ptr
.arg(&self.episode_starts_buf)
.arg(&self.current_timesteps)
.arg(&self.portfolio_states)
@@ -3811,8 +3817,8 @@ impl GpuExperienceCollector {
self.stream
.launch_builder(&self.expert_override_kernel)
.arg(&mut self.batch_actions)
.arg(targets_buf)
.arg(market_features_buf)
.arg(&targets_buf.dev_ptr) // mapped pinned dev_ptr
.arg(&market_features_buf.dev_ptr) // mapped pinned dev_ptr
.arg(&self.portfolio_states)
.arg(&self.episode_starts_buf)
.arg(&self.current_timesteps)
@@ -3845,7 +3851,7 @@ impl GpuExperienceCollector {
unsafe {
self.stream
.launch_builder(&self.env_step_kernel)
.arg(targets_buf)
.arg(&targets_buf.dev_ptr) // mapped pinned dev_ptr
.arg(&self.episode_starts_buf)
.arg(&mut self.current_timesteps)
.arg(&self.batch_actions)
@@ -3857,7 +3863,7 @@ impl GpuExperienceCollector {
.arg(&self.batch_states)
.arg(&max_pos)
.arg(&tx_cost)
.arg(market_features_buf)
.arg(&market_features_buf.dev_ptr) // mapped pinned dev_ptr
.arg(&md)
.arg(&l_i32)
.arg(&n_i32)
@@ -4500,8 +4506,12 @@ 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}")))?;
// Mapped pinned: host-write via host_ptr; kernel reads via dev_ptr.
// Replaces the 1-element placeholder allocated in the constructor.
// No DtoD copy per feedback_no_htod_htoh_only_mapped_pinned.
let gpu_buf = unsafe { super::mapped_pinned::MappedF32Buffer::new(ofi_flat.len()) }
.map_err(|e| MLError::ModelError(format!("upload_ofi_features alloc ({} f32): {e}", ofi_flat.len())))?;
gpu_buf.write_from_slice(ofi_flat);
info!(
"OFI features uploaded to GPU: {} bars x {} dims ({:.1} KB)",
ofi_flat.len() / self.ofi_dim,
@@ -4518,11 +4528,14 @@ impl GpuExperienceCollector {
/// Higher scores indicate more volatile / harder-to-trade bars. The output
/// can be used to bias episode start selection toward harder regions.
pub fn compute_difficulty_scores(
&self, targets: &CudaSlice<f32>, total_bars: usize,
&self,
targets: &super::mapped_pinned::MappedF32Buffer,
total_bars: usize,
) -> Result<CudaSlice<f32>, MLError> {
let scores = self.stream.alloc_zeros::<f32>(total_bars)
.map_err(|e| MLError::ModelError(format!("alloc difficulty_scores: {e}")))?;
let targets_ptr = targets.raw_ptr();
// Mapped-pinned dev_ptr aliases the host pages — kernels read it directly.
let targets_ptr = targets.dev_ptr;
let scores_ptr = scores.raw_ptr();
let n = total_bars as i32;
let blocks = ((total_bars + 255) / 256) as u32;

34
crates/ml/src/cuda_pipeline/gpu_weights.rs
View File

@@ -526,10 +526,10 @@ const RMSNORM_WEIGHT_NAMES: [&str; 4] = [
/// - advantage_rmsnorm:[128] (after advantage FC)
#[allow(missing_debug_implementations)]
pub struct RmsNormWeightSet {
pub gamma_s0: CudaSlice<f32>, // shared_rmsnorm_0.weight [shared_h1]
pub gamma_s1: CudaSlice<f32>, // shared_rmsnorm_1.weight [shared_h2]
pub gamma_v: CudaSlice<f32>, // value_rmsnorm.weight [value_h]
pub gamma_a: CudaSlice<f32>, // advantage_rmsnorm.weight [adv_h]
pub gamma_s0: super::mapped_pinned::MappedF32Buffer, // shared_rmsnorm_0.weight [shared_h1]
pub gamma_s1: super::mapped_pinned::MappedF32Buffer, // shared_rmsnorm_1.weight [shared_h2]
pub gamma_v: super::mapped_pinned::MappedF32Buffer, // value_rmsnorm.weight [value_h]
pub gamma_a: super::mapped_pinned::MappedF32Buffer, // advantage_rmsnorm.weight [adv_h]
}
impl RmsNormWeightSet {
@@ -537,17 +537,21 @@ impl RmsNormWeightSet {
///
/// Dimensions must match the actual network layer sizes to avoid GPU OOB reads.
/// `network_dims`: `(shared_h1, shared_h2, value_h, adv_h)`.
///
/// Each γ buffer is a `MappedF32Buffer` (cuMemHostAlloc DEVICEMAP). Host
/// writes the all-ones payload via `host_ptr`; the kernel reads `dev_ptr`
/// directly. No DtoD copy, no HtoD copy — graph-capture-safe.
pub fn ones(
stream: &Arc<CudaStream>,
_stream: &Arc<CudaStream>,
network_dims: (usize, usize, usize, usize),
) -> Result<Self, MLError> {
let (shared_h1, shared_h2, value_h, adv_h) = network_dims;
let alloc_ones = |size: usize| -> Result<CudaSlice<f32>, MLError> {
let alloc_ones = |size: usize| -> Result<super::mapped_pinned::MappedF32Buffer, MLError> {
let buf = unsafe { super::mapped_pinned::MappedF32Buffer::new(size) }
.map_err(|e| MLError::ModelError(format!("RMSNorm γ alloc ({size} f32): {e}")))?;
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}")))
buf.write_from_slice(&data);
Ok(buf)
};
Ok(Self {
gamma_s0: alloc_ones(shared_h1)?,
@@ -688,10 +692,12 @@ impl KernelWeightPack {
b2: raw_device_ptr(&curiosity.b2, stream),
},
rmsnorm: GpuRMSNormPtrs {
s0: raw_device_ptr(&rmsnorm.gamma_s0, stream),
s1: raw_device_ptr(&rmsnorm.gamma_s1, stream),
v: raw_device_ptr(&rmsnorm.gamma_v, stream),
a: raw_device_ptr(&rmsnorm.gamma_a, stream),
// MappedF32Buffer.dev_ptr is CUdeviceptr (= u64); no per-call
// device_ptr() guard needed (mapped pinned has no SyncOnDrop).
s0: rmsnorm.gamma_s0.dev_ptr,
s1: rmsnorm.gamma_s1.dev_ptr,
v: rmsnorm.gamma_v.dev_ptr,
a: rmsnorm.gamma_a.dev_ptr,
},
online_br: GpuBranchPtrs {
w_o1: online_br.w_bo1, b_o1: online_br.b_bo1,

13
crates/ml/src/trainers/dqn/trainer/mod.rs
View File

@@ -617,11 +617,16 @@ pub struct DQNTrainer {
/// Pre-uploaded GPU training data (set once, reused across epochs)
pub(crate) gpu_data: Option<DqnGpuData>,
/// Raw cudarc targets buffer for CUDA kernel (parallel to raw features buffer)
pub(crate) targets_raw_cuda: Option<cudarc::driver::CudaSlice<f32>>,
/// Raw targets buffer for CUDA kernel (parallel to raw features buffer).
/// `MappedF32Buffer` (cuMemHostAlloc DEVICEMAP): host writes via `host_ptr`
/// at fxcache load; kernels read via `dev_ptr`. No DtoD copy at upload —
/// satisfies `feedback_no_htod_htoh_only_mapped_pinned` and is graph-
/// capture safe.
pub(crate) targets_raw_cuda: Option<crate::cuda_pipeline::mapped_pinned::MappedF32Buffer>,
/// Raw cudarc features buffer for CUDA experience kernel [num_bars * 42]
pub(crate) features_raw_cuda: Option<cudarc::driver::CudaSlice<f32>>,
/// Raw features buffer for CUDA experience kernel `[num_bars * market_dim]`.
/// Same `MappedF32Buffer` storage pattern as `targets_raw_cuda` above.
pub(crate) features_raw_cuda: Option<crate::cuda_pipeline::mapped_pinned::MappedF32Buffer>,
/// GPU experience collector for zero-roundtrip CUDA kernel (Phase 2b)
pub(crate) gpu_experience_collector: Option<crate::cuda_pipeline::gpu_experience_collector::GpuExperienceCollector>,

126
crates/ml/src/trainers/dqn/trainer/training_loop.rs
View File

@@ -1291,10 +1291,12 @@ impl DQNTrainer {
if self.targets_raw_cuda.is_some() {
return Ok(());
}
let stream = match self.cuda_stream {
Some(ref s) => std::sync::Arc::clone(s),
None => return Ok(()),
};
// The mapped-pinned allocator only needs an active CUDA context on this
// thread; it does not consume the stream. Keep the early-return for the
// None-stream case to preserve prior semantics.
if self.cuda_stream.is_none() {
return Ok(());
}
let num_bars = features.len();
let flat_targets: Vec<f32> = targets.iter()
@@ -1304,14 +1306,31 @@ 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}"))?
);
// Mapped pinned (cuMemHostAlloc DEVICEMAP): host writes the f32 payload
// via `host_ptr`; kernels read via `dev_ptr`. No HtoD/DtoD copy — the
// device pointer aliases the same physical pages as the host pointer,
// so the kernel sees the data after a stream-sync barrier (mapped-pinned
// coherence). Replaces `clone_to_device_f32_via_pinned` per
// `feedback_no_htod_htoh_only_mapped_pinned` and the `check_no_dtod_via_pinned`
// pre-commit guard. Safety: cuda_stream-is-Some implies an active CUDA
// context on this thread (DQNTrainer construction sets it).
let targets_buf = unsafe {
crate::cuda_pipeline::mapped_pinned::MappedF32Buffer::new(flat_targets.len())
}
.map_err(|e| anyhow::anyhow!(
"targets_raw alloc ({} f32): {e}", flat_targets.len()
))?;
targets_buf.write_from_slice(&flat_targets);
self.targets_raw_cuda = Some(targets_buf);
let features_buf = unsafe {
crate::cuda_pipeline::mapped_pinned::MappedF32Buffer::new(flat_features.len())
}
.map_err(|e| anyhow::anyhow!(
"features_raw alloc ({} f32): {e}", flat_features.len()
))?;
features_buf.write_from_slice(&flat_features);
self.features_raw_cuda = Some(features_buf);
tracing::info!("CUDA raw buffers uploaded from slices: {} bars x 42+6", num_bars);
Ok(())
@@ -1763,6 +1782,70 @@ impl DQNTrainer {
"GPU zero-roundtrip collection FAILED (no CPU fallback): {e}"
))?;
// ── DIAG: pin Bug 2 (state[0]=raw_close vs log_return) ─────────────
// One-shot dump of first 5 samples × 5 cols of next_states + states.
// Goal: determine whether the rollout produces normalized log_return
// or raw price at state[0]. Compare against fxcache feat[0] stddev=1.0
// and aux label_scale=5300 (production observed value pre-fix).
// Triggers once via static AtomicBool. f32 dtoh download — slow but
// pre-graph-capture, no perf impact since it only fires step 1.
{
use std::sync::atomic::{AtomicBool, Ordering};
static DUMPED: AtomicBool = AtomicBool::new(false);
if !DUMPED.swap(true, Ordering::AcqRel) {
if let Some(ref stream) = self.cuda_stream {
let sd = gpu_batch.state_dim;
let n_dump = 5_usize.min(gpu_batch.n_episodes * gpu_batch.timesteps * 2);
let n_cols = 5_usize.min(sd);
// Pre-graph-capture one-shot DtoH download — the diagnostic
// is fine to use the slow path; it fires once per process.
// `clone_dtoh` replaces the deprecated `memcpy_dtov`.
let host_states: Vec<f32> = stream
.clone_dtoh(&gpu_batch.states)
.unwrap_or_default();
let host_next: Vec<f32> = stream
.clone_dtoh(&gpu_batch.next_states)
.unwrap_or_default();
if !host_states.is_empty() && !host_next.is_empty() {
// Per-sample first-5-col rows.
for i in 0..n_dump {
let s_off = i * sd;
let s_row: Vec<f32> = (0..n_cols)
.map(|j| host_states.get(s_off + j).copied().unwrap_or(f32::NAN))
.collect();
let n_row: Vec<f32> = (0..n_cols)
.map(|j| host_next.get(s_off + j).copied().unwrap_or(f32::NAN))
.collect();
tracing::warn!(
target: "DIAG_BUG2",
"sample[{i}] state[0..{n_cols}]={s_row:?} next_state[0..{n_cols}]={n_row:?}"
);
}
// Mean abs at col 0 across all samples — direct EMA approximation.
let n_total = host_next.len() / sd;
let sum_abs_s: f64 = (0..n_total).map(|i| host_states[i * sd].abs() as f64).sum();
let sum_abs_n: f64 = (0..n_total).map(|i| host_next[i * sd].abs() as f64).sum();
let mean_abs_s = sum_abs_s / (n_total.max(1) as f64);
let mean_abs_n = sum_abs_n / (n_total.max(1) as f64);
// Stddev too — quick pass.
let mean_s: f64 = (0..n_total).map(|i| host_states[i * sd] as f64).sum::<f64>() / (n_total.max(1) as f64);
let mean_n: f64 = (0..n_total).map(|i| host_next[i * sd] as f64).sum::<f64>() / (n_total.max(1) as f64);
let var_s: f64 = (0..n_total).map(|i| (host_states[i * sd] as f64 - mean_s).powi(2)).sum::<f64>() / (n_total.max(1) as f64);
let var_n: f64 = (0..n_total).map(|i| (host_next[i * sd] as f64 - mean_n).powi(2)).sum::<f64>() / (n_total.max(1) as f64);
tracing::warn!(
target: "DIAG_BUG2",
"col0 stats over {n_total} samples: states[mean={mean_s:.4e}, std={:.4e}, mean_abs={mean_abs_s:.4e}] next_states[mean={mean_n:.4e}, std={:.4e}, mean_abs={mean_abs_n:.4e}]",
var_s.sqrt(), var_n.sqrt()
);
tracing::warn!(
target: "DIAG_BUG2",
"interpretation: mean_abs ~0.001 = normalized log_return ✓ | mean_abs ~5000 = raw price ✗ (Bug 2 confirmed)"
);
}
}
}
}
let count = gpu_batch.n_episodes * gpu_batch.timesteps * 2; // counterfactual doubles experiences
// Rank normalization REMOVED — PopArt handles reward normalization in training.
@@ -3668,18 +3751,15 @@ impl DQNTrainer {
// features_raw_cuda lives on DQNTrainer; passed as dev ptr.
if let Some(ref features_buf) = self.features_raw_cuda {
let market_dim_usize = fused.trainer().config().market_dim.max(1);
let inferred_num_bars = features_buf.len() / market_dim_usize;
// MappedF32Buffer.len is a pub field (not a method) — the
// staging buffer aliases host pages, so the element count
// equals the f32 length passed to ::new().
let inferred_num_bars = features_buf.len / market_dim_usize;
if inferred_num_bars > 0 {
// Hold the StreamGuard for the duration of the kernel
// launch — pattern matches the device_ptr_mut idiom
// used elsewhere in this file (e.g. the SP4 producer
// wire-ups). Binding `_guard` keeps cudarc's borrow
// bookkeeping alive across the launch and lets it drop
// naturally at end of block.
use cudarc::driver::DevicePtr;
let stream_ref = self.cuda_stream.as_deref()
.expect("cuda_stream must be Some when fused_ctx is Some");
let (feat_dev_ptr, _guard) = features_buf.device_ptr(stream_ref);
// Mapped pinned exposes dev_ptr directly (CUdeviceptr).
// No StreamGuard / device_ptr() borrow needed — these
// pages have no SyncOnDrop bookkeeping.
let feat_dev_ptr = features_buf.dev_ptr;
let market_dim_i32 = market_dim_usize as i32;
let bar_idx_i32 = (inferred_num_bars - 1) as i32;
if let Err(e) = fused.trainer().launch_sp5_pearl_8_trail(

37
docs/dqn-gpu-hot-path-audit.md
View File

@@ -277,3 +277,40 @@ Two related changes addressing the EVAL_DIST=Hold collapse observed in `train-mu
**Bug 2 — state[0]=raw_close suspected; instrumented for runtime confirmation.** Production runs show `aux_label_scale=5300` (raw price magnitude) but the source path traces back through `next_states_buf[i][0]` ← `experience_state_gather(market_features[bar_idx][0])` ← fxcache feat[0] (z-normalized, stddev=1.0). The discrepancy can't be resolved by static code reading alone — there must be a runtime path that injects raw_close into state[0] between the gather and the aux read. Added one-shot debug print in `training_loop.rs` after first `collect_experiences_gpu` return: downloads first 5 samples × 5 columns of both `gpu_batch.states` and `gpu_batch.next_states`, computes mean/std/mean_abs over col 0, prints under `target: "DIAG_BUG2"`. Triggers once via static `AtomicBool`. f32 dtoh download is slow but pre-graph-capture so no hot-path impact. Interpretation key embedded in the log line: `mean_abs ~0.001 = log_return ✓ | mean_abs ~5000 = raw price ✗`.
**Why the smoke test passes but production fails.** The smoke harness `multi_fold_convergence::test_multi_fold_convergence` runs the same DQN trainer but bypasses the production `train_baseline_rl` rollout machinery. The smoke at facbf76eb showed label_scale~0.1 (consistent with normalized log_return). The production binary at the same commit shows label_scale~5300 (raw price). One of the production-only code paths is mutating state[0] post-gather. Bug 2 instrumentation will pin the exact mutation site at next L40S run.
### Fix 22 — features_raw_cuda + targets_raw_cuda → MappedF32Buffer (Bug 2 instrumentation also lands) (2026-05-02)
Cross-file structural conversion that closes the last two `clone_to_device_f32_via_pinned` call sites in the DQN production path (training_loop.rs:1308 + 1312, the residue from Fix 16's pre-guard pattern). Mirrors the SP6 IQN τ buffer fix at `facbf76eb` (audit-tracked under task #283) and the SP4 portfolio_state pearl pattern: `MappedF32Buffer` field stored directly + `write_from_slice` for host load + kernel reads via `.dev_ptr`. No DtoD copy, no synchronize, graph-capture safe.
**What changed.**
- `trainers/dqn/trainer/mod.rs` — `targets_raw_cuda` and `features_raw_cuda` field type `Option<CudaSlice<f32>>` → `Option<MappedF32Buffer>`.
- `trainers/dqn/trainer/training_loop.rs::init_gpu_raw_buffers_from_slices` — replaces both `clone_to_device_f32_via_pinned` calls with `MappedF32Buffer::new(n)` + `write_from_slice`. Pre-existing `if self.cuda_stream.is_none() { return Ok(()) }` guard preserved (the `stream` value itself isn't required by the mapped-pinned allocator, only an active CUDA context, which the `cuda_stream-is-Some` invariant implies).
- `cuda_pipeline/gpu_experience_collector.rs` — three signature changes propagated through the consumer chain: `collect_experiences_gpu`, `launch_timestep_loop`, `compute_difficulty_scores`. Five `.arg(buf)` launch sites (hindsight_relabel, state_gather, expert_override, env_step×1, env_step×2) rewritten to `.arg(&buf.dev_ptr)`. `targets.raw_ptr()` in `compute_difficulty_scores` → `targets.dev_ptr` (same `CUdeviceptr` u64, different field name).
- `cuda_pipeline/decision_transformer.rs::build_dt_trajectories` — signature `&CudaSlice<f32>` → `&MappedF32Buffer` for both `features_gpu` and `targets_gpu`. Two `raw_ptr_f32(buf, stream)` call sites converted to `buf.dev_ptr` (mapped-pinned has no SyncOnDrop guard, so the `_no_drop` ManuallyDrop dance isn't needed). Doc-comment updated to reflect mapped-pinned semantics.
- `trainers/dqn/trainer/training_loop.rs:3749` (SP5 Pearl 8 trail launch) — `features_buf.len()` (method) → `features_buf.len` (pub field); `features_buf.device_ptr(stream)` (StreamGuard idiom) → `features_buf.dev_ptr` direct. Removed unused `DevicePtr` import + `_guard` binding.
**Bonus deliverable: Bug 2 diagnostic.** Same commit lands the one-shot DIAG_BUG2 download at training_loop.rs:1768 (right after the first `collect_experiences_gpu` return). Static `AtomicBool` ensures it fires exactly once per process. Per-sample first-5-col rows + col-0 mean/stddev/mean_abs aggregates printed under `target: "DIAG_BUG2"`. Interpretation key: `mean_abs ~0.001 = normalized log_return ✓ | mean_abs ~5000 = raw price ✗`. Uses `clone_dtoh` (current cudarc API; replaced the deprecated `memcpy_dtov`). DtoH is a graph-capture violation but this is single-shot pre-capture, so OK.
**Adjacent buffers in the same commit (carried-over from prior agent attempts):**
- `gpu_dqn_trainer.rs::sel_clip_buf` — single-element `CudaSlice<f32>` (sel-grad clip norm constant 1.0) → `MappedF32Buffer`. Aux Adam-update kernel reads `clip_ptr = self.sel_clip_buf.dev_ptr`. Eliminates the `upload_f32_via_pinned` constructor call.
- `gpu_weights.rs::RmsNormWeightSet` — four γ buffers (s0, s1, v, a; each is an all-ones init for RMSNorm γ scaling factor) `CudaSlice<f32>` → `MappedF32Buffer`. `KernelWeightPack::new` reads `rmsnorm.gamma_*.dev_ptr` directly (no `raw_device_ptr(...)` helper needed for mapped-pinned). Eliminates four `clone_to_device_f32_via_pinned` calls in `RmsNormWeightSet::ones`.
**PPO trainer deferred.** `trainers/ppo.rs:221,223` carry the same field types; PPO is not on the eval-collapse hot path and is out of scope. Pre-commit guard checks staged files only, so PPO's existing `*_via_pinned` violations don't block. Tracked as a follow-up.
**Validation.** `cargo check -p ml --offline` clean (zero warnings post-`memcpy_dtov` → `clone_dtoh` migration). `cargo build -p ml --release --offline` clean. Pre-commit `check_no_dtod_via_pinned` passes (no `*_via_pinned` calls in any staged file). After this commit, the only remaining `*_via_pinned` callers in the entire ml crate are in `trainers/ppo.rs` and a long tail of cold-path init sites tracked under task #290.
### Fix 23 — `gpu_dqn_trainer.rs` spectral-norm + aux-head ctor sites + `gpu_experience_collector.rs` OFI upload → MappedF32Buffer (10 sites, 2026-05-02)
Final structural sweep that closes the residual `*_via_pinned` ctor / cold-path init call sites in the two files still violating the pre-commit `check_no_dtod_via_pinned` guard after Fix 22. Same canonical pattern as the τ-buffer fix at facbf76eb and the Fix 22 features/targets conversion: `MappedF32Buffer` field stored directly + `MappedF32Buffer::new(n)` + `write_from_slice` for host load + kernel reads via `.dev_ptr` (or via `spectral_norm_descriptors` u64 table for the spec_u/spec_v cluster). No DtoD, no synchronize, graph-capture safe.
**What changed (per buffer-migration group).**
- **Spectral-norm `spec_u_s1` / `spec_v_s1` / `spec_u_s2` / `spec_v_s2` (4 buffers).** Trunk power-iteration vectors for `w_a_h_s1` / `w_a_h_s2` (Linear_a matrices in the GRN trunk). Field type `CudaSlice<f32>` → `MappedF32Buffer`. Constructor (`gpu_dqn_trainer.rs` ~14421-14431) replaces `alloc_f32` + `upload_f32_via_pinned` with `MappedF32Buffer::new` + `write_from_slice`. Consumed only via `.raw_ptr()` → `.dev_ptr` in the `host_desc[78]` table that backs `spectral_norm_descriptors` — same `CUdeviceptr` u64 width, purely a structural rename.
- **Spectral-norm head/branch macro `alloc_spec_pair!` (16 buffers).** Macro at ~14443 is invoked 11 times to allocate value-head, advantage, and 4 branch-output spec_u/spec_v pairs (`spec_u_v1/v_v1`, `spec_u_v2/v_v2`, `spec_u_a1/v_a1`, `spec_u_a2/v_a2`, `spec_u_bo1..bg2`, `spec_u_bn/v_bn`). Macro body now allocates `MappedF32Buffer::new` + `write_from_slice` directly (no `&mut $u_name` upload helper needed — interior mutability via `write_volatile`). Same `host_desc` consumer chain: each `.raw_ptr()` → `.dev_ptr`.
- **`graph_params` (60 floats, 1 buffer).** Cross-branch graph message-passing edge params (4 edges × 15 floats = 60). Field type `CudaSlice<f32>` → `MappedF32Buffer`. Constructor (~15790) `MappedF32Buffer::new(60)` + `write_from_slice(&graph_params_host)`. Single consumer at `launch_graph_message_pass` reads `params_ptr = self.graph_params.dev_ptr`.
- **`denoise_params` (DENOISE_TOTAL_PARAMS=1800, 1 buffer).** Diffusion Q-refinement MLP weights (2 steps × 900 = W1[576]+b1[24]+W2[288]+b2[12]). Field type `CudaSlice<f32>` → `MappedF32Buffer`. Constructor (~15845) Xavier init host-side then `write_from_slice`. Four consumer sites: `launch_q_denoise`, `launch_q_denoise_backward`, `launch_q_denoise_backward_cublas`, aux Adam update — all read `params_ptr = self.denoise_params.dev_ptr`. Notable: the kernel writes back to this buffer during aux Adam updates; mapped-pinned semantics are kernel-write-coherent the same way as kernel-read (single mapping, both directions through `dev_ptr`).
- **`qlstm_weights` (QLSTM_WEIGHT_COUNT=528, 1 buffer).** xLSTM mLSTM-cell parameters (6 weight matrices × [11, 8] = 528). Field type `CudaSlice<f32>` → `MappedF32Buffer`. Constructor (~15992) Xavier uniform host-side then `write_from_slice`. Two consumer sites: `launch_qlstm_step`, `launch_qlstm_train_step` — both read `w_ptr = self.qlstm_weights.dev_ptr`. Kernel SGD-updates in place via the same dev_ptr mapping.
- **`gpu_experience_collector.rs::ofi_gpu` (1 buffer + 1 placeholder).** Pre-uploaded OFI features (`total_bars * OFI_DIM`). Field type `CudaSlice<f32>` → `MappedF32Buffer`. Constructor places a 1-element zero placeholder (`MappedF32Buffer::new(1)`); `upload_ofi_features` rewrites the field with a fresh `MappedF32Buffer` sized to `ofi_flat.len()`, then `write_from_slice(ofi_flat)`. Replaces the `clone_to_device_f32_via_pinned` call. The single per-step consumer at `launch_timestep_loop`'s state_gather invocation reads `ofi_ptr = self.ofi_gpu.dev_ptr`.
**Hot-path purity.** None of the converted buffers are written per-step from the host; `denoise_params` and `qlstm_weights` are mutated by kernel-side aux-Adam / SGD updates which use the same `dev_ptr` mapping the read-side does. `graph_params` and the spec_u/spec_v cluster are static post-init (graph_params is intentionally not trained; spec vectors are updated by the spectral-norm power-iteration kernel via the descriptor table, again through `dev_ptr`). `ofi_gpu` is rewritten exactly once (cold path) via `upload_ofi_features` — re-allocating the full mapped-pinned buffer and replacing the field on each call is fine because the call frequency is one-per-fold, never per-step.
**Validation.** `cargo check -p ml --offline` clean. Pre-commit `check_no_dtod_via_pinned` passes — zero `upload_*_via_pinned` / `clone_to_device_*_via_pinned` callers in either of the two staged files. The Bug 2 instrumentation commit currently in flight (Fix 21 / Fix 22 chain) can now proceed past the pre-commit guard.
**Out of scope.** `mapped_pinned.rs` retains the `upload_f32_via_pinned` / `upload_i32_via_pinned` / `clone_to_device_f32_via_pinned` / `clone_to_device_i32_via_pinned` helper definitions — `trainers/ppo.rs` and a small set of remaining cold-path callers still depend on them; the helpers are deleted under task #290 once the call count globally hits zero. Per `feedback_no_hiding` no suppression markers were added.