use cudarc::driver::PushKernelArg;
use crate::config::MambaConfig;
use crate::mamba_ssm::gpu::adamw::{AdamWBiasFactors, GpuAdamW, step_m1_capturable};
use crate::mamba_ssm::gpu::backward::gpu_backward_mamba_backbone;
use crate::mamba_ssm::gpu::backward_mixed::gpu_backward_mamba_backbone_mixed;
use crate::mamba_ssm::gpu::buffers::{GpuBuffer, GpuByteBuffer};
use crate::mamba_ssm::gpu::context::GpuCtx;
use crate::mamba_ssm::gpu::device::GpuDevice;
use crate::mamba_ssm::gpu::dtype::WeightDtype;
use crate::mamba_ssm::gpu::forward::{
GpuMambaBackboneActs, GpuMambaDims, GpuMambaScratch, GpuRecurrentState,
gpu_forward_mamba_backbone,
};
use crate::mamba_ssm::gpu::forward_mixed::{
GpuMambaBackboneMixedActs, GpuMambaMixedTrainScratch, gpu_forward_mamba_backbone_train_mixed,
};
use crate::mamba_ssm::gpu::grad_clip::{
GRAD_CLIP_PARTIALS, alloc_partials, global_grad_norm, scale_grads,
};
use crate::mamba_ssm::gpu::graph_capture::capture_into_graph;
use crate::mamba_ssm::gpu::launch::grid_1d;
use crate::mamba_ssm::gpu::weights::GpuMambaTrainLayerWeights;
fn recompute_a_neg_all(
ctx: &GpuCtx,
master_layers: &[GpuMambaTrainLayerWeights],
a_neg_all: &crate::mamba_ssm::gpu::buffers::GpuBuffer,
state_a_neg_all: &crate::mamba_ssm::gpu::buffers::GpuBuffer,
d_inner: usize,
d_state: usize,
) -> Result<(), String> {
let per_layer = d_inner * d_state;
if per_layer == 0 {
return Ok(());
}
let n_i32 = per_layer as i32;
for (li, mw) in master_layers.iter().enumerate() {
let src = mw.a_log.cached_ptr();
let dst_a = a_neg_all.inner_at(li * per_layer);
let mut b1 = ctx.stream.launch_builder(&ctx.kernels.exp_negate);
b1.arg(&dst_a);
b1.arg(&src);
b1.arg(&n_i32);
unsafe { b1.launch(grid_1d(per_layer)) }
.map_err(|e| format!("exp_negate self.a_neg_all L{li}: {e:?}"))?;
let dst_s = state_a_neg_all.inner_at(li * per_layer);
let mut b2 = ctx.stream.launch_builder(&ctx.kernels.exp_negate);
b2.arg(&dst_s);
b2.arg(&src);
b2.arg(&n_i32);
unsafe { b2.launch(grid_1d(per_layer)) }
.map_err(|e| format!("exp_negate state.a_neg_all L{li}: {e:?}"))?;
}
Ok(())
}
use crate::mamba_ssm::gpu::loss_scaler::{
DynamicLossScaler, OverflowFlag, UnscaleFactor, check_inf_nan_gpu, scale_grads_skip_gpu,
};
use crate::mamba_ssm::gpu::training_graph::{
GpuMambaF32TrainingStepGraph, GpuMambaTrainingStepGraph, MambaF32Capture, MambaF32Replay,
MambaMixedCapture, MambaMixedReplay,
};
use crate::mamba_ssm::gpu::weights::{GpuMambaGrads, GpuMambaTrainWeights};
use crate::mamba_ssm::gpu::weights_mixed_train::GpuMambaTrainMixedWeights;
use crate::weights::MambaWeights;
#[derive(Clone, Copy, Debug)]
pub struct TrainSessionCfg {
pub input_dim: usize,
pub batch: usize,
pub seq_len: usize,
pub lr: f32,
pub weight_decay: f32,
}
#[derive(Debug, Clone)]
pub struct StepMetrics {
pub step: u64,
pub graph_replayed: bool,
pub loss_scale: Option<f32>,
pub overflow_skipped: Option<bool>,
}
impl StepMetrics {
pub fn plain(step: u64, graph_replayed: bool) -> Self {
Self {
step,
graph_replayed,
loss_scale: None,
overflow_skipped: None,
}
}
}
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Default)]
pub struct BackwardOpts {
pub clip_max_norm: Option<f32>,
pub accumulate_only: bool,
}
impl BackwardOpts {
pub fn with_clip_max_norm(mut self, c: f32) -> Self {
self.clip_max_norm = Some(c);
self
}
pub fn with_accumulate_only(mut self, on: bool) -> Self {
self.accumulate_only = on;
self
}
}
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct BackwardMetrics {
pub step: u64,
pub optimizer_stepped: bool,
pub grad_norm: Option<f32>,
pub loss_scale: Option<f32>,
pub overflow_skipped: Option<bool>,
}
enum TrainerInner {
F32(Box<MambaTrainerF32>),
Mixed(Box<MambaTrainerMixed>),
}
pub struct MambaTrainer {
inner: TrainerInner,
}
impl MambaTrainer {
pub fn new_with_dtype(
gpu_ordinal: usize,
cpu_weights: &MambaWeights,
cfg: MambaConfig,
input_dim: usize,
batch: usize,
seq_len: usize,
dtype: WeightDtype,
) -> Result<Self, String> {
Self::new_full(
gpu_ordinal,
cpu_weights,
cfg,
TrainSessionCfg {
input_dim,
batch,
seq_len,
lr: 1e-3,
weight_decay: 1e-2,
},
dtype,
)
}
pub fn new_full(
gpu_ordinal: usize,
cpu_weights: &MambaWeights,
cfg: MambaConfig,
session: TrainSessionCfg,
dtype: WeightDtype,
) -> Result<Self, String> {
super::launch::validate_kernel_arg_capacity(
session.batch,
session.seq_len,
cfg.d_inner(),
cfg.d_state,
)?;
let inner = match dtype {
WeightDtype::F32 => TrainerInner::F32(Box::new(MambaTrainerF32::new_full(
gpu_ordinal,
cpu_weights,
cfg,
session,
)?)),
WeightDtype::Bf16 | WeightDtype::F16 => TrainerInner::Mixed(Box::new(
MambaTrainerMixed::new_full(gpu_ordinal, cpu_weights, cfg, session, dtype)?,
)),
};
Ok(Self { inner })
}
pub fn dtype(&self) -> WeightDtype {
match &self.inner {
TrainerInner::F32(_) => WeightDtype::F32,
TrainerInner::Mixed(t) => t.dtype,
}
}
pub fn batch(&self) -> usize {
match &self.inner {
TrainerInner::F32(t) => t.batch,
TrainerInner::Mixed(t) => t.batch,
}
}
pub fn seq_len(&self) -> usize {
match &self.inner {
TrainerInner::F32(t) => t.seq_len,
TrainerInner::Mixed(t) => t.seq_len,
}
}
pub fn ctx(&self) -> &GpuCtx {
match &self.inner {
TrainerInner::F32(t) => &t.ctx,
TrainerInner::Mixed(t) => &t.ctx,
}
}
pub fn has_graph(&self) -> bool {
match &self.inner {
TrainerInner::F32(t) => t.graph.is_some(),
TrainerInner::Mixed(t) => t.has_graph(),
}
}
pub fn set_lr(&mut self, lr: f32) -> Result<(), String> {
if !lr.is_finite() || lr <= 0.0 {
return Err(format!("set_lr: invalid learning rate {lr}"));
}
if self.has_graph() {
return Err(
"set_lr under a captured graph: the lr is baked by value into the \
captured AdamW kernel and a field write would silently not apply — \
drop_graph() first, then set_lr, then re-capture"
.into(),
);
}
match &mut self.inner {
TrainerInner::F32(t) => t.adam.lr = lr,
TrainerInner::Mixed(t) => t.adam.lr = lr,
}
Ok(())
}
pub fn lr(&self) -> f32 {
match &self.inner {
TrainerInner::F32(t) => t.adam.lr,
TrainerInner::Mixed(t) => t.adam.lr,
}
}
pub fn set_reference_no_decay(&mut self, on: bool) -> Result<(), String> {
if self.has_graph() {
return Err(
"set_reference_no_decay under a captured graph: the decay coefficient \
is baked by value into the captured AdamW launches — drop_graph() \
first, then toggle, then re-capture"
.into(),
);
}
match &mut self.inner {
TrainerInner::F32(t) => t.adam.reference_no_decay = on,
TrainerInner::Mixed(t) => t.adam.reference_no_decay = on,
}
Ok(())
}
pub fn drop_graph(&mut self) {
match &mut self.inner {
TrainerInner::F32(t) => t.graph = None,
TrainerInner::Mixed(t) => {
t.graph = None;
t.graph_f16 = None;
}
}
}
pub fn reset_state(&mut self) -> Result<(), String> {
match &mut self.inner {
TrainerInner::F32(t) => t.reset_state(),
TrainerInner::Mixed(t) => t.reset_state(),
}
}
pub fn capture_graph(&mut self) -> Result<(), String> {
match &mut self.inner {
TrainerInner::F32(t) => t.capture_graph(),
TrainerInner::Mixed(t) => t.capture_graph(),
}
}
pub fn step(&mut self, input: &[f32], d_temporal: &[f32]) -> Result<StepMetrics, String> {
match &mut self.inner {
TrainerInner::F32(t) => t.step(input, d_temporal),
TrainerInner::Mixed(t) => t.step(input, d_temporal),
}
}
pub fn forward(&mut self, input: &[f32], temporal_out: &mut [f32]) -> Result<(), String> {
match &mut self.inner {
TrainerInner::F32(t) => t.forward_split(input, temporal_out),
TrainerInner::Mixed(t) => t.forward_split(input, temporal_out),
}
}
pub fn backward_step(
&mut self,
d_temporal: &[f32],
opts: BackwardOpts,
) -> Result<BackwardMetrics, String> {
match &mut self.inner {
TrainerInner::F32(t) => t.backward_split(d_temporal, opts),
TrainerInner::Mixed(t) => t.backward_split(d_temporal, opts),
}
}
pub fn snapshot_master(&self) -> Result<MambaWeights, String> {
match &self.inner {
TrainerInner::F32(t) => t.snapshot_master(),
TrainerInner::Mixed(t) => t.snapshot_master(),
}
}
#[doc(hidden)]
pub fn debug_a_neg_all(&self) -> Result<Vec<f32>, String> {
match &self.inner {
TrainerInner::F32(t) => t.debug_a_neg_all(),
TrainerInner::Mixed(t) => t.debug_a_neg_all(),
}
}
pub fn scaler_state(&self) -> Option<(f32, u32)> {
match &self.inner {
TrainerInner::F32(_) => None,
TrainerInner::Mixed(t) => t.scaler_state(),
}
}
pub fn load_scaler_state(&mut self, scale: f32, growth_tracker: u32) {
if let TrainerInner::Mixed(ref mut t) = self.inner {
t.load_scaler_state(scale, growth_tracker);
}
}
}
pub(crate) struct MambaTrainerMixed {
ctx: GpuCtx,
cfg: MambaConfig,
batch: usize,
seq_len: usize,
dtype: WeightDtype,
pub weights: GpuMambaTrainMixedWeights,
pub grads: GpuMambaGrads,
pub adam: GpuAdamW,
bias: AdamWBiasFactors,
acts: GpuMambaBackboneMixedActs,
scratch: GpuMambaMixedTrainScratch,
state: GpuRecurrentState,
a_neg_all: GpuBuffer,
mamba_input: GpuBuffer,
d_temporal: GpuBuffer,
temporal_f32: GpuBuffer,
split_forward_pending: bool,
grads_dirty: bool,
clip_partials: GpuByteBuffer,
clip_partials_host: Vec<f64>,
graph: Option<GpuMambaTrainingStepGraph>,
scaler: Option<DynamicLossScaler>,
overflow_flag: Option<OverflowFlag>,
d_temporal_scaled: Option<GpuBuffer>,
graph_f16: Option<cudarc::driver::CudaGraph>,
unscale_factor: Option<UnscaleFactor>,
captured_f16_bias_ptr: u64,
captured_f16_unscale_ptr: u64,
captured_f16_overflow_ptr: u64,
captured_f16_grads_ptr: u64,
captured_f16_dt_scaled_ptr: u64,
}
impl MambaTrainerMixed {
fn new_full(
gpu_ordinal: usize,
cpu_weights: &MambaWeights,
cfg: MambaConfig,
session: TrainSessionCfg,
dtype: WeightDtype,
) -> Result<Self, String> {
let TrainSessionCfg {
input_dim,
batch,
seq_len,
lr,
weight_decay,
} = session;
assert!(
matches!(dtype, WeightDtype::Bf16 | WeightDtype::F16),
"MambaTrainerMixed accepts Bf16 or F16; got {dtype:?}"
);
let device = GpuDevice::new(gpu_ordinal)?;
let ctx = GpuCtx::new(&device)?;
let weights = GpuMambaTrainMixedWeights::from_cpu(&ctx.stream, cpu_weights, &cfg, dtype)?;
let d_inner = cfg.d_inner();
let d_state = cfg.d_state;
let d_conv = cfg.d_conv;
let n_layers = cfg.n_layers;
let dims = GpuMambaDims {
batch,
d_model: cfg.d_model,
d_inner,
d_state,
d_conv,
dt_rank: cfg.dt_rank(),
xdbl_dim: cfg.xdbl_dim(),
seq_len,
mamba_input_dim: input_dim,
n_layers,
scan_mode: cfg.scan_mode,
rms_norm_eps: cfg.rms_norm_eps,
};
let acts = GpuMambaBackboneMixedActs::new(&ctx.stream, &dims, dtype)?;
let scratch = GpuMambaMixedTrainScratch::new(&ctx.stream, &dims, dtype)?;
let mut a_neg_flat = vec![0.0f32; n_layers * d_inner * d_state];
for (l, lw) in cpu_weights.layers.iter().enumerate() {
for i in 0..d_inner * d_state {
a_neg_flat[l * d_inner * d_state + i] = -lw.a_log[i].exp();
}
}
let mut a_neg_all = GpuBuffer::zeros(&ctx.stream, n_layers * d_inner * d_state)?;
a_neg_all.upload(&ctx.stream, &a_neg_flat)?;
let mut state = GpuRecurrentState {
conv_states: GpuBuffer::zeros(&ctx.stream, n_layers * batch * d_inner * d_conv)?,
ssm_states: GpuBuffer::zeros(&ctx.stream, n_layers * batch * d_inner * d_state)?,
a_neg_all: GpuBuffer::zeros(&ctx.stream, n_layers * d_inner * d_state)?,
};
state.a_neg_all.upload(&ctx.stream, &a_neg_flat)?;
let mamba_input = GpuBuffer::zeros(&ctx.stream, batch * seq_len * input_dim)?;
let d_temporal = GpuBuffer::zeros(&ctx.stream, batch * seq_len * cfg.d_model)?;
let temporal_f32 = GpuBuffer::zeros(&ctx.stream, batch * seq_len * cfg.d_model)?;
let grads = GpuMambaGrads::new(&ctx.stream, &cfg, input_dim)?;
let adam = GpuAdamW::new(&ctx.stream, grads.flat.len())?
.with_lr(lr)
.with_weight_decay(weight_decay);
let bias = AdamWBiasFactors::new(&ctx.stream)?;
let (scaler, overflow_flag, d_temporal_scaled, unscale_factor) =
if matches!(dtype, WeightDtype::F16) {
let s = DynamicLossScaler::new();
let f = OverflowFlag::new(&ctx.stream)?;
let scaled = GpuBuffer::zeros(&ctx.stream, batch * seq_len * cfg.d_model)?;
let u = UnscaleFactor::new(&ctx.stream)?;
(Some(s), Some(f), Some(scaled), Some(u))
} else {
(None, None, None, None)
};
ctx.stream
.synchronize()
.map_err(|e| format!("sync: {e:?}"))?;
let clip_partials = alloc_partials(&ctx.stream)?;
Ok(Self {
ctx,
cfg,
batch,
seq_len,
dtype,
weights,
grads,
adam,
bias,
acts,
scratch,
state,
a_neg_all,
mamba_input,
d_temporal,
temporal_f32,
split_forward_pending: false,
grads_dirty: false,
clip_partials,
clip_partials_host: vec![0.0; GRAD_CLIP_PARTIALS],
graph: None,
scaler,
overflow_flag,
d_temporal_scaled,
graph_f16: None,
unscale_factor,
captured_f16_bias_ptr: 0,
captured_f16_unscale_ptr: 0,
captured_f16_overflow_ptr: 0,
captured_f16_grads_ptr: 0,
captured_f16_dt_scaled_ptr: 0,
})
}
pub fn has_graph(&self) -> bool {
self.graph.is_some() || self.graph_f16.is_some()
}
pub fn reset_state(&mut self) -> Result<(), String> {
self.state.conv_states.zero(&self.ctx.stream)?;
self.state.ssm_states.zero(&self.ctx.stream)?;
Ok(())
}
#[doc(hidden)]
pub fn debug_a_neg_all(&self) -> Result<Vec<f32>, String> {
self.ctx
.stream
.synchronize()
.map_err(|e| format!("debug_a_neg_all sync: {e:?}"))?;
self.a_neg_all.to_cpu(&self.ctx.stream)
}
pub fn scaler_state(&self) -> Option<(f32, u32)> {
self.scaler.as_ref().map(|s| s.state())
}
pub fn load_scaler_state(&mut self, scale: f32, growth_tracker: u32) {
if let Some(ref mut s) = self.scaler {
s.load_state(scale, growth_tracker);
if let Some(ref mut uf) = self.unscale_factor {
let unscale = 1.0 / s.scale();
let _ = uf.write(&self.ctx.stream, unscale);
}
}
}
pub fn capture_graph(&mut self) -> Result<(), String> {
if matches!(self.dtype, WeightDtype::F16) {
return self.capture_graph_f16();
}
self.bias.write(&self.ctx.stream, 1.0, 1.0)?;
let g = GpuMambaTrainingStepGraph::capture(
&self.ctx,
&self.cfg,
MambaMixedCapture {
train_w: &mut self.weights,
adam: &self.adam,
bias: &self.bias,
grads: &mut self.grads,
acts: &mut self.acts,
scratch: &mut self.scratch,
a_neg_all: &self.a_neg_all,
mamba_input: &self.mamba_input,
d_temporal: &mut self.d_temporal,
state: &mut self.state,
},
self.batch,
self.seq_len,
)?;
self.graph = Some(g);
Ok(())
}
pub fn step(&mut self, input: &[f32], d_temporal: &[f32]) -> Result<StepMetrics, String> {
assert_eq!(
input.len(),
self.mamba_input.len(),
"input shape mismatch: expected {} got {}",
self.mamba_input.len(),
input.len()
);
assert_eq!(
d_temporal.len(),
self.d_temporal.len(),
"d_temporal shape mismatch: expected {} got {}",
self.d_temporal.len(),
d_temporal.len()
);
if self.grads_dirty {
return Err(
"step(): an accumulate_only backward window is open — close it with \
backward_step(accumulate_only=false); the fused step zeroes the grad \
arena and would silently discard the accumulated gradients"
.into(),
);
}
if matches!(self.dtype, WeightDtype::F16) {
return self.step_f16(input, d_temporal);
}
self.mamba_input.upload(&self.ctx.stream, input)?;
self.d_temporal.upload(&self.ctx.stream, d_temporal)?;
let (step, bc1, bc2) = self.adam.advance();
self.bias.write(&self.ctx.stream, bc1, bc2)?;
let replayed = if let Some(ref g) = self.graph {
g.replay(
&self.ctx,
&MambaMixedReplay {
train_w: &self.weights,
adam: &self.adam,
bias: &self.bias,
grads: &self.grads,
a_neg_all: &self.a_neg_all,
mamba_input: &self.mamba_input,
d_temporal: &self.d_temporal,
state: &self.state,
},
)?;
true
} else {
self.step_eager()?;
false
};
Ok(StepMetrics::plain(step, replayed))
}
pub(crate) fn forward_split(
&mut self,
input: &[f32],
temporal_out: &mut [f32],
) -> Result<(), String> {
assert_eq!(
input.len(),
self.mamba_input.len(),
"input shape mismatch: expected batch*seq_len*input_dim={}, got {}",
self.mamba_input.len(),
input.len(),
);
assert_eq!(
temporal_out.len(),
self.temporal_f32.len(),
"temporal_out shape mismatch: expected batch*seq_len*d_model={}, got {}",
self.temporal_f32.len(),
temporal_out.len(),
);
self.mamba_input.upload(&self.ctx.stream, input)?;
self.eager_forward()?;
{
let dst = self.temporal_f32.cached_ptr();
let src = self.scratch.temporal_typed.cached_ptr();
let kernel = match self.scratch.temporal_typed.dtype() {
WeightDtype::Bf16 => &self.ctx.kernels.cast_bf16_to_f32,
WeightDtype::F16 => &self.ctx.kernels.cast_f16_to_f32,
WeightDtype::F32 => {
return Err("forward_split: unexpected f32 temporal_typed in Mixed".into());
}
};
let n = self.temporal_f32.len() as i32;
let mut b = self.ctx.stream.launch_builder(kernel);
b.arg(&dst);
b.arg(&src);
b.arg(&n);
unsafe { b.launch(grid_1d(self.temporal_f32.len())) }
.map_err(|e| format!("forward_split: temporal upcast: {e:?}"))?;
}
self.ctx
.stream
.synchronize()
.map_err(|e| format!("forward_split sync: {e:?}"))?;
self.temporal_f32.download(&self.ctx.stream, temporal_out)?;
self.split_forward_pending = true;
Ok(())
}
pub(crate) fn backward_split(
&mut self,
d_temporal: &[f32],
opts: BackwardOpts,
) -> Result<BackwardMetrics, String> {
if !self.split_forward_pending {
return Err(
"backward_step() without a pending forward() — the saved activations \
are stale or missing; call forward() first"
.into(),
);
}
if opts.clip_max_norm.is_some() && opts.accumulate_only {
return Err(
"clip_max_norm + accumulate_only is unsupported: the global norm is only \
defined over the COMPLETE accumulated gradient — request the clip on the \
final (applying) backward_step"
.into(),
);
}
assert_eq!(
d_temporal.len(),
self.d_temporal.len(),
"d_temporal shape mismatch: expected batch*seq_len*d_model={}, got {}",
self.d_temporal.len(),
d_temporal.len(),
);
if matches!(self.dtype, WeightDtype::F16) {
if opts.accumulate_only {
return Err(
"f16 + accumulate_only is unsupported: the loss-scale freeze window \
across micro-batches has no defined semantics"
.into(),
);
}
let m = self.backward_split_f16(d_temporal, opts.clip_max_norm)?;
self.split_forward_pending = false;
return Ok(m);
}
self.d_temporal.upload(&self.ctx.stream, d_temporal)?;
if !self.grads_dirty {
self.grads.zero(&self.ctx.stream)?;
}
self.eager_backward(false)?;
self.split_forward_pending = false;
if opts.accumulate_only {
self.grads_dirty = true;
Ok(BackwardMetrics {
step: self.adam.step,
optimizer_stepped: false,
grad_norm: None,
loss_scale: None,
overflow_skipped: None,
})
} else {
let grad_norm = match opts.clip_max_norm {
Some(c) => Some(self.apply_clip(c)?),
None => None,
};
let (step, bc1, bc2) = self.adam.advance();
self.bias.write(&self.ctx.stream, bc1, bc2)?;
self.eager_optimize()?;
self.grads_dirty = false;
Ok(BackwardMetrics {
step,
optimizer_stepped: true,
grad_norm,
loss_scale: None,
overflow_skipped: None,
})
}
}
fn apply_clip(&mut self, max_norm: f32) -> Result<f32, String> {
let norm = global_grad_norm(
&self.ctx,
&self.grads.flat,
&mut self.clip_partials,
&mut self.clip_partials_host,
)?;
if !norm.is_finite() {
return Err(format!(
"clip_max_norm: non-finite global grad norm ({norm})"
));
}
let coef = max_norm as f64 / (norm + 1e-6);
if coef < 1.0 {
scale_grads(&self.ctx, &mut self.grads.flat, coef as f32)?;
}
Ok(norm as f32)
}
fn backward_split_f16(
&mut self,
d_temporal: &[f32],
clip_max_norm: Option<f32>,
) -> Result<BackwardMetrics, String> {
let scale = self.scaler.as_ref().expect("f16 scaler").scale();
{
let dt_scaled = self.d_temporal_scaled.as_mut().expect("f16 dt_scaled");
dt_scaled.upload(&self.ctx.stream, d_temporal)?;
let n = d_temporal.len() as i32;
let mut builder = self
.ctx
.stream
.launch_builder(&self.ctx.kernels.scale_grads_f32);
builder.arg(dt_scaled.inner_mut());
builder.arg(&scale);
builder.arg(&n);
unsafe { builder.launch(grid_1d(d_temporal.len())) }
.map_err(|e| format!("scale d_temporal (f16 split): {e:?}"))?;
}
if let Some(ref mut u) = self.unscale_factor {
u.write(&self.ctx.stream, 1.0 / scale)?;
}
let prev_step = self.adam.step;
let (next_step, bc1, bc2) = self.adam.advance();
self.bias.write(&self.ctx.stream, bc1, bc2)?;
self.overflow_flag
.as_mut()
.expect("f16 overflow flag")
.zero(&self.ctx.stream)?;
self.grads.zero(&self.ctx.stream)?;
self.eager_backward(true)?;
check_inf_nan_gpu(
&self.ctx,
&self.ctx.kernels,
self.overflow_flag.as_mut().unwrap(),
&self.grads.flat,
)?;
let overflow = self
.overflow_flag
.as_ref()
.unwrap()
.read(&self.ctx.stream)?
!= 0;
let mut grad_norm = None;
if !overflow {
let unscale = self.unscale_factor.as_ref().expect("unscale buf");
scale_grads_skip_gpu(
&self.ctx,
&self.ctx.kernels,
self.overflow_flag.as_mut().unwrap(),
&mut self.grads.flat,
unscale,
)?;
if let Some(c) = clip_max_norm {
grad_norm = Some(self.apply_clip(c)?);
}
self.eager_optimize()?;
}
self.scaler.as_mut().expect("f16 scaler").update(overflow);
let final_step = if overflow {
self.adam.step = prev_step;
prev_step
} else {
next_step
};
Ok(BackwardMetrics {
step: final_step,
optimizer_stepped: !overflow,
grad_norm,
loss_scale: Some(scale),
overflow_skipped: Some(overflow),
})
}
fn step_f16(&mut self, input: &[f32], d_temporal: &[f32]) -> Result<StepMetrics, String> {
let scale = self.scaler.as_ref().expect("f16 scaler").scale();
self.mamba_input.upload(&self.ctx.stream, input)?;
{
let dt_scaled = self.d_temporal_scaled.as_mut().expect("f16 dt_scaled");
dt_scaled.upload(&self.ctx.stream, d_temporal)?;
let n = d_temporal.len() as i32;
let mut builder = self
.ctx
.stream
.launch_builder(&self.ctx.kernels.scale_grads_f32);
builder.arg(dt_scaled.inner_mut());
builder.arg(&scale);
builder.arg(&n);
unsafe { builder.launch(crate::mamba_ssm::gpu::launch::grid_1d(d_temporal.len())) }
.map_err(|e| format!("scale d_temporal (f16): {e:?}"))?;
}
if let Some(ref mut u) = self.unscale_factor {
u.write(&self.ctx.stream, 1.0 / scale)?;
}
let prev_step = self.adam.step;
let (next_step, bc1, bc2) = self.adam.advance();
self.bias.write(&self.ctx.stream, bc1, bc2)?;
self.overflow_flag
.as_mut()
.expect("f16 overflow flag")
.zero(&self.ctx.stream)?;
let (step, overflow, replayed) = if let Some(ref g) = self.graph_f16 {
assert_eq!(
self.bias.ptr(),
self.captured_f16_bias_ptr,
"f16 graph replay: bias pointer changed since capture"
);
assert_eq!(
self.unscale_factor.as_ref().unwrap().ptr(),
self.captured_f16_unscale_ptr,
"f16 graph replay: unscale_factor pointer changed since capture"
);
assert_eq!(
self.overflow_flag
.as_ref()
.unwrap()
.stable_ptr(&self.ctx.stream),
self.captured_f16_overflow_ptr,
"f16 graph replay: overflow_flag pointer changed since capture"
);
assert_eq!(
self.grads.flat.cached_ptr(),
self.captured_f16_grads_ptr,
"f16 graph replay: grads.flat pointer changed since capture"
);
assert_eq!(
self.d_temporal_scaled.as_ref().unwrap().cached_ptr(),
self.captured_f16_dt_scaled_ptr,
"f16 graph replay: d_temporal_scaled pointer changed since capture"
);
g.launch().map_err(|e| format!("f16 graph launch: {e:?}"))?;
let overflow = self
.overflow_flag
.as_ref()
.unwrap()
.read(&self.ctx.stream)?
!= 0;
self.scaler.as_mut().expect("f16 scaler").update(overflow);
(next_step, overflow, true)
} else {
self.grads.zero(&self.ctx.stream)?;
self.eager_forward()?;
self.eager_backward(true)?;
check_inf_nan_gpu(
&self.ctx,
&self.ctx.kernels,
self.overflow_flag.as_mut().unwrap(),
&self.grads.flat,
)?;
let overflow = self
.overflow_flag
.as_ref()
.unwrap()
.read(&self.ctx.stream)?
!= 0;
if !overflow {
let unscale = self.unscale_factor.as_ref().expect("unscale buf");
scale_grads_skip_gpu(
&self.ctx,
&self.ctx.kernels,
self.overflow_flag.as_mut().unwrap(),
&mut self.grads.flat,
unscale,
)?;
self.eager_optimize()?;
}
self.scaler.as_mut().expect("f16 scaler").update(overflow);
(next_step, overflow, false)
};
let final_step = if overflow {
self.adam.step = prev_step;
prev_step
} else {
step
};
Ok(StepMetrics {
step: final_step,
graph_replayed: replayed,
loss_scale: Some(scale),
overflow_skipped: Some(overflow),
})
}
fn capture_graph_f16(&mut self) -> Result<(), String> {
self.bias.write(&self.ctx.stream, 1.0, 1.0)?;
let init_unscale = 1.0 / self.scaler.as_ref().expect("f16 scaler").scale();
self.unscale_factor
.as_mut()
.expect("unscale buf")
.write(&self.ctx.stream, init_unscale)?;
self.overflow_flag
.as_mut()
.expect("overflow flag")
.zero(&self.ctx.stream)?;
let dummy = vec![0.0f32; self.d_temporal.len()];
self.d_temporal_scaled
.as_mut()
.expect("dt_scaled")
.upload(&self.ctx.stream, &dummy)?;
self.ctx
.presize_half_staging_for_train(&self.cfg, self.batch, self.seq_len, self.dtype)?;
let input_dim = self.mamba_input.len() / (self.batch * self.seq_len);
self.ctx.presize_bi_upcast_scratch_for_train_with_input(
&self.cfg,
self.batch,
self.seq_len,
input_dim,
self.dtype,
)?;
let snap_bias = self.bias.ptr();
let snap_unscale = self.unscale_factor.as_ref().unwrap().ptr();
let snap_overflow = self
.overflow_flag
.as_ref()
.unwrap()
.stable_ptr(&self.ctx.stream);
let snap_grads = self.grads.flat.cached_ptr();
let snap_dt_scaled = self.d_temporal_scaled.as_ref().unwrap().cached_ptr();
let stream = self.ctx.stream.clone();
let g = capture_into_graph(&stream, || {
self.grads.zero(&self.ctx.stream)?;
self.eager_forward()?;
self.eager_backward(true)?;
check_inf_nan_gpu(
&self.ctx,
&self.ctx.kernels,
self.overflow_flag.as_mut().unwrap(),
&self.grads.flat,
)?;
scale_grads_skip_gpu(
&self.ctx,
&self.ctx.kernels,
self.overflow_flag.as_mut().unwrap(),
&mut self.grads.flat,
self.unscale_factor.as_ref().unwrap(),
)?;
self.eager_optimize()?;
Ok(())
})?;
self.graph_f16 = Some(g);
self.captured_f16_bias_ptr = snap_bias;
self.captured_f16_unscale_ptr = snap_unscale;
self.captured_f16_overflow_ptr = snap_overflow;
self.captured_f16_grads_ptr = snap_grads;
self.captured_f16_dt_scaled_ptr = snap_dt_scaled;
Ok(())
}
fn eager_forward(&mut self) -> Result<(), String> {
gpu_forward_mamba_backbone_train_mixed(
&self.ctx,
&mut self.acts,
&self.weights,
&self.mamba_input,
&mut self.state,
&mut self.scratch,
)
}
fn eager_backward(&mut self, scaled: bool) -> Result<(), String> {
let d_temporal = if scaled {
self.d_temporal_scaled
.as_mut()
.expect("eager_backward(scaled): f16 d_temporal_scaled missing")
} else {
&mut self.d_temporal
};
gpu_backward_mamba_backbone_mixed(
&self.ctx,
d_temporal,
&self.grads,
&self.acts,
&self.weights.compute,
&self.a_neg_all,
&mut self.scratch,
)
}
fn eager_optimize(&mut self) -> Result<(), String> {
step_m1_capturable(
&self.ctx,
&self.ctx.kernels.adamw_step_f32_capturable,
&self.adam,
self.bias.ptr(),
&mut self.weights.master,
&self.grads,
)?;
self.weights.sync_master_to_compute(&self.ctx)?;
recompute_a_neg_all(
&self.ctx,
&self.weights.master.layers,
&self.a_neg_all,
&self.state.a_neg_all,
self.cfg.d_inner(),
self.cfg.d_state,
)
}
fn step_eager(&mut self) -> Result<(), String> {
self.grads.zero(&self.ctx.stream)?;
self.eager_forward()?;
self.eager_backward(false)?;
self.eager_optimize()
}
pub fn snapshot_master(&self) -> Result<MambaWeights, String> {
self.ctx
.stream
.synchronize()
.map_err(|e| format!("pre-snapshot sync: {e:?}"))?;
let master = &self.weights.master;
let mut out = MambaWeights::zeros(
&self.cfg,
self.mamba_input.len() / (self.batch * self.seq_len),
);
out.input_proj_w = master.input_proj_w.to_cpu(&self.ctx.stream)?;
out.input_proj_b = master.input_proj_b.to_cpu(&self.ctx.stream)?;
for (i, lw) in out.layers.iter_mut().enumerate() {
let g = &master.layers[i];
lw.norm_weight = g.norm_weight.to_cpu(&self.ctx.stream)?;
lw.in_proj_w = g.in_proj_w.to_cpu(&self.ctx.stream)?;
lw.conv1d_weight = g.conv1d_weight.to_cpu(&self.ctx.stream)?;
lw.conv1d_bias = g.conv1d_bias.to_cpu(&self.ctx.stream)?;
lw.x_proj_w = g.x_proj_w.to_cpu(&self.ctx.stream)?;
lw.dt_proj_w = g.dt_proj_w.to_cpu(&self.ctx.stream)?;
lw.dt_proj_b = g.dt_proj_b.to_cpu(&self.ctx.stream)?;
lw.a_log = g.a_log.to_cpu(&self.ctx.stream)?;
lw.d_param = g.d_param.to_cpu(&self.ctx.stream)?;
lw.out_proj_w = g.out_proj_w.to_cpu(&self.ctx.stream)?;
lw.a_neg = lw.a_log.iter().map(|v| -v.exp()).collect();
}
out.norm_f_weight = master.norm_f_weight.to_cpu(&self.ctx.stream)?;
Ok(out)
}
}
pub(crate) struct MambaTrainerF32 {
pub ctx: GpuCtx,
pub cfg: MambaConfig,
pub batch: usize,
pub seq_len: usize,
pub weights: GpuMambaTrainWeights,
pub grads: GpuMambaGrads,
pub adam: GpuAdamW,
bias: AdamWBiasFactors,
acts: GpuMambaBackboneActs,
scratch: GpuMambaScratch,
state: GpuRecurrentState,
a_neg_all: GpuBuffer,
temporal: GpuBuffer,
mamba_input: GpuBuffer,
d_temporal: GpuBuffer,
graph: Option<GpuMambaF32TrainingStepGraph>,
split_forward_pending: bool,
grads_dirty: bool,
clip_partials: GpuByteBuffer,
clip_partials_host: Vec<f64>,
}
impl MambaTrainerF32 {
fn new_full(
gpu_ordinal: usize,
cpu_weights: &MambaWeights,
cfg: MambaConfig,
session: TrainSessionCfg,
) -> Result<Self, String> {
let TrainSessionCfg {
input_dim,
batch,
seq_len,
lr,
weight_decay,
} = session;
let device = GpuDevice::new(gpu_ordinal)?;
let ctx = GpuCtx::new(&device)?;
let weights = GpuMambaTrainWeights::from_cpu(&ctx.stream, cpu_weights)?;
let d_inner = cfg.d_inner();
let d_state = cfg.d_state;
let d_conv = cfg.d_conv;
let n_layers = cfg.n_layers;
let dims = GpuMambaDims {
batch,
d_model: cfg.d_model,
d_inner,
d_state,
d_conv,
dt_rank: cfg.dt_rank(),
xdbl_dim: cfg.xdbl_dim(),
seq_len,
mamba_input_dim: input_dim,
n_layers,
scan_mode: cfg.scan_mode,
rms_norm_eps: cfg.rms_norm_eps,
};
let acts = GpuMambaBackboneActs::new(&ctx.stream, &dims)?;
let scratch = GpuMambaScratch::new(&ctx.stream, &dims)?;
let mut a_neg_flat = vec![0.0f32; n_layers * d_inner * d_state];
for (l, lw) in cpu_weights.layers.iter().enumerate() {
for i in 0..d_inner * d_state {
a_neg_flat[l * d_inner * d_state + i] = -lw.a_log[i].exp();
}
}
let mut a_neg_all = GpuBuffer::zeros(&ctx.stream, n_layers * d_inner * d_state)?;
a_neg_all.upload(&ctx.stream, &a_neg_flat)?;
let mut state = GpuRecurrentState {
conv_states: GpuBuffer::zeros(&ctx.stream, n_layers * batch * d_inner * d_conv)?,
ssm_states: GpuBuffer::zeros(&ctx.stream, n_layers * batch * d_inner * d_state)?,
a_neg_all: GpuBuffer::zeros(&ctx.stream, n_layers * d_inner * d_state)?,
};
state.a_neg_all.upload(&ctx.stream, &a_neg_flat)?;
let temporal = GpuBuffer::zeros(&ctx.stream, batch * seq_len * cfg.d_model)?;
let mamba_input = GpuBuffer::zeros(&ctx.stream, batch * seq_len * input_dim)?;
let d_temporal = GpuBuffer::zeros(&ctx.stream, batch * seq_len * cfg.d_model)?;
let grads = GpuMambaGrads::new(&ctx.stream, &cfg, input_dim)?;
let adam = GpuAdamW::new(&ctx.stream, grads.flat.len())?
.with_lr(lr)
.with_weight_decay(weight_decay);
let bias = AdamWBiasFactors::new(&ctx.stream)?;
ctx.stream
.synchronize()
.map_err(|e| format!("sync: {e:?}"))?;
let clip_partials = alloc_partials(&ctx.stream)?;
Ok(Self {
ctx,
cfg,
batch,
seq_len,
weights,
grads,
adam,
bias,
acts,
scratch,
state,
a_neg_all,
temporal,
mamba_input,
d_temporal,
graph: None,
split_forward_pending: false,
grads_dirty: false,
clip_partials,
clip_partials_host: vec![0.0; GRAD_CLIP_PARTIALS],
})
}
pub fn reset_state(&mut self) -> Result<(), String> {
self.state.conv_states.zero(&self.ctx.stream)?;
self.state.ssm_states.zero(&self.ctx.stream)?;
Ok(())
}
pub fn capture_graph(&mut self) -> Result<(), String> {
self.bias.write(&self.ctx.stream, 1.0, 1.0)?;
let g = GpuMambaF32TrainingStepGraph::capture(
&self.ctx,
&self.cfg,
MambaF32Capture {
weights: &mut self.weights,
adam: &self.adam,
bias: &self.bias,
grads: &mut self.grads,
acts: &mut self.acts,
scratch: &mut self.scratch,
a_neg_all: &self.a_neg_all,
temporal: &mut self.temporal,
mamba_input: &self.mamba_input,
d_temporal: &mut self.d_temporal,
state: &mut self.state,
},
self.batch,
self.seq_len,
)?;
self.graph = Some(g);
Ok(())
}
#[doc(hidden)]
pub fn debug_a_neg_all(&self) -> Result<Vec<f32>, String> {
self.ctx
.stream
.synchronize()
.map_err(|e| format!("debug_a_neg_all sync: {e:?}"))?;
self.a_neg_all.to_cpu(&self.ctx.stream)
}
pub fn step(&mut self, input: &[f32], d_temporal: &[f32]) -> Result<StepMetrics, String> {
assert_eq!(
input.len(),
self.mamba_input.len(),
"input shape mismatch: expected batch*seq_len*input_dim={}, got {}",
self.mamba_input.len(),
input.len(),
);
assert_eq!(
d_temporal.len(),
self.d_temporal.len(),
"d_temporal shape mismatch: expected batch*seq_len*d_model={}, got {}",
self.d_temporal.len(),
d_temporal.len(),
);
if self.grads_dirty {
return Err(
"step(): an accumulate_only backward window is open — close it with \
backward_step(accumulate_only=false); the fused step zeroes the grad \
arena and would silently discard the accumulated gradients"
.into(),
);
}
self.mamba_input.upload(&self.ctx.stream, input)?;
self.d_temporal.upload(&self.ctx.stream, d_temporal)?;
let (step, bc1, bc2) = self.adam.advance();
self.bias.write(&self.ctx.stream, bc1, bc2)?;
let replayed = if let Some(ref g) = self.graph {
g.replay(&MambaF32Replay {
weights: &self.weights,
adam: &self.adam,
bias: &self.bias,
grads: &self.grads,
temporal: &self.temporal,
a_neg_all: &self.a_neg_all,
mamba_input: &self.mamba_input,
d_temporal: &self.d_temporal,
state: &self.state,
})?;
true
} else {
self.step_eager()?;
false
};
Ok(StepMetrics::plain(step, replayed))
}
pub(crate) fn forward_split(
&mut self,
input: &[f32],
temporal_out: &mut [f32],
) -> Result<(), String> {
assert_eq!(
input.len(),
self.mamba_input.len(),
"input shape mismatch: expected batch*seq_len*input_dim={}, got {}",
self.mamba_input.len(),
input.len(),
);
assert_eq!(
temporal_out.len(),
self.temporal.len(),
"temporal_out shape mismatch: expected batch*seq_len*d_model={}, got {}",
self.temporal.len(),
temporal_out.len(),
);
self.mamba_input.upload(&self.ctx.stream, input)?;
self.eager_forward()?;
self.ctx
.stream
.synchronize()
.map_err(|e| format!("forward_split sync: {e:?}"))?;
self.temporal.download(&self.ctx.stream, temporal_out)?;
self.split_forward_pending = true;
Ok(())
}
pub(crate) fn backward_split(
&mut self,
d_temporal: &[f32],
opts: BackwardOpts,
) -> Result<BackwardMetrics, String> {
if !self.split_forward_pending {
return Err(
"backward_step() without a pending forward() — the saved activations \
are stale or missing; call forward() first"
.into(),
);
}
if opts.clip_max_norm.is_some() && opts.accumulate_only {
return Err(
"clip_max_norm + accumulate_only is unsupported: the global norm is only \
defined over the COMPLETE accumulated gradient — request the clip on the \
final (applying) backward_step"
.into(),
);
}
assert_eq!(
d_temporal.len(),
self.d_temporal.len(),
"d_temporal shape mismatch: expected batch*seq_len*d_model={}, got {}",
self.d_temporal.len(),
d_temporal.len(),
);
self.d_temporal.upload(&self.ctx.stream, d_temporal)?;
if !self.grads_dirty {
self.grads.zero(&self.ctx.stream)?;
}
self.eager_backward()?;
self.split_forward_pending = false;
if opts.accumulate_only {
self.grads_dirty = true;
Ok(BackwardMetrics {
step: self.adam.step,
optimizer_stepped: false,
grad_norm: None,
loss_scale: None,
overflow_skipped: None,
})
} else {
let grad_norm = match opts.clip_max_norm {
Some(c) => Some(self.apply_clip(c)?),
None => None,
};
let (step, bc1, bc2) = self.adam.advance();
self.bias.write(&self.ctx.stream, bc1, bc2)?;
self.eager_optimize()?;
self.grads_dirty = false;
Ok(BackwardMetrics {
step,
optimizer_stepped: true,
grad_norm,
loss_scale: None,
overflow_skipped: None,
})
}
}
fn eager_forward(&mut self) -> Result<(), String> {
gpu_forward_mamba_backbone(
&self.ctx,
&mut self.temporal,
&mut self.acts,
&self.weights,
&self.mamba_input,
&mut self.state,
&mut self.scratch,
)
}
fn eager_backward(&mut self) -> Result<(), String> {
gpu_backward_mamba_backbone(
&self.ctx,
&mut self.d_temporal,
&self.grads,
&self.acts,
&self.weights,
&self.a_neg_all,
&mut self.scratch,
)
}
fn eager_optimize(&mut self) -> Result<(), String> {
step_m1_capturable(
&self.ctx,
&self.ctx.kernels.adamw_step_f32_capturable,
&self.adam,
self.bias.ptr(),
&mut self.weights,
&self.grads,
)?;
recompute_a_neg_all(
&self.ctx,
&self.weights.layers,
&self.a_neg_all,
&self.state.a_neg_all,
self.cfg.d_inner(),
self.cfg.d_state,
)
}
fn step_eager(&mut self) -> Result<(), String> {
self.grads.zero(&self.ctx.stream)?;
self.eager_forward()?;
self.eager_backward()?;
self.eager_optimize()
}
fn apply_clip(&mut self, max_norm: f32) -> Result<f32, String> {
let norm = global_grad_norm(
&self.ctx,
&self.grads.flat,
&mut self.clip_partials,
&mut self.clip_partials_host,
)?;
if !norm.is_finite() {
return Err(format!(
"clip_max_norm: non-finite global grad norm ({norm})"
));
}
let coef = max_norm as f64 / (norm + 1e-6);
if coef < 1.0 {
scale_grads(&self.ctx, &mut self.grads.flat, coef as f32)?;
}
Ok(norm as f32)
}
pub fn snapshot_master(&self) -> Result<MambaWeights, String> {
self.ctx
.stream
.synchronize()
.map_err(|e| format!("pre-snapshot sync: {e:?}"))?;
let w = &self.weights;
let input_dim = self.mamba_input.len() / (self.batch * self.seq_len);
let mut out = MambaWeights::zeros(&self.cfg, input_dim);
out.input_proj_w = w.input_proj_w.to_cpu(&self.ctx.stream)?;
out.input_proj_b = w.input_proj_b.to_cpu(&self.ctx.stream)?;
for (i, lw) in out.layers.iter_mut().enumerate() {
let g = &w.layers[i];
lw.norm_weight = g.norm_weight.to_cpu(&self.ctx.stream)?;
lw.in_proj_w = g.in_proj_w.to_cpu(&self.ctx.stream)?;
lw.conv1d_weight = g.conv1d_weight.to_cpu(&self.ctx.stream)?;
lw.conv1d_bias = g.conv1d_bias.to_cpu(&self.ctx.stream)?;
lw.x_proj_w = g.x_proj_w.to_cpu(&self.ctx.stream)?;
lw.dt_proj_w = g.dt_proj_w.to_cpu(&self.ctx.stream)?;
lw.dt_proj_b = g.dt_proj_b.to_cpu(&self.ctx.stream)?;
lw.a_log = g.a_log.to_cpu(&self.ctx.stream)?;
lw.d_param = g.d_param.to_cpu(&self.ctx.stream)?;
lw.out_proj_w = g.out_proj_w.to_cpu(&self.ctx.stream)?;
lw.a_neg = lw.a_log.iter().map(|v| -v.exp()).collect();
}
out.norm_f_weight = w.norm_f_weight.to_cpu(&self.ctx.stream)?;
Ok(out)
}
}