Compare commits
14 Commits
worktree-a
...
ml-alpha-p
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d8247034f8 | ||
|
|
346e6670f5 | ||
|
|
54de55d4bc | ||
|
|
d76919d6a2 | ||
|
|
9c1b70edec | ||
|
|
7c96504155 | ||
|
|
f7427b27de | ||
|
|
5717bc07fe | ||
|
|
ff8cacb0e9 | ||
|
|
de378c5f62 | ||
|
|
6fca6a1d9b | ||
|
|
dfbc916227 | ||
|
|
90f178ae9e | ||
|
|
2bdb55cc5b |
@@ -112,6 +112,7 @@ const KERNELS: &[&str] = &[
|
||||
"rl_hindsight_track", // HER Phase 1: per-step mid-price ring + peak tracking for backward hindsight
|
||||
"rl_hindsight_inject", // HER Phase 2: backward inject — synthetic replay push on done if peak >> actual
|
||||
"rl_hindsight_forward", // HER Phase 3: forward continuation — evaluates closed trades after lookahead
|
||||
"rl_lr_from_mapped_pinned", // Mega-graph: LR controller reads loss from mapped-pinned dev_ptrs (eliminates host loss readback between reward + training graphs)
|
||||
"rl_increment_step", // device-resident step counter bump (ISV[548] += 1.0); graph-safe prereq — removes scalar current_step from all downstream kernel args
|
||||
"rl_fused_controllers", // fused kernel: 10 RL ISV controllers in one launch (gamma, tau, ppo_clip, entropy_coef, rollout_steps, per_alpha, reward_scale, ppo_ratio_clamp, gate_threshold, q_distill_lambda) — saves 9 kernel launch overheads (~40-80μs/step)
|
||||
"rl_popart_normalize", // PopArt: Welford-EMA reward normalization (replaces apply_reward_scale)
|
||||
|
||||
@@ -44,3 +44,41 @@ extern "C" __global__ void adamw_step(
|
||||
|
||||
theta[i] -= lr * (m_hat / (sqrtf(v_hat) + eps) + wd * theta[i]);
|
||||
}
|
||||
|
||||
// Mega-graph variant: reads LR from an ISV device pointer at a given
|
||||
// slot index instead of taking a scalar argument. This allows the LR
|
||||
// to vary across graph replays (the ISV slot is modified in-place by
|
||||
// the rl_lr_from_mapped_pinned controller kernel which is captured
|
||||
// earlier in the same graph). All other args are identical to adamw_step.
|
||||
extern "C" __global__ void adamw_step_isv_lr(
|
||||
float* __restrict__ theta,
|
||||
const float* __restrict__ grad,
|
||||
float* __restrict__ m,
|
||||
float* __restrict__ v,
|
||||
int n_params,
|
||||
const float* __restrict__ isv,
|
||||
int lr_slot,
|
||||
float beta1,
|
||||
float beta2,
|
||||
float eps,
|
||||
float wd,
|
||||
int step
|
||||
) {
|
||||
int i = blockIdx.x * blockDim.x + threadIdx.x;
|
||||
if (i >= n_params) return;
|
||||
|
||||
const float lr = isv[lr_slot];
|
||||
|
||||
const float g = grad[i];
|
||||
const float m_n = beta1 * m[i] + (1.0f - beta1) * g;
|
||||
const float v_n = beta2 * v[i] + (1.0f - beta2) * g * g;
|
||||
m[i] = m_n;
|
||||
v[i] = v_n;
|
||||
|
||||
const float bc1 = 1.0f - powf(beta1, (float) step);
|
||||
const float bc2 = 1.0f - powf(beta2, (float) step);
|
||||
const float m_hat = m_n / fmaxf(bc1, 1e-12f);
|
||||
const float v_hat = v_n / fmaxf(bc2, 1e-12f);
|
||||
|
||||
theta[i] -= lr * (m_hat / (sqrtf(v_hat) + eps) + wd * theta[i]);
|
||||
}
|
||||
|
||||
@@ -36,3 +36,20 @@ extern "C" __global__ void grad_h_accumulate_scaled(
|
||||
if (i >= n) return;
|
||||
grad_h_encoder[i] += lambda * grad_h_head[i];
|
||||
}
|
||||
|
||||
// Mega-graph variant: reads lambda from ISV device pointer at given slot.
|
||||
// This allows the lambda to vary across graph replays (ISV is modified
|
||||
// in-place by controller kernels captured earlier in the same graph).
|
||||
extern "C" __global__ void grad_h_accumulate_scaled_isv(
|
||||
const float* __restrict__ grad_h_head,
|
||||
const float* __restrict__ isv,
|
||||
int lambda_slot,
|
||||
float batch_inv, // 1.0 / b_size
|
||||
int n,
|
||||
float* __restrict__ grad_h_encoder
|
||||
) {
|
||||
const int i = blockIdx.x * blockDim.x + threadIdx.x;
|
||||
if (i >= n) return;
|
||||
const float lambda = isv[lambda_slot] * batch_inv;
|
||||
grad_h_encoder[i] += lambda * grad_h_head[i];
|
||||
}
|
||||
|
||||
@@ -96,7 +96,6 @@ extern "C" __global__ void rl_confidence_gate(
|
||||
|
||||
const int action = actions[b];
|
||||
if (action == ACTION_HOLD) return;
|
||||
if (action == 3 || action == 4) return; // FlatFromLong/Short: never gate exits
|
||||
|
||||
const float threshold = isv[RL_CONF_GATE_THRESHOLD_INDEX];
|
||||
const float lambda = isv[RL_CONF_GATE_LAMBDA_INDEX];
|
||||
|
||||
134
crates/ml-alpha/cuda/rl_lr_from_mapped_pinned.cu
Normal file
134
crates/ml-alpha/cuda/rl_lr_from_mapped_pinned.cu
Normal file
@@ -0,0 +1,134 @@
|
||||
// rl_lr_from_mapped_pinned.cu — GPU-side LR controller that reads loss
|
||||
// observations from mapped-pinned device pointers instead of host-
|
||||
// supplied scalar arguments. This eliminates the host-side loss readback
|
||||
// between the reward graph and training graph, enabling mega-graph
|
||||
// capture of the entire per-step pipeline.
|
||||
//
|
||||
// The mapped-pinned buffers (ss_q_loss_mapped, ss_pi_loss_mapped,
|
||||
// ss_v_loss_sum_mapped) have stable device pointers. The GPU writes
|
||||
// loss values during backward kernels; this kernel reads them from the
|
||||
// same physical memory via the device-side pointer. The reads are
|
||||
// coherent because this kernel launches AFTER the backward kernels on
|
||||
// the same stream (stream ordering guarantees visibility).
|
||||
//
|
||||
// Internally delegates to the same plateau_decay_head logic as
|
||||
// rl_lr_controller.cu. The only difference is the loss observation
|
||||
// source: pointer dereference instead of scalar argument.
|
||||
|
||||
#define RL_LR_BCE_INDEX 412
|
||||
#define RL_LR_Q_INDEX 413
|
||||
#define RL_LR_PI_INDEX 414
|
||||
#define RL_LR_V_INDEX 415
|
||||
#define RL_LR_AUX_INDEX 416
|
||||
|
||||
#define RL_LR_BOOTSTRAP_INDEX 462
|
||||
#define RL_LR_MIN_INDEX 463
|
||||
#define RL_LR_MAX_INDEX 464
|
||||
#define RL_LR_LOSS_EMA_ALPHA_INDEX 465
|
||||
#define RL_LR_DECAY_FACTOR_INDEX 466
|
||||
|
||||
#define RL_IMPROVEMENT_THRESHOLD_INDEX 455
|
||||
#define RL_PLATEAU_PATIENCE_INDEX 456
|
||||
#define RL_LR_WARMUP_STEPS_INDEX 461
|
||||
|
||||
__device__ __forceinline__ void plateau_decay_head(
|
||||
float* isv,
|
||||
int lr_idx,
|
||||
float observed_loss,
|
||||
int loss_ema_slot,
|
||||
int best_slot,
|
||||
int counter_slot,
|
||||
int warmup_slot
|
||||
) {
|
||||
const float lr_prev = isv[lr_idx];
|
||||
const float lr_bootstrap = isv[RL_LR_BOOTSTRAP_INDEX];
|
||||
|
||||
if (lr_prev == 0.0f) {
|
||||
isv[lr_idx] = lr_bootstrap;
|
||||
return;
|
||||
}
|
||||
|
||||
if (observed_loss == 0.0f) return;
|
||||
if (loss_ema_slot < 0) return;
|
||||
|
||||
const float loss_ema_prev = isv[loss_ema_slot];
|
||||
const float loss_ema_alpha = isv[RL_LR_LOSS_EMA_ALPHA_INDEX];
|
||||
float loss_ema_new;
|
||||
if (loss_ema_prev == 0.0f) {
|
||||
loss_ema_new = observed_loss;
|
||||
} else {
|
||||
loss_ema_new = (1.0f - loss_ema_alpha) * loss_ema_prev
|
||||
+ loss_ema_alpha * observed_loss;
|
||||
}
|
||||
isv[loss_ema_slot] = loss_ema_new;
|
||||
|
||||
const float warmup_prev = isv[warmup_slot];
|
||||
const float warmup_target = isv[RL_LR_WARMUP_STEPS_INDEX];
|
||||
if (warmup_prev < warmup_target) {
|
||||
isv[best_slot] = loss_ema_new;
|
||||
isv[counter_slot] = 0.0f;
|
||||
isv[warmup_slot] = warmup_prev + 1.0f;
|
||||
return;
|
||||
}
|
||||
|
||||
const float improvement_threshold = isv[RL_IMPROVEMENT_THRESHOLD_INDEX];
|
||||
const float plateau_patience = isv[RL_PLATEAU_PATIENCE_INDEX];
|
||||
const float best_prev = isv[best_slot];
|
||||
|
||||
if (loss_ema_new < best_prev * improvement_threshold) {
|
||||
isv[best_slot] = loss_ema_new;
|
||||
isv[counter_slot] = 0.0f;
|
||||
return;
|
||||
}
|
||||
|
||||
const float counter_next = isv[counter_slot] + 1.0f;
|
||||
if (counter_next >= plateau_patience) {
|
||||
const float lr_min = isv[RL_LR_MIN_INDEX];
|
||||
const float decay_factor = isv[RL_LR_DECAY_FACTOR_INDEX];
|
||||
const float lr_new = fmaxf(lr_min, lr_prev * decay_factor);
|
||||
isv[lr_idx] = lr_new;
|
||||
isv[counter_slot] = 0.0f;
|
||||
} else {
|
||||
isv[counter_slot] = counter_next;
|
||||
}
|
||||
}
|
||||
|
||||
extern "C" __global__ void rl_lr_from_mapped_pinned(
|
||||
float* __restrict__ isv,
|
||||
const float* __restrict__ q_loss_ptr, // mapped-pinned dev_ptr (1 float)
|
||||
const float* __restrict__ pi_loss_ptr, // mapped-pinned dev_ptr (1 float)
|
||||
const float* __restrict__ v_loss_sum_ptr, // mapped-pinned dev_ptr (1 float)
|
||||
int b_size, // batch size for V loss normalization
|
||||
int q_loss_ema_slot,
|
||||
int q_best_slot,
|
||||
int q_counter_slot,
|
||||
int q_warmup_slot,
|
||||
int pi_loss_ema_slot,
|
||||
int pi_best_slot,
|
||||
int pi_counter_slot,
|
||||
int pi_warmup_slot,
|
||||
int v_loss_ema_slot,
|
||||
int v_best_slot,
|
||||
int v_counter_slot,
|
||||
int v_warmup_slot
|
||||
) {
|
||||
if (threadIdx.x != 0 || blockIdx.x != 0) return;
|
||||
|
||||
// Read losses from mapped-pinned device pointers. These are the
|
||||
// same physical pages the backward kernels wrote to — coherent
|
||||
// because this kernel is stream-ordered after them.
|
||||
const float observed_loss_q = *q_loss_ptr;
|
||||
const float observed_loss_pi = *pi_loss_ptr;
|
||||
// V loss is stored as a sum across the batch; normalize to per-sample.
|
||||
const float observed_loss_v = (b_size > 0) ? (*v_loss_sum_ptr / (float)b_size) : 0.0f;
|
||||
|
||||
// BCE and AUX: perception-owned heads, pass slot=-1 to skip.
|
||||
plateau_decay_head(isv, RL_LR_BCE_INDEX, 0.0f, -1, -1, -1, -1);
|
||||
plateau_decay_head(isv, RL_LR_Q_INDEX, observed_loss_q,
|
||||
q_loss_ema_slot, q_best_slot, q_counter_slot, q_warmup_slot);
|
||||
plateau_decay_head(isv, RL_LR_PI_INDEX, observed_loss_pi,
|
||||
pi_loss_ema_slot, pi_best_slot, pi_counter_slot, pi_warmup_slot);
|
||||
plateau_decay_head(isv, RL_LR_V_INDEX, observed_loss_v,
|
||||
v_loss_ema_slot, v_best_slot, v_counter_slot, v_warmup_slot);
|
||||
plateau_decay_head(isv, RL_LR_AUX_INDEX, 0.0f, -1, -1, -1, -1);
|
||||
}
|
||||
@@ -1,25 +1,23 @@
|
||||
/**
|
||||
* rl_per_push_flush.cu — Coordinated coalesced replay write.
|
||||
* rl_per_push_flush.cu — Two-kernel coordinated coalesced replay write.
|
||||
*
|
||||
* Second kernel in the split rl_per_push pipeline. After rl_per_push_ring has
|
||||
* written data to the n-step ring and set flush_flags[b] for completed n-step
|
||||
* transitions, this kernel coordinates slot allocation and performs the actual
|
||||
* coalesced write to the replay buffer.
|
||||
* Split from the original single-kernel volatile-spin design into two
|
||||
* kernels with stream-ordered dependency (graph-safe):
|
||||
*
|
||||
* Architecture:
|
||||
* - Grid=(b_size, 1, 1), Block=(128, 1, 1). One block per batch element.
|
||||
* - Block 0 runs the prefix-sum to assign write slots (single-threaded, O(B)).
|
||||
* - All other blocks spin on a volatile global-memory ready flag (no atomics).
|
||||
* - Once ready, each flushing block does a coalesced 128-thread copy of h_t
|
||||
* and h_tp1, then thread 0 computes the n-step return and writes scalars +
|
||||
* priority + resets the ring count.
|
||||
* Kernel 1: rl_per_flush_prefix_sum
|
||||
* Grid=(1), Block=(1). Single thread computes exclusive prefix-sum of
|
||||
* flush_flags, advances write_head and replay_len, writes per-batch
|
||||
* offsets + base into write_offsets[0..b_size+1].
|
||||
*
|
||||
* Coordination: block-0 prefix-sum + volatile ready flag.
|
||||
* No atomicAdd — slot assignment is deterministic serial scan (feedback_no_atomicadd).
|
||||
* Kernel 2: rl_per_flush_write
|
||||
* Grid=(b_size), Block=(128). Each block reads its pre-computed offset
|
||||
* from write_offsets, then does the coalesced h_t/h_tp1 copy and
|
||||
* scalar write. No spin — the prefix-sum kernel completed before this
|
||||
* launch (stream ordering).
|
||||
*
|
||||
* No atomicAdd — slot assignment is deterministic serial scan.
|
||||
* No volatile spin — correctness from stream ordering / graph node deps.
|
||||
* Pre-compiled cubin only (feedback_no_nvrtc).
|
||||
*
|
||||
* Launch: CUDA_HOME=/usr/local/cuda nvcc -cubin -arch sm_86 -O3 \
|
||||
* --generate-line-info rl_per_push_flush.cu -o rl_per_push_flush.cubin
|
||||
*/
|
||||
|
||||
#define HIDDEN_DIM 128
|
||||
@@ -30,105 +28,89 @@
|
||||
#define RL_GAMMA_INDEX 400
|
||||
#define RL_N_STEP_INDEX 403
|
||||
|
||||
extern "C" __global__ void rl_per_push_flush(
|
||||
/* ── Flush coordination ── */
|
||||
const int* __restrict__ flush_flags, /* [B] from ring kernel: 1=flush, 0=skip */
|
||||
int* __restrict__ write_offsets, /* [B+2] output: per-batch offsets, base, ready flag */
|
||||
/* =====================================================================
|
||||
* Kernel 1: prefix-sum + write_head advance.
|
||||
*
|
||||
* Grid=(1,1,1), Block=(1,1,1). Single thread.
|
||||
*
|
||||
* Reads flush_flags[0..b_size], computes exclusive prefix-sum into
|
||||
* write_offsets[0..b_size], stores base write_head at write_offsets[b_size],
|
||||
* and advances the replay buffer's write_head + replay_len.
|
||||
* ===================================================================== */
|
||||
extern "C" __global__ void rl_per_flush_prefix_sum(
|
||||
const int* __restrict__ flush_flags, /* [B] */
|
||||
int* __restrict__ write_offsets, /* [B+1] output: offsets + base */
|
||||
unsigned int* __restrict__ write_head, /* [1] */
|
||||
unsigned int* __restrict__ replay_len, /* [1] */
|
||||
int b_size,
|
||||
int capacity
|
||||
)
|
||||
{
|
||||
unsigned int base = write_head[0];
|
||||
|
||||
/* ── N-step ring (read source) ── */
|
||||
const float* __restrict__ nstep_scalars, /* [B * N_STEP_MAX * 3]: (r_scaled, r_raw, done) per ring slot */
|
||||
/* Exclusive prefix-sum */
|
||||
int total = 0;
|
||||
for (int i = 0; i < b_size; ++i) {
|
||||
write_offsets[i] = total;
|
||||
total += flush_flags[i];
|
||||
}
|
||||
|
||||
/* Advance write head and replay length */
|
||||
if (total > 0) {
|
||||
write_head[0] = (base + (unsigned int)total) % (unsigned int)capacity;
|
||||
unsigned int len = replay_len[0] + (unsigned int)total;
|
||||
if (len > (unsigned int)capacity) len = (unsigned int)capacity;
|
||||
replay_len[0] = len;
|
||||
}
|
||||
|
||||
/* Store base write head for the write kernel */
|
||||
write_offsets[b_size] = (int)base;
|
||||
}
|
||||
|
||||
/* =====================================================================
|
||||
* Kernel 2: coalesced replay write.
|
||||
*
|
||||
* Grid=(b_size,1,1), Block=(128,1,1). One block per batch element.
|
||||
*
|
||||
* Reads pre-computed offset from write_offsets (populated by kernel 1).
|
||||
* No spin — stream ordering guarantees kernel 1 completed before this.
|
||||
* ===================================================================== */
|
||||
extern "C" __global__ void rl_per_flush_write(
|
||||
const int* __restrict__ flush_flags, /* [B] */
|
||||
const int* __restrict__ write_offsets, /* [B+1] */
|
||||
|
||||
/* N-step ring (read source) */
|
||||
const float* __restrict__ nstep_scalars, /* [B * N_STEP_MAX * 3] */
|
||||
const float* __restrict__ nstep_h_t, /* [B * N_STEP_MAX * HIDDEN_DIM] */
|
||||
const unsigned int* __restrict__ nstep_write_idx, /* [B] ring write cursor */
|
||||
const unsigned int* __restrict__ nstep_count, /* [B] items in ring */
|
||||
|
||||
/* ── Current step data (for h_tp1) ── */
|
||||
/* Current step data (for h_tp1) */
|
||||
const float* __restrict__ h_tp1_current, /* [B * HIDDEN_DIM] */
|
||||
const float* __restrict__ log_pi_old, /* [B] */
|
||||
const int* __restrict__ actions, /* [B] */
|
||||
const float* __restrict__ isv, /* ISV array */
|
||||
|
||||
/* ── Replay storage (write destination) ── */
|
||||
/* Replay storage (write destination) */
|
||||
float* __restrict__ replay_h_t, /* [capacity * HIDDEN_DIM] */
|
||||
float* __restrict__ replay_h_tp1, /* [capacity * HIDDEN_DIM] */
|
||||
float* __restrict__ replay_scalars, /* [capacity * SCALARS_PER_TRANSITION] */
|
||||
float* __restrict__ priority_tree, /* [2 * capacity] (leaf layer at offset capacity) */
|
||||
unsigned int* __restrict__ write_head, /* [1] circular write cursor */
|
||||
unsigned int* __restrict__ replay_len, /* [1] current buffer occupancy */
|
||||
float* __restrict__ max_priority, /* [1] running max priority for new inserts */
|
||||
float* __restrict__ max_priority, /* [1] running max priority */
|
||||
|
||||
/* ── N-step count reset (written on flush) ── */
|
||||
unsigned int* __restrict__ nstep_count_out, /* [B] — set to 0 for flushed batches */
|
||||
/* N-step count reset (written on flush) */
|
||||
unsigned int* __restrict__ nstep_count_out, /* [B] */
|
||||
|
||||
int b_size,
|
||||
int capacity
|
||||
)
|
||||
{
|
||||
/* ====================================================================
|
||||
* PHASE 1: Block 0 coordination (thread 0 only).
|
||||
*
|
||||
* Computes prefix-sum of flush_flags to assign contiguous write slots,
|
||||
* advances write_head and replay_len, stores base for other blocks,
|
||||
* and signals ready via volatile write.
|
||||
* ==================================================================== */
|
||||
if (blockIdx.x == 0 && threadIdx.x == 0) {
|
||||
unsigned int base = write_head[0];
|
||||
|
||||
/* Exclusive prefix-sum: write_offsets[i] = number of flushes before batch i */
|
||||
int total = 0;
|
||||
for (int i = 0; i < b_size; ++i) {
|
||||
write_offsets[i] = total;
|
||||
total += flush_flags[i];
|
||||
}
|
||||
|
||||
/* Advance write head and replay length */
|
||||
if (total > 0) {
|
||||
write_head[0] = (base + (unsigned int)total) % (unsigned int)capacity;
|
||||
unsigned int len = replay_len[0] + (unsigned int)total;
|
||||
if (len > (unsigned int)capacity) len = (unsigned int)capacity;
|
||||
replay_len[0] = len;
|
||||
}
|
||||
|
||||
/* Store base write head for other blocks at slot [b_size] */
|
||||
write_offsets[b_size] = (int)base;
|
||||
|
||||
/* Fence: ensure all writes above are globally visible before signaling */
|
||||
__threadfence();
|
||||
|
||||
/* Signal ready at slot [b_size + 1] */
|
||||
write_offsets[b_size + 1] = 1;
|
||||
__threadfence();
|
||||
}
|
||||
|
||||
/* ====================================================================
|
||||
* PHASE 2: All blocks wait for ready signal.
|
||||
*
|
||||
* Non-block-0 blocks spin on a volatile read of the ready flag.
|
||||
* Block 0 already wrote it, so it proceeds immediately.
|
||||
* ==================================================================== */
|
||||
if (threadIdx.x == 0 && blockIdx.x != 0) {
|
||||
volatile int* ready = (volatile int*)&write_offsets[b_size + 1];
|
||||
while (*ready == 0) {
|
||||
/* spin — volatile load, no atomic */
|
||||
}
|
||||
}
|
||||
__syncthreads();
|
||||
|
||||
/* ====================================================================
|
||||
* PHASE 3: Coalesced replay write (128 threads per block).
|
||||
*
|
||||
* Each flushing block writes:
|
||||
* - h_t[HIDDEN_DIM]: from oldest ring entry (coalesced, all 128 threads)
|
||||
* - h_tp1[HIDDEN_DIM]: from current step (coalesced, all 128 threads)
|
||||
* - scalars[7]: n-step return + metadata (thread 0 only)
|
||||
* - priority: max_priority into leaf layer (thread 0 only)
|
||||
* - reset: nstep_count_out[b] = 0 (thread 0 only)
|
||||
* ==================================================================== */
|
||||
const int b = blockIdx.x;
|
||||
|
||||
/* Early exit: nothing to flush for this batch */
|
||||
if (flush_flags[b] == 0) return;
|
||||
|
||||
/* Compute target replay slot */
|
||||
/* Compute target replay slot from pre-computed offsets */
|
||||
const unsigned int base = (unsigned int)write_offsets[b_size];
|
||||
const unsigned int my_offset = (unsigned int)write_offsets[b];
|
||||
const unsigned int my_slot = (base + my_offset) % (unsigned int)capacity;
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
/* =====================================================================
|
||||
* rl_per_sample.cu — GPU-resident PER: proportional sampling via sum-tree
|
||||
*
|
||||
* Grid=(b_size), Block=(1). One thread per sample.
|
||||
* Grid=(b_size), Block=(128). One block per sample.
|
||||
*
|
||||
* Each thread samples a leaf from the priority sum-tree using stratified
|
||||
* sampling (segment per thread) with xorshift32 PRNG, then gathers the
|
||||
* transition data from replay storage.
|
||||
* Thread 0 does the tree walk + scalar unpack. All 128 threads cooperate
|
||||
* on the coalesced h_t and h_tp1 gathers (128 floats = HIDDEN_DIM per
|
||||
* sample, one float per thread, single 512-byte coalesced transaction).
|
||||
*
|
||||
* ISV reads: none (alpha used only at priority-update time)
|
||||
* ===================================================================== */
|
||||
@@ -44,75 +44,87 @@ extern "C" __global__ void rl_per_sample(
|
||||
int capacity
|
||||
)
|
||||
{
|
||||
const int b = blockIdx.x * blockDim.x + threadIdx.x;
|
||||
const int b = blockIdx.x;
|
||||
const int tid = threadIdx.x;
|
||||
if (b >= b_size) return;
|
||||
|
||||
const unsigned int len = replay_len[0];
|
||||
const float total_priority = priority_tree[1]; /* root */
|
||||
/* ── Shared memory for broadcasting the sampled leaf index ────────── */
|
||||
__shared__ int s_safe_leaf;
|
||||
|
||||
/* Guard: empty or zero-priority replay */
|
||||
if (total_priority < 1e-9f || len == 0) {
|
||||
for (int i = 0; i < HIDDEN_DIM; ++i) {
|
||||
sampled_h_t[b * HIDDEN_DIM + i] = 0.0f;
|
||||
sampled_h_tp1[b * HIDDEN_DIM + i] = 0.0f;
|
||||
/* ── Thread 0: tree walk + scalar reads ──────────────────────────── */
|
||||
if (tid == 0) {
|
||||
const unsigned int len = replay_len[0];
|
||||
const float total_priority = priority_tree[1]; /* root */
|
||||
|
||||
/* Guard: empty or zero-priority replay */
|
||||
if (total_priority < 1e-9f || len == 0) {
|
||||
s_safe_leaf = -1; /* sentinel: zero-fill outputs */
|
||||
} else {
|
||||
/* Seed PRNG if cold */
|
||||
if (prng_state[b] == 0) {
|
||||
prng_state[b] = (unsigned int)(b + 1) * 2654435761u;
|
||||
}
|
||||
|
||||
/* Stratified sampling: draw u in [b*segment, (b+1)*segment) */
|
||||
const float segment = total_priority / (float)b_size;
|
||||
unsigned int rng = xorshift32(&prng_state[b]);
|
||||
const float u_frac = (float)(rng & 0x00FFFFFFu) / (float)0x01000000u;
|
||||
float u = ((float)b + u_frac) * segment;
|
||||
|
||||
/* Clamp to avoid floating-point overshoot */
|
||||
if (u >= total_priority) u = total_priority - 1e-6f;
|
||||
if (u < 0.0f) u = 0.0f;
|
||||
|
||||
/* Walk tree top-down */
|
||||
int idx = 1;
|
||||
while (idx < capacity) {
|
||||
const int left = 2 * idx;
|
||||
const float left_val = priority_tree[left];
|
||||
if (u <= left_val) {
|
||||
idx = left;
|
||||
} else {
|
||||
u -= left_val;
|
||||
idx = left + 1;
|
||||
}
|
||||
}
|
||||
const int leaf = idx - capacity;
|
||||
|
||||
/* Clamp leaf to valid range */
|
||||
const int safe_leaf = (leaf >= 0 && leaf < (int)len) ? leaf : 0;
|
||||
s_safe_leaf = safe_leaf;
|
||||
sample_indices[b] = (unsigned int)safe_leaf;
|
||||
|
||||
/* Unpack scalars (thread 0 only — small data, not worth parallelising) */
|
||||
const int sc_base = safe_leaf * SCALARS_PER_TRANSITION;
|
||||
sampled_actions[b] = (int)replay_scalars[sc_base + 0];
|
||||
sampled_rewards[b] = replay_scalars[sc_base + 1];
|
||||
sampled_dones[b] = replay_scalars[sc_base + 3];
|
||||
sampled_log_pi_old[b] = replay_scalars[sc_base + 4];
|
||||
sampled_n_step_gammas[b] = replay_scalars[sc_base + 5];
|
||||
}
|
||||
}
|
||||
__syncthreads();
|
||||
|
||||
const int safe_leaf = s_safe_leaf;
|
||||
|
||||
if (safe_leaf < 0) {
|
||||
/* Empty replay — zero-fill h_t and h_tp1 (coalesced, all 128 threads) */
|
||||
sampled_h_t[b * HIDDEN_DIM + tid] = 0.0f;
|
||||
sampled_h_tp1[b * HIDDEN_DIM + tid] = 0.0f;
|
||||
if (tid == 0) {
|
||||
sampled_rewards[b] = 0.0f;
|
||||
sampled_dones[b] = 0.0f;
|
||||
sampled_log_pi_old[b] = 0.0f;
|
||||
sampled_n_step_gammas[b] = 0.0f;
|
||||
sampled_actions[b] = 0;
|
||||
sample_indices[b] = 0;
|
||||
}
|
||||
sampled_rewards[b] = 0.0f;
|
||||
sampled_dones[b] = 0.0f;
|
||||
sampled_log_pi_old[b] = 0.0f;
|
||||
sampled_n_step_gammas[b] = 0.0f;
|
||||
sampled_actions[b] = 0;
|
||||
sample_indices[b] = 0;
|
||||
return;
|
||||
}
|
||||
|
||||
/* ── Seed PRNG if cold ────────────────────────────────────────────── */
|
||||
if (prng_state[b] == 0) {
|
||||
prng_state[b] = (unsigned int)(b + 1) * 2654435761u;
|
||||
}
|
||||
/* ── Coalesced h_t gather (128 threads, 1 float each) ────────────── */
|
||||
sampled_h_t[b * HIDDEN_DIM + tid] = replay_h_t[safe_leaf * HIDDEN_DIM + tid];
|
||||
|
||||
/* ── Stratified sampling: draw u in [b*segment, (b+1)*segment) ────── */
|
||||
const float segment = total_priority / (float)b_size;
|
||||
unsigned int rng = xorshift32(&prng_state[b]);
|
||||
const float u_frac = (float)(rng & 0x00FFFFFFu) / (float)0x01000000u; /* [0, 1) */
|
||||
float u = ((float)b + u_frac) * segment;
|
||||
|
||||
/* Clamp to avoid floating-point overshoot */
|
||||
if (u >= total_priority) u = total_priority - 1e-6f;
|
||||
if (u < 0.0f) u = 0.0f;
|
||||
|
||||
/* ── Walk tree top-down ───────────────────────────────────────────── */
|
||||
int idx = 1;
|
||||
while (idx < capacity) {
|
||||
const int left = 2 * idx;
|
||||
const float left_val = priority_tree[left];
|
||||
if (u <= left_val) {
|
||||
idx = left;
|
||||
} else {
|
||||
u -= left_val;
|
||||
idx = left + 1;
|
||||
}
|
||||
}
|
||||
const int leaf = idx - capacity;
|
||||
|
||||
/* Clamp leaf to valid range */
|
||||
const int safe_leaf = (leaf >= 0 && leaf < (int)len) ? leaf : 0;
|
||||
sample_indices[b] = (unsigned int)safe_leaf;
|
||||
|
||||
/* ── Gather h_t ──────────────────────────────────────────────────── */
|
||||
for (int i = 0; i < HIDDEN_DIM; ++i) {
|
||||
sampled_h_t[b * HIDDEN_DIM + i] = replay_h_t[safe_leaf * HIDDEN_DIM + i];
|
||||
}
|
||||
|
||||
/* ── Gather h_tp1 ────────────────────────────────────────────────── */
|
||||
for (int i = 0; i < HIDDEN_DIM; ++i) {
|
||||
sampled_h_tp1[b * HIDDEN_DIM + i] = replay_h_tp1[safe_leaf * HIDDEN_DIM + i];
|
||||
}
|
||||
|
||||
/* ── Unpack scalars ──────────────────────────────────────────────── */
|
||||
const int sc_base = safe_leaf * SCALARS_PER_TRANSITION;
|
||||
sampled_actions[b] = (int)replay_scalars[sc_base + 0];
|
||||
sampled_rewards[b] = replay_scalars[sc_base + 1]; /* n-step scaled return */
|
||||
sampled_dones[b] = replay_scalars[sc_base + 3];
|
||||
sampled_log_pi_old[b] = replay_scalars[sc_base + 4];
|
||||
sampled_n_step_gammas[b] = replay_scalars[sc_base + 5];
|
||||
/* ── Coalesced h_tp1 gather (128 threads, 1 float each) ──────────── */
|
||||
sampled_h_tp1[b * HIDDEN_DIM + tid] = replay_h_tp1[safe_leaf * HIDDEN_DIM + tid];
|
||||
}
|
||||
|
||||
@@ -1,44 +1,30 @@
|
||||
/* =====================================================================
|
||||
* rl_per_tree_rebuild.cu — GPU-resident PER: bottom-up parallel sum-tree rebuild
|
||||
* rl_per_tree_rebuild.cu — GPU-resident PER: per-level sum-tree rebuild
|
||||
*
|
||||
* Grid=(128), Block=(256). Total 32768 threads = capacity.
|
||||
* One kernel invocation per tree level. The host launches log2(capacity)
|
||||
* times, from bottom (level 0 = parents of leaves) to top (root).
|
||||
* Stream ordering between launches provides the device-wide barrier that
|
||||
* the previous __threadfence() approach lacked — each level's writes are
|
||||
* fully complete before the next level reads them.
|
||||
*
|
||||
* Rebuilds the entire sum-tree from leaf priorities (at indices
|
||||
* [capacity..2*capacity)) up to the root (index 1). Each level is
|
||||
* processed in parallel with __threadfence() device-wide barriers
|
||||
* between levels.
|
||||
* Graph-safe: each level is a separate kernel node in the CUDA graph.
|
||||
* The graph engine respects stream-order dependencies between nodes.
|
||||
*
|
||||
* No atomicAdd — each internal node is written by exactly one thread.
|
||||
* ===================================================================== */
|
||||
|
||||
extern "C" __global__ void rl_per_tree_rebuild(
|
||||
extern "C" __global__ void rl_per_tree_rebuild_level(
|
||||
float* __restrict__ priority_tree, /* [2 * capacity] */
|
||||
int capacity
|
||||
int start, /* first node index at this level */
|
||||
int nodes_at_level /* number of nodes to process at this level */
|
||||
)
|
||||
{
|
||||
const int tid_global = blockIdx.x * blockDim.x + threadIdx.x;
|
||||
const int total_threads = gridDim.x * blockDim.x;
|
||||
|
||||
/* Bottom-up: level 0 = parents of leaves, level (log2(cap)-1) = root */
|
||||
/* At level L, there are capacity >> (L+1) internal nodes. */
|
||||
/* Node indices at level L: [capacity >> (L+1) .. capacity >> L) */
|
||||
|
||||
int nodes_at_level = capacity >> 1; /* level 0: cap/2 nodes */
|
||||
int start = nodes_at_level; /* first node index at this level */
|
||||
|
||||
while (nodes_at_level >= 1) {
|
||||
/* Each thread processes multiple nodes via grid-stride loop */
|
||||
for (int i = tid_global; i < nodes_at_level; i += total_threads) {
|
||||
const int node = start + i;
|
||||
priority_tree[node] = priority_tree[2 * node] + priority_tree[2 * node + 1];
|
||||
}
|
||||
|
||||
/* Device-wide fence: all writes at this level visible before
|
||||
* any thread reads them at the next level */
|
||||
__threadfence();
|
||||
|
||||
/* Move up one level */
|
||||
nodes_at_level >>= 1;
|
||||
start >>= 1;
|
||||
/* Grid-stride loop: each thread processes one or more nodes */
|
||||
for (int i = tid_global; i < nodes_at_level; i += total_threads) {
|
||||
const int node = start + i;
|
||||
priority_tree[node] = priority_tree[2 * node] + priority_tree[2 * node + 1];
|
||||
}
|
||||
}
|
||||
|
||||
@@ -979,6 +979,11 @@ fn main() -> Result<()> {
|
||||
0
|
||||
};
|
||||
|
||||
// Mega-graph: capture the entire per-step pipeline into a single
|
||||
// cuGraphLaunch call. Eliminates ~150 individual kernel launches
|
||||
// and ~140ms of host-side Rust overhead per step.
|
||||
trainer.enable_mega_graph();
|
||||
|
||||
let t_start = std::time::Instant::now();
|
||||
for step in start_step..cli.n_steps {
|
||||
// GPU-resident data loading: sample_and_gather + gather_next +
|
||||
@@ -1016,30 +1021,38 @@ fn main() -> Result<()> {
|
||||
// so the diag copies have had the entire training step to finish —
|
||||
// the event sync is instant (~0 us). On step 0 this is a no-op
|
||||
// (event sync on a freshly-created event returns immediately).
|
||||
diag_staging.sync_and_swap().context("diag sync_and_swap")?;
|
||||
// Diag staging: only sync+snapshot every 10 steps or on log/checkpoint
|
||||
// boundaries. At mega-graph speeds (300+ sps), the per-step diag
|
||||
// sync_and_swap blocks for 100ms+ (diag DtoD copies haven't finished
|
||||
// in 3ms). Skipping 9/10 steps keeps the GPU saturated.
|
||||
let diag_this_step = step % 10 == 0
|
||||
|| step % cli.log_every == 0
|
||||
|| (cli.checkpoint_every > 0 && step % cli.checkpoint_every == 0)
|
||||
|| step == start_step
|
||||
|| step + 1 == cli.n_steps;
|
||||
|
||||
// Launch async DtoD copies of ALL diag buffers into the current
|
||||
// staging buffer. Non-blocking on the training stream — the copies
|
||||
// run on diag_staging's separate stream.
|
||||
diag_staging
|
||||
.snapshot_async(
|
||||
trainer.isv_dev_ptr,
|
||||
trainer.rewards_d.raw_ptr(),
|
||||
trainer.dones_d.raw_ptr(),
|
||||
trainer.actions_d.raw_ptr(),
|
||||
trainer.raw_rewards_d.raw_ptr(),
|
||||
trainer.trade_duration_emit_d.raw_ptr(),
|
||||
trainer.outcome_ema_d.raw_ptr(),
|
||||
trainer.prev_position_lots_d.raw_ptr(),
|
||||
trainer.pyramid_units_count_d.raw_ptr(),
|
||||
trainer.unit_entry_price_d.raw_ptr(),
|
||||
trainer.unit_entry_step_d.raw_ptr(),
|
||||
trainer.unit_lots_d.raw_ptr(),
|
||||
trainer.unit_trail_distance_d.raw_ptr(),
|
||||
trainer.close_unit_index_d.raw_ptr(),
|
||||
trainer.frd_logits_d.raw_ptr(),
|
||||
)
|
||||
.context("diag snapshot_async")?;
|
||||
if diag_this_step {
|
||||
diag_staging.sync_and_swap().context("diag sync_and_swap")?;
|
||||
diag_staging
|
||||
.snapshot_async(
|
||||
trainer.isv_dev_ptr,
|
||||
trainer.rewards_d.raw_ptr(),
|
||||
trainer.dones_d.raw_ptr(),
|
||||
trainer.actions_d.raw_ptr(),
|
||||
trainer.raw_rewards_d.raw_ptr(),
|
||||
trainer.trade_duration_emit_d.raw_ptr(),
|
||||
trainer.outcome_ema_d.raw_ptr(),
|
||||
trainer.prev_position_lots_d.raw_ptr(),
|
||||
trainer.pyramid_units_count_d.raw_ptr(),
|
||||
trainer.unit_entry_price_d.raw_ptr(),
|
||||
trainer.unit_entry_step_d.raw_ptr(),
|
||||
trainer.unit_lots_d.raw_ptr(),
|
||||
trainer.unit_trail_distance_d.raw_ptr(),
|
||||
trainer.close_unit_index_d.raw_ptr(),
|
||||
trainer.frd_logits_d.raw_ptr(),
|
||||
)
|
||||
.context("diag snapshot_async")?;
|
||||
}
|
||||
|
||||
// ── Snapshot diagnostic data into owned DiagFrame (~0.1ms). ──
|
||||
// Reads from the DiagStaging double-buffer (previous step's
|
||||
|
||||
@@ -47,7 +47,7 @@ impl GpuReplayBuffer {
|
||||
nstep_write_idx_d: stream.alloc_zeros(b_size).context("gpu_replay nstep_write_idx")?,
|
||||
nstep_count_d: stream.alloc_zeros(b_size).context("gpu_replay nstep_count")?,
|
||||
flush_flags_d: stream.alloc_zeros(b_size).context("gpu_replay flush_flags")?,
|
||||
write_offsets_d: stream.alloc_zeros(b_size + 2).context("gpu_replay write_offsets")?,
|
||||
write_offsets_d: stream.alloc_zeros(b_size + 1).context("gpu_replay write_offsets")?,
|
||||
sample_indices_d: stream.alloc_zeros(b_size).context("gpu_replay sample_indices")?,
|
||||
sample_prng_d: stream.alloc_zeros(b_size).context("gpu_replay sample_prng")?,
|
||||
})
|
||||
|
||||
@@ -46,9 +46,57 @@
|
||||
|
||||
use anyhow::Result;
|
||||
use cudarc::driver::CudaSlice;
|
||||
use cudarc::driver::sys::CUfunction;
|
||||
|
||||
use crate::cfc::snap_features::Mbp10RawInput;
|
||||
|
||||
/// Cached raw device pointers and kernel function handles for the lobsim.
|
||||
/// Extracted once per step to bypass trait dispatch + CudaSlice borrow
|
||||
/// overhead in the mega-graph hot path. All pointers are stable (pre-
|
||||
/// allocated at lobsim construction time, never reallocated).
|
||||
pub struct LobSimRawPtrs {
|
||||
// ── Book update (apply_snapshot_from_device) ────────────────────
|
||||
pub bid_px: u64,
|
||||
pub bid_sz: u64,
|
||||
pub ask_px: u64,
|
||||
pub ask_sz: u64,
|
||||
pub books: u64,
|
||||
pub prev_mid: u64,
|
||||
pub atr_mid_ema: u64,
|
||||
pub snapshots_skipped: u64,
|
||||
pub min_reasonable_px: u64,
|
||||
pub max_reasonable_px: u64,
|
||||
pub book_update_fn: CUfunction,
|
||||
|
||||
// ── Fill (step_fill_from_market_targets) ────────────────────────
|
||||
pub market_targets: u64,
|
||||
pub pos: u64,
|
||||
pub cost_per_lot_per_side: u64,
|
||||
pub total_fees_per_b: u64,
|
||||
pub submit_market_fn: CUfunction,
|
||||
|
||||
// ── PnL track (step_pnl_track) ─────────────────────────────────
|
||||
pub open_trade_state: u64,
|
||||
pub trade_log: u64,
|
||||
pub trade_log_head: u64,
|
||||
pub trail_hwm: u64,
|
||||
pub zero_vwap_at_open: u64,
|
||||
pub saturated_vwap_at_open: u64,
|
||||
pub defensive_exit_clamp: u64,
|
||||
pub conv_signed_ema: u64,
|
||||
pub diag_hold_hist: u64,
|
||||
pub diag_outcome_n: u64,
|
||||
pub diag_outcome_sum_pnl: u64,
|
||||
pub diag_outcome_n_wins: u64,
|
||||
pub pnl_track_fn: CUfunction,
|
||||
|
||||
// ── Dimensions ─────────────────────────────────────────────────
|
||||
pub n_backtests: i32,
|
||||
pub pos_bytes: i32,
|
||||
/// `TRADE_LOG_CAP` from ml-backtesting — passed to `pnl_track` kernel.
|
||||
pub trade_log_cap: i32,
|
||||
}
|
||||
|
||||
/// Narrow device-oriented backend the integrated RL trainer's
|
||||
/// `step_with_lobsim` invokes for its LOB-simulator interaction.
|
||||
/// Implemented for `LobSimCuda` in `ml-backtesting`. See module-level
|
||||
@@ -126,4 +174,10 @@ pub trait RlLobBackend {
|
||||
ask_px_src: u64,
|
||||
ask_sz_src: u64,
|
||||
) -> Result<()>;
|
||||
|
||||
/// Extract all raw device pointers and kernel function handles needed
|
||||
/// for mega-graph capture. Returns a [`LobSimRawPtrs`] struct that the
|
||||
/// trainer caches to bypass trait dispatch inside the captured section.
|
||||
/// All pointers are stable for the lobsim's lifetime.
|
||||
fn raw_ptrs(&self) -> LobSimRawPtrs;
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -122,6 +122,58 @@ impl AdamW {
|
||||
&mut self.v
|
||||
}
|
||||
|
||||
/// Mega-graph variant of `step()`. Reads LR from an ISV device
|
||||
/// pointer at `lr_slot` instead of the host-side `self.lr` field.
|
||||
/// This allows the LR to vary across graph replays because the ISV
|
||||
/// is modified in-place by the `rl_lr_from_mapped_pinned` controller
|
||||
/// kernel captured earlier in the same graph.
|
||||
///
|
||||
/// `isv_lr_fn` is the `adamw_step_isv_lr` kernel function handle.
|
||||
/// `isv_ptr` is the stable ISV device pointer.
|
||||
pub fn step_isv_lr(
|
||||
&mut self,
|
||||
theta: &mut CudaSlice<f32>,
|
||||
grad: &CudaSlice<f32>,
|
||||
isv_lr_fn: cudarc::driver::sys::CUfunction,
|
||||
isv_ptr: u64,
|
||||
lr_slot: i32,
|
||||
) -> Result<()> {
|
||||
let n = theta.len();
|
||||
assert_eq!(grad.len(), n, "grad/theta size mismatch");
|
||||
assert_eq!(self.m.len(), n, "m size mismatch");
|
||||
assert_eq!(self.v.len(), n, "v size mismatch");
|
||||
|
||||
self.step_count_host += 1;
|
||||
|
||||
let n_params_i = n as i32;
|
||||
let grid_x = (n as u32).div_ceil(256);
|
||||
{
|
||||
let mut args = RawArgs::new();
|
||||
args.push_ptr(theta.raw_ptr());
|
||||
args.push_ptr(grad.raw_ptr());
|
||||
args.push_ptr(self.m.raw_ptr());
|
||||
args.push_ptr(self.v.raw_ptr());
|
||||
args.push_i32(n_params_i);
|
||||
args.push_ptr(isv_ptr);
|
||||
args.push_i32(lr_slot);
|
||||
args.push_f32(self.beta1);
|
||||
args.push_f32(self.beta2);
|
||||
args.push_f32(self.eps);
|
||||
args.push_f32(self.wd);
|
||||
args.push_i32(self.step_count_host);
|
||||
let mut ptrs = args.build_arg_ptrs();
|
||||
unsafe {
|
||||
raw_launch(
|
||||
isv_lr_fn,
|
||||
(grid_x, 1, 1), (256, 1, 1), 0,
|
||||
self.raw_stream,
|
||||
&mut ptrs[..args.len()],
|
||||
).map_err(|e| anyhow::anyhow!("adamw_step_isv_lr launch: {:?}", e))?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Return the current step counter value. No GPU sync needed —
|
||||
/// the counter is host-resident.
|
||||
pub fn step_count(&self) -> i32 {
|
||||
|
||||
@@ -718,6 +718,13 @@ pub struct PerceptionTrainer {
|
||||
/// include cuBLAS calls.
|
||||
cublas_warmed: bool,
|
||||
|
||||
/// Mega-graph mode: when true, all perception sub-graph state
|
||||
/// machines (train_graph, forward_graph, forward_graph_no_scatter)
|
||||
/// run eagerly (no sub-capture, no sub-replay). The mega-graph in
|
||||
/// the integrated trainer captures the entire pipeline including
|
||||
/// perception kernels.
|
||||
pub mega_graph_enabled: bool,
|
||||
|
||||
/// Event recorded after every training graph launch in
|
||||
/// `step_batched_from_device`. The integrated trainer syncs on
|
||||
/// this event at the start of the NEXT step
|
||||
@@ -2316,6 +2323,7 @@ impl PerceptionTrainer {
|
||||
forward_graph_no_scatter: None,
|
||||
forward_no_scatter_warmed: false,
|
||||
cublas_warmed: false,
|
||||
mega_graph_enabled: false,
|
||||
training_done_event,
|
||||
|
||||
// AoS staging — single mapped-pinned buffer for B*K Mbp10RawInput.
|
||||
@@ -3235,7 +3243,7 @@ impl PerceptionTrainer {
|
||||
// replay (third+). The captured graph records all
|
||||
// in-graph kernel decisions at capture time per
|
||||
// `pearl_no_host_branches_in_captured_graph`.
|
||||
if self.train_graph.is_some() {
|
||||
if self.train_graph.is_some() && !self.mega_graph_enabled {
|
||||
self.train_graph
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
@@ -3245,6 +3253,9 @@ impl PerceptionTrainer {
|
||||
self.dispatch_train_step(b_sz, k_seq, total_snaps)
|
||||
.context("train warmup dispatch")?;
|
||||
self.cublas_warmed = true;
|
||||
} else if self.mega_graph_enabled {
|
||||
self.dispatch_train_step(b_sz, k_seq, total_snaps)
|
||||
.context("train eager dispatch (mega-graph mode)")?;
|
||||
} else {
|
||||
// Event tracking was disabled at trainer construction; the
|
||||
// trainer's CudaSlices have no read/write events, so neither
|
||||
@@ -3886,7 +3897,7 @@ impl PerceptionTrainer {
|
||||
// Three-state machine for the training graph — identical to
|
||||
// step_batched but dispatches `dispatch_train_step_no_scatter`
|
||||
// (skips the AoS→SoA scatter since SoA buffers are pre-filled).
|
||||
if self.train_graph.is_some() {
|
||||
if self.train_graph.is_some() && !self.mega_graph_enabled {
|
||||
self.train_graph
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
@@ -3896,6 +3907,11 @@ impl PerceptionTrainer {
|
||||
self.dispatch_train_step_no_scatter(b_sz, k_seq, total_snaps)
|
||||
.context("train warmup dispatch (from_device)")?;
|
||||
self.cublas_warmed = true;
|
||||
} else if self.mega_graph_enabled {
|
||||
// Mega-graph mode: always dispatch eagerly — the mega-graph
|
||||
// in the integrated trainer captures these kernel launches.
|
||||
self.dispatch_train_step_no_scatter(b_sz, k_seq, total_snaps)
|
||||
.context("train eager dispatch (mega-graph mode)")?;
|
||||
} else {
|
||||
let begin = self.stream.begin_capture(
|
||||
CUstreamCaptureMode::CU_STREAM_CAPTURE_MODE_RELAXED,
|
||||
@@ -4090,7 +4106,7 @@ impl PerceptionTrainer {
|
||||
// warmup → capture → replay with a new graph that excludes the
|
||||
// scatter. Let's use this approach since it's zero overhead
|
||||
// after the second call.
|
||||
if self.forward_graph_no_scatter.is_some() {
|
||||
if self.forward_graph_no_scatter.is_some() && !self.mega_graph_enabled {
|
||||
self.forward_graph_no_scatter
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
@@ -4100,6 +4116,10 @@ impl PerceptionTrainer {
|
||||
self.dispatch_forward_kernels_no_scatter(b_sz, k_seq, total_snaps)
|
||||
.context("forward_no_scatter warmup dispatch")?;
|
||||
self.forward_no_scatter_warmed = true;
|
||||
} else if self.mega_graph_enabled {
|
||||
// Mega-graph mode: always dispatch eagerly.
|
||||
self.dispatch_forward_kernels_no_scatter(b_sz, k_seq, total_snaps)
|
||||
.context("forward_no_scatter eager dispatch (mega-graph mode)")?;
|
||||
} else {
|
||||
use cudarc::driver::sys::{CUgraphInstantiate_flags, CUstreamCaptureMode};
|
||||
let begin = self.stream.begin_capture(
|
||||
|
||||
@@ -1433,6 +1433,43 @@ impl ml_alpha::rl::reward::RlLobBackend for LobSimCuda {
|
||||
ask_sz_src,
|
||||
)
|
||||
}
|
||||
|
||||
fn raw_ptrs(&self) -> ml_alpha::rl::reward::LobSimRawPtrs {
|
||||
ml_alpha::rl::reward::LobSimRawPtrs {
|
||||
bid_px: self.bid_px_d.raw_ptr(),
|
||||
bid_sz: self.bid_sz_d.raw_ptr(),
|
||||
ask_px: self.ask_px_d.raw_ptr(),
|
||||
ask_sz: self.ask_sz_d.raw_ptr(),
|
||||
books: self.books_d.raw_ptr(),
|
||||
prev_mid: self.prev_mid_d.raw_ptr(),
|
||||
atr_mid_ema: self.atr_mid_ema_d.raw_ptr(),
|
||||
snapshots_skipped: self.snapshots_skipped_d.raw_ptr(),
|
||||
min_reasonable_px: self.min_reasonable_px_d.raw_ptr(),
|
||||
max_reasonable_px: self.max_reasonable_px_d.raw_ptr(),
|
||||
book_update_fn: self.book_update_fn.cu_function(),
|
||||
market_targets: self.market_targets_d.raw_ptr(),
|
||||
pos: self.pos_d.raw_ptr(),
|
||||
cost_per_lot_per_side: self.cost_per_lot_per_side_d.raw_ptr(),
|
||||
total_fees_per_b: self.total_fees_per_b_d.raw_ptr(),
|
||||
submit_market_fn: self.submit_market_fn.cu_function(),
|
||||
open_trade_state: self.open_trade_state_d.raw_ptr(),
|
||||
trade_log: self.trade_log_d.raw_ptr(),
|
||||
trade_log_head: self.trade_log_head_d.raw_ptr(),
|
||||
trail_hwm: self.trail_hwm_d.raw_ptr(),
|
||||
zero_vwap_at_open: self.zero_vwap_at_open_d.raw_ptr(),
|
||||
saturated_vwap_at_open: self.saturated_vwap_at_open_d.raw_ptr(),
|
||||
defensive_exit_clamp: self.defensive_exit_clamp_d.raw_ptr(),
|
||||
conv_signed_ema: self.conv_signed_ema_d.raw_ptr(),
|
||||
diag_hold_hist: self.diag_hold_hist_d.raw_ptr(),
|
||||
diag_outcome_n: self.diag_outcome_n_d.raw_ptr(),
|
||||
diag_outcome_sum_pnl: self.diag_outcome_sum_pnl_d.raw_ptr(),
|
||||
diag_outcome_n_wins: self.diag_outcome_n_wins_d.raw_ptr(),
|
||||
pnl_track_fn: self.pnl_track_fn.cu_function(),
|
||||
n_backtests: self.n_backtests as i32,
|
||||
pos_bytes: std::mem::size_of::<crate::lob::PosFlat>() as i32,
|
||||
trade_log_cap: crate::lob::TRADE_LOG_CAP as i32,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Re-open the impl block for `LobSimCuda` so subsequent methods (if any
|
||||
|
||||
129
docs/superpowers/specs/2026-05-27-mega-graph-pipeline.md
Normal file
129
docs/superpowers/specs/2026-05-27-mega-graph-pipeline.md
Normal file
@@ -0,0 +1,129 @@
|
||||
# Mega-Graph CUDA Pipeline
|
||||
|
||||
**Date**: 2026-05-27
|
||||
**Status**: Draft
|
||||
**Goal**: Capture the entire per-step training pipeline into one CUDA graph launch, eliminating ~150 individual kernel launches and ~140ms of host-side overhead. Target: 50+ sps (from 6.4).
|
||||
|
||||
## Problem
|
||||
|
||||
nsys profiling shows GPU utilization at 8.5% (13ms kernel time per 156ms step). The host spends 143ms between kernel launches on: Rust arg assembly, Result unwrapping, conditional logic, 5 graph launches + ~20 individual raw_launch calls. The CUDA APIs themselves are fast (<1ms total); the overhead is Rust code between them.
|
||||
|
||||
## Current Per-Step Pipeline
|
||||
|
||||
```
|
||||
HOST GPU (13ms total)
|
||||
───────────────────────────── ────────────────────────────
|
||||
build 3 gather args (50μs) → sample_and_gather (13ms)
|
||||
gather_next (0.04ms)
|
||||
gather_current (0.04ms)
|
||||
graph_launch prefill (50μs) → GRAPH A: encoder + Q/V/π + actions (~2ms)
|
||||
build 4 gate args (100μs) → conf_gate + frd_gate + action_ent + log_pi (0.1ms)
|
||||
lobsim.apply_snapshot (20μs) → book_update (0.01ms)
|
||||
graph_launch postfill (50μs) → GRAPH A2: risk + trail + actions_to_market (~1ms)
|
||||
lobsim.step_fill (20μs) → fill + pnl_track (0.05ms)
|
||||
graph_launch reward (50μs) → GRAPH C: reward + scale + clamp + context (~2ms)
|
||||
build 2 PER args (20μs) → per_push_ring + per_push_flush (0.05ms)
|
||||
perception.step_batched (10μs)→ PERCEPTION GRAPH: encoder train (~2ms)
|
||||
graph_launch training (50μs) → GRAPH D: Q/π/V backward + Adam (~3ms)
|
||||
build 4 target args (50μs) → soft_update + q_divergence (0.1ms)
|
||||
DiagFrame memcpy (100μs) → (async, background thread)
|
||||
───────────────────────────
|
||||
Total host: ~570μs arg work
|
||||
But Rust overhead: ~155ms (!)
|
||||
```
|
||||
|
||||
The 570μs of pure arg work can't explain 155ms. The remaining ~154ms is:
|
||||
- `step_synthetic()` → reads ISV from mapped-pinned → launches LR controller → builds training graph args → launches GRAPH D → reads ISV for LR plateau
|
||||
- `step_with_lobsim_reward_and_train()` → builds reward graph args → conditional K-loop → event-based cross-stream sync → PER sample on train_stream
|
||||
|
||||
Each of these involves deep Rust function call chains (integrated.rs is 8400 lines) with Result unwrapping, debug_asserts, conditional branches, and method dispatch through trait objects.
|
||||
|
||||
## Approaches
|
||||
|
||||
### Approach A: Mega-Graph (Recommended)
|
||||
|
||||
Capture ALL kernels from a single step into ONE graph during warmup. On subsequent steps, launch the mega-graph with a single `cuGraphLaunch` call. Zero host work between kernels.
|
||||
|
||||
**Requirements**:
|
||||
1. ALL kernel arguments use stable device pointers (pre-allocated at init)
|
||||
2. No host-side conditional logic during the captured section
|
||||
3. No memory allocation/deallocation during capture
|
||||
4. K-loop fixed at K=1 (current config)
|
||||
5. cuBLAS workspace pre-allocated (already done)
|
||||
|
||||
**What changes per step** (must be handled):
|
||||
- `sample_and_gather` uses per-step PRNG state → device-side, graph-safe
|
||||
- Perception's `ts_ns` value → write to mapped-pinned staging, kernel reads from stable pointer
|
||||
- ISV bus → modified by controller kernels in-place, graph-safe
|
||||
- LR controller → reads host-side loss values → **BLOCKER**: needs mapped-pinned reads
|
||||
|
||||
**LR controller blocker**: The host reads `last_q_loss`, `last_pi_loss`, `last_v_loss` from mapped-pinned memory between the reward graph and the training graph. These feed the per-head LR scaling. Fix: move LR controller to a GPU kernel that reads from mapped-pinned device pointers directly. The host doesn't need to see the LR values — the Adam kernel reads LR from ISV.
|
||||
|
||||
**Lobsim blocker**: `apply_snapshot_from_device` and `step_fill_from_market_targets` are lobsim methods with internal state. These need to be either: (a) converted to raw_launch with cached pointers, or (b) kept outside the mega-graph as a "lobsim mini-graph."
|
||||
|
||||
**Estimated result**: 1 graph launch per step. Host does: graph launch (50μs) + DiagFrame memcpy (100μs) = 150μs/step → **~6600 sps** (GPU-bound at 13ms/step → ~77 sps with GPU saturation).
|
||||
|
||||
### Approach B: Coalesce Existing Graphs
|
||||
|
||||
Merge the 4 existing graphs + loose kernels into 2 larger graphs: "env_graph" (A + gates + A2 + fill + C) and "train_graph" (PER + perception + D + target).
|
||||
|
||||
**Simpler**: doesn't require solving the LR controller blocker or lobsim abstraction. Just expand the existing capture boundaries.
|
||||
|
||||
**Host work**: 2 graph launches + cross-stream event + DiagFrame = ~300μs. But the Rust overhead between the two launches is still significant (K-loop, LR reads, etc.).
|
||||
|
||||
**Estimated result**: ~20-30 sps (2× improvement, not 10×).
|
||||
|
||||
### Approach C: Async Host Pipeline
|
||||
|
||||
Keep the current graph structure but pipeline: host launches step N+1's gathers while step N's training graph runs. Double-buffer all state.
|
||||
|
||||
**Complex**: requires double-buffering 50+ device buffers. Weight updates from step N must be visible to step N+1's forward. Correctness is hard to verify.
|
||||
|
||||
**Estimated result**: ~12 sps (2× from pipelining).
|
||||
|
||||
## Recommendation: Approach A (Mega-Graph)
|
||||
|
||||
The only approach that reaches 50+ sps. The blockers are solvable:
|
||||
|
||||
1. **LR controller → GPU kernel**: already ISV-driven; just move the host-side `AdamW.lr` mutation to a kernel that writes the LR value into the Adam kernel's arg buffer. ~30 lines.
|
||||
|
||||
2. **Lobsim → raw_launch**: `apply_snapshot_from_device` is 2 DtoD copies + 1 kernel. `step_fill_from_market_targets` is 2 kernels. Convert both to raw_launch with cached pointers. ~100 lines.
|
||||
|
||||
3. **K-loop fixed at K=1**: already the case in production config. The mega-graph captures one iteration.
|
||||
|
||||
4. **Cross-stream PER**: PER sample runs on train_stream, but at K=1 it's just 1 sample per step. Can be moved to the main stream (the separate stream was for K>1 pipelining).
|
||||
|
||||
## Implementation Phases
|
||||
|
||||
### Phase 1: Convert lobsim methods to raw_launch (prerequisite)
|
||||
- `apply_snapshot_from_device` → 2 DtoD + 1 kernel via raw_launch
|
||||
- `step_fill_from_market_targets` → 2 kernels via raw_launch
|
||||
- Cache all lobsim internal pointers in LobPtrs
|
||||
|
||||
### Phase 2: Move LR controller to GPU kernel
|
||||
- New `rl_lr_from_mapped_pinned.cu` kernel reads loss from mapped-pinned device pointers
|
||||
- Writes per-head LR to ISV slots
|
||||
- Adam kernel reads LR from ISV (already supported)
|
||||
|
||||
### Phase 3: Move PER sample to main stream
|
||||
- At K=1, PER sample + priority_update + tree_rebuild run once per step
|
||||
- Move from train_stream to self.stream
|
||||
- Eliminate cross-stream events
|
||||
|
||||
### Phase 4: Mega-graph capture
|
||||
- Warmup step 0: eager execution (all kernels launch individually)
|
||||
- Warmup step 1: `begin_capture` → entire pipeline → `end_capture`
|
||||
- Step 2+: single `cuGraphLaunch` per step
|
||||
- Host per-step: ts_ns write to mapped-pinned + graph launch + DiagFrame
|
||||
|
||||
### Phase 5: Validation
|
||||
- nsys: confirm 1 cuGraphLaunch per step, GPU utilization >80%
|
||||
- Functional: wr/pnl match baseline within 1%
|
||||
- Smoke: 1k steps local, 5k steps L40S
|
||||
|
||||
## Anti-patterns to avoid
|
||||
|
||||
- No `cuGraphExecUpdate` (overkill — our topology is fixed)
|
||||
- No conditional graph nodes (CUDA 12.4 feature, complex, fragile)
|
||||
- No multi-stream graphs (correctness nightmare)
|
||||
- No host reads inside the captured section (use mapped-pinned reads OUTSIDE or defer)
|
||||
Reference in New Issue
Block a user