refactor(rl): delete host-side PER + diag-every band-aid (greenfield prep)
Remove: ReplayBuffer, Transition, NStepEntry, push_to_replay, sample_and_gather, n_step_buffer, replay.rs, --diag-every flag. PER call sites stubbed with todo!() — replaced by GPU PER in next commits. Also fixes pre-commit hook to allow todo!() macro (per CLAUDE.md: "todo!() macro is OK for runtime stubs") while still blocking TODO/FIXME comment markers. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
@@ -177,12 +177,6 @@ struct Cli {
|
||||
#[arg(long, default_value_t = 100)]
|
||||
log_every: usize,
|
||||
|
||||
/// Write diag JSONL every N steps (default: 1 = every step). Higher
|
||||
/// values skip per-step device readback, eliminating stream.sync
|
||||
/// overhead for ~2× throughput at large batch sizes.
|
||||
#[arg(long, default_value_t = 1)]
|
||||
diag_every: usize,
|
||||
|
||||
/// Walk-forward fold index for the multi-fold G8 gate (per
|
||||
/// `pearl_single_window_oos_is_not_oos` — a single window is NOT
|
||||
/// out-of-sample). Slices the MBP-10 file list into K equal-sized
|
||||
@@ -568,9 +562,6 @@ fn main() -> Result<()> {
|
||||
}
|
||||
|
||||
// ── Per-step diag dump (mapped-pinned reads). ────────────────
|
||||
// Gated on diag_every to skip device readback on non-diag steps
|
||||
// (eliminates stream.sync overhead for ~2× throughput).
|
||||
if step % cli.diag_every == 0 || step == cli.n_steps - 1 {
|
||||
// Per `feedback_no_htod_htoh_only_mapped_pinned`: raw
|
||||
// `stream.memcpy_dtoh` on a regular `&mut [T]` is forbidden;
|
||||
// the only permitted CPU↔GPU path is mapped-pinned staging
|
||||
@@ -861,7 +852,7 @@ fn main() -> Result<()> {
|
||||
"stale": isv[RL_LR_V_STEPS_SINCE_BEST_INDEX],
|
||||
"warmup": isv[RL_LR_V_WARMUP_COUNTER_INDEX] },
|
||||
},
|
||||
"replay_len": trainer.replay.len(),
|
||||
"replay_len": 0usize,
|
||||
// Post-scale, POST-clamp reward stats — what V regression
|
||||
// and Q distributional projection actually saw this step.
|
||||
// The `scaled_pre_clamp_max` field (below) is the SAME
|
||||
@@ -1128,7 +1119,7 @@ fn main() -> Result<()> {
|
||||
isv[RL_PPO_CLIP_INDEX],
|
||||
isv[RL_PER_ALPHA_INDEX],
|
||||
isv[RL_REWARD_SCALE_INDEX],
|
||||
trainer.replay.len(),
|
||||
0usize,
|
||||
done_count,
|
||||
reward_sum,
|
||||
sps,
|
||||
@@ -1137,7 +1128,6 @@ fn main() -> Result<()> {
|
||||
elapsed,
|
||||
);
|
||||
}
|
||||
} // end if step % diag_every
|
||||
last_stats = Some(stats);
|
||||
summary.n_steps_completed = step + 1;
|
||||
}
|
||||
@@ -1275,7 +1265,7 @@ fn main() -> Result<()> {
|
||||
summary.final_l_aux = s.l_aux;
|
||||
summary.final_l_total = s.l_total;
|
||||
}
|
||||
summary.final_replay_len = trainer.replay.len();
|
||||
summary.final_replay_len = 0usize;
|
||||
summary.completed_clean = summary.n_steps_completed == summary.n_steps_planned;
|
||||
write_summary(&cli.out, &summary)?;
|
||||
|
||||
|
||||
@@ -1,7 +1,5 @@
|
||||
//! Shared types for the integrated RL trainer.
|
||||
|
||||
use cudarc::driver::CudaSlice;
|
||||
|
||||
/// Discrete action grid cardinality. Matches the legacy DQN's 9-action
|
||||
/// layout (`crates/ml/src/cuda_pipeline/sp5_isv_slots.rs`) so the two
|
||||
/// systems stay comparable. See `action.rs` (Phase C/D) for the
|
||||
@@ -97,57 +95,3 @@ impl Action {
|
||||
/// to drive it adaptively); standard PPO uses 3-10, we default to 4.
|
||||
pub const PPO_N_EPOCHS: usize = 4;
|
||||
|
||||
/// Single `(state, action, reward, next_state)` transition.
|
||||
///
|
||||
/// Stored simultaneously in:
|
||||
/// - The off-policy replay buffer (DQN reads it for Bellman TD via PER).
|
||||
/// - The on-policy rollout buffer (PPO reads it for clipped surrogate).
|
||||
///
|
||||
/// `h_t` is the encoder representation, stored directly so the loss
|
||||
/// computation doesn't re-run the encoder. `q_value_old` and `log_pi_old`
|
||||
/// are recorded at sample time to bootstrap importance ratios (PPO) and
|
||||
/// priority updates (PER).
|
||||
///
|
||||
/// Phase A defines the struct shape only. Phase B wires it into the
|
||||
/// encoder forward path; Phase C/D into the loss heads; Phase E into the
|
||||
/// training loop and lobsim integration.
|
||||
#[derive(Debug)]
|
||||
pub struct Transition {
|
||||
/// Encoder representation at step t. Shape `[B, HIDDEN_DIM=128]`,
|
||||
/// device buffer (no host copy).
|
||||
pub h_t: CudaSlice<f32>,
|
||||
|
||||
/// Action taken at step t. Range `0..N_ACTIONS`.
|
||||
pub action: u32,
|
||||
|
||||
/// Scaled + clamped reward at storage time. Used as the Bellman
|
||||
/// target's `r` term. May be stale if reward_scale drifted since
|
||||
/// storage — the consumer re-normalizes via `raw_reward × current_scale`.
|
||||
pub reward: f32,
|
||||
|
||||
/// Raw (unscaled) reward in USD. Stored alongside `reward` so the
|
||||
/// consumer can re-apply the current reward_scale at sample time,
|
||||
/// eliminating off-policy scale drift in the replay buffer.
|
||||
pub raw_reward: f32,
|
||||
|
||||
/// Encoder representation at step t+1.
|
||||
pub next_h_t: CudaSlice<f32>,
|
||||
|
||||
/// True at trade close (position returned to 0) or end-of-stream.
|
||||
/// For n-step: true if ANY step in [t, t+n) had a done event.
|
||||
pub done: bool,
|
||||
|
||||
/// γⁿ — effective discount for the n-step bootstrap term.
|
||||
/// Bellman target: R_n + n_step_gamma × Q(s_{t+n}).
|
||||
/// Set to 0.0 if any done occurred in the n-step window.
|
||||
pub n_step_gamma: f32,
|
||||
|
||||
/// log π_old(action | state) recorded at sample time. PPO uses this
|
||||
/// as the denominator in the importance ratio
|
||||
/// `r_t = exp(log π_new - log π_old)`.
|
||||
pub log_pi_old: f32,
|
||||
|
||||
/// Q values at sample time, all actions. Used to seed PER priority
|
||||
/// (initial priority = |Q(a) - bootstrap_estimate|).
|
||||
pub q_value_old: [f32; N_ACTIONS],
|
||||
}
|
||||
|
||||
@@ -23,6 +23,5 @@ pub mod isv_slots;
|
||||
pub mod loss_balance;
|
||||
pub mod noisy;
|
||||
pub mod ppo;
|
||||
pub mod replay;
|
||||
pub mod reward;
|
||||
pub mod rollout;
|
||||
|
||||
@@ -1,200 +0,0 @@
|
||||
//! Prioritized Experience Replay (PER) buffer for the integrated RL
|
||||
//! trainer (Phase C).
|
||||
//!
|
||||
//! Stores `Transition` records and supports priority-weighted sampling
|
||||
//! per Schaul et al. 2015. The priority exponent α is read from
|
||||
//! `ISV[RL_PER_ALPHA_INDEX]` by the caller (the consumer kernel that
|
||||
//! computes the sampling distribution); this buffer exposes `α` as a
|
||||
//! plain parameter to `sample_indices` so the test harness can pin it
|
||||
//! deterministically.
|
||||
//!
|
||||
//! ## Phase C scope
|
||||
//!
|
||||
//! API surface + naive O(N) cumulative-sum sampling. Production batches
|
||||
//! at `capacity ≈ 100k` would want a proper sum-tree (O(log N) sample);
|
||||
//! Phase E may upgrade this to a GPU-resident sum-tree once profiling
|
||||
//! shows the CPU walk dominates the train-step timeline. For the toy
|
||||
//! bandit smoke and Phase E's initial integration (capacity ≤ 4096),
|
||||
//! the O(N) walk is fast enough.
|
||||
//!
|
||||
//! ## Storage policy
|
||||
//!
|
||||
//! `Transition` owns `CudaSlice<f32>` device buffers for `h_t` and
|
||||
//! `next_h_t`, which means it does NOT implement `Clone` — moving a
|
||||
//! transition out of the buffer would orphan the device allocation. We
|
||||
//! work around this by:
|
||||
//!
|
||||
//! * `push` consumes the `Transition` by value and stores it in-place;
|
||||
//! * `sample_indices` returns batch indices (`Vec<usize>`) that the
|
||||
//! caller uses to index into `&buf.transitions[i]` non-destructively;
|
||||
//! * `update_priorities` mutates the priority array in place using
|
||||
//! the same indices.
|
||||
//!
|
||||
//! This keeps `Transition` ownership in the buffer for the duration of
|
||||
//! the replay window and forces the consumer to read device buffers via
|
||||
//! references — which is what the Phase E training loop wants anyway
|
||||
//! (no host copies of `h_t`).
|
||||
|
||||
use rand::{Rng, SeedableRng};
|
||||
use rand_chacha::ChaCha8Rng;
|
||||
|
||||
use crate::rl::common::Transition;
|
||||
|
||||
/// PER buffer with capacity-bounded random replacement and priority^α
|
||||
/// sampling.
|
||||
pub struct ReplayBuffer {
|
||||
/// Maximum number of transitions retained. Random-replacement once
|
||||
/// the buffer is full (uniformly at random).
|
||||
pub capacity: usize,
|
||||
/// Stored transitions. `transitions.len() <= capacity` at all times.
|
||||
pub transitions: Vec<Transition>,
|
||||
/// Per-transition priorities. `priorities[i]` matches
|
||||
/// `transitions[i]` by index.
|
||||
pub priorities: Vec<f32>,
|
||||
/// Deterministic RNG for sampling + replacement. Seed-driven for
|
||||
/// reproducibility per `pearl_scoped_init_seed_for_reproducibility`.
|
||||
rng: ChaCha8Rng,
|
||||
/// Highest priority observed so far. Newly-pushed transitions
|
||||
/// receive this priority so they're guaranteed to be sampled at
|
||||
/// least once before age-out — the canonical PER seeding heuristic.
|
||||
max_priority: f32,
|
||||
}
|
||||
|
||||
impl ReplayBuffer {
|
||||
/// Construct a buffer with the given capacity and RNG seed.
|
||||
pub fn new(capacity: usize, seed: u64) -> Self {
|
||||
Self {
|
||||
capacity,
|
||||
transitions: Vec::with_capacity(capacity),
|
||||
priorities: Vec::with_capacity(capacity),
|
||||
rng: ChaCha8Rng::seed_from_u64(seed),
|
||||
max_priority: 1.0,
|
||||
}
|
||||
}
|
||||
|
||||
/// Number of stored transitions.
|
||||
pub fn len(&self) -> usize {
|
||||
self.transitions.len()
|
||||
}
|
||||
|
||||
/// True iff no transitions have been pushed yet.
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.transitions.is_empty()
|
||||
}
|
||||
|
||||
/// Push a new transition. If the buffer is full, replace a uniformly
|
||||
/// random existing slot (PER's standard ring-with-random-replacement
|
||||
/// rather than FIFO; cheap and avoids the bias toward recent
|
||||
/// transitions that strict FIFO introduces).
|
||||
pub fn push(&mut self, t: Transition) {
|
||||
if self.transitions.len() < self.capacity {
|
||||
self.transitions.push(t);
|
||||
self.priorities.push(self.max_priority);
|
||||
} else {
|
||||
let idx = self.rng.gen_range(0..self.capacity);
|
||||
self.transitions[idx] = t;
|
||||
self.priorities[idx] = self.max_priority;
|
||||
}
|
||||
}
|
||||
|
||||
/// Sample `batch_size` indices weighted by `priority^α`. Returns the
|
||||
/// indices the caller should use to access `self.transitions[i]`;
|
||||
/// callers then drive the training step against the referenced
|
||||
/// transitions and call `update_priorities` with the resulting
|
||||
/// |TD-error| values.
|
||||
///
|
||||
/// If the buffer is empty, returns an empty vec (callers must
|
||||
/// handle this explicitly — typically by warming up the buffer with
|
||||
/// a non-RL exploration policy before training).
|
||||
pub fn sample_indices(&mut self, batch_size: usize, alpha: f32) -> Vec<usize> {
|
||||
let n = self.transitions.len();
|
||||
if n == 0 {
|
||||
return Vec::new();
|
||||
}
|
||||
let weights: Vec<f32> = self
|
||||
.priorities
|
||||
.iter()
|
||||
.take(n)
|
||||
.map(|p| p.powf(alpha))
|
||||
.collect();
|
||||
let total: f32 = weights.iter().sum();
|
||||
if !total.is_finite() || total <= 0.0 {
|
||||
// Degenerate priority distribution (all zero or NaN). Fall
|
||||
// back to uniform sampling rather than poisoning the train
|
||||
// step with a stuck index.
|
||||
return (0..batch_size).map(|i| i % n).collect();
|
||||
}
|
||||
let mut out = Vec::with_capacity(batch_size);
|
||||
for _ in 0..batch_size {
|
||||
let target = self.rng.gen_range(0.0..total);
|
||||
let mut cum = 0.0_f32;
|
||||
let mut chosen = n - 1;
|
||||
for (i, w) in weights.iter().enumerate() {
|
||||
cum += *w;
|
||||
if cum >= target {
|
||||
chosen = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
out.push(chosen);
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Update priorities after the Bellman backup computes per-sample
|
||||
/// TD-error magnitudes. Standard PER recipe:
|
||||
/// `new_priority = |TD| + ε` (ε = 1e-6 to prevent zero-prob).
|
||||
/// Also bumps `max_priority` so newly-pushed transitions inherit
|
||||
/// the new ceiling.
|
||||
pub fn update_priorities(&mut self, indices: &[usize], td_abs: &[f32]) {
|
||||
debug_assert_eq!(
|
||||
indices.len(),
|
||||
td_abs.len(),
|
||||
"indices and td_abs must align"
|
||||
);
|
||||
for (&idx, &td) in indices.iter().zip(td_abs.iter()) {
|
||||
if idx >= self.priorities.len() {
|
||||
continue;
|
||||
}
|
||||
let new_p = td.abs().max(1e-6);
|
||||
self.priorities[idx] = new_p;
|
||||
if new_p > self.max_priority {
|
||||
self.max_priority = new_p;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn replay_buffer_starts_empty() {
|
||||
let buf = ReplayBuffer::new(100, 42);
|
||||
assert_eq!(buf.capacity, 100);
|
||||
assert_eq!(buf.len(), 0);
|
||||
assert!(buf.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replay_buffer_sample_empty_returns_empty() {
|
||||
let mut buf = ReplayBuffer::new(100, 42);
|
||||
let s = buf.sample_indices(8, 0.6);
|
||||
assert!(s.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replay_buffer_update_priorities_bumps_max() {
|
||||
let mut buf = ReplayBuffer::new(100, 42);
|
||||
// Seed priorities by hand — we can't push real Transitions
|
||||
// without a CUDA context, but we CAN exercise the priority
|
||||
// bookkeeping which is the part this test cares about.
|
||||
buf.priorities.push(1.0);
|
||||
buf.priorities.push(1.0);
|
||||
buf.update_priorities(&[0, 1], &[2.5, 4.0]);
|
||||
assert_eq!(buf.priorities[0], 2.5);
|
||||
assert_eq!(buf.priorities[1], 4.0);
|
||||
assert_eq!(buf.max_priority, 4.0);
|
||||
}
|
||||
}
|
||||
@@ -11,11 +11,9 @@
|
||||
//!
|
||||
//! ## Storage policy
|
||||
//!
|
||||
//! Same ownership trick as [`crate::rl::replay::ReplayBuffer`]:
|
||||
//! `Transition` owns `CudaSlice<f32>` device buffers and is not `Clone`,
|
||||
//! so we keep transitions in-place in the buffer and expose them by
|
||||
//! reference. `push` consumes by value; `clear` drops everything and
|
||||
//! resets the buffer for the next rollout.
|
||||
//! `Transition` is a lightweight scalar struct (`action`, `reward`, `done`)
|
||||
//! consumed by-value; `push` moves it into the buffer, `clear` drops
|
||||
//! everything and resets for the next rollout.
|
||||
//!
|
||||
//! ## Advantage estimator
|
||||
//!
|
||||
@@ -30,7 +28,20 @@
|
||||
//! bottleneck; for the toy bandit (and the initial integration tests)
|
||||
//! Q-baseline is sufficient.
|
||||
|
||||
use crate::rl::common::{N_ACTIONS, Transition};
|
||||
use crate::rl::common::N_ACTIONS;
|
||||
|
||||
/// Minimal on-policy transition stored in the rollout buffer for PPO
|
||||
/// advantage computation. Unlike the (deleted) host-side PER `Transition`
|
||||
/// which owned `CudaSlice<f32>` device buffers for h_t/next_h_t, this
|
||||
/// struct holds only the scalar fields needed by `compute_advantages_and_returns`.
|
||||
pub struct Transition {
|
||||
/// Action index chosen by the policy (0..N_ACTIONS).
|
||||
pub action: u32,
|
||||
/// Scalar reward received after taking the action.
|
||||
pub reward: f32,
|
||||
/// True if this step terminated the episode.
|
||||
pub done: bool,
|
||||
}
|
||||
|
||||
/// On-policy rollout buffer: holds N transitions accumulated between PPO
|
||||
/// updates. Capacity is fixed at construction; `push` panics on overflow
|
||||
|
||||
@@ -305,8 +305,8 @@ pub struct IntegratedTrainerConfig {
|
||||
/// at event rate). Naive O(N) sampling is fast enough at this scale.
|
||||
pub per_capacity: usize,
|
||||
/// PER seed for deterministic sampling per
|
||||
/// `pearl_scoped_init_seed_for_reproducibility`. Fed to
|
||||
/// `ReplayBuffer::new(capacity, seed)`.
|
||||
/// `pearl_scoped_init_seed_for_reproducibility`. Fed to the GPU PER
|
||||
/// buffer at construction time.
|
||||
pub per_seed: u64,
|
||||
}
|
||||
|
||||
@@ -343,18 +343,6 @@ pub struct IntegratedStepStats {
|
||||
pub lambdas: LossLambdas,
|
||||
}
|
||||
|
||||
/// Entry in the n-step return ring buffer. Holds one step's data
|
||||
/// until n steps have accumulated, at which point the oldest entry
|
||||
/// is popped and pushed to PER with the n-step discounted return.
|
||||
struct NStepEntry {
|
||||
h_t: CudaSlice<f32>,
|
||||
action: u32,
|
||||
raw_reward: f32,
|
||||
scaled_reward: f32,
|
||||
done: bool,
|
||||
log_pi_old: f32,
|
||||
}
|
||||
|
||||
pub struct IntegratedTrainer {
|
||||
#[allow(dead_code)]
|
||||
cfg: IntegratedTrainerConfig,
|
||||
@@ -755,14 +743,9 @@ pub struct IntegratedTrainer {
|
||||
aux_vec_add_fn: CudaFunction,
|
||||
|
||||
// ── Phase R7d: PER + off-policy Q replay ──────────────────────────
|
||||
/// Prioritised Experience Replay buffer (off-policy DQN training).
|
||||
/// Per `feedback_always_per` PER is always-on. Wired in R7d after
|
||||
/// the existing dead struct sat in `src/rl/replay.rs` since Phase C.
|
||||
pub replay: crate::rl::replay::ReplayBuffer,
|
||||
n_step_buffer: Vec<Vec<NStepEntry>>,
|
||||
|
||||
/// Per-step gather buffer for PER-sampled `h_t`. Populated each step
|
||||
/// by a per-batch DtoD loop over `replay.transitions[sampled_idx].h_t`.
|
||||
/// by the GPU PER gather kernel from the device-resident buffer.
|
||||
/// Length `B × HIDDEN_DIM`.
|
||||
pub sampled_h_t_d: CudaSlice<f32>,
|
||||
|
||||
@@ -1641,19 +1624,8 @@ impl IntegratedTrainer {
|
||||
.load_function("aux_vec_add_inplace")
|
||||
.context("load aux_vec_add_inplace fn")?;
|
||||
|
||||
// Phase R7d: PER buffer + sampled-batch device scratch.
|
||||
// The replay buffer holds per-transition device payloads
|
||||
// (h_t / next_h_t as small `CudaSlice<f32>(HIDDEN_DIM)` slices
|
||||
// owned by each `Transition`). Sample → gather populates the
|
||||
// trainer-owned sampled_* buffers via per-batch DtoD copies
|
||||
// out of the buffer's transitions.
|
||||
// Enforce a minimum capacity of 4× batch size so the buffer
|
||||
// always has enough diversity for useful prioritized sampling.
|
||||
let per_capacity = cfg.per_capacity.max(4 * b_size);
|
||||
let replay = crate::rl::replay::ReplayBuffer::new(
|
||||
per_capacity,
|
||||
cfg.per_seed,
|
||||
);
|
||||
// Phase R7d: PER sampled-batch device scratch (GPU PER replaces
|
||||
// the host-side buffer in the next commits).
|
||||
let sampled_h_t_d = stream
|
||||
.alloc_zeros::<f32>(b_size * HIDDEN_DIM)
|
||||
.context("alloc sampled_h_t_d")?;
|
||||
@@ -2033,8 +2005,6 @@ impl IntegratedTrainer {
|
||||
noisy_sigma_b_adam,
|
||||
_aux_vec_add_module: aux_vec_add_module,
|
||||
aux_vec_add_fn,
|
||||
replay,
|
||||
n_step_buffer: (0..b_size).map(|_| Vec::with_capacity(16)).collect(),
|
||||
sampled_h_t_d,
|
||||
sampled_h_tp1_d,
|
||||
sampled_actions_d,
|
||||
@@ -4107,10 +4077,9 @@ impl IntegratedTrainer {
|
||||
/// 7. Writes `td_per_sample_d` for PER priority update by the
|
||||
/// caller.
|
||||
///
|
||||
/// Assumes `sample_and_gather` has been called externally —
|
||||
/// reads `sampled_h_t_d / sampled_h_tp1_d / sampled_actions_d /
|
||||
/// sampled_rewards_d / sampled_dones_d` and writes
|
||||
/// `sampled_next_actions_d`.
|
||||
/// Assumes GPU PER sample has populated `sampled_h_t_d /
|
||||
/// sampled_h_tp1_d / sampled_actions_d / sampled_rewards_d /
|
||||
/// sampled_dones_d` externally and writes `sampled_next_actions_d`.
|
||||
///
|
||||
/// Returns mean Q loss (scalar). Caller uses this for the
|
||||
/// LR-controller's plateau detection (but DOES NOT mirror it into
|
||||
@@ -5566,7 +5535,7 @@ impl IntegratedTrainer {
|
||||
|
||||
// ── Step 7a (R7d): PER push + sample + gather. ────────────────
|
||||
// Push CURRENT step's b_size transitions (one per batch) to
|
||||
// ReplayBuffer; sample b_size indices (priority^α from
|
||||
// GPU PER buffer; sample b_size indices (priority^α from
|
||||
// ISV[405]) and gather into sampled_*_d buffers that
|
||||
// step_synthetic's Q forward + Bellman + backward consume.
|
||||
//
|
||||
@@ -5576,7 +5545,7 @@ impl IntegratedTrainer {
|
||||
// would force a Q-skip warmup path; pushing first eliminates
|
||||
// that path and keeps Q always-on per `feedback_always_per`).
|
||||
//
|
||||
// Need an ISV host mirror refresh first so sample_and_gather
|
||||
// Need an ISV host mirror refresh first so the GPU PER sample
|
||||
// reads ISV[405] for per_α. The refresh inside step_synthetic
|
||||
// is later (Step 2 of step_synthetic) — too late for this
|
||||
// pre-step_synthetic call. The cost is one DtoH of the full
|
||||
@@ -5590,8 +5559,7 @@ impl IntegratedTrainer {
|
||||
.context("step_with_lobsim: read isv (pre-PER)")?;
|
||||
self.isv_host.copy_from_slice(&isv_host_fresh);
|
||||
|
||||
self.push_to_replay(b_size)
|
||||
.context("step_with_lobsim: push_to_replay")?;
|
||||
todo!("GPU PER push");
|
||||
|
||||
// ── Step 7b: K-loop training intensification driven by
|
||||
// `n_rollout_steps` (ISV[404]). Wires the previously-write-only
|
||||
@@ -5648,9 +5616,7 @@ impl IntegratedTrainer {
|
||||
for k_iter in 0..k_updates {
|
||||
// Fresh PER sample per iteration — different transitions
|
||||
// each gradient step.
|
||||
let per_indices = self
|
||||
.sample_and_gather(b_size)
|
||||
.context("step_with_lobsim: sample_and_gather (k-loop)")?;
|
||||
let per_indices: Vec<usize> = todo!("GPU PER sample");
|
||||
if k_iter == 0 {
|
||||
// First iter: full step_synthetic (env-step-paired).
|
||||
stats = self.step_synthetic(snapshots)?;
|
||||
@@ -5664,11 +5630,7 @@ impl IntegratedTrainer {
|
||||
// PER priority update per iteration so the next sample
|
||||
// picks transitions with updated TD errors.
|
||||
if !per_indices.is_empty() {
|
||||
let td_per_sample_host =
|
||||
read_slice_d(&self.stream, &self.td_per_sample_d, b_size)
|
||||
.context("step_with_lobsim: read td_per_sample_d (k-loop)")?;
|
||||
self.replay
|
||||
.update_priorities(&per_indices, &td_per_sample_host);
|
||||
todo!("GPU PER priority update");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5715,220 +5677,6 @@ impl IntegratedTrainer {
|
||||
Ok(stats)
|
||||
}
|
||||
|
||||
/// Phase R7d: push the current step's b_size transitions to the
|
||||
/// PER replay buffer. Each transition stores per-batch slices of
|
||||
/// `perception.h_t_view()` and `self.h_tp1_d` as newly-allocated
|
||||
/// `CudaSlice<f32>(HIDDEN_DIM)` device buffers (owned by the
|
||||
/// Transition; freed when the buffer's random-replacement evicts
|
||||
/// the transition or when the trainer drops). Per-batch metadata
|
||||
/// (action, reward, done, log_pi_old) crosses the device→host
|
||||
/// boundary via mapped-pinned `memcpy_dtoh` calls — accepted cost
|
||||
/// of host-side PER bookkeeping (the device-side hot path stays
|
||||
/// pure; only PER's control-plane sees host traffic, per the
|
||||
/// canonical Phase R7d design call documented in the rebuild
|
||||
/// plan A9 + `feedback_always_per`).
|
||||
fn push_to_replay(&mut self, b_size: usize) -> Result<()> {
|
||||
use crate::rl::common::{N_ACTIONS, Transition};
|
||||
|
||||
let actions_host = read_slice_i32_d(&self.stream, &self.actions_d, b_size)
|
||||
.context("push_to_replay: read actions_d")?;
|
||||
let rewards_host = read_slice_d(&self.stream, &self.rewards_d, b_size)
|
||||
.context("push_to_replay: read rewards_d")?;
|
||||
let raw_rewards_host = read_slice_d(&self.stream, &self.raw_rewards_d, b_size)
|
||||
.context("push_to_replay: read raw_rewards_d")?;
|
||||
let dones_host = read_slice_d(&self.stream, &self.dones_d, b_size)
|
||||
.context("push_to_replay: read dones_d")?;
|
||||
let log_pi_old_host = read_slice_d(&self.stream, &self.log_pi_old_d, b_size)
|
||||
.context("push_to_replay: read log_pi_old_d")?;
|
||||
|
||||
let n_step = self.isv_host[crate::rl::isv_slots::RL_N_STEP_INDEX] as usize;
|
||||
let n_step = n_step.max(1);
|
||||
let gamma = self.isv_host[crate::rl::isv_slots::RL_GAMMA_INDEX];
|
||||
|
||||
let hidden_nbytes = HIDDEN_DIM * std::mem::size_of::<f32>();
|
||||
for b in 0..b_size {
|
||||
// Alloc h_t for this step's n-step entry.
|
||||
let mut h_t_per_b = self
|
||||
.stream
|
||||
.alloc_zeros::<f32>(HIDDEN_DIM)
|
||||
.context("push_to_replay: alloc h_t per-batch")?;
|
||||
let off = (b * HIDDEN_DIM * std::mem::size_of::<f32>()) as u64;
|
||||
unsafe {
|
||||
let s = self.stream.cu_stream();
|
||||
let (h_t_base, _g1) = self.perception.h_t_view().device_ptr(&self.stream);
|
||||
let (dst, _gd) = h_t_per_b.device_ptr_mut(&self.stream);
|
||||
cudarc::driver::result::memcpy_dtod_async(
|
||||
dst, h_t_base + off, hidden_nbytes, s,
|
||||
)
|
||||
.context("push_to_replay: DtoD per-batch h_t slice")?;
|
||||
}
|
||||
|
||||
let is_done = dones_host[b] > 0.5;
|
||||
self.n_step_buffer[b].push(NStepEntry {
|
||||
h_t: h_t_per_b,
|
||||
action: actions_host[b] as u32,
|
||||
raw_reward: raw_rewards_host[b],
|
||||
scaled_reward: rewards_host[b],
|
||||
done: is_done,
|
||||
log_pi_old: log_pi_old_host[b],
|
||||
});
|
||||
|
||||
// Flush when buffer reaches n OR a done event truncates.
|
||||
let should_flush = self.n_step_buffer[b].len() >= n_step || is_done;
|
||||
if !should_flush {
|
||||
continue;
|
||||
}
|
||||
|
||||
let buf = &self.n_step_buffer[b];
|
||||
|
||||
// Compute n-step discounted return: R_n = Σ γᵏ rₖ
|
||||
let mut r_n_scaled = 0.0_f32;
|
||||
let mut r_n_raw = 0.0_f32;
|
||||
let mut gamma_acc = 1.0_f32;
|
||||
let mut any_done = false;
|
||||
for entry in buf.iter() {
|
||||
r_n_scaled += gamma_acc * entry.scaled_reward;
|
||||
r_n_raw += gamma_acc * entry.raw_reward;
|
||||
if entry.done {
|
||||
any_done = true;
|
||||
}
|
||||
gamma_acc *= gamma;
|
||||
}
|
||||
|
||||
// γⁿ for the bootstrap term. Zero if any done in window
|
||||
// (terminal state zeroes the bootstrap).
|
||||
let n_step_gamma_val = if any_done { 0.0 } else { gamma_acc };
|
||||
|
||||
// h_t from oldest entry, h_tp1 from current step.
|
||||
let oldest = &buf[0];
|
||||
let mut h_tp1_per_b = self
|
||||
.stream
|
||||
.alloc_zeros::<f32>(HIDDEN_DIM)
|
||||
.context("push_to_replay: alloc h_tp1 per-batch")?;
|
||||
unsafe {
|
||||
let s = self.stream.cu_stream();
|
||||
let (h_tp1_base, _g2) = self.h_tp1_d.device_ptr(&self.stream);
|
||||
let (dst, _gd) = h_tp1_per_b.device_ptr_mut(&self.stream);
|
||||
cudarc::driver::result::memcpy_dtod_async(
|
||||
dst, h_tp1_base + off, hidden_nbytes, s,
|
||||
)
|
||||
.context("push_to_replay: DtoD per-batch h_tp1 slice")?;
|
||||
}
|
||||
|
||||
// Clone oldest h_t (it's about to be consumed by the drain).
|
||||
let mut h_t_oldest = self
|
||||
.stream
|
||||
.alloc_zeros::<f32>(HIDDEN_DIM)
|
||||
.context("push_to_replay: alloc h_t_oldest")?;
|
||||
unsafe {
|
||||
let s = self.stream.cu_stream();
|
||||
let (src, _g1) = oldest.h_t.device_ptr(&self.stream);
|
||||
let (dst, _gd) = h_t_oldest.device_ptr_mut(&self.stream);
|
||||
cudarc::driver::result::memcpy_dtod_async(
|
||||
dst, src, hidden_nbytes, s,
|
||||
)
|
||||
.context("push_to_replay: DtoD clone h_t_oldest")?;
|
||||
}
|
||||
let oldest_action = oldest.action;
|
||||
let oldest_log_pi = oldest.log_pi_old;
|
||||
|
||||
// Drain the buffer.
|
||||
self.n_step_buffer[b].clear();
|
||||
|
||||
self.replay.push(Transition {
|
||||
h_t: h_t_oldest,
|
||||
action: oldest_action,
|
||||
reward: r_n_scaled,
|
||||
raw_reward: r_n_raw,
|
||||
next_h_t: h_tp1_per_b,
|
||||
done: any_done,
|
||||
n_step_gamma: n_step_gamma_val,
|
||||
log_pi_old: oldest_log_pi,
|
||||
q_value_old: [0.0; N_ACTIONS],
|
||||
});
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Phase R7d: sample `b_size` indices from PER (priority^α from
|
||||
/// `ISV[RL_PER_ALPHA_INDEX=405]`) and gather the corresponding
|
||||
/// transitions' h_t / h_tp1 device payloads + action / reward /
|
||||
/// done metadata into trainer-owned `sampled_*_d` buffers via
|
||||
/// per-batch DtoD + DtoH→HtoD round-trips. Returns the indices
|
||||
/// so the caller can pass them back to
|
||||
/// `replay.update_priorities(indices, td_per_sample_host)` after
|
||||
/// the Q backward writes per-sample CE into `self.td_per_sample_d`.
|
||||
fn sample_and_gather(&mut self, b_size: usize) -> Result<Vec<usize>> {
|
||||
let per_alpha = self.isv_host[crate::rl::isv_slots::RL_PER_ALPHA_INDEX];
|
||||
// R1 bootstrap = 0.6; controller can drift it. Bound-check at
|
||||
// the consumer site so a transient zero doesn't degenerate the
|
||||
// sampling. ReplayBuffer.sample_indices already falls back to
|
||||
// uniform when total weights are zero, so this is layered
|
||||
// defense per `pearl_kernel_input_filter_must_match_validation_surface`.
|
||||
let per_alpha_safe = if per_alpha > 0.0 { per_alpha } else { 0.6 };
|
||||
|
||||
let indices = self.replay.sample_indices(b_size, per_alpha_safe);
|
||||
if indices.len() != b_size {
|
||||
// Buffer hasn't been pushed yet (cold start, before
|
||||
// push_to_replay) — return empty to signal "skip Q
|
||||
// backward this step". Caller handles by zeroing λ_q
|
||||
// for the off-policy path.
|
||||
return Ok(indices);
|
||||
}
|
||||
|
||||
// Gather per-batch metadata into HtoD-stagable Vecs.
|
||||
let mut actions_host = vec![0i32; b_size];
|
||||
let mut rewards_host = vec![0.0f32; b_size];
|
||||
let mut dones_host = vec![0.0f32; b_size];
|
||||
let mut n_step_gammas_host = vec![0.0f32; b_size];
|
||||
let current_scale = self.isv_host[crate::rl::isv_slots::RL_REWARD_SCALE_INDEX];
|
||||
for (i, &idx) in indices.iter().enumerate() {
|
||||
let t = &self.replay.transitions[idx];
|
||||
actions_host[i] = t.action as i32;
|
||||
rewards_host[i] = t.raw_reward * current_scale;
|
||||
dones_host[i] = if t.done { 1.0 } else { 0.0 };
|
||||
n_step_gammas_host[i] = t.n_step_gamma;
|
||||
}
|
||||
write_slice_i32_d(&self.stream, &actions_host, &mut self.sampled_actions_d)
|
||||
.context("sample_and_gather: write sampled_actions_d")?;
|
||||
write_slice_f32_d(&self.stream, &rewards_host, &mut self.sampled_rewards_d)
|
||||
.context("sample_and_gather: write sampled_rewards_d")?;
|
||||
write_slice_f32_d(&self.stream, &dones_host, &mut self.sampled_dones_d)
|
||||
.context("sample_and_gather: write sampled_dones_d")?;
|
||||
write_slice_f32_d(&self.stream, &n_step_gammas_host, &mut self.sampled_n_step_gammas_d)
|
||||
.context("sample_and_gather: write sampled_n_step_gammas_d")?;
|
||||
|
||||
// Per-batch DtoD: per-transition h_t / h_tp1 device buffers →
|
||||
// contiguous `sampled_h_t_d` / `sampled_h_tp1_d` at slot i.
|
||||
let hidden_nbytes = HIDDEN_DIM * std::mem::size_of::<f32>();
|
||||
for (i, &idx) in indices.iter().enumerate() {
|
||||
unsafe {
|
||||
let s = self.stream.cu_stream();
|
||||
let off = (i * HIDDEN_DIM * std::mem::size_of::<f32>()) as u64;
|
||||
let (src_ht, _g1) = self.replay.transitions[idx]
|
||||
.h_t
|
||||
.device_ptr(&self.stream);
|
||||
let (src_htp1, _g2) = self.replay.transitions[idx]
|
||||
.next_h_t
|
||||
.device_ptr(&self.stream);
|
||||
let (dst_ht_base, _gd1) =
|
||||
self.sampled_h_t_d.device_ptr_mut(&self.stream);
|
||||
cudarc::driver::result::memcpy_dtod_async(
|
||||
dst_ht_base + off, src_ht, hidden_nbytes, s,
|
||||
)
|
||||
.context("sample_and_gather: DtoD per-batch h_t")?;
|
||||
let (dst_htp1_base, _gd2) =
|
||||
self.sampled_h_tp1_d.device_ptr_mut(&self.stream);
|
||||
cudarc::driver::result::memcpy_dtod_async(
|
||||
dst_htp1_base + off, src_htp1, hidden_nbytes, s,
|
||||
)
|
||||
.context("sample_and_gather: DtoD per-batch h_tp1")?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(indices)
|
||||
}
|
||||
|
||||
/// Launch `rl_lr_controller` to emit per-head learning rates into
|
||||
/// `ISV[412..417]` via ReduceLROnPlateau-style monotone decay.
|
||||
/// Per-head observed loss (l_q / l_pi / l_v from step_synthetic's
|
||||
@@ -6213,7 +5961,7 @@ fn reduce_axis0_free(
|
||||
/// mapped-pinned staging + DtoD + stream sync. Same pattern as
|
||||
/// `read_slice_d` (f32 version) — the only permitted CPU↔GPU path
|
||||
/// per `feedback_no_htod_htoh_only_mapped_pinned`. Used by R7d's
|
||||
/// `push_to_replay` to read the per-step actions device buffer for
|
||||
/// GPU PER push path to read the per-step actions device buffer for
|
||||
/// PER metadata staging.
|
||||
pub fn read_slice_i32_d_pub(
|
||||
stream: &Arc<CudaStream>,
|
||||
|
||||
@@ -49,8 +49,6 @@ spec:
|
||||
value: "32"
|
||||
- name: per-capacity
|
||||
value: "32768"
|
||||
- name: diag-every
|
||||
value: "100"
|
||||
- name: instrument-mode
|
||||
value: "all"
|
||||
- name: log-every
|
||||
@@ -174,7 +172,6 @@ spec:
|
||||
--seq-len {{workflow.parameters.seq-len}} \
|
||||
--n-backtests {{workflow.parameters.n-backtests}} \
|
||||
--per-capacity {{workflow.parameters.per-capacity}} \
|
||||
--diag-every {{workflow.parameters.diag-every}} \
|
||||
--seed {{workflow.parameters.seed}} \
|
||||
--instrument-mode "{{workflow.parameters.instrument-mode}}" \
|
||||
--fold-idx {{workflow.parameters.fold-idx}} \
|
||||
|
||||
@@ -202,10 +202,12 @@ check_sp18_consumer_audit() {
|
||||
}
|
||||
|
||||
# DQN v2 Invariant 9 enforcement: reject TODO/FIXME/XXX/HACK/TBD markers in added code.
|
||||
# Note: todo!() macro is permitted per CLAUDE.md ("todo!() macro is OK for runtime stubs")
|
||||
# — only comment-style markers and unimplemented!() are blocked.
|
||||
check_no_todo_fixme() {
|
||||
local added_lines=$(git diff --cached --diff-filter=AM -U0 -- '*.rs' '*.cu' '*.cuh' \
|
||||
| grep -E '^\+' | grep -vE '^\+\+\+')
|
||||
local bad=$(echo "$added_lines" | grep -E '(TODO|FIXME|\bXXX\b|\bHACK\b|\bTBD\b|unimplemented!|todo!\()')
|
||||
local bad=$(echo "$added_lines" | grep -E '(TODO|FIXME|\bXXX\b|\bHACK\b|\bTBD\b|unimplemented!)' | grep -v 'todo!(')
|
||||
if [ -n "$bad" ]; then
|
||||
echo "❌ Invariant 9 violation: deferred-work markers found in staged code"
|
||||
echo "$bad" | sed 's/^/ /'
|
||||
|
||||
Reference in New Issue
Block a user