1use crate::Engine;
6use crate::model::{EmbedHost, GpuTensor, HostExps};
7use cudarc::driver::CudaSlice;
8use memra_gguf::config::{ModelConfig, SwigluClamp};
9use memra_gguf::model_plan::{AttentionPlan, MlpPlan};
10use memra_gguf::source::{GgufSource, TensorSource};
11use memra_gguf::{GgmlType, GgufFile};
12use std::collections::HashMap;
13use std::sync::Arc;
14
15fn load_t(
18 e: &Engine,
19 src: &dyn TensorSource,
20 name: &str,
21) -> Result<GpuTensor, Box<dyn std::error::Error>> {
22 GpuTensor::load_from_source(e, src, name)
23}
24fn load_opt(
25 e: &Engine,
26 src: &dyn TensorSource,
27 name: &str,
28) -> Result<Option<GpuTensor>, Box<dyn std::error::Error>> {
29 GpuTensor::load_opt_from_source(e, src, name)
30}
31
32struct ResidencyBytes {
33 experts: HashMap<usize, usize>,
34 rest: usize,
35 saw_experts: bool,
36}
37
38fn block_index(name: &str) -> Option<usize> {
39 name.strip_prefix("blk.")?.split('.').next()?.parse().ok()
40}
41
42fn residency_bytes_by_device<'a>(
43 tensors: impl IntoIterator<Item = (&'a str, usize)>,
44 layer_devices: &[usize],
45 primary_device: usize,
46) -> ResidencyBytes {
47 let mut out = ResidencyBytes {
48 experts: HashMap::new(),
49 rest: 0,
50 saw_experts: false,
51 };
52 for (name, bytes) in tensors {
53 if name.starts_with("blk.") && name.contains("_exps.") {
54 let device = block_index(name)
55 .and_then(|il| layer_devices.get(il).copied())
56 .unwrap_or(primary_device);
57 *out.experts.entry(device).or_default() += bytes;
58 out.saw_experts = true;
59 } else {
60 out.rest += bytes;
61 }
62 }
63 out
64}
65
66pub(crate) struct ResidentPlan {
69 primary_device: usize,
70 layer_devices: Vec<usize>,
71 layer_counts: HashMap<usize, usize>,
72 exact_expert_bytes: Option<HashMap<usize, usize>>,
73 trunk_bytes: usize,
74 decisions: HashMap<usize, bool>,
75 pp: bool,
76}
77
78#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
88enum StepExpertArtifact {
89 #[default]
90 E4m3,
91 Nvfp4,
92}
93
94#[derive(Clone, Debug, Default)]
95struct StepParallelLoadConfig {
96 ep_specs: Vec<crate::tp::StepEpLayerSpec>,
97 tp_specs: Vec<crate::tp::StepTpLayerSpec>,
98 native_p2p: bool,
99 ep_device_arithmetic: bool,
100 f32_mirror: bool,
101 bulk_p2p: bool,
102 nvfp4_device_routes: bool,
103 auto_parallel: bool,
104 tp_attention_expert_overlap: bool,
105 expert_artifact: StepExpertArtifact,
106}
107
108#[derive(Default)]
109pub(crate) struct StepParallelRuntimeRegistry {
110 config: StepParallelLoadConfig,
111 runtimes: HashMap<(Vec<usize>, bool, bool, bool), Arc<crate::tp::TpE4m3HostBounce>>,
112}
113
114#[derive(Clone, Copy, Debug, PartialEq, Eq)]
115enum StepExpertLayout {
116 TensorParallel,
117 ExpertParallel,
118}
119
120#[derive(Clone, Debug, PartialEq, Eq)]
121struct StepExpertSelection {
122 spec: crate::tp::StepEpLayerSpec,
123 layout: StepExpertLayout,
124 configured_by_tp: bool,
125}
126
127fn select_step_expert_layout_inner(
128 layer: usize,
129 ep_specs: &[crate::tp::StepEpLayerSpec],
130 tp_specs: &[crate::tp::StepTpLayerSpec],
131 allow_attention_ep_overlap: bool,
132) -> Result<Option<StepExpertSelection>, String> {
133 let ep = ep_specs.iter().find(|spec| spec.layer == layer);
134 let tp = tp_specs.iter().find(|spec| spec.layer == layer);
135 Ok(match (ep, tp) {
136 (Some(spec), None) => Some(StepExpertSelection {
137 spec: spec.clone(),
138 layout: StepExpertLayout::ExpertParallel,
139 configured_by_tp: false,
140 }),
141 (None, Some(spec)) => Some(StepExpertSelection {
142 spec: spec.clone(),
143 layout: if spec.devices.len() > 2 {
144 StepExpertLayout::ExpertParallel
145 } else {
146 StepExpertLayout::TensorParallel
147 },
148 configured_by_tp: true,
149 }),
150 (None, None) => None,
151 (Some(ep), Some(tp)) => {
152 if !allow_attention_ep_overlap {
153 return Err(format!(
154 "Step layer {layer} cannot enable MEMRA_STEP_EP and MEMRA_STEP_TP together"
155 ));
156 }
157 if ep.devices.first() != tp.devices.first()
158 || tp.devices.iter().any(|device| !ep.devices.contains(device))
159 {
160 return Err(format!(
161 "automatic TP-attention/EP overlap at layer {layer} requires the attention \
162 ranks {:?} to be an owner-first subset of expert ranks {:?}",
163 tp.devices, ep.devices
164 ));
165 }
166 Some(StepExpertSelection {
167 spec: ep.clone(),
168 layout: StepExpertLayout::ExpertParallel,
169 configured_by_tp: false,
170 })
171 }
172 })
173}
174
175#[cfg(test)]
176fn select_step_expert_layout(
177 layer: usize,
178 ep_specs: &[crate::tp::StepEpLayerSpec],
179 tp_specs: &[crate::tp::StepTpLayerSpec],
180) -> Result<Option<StepExpertSelection>, String> {
181 select_step_expert_layout_inner(layer, ep_specs, tp_specs, false)
182}
183
184impl StepParallelRuntimeRegistry {
185 fn with_config(config: StepParallelLoadConfig) -> Self {
186 Self {
187 config,
188 runtimes: HashMap::new(),
189 }
190 }
191
192 fn tp_spec(&self, layer: usize) -> Option<&crate::tp::StepTpLayerSpec> {
193 self.config.tp_specs.iter().find(|spec| spec.layer == layer)
194 }
195
196 fn expert_selection(&self, layer: usize) -> Result<Option<StepExpertSelection>, String> {
197 select_step_expert_layout_inner(
198 layer,
199 &self.config.ep_specs,
200 &self.config.tp_specs,
201 self.config.tp_attention_expert_overlap,
202 )
203 }
204
205 fn runtime(
206 &mut self,
207 devices: &[usize],
208 native_p2p: bool,
209 ep_device_arithmetic: bool,
210 ) -> Result<Arc<crate::tp::TpE4m3HostBounce>, Box<dyn std::error::Error>> {
211 let bulk_p2p = self.config.bulk_p2p && native_p2p;
212 let key = (devices.to_vec(), native_p2p, ep_device_arithmetic, bulk_p2p);
213 if let Some(runtime) = self.runtimes.get(&key) {
214 return Ok(Arc::clone(runtime));
215 }
216 let runtime = Arc::new(crate::tp::TpE4m3HostBounce::new_configured(
217 devices,
218 native_p2p,
219 ep_device_arithmetic,
220 bulk_p2p,
221 )?);
222 let names = runtime.device_names()?;
223 if names
224 .iter()
225 .any(|name| !name.contains("RTX PRO 6000") || !name.contains("Blackwell"))
226 {
227 return Err(format!(
228 "Step distributed execution is qualified only on RTX PRO 6000 Blackwell, \
229 got {names:?}"
230 )
231 .into());
232 }
233 self.runtimes.insert(key, Arc::clone(&runtime));
234 Ok(runtime)
235 }
236}
237
238impl ResidentPlan {
239 fn from_layout(
240 src: &dyn TensorSource,
241 primary_device: usize,
242 layer_devices: Vec<usize>,
243 pp: bool,
244 ) -> Self {
245 let mut layer_counts = HashMap::new();
246 for &device in &layer_devices {
247 *layer_counts.entry(device).or_default() += 1;
248 }
249 let (exact_expert_bytes, trunk_bytes) = match src.gguf() {
250 Some(g) => {
251 let bytes = residency_bytes_by_device(
252 g.tensors
253 .iter()
254 .map(|t| (t.name.as_str(), t.n_bytes as usize)),
255 &layer_devices,
256 primary_device,
257 );
258 if bytes.saw_experts {
259 (Some(bytes.experts), bytes.rest)
260 } else {
261 (None, 0)
262 }
263 }
264 None => (None, 0),
265 };
266 Self {
267 primary_device,
268 layer_devices,
269 layer_counts,
270 exact_expert_bytes,
271 trunk_bytes,
272 decisions: HashMap::new(),
273 pp,
274 }
275 }
276
277 pub(crate) fn unsharded(e: &Engine, src: &dyn TensorSource, cfg: &ModelConfig) -> Self {
278 let device = e.ctx().ordinal();
279 Self::from_layout(src, device, vec![device; cfg.n_layer as usize], false)
280 }
281
282 pub(crate) fn pp(
283 e: &Engine,
284 src: &dyn TensorSource,
285 cfg: &ModelConfig,
286 n_trunk: usize,
287 ) -> Result<Self, Box<dyn std::error::Error>> {
288 let primary = e.ctx().ordinal();
289 let Some(_fence) = crate::pp::pp_cuts(n_trunk) else {
290 return Ok(Self::unsharded(e, src, cfg));
291 };
292 let mut layer_devices = vec![primary; cfg.n_layer as usize];
293 for (il, device) in layer_devices.iter_mut().take(n_trunk).enumerate() {
294 *device = crate::pp::layer_engine(e, n_trunk, il)?.ctx().ordinal();
295 }
296 Ok(Self::from_layout(src, primary, layer_devices, true))
297 }
298
299 fn exclude_distributed_expert_layers(&mut self, specs: impl IntoIterator<Item = usize>) {
303 for layer in specs {
304 let device = self
305 .layer_devices
306 .get(layer)
307 .copied()
308 .unwrap_or(self.primary_device);
309 if let Some(count) = self.layer_counts.get_mut(&device) {
310 *count = count.saturating_sub(1);
311 }
312 }
313 }
314
315 fn should_reside(&mut self, e: &Engine, il: usize, per_layer: usize) -> bool {
316 let device = self
317 .layer_devices
318 .get(il)
319 .copied()
320 .unwrap_or(self.primary_device);
321 debug_assert_eq!(e.ctx().ordinal(), device);
322 if let Some(&decision) = self.decisions.get(&device) {
323 return decision;
324 }
325 if std::env::var("MEMRA_MOE_RESIDENT").as_deref() == Ok("0") {
326 self.decisions.insert(device, false);
327 return false;
328 }
329 let (free, _total) = match e.ctx().mem_get_info() {
330 Ok(v) => v,
331 Err(_) => {
332 self.decisions.insert(device, false);
333 return false;
334 }
335 };
336 let projected = self
337 .exact_expert_bytes
338 .as_ref()
339 .map(|bytes| bytes.get(&device).copied().unwrap_or(0))
340 .unwrap_or(per_layer * self.layer_counts.get(&device).copied().unwrap_or(1));
341 let budget = std::env::var("MEMRA_MOE_RESIDENT_GB")
342 .ok()
343 .and_then(|v| v.parse::<f64>().ok())
344 .map(|gb| (gb * 1e9) as usize)
345 .unwrap_or_else(|| {
346 let reserve = std::env::var("MEMRA_MOE_RESIDENT_HEADROOM_GB")
347 .ok()
348 .and_then(|v| v.parse::<f64>().ok())
349 .map(|gb| (gb * 1e9) as usize)
350 .unwrap_or(2_000_000_000);
351 free.saturating_sub(self.trunk_bytes + reserve)
352 });
353 let ok = projected <= budget;
354 eprintln!(
355 "[moe] resident-experts decision ({}dev{}): experts {:.2}GB + trunk {:.2}GB vs free {:.2}GB (expert budget {:.2}GB) -> {}",
356 if self.pp { "PP " } else { "" },
357 device,
358 projected as f64 / 1e9,
359 self.trunk_bytes as f64 / 1e9,
360 free as f64 / 1e9,
361 budget as f64 / 1e9,
362 if ok { "RESIDENT" } else { "SLRU cache" }
363 );
364 self.decisions.insert(device, ok);
365 ok
366 }
367}
368
369fn load_mixer_kind(
371 e: &Engine,
372 src: &dyn TensorSource,
373 cfg: &ModelConfig,
374 il: u32,
375 attention: &AttentionPlan,
376 step_runtimes: &mut StepParallelRuntimeRegistry,
377) -> Result<Mixer, Box<dyn std::error::Error>> {
378 let p = |s: &str| format!("blk.{il}.{s}");
379 Ok(match attention {
380 AttentionPlan::Mla(mla) => Mixer::Mla(MlaAttnLayer::load(e, src, il, mla)?),
381 AttentionPlan::Full(full)
382 | AttentionPlan::SlidingWindow {
383 attention: full, ..
384 } => {
385 Mixer::Full(FullAttnLayer {
386 wq: load_t(e, src, &p("attn_q.weight"))?,
387 wk: load_t(e, src, &p("attn_k.weight"))?,
388 wv: match load_opt(e, src, &p("attn_v.weight"))? {
393 Some(v) => v,
394 None => load_t(e, src, &p("attn_k.weight"))?,
395 },
396 wo: load_t(e, src, &p("attn_output.weight"))?,
397 q_norm: load_t(e, src, &p("attn_q_norm.weight"))?,
398 k_norm: load_t(e, src, &p("attn_k_norm.weight"))?,
399 attn_gate: if full.output_gate
403 == memra_gguf::config::AttentionGateKind::SeparateHead
404 {
405 Some(load_t(e, src, &p("attn_gate.weight"))?)
406 } else {
407 None
408 },
409 step_tp_qkv: build_step_tp_qkv(e, src, cfg, il as usize, step_runtimes)?,
410 })
411 }
412 AttentionPlan::KimiDeltaNet(kda) => {
415 Mixer::Kda(crate::kda::KdaAttnLayer::load(e, src, il, kda)?)
416 }
417 AttentionPlan::GatedDeltaNet(geometry) => Mixer::Linear(LinearAttnLayer {
418 geometry: *geometry,
419 wqkv: load_t(e, src, &p("attn_qkv.weight"))?,
420 wqkv_gate: load_t(e, src, &p("attn_gate.weight"))?,
421 ssm_beta: load_t(e, src, &p("ssm_beta.weight"))?,
422 ssm_alpha: load_t(e, src, &p("ssm_alpha.weight"))?,
423 ssm_a: load_t(e, src, &p("ssm_a"))?,
424 ssm_dt: load_t(e, src, &p("ssm_dt.bias"))?,
425 ssm_conv1d: load_t(e, src, &p("ssm_conv1d.weight"))?,
426 ssm_norm: load_t(e, src, &p("ssm_norm.weight"))?,
427 ssm_out: load_t(e, src, &p("ssm_out.weight"))?,
428 }),
429 })
430}
431
432#[allow(clippy::too_many_arguments)] pub(crate) fn load_ffn(
440 e: &Engine,
441 src: &dyn TensorSource,
442 cfg: &ModelConfig,
443 mlp: &MlpPlan,
444 il: u32,
445 spill: Option<(&GgufFile, &mut crate::spill::SpillCtx)>,
446 resident: &mut ResidentPlan,
447 step_runtimes: &mut StepParallelRuntimeRegistry,
448) -> Result<Ffn, Box<dyn std::error::Error>> {
449 let p = |s: &str| format!("blk.{il}.{s}");
450 let artifact_dense = matches!(mlp, MlpPlan::Moe(_))
457 && !src.has(&p("ffn_gate_exps.weight"))
458 && !src.has(&p("ffn_gate_up_exps.weight"))
459 && src.has(&p("ffn_gate.weight"));
460 Ok(if artifact_dense {
461 Ffn::Dense {
462 ffn_gate: GpuTensor::load_from_source(e, src, &p("ffn_gate.weight"))?,
463 ffn_up: GpuTensor::load_from_source(e, src, &p("ffn_up.weight"))?,
464 ffn_down: GpuTensor::load_from_source(e, src, &p("ffn_down.weight"))?,
465 }
466 } else if let MlpPlan::Moe(moe) = mlp {
467 let n_expert = moe.expert_count as usize;
468 let (gate_exps, up_exps, down_exps) = match spill {
474 Some((g, ctx)) => (
475 HostExps::load_tiered(e, g, &p("ffn_gate_exps.weight"), ctx)?,
476 HostExps::load_tiered(e, g, &p("ffn_up_exps.weight"), ctx)?,
477 HostExps::load_tiered(e, g, &p("ffn_down_exps.weight"), ctx)?,
478 ),
479 None => {
480 let exps = |e: &Engine, n: &str| -> Result<HostExps, Box<dyn std::error::Error>> {
481 if src.has(n) {
482 HostExps::load_stacked_from_source(e, src, n)
483 } else {
484 HostExps::load_from_source(e, src, n, n_expert)
485 }
486 };
487 let fused = p("ffn_gate_up_exps.weight");
489 if !src.has(&p("ffn_gate_exps.weight")) && src.has(&fused) {
490 let ff = moe.expert_intermediate_size as usize;
491 (
492 HostExps::load_stacked_split_from_source(e, src, &fused, 0, ff)?,
493 HostExps::load_stacked_split_from_source(e, src, &fused, ff, 2 * ff)?,
494 exps(e, &p("ffn_down_exps.weight"))?,
495 )
496 } else {
497 (
498 exps(e, &p("ffn_gate_exps.weight"))?,
499 exps(e, &p("ffn_up_exps.weight"))?,
500 exps(e, &p("ffn_down_exps.weight"))?,
501 )
502 }
503 }
504 };
505 let (step_ep, step_tp) = build_step_distributed_exps(
506 e,
507 cfg,
508 src,
509 il as usize,
510 &gate_exps,
511 &up_exps,
512 &down_exps,
513 step_runtimes,
514 )?;
515 let dev_exps = if step_ep.is_some() || step_tp.is_some() {
521 None
522 } else {
523 build_dev_exps(e, resident, il as usize, &gate_exps, &up_exps, &down_exps)?
524 };
525 let mut macro_row = vec![1.0f32; 3 * n_expert];
527 for (slot, exps) in [(0usize, &gate_exps), (1, &up_exps), (2, &down_exps)] {
528 if let Some(ms) = exps.macros.as_ref() {
529 macro_row[slot * n_expert..(slot + 1) * n_expert].copy_from_slice(ms);
530 }
531 }
532 let has_macros = macro_row.iter().any(|&m| m != 1.0);
533 let dev_macros = e.htod(¯o_row)?;
534 let exp_probs_b = src
537 .find(&p("exp_probs_b.bias"))
538 .map(|v| memra_gguf::dequant::dequantize(v.ggml_type, &v.bytes, n_expert));
539 if exp_probs_b.is_none()
548 && matches!(
549 moe.router,
550 memra_gguf::model_plan::RouterPlan::Sigmoid {
551 selection_bias: true,
552 ..
553 } | memra_gguf::model_plan::RouterPlan::SqrtSoftplus {
554 selection_bias: true,
555 ..
556 }
557 )
558 {
559 return Err(format!(
560 "layer {il}: {} is absent, but the compiled ModelPlan declares a router with a \
561 selection bias ({:?}). Refusing to load: a zero-filled bias would route to \
562 different experts than this model does, silently. Either the checkpoint does \
563 not carry the tensor, or this arch has no `exp_probs_b.bias` entry in \
564 hf_mapping's ggml->HF map",
565 p("exp_probs_b.bias"),
566 moe.router
567 )
568 .into());
569 }
570 let active_experts = src.active_experts(il).map(<[bool]>::to_vec);
571 let route_bias = exp_probs_b.clone().unwrap_or_else(|| vec![0.0; n_expert]);
572 let active_row: Vec<u8> = active_experts
573 .as_ref()
574 .map(|mask| mask.iter().map(|&is_active| u8::from(is_active)).collect())
575 .unwrap_or_else(|| vec![1; n_expert]);
576 let exp_probs_b_dev = e.htod(&route_bias)?;
577 let active_experts_dev = e.htod_bytes(&active_row)?;
578 let gate_shexp = load_opt(e, src, &p("ffn_gate_shexp.weight"))?;
579 let up_shexp = load_opt(e, src, &p("ffn_up_shexp.weight"))?;
580 let down_shexp = load_opt(e, src, &p("ffn_down_shexp.weight"))?;
581 if moe.shared.is_some()
589 && (gate_shexp.is_none() || up_shexp.is_none() || down_shexp.is_none())
590 {
591 return Err(format!(
592 "layer {il}: the compiled ModelPlan declares an always-on shared expert, but \
593 {}{}{} could not be resolved in the checkpoint. Refusing to load: dropping the \
594 shared branch computes a different model, silently. Either the checkpoint does \
595 not carry it, or this arch's shared-expert spelling is missing from \
596 hf_mapping's ggml->HF map",
597 if gate_shexp.is_none() {
598 format!("{} ", p("ffn_gate_shexp.weight"))
599 } else {
600 String::new()
601 },
602 if up_shexp.is_none() {
603 format!("{} ", p("ffn_up_shexp.weight"))
604 } else {
605 String::new()
606 },
607 if down_shexp.is_none() {
608 p("ffn_down_shexp.weight")
609 } else {
610 String::new()
611 },
612 )
613 .into());
614 }
615 Ffn::Moe(MoeWeights {
616 gate_inp: load_t(e, src, &p("ffn_gate_inp.weight"))?,
617 gate_inp_shexp: load_opt(e, src, &p("ffn_gate_inp_shexp.weight"))?,
618 exp_probs_b,
619 exp_probs_b_dev,
620 active_experts,
621 active_experts_dev,
622 gate_exps,
623 up_exps,
624 down_exps,
625 gate_shexp,
626 up_shexp,
627 down_shexp,
628 dev_exps,
629 step_ep,
630 step_tp,
631 glm5_ep: None,
632 dev_macros,
633 has_macros,
634 w4a16_bf16_activations: matches!(
635 src.expert_activation_precision(),
636 memra_gguf::source::ExpertActivationPrecision::Bf16
637 ),
638 })
639 } else {
640 Ffn::Dense {
641 ffn_gate: load_t(e, src, &p("ffn_gate.weight"))?,
642 ffn_up: load_t(e, src, &p("ffn_up.weight"))?,
643 ffn_down: load_t(e, src, &p("ffn_down.weight"))?,
644 }
645 })
646}
647
648fn host_e4m3_bank(
649 exps: &HostExps,
650) -> Result<crate::tp::E4m3ExpertBank<'_>, Box<dyn std::error::Error>> {
651 if exps.qtype != crate::QT_F8_E4M3_BLK {
652 return Err(format!(
653 "Step EP requires native block-E4M3 expert banks, got qtype {}",
654 exps.qtype
655 )
656 .into());
657 }
658 let scales = exps
659 .fp8_blk
660 .as_ref()
661 .ok_or("Step EP native expert bank has no block-E4M3 scale plane")?;
662 Ok(crate::tp::E4m3ExpertBank {
663 codes: exps.bytes.as_bytes(),
664 scales: &scales.scales,
665 expert_count: exps.n_expert,
666 out_features: exps.out_f,
667 in_features: exps.in_f,
668 })
669}
670
671fn validate_step_expert_specs(
672 contract: &crate::parallel::ModelParallelContract,
673 flag: &str,
674 specs: &[crate::tp::StepEpLayerSpec],
675 allow_dense_attention_only: bool,
676) -> Result<(), Box<dyn std::error::Error>> {
677 for candidate in specs {
678 if candidate.layer >= contract.trunk_layers {
679 return Err(format!(
680 "{flag} layer {} is outside Step trunk layers 0..{}",
681 candidate.layer, contract.trunk_layers
682 )
683 .into());
684 }
685 if candidate.layer < contract.dense_prefix_layers {
686 if allow_dense_attention_only {
687 continue;
688 }
689 return Err(format!(
690 "{flag} layer {} is outside Step routed-expert layers {}..{}",
691 candidate.layer, contract.dense_prefix_layers, contract.trunk_layers
692 )
693 .into());
694 }
695 }
696 Ok(())
697}
698
699fn validate_step_expert_activation_layout(
700 cfg: &ModelConfig,
701 flag: &str,
702 selection: &StepExpertSelection,
703) -> Result<(), Box<dyn std::error::Error>> {
704 let _ = (cfg, flag, selection);
710 Ok(())
711}
712
713fn parse_auto_w4a16_bf16_mmv(value: Option<&str>) -> Result<bool, String> {
714 match value {
715 None => Ok(true),
716 Some("0") => Ok(false),
717 Some("1") => Ok(true),
718 Some(value) => Err(format!(
719 "MEMRA_BF16_MMV={value:?} is invalid under MEMRA_PARALLEL=auto; expected 0 or 1"
720 )),
721 }
722}
723
724fn parse_auto_parallel_tp_attention(value: Option<&str>) -> Result<bool, String> {
725 match value {
726 None | Some("") | Some("0") => Ok(false),
727 Some("1") => Ok(true),
728 Some(value) => Err(format!(
729 "MEMRA_PARALLEL_TP_ATTENTION={value:?} is invalid; expected 0 or 1"
730 )),
731 }
732}
733
734fn auto_parallel_tp_attention_enabled() -> Result<bool, String> {
735 parse_auto_parallel_tp_attention(std::env::var("MEMRA_PARALLEL_TP_ATTENTION").ok().as_deref())
736}
737
738fn parse_auto_parallel_tp_attention_ranks(value: Option<&str>) -> Result<Option<usize>, String> {
739 match value {
740 None => Ok(None),
741 Some("2") => Ok(Some(2)),
742 Some("3") => Ok(Some(3)),
743 Some("4") => Ok(Some(4)),
744 Some(value) => Err(format!(
745 "MEMRA_PARALLEL_TP_ATTENTION_RANKS={value:?} is invalid; expected 2, 3, or 4"
746 )),
747 }
748}
749
750fn auto_parallel_tp_attention_ranks() -> Result<Option<usize>, String> {
751 parse_auto_parallel_tp_attention_ranks(
752 std::env::var("MEMRA_PARALLEL_TP_ATTENTION_RANKS")
753 .ok()
754 .as_deref(),
755 )
756}
757
758fn prepare_auto_parallel(
764 src: &dyn TensorSource,
765 cfg: &ModelConfig,
766 plan: &memra_gguf::model_plan::ModelPlan,
767) -> Result<Option<crate::parallel::AutoParallelPlacement>, Box<dyn std::error::Error>> {
768 let Some(devices) = crate::tp::auto_parallel_devices()? else {
769 return Ok(None);
770 };
771 if std::env::var_os("MEMRA_PP_STAGES").is_some()
772 || std::env::var_os("MEMRA_PP_DEVICES").is_some()
773 || std::env::var_os("MEMRA_PP_SPLITS").is_some()
774 {
775 return Err(
776 "MEMRA_PARALLEL=auto cannot be combined with MEMRA_PP_STAGES, MEMRA_PP_DEVICES, or \
777 MEMRA_PP_SPLITS"
778 .into(),
779 );
780 }
781 let placement = crate::parallel::plan_auto_parallel(src, cfg, plan, &devices)?;
782 let auto_w4a16_bf16 = placement.backend == crate::parallel::AutoParallelBackend::ExpertParallel
783 && matches!(
784 src.expert_activation_precision(),
785 memra_gguf::source::ExpertActivationPrecision::Bf16
786 );
787 let bf16_nonexpert = if auto_w4a16_bf16 {
788 let explicit = match std::env::var("MEMRA_BF16_MMV") {
789 Ok(value) => Some(value),
790 Err(std::env::VarError::NotPresent) => None,
791 Err(error) => return Err(format!("cannot read MEMRA_BF16_MMV: {error}").into()),
792 };
793 let enabled = parse_auto_w4a16_bf16_mmv(explicit.as_deref())?;
794 if enabled && explicit.is_none() {
795 unsafe {
798 std::env::set_var("MEMRA_BF16_MMV", "1");
799 }
800 }
801 match (enabled, explicit.is_some()) {
802 (true, false) => "bf16-resident(auto)",
803 (true, true) => "bf16-resident(explicit)",
804 (false, true) => "f32-expanded(explicit-rollback)",
805 (false, false) => unreachable!("unset auto W4A16 defaults BF16 residency on"),
806 }
807 } else {
808 "placement-default"
809 };
810 if placement.backend == crate::parallel::AutoParallelBackend::Pipeline {
811 let stages = placement.devices.len();
812 let device_list = placement
813 .devices
814 .iter()
815 .map(usize::to_string)
816 .collect::<Vec<_>>()
817 .join(",");
818 let splits = placement
819 .pipeline_splits
820 .iter()
821 .map(usize::to_string)
822 .collect::<Vec<_>>()
823 .join(",");
824 unsafe {
827 std::env::set_var("MEMRA_PP_STAGES", stages.to_string());
828 std::env::set_var("MEMRA_PP_DEVICES", &device_list);
829 std::env::set_var("MEMRA_PP_SPLITS", &splits);
830 }
831 }
832 let family = if placement.routed_layers.is_empty() {
833 "dense-transformer"
834 } else {
835 "routed-moe"
836 };
837 eprintln!(
838 "[parallel-auto] family={family} variant={:?} devices={:?} placement={} \
839 checkpoint_peak={:.2}GB ep_root={:.2}GB ep_peer={:.2}GB reserve={:.2}GB \
840 capacity={:?} splits={:?} bf16_nonexpert={bf16_nonexpert} \
841 wavefront=off(default) performance_claim=false",
842 cfg.name,
843 placement.devices,
844 match placement.backend {
845 crate::parallel::AutoParallelBackend::Pipeline => "pipeline",
846 crate::parallel::AutoParallelBackend::ExpertParallel => "expert-parallel",
847 },
848 placement.checkpoint_peak_bytes as f64 / 1e9,
849 placement.expert_root_bytes as f64 / 1e9,
850 placement.expert_peer_bytes as f64 / 1e9,
851 placement.reserve_bytes as f64 / 1e9,
852 placement.device_capacity_bytes,
853 placement.pipeline_splits,
854 );
855 Ok(Some(placement))
856}
857
858fn prepare_step_parallel_load(
859 e: &Engine,
860 src: &dyn TensorSource,
861 cfg: &ModelConfig,
862 trunk_layers: usize,
863 auto_placement: Option<&crate::parallel::AutoParallelPlacement>,
864) -> Result<StepParallelLoadConfig, Box<dyn std::error::Error>> {
865 let mut tp_specs = crate::tp::step_tp_layer_specs()?;
866 let mut ep_specs = crate::tp::step_ep_layer_specs()?;
867 let device_arithmetic = crate::tp::step_ep_device_arithmetic_enabled()?;
868 let f32_mirror = crate::tp::step_tp_f32_mirror_enabled()?;
869 let bulk_p2p = crate::tp::step_tp_bulk_p2p_enabled()?;
870 let mut native_p2p = crate::tp::step_tp_native_p2p_enabled()?;
871 let mut nvfp4_device_routes = crate::tp::step_nvfp4_dev_routes_enabled()?;
872 let auto_tp_attention = auto_parallel_tp_attention_enabled()?;
873 let requested_attention_ranks = auto_parallel_tp_attention_ranks()?;
874 let mut auto_parallel = false;
875 let mut tp_attention_expert_overlap = false;
876 if requested_attention_ranks.is_some() && !auto_tp_attention {
877 return Err(
878 "MEMRA_PARALLEL_TP_ATTENTION_RANKS requires MEMRA_PARALLEL_TP_ATTENTION=1".into(),
879 );
880 }
881 if auto_tp_attention && auto_placement.is_none() {
882 return Err(
883 "MEMRA_PARALLEL_TP_ATTENTION=1 requires MEMRA_PARALLEL=auto; explicit per-layer \
884 recipes remain under MEMRA_STEP_TP"
885 .into(),
886 );
887 }
888 if let Some(placement) = auto_placement {
889 if !tp_specs.is_empty() || !ep_specs.is_empty() {
890 return Err(
891 "MEMRA_PARALLEL=auto cannot be combined with MEMRA_STEP_EP or MEMRA_STEP_TP".into(),
892 );
893 }
894 if placement.backend == crate::parallel::AutoParallelBackend::Pipeline {
895 if auto_tp_attention {
896 return Err(
897 "MEMRA_PARALLEL_TP_ATTENTION=1 requires automatic whole-expert EP; the \
898 selected checkpoint fits only the pipeline backend"
899 .into(),
900 );
901 }
902 return Ok(StepParallelLoadConfig::default());
903 }
904 if auto_tp_attention {
905 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
906 if !contract.tensor_attention_supported {
907 return Err(format!(
908 "MEMRA_PARALLEL_TP_ATTENTION=1 cannot shard attention for {:?}: the \
909 compiled ModelPlan has no generic tensor-attention contract",
910 cfg.name
911 )
912 .into());
913 }
914 let attention_ranks = requested_attention_ranks.unwrap_or(placement.devices.len());
915 if attention_ranks > placement.devices.len() {
916 return Err(format!(
917 "MEMRA_PARALLEL_TP_ATTENTION_RANKS={attention_ranks} exceeds the automatic \
918 placement width {}",
919 placement.devices.len()
920 )
921 .into());
922 }
923 let attention_devices = placement.devices[..attention_ranks].to_vec();
924 tp_specs = (0..trunk_layers)
925 .map(|layer| crate::tp::StepTpLayerSpec {
926 layer,
927 devices: attention_devices.clone(),
928 })
929 .collect();
930 if attention_ranks < placement.devices.len() {
931 ep_specs = placement
932 .routed_layers
933 .iter()
934 .map(|&layer| crate::tp::StepEpLayerSpec {
935 layer,
936 devices: placement.devices.clone(),
937 })
938 .collect();
939 tp_attention_expert_overlap = true;
940 } else {
941 ep_specs.clear();
942 }
943 } else {
944 ep_specs = placement
945 .routed_layers
946 .iter()
947 .map(|&layer| crate::tp::StepEpLayerSpec {
948 layer,
949 devices: placement.devices.clone(),
950 })
951 .collect();
952 }
953 auto_parallel = true;
954 native_p2p = true;
955 nvfp4_device_routes = matches!(
956 src.expert_activation_precision(),
957 memra_gguf::source::ExpertActivationPrecision::Bf16
958 );
959 eprintln!(
960 "[parallel-auto-backend] devices={:?} routed_layers={} native_p2p=true \
961 artifact_activation={:?} attention_layout={} attention_devices={:?} \
962 expert_layout=expert-parallel expert_devices={:?} \
963 backend={} performance_claim=false",
964 placement.devices,
965 placement.routed_layers.len(),
966 src.expert_activation_precision(),
967 if auto_tp_attention {
968 "tensor-parallel"
969 } else {
970 "root-local"
971 },
972 tp_specs
973 .first()
974 .map(|spec| spec.devices.as_slice())
975 .unwrap_or(&[]),
976 ep_specs
977 .first()
978 .map(|spec| spec.devices.as_slice())
979 .unwrap_or(placement.devices.as_slice()),
980 if nvfp4_device_routes {
981 "nvfp4-w4a16"
982 } else {
983 "artifact-selected-host-oracle"
984 },
985 );
986 }
987 if tp_specs.is_empty() {
988 if auto_tp_attention {
989 return Err("MEMRA_PARALLEL_TP_ATTENTION=1 produced no tensor-parallel layers".into());
990 }
991 if device_arithmetic || f32_mirror || bulk_p2p {
992 return Err(
993 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1, MEMRA_STEP_TP_F32_MIRROR=1, or \
994 MEMRA_STEP_TP_BULK_P2P=1 requires MEMRA_STEP_TP; device arithmetic and bulk \
995 transport also require MEMRA_STEP_TP_NATIVE_P2P=1"
996 .into(),
997 );
998 }
999 if nvfp4_device_routes && ep_specs.is_empty() {
1000 return Err(
1001 "MEMRA_STEP_NVFP4_DEV_ROUTES=1 requires MEMRA_STEP_EP or MEMRA_STEP_TP".into(),
1002 );
1003 }
1004 if nvfp4_device_routes && !native_p2p {
1005 return Err("MEMRA_STEP_NVFP4_DEV_ROUTES=1 with explicit EP requires \
1006 MEMRA_STEP_TP_NATIVE_P2P=1"
1007 .into());
1008 }
1009 let expert_artifact = if ep_specs.is_empty() {
1012 StepExpertArtifact::default()
1013 } else if nvfp4_device_routes
1014 && matches!(
1015 src.expert_activation_precision(),
1016 memra_gguf::source::ExpertActivationPrecision::Bf16
1017 )
1018 {
1019 StepExpertArtifact::Nvfp4
1024 } else {
1025 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
1026 validate_step_expert_specs(&contract, "MEMRA_STEP_EP", &ep_specs, false)?;
1027 let layer_owners = (0..trunk_layers)
1028 .map(|layer| {
1029 crate::pp::layer_engine(e, trunk_layers, layer)
1030 .map(|engine| engine.ctx().ordinal())
1031 })
1032 .collect::<Result<Vec<_>, _>>()?;
1033 let mut runtime_groups = Vec::<Vec<usize>>::new();
1034 for spec in &ep_specs {
1035 let owner = layer_owners[spec.layer];
1036 if !spec.devices.contains(&owner) {
1037 return Err(format!(
1038 "MEMRA_STEP_EP layer {} owning device {owner} is absent from {:?}",
1039 spec.layer, spec.devices
1040 )
1041 .into());
1042 }
1043 if nvfp4_device_routes && spec.devices.first().copied() != Some(owner) {
1044 return Err(format!(
1045 "MEMRA_STEP_NVFP4_DEV_ROUTES=1 requires the owning device first; \
1046 layer {} owner={owner} devices={:?}",
1047 spec.layer, spec.devices
1048 )
1049 .into());
1050 }
1051 if !runtime_groups.contains(&spec.devices) {
1052 runtime_groups.push(spec.devices.clone());
1053 }
1054 }
1055 for devices in &runtime_groups {
1056 let hardware = crate::parallel::detect_uniform_hardware(devices)?;
1057 if !contract.hardware_targets.contains(&hardware) {
1058 return Err(format!(
1059 "{} has no qualified {hardware:?} EP contract for devices {devices:?}",
1060 contract.variant
1061 )
1062 .into());
1063 }
1064 }
1065 let artifact = match crate::parallel::validate_fp8_expert_checkpoint(src, &contract) {
1066 Ok(_) => StepExpertArtifact::E4m3,
1067 Err(fp8_error) => {
1068 match crate::parallel::validate_nvfp4_expert_checkpoint(src, &contract) {
1069 Ok(_) => StepExpertArtifact::Nvfp4,
1070 Err(nvfp4_error) => {
1071 return Err(format!(
1072 "Step checkpoint qualifies as neither native expert artifact \
1073 class: [E4M3] {fp8_error} [NVFP4] {nvfp4_error}"
1074 )
1075 .into());
1076 }
1077 }
1078 }
1079 };
1080 if nvfp4_device_routes && artifact != StepExpertArtifact::Nvfp4 {
1081 return Err(
1082 "MEMRA_STEP_NVFP4_DEV_ROUTES=1 requires a native ModelOpt NVFP4 expert \
1083 artifact"
1084 .into(),
1085 );
1086 }
1087 artifact
1088 };
1089 return Ok(StepParallelLoadConfig {
1090 ep_specs,
1091 native_p2p,
1092 nvfp4_device_routes,
1093 auto_parallel,
1094 expert_artifact,
1095 ..StepParallelLoadConfig::default()
1096 });
1097 }
1098 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
1099 validate_step_expert_specs(&contract, "MEMRA_STEP_EP", &ep_specs, false)?;
1100 validate_step_expert_specs(&contract, "MEMRA_STEP_TP", &tp_specs, true)?;
1101 for spec in &tp_specs {
1102 let selection = select_step_expert_layout_inner(
1103 spec.layer,
1104 &ep_specs,
1105 &tp_specs,
1106 tp_attention_expert_overlap,
1107 )?
1108 .ok_or("Step TP expert selection disappeared during preflight")?;
1109 validate_step_expert_activation_layout(cfg, "MEMRA_STEP_TP", &selection)?;
1110 }
1111
1112 let layer_owners = (0..trunk_layers)
1113 .map(|layer| {
1114 crate::pp::layer_engine(e, trunk_layers, layer).map(|engine| engine.ctx().ordinal())
1115 })
1116 .collect::<Result<Vec<_>, _>>()?;
1117 let plan = contract.preflight_step_tp_specs(
1118 tp_specs
1119 .iter()
1120 .map(|spec| (spec.layer, spec.devices.as_slice())),
1121 &layer_owners,
1122 )?;
1123
1124 for devices in &plan.runtime_groups {
1125 let hardware = crate::parallel::detect_uniform_hardware(devices)?;
1126 if !contract.hardware_targets.contains(&hardware) {
1127 return Err(format!(
1128 "{} has no qualified {hardware:?} TP contract for devices {devices:?}",
1129 contract.variant
1130 )
1131 .into());
1132 }
1133 }
1134
1135 if bulk_p2p && !native_p2p {
1136 return Err("MEMRA_STEP_TP_BULK_P2P=1 requires MEMRA_STEP_TP_NATIVE_P2P=1".into());
1137 }
1138 if device_arithmetic
1139 && (!ep_specs.is_empty()
1140 || !native_p2p
1141 || plan.expert_parallel_layers() == 0
1142 || plan.tensor_parallel_expert_layers() != 0)
1143 {
1144 return Err(
1145 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires native-P2P TP4/TP8 \
1146 expert ownership for every selected routed-expert layer"
1147 .into(),
1148 );
1149 }
1150 let (qualified_experts, expert_artifact) =
1154 match crate::parallel::validate_fp8_expert_checkpoint(src, &contract) {
1155 Ok(qualified) => (qualified, StepExpertArtifact::E4m3),
1156 Err(fp8_error) => {
1157 match crate::parallel::validate_nvfp4_expert_checkpoint(src, &contract) {
1158 Ok(qualified) => (qualified, StepExpertArtifact::Nvfp4),
1159 Err(nvfp4_error) => {
1160 return Err(format!(
1161 "Step checkpoint qualifies as neither native expert artifact class: \
1162 [E4M3] {fp8_error} [NVFP4] {nvfp4_error}"
1163 )
1164 .into());
1165 }
1166 }
1167 }
1168 };
1169 if expert_artifact == StepExpertArtifact::Nvfp4 {
1170 if device_arithmetic {
1171 return Err(
1172 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 is qualified for the E4M3 expert artifact \
1173 only; the NVFP4 expert program is host-canonical in this increment"
1174 .into(),
1175 );
1176 }
1177 if bulk_p2p {
1182 return Err(
1183 "MEMRA_STEP_TP_BULK_P2P=1 is qualified for the E4M3 expert artifact only; the \
1184 NVFP4 bank transport increment has not landed"
1185 .into(),
1186 );
1187 }
1188 }
1189
1190 if f32_mirror {
1191 eprintln!(
1192 "[step-tp-preflight] layers={} full_trunk={} runtime_groups={} \
1193 dense_attention_layers={} tensor_expert_layers={} expert_owner_layers={} \
1194 qualified_fp8_expert_projection_slices={} owner_first=true \
1195 hardware=rtx-pro-6000-blackwell \
1196 native_p2p={} bulk_p2p={} device_arithmetic={} bf16_residency=f32-mirror \
1197 weights_loaded=false performance_claim=false",
1198 plan.layers.len(),
1199 plan.full_trunk,
1200 plan.runtime_groups.len(),
1201 plan.dense_attention_layers(),
1202 plan.tensor_parallel_expert_layers(),
1203 plan.expert_parallel_layers(),
1204 qualified_experts,
1205 native_p2p,
1206 bulk_p2p,
1207 device_arithmetic,
1208 );
1209 } else {
1210 eprintln!(
1211 "[step-tp-preflight] layers={} full_trunk={} runtime_groups={} \
1212 dense_attention_layers={} tensor_expert_layers={} expert_owner_layers={} \
1213 qualified_fp8_expert_projection_slices={} owner_first=true \
1214 hardware=rtx-pro-6000-blackwell \
1215 native_p2p={} bulk_p2p={} device_arithmetic={} \
1216 weights_loaded=false performance_claim=false",
1217 plan.layers.len(),
1218 plan.full_trunk,
1219 plan.runtime_groups.len(),
1220 plan.dense_attention_layers(),
1221 plan.tensor_parallel_expert_layers(),
1222 plan.expert_parallel_layers(),
1223 qualified_experts,
1224 native_p2p,
1225 bulk_p2p,
1226 device_arithmetic,
1227 );
1228 }
1229 Ok(StepParallelLoadConfig {
1230 ep_specs,
1231 tp_specs,
1232 native_p2p,
1233 ep_device_arithmetic: device_arithmetic,
1234 f32_mirror,
1235 bulk_p2p,
1236 nvfp4_device_routes,
1237 auto_parallel,
1238 tp_attention_expert_overlap,
1239 expert_artifact,
1240 })
1241}
1242
1243fn nvfp4_native_expert_bank<'a>(
1245 src: &'a dyn TensorSource,
1246 layer: usize,
1247 proj: &str,
1248) -> Result<memra_gguf::source::Nvfp4StackedNative<'a>, Box<dyn std::error::Error>> {
1249 let name = format!("blk.{layer}.ffn_{proj}_exps.weight");
1250 src.find_nvfp4_stacked_native(&name)
1251 .ok_or_else(|| format!("NVFP4 expert backend is missing native bank {name}").into())
1252}
1253
1254fn nvfp4_expert_bank_view<'a>(
1256 native: &'a memra_gguf::source::Nvfp4StackedNative<'a>,
1257) -> crate::tp::Nvfp4ExpertBank<'a> {
1258 crate::tp::Nvfp4ExpertBank {
1259 codes: native.codes,
1260 scales: native.scales,
1261 macros: &native.macros,
1262 expert_count: native.n_expert,
1263 out_features: native.out_f,
1264 in_features: native.in_f,
1265 }
1266}
1267
1268#[allow(clippy::too_many_arguments)] fn build_step_distributed_exps(
1270 e: &Engine,
1271 cfg: &ModelConfig,
1272 src: &dyn TensorSource,
1273 layer: usize,
1274 gate: &HostExps,
1275 up: &HostExps,
1276 down: &HostExps,
1277 step_runtimes: &mut StepParallelRuntimeRegistry,
1278) -> Result<(Option<StepEpExps>, Option<StepTpExps>), Box<dyn std::error::Error>> {
1279 let ep_device_arithmetic = step_runtimes.config.ep_device_arithmetic;
1280 if step_runtimes.config.ep_specs.is_empty() && step_runtimes.config.tp_specs.is_empty() {
1281 if ep_device_arithmetic {
1282 return Err(
1283 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires MEMRA_STEP_TP and \
1284 MEMRA_STEP_TP_NATIVE_P2P=1"
1285 .into(),
1286 );
1287 }
1288 return Ok((None, None));
1289 }
1290 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
1291 validate_step_expert_specs(
1292 &contract,
1293 "MEMRA_STEP_EP",
1294 &step_runtimes.config.ep_specs,
1295 false,
1296 )?;
1297 validate_step_expert_specs(
1298 &contract,
1299 "MEMRA_STEP_TP",
1300 &step_runtimes.config.tp_specs,
1301 true,
1302 )?;
1303 let Some(selection) = step_runtimes.expert_selection(layer)? else {
1304 return Ok((None, None));
1305 };
1306 validate_step_expert_activation_layout(
1307 cfg,
1308 if selection.configured_by_tp {
1309 "MEMRA_STEP_TP"
1310 } else {
1311 "MEMRA_STEP_EP"
1312 },
1313 &selection,
1314 )?;
1315 let activation_limit = match cfg.clamp_exp_at(layer as u32) {
1320 None => None,
1321 Some(SwigluClamp::Post(l)) => Some(l),
1322 Some(SwigluClamp::Pre(_)) => {
1323 return Err(format!(
1324 "MEMRA_STEP_EP/TP layer {layer}: glm5_next PRE-clamped SwiGLU has no \
1325 expert-parallel arm (the banks encode step35's post-clamp form)"
1326 )
1327 .into());
1328 }
1329 };
1330 let owner = e.ctx().ordinal();
1331 if !selection.spec.devices.contains(&owner) {
1332 let flag = if selection.configured_by_tp {
1333 "MEMRA_STEP_TP"
1334 } else {
1335 "MEMRA_STEP_EP"
1336 };
1337 return Err(format!(
1338 "{flag} layer {layer} owning PP device {owner} is absent from rank devices {:?}",
1339 selection.spec.devices
1340 )
1341 .into());
1342 }
1343 let expert_parallel = selection.layout == StepExpertLayout::ExpertParallel;
1344 if selection.configured_by_tp {
1345 contract.plan(crate::parallel::TopologyRequest {
1346 pipeline: 1,
1347 tensor: selection.spec.devices.len(),
1348 expert_parallel,
1349 available_devices: selection.spec.devices.len(),
1350 hardware: crate::parallel::HardwareTarget::RtxPro6000Blackwell,
1351 })?;
1352 }
1353 let native_p2p = selection.configured_by_tp && step_runtimes.config.native_p2p;
1354 if ep_device_arithmetic
1355 && (!selection.configured_by_tp
1356 || selection.layout != StepExpertLayout::ExpertParallel
1357 || !native_p2p)
1358 {
1359 return Err(
1360 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires a MEMRA_STEP_TP TP4/TP8 \
1361 expert-owner layer and MEMRA_STEP_TP_NATIVE_P2P=1"
1362 .into(),
1363 );
1364 }
1365 let expert_artifact = step_runtimes.config.expert_artifact;
1366 match selection.layout {
1367 StepExpertLayout::ExpertParallel => {
1368 if expert_artifact == StepExpertArtifact::Nvfp4 {
1369 let w4a16_device_routes = step_runtimes.config.nvfp4_device_routes;
1373 if w4a16_device_routes
1374 && !matches!(
1375 src.expert_activation_precision(),
1376 memra_gguf::source::ExpertActivationPrecision::Bf16
1377 )
1378 {
1379 return Err(
1380 "explicit-EP MEMRA_STEP_NVFP4_DEV_ROUTES=1 requires an artifact that \
1381 declares BF16 routed-expert activations; TP keeps its separately gated \
1382 quantized-activation path"
1383 .into(),
1384 );
1385 }
1386 let runtime = step_runtimes.runtime(
1391 &selection.spec.devices,
1392 step_runtimes.config.native_p2p,
1393 false,
1394 )?;
1395 let experts = runtime.upload_expert_parallel_nvfp4_normalized(gate, up, down)?;
1396 let marker = if step_runtimes.config.auto_parallel {
1397 "parallel-ep"
1398 } else {
1399 "step-ep"
1400 };
1401 eprintln!(
1402 "[{marker}] layer={layer} devices={:?} experts={} artifact=nvfp4 \
1403 expert_layout=expert-parallel expert_transport={} \
1404 macro_fold=post-kernel-once native_p2p={} w4a16_device_routes={} \
1405 performance_claim=false",
1406 selection.spec.devices,
1407 contract.expert_count,
1408 runtime.transport_label(),
1409 runtime.native_p2p(),
1410 w4a16_device_routes,
1411 );
1412 if let Some(limit) = activation_limit {
1413 eprintln!(
1414 "[step-ep-clamp] load layer={layer} routed_clamp={limit} \
1415 formula=min-silu-times-clamped-up performance_claim=false"
1416 );
1417 }
1418 return Ok((
1419 Some(StepEpExps {
1420 runtime,
1421 experts: StepEpExpertBank::Nvfp4(experts),
1422 devices: selection.spec.devices,
1423 configured_by_tp: selection.configured_by_tp,
1424 activation_limit,
1425 nvfp4_device_routes: w4a16_device_routes,
1426 grouped_decode: None,
1427 }),
1428 None,
1429 ));
1430 }
1431 let runtime =
1432 step_runtimes.runtime(&selection.spec.devices, native_p2p, ep_device_arithmetic)?;
1433 let experts = runtime.upload_expert_parallel(
1434 host_e4m3_bank(gate)?,
1435 host_e4m3_bank(up)?,
1436 host_e4m3_bank(down)?,
1437 )?;
1438 let grouped_decode = if ep_device_arithmetic {
1439 let tokens = 1;
1440 let selected = (0..contract.experts_per_token).collect::<Vec<_>>();
1441 let input = vec![0.0f32; contract.hidden_size];
1442 let route_weights = vec![1.0f32; contract.experts_per_token];
1443 let projection = runtime.prepare_step_grouped_expert_parallel_gate_with_capacity(
1444 &experts,
1445 &input,
1446 tokens,
1447 &selected,
1448 activation_limit,
1449 tokens,
1450 )?;
1451 let combine = runtime
1452 .prepare_step_grouped_expert_parallel_combine(&projection, &route_weights)?;
1453 Some(std::sync::Mutex::new(StepEpGroupedDecode {
1454 projection,
1455 combine,
1456 }))
1457 } else {
1458 None
1459 };
1460 if selection.configured_by_tp {
1461 eprintln!(
1462 "[step-tp-ep] layer={layer} devices={:?} experts={} tp={} \
1463 attention_layout=tensor-parallel expert_layout=expert-parallel \
1464 expert_transport={} tp_transport={} native_p2p={} \
1465 activation={} accumulation={} output={} \
1466 grouped_decode_prepared={} grouped_decode_capacity=1 \
1467 performance_claim=false",
1468 selection.spec.devices,
1469 contract.expert_count,
1470 selection.spec.devices.len(),
1471 runtime.transport_label(),
1472 runtime.transport_label(),
1473 runtime.native_p2p(),
1474 runtime.expert_activation_label(),
1475 runtime.expert_accumulation_label(),
1476 runtime.expert_output_label(),
1477 grouped_decode.is_some(),
1478 );
1479 } else {
1480 eprintln!(
1481 "[step-ep] layer={layer} devices={:?} experts={} \
1482 expert_layout=expert-parallel expert_transport=host-bounce \
1483 native_p2p=false performance_claim=false",
1484 selection.spec.devices, contract.expert_count
1485 );
1486 }
1487 if let Some(limit) = activation_limit {
1488 eprintln!(
1489 "[step-ep-clamp] load layer={layer} routed_clamp={limit} \
1490 formula=min-silu-times-clamped-up performance_claim=false"
1491 );
1492 }
1493 Ok((
1494 Some(StepEpExps {
1495 runtime,
1496 experts: StepEpExpertBank::E4m3(experts),
1497 devices: selection.spec.devices,
1498 configured_by_tp: selection.configured_by_tp,
1499 activation_limit,
1500 nvfp4_device_routes: false,
1501 grouped_decode,
1502 }),
1503 None,
1504 ))
1505 }
1506 StepExpertLayout::TensorParallel => {
1507 let runtime =
1508 step_runtimes.runtime(&selection.spec.devices, native_p2p, ep_device_arithmetic)?;
1509 if activation_limit.is_some() && expert_artifact == StepExpertArtifact::E4m3 {
1510 return Err(format!(
1511 "layer {layer} uses the routed SwiGLU clamp and the E4M3 TP expert \
1512 program has no clamp arm; select EP for this layer (the NVFP4 TP \
1513 program carries the clamp)"
1514 )
1515 .into());
1516 }
1517 let experts = if expert_artifact == StepExpertArtifact::Nvfp4 {
1518 let gate_native = nvfp4_native_expert_bank(src, layer, "gate")?;
1519 let up_native = nvfp4_native_expert_bank(src, layer, "up")?;
1520 let down_native = nvfp4_native_expert_bank(src, layer, "down")?;
1521 StepTpExpertBank::Nvfp4(runtime.upload_tensor_parallel_nvfp4(
1522 nvfp4_expert_bank_view(&gate_native),
1523 nvfp4_expert_bank_view(&up_native),
1524 nvfp4_expert_bank_view(&down_native),
1525 )?)
1526 } else {
1527 StepTpExpertBank::E4m3(runtime.upload_tensor_parallel(
1528 host_e4m3_bank(gate)?,
1529 host_e4m3_bank(up)?,
1530 host_e4m3_bank(down)?,
1531 )?)
1532 };
1533 eprintln!(
1534 "[step-tp] layer={layer} devices={:?} experts={} tp={} artifact={} \
1535 expert_layout=tensor-parallel transport={} native_p2p={} \
1536 performance_claim=false",
1537 selection.spec.devices,
1538 contract.expert_count,
1539 selection.spec.devices.len(),
1540 match expert_artifact {
1541 StepExpertArtifact::E4m3 => "e4m3",
1542 StepExpertArtifact::Nvfp4 => "nvfp4",
1543 },
1544 runtime.transport_label(),
1545 runtime.native_p2p(),
1546 );
1547 if let Some(limit) = activation_limit {
1548 eprintln!(
1549 "[step-tp-clamp] load layer={layer} routed_clamp={limit} \
1550 formula=min-silu-times-clamped-up performance_claim=false"
1551 );
1552 }
1553 Ok((
1554 None,
1555 Some(StepTpExps {
1556 runtime,
1557 experts,
1558 devices: selection.spec.devices,
1559 activation_limit,
1560 }),
1561 ))
1562 }
1563 }
1564}
1565
1566fn upload_step_bf16_column(
1567 runtime: &crate::tp::TpE4m3HostBounce,
1568 src: &dyn TensorSource,
1569 name: &str,
1570 expected_in: usize,
1571 expected_out: usize,
1572 f32_mirror: bool,
1573) -> Result<crate::tp::ResidentBf16ColumnParallel, Box<dyn std::error::Error>> {
1574 let tensor = src
1575 .find(name)
1576 .ok_or_else(|| format!("Step TP projection is missing {name}"))?;
1577 if tensor.ggml_type != GgmlType::BF16 {
1578 return Err(format!(
1579 "Step TP projection {name} must preserve checkpoint BF16 bytes, got {:?}",
1580 tensor.ggml_type
1581 )
1582 .into());
1583 }
1584 if tensor.ne.len() != 2 {
1585 return Err(format!(
1586 "Step TP projection {name} must be a 2-D matrix, got shape {:?}",
1587 tensor.ne
1588 )
1589 .into());
1590 }
1591 let matrix = crate::tp::Bf16Matrix {
1592 bytes: tensor.bytes.as_ref(),
1593 in_features: tensor.ne[0] as usize,
1594 out_features: tensor.ne[1] as usize,
1595 };
1596 matrix.validate()?;
1597 if matrix.in_features != expected_in || matrix.out_features != expected_out {
1598 return Err(format!(
1599 "Step TP projection {name} shape {}x{} != registered {expected_out}x{expected_in}",
1600 matrix.out_features, matrix.in_features
1601 )
1602 .into());
1603 }
1604 Ok(if f32_mirror {
1605 runtime.upload_step_bf16_column_parallel_f32_mirror(matrix)?
1606 } else {
1607 runtime.upload_step_bf16_column_parallel(matrix)?
1608 })
1609}
1610
1611fn upload_step_bf16_row(
1612 runtime: &crate::tp::TpE4m3HostBounce,
1613 src: &dyn TensorSource,
1614 name: &str,
1615 expected_in: usize,
1616 expected_out: usize,
1617 f32_mirror: bool,
1618) -> Result<crate::tp::ResidentStepBf16RowParallel, Box<dyn std::error::Error>> {
1619 let tensor = src
1620 .find(name)
1621 .ok_or_else(|| format!("Step TP projection is missing {name}"))?;
1622 if tensor.ggml_type != GgmlType::BF16 {
1623 return Err(format!(
1624 "Step TP projection {name} must preserve checkpoint BF16 bytes, got {:?}",
1625 tensor.ggml_type
1626 )
1627 .into());
1628 }
1629 if tensor.ne.len() != 2 {
1630 return Err(format!(
1631 "Step TP projection {name} must be a 2-D matrix, got shape {:?}",
1632 tensor.ne
1633 )
1634 .into());
1635 }
1636 let matrix = crate::tp::Bf16Matrix {
1637 bytes: tensor.bytes.as_ref(),
1638 in_features: tensor.ne[0] as usize,
1639 out_features: tensor.ne[1] as usize,
1640 };
1641 matrix.validate()?;
1642 if matrix.in_features != expected_in || matrix.out_features != expected_out {
1643 return Err(format!(
1644 "Step TP projection {name} shape {}x{} != registered {expected_out}x{expected_in}",
1645 matrix.out_features, matrix.in_features
1646 )
1647 .into());
1648 }
1649 Ok(if f32_mirror {
1650 runtime.upload_step_bf16_row_parallel_f32_mirror(matrix)?
1651 } else {
1652 runtime.upload_step_bf16_row_parallel(matrix)?
1653 })
1654}
1655
1656fn upload_step_tp_f32_copies(
1657 runtime: &crate::tp::TpE4m3HostBounce,
1658 src: &dyn TensorSource,
1659 name: &str,
1660 expected: usize,
1661) -> Result<Vec<CudaSlice<f32>>, Box<dyn std::error::Error>> {
1662 let tensor = src
1663 .find(name)
1664 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1665 let values = memra_gguf::dequant::dequantize(
1666 tensor.ggml_type,
1667 &tensor.bytes,
1668 tensor.ne.iter().product::<u64>() as usize,
1669 );
1670 if values.len() != expected || values.iter().any(|value| !value.is_finite()) {
1671 return Err(format!(
1672 "Step TP attention {name} has {} finite values, expected {expected}",
1673 values.len()
1674 )
1675 .into());
1676 }
1677 let mut copies = Vec::with_capacity(runtime.devices().len());
1678 for rank in 0..runtime.devices().len() {
1679 let engine = runtime
1680 .rank_engine(rank)
1681 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1682 let _main = engine.gpu.enter_main()?;
1683 copies.push(engine.htod(&values)?);
1684 }
1685 Ok(copies)
1686}
1687
1688#[allow(clippy::manual_is_multiple_of)] fn upload_step_tp_f32_row_shards(
1694 runtime: &crate::tp::TpE4m3HostBounce,
1695 src: &dyn TensorSource,
1696 name: &str,
1697 rows: usize,
1698 cols: usize,
1699) -> Result<Vec<CudaSlice<f32>>, Box<dyn std::error::Error>> {
1700 let tensor = src
1701 .find(name)
1702 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1703 let values = memra_gguf::dequant::dequantize(
1704 tensor.ggml_type,
1705 &tensor.bytes,
1706 tensor.ne.iter().product::<u64>() as usize,
1707 );
1708 let world = runtime.devices().len();
1709 if values.len() != rows * cols || rows % world != 0 || values.iter().any(|v| !v.is_finite()) {
1710 return Err(format!(
1711 "Step TP attention {name} has {} finite values, expected {rows}x{cols} \
1712 (rows divisible by world {world})",
1713 values.len()
1714 )
1715 .into());
1716 }
1717 let local_rows = rows / world;
1718 let mut shards = Vec::with_capacity(world);
1719 for rank in 0..world {
1720 let engine = runtime
1721 .rank_engine(rank)
1722 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1723 let _main = engine.gpu.enter_main()?;
1724 shards
1725 .push(engine.htod(&values[rank * local_rows * cols..(rank + 1) * local_rows * cols])?);
1726 }
1727 Ok(shards)
1728}
1729
1730#[allow(clippy::manual_is_multiple_of)] fn upload_step_tp_bf16_row_shards(
1733 runtime: &crate::tp::TpE4m3HostBounce,
1734 src: &dyn TensorSource,
1735 name: &str,
1736 rows: usize,
1737 cols: usize,
1738) -> Result<Vec<CudaSlice<u8>>, Box<dyn std::error::Error>> {
1739 let tensor = src
1740 .find(name)
1741 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1742 if tensor.ggml_type != memra_gguf::GgmlType::BF16 || tensor.bytes.len() != rows * cols * 2 {
1743 return Err(format!(
1744 "Step TP attention {name} is not a bf16 [{rows}, {cols}] tensor ({} bytes, {:?})",
1745 tensor.bytes.len(),
1746 tensor.ggml_type
1747 )
1748 .into());
1749 }
1750 let world = runtime.devices().len();
1751 if rows % world != 0 {
1752 return Err(format!("{name} rows {rows} not divisible by world {world}").into());
1753 }
1754 let local = rows / world * cols * 2;
1755 let mut shards = Vec::with_capacity(world);
1756 for rank in 0..world {
1757 let engine = runtime
1758 .rank_engine(rank)
1759 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1760 let _main = engine.gpu.enter_main()?;
1761 shards.push(engine.htod_bytes(&tensor.bytes[rank * local..(rank + 1) * local])?);
1762 }
1763 Ok(shards)
1764}
1765
1766#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1767enum StepTpAttentionPlacement {
1768 RankLocalGlobal,
1769 RankLocalSwa,
1770 OwnerSwa,
1771 OwnerTransportFallback,
1772}
1773
1774impl StepTpAttentionPlacement {
1775 fn resolve(native_p2p: bool, window: Option<u32>) -> Self {
1776 match (native_p2p, window.is_some()) {
1777 (true, true) => Self::RankLocalSwa,
1778 (false, true) => Self::OwnerSwa,
1779 (true, false) => Self::RankLocalGlobal,
1780 (false, false) => Self::OwnerTransportFallback,
1781 }
1782 }
1783
1784 fn is_rank_local(self) -> bool {
1785 matches!(self, Self::RankLocalGlobal | Self::RankLocalSwa)
1786 }
1787
1788 fn label(self) -> &'static str {
1789 match self {
1790 Self::RankLocalGlobal => "rank-local-global",
1791 Self::RankLocalSwa => "rank-local-swa-ring",
1792 Self::OwnerSwa => "owner-swa",
1793 Self::OwnerTransportFallback => "owner-transport-fallback",
1794 }
1795 }
1796}
1797
1798fn build_step_tp_qkv(
1799 e: &Engine,
1800 src: &dyn TensorSource,
1801 cfg: &ModelConfig,
1802 layer: usize,
1803 step_runtimes: &mut StepParallelRuntimeRegistry,
1804) -> Result<Option<StepTpQkv>, Box<dyn std::error::Error>> {
1805 let Some(spec) = step_runtimes.tp_spec(layer).cloned() else {
1806 return Ok(None);
1807 };
1808 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
1809 if layer >= contract.trunk_layers {
1810 return Err(format!(
1811 "MEMRA_STEP_TP layer {layer} is outside Step trunk layers 0..{}",
1812 contract.trunk_layers
1813 )
1814 .into());
1815 }
1816 let owner = e.ctx().ordinal();
1817 if spec.devices.first().copied() != Some(owner) {
1818 return Err(format!(
1819 "MEMRA_STEP_TP layer {layer} owning PP device {owner} must be the first QKV rank, \
1820 got {:?}",
1821 spec.devices
1822 )
1823 .into());
1824 }
1825 let plan = contract.plan(crate::parallel::TopologyRequest {
1826 pipeline: 1,
1827 tensor: spec.devices.len(),
1828 expert_parallel: spec.devices.len() > 2,
1829 available_devices: spec.devices.len(),
1830 hardware: crate::parallel::HardwareTarget::RtxPro6000Blackwell,
1831 })?;
1832 for rank in 0..spec.devices.len() {
1833 plan.query_head_range(layer, rank).ok_or_else(|| {
1834 format!("Step TP layer {layer} has no query-head range for rank {rank}")
1835 })?;
1836 plan.kv_head_range(layer, rank)
1837 .ok_or_else(|| format!("Step TP layer {layer} has no KV-head range for rank {rank}"))?;
1838 }
1839 let native_p2p = step_runtimes.config.native_p2p;
1840 let ep_device_arithmetic = step_runtimes.config.ep_device_arithmetic;
1841 let f32_mirror = step_runtimes.config.f32_mirror;
1842 if ep_device_arithmetic && (!native_p2p || !matches!(spec.devices.len(), 4 | 8)) {
1843 return Err(
1844 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires a MEMRA_STEP_TP TP4/TP8 \
1845 expert-owner layer and MEMRA_STEP_TP_NATIVE_P2P=1"
1846 .into(),
1847 );
1848 }
1849 let runtime = step_runtimes.runtime(&spec.devices, native_p2p, ep_device_arithmetic)?;
1850 let p = |suffix: &str| format!("blk.{layer}.{suffix}");
1851 let q = upload_step_bf16_column(
1852 &runtime,
1853 src,
1854 &p("attn_q.weight"),
1855 contract.hidden_size,
1856 contract.query_heads[layer] * contract.head_dim,
1857 f32_mirror,
1858 )?;
1859 let k = upload_step_bf16_column(
1860 &runtime,
1861 src,
1862 &p("attn_k.weight"),
1863 contract.hidden_size,
1864 contract.kv_heads[layer] * contract.head_dim,
1865 f32_mirror,
1866 )?;
1867 let v = upload_step_bf16_column(
1868 &runtime,
1869 src,
1870 &p("attn_v.weight"),
1871 contract.hidden_size,
1872 contract.kv_heads[layer] * contract.head_dim,
1873 f32_mirror,
1874 )?;
1875 let o = upload_step_bf16_row(
1876 &runtime,
1877 src,
1878 &p("attn_output.weight"),
1879 contract.query_heads[layer] * contract.head_dim,
1880 contract.hidden_size,
1881 f32_mirror,
1882 )?;
1883 let geometry = cfg.full_attention_geometry_at(layer as u32);
1884 let attention_placement =
1885 StepTpAttentionPlacement::resolve(runtime.native_p2p(), geometry.window);
1886 let attention = if attention_placement.is_rank_local() {
1887 let decode_input = if ep_device_arithmetic || crate::tp::step_tp_decode_v2_enabled()? {
1892 Some(std::sync::Mutex::new(
1893 runtime.allocate_replicated_device_rows(1, contract.hidden_size)?,
1894 ))
1895 } else {
1896 None
1897 };
1898 let gate_fused =
1901 crate::tp::step_tp_qkv_fused_enabled()? && src.find(&p("attn_gate.weight")).is_some();
1902 let gate_shards = if gate_fused && f32_mirror {
1903 Some(upload_step_tp_f32_row_shards(
1904 &runtime,
1905 src,
1906 &p("attn_gate.weight"),
1907 contract.query_heads[layer],
1908 contract.hidden_size,
1909 )?)
1910 } else {
1911 None
1912 };
1913 let gate_shards_bf16 = if gate_fused && !f32_mirror {
1914 Some(upload_step_tp_bf16_row_shards(
1915 &runtime,
1916 src,
1917 &p("attn_gate.weight"),
1918 contract.query_heads[layer],
1919 contract.hidden_size,
1920 )?)
1921 } else {
1922 None
1923 };
1924 Some(StepTpAttention {
1925 q_norm: upload_step_tp_f32_copies(
1926 &runtime,
1927 src,
1928 &p("attn_q_norm.weight"),
1929 contract.head_dim,
1930 )?,
1931 k_norm: upload_step_tp_f32_copies(
1932 &runtime,
1933 src,
1934 &p("attn_k_norm.weight"),
1935 contract.head_dim,
1936 )?,
1937 decode_input,
1938 gate_shards,
1939 gate_shards_bf16,
1940 })
1941 } else {
1942 None
1943 };
1944 if f32_mirror {
1945 eprintln!(
1946 "[step-tp-qkv] load layer={layer} devices={:?} projections=qkv \
1947 qkv_tensor_parallel=true attention_local=true kv_local=true output_local=true \
1948 transport={} native_p2p={} bf16_residency=f32-mirror \
1949 output=root-readback performance_claim=false",
1950 spec.devices,
1951 runtime.transport_label(),
1952 runtime.native_p2p(),
1953 );
1954 } else {
1955 eprintln!(
1956 "[step-tp-qkv] load layer={layer} devices={:?} projections=qkv \
1957 qkv_tensor_parallel=true attention_local=true kv_local=true output_local=true \
1958 transport={} native_p2p={} output=root-readback performance_claim=false",
1959 spec.devices,
1960 runtime.transport_label(),
1961 runtime.native_p2p(),
1962 );
1963 }
1964 eprintln!(
1965 "[step-tp-attn-plan] load layer={layer} devices={:?} \
1966 qkv_tensor_parallel=true attention_tensor_parallel={} kv_cache_distributed={} \
1967 attention_scope={} transport={} native_p2p={} replicated_decode_input_prepared={} \
1968 performance_claim=false",
1969 spec.devices,
1970 attention_placement.is_rank_local(),
1971 attention_placement.is_rank_local(),
1972 attention_placement.label(),
1973 runtime.transport_label(),
1974 runtime.native_p2p(),
1975 attention
1976 .as_ref()
1977 .is_some_and(|attention| attention.decode_input.is_some()),
1978 );
1979 if f32_mirror {
1980 eprintln!(
1981 "[step-tp-o] load layer={layer} devices={:?} projection=o \
1982 o_tensor_parallel=true attention_local=true kv_local=true \
1983 transport={} native_p2p={} reduction=global-tp8-block-order \
1984 bf16_residency=f32-mirror output=root-readback performance_claim=false",
1985 spec.devices,
1986 runtime.transport_label(),
1987 runtime.native_p2p(),
1988 );
1989 } else {
1990 eprintln!(
1991 "[step-tp-o] load layer={layer} devices={:?} projection=o \
1992 o_tensor_parallel=true attention_local=true kv_local=true \
1993 transport={} native_p2p={} reduction=global-tp8-block-order \
1994 output=root-readback performance_claim=false",
1995 spec.devices,
1996 runtime.transport_label(),
1997 runtime.native_p2p(),
1998 );
1999 }
2000 Ok(Some(StepTpQkv {
2001 runtime,
2002 q,
2003 k,
2004 v,
2005 o,
2006 attention,
2007 devices: spec.devices,
2008 layer,
2009 }))
2010}
2011
2012fn build_dev_exps(
2025 e: &Engine,
2026 resident: &mut ResidentPlan,
2027 il: usize,
2028 gate: &HostExps,
2029 up: &HostExps,
2030 down: &HostExps,
2031) -> Result<Option<crate::hybrid::DevExps>, Box<dyn std::error::Error>> {
2032 if !gate.is_uniform_layout() || !up.is_uniform_layout() || !down.is_uniform_layout() {
2035 return Ok(None);
2036 }
2037 let fp8_host = match (&gate.fp8_blk, &up.fp8_blk, &down.fp8_blk) {
2038 (None, None, None) => None,
2039 (Some(g), Some(u), Some(d)) => Some((g, u, d)),
2040 _ => {
2041 return Err("resident expert projections disagree on block-E4M3 scale carriage".into());
2042 }
2043 };
2044 let scale_bytes = fp8_host
2045 .map(|(g, u, d)| (g.scales.len() + u.scales.len() + d.scales.len()) * size_of::<f32>())
2046 .unwrap_or(0);
2047 let per_layer = gate.bytes.as_bytes().len()
2048 + up.bytes.as_bytes().len()
2049 + down.bytes.as_bytes().len()
2050 + scale_bytes;
2051 if gate.tiers.is_some() {
2052 return Ok(None); }
2054 let fits = resident.should_reside(e, il, per_layer);
2055 if !fits {
2056 return Ok(None);
2057 }
2058 use cudarc::driver::DevicePtr;
2059 let gu_il = std::env::var("MEMRA_MOE_GU_IL").as_deref() == Ok("1")
2060 && gate.out_f == up.out_f
2061 && gate.in_f == up.in_f
2062 && fp8_host.is_none();
2063 let n_expert = gate.n_expert;
2064 let (g, u) = if gu_il {
2065 let (rbg, rbu) = (gate.row_bytes, up.row_bytes);
2067 let n_rows = gate.out_f;
2068 let gb = gate.bytes.as_bytes();
2069 let ub = up.bytes.as_bytes();
2070 let mut il = vec![0u8; n_expert * n_rows * (rbg + rbu)];
2071 for ex in 0..n_expert {
2072 for o in 0..n_rows {
2073 let dst = (ex * n_rows + o) * (rbg + rbu);
2074 let sg = ex * gate.expert_stride + o * rbg;
2075 let su = ex * up.expert_stride + o * rbu;
2076 il[dst..dst + rbg].copy_from_slice(&gb[sg..sg + rbg]);
2077 il[dst + rbg..dst + rbg + rbu].copy_from_slice(&ub[su..su + rbu]);
2078 }
2079 }
2080 let ild = e.htod_bytes_padded(&il, 8)?;
2081 (ild, e.htod_bytes(&[0u8; 16])?)
2084 } else {
2085 (
2086 e.htod_bytes_padded(gate.bytes.as_bytes(), 8)?,
2087 e.htod_bytes_padded(up.bytes.as_bytes(), 8)?,
2088 )
2089 };
2090 let d = e.htod_bytes_padded(down.bytes.as_bytes(), 144)?;
2095 let fp8_blk = match fp8_host {
2096 Some((gate, up, down)) => {
2097 if e.fp8_blk_nan_count(&g)? != 0
2098 || e.fp8_blk_nan_count(&u)? != 0
2099 || e.fp8_blk_nan_count(&d)? != 0
2100 {
2101 return Err("native stacked block-E4M3 expert bank contains NaN codes".into());
2102 }
2103 Some(DevExpertFp8BlockScales {
2104 gate: DevExpertFp8ProjectionScales::upload(e, gate, n_expert)?,
2105 up: DevExpertFp8ProjectionScales::upload(e, up, n_expert)?,
2106 down: DevExpertFp8ProjectionScales::upload(e, down, n_expert)?,
2107 })
2108 }
2109 None => None,
2110 };
2111 let mut host = vec![0u64; 3 * n_expert];
2112 let (pg, pu, pd) = {
2113 let __s_e0 = e.stream();
2114 let (pg, _e0) = g.device_ptr(&__s_e0);
2115 let __s_e1 = e.stream();
2116 let (pu, _e1) = u.device_ptr(&__s_e1);
2117 let __s_e2 = e.stream();
2118 let (pd, _e2) = d.device_ptr(&__s_e2);
2119 (pg, pu, pd)
2120 };
2121 for ex in 0..n_expert {
2122 if gu_il {
2123 let stride = gate.out_f * (gate.row_bytes + up.row_bytes);
2124 host[ex] = pg + (ex * stride) as u64;
2125 host[n_expert + ex] = pg + (ex * stride + gate.row_bytes) as u64;
2126 } else {
2127 host[ex] = pg + (ex * gate.expert_stride) as u64;
2128 host[n_expert + ex] = pu + (ex * up.expert_stride) as u64;
2129 }
2130 host[2 * n_expert + ex] = pd + (ex * down.expert_stride) as u64;
2131 }
2132 if gu_il {
2133 eprintln!("[moe] gate/up dev slab INTERLEAVED (MEMRA_MOE_GU_IL)");
2134 }
2135 let ptr_row = e.htod_u64(&host)?;
2136 Ok(Some(crate::hybrid::DevExps {
2137 gate: g,
2138 up: u,
2139 down: d,
2140 ptr_row,
2141 gu_il,
2142 dev: e.ctx().ordinal(),
2143 fp8_blk,
2144 }))
2145}
2146
2147pub struct FullAttnLayer {
2148 pub wq: GpuTensor,
2149 pub wk: GpuTensor,
2150 pub wv: GpuTensor,
2151 pub wo: GpuTensor,
2152 pub q_norm: GpuTensor,
2153 pub k_norm: GpuTensor,
2154 pub attn_gate: Option<GpuTensor>,
2165 pub step_tp_qkv: Option<StepTpQkv>,
2169}
2170
2171pub struct StepTpQkv {
2172 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
2173 pub q: crate::tp::ResidentBf16ColumnParallel,
2174 pub k: crate::tp::ResidentBf16ColumnParallel,
2175 pub v: crate::tp::ResidentBf16ColumnParallel,
2176 pub o: crate::tp::ResidentStepBf16RowParallel,
2177 pub attention: Option<StepTpAttention>,
2178 pub devices: Vec<usize>,
2179 pub layer: usize,
2180}
2181
2182pub struct StepTpAttention {
2183 pub q_norm: Vec<CudaSlice<f32>>,
2184 pub k_norm: Vec<CudaSlice<f32>>,
2185 pub decode_input: Option<std::sync::Mutex<crate::tp::ResidentReplicatedDeviceRows>>,
2186 pub gate_shards: Option<Vec<CudaSlice<f32>>>,
2189 pub gate_shards_bf16: Option<Vec<CudaSlice<u8>>>,
2191}
2192
2193#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2194pub struct StepTpKvDeviceAdmission {
2195 pub device: usize,
2196 pub bytes: usize,
2197}
2198
2199#[derive(Clone, Copy, Debug)]
2203pub struct MlaGeom {
2204 pub n_head: usize, pub d_nope: usize, pub d_rope: usize, pub d_v: usize, pub kv_rank: usize, pub latent_dim: usize, pub scale: f32, }
2212
2213#[derive(Clone, Copy, Debug)]
2218pub struct MlaIndexerGeom {
2219 pub heads: usize, pub head_dim: usize, pub top_k: usize, pub pool: usize, pub always_select_tail: bool,
2224}
2225
2226impl MlaIndexerGeom {
2227 pub fn select_k(&self, n_pools: usize) -> usize {
2229 (self.top_k / self.pool).min(n_pools)
2230 }
2231
2232 pub fn index_width(&self, n_pools: usize) -> usize {
2234 self.select_k(n_pools) * self.pool
2235 + if self.always_select_tail {
2236 self.pool - 1
2237 } else {
2238 0
2239 }
2240 }
2241
2242 pub fn state_width(&self) -> usize {
2244 2 * self.head_dim
2245 }
2246}
2247
2248pub struct MlaIndexer {
2252 pub wq_b: GpuTensor, pub wk: GpuTensor, pub k_norm_w: GpuTensor, pub k_norm_b: GpuTensor, pub weights_proj: GpuTensor, pub kpool_gate: GpuTensor, pub kpool_ape: GpuTensor, pub geom: MlaIndexerGeom,
2260}
2261
2262pub struct MlaAttnLayer {
2263 pub wq_a: GpuTensor, pub q_a_norm: GpuTensor, pub wq_b: GpuTensor, pub wkv_a: GpuTensor, pub kv_a_norm: GpuTensor, pub wk_b: GpuTensor, pub wv_b: GpuTensor, pub wo: GpuTensor, pub geom: MlaGeom,
2273 pub index: Option<MlaIndexer>,
2276 pub tp: Option<Box<crate::glm5_tp::Glm5TpMla>>,
2281 pub tp_shard: bool,
2288}
2289
2290impl MlaAttnLayer {
2291 pub fn load(
2303 e: &Engine,
2304 src: &dyn TensorSource,
2305 il: u32,
2306 plan: &memra_gguf::model_plan::MlaAttentionPlan,
2307 ) -> Result<Self, Box<dyn std::error::Error>> {
2308 let memra_gguf::model_plan::MlaAttentionPlan::LatentKv {
2309 query_heads,
2310 q_lora_rank,
2311 kv_lora_rank,
2312 qk_head_dim,
2313 rope_head_dim,
2314 value_head_dim,
2315 sparse_index,
2316 ..
2317 } = plan
2318 else {
2319 return Err(format!(
2320 "native MLA loader has no compressed-KV implementation for block {il}"
2321 )
2322 .into());
2323 };
2324 let d_nope = qk_head_dim
2325 .checked_sub(*rope_head_dim)
2326 .ok_or("MLA rope head width exceeds total QK head width")?;
2327 let p = |s: &str| format!("blk.{il}.{s}");
2328 let geom = MlaGeom {
2329 n_head: *query_heads as usize,
2330 d_nope: d_nope as usize,
2331 d_rope: *rope_head_dim as usize,
2332 d_v: *value_head_dim as usize,
2333 kv_rank: *kv_lora_rank as usize,
2334 latent_dim: (*kv_lora_rank + *rope_head_dim) as usize,
2335 scale: 1.0 / (*qk_head_dim as f32).sqrt(),
2336 };
2337 let wq_a = load_t(e, src, &p("attn_q_a.weight"))?;
2338 let wq_b = load_t(e, src, &p("attn_q_b.weight"))?;
2339 let wkv_a = load_t(e, src, &p("attn_kv_a_mqa.weight"))?;
2340 let wk_b = load_t(e, src, &p("attn_k_b.weight"))?;
2341 let wv_b = load_t(e, src, &p("attn_v_b.weight"))?;
2342 let wo = load_t(e, src, &p("attn_output.weight"))?;
2343 for (w, tensor) in [(&wk_b, "attn_k_b"), (&wv_b, "attn_v_b")] {
2352 if !matches!(w, GpuTensor::Float { .. }) {
2353 return Err(format!(
2354 "blk.{il}.{tensor}.weight is not f32-resident. The MLA conversion-split \
2355 operands feed f32-only absorb/decompress kernels; the checkpoint source must \
2356 dequantize them (TensorTransform::SplitMlaKv) rather than hand the engine a \
2357 quantized plane"
2358 )
2359 .into());
2360 }
2361 }
2362 let n_head = wq_b.out_features() / (geom.d_nope + geom.d_rope);
2364 assert_eq!(
2365 wq_b.out_features(),
2366 n_head * (geom.d_nope + geom.d_rope),
2367 "wq_b out {} not a multiple of qk_head_dim {}",
2368 wq_b.out_features(),
2369 geom.d_nope + geom.d_rope
2370 );
2371 assert_eq!(
2372 wq_a.in_features(),
2373 wkv_a.in_features(),
2374 "q_a/kv_a hidden mismatch"
2375 );
2376 assert_eq!(
2377 wq_b.in_features(),
2378 *q_lora_rank as usize,
2379 "wq_b in != q_lora_rank"
2380 );
2381 assert_eq!(
2382 n_head, geom.n_head,
2383 "MLA checkpoint head count != ModelPlan"
2384 );
2385 assert_eq!(
2386 wkv_a.out_features(),
2387 geom.latent_dim,
2388 "wkv_a out != kv_lora_rank + rope"
2389 );
2390 assert_eq!(
2391 wk_b.ne(),
2392 &[geom.d_nope as u64, geom.kv_rank as u64, n_head as u64],
2393 "attn_k_b must be the TRANSPOSED (nope, kv_rank, head) conversion split"
2394 );
2395 assert_eq!(
2396 wv_b.ne(),
2397 &[geom.kv_rank as u64, geom.d_v as u64, n_head as u64],
2398 "attn_v_b must be the (kv_rank, v, head) conversion split"
2399 );
2400 assert_eq!(
2401 wo.in_features(),
2402 n_head * geom.d_v,
2403 "wo in != n_head * v_head_dim"
2404 );
2405 let index = Self::load_indexer(e, src, il, sparse_index, *q_lora_rank)?;
2406 Ok(MlaAttnLayer {
2407 wq_a,
2408 q_a_norm: load_t(e, src, &p("attn_q_a_norm.weight"))?,
2409 wq_b,
2410 wkv_a,
2411 kv_a_norm: load_t(e, src, &p("attn_kv_a_norm.weight"))?,
2412 wk_b,
2413 wv_b,
2414 wo,
2415 geom,
2416 index,
2417 tp: None,
2418 tp_shard: false,
2419 })
2420 }
2421
2422 fn load_indexer(
2434 e: &Engine,
2435 src: &dyn TensorSource,
2436 il: u32,
2437 sparse_index: &memra_gguf::model_plan::SparseIndexPlan,
2438 q_lora_rank: u32,
2439 ) -> Result<Option<MlaIndexer>, Box<dyn std::error::Error>> {
2440 let memra_gguf::model_plan::SparseIndexPlan::Own {
2441 heads,
2442 head_dim,
2443 top_k,
2444 kpool: Some(kpool),
2445 } = sparse_index
2446 else {
2447 return Ok(None);
2448 };
2449 let geom = MlaIndexerGeom {
2450 heads: *heads as usize,
2451 head_dim: *head_dim as usize,
2452 top_k: *top_k as usize,
2453 pool: kpool.pool as usize,
2454 always_select_tail: kpool.always_select_tail,
2455 };
2456 if geom.heads == 0 || geom.head_dim == 0 || geom.pool == 0 || geom.top_k < geom.pool {
2457 return Err(format!(
2458 "blk.{il}: SparseIndexPlan::Own declares an unusable k-pool indexer \
2459 (heads {}, head_dim {}, pool {}, top_k {}) — heads/head_dim/pool must be \
2460 positive and top_k must admit at least one pool",
2461 geom.heads, geom.head_dim, geom.pool, geom.top_k
2462 )
2463 .into());
2464 }
2465 let need = |suffix: &str| -> Result<GpuTensor, Box<dyn std::error::Error>> {
2470 let name = format!("blk.{il}.{suffix}");
2471 if !src.has(&name) {
2472 return Err(format!(
2473 "blk.{il}: the layer's ModelPlan declares a DSA k-pool indexer but the \
2474 checkpoint has no `{name}`. This layer MUST NOT fall back to dense \
2475 attention: dense and indexed attention are the same function only below \
2476 index_topk ({}), and glm5_next serves a 1,048,576-token context",
2477 geom.top_k
2478 )
2479 .into());
2480 }
2481 load_t(e, src, &name).map_err(|source| -> Box<dyn std::error::Error> {
2482 format!("blk.{il}: DSA k-pool indexer tensor `{name}` failed to load: {source}")
2483 .into()
2484 })
2485 };
2486 let wq_b = need("indexer.attn_q_b.weight")?;
2487 let wk = need("indexer.attn_k.weight")?;
2488 let k_norm_w = need("indexer.k_norm.weight")?;
2489 let k_norm_b = need("indexer.k_norm.bias")?;
2490 let weights_proj = need("indexer.proj.weight")?;
2491 let kpool_gate = need("indexer.kpool_gate.weight")?;
2492 let kpool_ape = need("indexer.kpool_ape.weight")?;
2493 for (w, name) in [
2496 (&k_norm_w, "indexer.k_norm.weight"),
2497 (&k_norm_b, "indexer.k_norm.bias"),
2498 (&kpool_ape, "indexer.kpool_ape.weight"),
2499 ] {
2500 if !matches!(w, GpuTensor::Float { .. }) {
2501 return Err(format!(
2502 "blk.{il}.{name} is not f32-resident. The indexer's LayerNorm affine and \
2503 k-pool positional embedding feed f32-only kernels"
2504 )
2505 .into());
2506 }
2507 }
2508 assert_eq!(
2509 wq_b.in_features(),
2510 q_lora_rank as usize,
2511 "blk.{il}.indexer.attn_q_b in != q_lora_rank"
2512 );
2513 assert_eq!(
2514 wq_b.out_features(),
2515 geom.heads * geom.head_dim,
2516 "blk.{il}.indexer.attn_q_b out != index heads * head_dim"
2517 );
2518 assert_eq!(
2519 wk.out_features(),
2520 geom.head_dim,
2521 "blk.{il}.indexer.attn_k out != index head_dim"
2522 );
2523 assert_eq!(
2524 weights_proj.out_features(),
2525 geom.heads,
2526 "blk.{il}.indexer.proj out != index heads"
2527 );
2528 assert_eq!(
2529 kpool_gate.out_features(),
2530 geom.head_dim,
2531 "blk.{il}.indexer.kpool_gate out != index head_dim"
2532 );
2533 assert_eq!(
2534 kpool_ape.float_data().len(),
2535 geom.pool * geom.head_dim,
2536 "blk.{il}.indexer.kpool_ape must hold pool * head_dim elements"
2537 );
2538 Ok(Some(MlaIndexer {
2539 wq_b,
2540 wk,
2541 k_norm_w,
2542 k_norm_b,
2543 weights_proj,
2544 kpool_gate,
2545 kpool_ape,
2546 geom,
2547 }))
2548 }
2549}
2550
2551#[track_caller]
2559pub(crate) fn mla_path_unimplemented(path: &str) -> ! {
2560 panic!(
2561 "Mixer::Mla has no {path} arm — the MLA forward is wired for the stateless forward, \
2562 the stateful prime and T=1 decode only (cu/mla_attn.cu, increment 4); this path needs \
2563 its own parity gate before it may run \
2564 (research/mla-bringup-20260801/DESIGN.md §4, increment 7)"
2565 )
2566}
2567
2568#[track_caller]
2574pub(crate) fn kda_path_unimplemented(path: &str) -> ! {
2575 panic!(
2576 "Mixer::Kda has no {path} arm — glm5_next KDA is wired for the stateless forward, the \
2577 stateful prime and T=1 decode only (crates/memra-engine/src/kda.rs); this path needs \
2578 its own parity gate before it may run"
2579 )
2580}
2581
2582pub struct LinearAttnLayer {
2583 pub geometry: memra_gguf::model_plan::GatedDeltaNetPlan,
2584 pub wqkv: GpuTensor, pub wqkv_gate: GpuTensor, pub ssm_beta: GpuTensor, pub ssm_alpha: GpuTensor, pub ssm_a: GpuTensor, pub ssm_dt: GpuTensor, pub ssm_conv1d: GpuTensor, pub ssm_norm: GpuTensor, pub ssm_out: GpuTensor, }
2594
2595#[allow(clippy::large_enum_variant)] pub enum Mixer {
2597 Full(FullAttnLayer),
2598 Linear(LinearAttnLayer),
2599 Mla(MlaAttnLayer),
2601 Kda(crate::kda::KdaAttnLayer),
2603}
2604
2605pub struct MoeWeights {
2612 pub gate_inp: GpuTensor, pub gate_inp_shexp: Option<GpuTensor>, pub exp_probs_b: Option<Vec<f32>>,
2618 pub exp_probs_b_dev: CudaSlice<f32>,
2619 pub active_experts: Option<Vec<bool>>,
2623 pub active_experts_dev: CudaSlice<u8>,
2624 pub gate_exps: HostExps, pub up_exps: HostExps, pub down_exps: HostExps, pub gate_shexp: Option<GpuTensor>,
2628 pub up_shexp: Option<GpuTensor>,
2629 pub down_shexp: Option<GpuTensor>,
2630 pub dev_exps: Option<DevExps>,
2637 pub step_ep: Option<StepEpExps>,
2641 pub step_tp: Option<StepTpExps>,
2645 pub glm5_ep: Option<crate::glm5_tp::Glm5EpExps>,
2650 pub dev_macros: cudarc::driver::CudaSlice<f32>,
2656 pub has_macros: bool,
2657 pub w4a16_bf16_activations: bool,
2660}
2661
2662#[allow(clippy::large_enum_variant)] pub enum StepEpExpertBank {
2665 E4m3(crate::tp::ResidentExpertParallel),
2666 Nvfp4(crate::tp::ResidentNvfp4ExpertParallel),
2667}
2668
2669impl StepEpExpertBank {
2670 pub fn e4m3(&self) -> Result<&crate::tp::ResidentExpertParallel, String> {
2674 match self {
2675 Self::E4m3(bank) => Ok(bank),
2676 Self::Nvfp4(_) => Err(
2677 "Step grouped expert program reached an NVFP4 bank; this path is qualified \
2678 for the E4M3 artifact only"
2679 .to_string(),
2680 ),
2681 }
2682 }
2683}
2684
2685pub struct StepEpExps {
2686 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
2687 pub experts: StepEpExpertBank,
2688 pub devices: Vec<usize>,
2689 pub configured_by_tp: bool,
2690 pub activation_limit: Option<f32>,
2691 pub nvfp4_device_routes: bool,
2693 pub grouped_decode: Option<std::sync::Mutex<StepEpGroupedDecode>>,
2696}
2697
2698pub struct StepEpGroupedDecode {
2699 pub(crate) projection: crate::tp::PreparedStepGroupedExpertParallelGate,
2700 pub(crate) combine: crate::tp::PreparedPeerWeightedRouteCombine,
2701}
2702
2703#[derive(Default)]
2704pub(crate) struct StepEpGroupedPrefill {
2705 pub(crate) state: Option<StepEpGroupedPrefillState>,
2706}
2707
2708pub(crate) struct StepEpGroupedPrefillState {
2709 pub(crate) devices: Vec<usize>,
2710 pub(crate) grouped: StepEpGroupedDecode,
2711}
2712
2713#[allow(clippy::large_enum_variant)] pub enum StepTpExpertBank {
2716 E4m3(crate::tp::ResidentTensorParallel),
2717 Nvfp4(crate::tp::ResidentNvfp4TensorParallel),
2718}
2719
2720pub struct StepTpExps {
2721 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
2722 pub experts: StepTpExpertBank,
2723 pub devices: Vec<usize>,
2724 pub activation_limit: Option<f32>,
2727}
2728
2729impl MoeWeights {
2730 #[inline]
2731 pub fn has_uniform_expert_layout(&self) -> bool {
2732 self.gate_exps.is_uniform_layout()
2733 && self.up_exps.is_uniform_layout()
2734 && self.down_exps.is_uniform_layout()
2735 }
2736
2737 #[inline]
2738 pub fn active_count(&self) -> usize {
2739 self.active_experts
2740 .as_ref()
2741 .map(|mask| mask.iter().filter(|&&active| active).count())
2742 .unwrap_or(self.gate_exps.n_expert)
2743 }
2744
2745 #[allow(clippy::too_many_arguments)]
2746 pub(crate) fn qmatvec_view(
2747 &self,
2748 e: &Engine,
2749 w: &CudaSlice<u8>,
2750 range: std::ops::Range<usize>,
2751 x: &cudarc::driver::CudaView<f32>,
2752 m: usize,
2753 in_f: usize,
2754 out_f: usize,
2755 qtype: i32,
2756 row_bytes: usize,
2757 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
2758 if self.w4a16_bf16_activations && qtype == crate::QT_NVFP4 {
2759 e.qmatvec_view_bf16_activation(w, range, x, m, in_f, out_f, qtype, row_bytes)
2760 } else {
2761 e.qmatvec_view(w, range, x, m, in_f, out_f, qtype, row_bytes)
2762 }
2763 }
2764}
2765
2766pub struct DevExps {
2769 pub gate: CudaSlice<u8>,
2770 pub up: CudaSlice<u8>,
2771 pub down: CudaSlice<u8>,
2772 pub ptr_row: CudaSlice<u64>,
2774 pub dev: usize,
2782 pub gu_il: bool,
2788 pub fp8_blk: Option<DevExpertFp8BlockScales>,
2792}
2793
2794pub struct DevExpertFp8BlockScales {
2795 pub gate: DevExpertFp8ProjectionScales,
2796 pub up: DevExpertFp8ProjectionScales,
2797 pub down: DevExpertFp8ProjectionScales,
2798}
2799
2800pub struct DevExpertFp8ProjectionScales {
2801 pub scales: CudaSlice<f32>,
2802 pub rows: usize,
2803 pub cols: usize,
2804 pub expert_stride: usize,
2805}
2806
2807impl DevExpertFp8ProjectionScales {
2808 fn validate(
2809 host: &crate::model::HostExpertFp8BlockScales,
2810 n_expert: usize,
2811 ) -> Result<(), String> {
2812 if host.expert_stride == 0 {
2813 return Err("block-E4M3 expert scale stride must be nonzero".into());
2814 }
2815 if host.rows * host.cols != host.expert_stride {
2816 return Err(format!(
2817 "block-E4M3 expert scale stride mismatch: {}x{} != {}",
2818 host.rows, host.cols, host.expert_stride
2819 ));
2820 }
2821 let want = n_expert
2822 .checked_mul(host.expert_stride)
2823 .ok_or("block-E4M3 expert scale slab length overflow")?;
2824 if host.scales.len() != want {
2825 return Err(format!(
2826 "block-E4M3 scale slab length mismatch: got {}, want {n_expert}x{}={want}",
2827 host.scales.len(),
2828 host.expert_stride
2829 ));
2830 }
2831 Ok(())
2832 }
2833
2834 fn upload(
2835 e: &Engine,
2836 host: &crate::model::HostExpertFp8BlockScales,
2837 n_expert: usize,
2838 ) -> Result<Self, Box<dyn std::error::Error>> {
2839 Self::validate(host, n_expert)?;
2840 Ok(Self {
2841 scales: e.htod(&host.scales)?,
2842 rows: host.rows,
2843 cols: host.cols,
2844 expert_stride: host.expert_stride,
2845 })
2846 }
2847}
2848
2849#[allow(clippy::large_enum_variant)] pub enum Ffn {
2852 Dense {
2853 ffn_gate: GpuTensor,
2854 ffn_up: GpuTensor,
2855 ffn_down: GpuTensor,
2856 },
2857 Moe(MoeWeights),
2858}
2859
2860pub struct HybridLayer {
2861 pub attn_norm: GpuTensor,
2862 pub post_attn_norm: GpuTensor, pub mixer: Mixer,
2864 pub ffn: Ffn,
2865 pub gemma4: Option<Gemma4LayerBits>,
2866 pub hyper: Option<crate::hyper::HyperLayer>,
2871}
2872
2873pub struct Gemma4LayerBits {
2877 pub ffn_norm: GpuTensor, pub post_ffw_norm: GpuTensor, pub moe_bits: Option<Gemma4MoeBits>,
2882 pub layer_scale: f32, pub e4b: Option<Gemma4E4bLayer>,
2885}
2886
2887pub struct Gemma4E4bLayer {
2892 pub inp_gate: GpuTensor, pub proj: GpuTensor, pub post_norm: GpuTensor, pub qkv_cat: Option<GpuTensor>,
2899 pub kv_share: Option<u32>,
2903}
2904
2905pub struct Gemma4E4bModel {
2909 pub tok_tbl_gpu: std::sync::OnceLock<CudaSlice<u8>>,
2912 pub tok_embd_bytes: Vec<u8>,
2913 pub tok_embd_qt: i32,
2914 pub tok_embd_row_bytes: usize,
2915 pub model_proj: GpuTensor, pub proj_norm: GpuTensor, pub n_epl: usize,
2918}
2919
2920pub struct Gemma4MoeBits {
2921 pub post_ffw_norm_1: GpuTensor, pub pre_ffw_norm_2: GpuTensor, pub post_ffw_norm_2: GpuTensor, pub shared_gate: GpuTensor,
2925 pub shared_up: GpuTensor,
2926 pub shared_down: GpuTensor,
2927 pub router_scale_pre: CudaSlice<f32>,
2932 pub per_expert_scale: Vec<f32>, pub per_expert_scale_d: CudaSlice<f32>, }
2935
2936fn load_mtp_head_maybe_nvfp4(
2949 e: &Engine,
2950 src: &dyn TensorSource,
2951 name: &str,
2952) -> Result<Option<GpuTensor>, Box<dyn std::error::Error>> {
2953 if !{
2954 static ENV: std::sync::OnceLock<Option<bool>> = std::sync::OnceLock::new();
2955 crate::step37_door(&ENV, "MEMRA_MTP_HEAD_NVFP4")
2956 } {
2957 return load_opt(e, src, name);
2958 }
2959 let Some(v) = src.find(name) else {
2960 return Ok(None);
2961 };
2962 if !matches!(v.ggml_type, GgmlType::BF16) || v.ne[0] % 64 != 0 {
2963 return load_opt(e, src, name);
2964 }
2965 let vals: Vec<f32> = v
2966 .bytes
2967 .chunks_exact(2)
2968 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
2969 .collect();
2970 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
2971 eprintln!(
2972 "[mtp-head] {name}: BF16 -> NVFP4 ({} MiB, was {} MiB)",
2973 blocks.len() >> 20,
2974 v.bytes.len() >> 20
2975 );
2976 Ok(Some(GpuTensor::from_quant_bytes(
2977 e,
2978 &blocks,
2979 GgmlType::NVFP4,
2980 v.ne[0],
2981 v.ne[1],
2982 1.0,
2983 )?))
2984}
2985
2986pub(crate) fn sha256_file_hex8(
2994 path: &std::path::Path,
2995) -> Result<String, Box<dyn std::error::Error>> {
2996 use sha2::{Digest, Sha256};
2997 let mut file = std::fs::File::open(path)?;
2998 let mut hasher = Sha256::new();
2999 std::io::copy(&mut file, &mut hasher)?;
3000 let digest = hasher.finalize();
3001 Ok(digest
3002 .iter()
3003 .take(4)
3004 .map(|byte| format!("{byte:02x}"))
3005 .collect())
3006}
3007
3008pub(crate) fn frspec_trim_own_head_name(n_trunk: usize) -> String {
3009 format!("blk.{n_trunk}.nextn.shared_head_head.weight")
3010}
3011
3012pub struct DflashTrimHead {
3023 pub head: GpuTensor,
3026 pub d2t: Vec<u32>,
3028}
3029
3030fn frspec_read_d2t(path: &str) -> Result<Vec<u32>, Box<dyn std::error::Error>> {
3035 Ok(if path.ends_with(".txt") {
3036 std::fs::read_to_string(path)?
3037 .lines()
3038 .filter_map(|l| l.trim().parse::<u32>().ok())
3039 .collect()
3040 } else {
3041 let tg = GgufFile::open(path)?;
3042 let d2t_t = tg
3043 .find("d2t")
3044 .expect("MEMRA_FRSPEC_TRIM file has no d2t tensor");
3045 let d2t_bytes = tg.tensor_data(d2t_t);
3046 match d2t_t.ggml_type {
3047 GgmlType::I32 => d2t_bytes
3048 .chunks_exact(4)
3049 .map(|c| i32::from_le_bytes(c.try_into().unwrap()) as u32)
3050 .collect(),
3051 GgmlType::I64 => d2t_bytes
3052 .chunks_exact(8)
3053 .map(|c| i64::from_le_bytes(c.try_into().unwrap()) as u32)
3054 .collect(),
3055 other => panic!("d2t must be I32/I64, got {other:?}"),
3056 }
3057 })
3058}
3059
3060#[allow(clippy::type_complexity)] fn frspec_gather_trimmed_head(
3069 e: &Engine,
3070 v: &memra_gguf::source::TensorView<'_>,
3071 d2t: &[u32],
3072 want_nvfp4_env: bool,
3073 macro_scale: f32,
3074) -> Result<(GpuTensor, Option<(usize, usize)>), Box<dyn std::error::Error>> {
3075 let out_f = v.ne[1] as usize;
3076 let row_bytes = v.bytes.len() / out_f;
3077 assert!(
3078 d2t.iter().all(|&t| (t as usize) < out_f),
3079 "d2t token id >= lm_head rows {out_f}"
3080 );
3081 let mut gathered = Vec::with_capacity(d2t.len() * row_bytes);
3082 for &t in d2t {
3083 let off = t as usize * row_bytes;
3084 gathered.extend_from_slice(&v.bytes[off..off + row_bytes]);
3085 }
3086 let want_nvfp4 =
3087 want_nvfp4_env && matches!(v.ggml_type, GgmlType::BF16) && v.ne[0].is_multiple_of(64);
3088 if want_nvfp4 {
3089 let in_f = v.ne[0] as usize;
3090 let vals: Vec<f32> = gathered
3091 .chunks_exact(2)
3092 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
3093 .collect();
3094 debug_assert_eq!(vals.len(), d2t.len() * in_f);
3095 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
3096 let sizes = (blocks.len(), gathered.len());
3097 let trimmed = GpuTensor::from_quant_bytes(
3098 e,
3099 &blocks,
3100 GgmlType::NVFP4,
3101 v.ne[0],
3102 d2t.len() as u64,
3103 1.0,
3104 )?;
3105 Ok((trimmed, Some(sizes)))
3106 } else {
3107 let trimmed = match v.ggml_type {
3108 GgmlType::BF16 => GpuTensor::FloatBf16 {
3109 data: e.htod_bytes(&gathered)?,
3110 ne: vec![v.ne[0], d2t.len() as u64],
3111 },
3112 GgmlType::F32 => GpuTensor::Float {
3113 data: e.htod(
3114 &gathered
3115 .chunks_exact(4)
3116 .map(|c| f32::from_le_bytes(c.try_into().unwrap()))
3117 .collect::<Vec<f32>>(),
3118 )?,
3119 ne: vec![v.ne[0], d2t.len() as u64],
3120 },
3121 _ => GpuTensor::from_quant_bytes(
3122 e,
3123 &gathered,
3124 v.ggml_type,
3125 v.ne[0],
3126 d2t.len() as u64,
3127 macro_scale,
3128 )?,
3129 };
3130 Ok((trimmed, None))
3131 }
3132}
3133
3134pub struct MtpHead {
3135 pub enorm: GpuTensor, pub hnorm: GpuTensor, pub eh_proj: GpuTensor, pub attn_norm: GpuTensor, pub post_attn_norm: GpuTensor, pub mixer: Mixer, pub ffn: Ffn, pub shared_head_norm: Option<GpuTensor>, pub shared_head_head: Option<GpuTensor>, pub d2t: Option<Vec<u32>>,
3149 pub d2t_from_target_head: bool,
3153 pub geom: Option<DraftGeom>,
3159 pub step35: Option<Step35MtpGeom>,
3164}
3165
3166#[derive(Debug, Clone)]
3180pub struct Step35MtpGeom {
3181 pub il: u32,
3183 pub n_head: usize, pub n_head_kv: usize, pub n_rot: usize, pub rope_base: f32, pub swa: bool, pub window: usize, pub clamp_shexp: Option<f32>,
3194}
3195
3196impl Step35MtpGeom {
3197 pub fn from_plan(layer: &memra_gguf::model_plan::LayerPlan) -> Result<Self, String> {
3199 use memra_gguf::model_plan::{ActivationPlan, AttentionPlan};
3200
3201 let (attention, window) = match &layer.attention {
3202 AttentionPlan::Full(attention) => (attention, None),
3203 AttentionPlan::SlidingWindow { attention, window } => (attention, Some(*window)),
3204 other => {
3205 return Err(format!(
3206 "MTP block {} has unsupported tuned attention {other:?}",
3207 layer.index
3208 ));
3209 }
3210 };
3211 if attention.output_gate != memra_gguf::config::AttentionGateKind::SeparateHead {
3212 return Err(format!(
3213 "MTP block {} does not declare a separate attention gate",
3214 layer.index
3215 ));
3216 }
3217 let activation = match &layer.mlp {
3218 MlpPlan::Dense(dense) => &dense.activation,
3219 MlpPlan::Moe(moe) => &moe.activation,
3220 };
3221 let clamp_shexp = match activation {
3222 ActivationPlan::SwiGluClamped { limit } if *limit > 0.0 => Some(*limit),
3223 _ => None,
3224 };
3225 Ok(Step35MtpGeom {
3226 il: layer.index,
3227 n_head: attention.query_heads as usize,
3228 n_head_kv: attention.kv_heads as usize,
3229 n_rot: attention.rope.dimensions as usize,
3230 rope_base: attention.rope.base,
3231 swa: window.is_some(),
3232 window: window.unwrap_or(0) as usize,
3233 clamp_shexp,
3234 })
3235 }
3236}
3237
3238pub struct DraftGeom {
3240 pub d_inner: usize, pub n_head: usize, pub n_head_kv: usize,
3243 pub out_up: GpuTensor, }
3245
3246pub fn draft_head_tensor(has: impl Fn(&str) -> bool, n: u32) -> String {
3255 let own = format!("blk.{n}.nextn.shared_head_head.weight");
3256 if has(&own) {
3257 return own;
3258 }
3259 let legacy = format!("blk.{n}.nextn.shared_head.weight");
3262 if has(&legacy) {
3263 return legacy;
3264 }
3265 "output.weight".to_string()
3267}
3268
3269impl MtpHead {
3270 pub fn load_draft(
3277 e: &Engine,
3278 g: &GgufFile,
3279 main_cfg: &ModelConfig,
3280 ) -> Result<Self, Box<dyn std::error::Error>> {
3281 let src = GgufSource(g);
3282 let dcfg = src.try_config().map_err(std::io::Error::other)?;
3283 let draft_plan = match memra_gguf::model_packs::for_config(&dcfg) {
3284 Some(pack) => pack.compile_plan(&dcfg)?,
3285 None => memra_gguf::model_plan::ModelPlan::compile(&dcfg)?,
3286 };
3287 let main_plan = match memra_gguf::model_packs::for_config(main_cfg) {
3288 Some(pack) => pack.compile_plan(main_cfg)?,
3289 None => memra_gguf::model_plan::ModelPlan::compile(main_cfg)?,
3290 };
3291 if dcfg.nextn_predict_layers == 0 {
3296 return Err(format!(
3297 "draft GGUF has no nextn_predict_layers (arch {:?}) — not a NextN/MTP regime \
3298 draft; gemma assistant drafters attach via MEMRA_DRAFT, not '+draft'",
3299 g.arch()
3300 )
3301 .into());
3302 }
3303 let n = dcfg.n_layer - dcfg.nextn_predict_layers;
3304 let draft_block = draft_plan
3305 .mtp_blocks
3306 .iter()
3307 .find(|block| block.layer.index == n)
3308 .ok_or_else(|| format!("draft ModelPlan has no MTP block {n}"))?;
3309 let p = |s: &str| format!("blk.{n}.{s}");
3310
3311 let student = src.has(&p("nextn.out_up.weight"));
3315 assert_eq!(dcfg.n_embd, main_cfg.n_embd, "draft n_embd != model n_embd");
3316 assert_eq!(
3317 dcfg.head_dim_k, main_cfg.head_dim_k,
3318 "draft head_dim != model head_dim"
3319 );
3320 let main_sliding_gated = crate::plan_backend::decode_batch_program(&main_plan)
3327 == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
3328 let draft_sliding_gated = crate::plan_backend::decode_batch_program(&draft_plan)
3329 == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
3330 let step35 = match (main_sliding_gated, draft_sliding_gated) {
3331 (true, true) => {
3332 let g = Step35MtpGeom::from_plan(&draft_block.layer)?;
3333 let out_f = |t: &str| -> Option<usize> {
3335 src.find(&p(t))
3336 .and_then(|v| v.ne.get(1).copied())
3337 .map(|x| x as usize)
3338 };
3339 let hd = dcfg.head_dim_k as usize;
3340 let wq_out =
3341 out_f("attn_q.weight").ok_or("step35 draft block has no attn_q.weight")?;
3342 assert_eq!(
3343 wq_out,
3344 g.n_head * hd,
3345 "step35 draft blk.{n}: attn_q out {wq_out} != n_head({}) * head_dim({hd}) — \
3346 the draft file's head_count array disagrees with its own tensors",
3347 g.n_head
3348 );
3349 let wg_out = out_f("attn_gate.weight")
3352 .ok_or("step35 draft block has no attn_gate.weight (head-wise gate)")?;
3353 assert_eq!(
3354 wg_out, g.n_head,
3355 "step35 draft blk.{n}: attn_gate out {wg_out} != n_head({})",
3356 g.n_head
3357 );
3358 assert_eq!(
3362 g.n_head_kv, main_cfg.n_head_kv as usize,
3363 "step35 draft blk.{n} KV heads {} != trunk n_head_kv {} — the MTP scratch \
3364 rows are sized from the trunk cfg, so a differing draft KV width would \
3365 write past the row",
3366 g.n_head_kv, main_cfg.n_head_kv
3367 );
3368 eprintln!(
3369 "[mtp-draft] step35 MTP geometry blk.{n}: n_head={} n_head_kv={} n_rot={} \
3370 rope_base={:.0} swa={} window={}",
3371 g.n_head, g.n_head_kv, g.n_rot, g.rope_base, g.swa, g.window
3372 );
3373 Some(g)
3374 }
3375 (true, false) => {
3376 return Err(format!(
3377 "MEMRA_MTP_DRAFT operations are incompatible with the model's \
3378 sliding-gated-MoE program (draft arch {:?})",
3379 g.arch()
3380 )
3381 .into());
3382 }
3383 (false, true) => {
3384 return Err(
3385 "MEMRA_MTP_DRAFT requires sliding-gated-MoE operations but the model does not"
3386 .into(),
3387 );
3388 }
3389 (false, false) => None,
3390 };
3391 if step35.is_none() && !student {
3392 assert_eq!(dcfg.n_head, main_cfg.n_head, "draft n_head != model n_head");
3395 assert_eq!(
3396 dcfg.n_head_kv, main_cfg.n_head_kv,
3397 "draft n_head_kv != model n_head_kv"
3398 );
3399 }
3400
3401 let head_name = draft_head_tensor(|t| src.has(t), n);
3428 let head = load_t(e, &src, &head_name)?;
3429 let head_norm = match load_opt(e, &src, &p("nextn.shared_head_norm.weight"))? {
3430 Some(t) => Some(t),
3431 None => load_opt(e, &src, "output_norm.weight")?,
3432 };
3433
3434 let d2t: Option<Vec<u32>> = g.find("d2t").map(|t| {
3436 let bytes = g.tensor_data(t);
3437 match t.ggml_type {
3438 GgmlType::I32 => bytes
3439 .chunks_exact(4)
3440 .map(|c| i32::from_le_bytes(c.try_into().unwrap()) as u32)
3441 .collect(),
3442 GgmlType::I64 => bytes
3443 .chunks_exact(8)
3444 .map(|c| i64::from_le_bytes(c.try_into().unwrap()) as u32)
3445 .collect(),
3446 other => panic!("d2t must be I32/I64, got {other:?}"),
3447 }
3448 });
3449 if let Some(map) = &d2t {
3450 assert_eq!(
3451 map.len(),
3452 head.out_features(),
3453 "d2t len {} != draft head rows {}",
3454 map.len(),
3455 head.out_features()
3456 );
3457 let n_vocab = main_cfg.n_vocab as u64;
3458 assert!(
3459 map.iter().all(|&t| (t as u64) < n_vocab),
3460 "d2t contains token id >= model n_vocab {n_vocab}"
3461 );
3462 }
3463 let eh_proj = load_t(e, &src, &p("nextn.eh_proj.weight"))?;
3464 assert_eq!(
3467 eh_proj.in_features(),
3468 2 * main_cfg.n_embd as usize,
3469 "eh_proj in dim != 2*n_embd"
3470 );
3471 let geom = if student {
3472 let out_up = load_t(e, &src, &p("nextn.out_up.weight"))?;
3473 let d_inner = eh_proj.out_features();
3474 assert_eq!(
3475 out_up.out_features(),
3476 main_cfg.n_embd as usize,
3477 "out_up out dim != n_embd"
3478 );
3479 assert_eq!(
3480 out_up.in_features(),
3481 d_inner,
3482 "out_up in dim != eh_proj out dim (d_inner)"
3483 );
3484 assert!(
3485 dcfg.n_head >= 1 && dcfg.n_head_kv >= 1 && dcfg.n_head % dcfg.n_head_kv == 0,
3486 "student head counts malformed ({}/{})",
3487 dcfg.n_head,
3488 dcfg.n_head_kv
3489 );
3490 Some(DraftGeom {
3491 d_inner,
3492 n_head: dcfg.n_head as usize,
3493 n_head_kv: dcfg.n_head_kv as usize,
3494 out_up,
3495 })
3496 } else {
3497 None
3498 };
3499 let blk_prefix = format!("blk.{n}.");
3503 let head_src = head_name.strip_prefix(&blk_prefix).unwrap_or(&head_name);
3504 eprintln!(
3505 "[mtp-draft] external draft head: blk.{n}, source={}, head_vocab={}{}{}",
3506 head_src,
3507 head.out_features(),
3508 if d2t.is_some() {
3509 " (trimmed, d2t map)"
3510 } else {
3511 " (full)"
3512 },
3513 match &geom {
3514 Some(g) => format!(
3515 " (student d_inner={} heads={}/{})",
3516 g.d_inner, g.n_head, g.n_head_kv
3517 ),
3518 None => String::new(),
3519 }
3520 );
3521
3522 let mut resident = ResidentPlan::unsharded(e, &src, &dcfg);
3523 let mut step_runtimes = StepParallelRuntimeRegistry::default();
3524 Ok(MtpHead {
3525 enorm: load_t(e, &src, &p("nextn.enorm.weight"))?,
3526 hnorm: load_t(e, &src, &p("nextn.hnorm.weight"))?,
3527 eh_proj,
3528 attn_norm: load_t(e, &src, &p("attn_norm.weight"))?,
3529 post_attn_norm: load_opt(e, &src, &p("post_attention_norm.weight"))?
3530 .or(load_opt(e, &src, &p("ffn_norm.weight"))?)
3531 .expect("draft NextN block needs post_attention_norm or ffn_norm"),
3532 mixer: load_mixer_kind(
3533 e,
3534 &src,
3535 &dcfg,
3536 n,
3537 &draft_block.layer.attention,
3538 &mut step_runtimes,
3539 )?,
3540 ffn: load_ffn(
3541 e,
3542 &src,
3543 &dcfg,
3544 &draft_block.layer.mlp,
3545 n,
3546 None,
3547 &mut resident,
3548 &mut step_runtimes,
3549 )?,
3550 shared_head_norm: head_norm,
3551 shared_head_head: Some(head),
3552 d2t,
3553 d2t_from_target_head: false,
3554 geom,
3555 step35,
3556 })
3557 }
3558}
3559
3560pub struct GemmaAux {
3562 pub rope_freqs: Option<Vec<(usize, CudaSlice<f32>)>>,
3565 pub ones: Vec<(usize, CudaSlice<f32>)>,
3568 pub suppress_d: Option<(CudaSlice<i32>, usize)>,
3571 pub e4b: Option<Gemma4E4bModel>,
3573}
3574
3575impl GemmaAux {
3576 pub fn rope_freqs(&self, e: &Engine) -> Option<&CudaSlice<f32>> {
3577 self.rope_freqs.as_ref().map(|copies| {
3578 let dev = e.ctx().ordinal();
3579 &copies
3580 .iter()
3581 .find(|(d, _)| *d == dev)
3582 .unwrap_or_else(|| panic!("gemma4 rope_freqs has no local copy for device {dev}"))
3583 .1
3584 })
3585 }
3586
3587 pub fn ones(&self, e: &Engine) -> &CudaSlice<f32> {
3588 let dev = e.ctx().ordinal();
3589 &self
3590 .ones
3591 .iter()
3592 .find(|(d, _)| *d == dev)
3593 .unwrap_or_else(|| panic!("gemma4 ones has no local copy for device {dev}"))
3594 .1
3595 }
3596}
3597
3598pub struct Step35Aux {
3601 pub rope_freqs: Option<Vec<(usize, CudaSlice<f32>)>>,
3607}
3608
3609impl Step35Aux {
3610 pub fn rope_freqs(&self, e: &Engine) -> Option<&CudaSlice<f32>> {
3611 self.rope_freqs.as_ref().map(|copies| {
3612 let dev = e.ctx().ordinal();
3613 &copies
3614 .iter()
3615 .find(|(d, _)| *d == dev)
3616 .unwrap_or_else(|| panic!("step35 rope_freqs has no local copy for device {dev}"))
3617 .1
3618 })
3619 }
3620}
3621
3622pub struct HybridModel {
3623 pub cfg: ModelConfig,
3624 pub plan: memra_gguf::model_plan::ModelPlan,
3625 pub rewrite_qualifications: Option<memra_gguf::execution_manifest::RewriteQualifications>,
3626 pub embd: EmbedHost,
3627 pub output_norm: GpuTensor,
3628 pub output: GpuTensor,
3629 pub layers: Vec<HybridLayer>,
3630 pub mtp: Option<MtpHead>, pub mtp_extra: Vec<MtpHead>,
3634 pub dflash_trim: Option<DflashTrimHead>,
3638 pub embd_gpu: std::sync::OnceLock<cudarc::driver::CudaSlice<u8>>,
3641 pub gemma4_aux: Option<GemmaAux>,
3642 pub step35_aux: Option<Step35Aux>,
3644 pub prime_slabs: std::sync::Mutex<
3652 std::collections::HashMap<
3653 usize,
3654 std::sync::Arc<std::sync::Mutex<crate::hybrid_forward::PrimeSlabs>>,
3655 >,
3656 >,
3657 pub(crate) dspark_vgraphs: std::sync::Mutex<Option<crate::spec::DsparkVerifyGraphs>>,
3670 pub(crate) step_grouped_prefill: std::sync::Mutex<StepEpGroupedPrefill>,
3676 pub(crate) step35_token_graph:
3679 std::sync::Mutex<Option<crate::hybrid_forward::Step35TokenGraphState>>,
3680 pub hyper: Option<crate::hyper::HyperTopology>,
3685 pub hyper_head: Option<crate::hyper::HyperHead>,
3688 pub glm5_dflash: Option<crate::glm_spec::Glm5DflashDrafter>,
3695 pub(crate) draft_state_bytes: std::sync::atomic::AtomicUsize,
3704}
3705
3706impl HybridModel {
3707 pub fn install_rewrite_bundle(
3708 &mut self,
3709 bundle: &std::path::Path,
3710 ) -> Result<(), Box<dyn std::error::Error>> {
3711 self.rewrite_qualifications = Some(
3712 memra_gguf::execution_manifest::RewriteQualifications::load(bundle, &self.plan)
3713 .map_err(|error| format!("rewrite qualification: {error}"))?,
3714 );
3715 Ok(())
3716 }
3717
3718 pub fn rewrite_allowed(&self, surface: memra_gguf::execution_manifest::RewriteSurface) -> bool {
3719 self.rewrite_qualifications
3720 .as_ref()
3721 .is_none_or(|qualifications| qualifications.allows(surface))
3722 }
3723
3724 pub fn record_draft_state_bytes(&self, observed: usize) -> Option<usize> {
3729 use std::sync::atomic::Ordering;
3730 let prev = self
3731 .draft_state_bytes
3732 .fetch_max(observed, Ordering::Relaxed);
3733 (observed > prev).then_some(observed)
3734 }
3735
3736 pub fn draft_session_admission_bytes(&self) -> usize {
3742 self.draft_state_bytes
3743 .load(std::sync::atomic::Ordering::Relaxed)
3744 }
3745
3746 pub fn step_tp_unmaterialized_kv_bytes(
3752 &self,
3753 cache: Option<&crate::cache::Cache>,
3754 capacity: usize,
3755 ) -> Result<Vec<StepTpKvDeviceAdmission>, String> {
3756 if let Some(cache) = cache
3757 && cache.tp_kv.len() < self.layers.len()
3758 {
3759 return Err(format!(
3760 "Step TP admission cache has {} layers, model trunk has {}",
3761 cache.tp_kv.len(),
3762 self.layers.len()
3763 ));
3764 }
3765
3766 let mut by_device: HashMap<usize, usize> = HashMap::new();
3767 for (layer, weights) in self.layers.iter().enumerate() {
3768 let Mixer::Full(attention) = &weights.mixer else {
3769 continue;
3770 };
3771 let Some(tp) = attention
3772 .step_tp_qkv
3773 .as_ref()
3774 .filter(|tp| tp.attention.is_some())
3775 else {
3776 continue;
3777 };
3778 if cache.is_some_and(|cache| cache.tp_kv[layer].is_some()) {
3779 continue;
3780 }
3781 let geometry = self.cfg.full_attention_geometry_at(layer as u32);
3782 let shape = crate::cache::tp_kv_rank_allocation_shape(
3783 geometry.n_head_kv as usize * geometry.head_dim_k as usize,
3784 geometry.n_head_kv as usize * geometry.head_dim_v as usize,
3785 tp.devices.len(),
3786 )?;
3787 let physical_rows = geometry
3788 .window
3789 .map(|window| crate::cache::swa_ring_rows(window as usize, capacity))
3790 .unwrap_or(capacity);
3791 let bytes = shape.allocation_bytes(physical_rows);
3792 for &device in &tp.devices {
3793 let total = by_device.entry(device).or_default();
3794 *total = total.saturating_add(bytes);
3795 }
3796 }
3797
3798 let mut out: Vec<_> = by_device
3799 .into_iter()
3800 .map(|(device, bytes)| StepTpKvDeviceAdmission { device, bytes })
3801 .collect();
3802 out.sort_unstable_by_key(|charge| charge.device);
3803 Ok(out)
3804 }
3805
3806 pub fn step_tp_rank_engine(&self, device: usize) -> Option<&Engine> {
3808 self.layers.iter().find_map(|weights| {
3809 let Mixer::Full(attention) = &weights.mixer else {
3810 return None;
3811 };
3812 let tp = attention.step_tp_qkv.as_ref()?;
3813 let rank = tp
3814 .runtime
3815 .devices()
3816 .iter()
3817 .position(|&rank| rank == device)?;
3818 tp.runtime.rank_engine(rank)
3819 })
3820 }
3821
3822 pub(crate) fn step_tp_runtime_for_layer(
3823 &self,
3824 layer: usize,
3825 ) -> Option<&crate::tp::TpE4m3HostBounce> {
3826 let Mixer::Full(attention) = &self.layers.get(layer)?.mixer else {
3827 return None;
3828 };
3829 let tp = attention.step_tp_qkv.as_ref()?;
3830 tp.attention.as_ref()?;
3831 Some(tp.runtime.as_ref())
3832 }
3833
3834 pub fn decode_batch_program(&self) -> crate::plan_backend::DecodeBatchProgram {
3835 crate::plan_backend::decode_batch_program(&self.plan)
3836 }
3837
3838 pub fn uses_gemma_program(&self) -> bool {
3839 self.decode_batch_program() == crate::plan_backend::DecodeBatchProgram::Gemma
3840 }
3841
3842 pub fn uses_sliding_gated_moe_program(&self) -> bool {
3843 self.decode_batch_program() == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe
3844 }
3845
3846 pub fn has_plan_operation(&self, operation: memra_gguf::model_plan::OperationKind) -> bool {
3847 self.plan.trunk_operations().contains(&operation)
3848 }
3849
3850 pub fn load(e: &Engine, g: &GgufFile) -> Result<Self, Box<dyn std::error::Error>> {
3852 Self::load_from_source(e, &GgufSource(g))
3853 }
3854
3855 pub fn load_without_mtp(e: &Engine, g: &GgufFile) -> Result<Self, Box<dyn std::error::Error>> {
3858 Self::load_from_source_impl(e, &GgufSource(g), false)
3859 }
3860
3861 pub fn load_from_source(
3865 e: &Engine,
3866 src: &dyn TensorSource,
3867 ) -> Result<Self, Box<dyn std::error::Error>> {
3868 Self::load_from_source_impl(e, src, true)
3869 }
3870
3871 pub fn load_from_source_without_mtp(
3873 e: &Engine,
3874 src: &dyn TensorSource,
3875 ) -> Result<Self, Box<dyn std::error::Error>> {
3876 Self::load_from_source_impl(e, src, false)
3877 }
3878
3879 fn load_from_source_impl(
3880 e: &Engine,
3881 src: &dyn TensorSource,
3882 load_mtp: bool,
3883 ) -> Result<Self, Box<dyn std::error::Error>> {
3884 let cfg = src.try_config().map_err(std::io::Error::other)?;
3885 let plan = match memra_gguf::model_packs::for_config(&cfg) {
3886 Some(pack) => pack.compile_plan(&cfg)?,
3887 None => memra_gguf::model_plan::ModelPlan::compile(&cfg)?,
3888 };
3889 let auto_parallel = prepare_auto_parallel(src, &cfg, &plan)?;
3890 let batch_program = crate::plan_backend::decode_batch_program(&plan);
3891 let gemma_program = batch_program == crate::plan_backend::DecodeBatchProgram::Gemma;
3892 let sliding_gated_moe_program =
3893 batch_program == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
3894 if matches!(
3895 src.expert_activation_precision(),
3896 memra_gguf::source::ExpertActivationPrecision::Bf16
3897 ) {
3898 eprintln!(
3899 "[w4a16] artifact contract accepted: expert_weights=nvfp4 \
3900 expert_activations=bf16-rounded q8_expert_program=disabled"
3901 );
3902 }
3903 if sliding_gated_moe_program {
3908 crate::arm_step37_serving_defaults();
3909 }
3910 cfg.validate_attention_gate_layout()?;
3915 if cfg.sigmoid_router().is_some() {
3922 let host_oracle = std::env::var("MEMRA_SIG_ROUTER").as_deref() == Ok("0");
3923 match crate::sigrouter_contract::verify_host_expf() {
3924 Ok(()) => {}
3925 Err(e) if host_oracle => return Err(e.into()),
3926 Err(e) => eprintln!(
3927 "[sigrouter] WARN: host expf probe mismatch ({e}); device routing is \
3928 unaffected, but host-oracle replay/comparison cells are invalid on this host"
3929 ),
3930 }
3931 }
3932 if std::env::var("MEMRA_DRAFT").is_ok() && std::env::var("MEMRA_MMQ_SK").is_err() {
3941 let force = if cfg.n_embd >= 3500 { 0i8 } else { -1i8 };
3942 crate::MMQ_SK_FORCE.store(force, std::sync::atomic::Ordering::Relaxed);
3943 }
3944 crate::KV_FP8_FORCE.store(0, std::sync::atomic::Ordering::Relaxed);
3948
3949 let n_trunk = (cfg.n_layer - cfg.nextn_predict_layers) as usize;
3954 let mtp_skip_requested = load_mtp
3964 && match std::env::var("MEMRA_MTP_SKIP").ok().as_deref() {
3965 None | Some("") | Some("0") => false,
3966 Some("1") => true,
3967 Some(other) => {
3968 return Err(format!(
3969 "MEMRA_MTP_SKIP={other:?}: expected 1 (skip the embedded MTP block) or \
3970 0/unset (load it); refusing to guess"
3971 )
3972 .into());
3973 }
3974 };
3975 if mtp_skip_requested && std::env::var("MEMRA_MTP_DRAFT").is_ok_and(|p| !p.is_empty()) {
3976 return Err(
3977 "MEMRA_MTP_SKIP=1 together with MEMRA_MTP_DRAFT is contradictory: the skip \
3978 removes the MTP head to reclaim VRAM while MEMRA_MTP_DRAFT attaches an \
3979 external MTP head for MTP spec decode; unset one"
3980 .into(),
3981 );
3982 }
3983 if mtp_skip_requested && cfg.nextn_predict_layers > 0 {
3984 let prefixes: Vec<String> = (0..cfg.nextn_predict_layers)
3989 .map(|off| format!("blk.{}.", n_trunk as u32 + off))
3990 .collect();
3991 let skipped_bytes: Option<u64> = src.gguf().map(|g| {
3992 g.tensors
3993 .iter()
3994 .filter(|t| prefixes.iter().any(|p| t.name.starts_with(p.as_str())))
3995 .map(|t| t.n_bytes)
3996 .sum()
3997 });
3998 eprintln!(
3999 "[mtp-skip] MEMRA_MTP_SKIP=1: skipping {} embedded MTP/NextN block(s) \
4000 blk.{}..=blk.{} ({}); MTP spec decode is unavailable for this model \
4001 (dspark/DFlash2 drafting keeps its trimmed head via the MEMRA_FRSPEC_TRIM stub)",
4002 cfg.nextn_predict_layers,
4003 n_trunk,
4004 n_trunk as u32 + cfg.nextn_predict_layers - 1,
4005 match skipped_bytes {
4006 Some(b) => format!("~{} MiB of weights not loaded", b >> 20),
4007 None => "size unknown: non-GGUF source".to_string(),
4008 },
4009 );
4010 }
4011 let mtp_skip_trim_d2t: Option<Vec<u32>> = if mtp_skip_requested
4026 && cfg.nextn_predict_layers > 0
4027 && !crate::model::full_prec_enabled()
4028 {
4029 match std::env::var("MEMRA_FRSPEC_TRIM") {
4030 Ok(path) if !path.is_empty() => {
4031 let path = memra_gguf::hf::resolve_arg(&path)
4032 .map_err(|err| format!("MEMRA_FRSPEC_TRIM={path:?}: {err}"))?;
4033 let own_head_name = frspec_trim_own_head_name(n_trunk);
4034 if src.has(&own_head_name) {
4035 return Err(format!(
4036 "MEMRA_MTP_SKIP=1 with MEMRA_FRSPEC_TRIM: this artifact ships its \
4037 own MTP-block lm_head ({own_head_name}), so the trimmed draft rows \
4038 live in the block being skipped; gathering trunk rows instead is \
4039 the wrong-head bug (acceptance 0/248 receipt, \
4040 frspec_trim_own_head_name). Unset MEMRA_MTP_SKIP or \
4041 MEMRA_FRSPEC_TRIM"
4042 )
4043 .into());
4044 }
4045 if !src.has("output.weight") && !src.has("token_embd.weight") {
4046 return Err("MEMRA_MTP_SKIP=1 with MEMRA_FRSPEC_TRIM: model has no \
4047 output.weight (or tied token_embd.weight) to gather trimmed draft \
4048 rows from"
4049 .into());
4050 }
4051 let d2t = frspec_read_d2t(&path)?;
4052 if d2t.is_empty() {
4053 return Err(format!(
4054 "MEMRA_MTP_SKIP=1 with MEMRA_FRSPEC_TRIM={path}: the rank artifact \
4055 yields an EMPTY d2t list, so no stub draft head can be built; fix \
4056 the artifact or unset MEMRA_MTP_SKIP"
4057 )
4058 .into());
4059 }
4060 Some(d2t)
4061 }
4062 _ => None,
4063 }
4064 } else {
4065 None
4066 };
4067 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
4068 let pipeline = crate::plan_backend::PIPELINE
4069 .trunk_capabilities(&plan)
4070 .pipeline;
4071 let qualified_gemma_pp2 = gemma_program && fence.len() == 3;
4075 if !pipeline.supported && !qualified_gemma_pp2 {
4076 return Err(format!(
4077 "pipeline placement is unsupported for plan operations {:?}; blockers={:?}",
4078 plan.trunk_operations(),
4079 pipeline.blockers,
4080 )
4081 .into());
4082 }
4083 let illegal = illegal_pipeline_cuts(&fence, &plan.partition_boundaries);
4084 if !illegal.is_empty() {
4085 return Err(format!(
4086 "pipeline placement cuts {illegal:?} split outside ModelPlan legal boundaries {:?}",
4087 plan.partition_boundaries,
4088 )
4089 .into());
4090 }
4091 }
4092 crate::pp::init_model_transport(e, &cfg, n_trunk)?;
4093 let step_parallel =
4094 prepare_step_parallel_load(e, src, &cfg, n_trunk, auto_parallel.as_ref())?;
4095 let glm5_tp = if crate::glm5_tp::glm5_tp_armed() {
4099 use memra_gguf::model_plan::{AttentionPlan, MlpPlan};
4100 let moe = cfg.moe.as_ref().ok_or(
4101 "MEMRA_GLM5_TP requires a MoE model (glm5_next); this plan carries no MoE \
4102 metadata",
4103 )?;
4104 let mut layer_class = Vec::with_capacity(n_trunk);
4105 let mut layer_is_moe = Vec::with_capacity(n_trunk);
4106 let (mut kda_heads, mut kda_head_dim, mut mla_heads) = (0usize, 0usize, 0usize);
4107 for (il, lp) in plan.layers.iter().take(n_trunk).enumerate() {
4108 match &lp.attention {
4109 AttentionPlan::KimiDeltaNet(k) => {
4110 layer_class.push(crate::glm5_tp::Glm5LayerClass::Kda);
4111 kda_heads = k.num_heads as usize;
4112 kda_head_dim = k.head_dim as usize;
4113 }
4114 AttentionPlan::Mla(memra_gguf::model_plan::MlaAttentionPlan::LatentKv {
4115 query_heads,
4116 ..
4117 }) => {
4118 layer_class.push(crate::glm5_tp::Glm5LayerClass::Mla);
4119 mla_heads = *query_heads as usize;
4120 }
4121 other => {
4122 return Err(format!(
4123 "MEMRA_GLM5_TP requires a glm5_next-class plan (KDA/MLA mixers): \
4124 trunk layer {il} declares {other:?}"
4125 )
4126 .into());
4127 }
4128 }
4129 layer_is_moe.push(matches!(&lp.mlp, MlpPlan::Moe(_)));
4130 }
4131 let view = crate::glm5_tp::Glm5TpModelView {
4132 trunk_layers: n_trunk,
4133 layer_class,
4134 layer_is_moe,
4135 kda_heads,
4136 kda_head_dim,
4137 mla_heads,
4138 n_routed_experts: moe.expert_count as usize,
4139 top_k: moe.expert_used_count as usize,
4140 };
4141 crate::glm5_tp::prepare_glm5_tp_load(e, &view)?
4142 } else {
4143 let glm5_class = plan.layers.iter().take(n_trunk).any(|lp| {
4150 matches!(
4151 lp.attention,
4152 memra_gguf::model_plan::AttentionPlan::KimiDeltaNet(_)
4153 )
4154 });
4155 let ep_map_armed = crate::ep_map::ep_map_env()?;
4156 if let Some((flag, _)) = ep_map_armed
4157 && glm5_class
4158 {
4159 return Err(format!(
4160 "{flag} is set but MEMRA_GLM5_TP is off: the map cannot \
4161 engage, and a placement that silently reverts to the even split is \
4162 refused by name (unset one of the two)"
4163 )
4164 .into());
4165 }
4166 if glm5_class {
4174 for (armed, flag) in [crate::ep_diet_armed(), crate::ep_grouped_prime_armed()] {
4175 if armed {
4176 return Err(format!(
4177 "{flag}=1 is set but MEMRA_GLM5_TP is off: the EP dispatch \
4178 diet only exists inside the TP-2 EP walk and cannot engage \
4179 (unset one of the two)"
4180 )
4181 .into());
4182 }
4183 }
4184 }
4185 None
4186 };
4187 let embd = EmbedHost::from_source(src, "token_embd.weight");
4188 let e_head = crate::pp::layer_engine(e, n_trunk, n_trunk - 1)?;
4192 let output_norm = load_t(e_head, src, "output_norm.weight")?;
4193 let mut output = if src.has("output.weight") {
4195 load_t(e_head, src, "output.weight")?
4196 } else {
4197 load_t(e_head, src, "token_embd.weight")?
4198 };
4199 let mut resident = ResidentPlan::pp(e, src, &cfg, n_trunk)?;
4200 resident.exclude_distributed_expert_layers(
4201 step_parallel
4202 .ep_specs
4203 .iter()
4204 .map(|spec| spec.layer)
4205 .chain(step_parallel.tp_specs.iter().map(|spec| spec.layer)),
4206 );
4207 let mut step_runtimes = StepParallelRuntimeRegistry::with_config(step_parallel);
4208
4209 let gguf: Option<&GgufFile> = src.gguf();
4216 let mut spill: Option<crate::spill::SpillCtx> = if cfg
4219 .moe
4220 .as_ref()
4221 .is_some_and(|m| m.expert_count > 0)
4222 && crate::spill::disk_tier_enabled()
4223 && gguf.is_some()
4224 {
4225 let budget = crate::spill::MemBudget::probe(e)?;
4226 #[allow(clippy::unnecessary_unwrap)]
4227 let ctx = crate::spill::SpillCtx::open(gguf.unwrap(), &budget)?;
4229 eprintln!(
4230 "[spill] disk tier ON: free_vram={} MiB free_pinnable_ram={} MiB (MemAvailable*resolved_frac)",
4231 budget.free_vram >> 20,
4232 budget.free_pinnable_ram >> 20
4233 );
4234 Some(ctx)
4235 } else {
4236 None
4237 };
4238
4239 let hyper = crate::hyper::HyperTopology::from_plan(&plan)?;
4246 let hyper_head = match hyper.as_ref() {
4247 Some(topology) => {
4248 crate::hyper::HyperHead::load(e_head, src, topology, cfg.n_embd as usize)?
4249 }
4250 None => None,
4251 };
4252 let mut layers = Vec::with_capacity(n_trunk);
4253 for il in 0..n_trunk as u32 {
4254 let p = |s: &str| format!("blk.{il}.{s}");
4255 let layer_plan = plan
4256 .layers
4257 .get(il as usize)
4258 .ok_or_else(|| format!("ModelPlan has no trunk layer {il}"))?;
4259 let e = crate::pp::layer_engine(e, n_trunk, il as usize)?;
4263 layers.push(HybridLayer {
4265 attn_norm: load_t(e, src, &p("attn_norm.weight"))?,
4266 post_attn_norm: load_opt(e, src, &p("post_attention_norm.weight"))?
4267 .or(load_opt(e, src, &p("ffn_norm.weight"))?)
4268 .expect("need post_attention_norm or ffn_norm"),
4269 mixer: {
4270 let g4_shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
4274 let kv_from = n_trunk as u32 - g4_shared;
4275 if g4_shared > 0
4276 && il >= kv_from
4277 && !src.has(&format!("blk.{il}.attn_k.weight"))
4278 {
4279 let g4 = cfg.gemma4.as_ref().unwrap();
4280 let swa = g4.swa_pattern.get(il as usize).copied().unwrap_or(true);
4281 let tgt = kv_from - if swa { 2 } else { 1 };
4282 let tp = |s: &str| format!("blk.{tgt}.{s}");
4283 Mixer::Full(FullAttnLayer {
4284 wq: load_t(e, src, &p("attn_q.weight"))?,
4285 wk: load_t(e, src, &tp("attn_k.weight"))?,
4286 wv: load_t(e, src, &tp("attn_v.weight"))?,
4287 wo: load_t(e, src, &p("attn_output.weight"))?,
4288 q_norm: load_t(e, src, &p("attn_q_norm.weight"))?,
4289 k_norm: load_t(e, src, &tp("attn_k_norm.weight"))?,
4290 attn_gate: None, step_tp_qkv: None,
4292 })
4293 } else {
4294 load_mixer_kind(
4295 e,
4296 src,
4297 &cfg,
4298 il,
4299 &layer_plan.attention,
4300 &mut step_runtimes,
4301 )?
4302 }
4303 },
4304 ffn: load_ffn(
4305 e,
4306 src,
4307 &cfg,
4308 &layer_plan.mlp,
4309 il,
4310 spill.as_mut().map(|c| (gguf.unwrap(), c)),
4311 &mut resident,
4312 &mut step_runtimes,
4313 )?,
4314 gemma4: if gemma_program {
4315 let scalar = |n: &str| -> f32 {
4316 let t = src.find(&p(n)).unwrap_or_else(|| panic!("missing {n}"));
4317 memra_gguf::dequant::dequantize(t.ggml_type, &t.bytes, 1)[0]
4318 };
4319 let vecf = |n: &str| -> Vec<f32> {
4320 let t = src.find(&p(n)).unwrap_or_else(|| panic!("missing {n}"));
4321 memra_gguf::dequant::dequantize(
4322 t.ggml_type,
4323 &t.bytes,
4324 t.ne.iter().product::<u64>() as usize,
4325 )
4326 };
4327 let moe_bits = if src.find(&p("ffn_gate_inp.scale")).is_some() {
4328 Some(crate::hybrid::Gemma4MoeBits {
4329 post_ffw_norm_1: load_t(e, src, &p("post_ffw_norm_1.weight"))?,
4330 pre_ffw_norm_2: load_t(e, src, &p("pre_ffw_norm_2.weight"))?,
4331 post_ffw_norm_2: load_t(e, src, &p("post_ffw_norm_2.weight"))?,
4332 shared_gate: load_t(e, src, &p("ffn_gate.weight"))?,
4333 shared_up: load_t(e, src, &p("ffn_up.weight"))?,
4334 shared_down: load_t(e, src, &p("ffn_down.weight"))?,
4335 router_scale_pre: {
4336 let inv = 1.0 / (cfg.n_embd as f32).sqrt();
4337 let v: Vec<f32> =
4338 vecf("ffn_gate_inp.scale").iter().map(|x| x * inv).collect();
4339 e.htod(&v)?
4340 },
4341 per_expert_scale: vecf("ffn_down_exps.scale"),
4342 per_expert_scale_d: e.htod(&vecf("ffn_down_exps.scale"))?,
4343 })
4344 } else {
4345 None
4346 };
4347 let e4b = if src.has(&p("inp_gate.weight")) {
4349 let g4 = cfg.gemma4.as_ref().unwrap();
4350 let kv_from = n_trunk as u32 - g4.shared_kv_layers;
4351 let kv_share = if g4.shared_kv_layers > 0 && il >= kv_from {
4352 let swa = g4.swa_pattern.get(il as usize).copied().unwrap_or(true);
4353 Some(kv_from - if swa { 2 } else { 1 })
4354 } else {
4355 None
4356 };
4357 Some(crate::hybrid::Gemma4E4bLayer {
4358 inp_gate: load_t(e, src, &p("inp_gate.weight"))?,
4359 proj: load_t(e, src, &p("proj.weight"))?,
4360 post_norm: load_t(e, src, &p("post_norm.weight"))?,
4361 kv_share,
4362 qkv_cat: None, })
4364 } else {
4365 None
4366 };
4367 Some(Gemma4LayerBits {
4368 ffn_norm: load_t(e, src, &p("ffn_norm.weight"))?,
4369 post_ffw_norm: load_t(e, src, &p("post_ffw_norm.weight"))?,
4370 moe_bits,
4371 layer_scale: scalar("layer_output_scale.weight"),
4372 e4b,
4373 })
4374 } else {
4375 None
4376 },
4377 hyper: match hyper.as_ref() {
4378 Some(topology) => Some(crate::hyper::HyperLayer::load(
4379 e,
4380 src,
4381 il,
4382 topology,
4383 cfg.n_embd as usize,
4384 )?),
4385 None => None,
4386 },
4387 });
4388 if let Some(tp_plan) = &glm5_tp
4391 && tp_plan.layers.contains(&(il as usize))
4392 {
4393 let mut layer = layers.pop().expect("layer just pushed");
4394 layer.mixer = match layer.mixer {
4395 Mixer::Kda(la) => {
4396 Mixer::Kda(crate::glm5_tp::shard_kda_layer(e, &tp_plan.rt, la)?)
4397 }
4398 Mixer::Mla(la) => {
4399 Mixer::Mla(crate::glm5_tp::shard_mla_layer(e, &tp_plan.rt, la)?)
4400 }
4401 _ => {
4402 return Err(format!(
4403 "MEMRA_GLM5_TP selected layer {il}, whose loaded mixer is not \
4404 KDA/MLA — preflight and loader disagree (wiring bug)"
4405 )
4406 .into());
4407 }
4408 };
4409 if let Ffn::Moe(m) = &mut layer.ffn {
4410 let placement = match &tp_plan.ep_map {
4414 Some(map) => Some(
4415 map.layers
4416 .get(&(il as usize))
4417 .ok_or_else(|| {
4418 format!(
4419 "glm5-tp EP: preflight-validated map lost layer {il} \
4420 (wiring bug)"
4421 )
4422 })?
4423 .as_slice(),
4424 ),
4425 None => None,
4426 };
4427 crate::glm5_tp::arm_moe_ep(e, &tp_plan.rt, m, placement)?;
4428 }
4429 layers.push(layer);
4430 }
4431 }
4432
4433 let external_mtp_requested =
4437 load_mtp && std::env::var("MEMRA_MTP_DRAFT").is_ok_and(|path| !path.is_empty());
4438 let trim_mtp_requested = load_mtp
4439 && !crate::model::full_prec_enabled()
4440 && std::env::var("MEMRA_FRSPEC_TRIM").is_ok_and(|path| !path.is_empty());
4441 let _ = trim_mtp_requested;
4452 let glm5_mtp_requested =
4462 !cfg.arch.is_glm5_next() || std::env::var("MEMRA_GLM5_MTP").as_deref() == Ok("1");
4463 let embedded_head_count =
4466 if external_mtp_requested || !glm5_mtp_requested || mtp_skip_requested {
4467 0
4468 } else {
4469 cfg.nextn_predict_layers
4470 };
4471 if cfg.arch.is_glm5_next()
4472 && glm5_mtp_requested
4473 && !mtp_skip_requested
4474 && cfg.nextn_predict_layers > 0
4475 {
4476 eprintln!("[mtp-glm5] MEMRA_GLM5_MTP=1: loading the glm5_next NextN block");
4477 }
4478 let embedded_head_count = match std::env::var("MEMRA_MTP_HEADS")
4483 .ok()
4484 .and_then(|v| v.parse::<u32>().ok())
4485 .filter(|&n| n > 0)
4486 {
4487 Some(cap) if cap < embedded_head_count => {
4488 eprintln!(
4489 "[mtp-chain] MEMRA_MTP_HEADS={cap}: capping the embedded chain from \
4490 {embedded_head_count} heads (measurement knob)"
4491 );
4492 cap
4493 }
4494 _ => embedded_head_count,
4495 };
4496 let mut embedded_mtp = Vec::new();
4497 if load_mtp && embedded_head_count > 0 {
4498 for offset in 0..embedded_head_count {
4499 let n = n_trunk as u32 + offset;
4500 let e = crate::pp::layer_engine(e, n_trunk, n as usize)?;
4506 let p = |s: &str| format!("blk.{n}.{s}");
4507 let mtp_plan = plan
4508 .mtp_blocks
4509 .iter()
4510 .find(|block| block.layer.index == n)
4511 .ok_or_else(|| format!("ModelPlan has no embedded MTP block {n}"))?;
4512 if !src.has(&p("nextn.eh_proj.weight")) {
4513 if offset == 0 {
4514 break;
4515 }
4516 return Err(format!(
4517 "embedded MTP chain declares {} heads but blk.{n} has no \
4518 nextn.eh_proj.weight",
4519 cfg.nextn_predict_layers
4520 )
4521 .into());
4522 }
4523 embedded_mtp.push(MtpHead {
4524 enorm: load_t(e, src, &p("nextn.enorm.weight"))?,
4525 hnorm: load_t(e, src, &p("nextn.hnorm.weight"))?,
4526 eh_proj: load_t(e, src, &p("nextn.eh_proj.weight"))?,
4527 attn_norm: load_t(e, src, &p("attn_norm.weight"))?,
4528 post_attn_norm: load_opt(e, src, &p("post_attention_norm.weight"))?
4529 .or(load_opt(e, src, &p("ffn_norm.weight"))?)
4530 .expect("MTP block needs post_attention_norm or ffn_norm"),
4531 mixer: load_mixer_kind(
4532 e,
4533 src,
4534 &cfg,
4535 n,
4536 &mtp_plan.layer.attention,
4537 &mut step_runtimes,
4538 )?,
4539 ffn: load_ffn(
4540 e,
4541 src,
4542 &cfg,
4543 &mtp_plan.layer.mlp,
4544 n,
4545 spill.as_mut().map(|c| (gguf.unwrap(), c)),
4546 &mut resident,
4547 &mut step_runtimes,
4548 )?,
4549 shared_head_norm: load_opt(e, src, &p("nextn.shared_head_norm.weight"))?,
4550 shared_head_head: load_mtp_head_maybe_nvfp4(
4559 e,
4560 src,
4561 &p("nextn.shared_head_head.weight"),
4562 )?
4563 .or(load_opt(e, src, &p("nextn.shared_head.weight"))?),
4564 d2t: None,
4565 d2t_from_target_head: false,
4566 geom: None,
4567 step35: if sliding_gated_moe_program {
4568 Some(Step35MtpGeom::from_plan(&mtp_plan.layer)?)
4569 } else {
4570 None
4571 },
4572 });
4573 }
4574 }
4575 let mut embedded_mtp = embedded_mtp.into_iter();
4576 let mut mtp = embedded_mtp.next();
4577 let mut mtp_extra: Vec<MtpHead> = embedded_mtp.collect();
4578
4579 mtp = if load_mtp {
4583 match std::env::var("MEMRA_MTP_DRAFT") {
4584 Ok(path) if !path.is_empty() => {
4585 eprintln!("[mtp-draft] loading external MTP draft: {path}");
4586 let dg = GgufFile::open(&path)?;
4587 mtp_extra.clear();
4588 Some(MtpHead::load_draft(e, &dg, &cfg)?)
4589 }
4590 _ => mtp,
4591 }
4592 } else {
4593 None
4594 };
4595
4596 let trim_env = if load_mtp {
4607 std::env::var("MEMRA_FRSPEC_TRIM")
4608 } else {
4609 Err(std::env::VarError::NotPresent)
4610 };
4611 if crate::model::full_prec_enabled()
4612 && trim_env.as_deref().map(|p| !p.is_empty()).unwrap_or(false)
4613 {
4614 eprintln!(
4615 "[frspec-trim] DISABLED under MEMRA_FULL_PREC — using the natural full MTP head"
4616 );
4617 }
4618 mtp = match (
4619 if crate::model::full_prec_enabled() {
4620 Err(std::env::VarError::NotPresent)
4621 } else {
4622 trim_env
4623 },
4624 mtp,
4625 ) {
4626 (Ok(path), Some(mut head)) if !path.is_empty() => {
4627 let e = crate::pp::layer_engine(e, n_trunk, n_trunk)?;
4631 let path = memra_gguf::hf::resolve_arg(&path)
4635 .map_err(|err| format!("MEMRA_FRSPEC_TRIM={path:?}: {err}"))?;
4636 let d2t: Vec<u32> = frspec_read_d2t(&path)?;
4640 let own_head_name = frspec_trim_own_head_name(n_trunk);
4649 let own_head = src.find(&own_head_name);
4650 let from_own_head = own_head.is_some();
4651 let v = own_head
4652 .or_else(|| src.find("output.weight"))
4653 .or_else(|| src.find("token_embd.weight"))
4654 .expect("model has no output.weight for FR-Spec trim");
4655 let (trimmed, nvfp4_sizes) = frspec_gather_trimmed_head(
4675 e,
4676 &v,
4677 &d2t,
4678 std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1"),
4679 match src.find("output.scale") {
4681 Some(sv) => f32::from_le_bytes(sv.bytes[..4].try_into().unwrap()),
4682 None => 1.0,
4683 },
4684 )?;
4685 match nvfp4_sizes {
4686 Some((nvfp4_bytes, gathered_bytes)) => eprintln!(
4687 "[frspec-trim] self-trimmed head: {} rows of {} re-quantized BF16 -> NVFP4 \
4688 ({} MiB, was {} MiB)",
4689 d2t.len(),
4690 if from_own_head {
4691 own_head_name.as_str()
4692 } else {
4693 "main output.weight"
4694 },
4695 nvfp4_bytes >> 20,
4696 gathered_bytes >> 20,
4697 ),
4698 None => eprintln!(
4699 "[frspec-trim] self-trimmed head: {} rows of {} ({:?})",
4700 d2t.len(),
4701 if from_own_head {
4702 own_head_name.as_str()
4703 } else {
4704 "main output.weight"
4705 },
4706 v.ggml_type
4707 ),
4708 }
4709 head.shared_head_head = Some(trimmed);
4710 head.d2t = Some(d2t);
4711 head.d2t_from_target_head = !from_own_head;
4714 Some(head)
4715 }
4716 (_, m) => m,
4717 };
4718 let dflash_trim: Option<DflashTrimHead> = match mtp_skip_trim_d2t {
4729 Some(d2t) => {
4730 let v = src
4731 .find("output.weight")
4732 .or_else(|| src.find("token_embd.weight"))
4733 .ok_or("model has no output.weight for FR-Spec trim")?;
4734 let (head, nvfp4_sizes) = frspec_gather_trimmed_head(
4735 e,
4736 &v,
4737 &d2t,
4738 std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1"),
4739 match src.find("output.scale") {
4740 Some(sv) => f32::from_le_bytes(sv.bytes[..4].try_into().unwrap()),
4741 None => 1.0,
4742 },
4743 )?;
4744 eprintln!(
4745 "[mtp-skip] FR-Spec stub draft head built: {} rows of main output.weight \
4746 ({}); DFlash2 trim serves without the embedded MTP block",
4747 d2t.len(),
4748 match nvfp4_sizes {
4749 Some((nvfp4_bytes, gathered_bytes)) => format!(
4750 "re-quantized BF16 -> NVFP4, {} MiB, was {} MiB",
4751 nvfp4_bytes >> 20,
4752 gathered_bytes >> 20
4753 ),
4754 None => format!("{:?}", v.ggml_type),
4755 },
4756 );
4757 Some(DflashTrimHead { head, d2t })
4758 }
4759 None => None,
4760 };
4761 if let Some(d2t) = mtp.as_ref().and_then(|head| head.d2t.clone()) {
4774 let want_nvfp4_env = std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1");
4775 let mut kept = 0usize;
4776 let e = crate::pp::layer_engine(e, n_trunk, n_trunk)?;
4779 for (i, head) in mtp_extra.iter_mut().enumerate() {
4780 let name = frspec_trim_own_head_name(n_trunk + 1 + i);
4781 let Some(v) = src.find(&name) else { break };
4782 let out_f = v.ne[1] as usize;
4783 let row_bytes = v.bytes.len() / out_f;
4784 if d2t.iter().any(|&t| (t as usize) >= out_f) {
4785 break;
4786 }
4787 let mut gathered = Vec::with_capacity(d2t.len() * row_bytes);
4788 for &t in &d2t {
4789 let off = t as usize * row_bytes;
4790 gathered.extend_from_slice(&v.bytes[off..off + row_bytes]);
4791 }
4792 let want_nvfp4 =
4793 want_nvfp4_env && matches!(v.ggml_type, GgmlType::BF16) && v.ne[0] % 64 == 0;
4794 let trimmed = if want_nvfp4 {
4795 let vals: Vec<f32> = gathered
4796 .chunks_exact(2)
4797 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
4798 .collect();
4799 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
4800 GpuTensor::from_quant_bytes(
4801 e,
4802 &blocks,
4803 GgmlType::NVFP4,
4804 v.ne[0],
4805 d2t.len() as u64,
4806 1.0,
4807 )?
4808 } else {
4809 match v.ggml_type {
4810 GgmlType::BF16 => GpuTensor::FloatBf16 {
4811 data: e.htod_bytes(&gathered)?,
4812 ne: vec![v.ne[0], d2t.len() as u64],
4813 },
4814 GgmlType::F32 => GpuTensor::Float {
4815 data: e.htod(
4816 &gathered
4817 .chunks_exact(4)
4818 .map(|c| f32::from_le_bytes(c.try_into().unwrap()))
4819 .collect::<Vec<f32>>(),
4820 )?,
4821 ne: vec![v.ne[0], d2t.len() as u64],
4822 },
4823 _ => GpuTensor::from_quant_bytes(
4824 e,
4825 &gathered,
4826 v.ggml_type,
4827 v.ne[0],
4828 d2t.len() as u64,
4829 1.0,
4830 )?,
4831 }
4832 };
4833 head.shared_head_head = Some(trimmed);
4834 head.d2t = Some(d2t.clone());
4835 head.d2t_from_target_head = false;
4836 kept += 1;
4837 }
4838 let dropped = mtp_extra.len() - kept;
4839 mtp_extra.truncate(kept);
4840 eprintln!(
4841 "[frspec-trim] per-head trim: {kept} extra chain head(s) gathered from their own \
4842 blocks{}",
4843 if dropped > 0 {
4844 format!(" ({dropped} dropped: no own-head tensor)")
4845 } else {
4846 String::new()
4847 }
4848 );
4849 }
4850 if !mtp_extra.is_empty() {
4851 if plan.draft_source != memra_gguf::model_plan::DraftSourcePlan::Embedded
4852 || plan.mtp_blocks.len() != 1 + mtp_extra.len()
4853 || plan
4854 .mtp_blocks
4855 .iter()
4856 .any(|block| !matches!(block.layer.mlp, MlpPlan::Dense(_)))
4857 || mtp
4858 .iter()
4859 .chain(mtp_extra.iter())
4860 .any(|head| !matches!(head.ffn, Ffn::Dense { .. }))
4861 {
4862 return Err(
4863 "multi-head MTP requires embedded dense canonical blocks and matching loaded heads"
4864 .into(),
4865 );
4866 }
4867 eprintln!(
4868 "[mtp-draft] embedded chain: heads={} blocks={}..={} scratch=per-head",
4869 1 + mtp_extra.len(),
4870 n_trunk,
4871 n_trunk + mtp_extra.len()
4872 );
4873 }
4874
4875 let glm5_dflash = match std::env::var("MEMRA_GLM5_DFLASH") {
4891 Ok(spec) if !spec.is_empty() && cfg.arch.is_glm5_next() => {
4892 let dpath = memra_gguf::hf::resolve_arg(&spec)
4893 .map_err(|err| format!("MEMRA_GLM5_DFLASH={spec:?}: {err}"))?;
4894 let de = crate::pp::layer_engine(e, n_trunk, n_trunk)?;
4895 Some(crate::dflash::load_drafter(
4896 de,
4897 std::path::Path::new(&dpath),
4898 "MEMRA_GLM5_DFLASH",
4899 n_trunk,
4900 cfg.n_embd as usize,
4901 output.out_features(),
4902 )?)
4903 }
4904 _ => None,
4905 };
4906
4907 if cfg.arch.is_glm5_next() && crate::glm_spec::glm5_spec_on() {
4915 match (glm5_dflash.as_ref(), mtp.as_ref()) {
4916 (Some(dr), head) => {
4917 let trim_note = match head.and_then(|h| h.d2t.as_ref()) {
4918 Some(map) => {
4919 format!("draft head TRIMMED to {} rows (FR-Spec d2t)", map.len())
4920 }
4921 None => "draft head FULL target vocab".to_string(),
4922 };
4923 eprintln!(
4924 "[glm5-spec] serve route ARMED: draft source = dflash2 @ {}; {trim_note}; \
4925 native MTP head {}",
4926 dr.sha8,
4927 if head.is_some() {
4928 "ALSO loaded (idle for drafting — dflash2 wins by selection)"
4929 } else {
4930 "NOT loaded (the q38 pattern: a full MoE trunk layer of VRAM saved)"
4931 }
4932 );
4933 }
4934 (None, Some(head)) => {
4935 match head.d2t.as_ref() {
4936 Some(map) => eprintln!(
4937 "[glm5-spec] serve route ARMED: MTP head loaded; draft head TRIMMED \
4938 to {} rows (FR-Spec d2t engaged)",
4939 map.len()
4940 ),
4941 None => eprintln!(
4942 "[glm5-spec] serve route ARMED: MTP head loaded; draft head FULL \
4943 target vocab (no FR-Spec trim)"
4944 ),
4945 }
4946 eprintln!("[glm5-spec] draft source = native-mtp");
4947 }
4948 (None, None) => eprintln!(
4949 "[glm5-spec] MEMRA_GLM5_SPEC=1 but no MTP head loaded \
4950 (set MEMRA_GLM5_MTP=1 or MEMRA_GLM5_DFLASH=<drafter>) — route stays \
4951 fail-closed, plain serving"
4952 ),
4953 }
4954 }
4955
4956 if let Some(ctx) = spill.as_ref() {
4957 eprintln!(
4958 "[spill] experts placed: {} pinned (Tier 1), {} mmap'd from disk (Tier 2, {} MiB)",
4959 ctx.n_pinned,
4960 ctx.n_mmap,
4961 ctx.mmap_bytes >> 20
4962 );
4963 }
4964
4965 if cfg.n_head_kv > 0 && cfg.n_head / cfg.n_head_kv > 8 {
4979 crate::FA_V4_MAX_DEFAULT.store(0, std::sync::atomic::Ordering::Relaxed);
4980 eprintln!(
4981 "[fa] v4 decode family disabled: gqa {} > fa_v4_smem capacity 8 (v3 lane serves)",
4982 cfg.n_head / cfg.n_head_kv
4983 );
4984 }
4985
4986 if gemma_program {
4987 crate::FA_VEC_MIN_DEFAULT.store(1, std::sync::atomic::Ordering::Relaxed);
4989 let real_moe = plan
4992 .trunk_operations()
4993 .contains(&memra_gguf::model_plan::OperationKind::MoeMlp);
4994 crate::FA_SPW_DEFAULT.store(
4995 if real_moe { 32 } else { 64 },
4996 std::sync::atomic::Ordering::Relaxed,
4997 );
4998 crate::FA_SP512_DEFAULT.store(
5000 if real_moe { 16 } else { 32 },
5001 std::sync::atomic::Ordering::Relaxed,
5002 );
5003 crate::FUSED_MR1_DEFAULT.store(!real_moe, std::sync::atomic::Ordering::Relaxed);
5013 crate::RMS_BLOCK_DEFAULT.store(1024, std::sync::atomic::Ordering::Relaxed);
5015 crate::FA_SP_GEMMA.store(true, std::sync::atomic::Ordering::Relaxed);
5017 }
5021 let force_embd_gpu = gemma_program;
5024 let gemma4_aux = if gemma_program {
5025 let rope_freqs = match src.find("rope_freqs.weight") {
5026 Some(t) => {
5027 let host = memra_gguf::dequant::dequantize(
5028 t.ggml_type,
5029 &t.bytes,
5030 t.ne.iter().product::<u64>() as usize,
5031 );
5032 let mut copies = Vec::new();
5033 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
5034 #[allow(clippy::needless_range_loop)]
5035 for s in 0..fence.len() - 1 {
5037 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
5038 let dev = owner.ctx().ordinal();
5039 if copies.iter().all(|(d, _)| *d != dev) {
5040 copies.push((dev, owner.htod(&host)?));
5041 }
5042 }
5043 } else {
5044 copies.push((e.ctx().ordinal(), e.htod(&host)?));
5045 }
5046 Some(copies)
5047 }
5048 None => {
5056 let g4 = cfg.gemma4.as_ref().unwrap();
5057 let n = (g4.rope_dims_global / 2) as usize;
5058 let keep =
5059 ((n as f32) * g4.partial_rotary_global.clamp(0.0, 1.0)).round() as usize;
5060 let host: Vec<f32> = (0..n)
5061 .map(|i| if i < keep { 1.0 } else { 1.0e30 })
5062 .collect();
5063 eprintln!(
5064 "[gemma4] rope_freqs.weight synthesized ({n} factors, first {keep} \
5065 rotate; source ships none — native checkpoint)"
5066 );
5067 let mut copies = Vec::new();
5068 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
5069 #[allow(clippy::needless_range_loop)]
5070 for s in 0..fence.len() - 1 {
5072 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
5073 let dev = owner.ctx().ordinal();
5074 if copies.iter().all(|(d, _)| *d != dev) {
5075 copies.push((dev, owner.htod(&host)?));
5076 }
5077 }
5078 } else {
5079 copies.push((e.ctx().ordinal(), e.htod(&host)?));
5080 }
5081 Some(copies)
5082 }
5083 };
5084 let e4b = match src.find("per_layer_token_embd.weight") {
5086 Some(t) => {
5087 let n_epl = cfg
5088 .gemma4
5089 .as_ref()
5090 .map(|g| g.n_embd_per_layer as usize)
5091 .unwrap_or(0);
5092 let row = t.ne[0] as usize; let row_bytes = t.bytes.len() / (t.ne[1] as usize);
5094 eprintln!(
5095 "[gemma4-e4b] per-layer-embed model detected (n_epl={n_epl}, row {row}) — \
5096 first-light forward (eager decode + prime); dc/graph/spec unwired \
5097 (HANDOVER-E4B.md)"
5098 );
5099 Some(crate::hybrid::Gemma4E4bModel {
5100 tok_tbl_gpu: std::sync::OnceLock::new(),
5101 tok_embd_bytes: t.bytes.to_vec(),
5102 tok_embd_qt: match t.ggml_type {
5103 memra_gguf::GgmlType::Q6_K => crate::QT_Q6_K,
5104 memra_gguf::GgmlType::Q8_0 => crate::QT_Q8_0,
5105 other => panic!("e4b per-layer tok embd: unhandled dtype {other:?}"),
5106 },
5107 tok_embd_row_bytes: row_bytes,
5108 model_proj: load_t(e, src, "per_layer_model_proj.weight")?,
5109 proj_norm: load_t(e, src, "per_layer_proj_norm.weight")?,
5110 n_epl,
5111 })
5112 }
5113 None => None,
5114 };
5115 let suppress_d = {
5116 let sup = &cfg.gemma4.as_ref().unwrap().suppress_tokens;
5117 if sup.is_empty() {
5118 None
5119 } else {
5120 let ids: Vec<i32> = sup.iter().map(|&x| x as i32).collect();
5121 eprintln!(
5122 "[gemma4] suppress_tokens: {} ids masked at sampling",
5123 ids.len()
5124 );
5125 Some((e.htod_i32(&ids)?, ids.len()))
5126 }
5127 };
5128 let ones_host = [1.0f32; 512];
5129 let mut ones = Vec::new();
5130 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
5131 #[allow(clippy::needless_range_loop)]
5132 for s in 0..fence.len() - 1 {
5134 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
5135 let dev = owner.ctx().ordinal();
5136 if ones.iter().all(|(d, _)| *d != dev) {
5137 ones.push((dev, owner.htod(&ones_host)?));
5138 }
5139 }
5140 } else {
5141 ones.push((e.ctx().ordinal(), e.htod(&ones_host)?));
5142 }
5143 Some(GemmaAux {
5144 rope_freqs,
5145 ones,
5146 suppress_d,
5147 e4b,
5148 })
5149 } else {
5150 None
5151 };
5152 let step35_aux = if sliding_gated_moe_program {
5156 let rope_freqs = match src.find("rope_freqs.weight") {
5157 Some(t) => {
5158 let host = memra_gguf::dequant::dequantize(
5159 t.ggml_type,
5160 &t.bytes,
5161 t.ne.iter().product::<u64>() as usize,
5162 );
5163 let mut copies = Vec::new();
5164 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
5165 #[allow(clippy::needless_range_loop)]
5166 for s in 0..fence.len() - 1 {
5168 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
5169 let dev = owner.ctx().ordinal();
5170 if copies.iter().all(|(d, _)| *d != dev) {
5171 copies.push((dev, owner.htod(&host)?));
5172 }
5173 }
5174 } else {
5175 copies.push((e.ctx().ordinal(), e.htod(&host)?));
5176 }
5177 Some(copies)
5178 }
5179 None => None,
5180 };
5181 Some(Step35Aux { rope_freqs })
5182 } else {
5183 None
5184 };
5185 let mut layers = layers;
5186 {
5193 let q8rp_on = match std::env::var("MEMRA_Q8RP").as_deref() {
5194 Ok("0") => false,
5195 Ok(_) => true,
5196 Err(_) => {
5204 cfg!(memra_hopper_mma) || {
5205 let q8b = |w: &crate::model::GpuTensor| -> usize {
5206 match w {
5207 crate::model::GpuTensor::Quant {
5208 bytes,
5209 qtype,
5210 row_bytes,
5211 ne,
5212 rp4: None,
5213 ..
5214 } if *qtype == crate::QT_Q8_0
5215 && ne.len() == 2
5216 && (ne[0] as usize).is_multiple_of(32)
5217 && *row_bytes == (ne[0] as usize / 32) * 34 =>
5218 {
5219 bytes.len()
5220 }
5221 _ => 0,
5222 }
5223 };
5224 let mut need = q8b(&output);
5225 for layer in layers.iter() {
5226 match &layer.mixer {
5227 Mixer::Full(fa) => {
5228 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
5229 need += q8b(w);
5230 }
5231 }
5232 Mixer::Linear(la) => {
5233 for w in [
5234 &la.wqkv,
5235 &la.wqkv_gate,
5236 &la.ssm_beta,
5237 &la.ssm_alpha,
5238 &la.ssm_out,
5239 ] {
5240 need += q8b(w);
5241 }
5242 }
5243 Mixer::Mla(_) => {}
5244 Mixer::Kda(_) => {} }
5246 if let Ffn::Dense {
5247 ffn_gate,
5248 ffn_up,
5249 ffn_down,
5250 } = &layer.ffn
5251 {
5252 for w in [ffn_gate, ffn_up, ffn_down] {
5253 need += q8b(w);
5254 }
5255 }
5256 }
5257 need > 0
5258 && e.ctx()
5259 .mem_get_info()
5260 .map(|(free, _)| free >= need + (8usize << 30))
5261 .unwrap_or(false)
5262 }
5263 }
5264 };
5265 let kqrp_on = crate::Engine::kqrp_enabled() || {
5275 std::env::var("MEMRA_KQRP").is_err() && {
5276 let kqb = |w: &crate::model::GpuTensor| -> usize {
5277 match w {
5278 crate::model::GpuTensor::Quant {
5279 bytes,
5280 qtype,
5281 row_bytes,
5282 ne,
5283 rp4: None,
5284 ..
5285 } if ne.len() == 2 && (ne[0] as usize).is_multiple_of(256) => {
5286 let sb = if *qtype == crate::QT_Q4_K {
5287 144
5288 } else if *qtype == crate::QT_Q6_K {
5289 210
5290 } else {
5291 return 0;
5292 };
5293 if *row_bytes == (ne[0] as usize / 256) * sb {
5294 bytes.len()
5295 } else {
5296 0
5297 }
5298 }
5299 _ => 0,
5300 }
5301 };
5302 let mut need = kqb(&output);
5303 for layer in layers.iter() {
5304 if let Mixer::Full(fa) = &layer.mixer {
5305 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
5306 need += kqb(w);
5307 }
5308 }
5309 if let Ffn::Dense {
5310 ffn_gate,
5311 ffn_up,
5312 ffn_down,
5313 } = &layer.ffn
5314 {
5315 for w in [ffn_gate, ffn_up, ffn_down] {
5316 need += kqb(w);
5317 }
5318 }
5319 }
5320 need > 0
5321 && e.ctx()
5322 .mem_get_info()
5323 .map(|(free, _)| free >= need + (8usize << 30))
5324 .unwrap_or(false)
5325 }
5326 };
5327 if q8rp_on || kqrp_on {
5328 let f16_model_ok = gemma_program
5335 || plan
5336 .trunk_operations()
5337 .contains(&memra_gguf::model_plan::OperationKind::MoeMlp)
5338 || std::env::var("MEMRA_PP_F16").as_deref() == Ok("1");
5339 let mut nmir = 0usize;
5340 let mut mir = |e_ref: &crate::Engine,
5344 w: &mut crate::model::GpuTensor|
5345 -> Result<(), Box<dyn std::error::Error>> {
5346 let before = matches!(w, crate::model::GpuTensor::Quant { rp4: Some(_), .. });
5347 if q8rp_on {
5348 e_ref.build_q8_rp4(w)?;
5349 }
5350 if kqrp_on {
5351 e_ref.build_q4k_rp4(w)?;
5352 e_ref.build_q6k_rp4(w)?;
5353 }
5354 let q6k = matches!(w, crate::model::GpuTensor::Quant { qtype, .. }
5359 if *qtype == crate::QT_Q6_K);
5360 if q8rp_on && crate::f16_ffi::pp_f16_enabled() && (f16_model_ok || q6k) {
5361 e_ref.build_q8_f16(w)?;
5362 }
5363 if !before && matches!(w, crate::model::GpuTensor::Quant { rp4: Some(_), .. }) {
5364 nmir += 1;
5365 }
5366 Ok(())
5367 };
5368 for (il, layer) in layers.iter_mut().enumerate() {
5369 let el = crate::pp::layer_engine(e, n_trunk, il)?;
5370 match &mut layer.mixer {
5371 Mixer::Full(fa) => {
5372 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
5373 mir(el, w)?;
5374 }
5375 }
5376 Mixer::Linear(la) => {
5377 for w in [
5378 &mut la.wqkv,
5379 &mut la.wqkv_gate,
5380 &mut la.ssm_beta,
5381 &mut la.ssm_alpha,
5382 &mut la.ssm_out,
5383 ] {
5384 mir(el, w)?;
5385 }
5386 }
5387 Mixer::Mla(_) => {}
5390 Mixer::Kda(_) => {} }
5392 if let Ffn::Dense {
5393 ffn_gate,
5394 ffn_up,
5395 ffn_down,
5396 } = &mut layer.ffn
5397 {
5398 for w in [ffn_gate, ffn_up, ffn_down] {
5399 mir(el, w)?;
5400 }
5401 }
5402 }
5403 mir(e_head, &mut output)?;
5404 if nmir > 0 {
5405 eprintln!("[q8rp] split-plane decode mirrors built: {nmir} tensors");
5406 }
5407 if q8rp_on && crate::f16_ffi::pp_f16_enabled() {
5422 for (want, tag) in [(crate::QT_Q4_K, "q4kf16"), (crate::QT_Q5_K, "q5kf16")] {
5423 let (mut n4, mut b4) = (0usize, 0usize);
5424 let mut mirk =
5425 |e_ref: &crate::Engine,
5426 w: &mut crate::model::GpuTensor|
5427 -> Result<(), Box<dyn std::error::Error>> {
5428 if matches!(w, crate::model::GpuTensor::Quant { qtype, f16: None, .. }
5429 if *qtype == want)
5430 {
5431 e_ref.build_q8_f16(w)?;
5432 if let crate::model::GpuTensor::Quant { f16: Some(m), .. } = w {
5433 n4 += 1;
5434 b4 += m.len();
5435 }
5436 }
5437 Ok(())
5438 };
5439 for (il, layer) in layers.iter_mut().enumerate() {
5440 let el = crate::pp::layer_engine(e, n_trunk, il)?;
5441 match &mut layer.mixer {
5442 Mixer::Full(fa) => {
5443 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
5444 mirk(el, w)?;
5445 }
5446 }
5447 Mixer::Linear(la) => {
5448 for w in [
5449 &mut la.wqkv,
5450 &mut la.wqkv_gate,
5451 &mut la.ssm_beta,
5452 &mut la.ssm_alpha,
5453 &mut la.ssm_out,
5454 ] {
5455 mirk(el, w)?;
5456 }
5457 }
5458 Mixer::Mla(_) => {} Mixer::Kda(_) => {} }
5461 if let Ffn::Dense {
5462 ffn_gate,
5463 ffn_up,
5464 ffn_down,
5465 } = &mut layer.ffn
5466 {
5467 for w in [ffn_gate, ffn_up, ffn_down] {
5468 mirk(el, w)?;
5469 }
5470 }
5471 }
5472 mirk(e_head, &mut output)?;
5473 if n4 > 0 {
5474 eprintln!(
5475 "[{tag}] prefill fp16 mirrors built: {n4} tensors \
5476 ({} MB)",
5477 b4 >> 20
5478 );
5479 }
5480 }
5481 }
5482 }
5483 }
5484 if gemma_program && crate::Engine::q4rp_enabled() {
5491 let mut nmir = 0usize;
5492 for (il, layer) in layers.iter_mut().enumerate() {
5493 let e = crate::pp::layer_engine(e, n_trunk, il)?;
5495 let is_moe26 = layer.gemma4.as_ref().is_some_and(|g| g.moe_bits.is_some());
5504 let is_e4b = layer.gemma4.as_ref().is_some_and(|g| g.e4b.is_some());
5505 if !(is_moe26 || is_e4b) {
5506 continue;
5507 }
5508 if let Mixer::Full(fa) = &mut layer.mixer {
5509 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
5510 e.build_q4_rp4(w)?;
5511 nmir += 1;
5512 }
5513 }
5514 if is_e4b {
5515 let own_kv = layer
5517 .gemma4
5518 .as_ref()
5519 .unwrap()
5520 .e4b
5521 .as_ref()
5522 .is_some_and(|e4| e4.kv_share.is_none());
5523 if own_kv
5524 && let Mixer::Full(fa) = &layer.mixer
5525 && let Some(mut cat) = e.build_q4_out_concat3(&fa.wq, &fa.wk, &fa.wv)?
5526 {
5527 e.build_q4_rp4(&mut cat)?;
5528 nmir += 1;
5529 layer.gemma4.as_mut().unwrap().e4b.as_mut().unwrap().qkv_cat = Some(cat);
5530 }
5531 if let Ffn::Dense {
5532 ffn_gate,
5533 ffn_up,
5534 ffn_down,
5535 } = &mut layer.ffn
5536 {
5537 for w in [ffn_gate, ffn_up, ffn_down] {
5538 e.build_q4_rp4(w)?;
5539 nmir += 1;
5540 }
5541 }
5542 let e4 = layer.gemma4.as_mut().unwrap().e4b.as_mut().unwrap();
5543 for w in [&mut e4.inp_gate, &mut e4.proj] {
5544 e.build_q4_rp4(w)?;
5545 nmir += 1;
5546 }
5547 }
5548 if let Some(mb) = layer.gemma4.as_mut().unwrap().moe_bits.as_mut() {
5549 for w in [&mut mb.shared_gate, &mut mb.shared_up, &mut mb.shared_down] {
5550 e.build_q4_rp4(w)?;
5551 nmir += 1;
5552 }
5553 }
5554 }
5555 if nmir > 0 {
5556 eprintln!("[q4rp] split-plane decode mirrors built: {nmir} trunk tensors");
5557 }
5558 let fast_on = std::env::var("MEMRA_FAST").as_deref() != Ok("0");
5565 if fast_on {
5566 let mut nswap = 0usize;
5567 let mut nf16 = 0usize;
5568 let q4f16_model_ok = matches!(cfg.n_embd, 3840 | 5376); if let Ok(v) = std::env::var("MEMRA_Q4F16")
5587 && v != "0"
5588 && v != "1"
5589 {
5590 return Err(format!(
5591 "MEMRA_Q4F16={v} is not 0 or 1 — this env selects the prefill \
5592 ARITHMETIC (fp16 mirrors vs int8 MMQ) and must never be guessed"
5593 )
5594 .into());
5595 }
5596 let f16_need = {
5597 let f16b = |w: &crate::model::GpuTensor| -> usize {
5598 match w {
5599 crate::model::GpuTensor::Quant {
5600 qtype,
5601 ne,
5602 f16: None,
5603 ..
5604 } if ne.len() == 2
5605 && matches!(
5606 *qtype,
5607 crate::QT_Q8_0
5608 | crate::QT_Q4_0
5609 | crate::QT_Q6_K
5610 | crate::QT_Q4_K
5611 | crate::QT_Q5_K
5612 ) =>
5613 {
5614 (ne[0] as usize) * (ne[1] as usize) * 2
5615 }
5616 _ => 0,
5617 }
5618 };
5619 let mut need = 0usize;
5620 for layer in layers.iter() {
5621 if layer.gemma4.as_ref().is_none_or(|g| g.moe_bits.is_some()) {
5622 continue;
5623 }
5624 if let Mixer::Full(fa) = &layer.mixer {
5625 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
5626 need += f16b(w);
5627 }
5628 }
5629 if let Ffn::Dense {
5630 ffn_gate,
5631 ffn_up,
5632 ffn_down,
5633 } = &layer.ffn
5634 {
5635 for w in [ffn_gate, ffn_up, ffn_down] {
5636 need += f16b(w);
5637 }
5638 }
5639 }
5640 need
5641 };
5642 let f16_free = e.ctx().mem_get_info().map(|(free, _)| free).unwrap_or(0);
5643 let f16_auto = q4f16_model_ok
5644 && std::env::var("MEMRA_Q4F16").is_err()
5645 && crate::f16_ffi::pp_f16_capacity_ok(f16_free, f16_need);
5646 let (f16_on, f16_why) = match std::env::var("MEMRA_Q4F16").as_deref() {
5652 Ok("1") => (true, "env MEMRA_Q4F16=1"),
5653 Ok("0") => (false, "env MEMRA_Q4F16=0"),
5654 _ if crate::f16_ffi::pp_f16_enabled() && q4f16_model_ok => {
5655 (true, "env MEMRA_PP_F16")
5656 }
5657 _ if f16_auto => (true, "capacity-keyed auto (UNPINNED)"),
5658 _ if !q4f16_model_ok => (false, "model geometry not eligible"),
5659 _ => (false, "capacity-keyed auto REFUSED (UNPINNED)"),
5660 };
5661 eprintln!(
5668 "[q4f16] prefill program = {} (reason: {}); free {} MiB, mirror mass {} MiB, \
5669 capacity threshold {} MiB (mass + 8192 headroom) — SELECTS PREFILL ARITHMETIC",
5670 if f16_on {
5671 "FP16 MIRRORS"
5672 } else {
5673 "INT8 MMQ (no f16 mirrors)"
5674 },
5675 f16_why,
5676 f16_free >> 20,
5677 f16_need >> 20,
5678 (f16_need + (8usize << 30)) >> 20,
5679 );
5680 for (il, layer) in layers.iter_mut().enumerate() {
5681 let e = crate::pp::layer_engine(e, n_trunk, il)?;
5683 let dense_gemma = layer.gemma4.as_ref().is_some_and(|g| g.moe_bits.is_none());
5684 if !dense_gemma {
5685 continue;
5686 }
5687 if let Mixer::Full(fa) = &mut layer.mixer {
5688 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
5689 if f16_on {
5690 e.build_q8_f16(w)?;
5691 if matches!(w, crate::model::GpuTensor::Quant { f16: Some(_), .. })
5692 {
5693 nf16 += 1;
5694 }
5695 }
5696 if e.build_q4_rp_swap(w)? {
5697 nswap += 1;
5698 }
5699 }
5700 }
5701 if let Ffn::Dense {
5702 ffn_gate,
5703 ffn_up,
5704 ffn_down,
5705 } = &mut layer.ffn
5706 {
5707 for w in [ffn_gate, ffn_up, ffn_down] {
5708 if f16_on {
5709 e.build_q8_f16(w)?;
5710 if matches!(w, crate::model::GpuTensor::Quant { f16: Some(_), .. })
5711 {
5712 nf16 += 1;
5713 }
5714 }
5715 if e.build_q4_rp_swap(w)? {
5716 nswap += 1;
5717 }
5718 }
5719 }
5720 }
5721 if nswap > 0 {
5722 eprintln!("[q4rp] split-plane IN-PLACE swap: {nswap} dense trunk tensors");
5723 }
5724 if nf16 > 0 {
5725 eprintln!("[q4f16] prefill fp16 mirrors built: {nf16} dense trunk tensors");
5726 }
5727 }
5728 }
5729 let model = HybridModel {
5730 cfg,
5731 plan,
5732 rewrite_qualifications: None,
5733 embd,
5734 output_norm,
5735 output,
5736 layers,
5737 mtp,
5738 mtp_extra,
5739 dflash_trim,
5740 embd_gpu: std::sync::OnceLock::new(),
5741 gemma4_aux,
5742 step35_aux,
5743 prime_slabs: std::sync::Mutex::new(std::collections::HashMap::new()),
5744 dspark_vgraphs: std::sync::Mutex::new(None),
5745 step_grouped_prefill: std::sync::Mutex::new(StepEpGroupedPrefill::default()),
5746 step35_token_graph: std::sync::Mutex::new(None),
5747 hyper,
5748 hyper_head,
5749 glm5_dflash,
5750 draft_state_bytes: std::sync::atomic::AtomicUsize::new(0),
5751 };
5752 e.configure_moe_cache_layout(model.moe_cache_block_sizes());
5753 if force_embd_gpu {
5754 let _ = model
5755 .embd_gpu
5756 .get_or_init(|| e.upload_u8(&model.embd.raw).expect("embed table upload"));
5757 }
5758 crate::pp::sync_stages_after_load(e, n_trunk)?;
5764 Ok(model)
5765 }
5766
5767 pub fn ensure_embed_resident(&self, e: &Engine) -> Result<(), Box<dyn std::error::Error>> {
5777 if std::env::var("MEMRA_EMBED_DEV").as_deref() == Ok("0") {
5778 return Ok(());
5779 }
5780 if self.embd_gpu.get().is_none() {
5781 let buf = e.upload_u8(&self.embd.raw)?;
5782 let _ = self.embd_gpu.set(buf); }
5784 Ok(())
5785 }
5786
5787 pub fn embed(
5788 &self,
5789 e: &Engine,
5790 tokens: &[u32],
5791 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
5792 let n_embd = self.cfg.n_embd as usize;
5793 if std::env::var("MEMRA_EMBED_DEV").as_deref() != Ok("0") {
5799 let tbl = self
5800 .embd_gpu
5801 .get_or_init(|| e.upload_u8(&self.embd.raw).expect("embed table upload"));
5802 let tok_d = e.htod_u32_v(tokens)?;
5803 let (qt, rb) = self.embd.qt_and_row_bytes(n_embd);
5804 return e.embed_gather_device_td(tbl, &tok_d, tokens.len(), n_embd, qt, rb);
5805 }
5806 let x = self.embd.gather(n_embd, tokens);
5807 e.htod(&x)
5808 }
5809}
5810
5811fn illegal_pipeline_cuts(fence: &[usize], legal_boundaries: &[usize]) -> Vec<usize> {
5812 fence
5813 .get(1..fence.len().saturating_sub(1))
5814 .unwrap_or_default()
5815 .iter()
5816 .copied()
5817 .filter(|cut| !legal_boundaries.contains(cut))
5818 .collect()
5819}
5820
5821#[cfg(test)]
5822mod pipeline_cut_tests {
5823 use super::illegal_pipeline_cuts;
5824
5825 #[test]
5826 fn manual_pipeline_cuts_cannot_bypass_model_plan_boundaries() {
5827 assert!(illegal_pipeline_cuts(&[0, 8, 16, 24], &[8, 16]).is_empty());
5828 assert_eq!(illegal_pipeline_cuts(&[0, 7, 16, 24], &[8, 16]), vec![7]);
5829 assert_eq!(
5830 illegal_pipeline_cuts(&[0, 7, 15, 24], &[8, 16]),
5831 vec![7, 15]
5832 );
5833 }
5834}
5835
5836#[cfg(test)]
5837mod auto_parallel_policy_tests {
5838 use super::{
5839 parse_auto_parallel_tp_attention, parse_auto_parallel_tp_attention_ranks,
5840 parse_auto_w4a16_bf16_mmv,
5841 };
5842
5843 #[test]
5844 fn automatic_w4a16_bf16_residency_defaults_on_with_explicit_rollback() {
5845 assert!(parse_auto_w4a16_bf16_mmv(None).unwrap());
5846 assert!(!parse_auto_w4a16_bf16_mmv(Some("0")).unwrap());
5847 assert!(parse_auto_w4a16_bf16_mmv(Some("1")).unwrap());
5848 assert!(parse_auto_w4a16_bf16_mmv(Some("true")).is_err());
5849 assert!(parse_auto_w4a16_bf16_mmv(Some("")).is_err());
5850 }
5851
5852 #[test]
5853 fn automatic_tp_attention_is_strict_and_defaults_off() {
5854 assert!(!parse_auto_parallel_tp_attention(None).unwrap());
5855 assert!(!parse_auto_parallel_tp_attention(Some("")).unwrap());
5856 assert!(!parse_auto_parallel_tp_attention(Some("0")).unwrap());
5857 assert!(parse_auto_parallel_tp_attention(Some("1")).unwrap());
5858 assert!(parse_auto_parallel_tp_attention(Some("true")).is_err());
5859 assert!(parse_auto_parallel_tp_attention(Some("2")).is_err());
5860 }
5861
5862 #[test]
5863 fn automatic_tp_attention_rank_count_is_explicit_and_bounded() {
5864 assert_eq!(parse_auto_parallel_tp_attention_ranks(None).unwrap(), None);
5865 assert_eq!(
5866 parse_auto_parallel_tp_attention_ranks(Some("2")).unwrap(),
5867 Some(2)
5868 );
5869 assert_eq!(
5870 parse_auto_parallel_tp_attention_ranks(Some("3")).unwrap(),
5871 Some(3)
5872 );
5873 assert_eq!(
5874 parse_auto_parallel_tp_attention_ranks(Some("4")).unwrap(),
5875 Some(4)
5876 );
5877 for bad in ["", "0", "1", "5", "all"] {
5878 assert!(parse_auto_parallel_tp_attention_ranks(Some(bad)).is_err());
5879 }
5880 }
5881}
5882
5883#[cfg(test)]
5884mod step_expert_selection_tests {
5885 use super::{
5886 StepExpertArtifact, StepExpertLayout, StepParallelLoadConfig, StepParallelRuntimeRegistry,
5887 StepTpAttentionPlacement, select_step_expert_layout, select_step_expert_layout_inner,
5888 };
5889 use crate::tp::StepEpLayerSpec;
5890
5891 fn spec(layer: usize, ranks: usize) -> StepEpLayerSpec {
5892 StepEpLayerSpec {
5893 layer,
5894 devices: (0..ranks).collect(),
5895 }
5896 }
5897
5898 #[test]
5899 fn tp2_keeps_projection_sharded_experts() {
5900 let selection = select_step_expert_layout(24, &[], &[spec(24, 2)])
5901 .unwrap()
5902 .unwrap();
5903 assert_eq!(selection.layout, StepExpertLayout::TensorParallel);
5904 assert!(selection.configured_by_tp);
5905 }
5906
5907 #[test]
5908 fn tp4_and_tp8_use_expert_ownership_without_a_second_flag() {
5909 for ranks in [4, 8] {
5910 let selection = select_step_expert_layout(24, &[], &[spec(24, ranks)])
5911 .unwrap()
5912 .unwrap();
5913 assert_eq!(selection.layout, StepExpertLayout::ExpertParallel);
5914 assert!(selection.configured_by_tp);
5915 assert_eq!(selection.spec.devices.len(), ranks);
5916 }
5917 }
5918
5919 #[test]
5920 fn explicit_ep_remains_expert_parallel() {
5921 let selection = select_step_expert_layout(24, &[spec(24, 2)], &[])
5922 .unwrap()
5923 .unwrap();
5924 assert_eq!(selection.layout, StepExpertLayout::ExpertParallel);
5925 assert!(!selection.configured_by_tp);
5926 }
5927
5928 #[test]
5929 fn conflicting_ep_and_tp_assignments_fail_closed() {
5930 let error = select_step_expert_layout(24, &[spec(24, 4)], &[spec(24, 4)]).unwrap_err();
5931 assert!(error.contains("cannot enable MEMRA_STEP_EP and MEMRA_STEP_TP together"));
5932 }
5933
5934 #[test]
5935 fn automatic_tp2_attention_can_overlap_ep4_expert_ownership() {
5936 let selection = select_step_expert_layout_inner(24, &[spec(24, 4)], &[spec(24, 2)], true)
5937 .unwrap()
5938 .unwrap();
5939 assert_eq!(selection.layout, StepExpertLayout::ExpertParallel);
5940 assert!(!selection.configured_by_tp);
5941 assert_eq!(selection.spec.devices, vec![0, 1, 2, 3]);
5942
5943 let error =
5944 select_step_expert_layout_inner(24, &[spec(24, 4)], &[spec(24, 2)], false).unwrap_err();
5945 assert!(error.contains("cannot enable MEMRA_STEP_EP and MEMRA_STEP_TP together"));
5946 }
5947
5948 #[test]
5949 fn runtime_registry_owns_one_immutable_load_snapshot() {
5950 let mut source_specs = vec![spec(24, 8)];
5951 let registry = StepParallelRuntimeRegistry::with_config(StepParallelLoadConfig {
5952 ep_specs: Vec::new(),
5953 tp_specs: source_specs.clone(),
5954 native_p2p: true,
5955 ep_device_arithmetic: true,
5956 f32_mirror: true,
5957 bulk_p2p: true,
5958 nvfp4_device_routes: true,
5959 auto_parallel: true,
5960 tp_attention_expert_overlap: false,
5961 expert_artifact: StepExpertArtifact::default(),
5962 });
5963 source_specs[0].devices.clear();
5964
5965 let stored = registry.tp_spec(24).unwrap();
5966 assert_eq!(stored.devices, (0..8).collect::<Vec<_>>());
5967 assert!(registry.config.native_p2p);
5968 assert!(registry.config.ep_device_arithmetic);
5969 assert!(registry.config.f32_mirror);
5970 assert!(registry.config.bulk_p2p);
5971 assert!(registry.config.nvfp4_device_routes);
5972 assert!(registry.config.auto_parallel);
5973 assert_eq!(
5974 registry.expert_selection(24).unwrap().unwrap().layout,
5975 StepExpertLayout::ExpertParallel
5976 );
5977
5978 let standalone = StepParallelRuntimeRegistry::default();
5979 assert!(standalone.tp_spec(24).is_none());
5980 assert!(!standalone.config.native_p2p);
5981 assert!(!standalone.config.ep_device_arithmetic);
5982 assert!(!standalone.config.f32_mirror);
5983 assert!(!standalone.config.bulk_p2p);
5984 }
5985
5986 #[test]
5987 fn rank_local_attention_uses_bounded_swa_rings_only_with_native_p2p() {
5988 assert_eq!(
5989 StepTpAttentionPlacement::resolve(true, None),
5990 StepTpAttentionPlacement::RankLocalGlobal
5991 );
5992 assert_eq!(
5993 StepTpAttentionPlacement::resolve(true, Some(512)),
5994 StepTpAttentionPlacement::RankLocalSwa
5995 );
5996 assert_eq!(
5997 StepTpAttentionPlacement::resolve(false, None),
5998 StepTpAttentionPlacement::OwnerTransportFallback
5999 );
6000 assert_eq!(
6001 StepTpAttentionPlacement::resolve(false, Some(512)),
6002 StepTpAttentionPlacement::OwnerSwa
6003 );
6004 }
6005}
6006
6007#[cfg(test)]
6008mod residency_tests {
6009 use super::{DevExpertFp8ProjectionScales, ResidentPlan, residency_bytes_by_device};
6010 use crate::model::HostExpertFp8BlockScales;
6011 use std::collections::HashMap;
6012
6013 #[test]
6014 fn pp_residency_counts_only_each_devices_expert_slice() {
6015 let tensors = [
6016 ("blk.0.ffn_gate_exps.weight", 10usize),
6017 ("blk.0.ffn_up_exps.weight", 20),
6018 ("blk.1.ffn_down_exps.weight", 30),
6019 ("blk.2.ffn_gate_exps.weight", 40),
6020 ("blk.3.ffn_up_exps.weight", 50),
6021 ("blk.0.attn_q.weight", 7),
6022 ("output.weight", 11),
6023 ];
6024 let bytes = residency_bytes_by_device(tensors, &[0, 0, 1, 1], 0);
6025 assert_eq!(bytes.experts.get(&0), Some(&60));
6026 assert_eq!(bytes.experts.get(&1), Some(&90));
6027 assert_eq!(bytes.rest, 18);
6028 assert!(bytes.saw_experts);
6029 }
6030
6031 #[test]
6032 fn pp_residency_combines_stages_that_share_one_device() {
6033 let tensors = [
6034 ("blk.0.ffn_gate_exps.weight", 10usize),
6035 ("blk.1.ffn_gate_exps.weight", 20),
6036 ("blk.2.ffn_gate_exps.weight", 30),
6037 ("blk.3.ffn_gate_exps.weight", 40),
6038 ];
6039 let bytes = residency_bytes_by_device(tensors, &[0, 0, 0, 0], 0);
6040 assert_eq!(bytes.experts.get(&0), Some(&100));
6041 assert_eq!(bytes.experts.len(), 1);
6042 }
6043
6044 #[test]
6045 fn distributed_trunk_layers_do_not_poison_local_mtp_residency_estimates() {
6046 let mut plan = ResidentPlan {
6047 primary_device: 0,
6048 layer_devices: vec![0; 81],
6049 layer_counts: HashMap::from([(0, 81)]),
6050 exact_expert_bytes: None,
6051 trunk_bytes: 0,
6052 decisions: HashMap::new(),
6053 pp: false,
6054 };
6055 plan.exclude_distributed_expert_layers(1..80);
6056 assert_eq!(plan.layer_counts.get(&0), Some(&2));
6057 }
6058
6059 #[test]
6060 fn resident_fp8_scale_slab_must_match_every_expert() {
6061 let valid = HostExpertFp8BlockScales {
6062 scales: vec![1.0; 12],
6063 rows: 2,
6064 cols: 3,
6065 expert_stride: 6,
6066 };
6067 DevExpertFp8ProjectionScales::validate(&valid, 2).unwrap();
6068
6069 let short = HostExpertFp8BlockScales {
6070 scales: vec![1.0; 11],
6071 ..valid
6072 };
6073 assert_eq!(
6074 DevExpertFp8ProjectionScales::validate(&short, 2).unwrap_err(),
6075 "block-E4M3 scale slab length mismatch: got 11, want 2x6=12"
6076 );
6077 }
6078
6079 #[test]
6080 fn resident_fp8_scale_stride_must_match_its_grid() {
6081 let invalid = HostExpertFp8BlockScales {
6082 scales: vec![1.0; 8],
6083 rows: 2,
6084 cols: 2,
6085 expert_stride: 0,
6086 };
6087 assert_eq!(
6088 DevExpertFp8ProjectionScales::validate(&invalid, 2).unwrap_err(),
6089 "block-E4M3 expert scale stride must be nonzero"
6090 );
6091 }
6092}
6093
6094#[cfg(test)]
6095mod draft_head_tests {
6096 use super::{draft_head_tensor, frspec_trim_own_head_name};
6097
6098 const STEP37_DRAFTER: &[&str] = &[
6105 "output.weight",
6106 "output_norm.weight",
6107 "token_embd.weight",
6108 "blk.45.nextn.shared_head_norm.weight",
6109 "blk.45.nextn.shared_head_head.weight",
6110 "blk.46.nextn.shared_head_head.weight",
6111 "blk.47.nextn.shared_head_head.weight",
6112 ];
6113
6114 fn present(names: &'static [&'static str]) -> impl Fn(&str) -> bool {
6115 move |t: &str| names.contains(&t)
6116 }
6117
6118 #[test]
6126 fn step37_drafter_prefers_the_blocks_own_nextn_head_over_file_level_output() {
6127 assert_eq!(
6128 draft_head_tensor(present(STEP37_DRAFTER), 45),
6129 "blk.45.nextn.shared_head_head.weight"
6130 );
6131 }
6132
6133 #[test]
6137 fn each_nextn_block_selects_its_own_head() {
6138 for n in 45..=47u32 {
6139 assert_eq!(
6140 draft_head_tensor(present(STEP37_DRAFTER), n),
6141 format!("blk.{n}.nextn.shared_head_head.weight")
6142 );
6143 }
6144 }
6145
6146 #[test]
6150 fn draft_without_a_nextn_head_falls_back_to_file_level_output() {
6151 let fr_spec: &[&str] = &["output.weight", "output_norm.weight", "d2t.weight"];
6152 assert_eq!(draft_head_tensor(present(fr_spec), 45), "output.weight");
6153 }
6154
6155 #[test]
6160 fn legacy_shared_head_is_probed_but_loses_to_shared_head_head() {
6161 let legacy_only: &[&str] = &["output.weight", "blk.45.nextn.shared_head.weight"];
6162 assert_eq!(
6163 draft_head_tensor(present(legacy_only), 45),
6164 "blk.45.nextn.shared_head.weight"
6165 );
6166
6167 let both: &[&str] = &[
6168 "output.weight",
6169 "blk.45.nextn.shared_head.weight",
6170 "blk.45.nextn.shared_head_head.weight",
6171 ];
6172 assert_eq!(
6173 draft_head_tensor(present(both), 45),
6174 "blk.45.nextn.shared_head_head.weight"
6175 );
6176 }
6177
6178 #[test]
6182 fn a_different_blocks_nextn_head_is_never_borrowed() {
6183 let wrong_block: &[&str] = &[
6184 "output.weight",
6185 "blk.46.nextn.shared_head_head.weight",
6186 "blk.47.nextn.shared_head_head.weight",
6187 ];
6188 assert_eq!(draft_head_tensor(present(wrong_block), 45), "output.weight");
6189 }
6190
6191 #[test]
6196 fn frspec_trim_prefers_the_nextn_blocks_own_head_name() {
6197 assert_eq!(
6198 frspec_trim_own_head_name(45),
6199 "blk.45.nextn.shared_head_head.weight"
6200 );
6201 assert_eq!(
6203 frspec_trim_own_head_name(45),
6204 format!("blk.{}.nextn.shared_head_head.weight", 45)
6205 );
6206 assert_eq!(
6207 frspec_trim_own_head_name(40),
6208 "blk.40.nextn.shared_head_head.weight"
6209 );
6210 }
6211}