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 expert_artifact: StepExpertArtifact,
105}
106
107#[derive(Default)]
108pub(crate) struct StepParallelRuntimeRegistry {
109 config: StepParallelLoadConfig,
110 runtimes: HashMap<(Vec<usize>, bool, bool, bool), Arc<crate::tp::TpE4m3HostBounce>>,
111}
112
113#[derive(Clone, Copy, Debug, PartialEq, Eq)]
114enum StepExpertLayout {
115 TensorParallel,
116 ExpertParallel,
117}
118
119#[derive(Clone, Debug, PartialEq, Eq)]
120struct StepExpertSelection {
121 spec: crate::tp::StepEpLayerSpec,
122 layout: StepExpertLayout,
123 configured_by_tp: bool,
124}
125
126fn select_step_expert_layout(
127 layer: usize,
128 ep_specs: &[crate::tp::StepEpLayerSpec],
129 tp_specs: &[crate::tp::StepTpLayerSpec],
130) -> Result<Option<StepExpertSelection>, String> {
131 let ep = ep_specs.iter().find(|spec| spec.layer == layer);
132 let tp = tp_specs.iter().find(|spec| spec.layer == layer);
133 if ep.is_some() && tp.is_some() {
134 return Err(format!(
135 "Step layer {layer} cannot enable MEMRA_STEP_EP and MEMRA_STEP_TP together"
136 ));
137 }
138 Ok(match (ep, tp) {
139 (Some(spec), None) => Some(StepExpertSelection {
140 spec: spec.clone(),
141 layout: StepExpertLayout::ExpertParallel,
142 configured_by_tp: false,
143 }),
144 (None, Some(spec)) => Some(StepExpertSelection {
145 spec: spec.clone(),
146 layout: if spec.devices.len() > 2 {
147 StepExpertLayout::ExpertParallel
148 } else {
149 StepExpertLayout::TensorParallel
150 },
151 configured_by_tp: true,
152 }),
153 (None, None) => None,
154 (Some(_), Some(_)) => unreachable!(),
155 })
156}
157
158impl StepParallelRuntimeRegistry {
159 fn with_config(config: StepParallelLoadConfig) -> Self {
160 Self {
161 config,
162 runtimes: HashMap::new(),
163 }
164 }
165
166 fn tp_spec(&self, layer: usize) -> Option<&crate::tp::StepTpLayerSpec> {
167 self.config.tp_specs.iter().find(|spec| spec.layer == layer)
168 }
169
170 fn expert_selection(&self, layer: usize) -> Result<Option<StepExpertSelection>, String> {
171 select_step_expert_layout(layer, &self.config.ep_specs, &self.config.tp_specs)
172 }
173
174 fn runtime(
175 &mut self,
176 devices: &[usize],
177 native_p2p: bool,
178 ep_device_arithmetic: bool,
179 ) -> Result<Arc<crate::tp::TpE4m3HostBounce>, Box<dyn std::error::Error>> {
180 let bulk_p2p = self.config.bulk_p2p && native_p2p;
181 let key = (devices.to_vec(), native_p2p, ep_device_arithmetic, bulk_p2p);
182 if let Some(runtime) = self.runtimes.get(&key) {
183 return Ok(Arc::clone(runtime));
184 }
185 let runtime = Arc::new(crate::tp::TpE4m3HostBounce::new_configured(
186 devices,
187 native_p2p,
188 ep_device_arithmetic,
189 bulk_p2p,
190 )?);
191 let names = runtime.device_names()?;
192 if names
193 .iter()
194 .any(|name| !name.contains("RTX PRO 6000") || !name.contains("Blackwell"))
195 {
196 return Err(format!(
197 "Step distributed execution is qualified only on RTX PRO 6000 Blackwell, \
198 got {names:?}"
199 )
200 .into());
201 }
202 self.runtimes.insert(key, Arc::clone(&runtime));
203 Ok(runtime)
204 }
205}
206
207impl ResidentPlan {
208 fn from_layout(
209 src: &dyn TensorSource,
210 primary_device: usize,
211 layer_devices: Vec<usize>,
212 pp: bool,
213 ) -> Self {
214 let mut layer_counts = HashMap::new();
215 for &device in &layer_devices {
216 *layer_counts.entry(device).or_default() += 1;
217 }
218 let (exact_expert_bytes, trunk_bytes) = match src.gguf() {
219 Some(g) => {
220 let bytes = residency_bytes_by_device(
221 g.tensors
222 .iter()
223 .map(|t| (t.name.as_str(), t.n_bytes as usize)),
224 &layer_devices,
225 primary_device,
226 );
227 if bytes.saw_experts {
228 (Some(bytes.experts), bytes.rest)
229 } else {
230 (None, 0)
231 }
232 }
233 None => (None, 0),
234 };
235 Self {
236 primary_device,
237 layer_devices,
238 layer_counts,
239 exact_expert_bytes,
240 trunk_bytes,
241 decisions: HashMap::new(),
242 pp,
243 }
244 }
245
246 pub(crate) fn unsharded(e: &Engine, src: &dyn TensorSource, cfg: &ModelConfig) -> Self {
247 let device = e.ctx().ordinal();
248 Self::from_layout(src, device, vec![device; cfg.n_layer as usize], false)
249 }
250
251 pub(crate) fn pp(
252 e: &Engine,
253 src: &dyn TensorSource,
254 cfg: &ModelConfig,
255 n_trunk: usize,
256 ) -> Result<Self, Box<dyn std::error::Error>> {
257 let primary = e.ctx().ordinal();
258 let Some(_fence) = crate::pp::pp_cuts(n_trunk) else {
259 return Ok(Self::unsharded(e, src, cfg));
260 };
261 let mut layer_devices = vec![primary; cfg.n_layer as usize];
262 for (il, device) in layer_devices.iter_mut().take(n_trunk).enumerate() {
263 *device = crate::pp::layer_engine(e, n_trunk, il)?.ctx().ordinal();
264 }
265 Ok(Self::from_layout(src, primary, layer_devices, true))
266 }
267
268 fn exclude_distributed_expert_layers(&mut self, specs: impl IntoIterator<Item = usize>) {
272 for layer in specs {
273 let device = self
274 .layer_devices
275 .get(layer)
276 .copied()
277 .unwrap_or(self.primary_device);
278 if let Some(count) = self.layer_counts.get_mut(&device) {
279 *count = count.saturating_sub(1);
280 }
281 }
282 }
283
284 fn should_reside(&mut self, e: &Engine, il: usize, per_layer: usize) -> bool {
285 let device = self
286 .layer_devices
287 .get(il)
288 .copied()
289 .unwrap_or(self.primary_device);
290 debug_assert_eq!(e.ctx().ordinal(), device);
291 if let Some(&decision) = self.decisions.get(&device) {
292 return decision;
293 }
294 if std::env::var("MEMRA_MOE_RESIDENT").as_deref() == Ok("0") {
295 self.decisions.insert(device, false);
296 return false;
297 }
298 let (free, _total) = match e.ctx().mem_get_info() {
299 Ok(v) => v,
300 Err(_) => {
301 self.decisions.insert(device, false);
302 return false;
303 }
304 };
305 let projected = self
306 .exact_expert_bytes
307 .as_ref()
308 .map(|bytes| bytes.get(&device).copied().unwrap_or(0))
309 .unwrap_or(per_layer * self.layer_counts.get(&device).copied().unwrap_or(1));
310 let budget = std::env::var("MEMRA_MOE_RESIDENT_GB")
311 .ok()
312 .and_then(|v| v.parse::<f64>().ok())
313 .map(|gb| (gb * 1e9) as usize)
314 .unwrap_or_else(|| {
315 let reserve = std::env::var("MEMRA_MOE_RESIDENT_HEADROOM_GB")
316 .ok()
317 .and_then(|v| v.parse::<f64>().ok())
318 .map(|gb| (gb * 1e9) as usize)
319 .unwrap_or(2_000_000_000);
320 free.saturating_sub(self.trunk_bytes + reserve)
321 });
322 let ok = projected <= budget;
323 eprintln!(
324 "[moe] resident-experts decision ({}dev{}): experts {:.2}GB + trunk {:.2}GB vs free {:.2}GB (expert budget {:.2}GB) -> {}",
325 if self.pp { "PP " } else { "" },
326 device,
327 projected as f64 / 1e9,
328 self.trunk_bytes as f64 / 1e9,
329 free as f64 / 1e9,
330 budget as f64 / 1e9,
331 if ok { "RESIDENT" } else { "SLRU cache" }
332 );
333 self.decisions.insert(device, ok);
334 ok
335 }
336}
337
338fn load_mixer_kind(
340 e: &Engine,
341 src: &dyn TensorSource,
342 cfg: &ModelConfig,
343 il: u32,
344 attention: &AttentionPlan,
345 step_runtimes: &mut StepParallelRuntimeRegistry,
346) -> Result<Mixer, Box<dyn std::error::Error>> {
347 let p = |s: &str| format!("blk.{il}.{s}");
348 Ok(match attention {
349 AttentionPlan::Mla(mla) => Mixer::Mla(MlaAttnLayer::load(e, src, il, mla)?),
350 AttentionPlan::Full(full)
351 | AttentionPlan::SlidingWindow {
352 attention: full, ..
353 } => {
354 Mixer::Full(FullAttnLayer {
355 wq: load_t(e, src, &p("attn_q.weight"))?,
356 wk: load_t(e, src, &p("attn_k.weight"))?,
357 wv: match load_opt(e, src, &p("attn_v.weight"))? {
362 Some(v) => v,
363 None => load_t(e, src, &p("attn_k.weight"))?,
364 },
365 wo: load_t(e, src, &p("attn_output.weight"))?,
366 q_norm: load_t(e, src, &p("attn_q_norm.weight"))?,
367 k_norm: load_t(e, src, &p("attn_k_norm.weight"))?,
368 attn_gate: if full.output_gate
372 == memra_gguf::config::AttentionGateKind::SeparateHead
373 {
374 Some(load_t(e, src, &p("attn_gate.weight"))?)
375 } else {
376 None
377 },
378 step_tp_qkv: build_step_tp_qkv(e, src, cfg, il as usize, step_runtimes)?,
379 })
380 }
381 AttentionPlan::KimiDeltaNet(kda) => {
384 Mixer::Kda(crate::kda::KdaAttnLayer::load(e, src, il, kda)?)
385 }
386 AttentionPlan::GatedDeltaNet(geometry) => Mixer::Linear(LinearAttnLayer {
387 geometry: *geometry,
388 wqkv: load_t(e, src, &p("attn_qkv.weight"))?,
389 wqkv_gate: load_t(e, src, &p("attn_gate.weight"))?,
390 ssm_beta: load_t(e, src, &p("ssm_beta.weight"))?,
391 ssm_alpha: load_t(e, src, &p("ssm_alpha.weight"))?,
392 ssm_a: load_t(e, src, &p("ssm_a"))?,
393 ssm_dt: load_t(e, src, &p("ssm_dt.bias"))?,
394 ssm_conv1d: load_t(e, src, &p("ssm_conv1d.weight"))?,
395 ssm_norm: load_t(e, src, &p("ssm_norm.weight"))?,
396 ssm_out: load_t(e, src, &p("ssm_out.weight"))?,
397 }),
398 })
399}
400
401#[allow(clippy::too_many_arguments)] pub(crate) fn load_ffn(
409 e: &Engine,
410 src: &dyn TensorSource,
411 cfg: &ModelConfig,
412 mlp: &MlpPlan,
413 il: u32,
414 spill: Option<(&GgufFile, &mut crate::spill::SpillCtx)>,
415 resident: &mut ResidentPlan,
416 step_runtimes: &mut StepParallelRuntimeRegistry,
417) -> Result<Ffn, Box<dyn std::error::Error>> {
418 let p = |s: &str| format!("blk.{il}.{s}");
419 let artifact_dense = matches!(mlp, MlpPlan::Moe(_))
426 && !src.has(&p("ffn_gate_exps.weight"))
427 && !src.has(&p("ffn_gate_up_exps.weight"))
428 && src.has(&p("ffn_gate.weight"));
429 Ok(if artifact_dense {
430 Ffn::Dense {
431 ffn_gate: GpuTensor::load_from_source(e, src, &p("ffn_gate.weight"))?,
432 ffn_up: GpuTensor::load_from_source(e, src, &p("ffn_up.weight"))?,
433 ffn_down: GpuTensor::load_from_source(e, src, &p("ffn_down.weight"))?,
434 }
435 } else if let MlpPlan::Moe(moe) = mlp {
436 let n_expert = moe.expert_count as usize;
437 let (gate_exps, up_exps, down_exps) = match spill {
443 Some((g, ctx)) => (
444 HostExps::load_tiered(e, g, &p("ffn_gate_exps.weight"), ctx)?,
445 HostExps::load_tiered(e, g, &p("ffn_up_exps.weight"), ctx)?,
446 HostExps::load_tiered(e, g, &p("ffn_down_exps.weight"), ctx)?,
447 ),
448 None => {
449 let exps = |e: &Engine, n: &str| -> Result<HostExps, Box<dyn std::error::Error>> {
450 if src.has(n) {
451 HostExps::load_stacked_from_source(e, src, n)
452 } else {
453 HostExps::load_from_source(e, src, n, n_expert)
454 }
455 };
456 let fused = p("ffn_gate_up_exps.weight");
458 if !src.has(&p("ffn_gate_exps.weight")) && src.has(&fused) {
459 let ff = moe.expert_intermediate_size as usize;
460 (
461 HostExps::load_stacked_split_from_source(e, src, &fused, 0, ff)?,
462 HostExps::load_stacked_split_from_source(e, src, &fused, ff, 2 * ff)?,
463 exps(e, &p("ffn_down_exps.weight"))?,
464 )
465 } else {
466 (
467 exps(e, &p("ffn_gate_exps.weight"))?,
468 exps(e, &p("ffn_up_exps.weight"))?,
469 exps(e, &p("ffn_down_exps.weight"))?,
470 )
471 }
472 }
473 };
474 let (step_ep, step_tp) = build_step_distributed_exps(
475 e,
476 cfg,
477 src,
478 il as usize,
479 &gate_exps,
480 &up_exps,
481 &down_exps,
482 step_runtimes,
483 )?;
484 let dev_exps = if step_ep.is_some() || step_tp.is_some() {
490 None
491 } else {
492 build_dev_exps(e, resident, il as usize, &gate_exps, &up_exps, &down_exps)?
493 };
494 let mut macro_row = vec![1.0f32; 3 * n_expert];
496 for (slot, exps) in [(0usize, &gate_exps), (1, &up_exps), (2, &down_exps)] {
497 if let Some(ms) = exps.macros.as_ref() {
498 macro_row[slot * n_expert..(slot + 1) * n_expert].copy_from_slice(ms);
499 }
500 }
501 let has_macros = macro_row.iter().any(|&m| m != 1.0);
502 let dev_macros = e.htod(¯o_row)?;
503 let exp_probs_b = src
506 .find(&p("exp_probs_b.bias"))
507 .map(|v| memra_gguf::dequant::dequantize(v.ggml_type, &v.bytes, n_expert));
508 if exp_probs_b.is_none()
517 && matches!(
518 moe.router,
519 memra_gguf::model_plan::RouterPlan::Sigmoid {
520 selection_bias: true,
521 ..
522 } | memra_gguf::model_plan::RouterPlan::SqrtSoftplus {
523 selection_bias: true,
524 ..
525 }
526 )
527 {
528 return Err(format!(
529 "layer {il}: {} is absent, but the compiled ModelPlan declares a router with a \
530 selection bias ({:?}). Refusing to load: a zero-filled bias would route to \
531 different experts than this model does, silently. Either the checkpoint does \
532 not carry the tensor, or this arch has no `exp_probs_b.bias` entry in \
533 hf_mapping's ggml->HF map",
534 p("exp_probs_b.bias"),
535 moe.router
536 )
537 .into());
538 }
539 let active_experts = src.active_experts(il).map(<[bool]>::to_vec);
540 let route_bias = exp_probs_b.clone().unwrap_or_else(|| vec![0.0; n_expert]);
541 let active_row: Vec<u8> = active_experts
542 .as_ref()
543 .map(|mask| mask.iter().map(|&is_active| u8::from(is_active)).collect())
544 .unwrap_or_else(|| vec![1; n_expert]);
545 let exp_probs_b_dev = e.htod(&route_bias)?;
546 let active_experts_dev = e.htod_bytes(&active_row)?;
547 let gate_shexp = load_opt(e, src, &p("ffn_gate_shexp.weight"))?;
548 let up_shexp = load_opt(e, src, &p("ffn_up_shexp.weight"))?;
549 let down_shexp = load_opt(e, src, &p("ffn_down_shexp.weight"))?;
550 if moe.shared.is_some()
558 && (gate_shexp.is_none() || up_shexp.is_none() || down_shexp.is_none())
559 {
560 return Err(format!(
561 "layer {il}: the compiled ModelPlan declares an always-on shared expert, but \
562 {}{}{} could not be resolved in the checkpoint. Refusing to load: dropping the \
563 shared branch computes a different model, silently. Either the checkpoint does \
564 not carry it, or this arch's shared-expert spelling is missing from \
565 hf_mapping's ggml->HF map",
566 if gate_shexp.is_none() {
567 format!("{} ", p("ffn_gate_shexp.weight"))
568 } else {
569 String::new()
570 },
571 if up_shexp.is_none() {
572 format!("{} ", p("ffn_up_shexp.weight"))
573 } else {
574 String::new()
575 },
576 if down_shexp.is_none() {
577 p("ffn_down_shexp.weight")
578 } else {
579 String::new()
580 },
581 )
582 .into());
583 }
584 Ffn::Moe(MoeWeights {
585 gate_inp: load_t(e, src, &p("ffn_gate_inp.weight"))?,
586 gate_inp_shexp: load_opt(e, src, &p("ffn_gate_inp_shexp.weight"))?,
587 exp_probs_b,
588 exp_probs_b_dev,
589 active_experts,
590 active_experts_dev,
591 gate_exps,
592 up_exps,
593 down_exps,
594 gate_shexp,
595 up_shexp,
596 down_shexp,
597 dev_exps,
598 step_ep,
599 step_tp,
600 glm5_ep: None,
601 dev_macros,
602 has_macros,
603 w4a16_bf16_activations: matches!(
604 src.expert_activation_precision(),
605 memra_gguf::source::ExpertActivationPrecision::Bf16
606 ),
607 })
608 } else {
609 Ffn::Dense {
610 ffn_gate: load_t(e, src, &p("ffn_gate.weight"))?,
611 ffn_up: load_t(e, src, &p("ffn_up.weight"))?,
612 ffn_down: load_t(e, src, &p("ffn_down.weight"))?,
613 }
614 })
615}
616
617fn host_e4m3_bank(
618 exps: &HostExps,
619) -> Result<crate::tp::E4m3ExpertBank<'_>, Box<dyn std::error::Error>> {
620 if exps.qtype != crate::QT_F8_E4M3_BLK {
621 return Err(format!(
622 "Step EP requires native block-E4M3 expert banks, got qtype {}",
623 exps.qtype
624 )
625 .into());
626 }
627 let scales = exps
628 .fp8_blk
629 .as_ref()
630 .ok_or("Step EP native expert bank has no block-E4M3 scale plane")?;
631 Ok(crate::tp::E4m3ExpertBank {
632 codes: exps.bytes.as_bytes(),
633 scales: &scales.scales,
634 expert_count: exps.n_expert,
635 out_features: exps.out_f,
636 in_features: exps.in_f,
637 })
638}
639
640fn validate_step_expert_specs(
641 contract: &crate::parallel::ModelParallelContract,
642 flag: &str,
643 specs: &[crate::tp::StepEpLayerSpec],
644 allow_dense_attention_only: bool,
645) -> Result<(), Box<dyn std::error::Error>> {
646 for candidate in specs {
647 if candidate.layer >= contract.trunk_layers {
648 return Err(format!(
649 "{flag} layer {} is outside Step trunk layers 0..{}",
650 candidate.layer, contract.trunk_layers
651 )
652 .into());
653 }
654 if candidate.layer < contract.dense_prefix_layers {
655 if allow_dense_attention_only {
656 continue;
657 }
658 return Err(format!(
659 "{flag} layer {} is outside Step routed-expert layers {}..{}",
660 candidate.layer, contract.dense_prefix_layers, contract.trunk_layers
661 )
662 .into());
663 }
664 }
665 Ok(())
666}
667
668fn validate_step_expert_activation_layout(
669 cfg: &ModelConfig,
670 flag: &str,
671 selection: &StepExpertSelection,
672) -> Result<(), Box<dyn std::error::Error>> {
673 let _ = (cfg, flag, selection);
679 Ok(())
680}
681
682fn parse_auto_w4a16_bf16_mmv(value: Option<&str>) -> Result<bool, String> {
683 match value {
684 None => Ok(true),
685 Some("0") => Ok(false),
686 Some("1") => Ok(true),
687 Some(value) => Err(format!(
688 "MEMRA_BF16_MMV={value:?} is invalid under MEMRA_PARALLEL=auto; expected 0 or 1"
689 )),
690 }
691}
692
693fn parse_auto_parallel_tp_attention(value: Option<&str>) -> Result<bool, String> {
694 match value {
695 None | Some("") | Some("0") => Ok(false),
696 Some("1") => Ok(true),
697 Some(value) => Err(format!(
698 "MEMRA_PARALLEL_TP_ATTENTION={value:?} is invalid; expected 0 or 1"
699 )),
700 }
701}
702
703fn auto_parallel_tp_attention_enabled() -> Result<bool, String> {
704 parse_auto_parallel_tp_attention(std::env::var("MEMRA_PARALLEL_TP_ATTENTION").ok().as_deref())
705}
706
707fn prepare_auto_parallel(
713 src: &dyn TensorSource,
714 cfg: &ModelConfig,
715 plan: &memra_gguf::model_plan::ModelPlan,
716) -> Result<Option<crate::parallel::AutoParallelPlacement>, Box<dyn std::error::Error>> {
717 let Some(devices) = crate::tp::auto_parallel_devices()? else {
718 return Ok(None);
719 };
720 if std::env::var_os("MEMRA_PP_STAGES").is_some()
721 || std::env::var_os("MEMRA_PP_DEVICES").is_some()
722 || std::env::var_os("MEMRA_PP_SPLITS").is_some()
723 {
724 return Err(
725 "MEMRA_PARALLEL=auto cannot be combined with MEMRA_PP_STAGES, MEMRA_PP_DEVICES, or \
726 MEMRA_PP_SPLITS"
727 .into(),
728 );
729 }
730 let placement = crate::parallel::plan_auto_parallel(src, cfg, plan, &devices)?;
731 let auto_w4a16_bf16 = placement.backend == crate::parallel::AutoParallelBackend::ExpertParallel
732 && matches!(
733 src.expert_activation_precision(),
734 memra_gguf::source::ExpertActivationPrecision::Bf16
735 );
736 let bf16_nonexpert = if auto_w4a16_bf16 {
737 let explicit = match std::env::var("MEMRA_BF16_MMV") {
738 Ok(value) => Some(value),
739 Err(std::env::VarError::NotPresent) => None,
740 Err(error) => return Err(format!("cannot read MEMRA_BF16_MMV: {error}").into()),
741 };
742 let enabled = parse_auto_w4a16_bf16_mmv(explicit.as_deref())?;
743 if enabled && explicit.is_none() {
744 unsafe {
747 std::env::set_var("MEMRA_BF16_MMV", "1");
748 }
749 }
750 match (enabled, explicit.is_some()) {
751 (true, false) => "bf16-resident(auto)",
752 (true, true) => "bf16-resident(explicit)",
753 (false, true) => "f32-expanded(explicit-rollback)",
754 (false, false) => unreachable!("unset auto W4A16 defaults BF16 residency on"),
755 }
756 } else {
757 "placement-default"
758 };
759 if placement.backend == crate::parallel::AutoParallelBackend::Pipeline {
760 let stages = placement.devices.len();
761 let device_list = placement
762 .devices
763 .iter()
764 .map(usize::to_string)
765 .collect::<Vec<_>>()
766 .join(",");
767 let splits = placement
768 .pipeline_splits
769 .iter()
770 .map(usize::to_string)
771 .collect::<Vec<_>>()
772 .join(",");
773 unsafe {
776 std::env::set_var("MEMRA_PP_STAGES", stages.to_string());
777 std::env::set_var("MEMRA_PP_DEVICES", &device_list);
778 std::env::set_var("MEMRA_PP_SPLITS", &splits);
779 }
780 }
781 let family = if placement.routed_layers.is_empty() {
782 "dense-transformer"
783 } else {
784 "routed-moe"
785 };
786 eprintln!(
787 "[parallel-auto] family={family} variant={:?} devices={:?} placement={} \
788 checkpoint_peak={:.2}GB ep_root={:.2}GB ep_peer={:.2}GB reserve={:.2}GB \
789 capacity={:?} splits={:?} bf16_nonexpert={bf16_nonexpert} \
790 wavefront=off(default) performance_claim=false",
791 cfg.name,
792 placement.devices,
793 match placement.backend {
794 crate::parallel::AutoParallelBackend::Pipeline => "pipeline",
795 crate::parallel::AutoParallelBackend::ExpertParallel => "expert-parallel",
796 },
797 placement.checkpoint_peak_bytes as f64 / 1e9,
798 placement.expert_root_bytes as f64 / 1e9,
799 placement.expert_peer_bytes as f64 / 1e9,
800 placement.reserve_bytes as f64 / 1e9,
801 placement.device_capacity_bytes,
802 placement.pipeline_splits,
803 );
804 Ok(Some(placement))
805}
806
807fn prepare_step_parallel_load(
808 e: &Engine,
809 src: &dyn TensorSource,
810 cfg: &ModelConfig,
811 trunk_layers: usize,
812 auto_placement: Option<&crate::parallel::AutoParallelPlacement>,
813) -> Result<StepParallelLoadConfig, Box<dyn std::error::Error>> {
814 let mut tp_specs = crate::tp::step_tp_layer_specs()?;
815 let mut ep_specs = crate::tp::step_ep_layer_specs()?;
816 let device_arithmetic = crate::tp::step_ep_device_arithmetic_enabled()?;
817 let f32_mirror = crate::tp::step_tp_f32_mirror_enabled()?;
818 let bulk_p2p = crate::tp::step_tp_bulk_p2p_enabled()?;
819 let mut native_p2p = crate::tp::step_tp_native_p2p_enabled()?;
820 let mut nvfp4_device_routes = crate::tp::step_nvfp4_dev_routes_enabled()?;
821 let auto_tp_attention = auto_parallel_tp_attention_enabled()?;
822 let mut auto_parallel = false;
823 if auto_tp_attention && auto_placement.is_none() {
824 return Err(
825 "MEMRA_PARALLEL_TP_ATTENTION=1 requires MEMRA_PARALLEL=auto; explicit per-layer \
826 recipes remain under MEMRA_STEP_TP"
827 .into(),
828 );
829 }
830 if let Some(placement) = auto_placement {
831 if !tp_specs.is_empty() || !ep_specs.is_empty() {
832 return Err(
833 "MEMRA_PARALLEL=auto cannot be combined with MEMRA_STEP_EP or MEMRA_STEP_TP".into(),
834 );
835 }
836 if placement.backend == crate::parallel::AutoParallelBackend::Pipeline {
837 if auto_tp_attention {
838 return Err(
839 "MEMRA_PARALLEL_TP_ATTENTION=1 requires automatic whole-expert EP; the \
840 selected checkpoint fits only the pipeline backend"
841 .into(),
842 );
843 }
844 return Ok(StepParallelLoadConfig::default());
845 }
846 if auto_tp_attention {
847 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
848 if !contract.tensor_attention_supported {
849 return Err(format!(
850 "MEMRA_PARALLEL_TP_ATTENTION=1 cannot shard attention for {:?}: the \
851 compiled ModelPlan has no generic tensor-attention contract",
852 cfg.name
853 )
854 .into());
855 }
856 tp_specs = (0..trunk_layers)
857 .map(|layer| crate::tp::StepTpLayerSpec {
858 layer,
859 devices: placement.devices.clone(),
860 })
861 .collect();
862 ep_specs.clear();
863 } else {
864 ep_specs = placement
865 .routed_layers
866 .iter()
867 .map(|&layer| crate::tp::StepEpLayerSpec {
868 layer,
869 devices: placement.devices.clone(),
870 })
871 .collect();
872 }
873 auto_parallel = true;
874 native_p2p = true;
875 nvfp4_device_routes = matches!(
876 src.expert_activation_precision(),
877 memra_gguf::source::ExpertActivationPrecision::Bf16
878 );
879 eprintln!(
880 "[parallel-auto-backend] devices={:?} routed_layers={} native_p2p=true \
881 artifact_activation={:?} attention_layout={} expert_layout=expert-parallel \
882 backend={} performance_claim=false",
883 placement.devices,
884 placement.routed_layers.len(),
885 src.expert_activation_precision(),
886 if auto_tp_attention {
887 "tensor-parallel"
888 } else {
889 "root-local"
890 },
891 if nvfp4_device_routes {
892 "nvfp4-w4a16"
893 } else {
894 "artifact-selected-host-oracle"
895 },
896 );
897 }
898 if tp_specs.is_empty() {
899 if auto_tp_attention {
900 return Err("MEMRA_PARALLEL_TP_ATTENTION=1 produced no tensor-parallel layers".into());
901 }
902 if device_arithmetic || f32_mirror || bulk_p2p {
903 return Err(
904 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1, MEMRA_STEP_TP_F32_MIRROR=1, or \
905 MEMRA_STEP_TP_BULK_P2P=1 requires MEMRA_STEP_TP; device arithmetic and bulk \
906 transport also require MEMRA_STEP_TP_NATIVE_P2P=1"
907 .into(),
908 );
909 }
910 if nvfp4_device_routes && ep_specs.is_empty() {
911 return Err(
912 "MEMRA_STEP_NVFP4_DEV_ROUTES=1 requires MEMRA_STEP_EP or MEMRA_STEP_TP".into(),
913 );
914 }
915 if nvfp4_device_routes && !native_p2p {
916 return Err("MEMRA_STEP_NVFP4_DEV_ROUTES=1 with explicit EP requires \
917 MEMRA_STEP_TP_NATIVE_P2P=1"
918 .into());
919 }
920 let expert_artifact = if ep_specs.is_empty() {
923 StepExpertArtifact::default()
924 } else if nvfp4_device_routes
925 && matches!(
926 src.expert_activation_precision(),
927 memra_gguf::source::ExpertActivationPrecision::Bf16
928 )
929 {
930 StepExpertArtifact::Nvfp4
935 } else {
936 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
937 validate_step_expert_specs(&contract, "MEMRA_STEP_EP", &ep_specs, false)?;
938 let layer_owners = (0..trunk_layers)
939 .map(|layer| {
940 crate::pp::layer_engine(e, trunk_layers, layer)
941 .map(|engine| engine.ctx().ordinal())
942 })
943 .collect::<Result<Vec<_>, _>>()?;
944 let mut runtime_groups = Vec::<Vec<usize>>::new();
945 for spec in &ep_specs {
946 let owner = layer_owners[spec.layer];
947 if !spec.devices.contains(&owner) {
948 return Err(format!(
949 "MEMRA_STEP_EP layer {} owning device {owner} is absent from {:?}",
950 spec.layer, spec.devices
951 )
952 .into());
953 }
954 if nvfp4_device_routes && spec.devices.first().copied() != Some(owner) {
955 return Err(format!(
956 "MEMRA_STEP_NVFP4_DEV_ROUTES=1 requires the owning device first; \
957 layer {} owner={owner} devices={:?}",
958 spec.layer, spec.devices
959 )
960 .into());
961 }
962 if !runtime_groups.contains(&spec.devices) {
963 runtime_groups.push(spec.devices.clone());
964 }
965 }
966 for devices in &runtime_groups {
967 let hardware = crate::parallel::detect_uniform_hardware(devices)?;
968 if !contract.hardware_targets.contains(&hardware) {
969 return Err(format!(
970 "{} has no qualified {hardware:?} EP contract for devices {devices:?}",
971 contract.variant
972 )
973 .into());
974 }
975 }
976 let artifact = match crate::parallel::validate_fp8_expert_checkpoint(src, &contract) {
977 Ok(_) => StepExpertArtifact::E4m3,
978 Err(fp8_error) => {
979 match crate::parallel::validate_nvfp4_expert_checkpoint(src, &contract) {
980 Ok(_) => StepExpertArtifact::Nvfp4,
981 Err(nvfp4_error) => {
982 return Err(format!(
983 "Step checkpoint qualifies as neither native expert artifact \
984 class: [E4M3] {fp8_error} [NVFP4] {nvfp4_error}"
985 )
986 .into());
987 }
988 }
989 }
990 };
991 if nvfp4_device_routes && artifact != StepExpertArtifact::Nvfp4 {
992 return Err(
993 "MEMRA_STEP_NVFP4_DEV_ROUTES=1 requires a native ModelOpt NVFP4 expert \
994 artifact"
995 .into(),
996 );
997 }
998 artifact
999 };
1000 return Ok(StepParallelLoadConfig {
1001 ep_specs,
1002 native_p2p,
1003 nvfp4_device_routes,
1004 auto_parallel,
1005 expert_artifact,
1006 ..StepParallelLoadConfig::default()
1007 });
1008 }
1009 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
1010 validate_step_expert_specs(&contract, "MEMRA_STEP_EP", &ep_specs, false)?;
1011 validate_step_expert_specs(&contract, "MEMRA_STEP_TP", &tp_specs, true)?;
1012 for spec in &tp_specs {
1013 let selection = select_step_expert_layout(spec.layer, &ep_specs, &tp_specs)?
1014 .ok_or("Step TP expert selection disappeared during preflight")?;
1015 validate_step_expert_activation_layout(cfg, "MEMRA_STEP_TP", &selection)?;
1016 }
1017
1018 let layer_owners = (0..trunk_layers)
1019 .map(|layer| {
1020 crate::pp::layer_engine(e, trunk_layers, layer).map(|engine| engine.ctx().ordinal())
1021 })
1022 .collect::<Result<Vec<_>, _>>()?;
1023 let plan = contract.preflight_step_tp_specs(
1024 tp_specs
1025 .iter()
1026 .map(|spec| (spec.layer, spec.devices.as_slice())),
1027 &layer_owners,
1028 )?;
1029
1030 for devices in &plan.runtime_groups {
1031 let hardware = crate::parallel::detect_uniform_hardware(devices)?;
1032 if !contract.hardware_targets.contains(&hardware) {
1033 return Err(format!(
1034 "{} has no qualified {hardware:?} TP contract for devices {devices:?}",
1035 contract.variant
1036 )
1037 .into());
1038 }
1039 }
1040
1041 if bulk_p2p && !native_p2p {
1042 return Err("MEMRA_STEP_TP_BULK_P2P=1 requires MEMRA_STEP_TP_NATIVE_P2P=1".into());
1043 }
1044 if device_arithmetic
1045 && (!ep_specs.is_empty()
1046 || !native_p2p
1047 || plan.expert_parallel_layers() == 0
1048 || plan.tensor_parallel_expert_layers() != 0)
1049 {
1050 return Err(
1051 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires native-P2P TP4/TP8 \
1052 expert ownership for every selected routed-expert layer"
1053 .into(),
1054 );
1055 }
1056 let (qualified_experts, expert_artifact) =
1060 match crate::parallel::validate_fp8_expert_checkpoint(src, &contract) {
1061 Ok(qualified) => (qualified, StepExpertArtifact::E4m3),
1062 Err(fp8_error) => {
1063 match crate::parallel::validate_nvfp4_expert_checkpoint(src, &contract) {
1064 Ok(qualified) => (qualified, StepExpertArtifact::Nvfp4),
1065 Err(nvfp4_error) => {
1066 return Err(format!(
1067 "Step checkpoint qualifies as neither native expert artifact class: \
1068 [E4M3] {fp8_error} [NVFP4] {nvfp4_error}"
1069 )
1070 .into());
1071 }
1072 }
1073 }
1074 };
1075 if expert_artifact == StepExpertArtifact::Nvfp4 {
1076 if device_arithmetic {
1077 return Err(
1078 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 is qualified for the E4M3 expert artifact \
1079 only; the NVFP4 expert program is host-canonical in this increment"
1080 .into(),
1081 );
1082 }
1083 if bulk_p2p {
1088 return Err(
1089 "MEMRA_STEP_TP_BULK_P2P=1 is qualified for the E4M3 expert artifact only; the \
1090 NVFP4 bank transport increment has not landed"
1091 .into(),
1092 );
1093 }
1094 }
1095
1096 if f32_mirror {
1097 eprintln!(
1098 "[step-tp-preflight] layers={} full_trunk={} runtime_groups={} \
1099 dense_attention_layers={} tensor_expert_layers={} expert_owner_layers={} \
1100 qualified_fp8_expert_projection_slices={} owner_first=true \
1101 hardware=rtx-pro-6000-blackwell \
1102 native_p2p={} bulk_p2p={} device_arithmetic={} bf16_residency=f32-mirror \
1103 weights_loaded=false performance_claim=false",
1104 plan.layers.len(),
1105 plan.full_trunk,
1106 plan.runtime_groups.len(),
1107 plan.dense_attention_layers(),
1108 plan.tensor_parallel_expert_layers(),
1109 plan.expert_parallel_layers(),
1110 qualified_experts,
1111 native_p2p,
1112 bulk_p2p,
1113 device_arithmetic,
1114 );
1115 } else {
1116 eprintln!(
1117 "[step-tp-preflight] layers={} full_trunk={} runtime_groups={} \
1118 dense_attention_layers={} tensor_expert_layers={} expert_owner_layers={} \
1119 qualified_fp8_expert_projection_slices={} owner_first=true \
1120 hardware=rtx-pro-6000-blackwell \
1121 native_p2p={} bulk_p2p={} device_arithmetic={} \
1122 weights_loaded=false performance_claim=false",
1123 plan.layers.len(),
1124 plan.full_trunk,
1125 plan.runtime_groups.len(),
1126 plan.dense_attention_layers(),
1127 plan.tensor_parallel_expert_layers(),
1128 plan.expert_parallel_layers(),
1129 qualified_experts,
1130 native_p2p,
1131 bulk_p2p,
1132 device_arithmetic,
1133 );
1134 }
1135 Ok(StepParallelLoadConfig {
1136 ep_specs,
1137 tp_specs,
1138 native_p2p,
1139 ep_device_arithmetic: device_arithmetic,
1140 f32_mirror,
1141 bulk_p2p,
1142 nvfp4_device_routes,
1143 auto_parallel,
1144 expert_artifact,
1145 })
1146}
1147
1148fn nvfp4_native_expert_bank<'a>(
1150 src: &'a dyn TensorSource,
1151 layer: usize,
1152 proj: &str,
1153) -> Result<memra_gguf::source::Nvfp4StackedNative<'a>, Box<dyn std::error::Error>> {
1154 let name = format!("blk.{layer}.ffn_{proj}_exps.weight");
1155 src.find_nvfp4_stacked_native(&name)
1156 .ok_or_else(|| format!("NVFP4 expert backend is missing native bank {name}").into())
1157}
1158
1159fn nvfp4_expert_bank_view<'a>(
1161 native: &'a memra_gguf::source::Nvfp4StackedNative<'a>,
1162) -> crate::tp::Nvfp4ExpertBank<'a> {
1163 crate::tp::Nvfp4ExpertBank {
1164 codes: native.codes,
1165 scales: native.scales,
1166 macros: &native.macros,
1167 expert_count: native.n_expert,
1168 out_features: native.out_f,
1169 in_features: native.in_f,
1170 }
1171}
1172
1173#[allow(clippy::too_many_arguments)] fn build_step_distributed_exps(
1175 e: &Engine,
1176 cfg: &ModelConfig,
1177 src: &dyn TensorSource,
1178 layer: usize,
1179 gate: &HostExps,
1180 up: &HostExps,
1181 down: &HostExps,
1182 step_runtimes: &mut StepParallelRuntimeRegistry,
1183) -> Result<(Option<StepEpExps>, Option<StepTpExps>), Box<dyn std::error::Error>> {
1184 let ep_device_arithmetic = step_runtimes.config.ep_device_arithmetic;
1185 if step_runtimes.config.ep_specs.is_empty() && step_runtimes.config.tp_specs.is_empty() {
1186 if ep_device_arithmetic {
1187 return Err(
1188 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires MEMRA_STEP_TP and \
1189 MEMRA_STEP_TP_NATIVE_P2P=1"
1190 .into(),
1191 );
1192 }
1193 return Ok((None, None));
1194 }
1195 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
1196 validate_step_expert_specs(
1197 &contract,
1198 "MEMRA_STEP_EP",
1199 &step_runtimes.config.ep_specs,
1200 false,
1201 )?;
1202 validate_step_expert_specs(
1203 &contract,
1204 "MEMRA_STEP_TP",
1205 &step_runtimes.config.tp_specs,
1206 true,
1207 )?;
1208 let Some(selection) = step_runtimes.expert_selection(layer)? else {
1209 return Ok((None, None));
1210 };
1211 validate_step_expert_activation_layout(
1212 cfg,
1213 if selection.configured_by_tp {
1214 "MEMRA_STEP_TP"
1215 } else {
1216 "MEMRA_STEP_EP"
1217 },
1218 &selection,
1219 )?;
1220 let activation_limit = match cfg.clamp_exp_at(layer as u32) {
1225 None => None,
1226 Some(SwigluClamp::Post(l)) => Some(l),
1227 Some(SwigluClamp::Pre(_)) => {
1228 return Err(format!(
1229 "MEMRA_STEP_EP/TP layer {layer}: glm5_next PRE-clamped SwiGLU has no \
1230 expert-parallel arm (the banks encode step35's post-clamp form)"
1231 )
1232 .into());
1233 }
1234 };
1235 let owner = e.ctx().ordinal();
1236 if !selection.spec.devices.contains(&owner) {
1237 let flag = if selection.configured_by_tp {
1238 "MEMRA_STEP_TP"
1239 } else {
1240 "MEMRA_STEP_EP"
1241 };
1242 return Err(format!(
1243 "{flag} layer {layer} owning PP device {owner} is absent from rank devices {:?}",
1244 selection.spec.devices
1245 )
1246 .into());
1247 }
1248 let expert_parallel = selection.layout == StepExpertLayout::ExpertParallel;
1249 if selection.configured_by_tp {
1250 contract.plan(crate::parallel::TopologyRequest {
1251 pipeline: 1,
1252 tensor: selection.spec.devices.len(),
1253 expert_parallel,
1254 available_devices: selection.spec.devices.len(),
1255 hardware: crate::parallel::HardwareTarget::RtxPro6000Blackwell,
1256 })?;
1257 }
1258 let native_p2p = selection.configured_by_tp && step_runtimes.config.native_p2p;
1259 if ep_device_arithmetic
1260 && (!selection.configured_by_tp
1261 || selection.layout != StepExpertLayout::ExpertParallel
1262 || !native_p2p)
1263 {
1264 return Err(
1265 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires a MEMRA_STEP_TP TP4/TP8 \
1266 expert-owner layer and MEMRA_STEP_TP_NATIVE_P2P=1"
1267 .into(),
1268 );
1269 }
1270 let expert_artifact = step_runtimes.config.expert_artifact;
1271 match selection.layout {
1272 StepExpertLayout::ExpertParallel => {
1273 if expert_artifact == StepExpertArtifact::Nvfp4 {
1274 let w4a16_device_routes = step_runtimes.config.nvfp4_device_routes;
1278 if w4a16_device_routes
1279 && !matches!(
1280 src.expert_activation_precision(),
1281 memra_gguf::source::ExpertActivationPrecision::Bf16
1282 )
1283 {
1284 return Err(
1285 "explicit-EP MEMRA_STEP_NVFP4_DEV_ROUTES=1 requires an artifact that \
1286 declares BF16 routed-expert activations; TP keeps its separately gated \
1287 quantized-activation path"
1288 .into(),
1289 );
1290 }
1291 let runtime = step_runtimes.runtime(
1296 &selection.spec.devices,
1297 step_runtimes.config.native_p2p,
1298 false,
1299 )?;
1300 let experts = runtime.upload_expert_parallel_nvfp4_normalized(gate, up, down)?;
1301 let marker = if step_runtimes.config.auto_parallel {
1302 "parallel-ep"
1303 } else {
1304 "step-ep"
1305 };
1306 eprintln!(
1307 "[{marker}] layer={layer} devices={:?} experts={} artifact=nvfp4 \
1308 expert_layout=expert-parallel expert_transport={} \
1309 macro_fold=post-kernel-once native_p2p={} w4a16_device_routes={} \
1310 performance_claim=false",
1311 selection.spec.devices,
1312 contract.expert_count,
1313 runtime.transport_label(),
1314 runtime.native_p2p(),
1315 w4a16_device_routes,
1316 );
1317 if let Some(limit) = activation_limit {
1318 eprintln!(
1319 "[step-ep-clamp] load layer={layer} routed_clamp={limit} \
1320 formula=min-silu-times-clamped-up performance_claim=false"
1321 );
1322 }
1323 return Ok((
1324 Some(StepEpExps {
1325 runtime,
1326 experts: StepEpExpertBank::Nvfp4(experts),
1327 devices: selection.spec.devices,
1328 configured_by_tp: selection.configured_by_tp,
1329 activation_limit,
1330 nvfp4_device_routes: w4a16_device_routes,
1331 grouped_decode: None,
1332 }),
1333 None,
1334 ));
1335 }
1336 let runtime =
1337 step_runtimes.runtime(&selection.spec.devices, native_p2p, ep_device_arithmetic)?;
1338 let experts = runtime.upload_expert_parallel(
1339 host_e4m3_bank(gate)?,
1340 host_e4m3_bank(up)?,
1341 host_e4m3_bank(down)?,
1342 )?;
1343 let grouped_decode = if ep_device_arithmetic {
1344 let tokens = 1;
1345 let selected = (0..contract.experts_per_token).collect::<Vec<_>>();
1346 let input = vec![0.0f32; contract.hidden_size];
1347 let route_weights = vec![1.0f32; contract.experts_per_token];
1348 let projection = runtime.prepare_step_grouped_expert_parallel_gate_with_capacity(
1349 &experts,
1350 &input,
1351 tokens,
1352 &selected,
1353 activation_limit,
1354 tokens,
1355 )?;
1356 let combine = runtime
1357 .prepare_step_grouped_expert_parallel_combine(&projection, &route_weights)?;
1358 Some(std::sync::Mutex::new(StepEpGroupedDecode {
1359 projection,
1360 combine,
1361 }))
1362 } else {
1363 None
1364 };
1365 if selection.configured_by_tp {
1366 eprintln!(
1367 "[step-tp-ep] layer={layer} devices={:?} experts={} tp={} \
1368 attention_layout=tensor-parallel expert_layout=expert-parallel \
1369 expert_transport={} tp_transport={} native_p2p={} \
1370 activation={} accumulation={} output={} \
1371 grouped_decode_prepared={} grouped_decode_capacity=1 \
1372 performance_claim=false",
1373 selection.spec.devices,
1374 contract.expert_count,
1375 selection.spec.devices.len(),
1376 runtime.transport_label(),
1377 runtime.transport_label(),
1378 runtime.native_p2p(),
1379 runtime.expert_activation_label(),
1380 runtime.expert_accumulation_label(),
1381 runtime.expert_output_label(),
1382 grouped_decode.is_some(),
1383 );
1384 } else {
1385 eprintln!(
1386 "[step-ep] layer={layer} devices={:?} experts={} \
1387 expert_layout=expert-parallel expert_transport=host-bounce \
1388 native_p2p=false performance_claim=false",
1389 selection.spec.devices, contract.expert_count
1390 );
1391 }
1392 if let Some(limit) = activation_limit {
1393 eprintln!(
1394 "[step-ep-clamp] load layer={layer} routed_clamp={limit} \
1395 formula=min-silu-times-clamped-up performance_claim=false"
1396 );
1397 }
1398 Ok((
1399 Some(StepEpExps {
1400 runtime,
1401 experts: StepEpExpertBank::E4m3(experts),
1402 devices: selection.spec.devices,
1403 configured_by_tp: selection.configured_by_tp,
1404 activation_limit,
1405 nvfp4_device_routes: false,
1406 grouped_decode,
1407 }),
1408 None,
1409 ))
1410 }
1411 StepExpertLayout::TensorParallel => {
1412 let runtime =
1413 step_runtimes.runtime(&selection.spec.devices, native_p2p, ep_device_arithmetic)?;
1414 if activation_limit.is_some() && expert_artifact == StepExpertArtifact::E4m3 {
1415 return Err(format!(
1416 "layer {layer} uses the routed SwiGLU clamp and the E4M3 TP expert \
1417 program has no clamp arm; select EP for this layer (the NVFP4 TP \
1418 program carries the clamp)"
1419 )
1420 .into());
1421 }
1422 let experts = if expert_artifact == StepExpertArtifact::Nvfp4 {
1423 let gate_native = nvfp4_native_expert_bank(src, layer, "gate")?;
1424 let up_native = nvfp4_native_expert_bank(src, layer, "up")?;
1425 let down_native = nvfp4_native_expert_bank(src, layer, "down")?;
1426 StepTpExpertBank::Nvfp4(runtime.upload_tensor_parallel_nvfp4(
1427 nvfp4_expert_bank_view(&gate_native),
1428 nvfp4_expert_bank_view(&up_native),
1429 nvfp4_expert_bank_view(&down_native),
1430 )?)
1431 } else {
1432 StepTpExpertBank::E4m3(runtime.upload_tensor_parallel(
1433 host_e4m3_bank(gate)?,
1434 host_e4m3_bank(up)?,
1435 host_e4m3_bank(down)?,
1436 )?)
1437 };
1438 eprintln!(
1439 "[step-tp] layer={layer} devices={:?} experts={} tp={} artifact={} \
1440 expert_layout=tensor-parallel transport={} native_p2p={} \
1441 performance_claim=false",
1442 selection.spec.devices,
1443 contract.expert_count,
1444 selection.spec.devices.len(),
1445 match expert_artifact {
1446 StepExpertArtifact::E4m3 => "e4m3",
1447 StepExpertArtifact::Nvfp4 => "nvfp4",
1448 },
1449 runtime.transport_label(),
1450 runtime.native_p2p(),
1451 );
1452 if let Some(limit) = activation_limit {
1453 eprintln!(
1454 "[step-tp-clamp] load layer={layer} routed_clamp={limit} \
1455 formula=min-silu-times-clamped-up performance_claim=false"
1456 );
1457 }
1458 Ok((
1459 None,
1460 Some(StepTpExps {
1461 runtime,
1462 experts,
1463 devices: selection.spec.devices,
1464 activation_limit,
1465 }),
1466 ))
1467 }
1468 }
1469}
1470
1471fn upload_step_bf16_column(
1472 runtime: &crate::tp::TpE4m3HostBounce,
1473 src: &dyn TensorSource,
1474 name: &str,
1475 expected_in: usize,
1476 expected_out: usize,
1477 f32_mirror: bool,
1478) -> Result<crate::tp::ResidentBf16ColumnParallel, Box<dyn std::error::Error>> {
1479 let tensor = src
1480 .find(name)
1481 .ok_or_else(|| format!("Step TP projection is missing {name}"))?;
1482 if tensor.ggml_type != GgmlType::BF16 {
1483 return Err(format!(
1484 "Step TP projection {name} must preserve checkpoint BF16 bytes, got {:?}",
1485 tensor.ggml_type
1486 )
1487 .into());
1488 }
1489 if tensor.ne.len() != 2 {
1490 return Err(format!(
1491 "Step TP projection {name} must be a 2-D matrix, got shape {:?}",
1492 tensor.ne
1493 )
1494 .into());
1495 }
1496 let matrix = crate::tp::Bf16Matrix {
1497 bytes: tensor.bytes.as_ref(),
1498 in_features: tensor.ne[0] as usize,
1499 out_features: tensor.ne[1] as usize,
1500 };
1501 matrix.validate()?;
1502 if matrix.in_features != expected_in || matrix.out_features != expected_out {
1503 return Err(format!(
1504 "Step TP projection {name} shape {}x{} != registered {expected_out}x{expected_in}",
1505 matrix.out_features, matrix.in_features
1506 )
1507 .into());
1508 }
1509 Ok(if f32_mirror {
1510 runtime.upload_step_bf16_column_parallel_f32_mirror(matrix)?
1511 } else {
1512 runtime.upload_step_bf16_column_parallel(matrix)?
1513 })
1514}
1515
1516fn upload_step_bf16_row(
1517 runtime: &crate::tp::TpE4m3HostBounce,
1518 src: &dyn TensorSource,
1519 name: &str,
1520 expected_in: usize,
1521 expected_out: usize,
1522 f32_mirror: bool,
1523) -> Result<crate::tp::ResidentStepBf16RowParallel, Box<dyn std::error::Error>> {
1524 let tensor = src
1525 .find(name)
1526 .ok_or_else(|| format!("Step TP projection is missing {name}"))?;
1527 if tensor.ggml_type != GgmlType::BF16 {
1528 return Err(format!(
1529 "Step TP projection {name} must preserve checkpoint BF16 bytes, got {:?}",
1530 tensor.ggml_type
1531 )
1532 .into());
1533 }
1534 if tensor.ne.len() != 2 {
1535 return Err(format!(
1536 "Step TP projection {name} must be a 2-D matrix, got shape {:?}",
1537 tensor.ne
1538 )
1539 .into());
1540 }
1541 let matrix = crate::tp::Bf16Matrix {
1542 bytes: tensor.bytes.as_ref(),
1543 in_features: tensor.ne[0] as usize,
1544 out_features: tensor.ne[1] as usize,
1545 };
1546 matrix.validate()?;
1547 if matrix.in_features != expected_in || matrix.out_features != expected_out {
1548 return Err(format!(
1549 "Step TP projection {name} shape {}x{} != registered {expected_out}x{expected_in}",
1550 matrix.out_features, matrix.in_features
1551 )
1552 .into());
1553 }
1554 Ok(if f32_mirror {
1555 runtime.upload_step_bf16_row_parallel_f32_mirror(matrix)?
1556 } else {
1557 runtime.upload_step_bf16_row_parallel(matrix)?
1558 })
1559}
1560
1561fn upload_step_tp_f32_copies(
1562 runtime: &crate::tp::TpE4m3HostBounce,
1563 src: &dyn TensorSource,
1564 name: &str,
1565 expected: usize,
1566) -> Result<Vec<CudaSlice<f32>>, Box<dyn std::error::Error>> {
1567 let tensor = src
1568 .find(name)
1569 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1570 let values = memra_gguf::dequant::dequantize(
1571 tensor.ggml_type,
1572 &tensor.bytes,
1573 tensor.ne.iter().product::<u64>() as usize,
1574 );
1575 if values.len() != expected || values.iter().any(|value| !value.is_finite()) {
1576 return Err(format!(
1577 "Step TP attention {name} has {} finite values, expected {expected}",
1578 values.len()
1579 )
1580 .into());
1581 }
1582 let mut copies = Vec::with_capacity(runtime.devices().len());
1583 for rank in 0..runtime.devices().len() {
1584 let engine = runtime
1585 .rank_engine(rank)
1586 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1587 let _main = engine.gpu.enter_main()?;
1588 copies.push(engine.htod(&values)?);
1589 }
1590 Ok(copies)
1591}
1592
1593#[allow(clippy::manual_is_multiple_of)] fn upload_step_tp_f32_row_shards(
1599 runtime: &crate::tp::TpE4m3HostBounce,
1600 src: &dyn TensorSource,
1601 name: &str,
1602 rows: usize,
1603 cols: usize,
1604) -> Result<Vec<CudaSlice<f32>>, Box<dyn std::error::Error>> {
1605 let tensor = src
1606 .find(name)
1607 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1608 let values = memra_gguf::dequant::dequantize(
1609 tensor.ggml_type,
1610 &tensor.bytes,
1611 tensor.ne.iter().product::<u64>() as usize,
1612 );
1613 let world = runtime.devices().len();
1614 if values.len() != rows * cols || rows % world != 0 || values.iter().any(|v| !v.is_finite()) {
1615 return Err(format!(
1616 "Step TP attention {name} has {} finite values, expected {rows}x{cols} \
1617 (rows divisible by world {world})",
1618 values.len()
1619 )
1620 .into());
1621 }
1622 let local_rows = rows / world;
1623 let mut shards = Vec::with_capacity(world);
1624 for rank in 0..world {
1625 let engine = runtime
1626 .rank_engine(rank)
1627 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1628 let _main = engine.gpu.enter_main()?;
1629 shards
1630 .push(engine.htod(&values[rank * local_rows * cols..(rank + 1) * local_rows * cols])?);
1631 }
1632 Ok(shards)
1633}
1634
1635#[allow(clippy::manual_is_multiple_of)] fn upload_step_tp_bf16_row_shards(
1638 runtime: &crate::tp::TpE4m3HostBounce,
1639 src: &dyn TensorSource,
1640 name: &str,
1641 rows: usize,
1642 cols: usize,
1643) -> Result<Vec<CudaSlice<u8>>, Box<dyn std::error::Error>> {
1644 let tensor = src
1645 .find(name)
1646 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1647 if tensor.ggml_type != memra_gguf::GgmlType::BF16 || tensor.bytes.len() != rows * cols * 2 {
1648 return Err(format!(
1649 "Step TP attention {name} is not a bf16 [{rows}, {cols}] tensor ({} bytes, {:?})",
1650 tensor.bytes.len(),
1651 tensor.ggml_type
1652 )
1653 .into());
1654 }
1655 let world = runtime.devices().len();
1656 if rows % world != 0 {
1657 return Err(format!("{name} rows {rows} not divisible by world {world}").into());
1658 }
1659 let local = rows / world * cols * 2;
1660 let mut shards = Vec::with_capacity(world);
1661 for rank in 0..world {
1662 let engine = runtime
1663 .rank_engine(rank)
1664 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1665 let _main = engine.gpu.enter_main()?;
1666 shards.push(engine.htod_bytes(&tensor.bytes[rank * local..(rank + 1) * local])?);
1667 }
1668 Ok(shards)
1669}
1670
1671#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1672enum StepTpAttentionPlacement {
1673 RankLocalGlobal,
1674 RankLocalSwa,
1675 OwnerSwa,
1676 OwnerTransportFallback,
1677}
1678
1679impl StepTpAttentionPlacement {
1680 fn resolve(native_p2p: bool, window: Option<u32>) -> Self {
1681 match (native_p2p, window.is_some()) {
1682 (true, true) => Self::RankLocalSwa,
1683 (false, true) => Self::OwnerSwa,
1684 (true, false) => Self::RankLocalGlobal,
1685 (false, false) => Self::OwnerTransportFallback,
1686 }
1687 }
1688
1689 fn is_rank_local(self) -> bool {
1690 matches!(self, Self::RankLocalGlobal | Self::RankLocalSwa)
1691 }
1692
1693 fn label(self) -> &'static str {
1694 match self {
1695 Self::RankLocalGlobal => "rank-local-global",
1696 Self::RankLocalSwa => "rank-local-swa-ring",
1697 Self::OwnerSwa => "owner-swa",
1698 Self::OwnerTransportFallback => "owner-transport-fallback",
1699 }
1700 }
1701}
1702
1703fn build_step_tp_qkv(
1704 e: &Engine,
1705 src: &dyn TensorSource,
1706 cfg: &ModelConfig,
1707 layer: usize,
1708 step_runtimes: &mut StepParallelRuntimeRegistry,
1709) -> Result<Option<StepTpQkv>, Box<dyn std::error::Error>> {
1710 let Some(spec) = step_runtimes.tp_spec(layer).cloned() else {
1711 return Ok(None);
1712 };
1713 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
1714 if layer >= contract.trunk_layers {
1715 return Err(format!(
1716 "MEMRA_STEP_TP layer {layer} is outside Step trunk layers 0..{}",
1717 contract.trunk_layers
1718 )
1719 .into());
1720 }
1721 let owner = e.ctx().ordinal();
1722 if spec.devices.first().copied() != Some(owner) {
1723 return Err(format!(
1724 "MEMRA_STEP_TP layer {layer} owning PP device {owner} must be the first QKV rank, \
1725 got {:?}",
1726 spec.devices
1727 )
1728 .into());
1729 }
1730 let plan = contract.plan(crate::parallel::TopologyRequest {
1731 pipeline: 1,
1732 tensor: spec.devices.len(),
1733 expert_parallel: spec.devices.len() > 2,
1734 available_devices: spec.devices.len(),
1735 hardware: crate::parallel::HardwareTarget::RtxPro6000Blackwell,
1736 })?;
1737 for rank in 0..spec.devices.len() {
1738 plan.query_head_range(layer, rank).ok_or_else(|| {
1739 format!("Step TP layer {layer} has no query-head range for rank {rank}")
1740 })?;
1741 plan.kv_head_range(layer, rank)
1742 .ok_or_else(|| format!("Step TP layer {layer} has no KV-head range for rank {rank}"))?;
1743 }
1744 let native_p2p = step_runtimes.config.native_p2p;
1745 let ep_device_arithmetic = step_runtimes.config.ep_device_arithmetic;
1746 let f32_mirror = step_runtimes.config.f32_mirror;
1747 if ep_device_arithmetic && (!native_p2p || !matches!(spec.devices.len(), 4 | 8)) {
1748 return Err(
1749 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires a MEMRA_STEP_TP TP4/TP8 \
1750 expert-owner layer and MEMRA_STEP_TP_NATIVE_P2P=1"
1751 .into(),
1752 );
1753 }
1754 let runtime = step_runtimes.runtime(&spec.devices, native_p2p, ep_device_arithmetic)?;
1755 let p = |suffix: &str| format!("blk.{layer}.{suffix}");
1756 let q = upload_step_bf16_column(
1757 &runtime,
1758 src,
1759 &p("attn_q.weight"),
1760 contract.hidden_size,
1761 contract.query_heads[layer] * contract.head_dim,
1762 f32_mirror,
1763 )?;
1764 let k = upload_step_bf16_column(
1765 &runtime,
1766 src,
1767 &p("attn_k.weight"),
1768 contract.hidden_size,
1769 contract.kv_heads[layer] * contract.head_dim,
1770 f32_mirror,
1771 )?;
1772 let v = upload_step_bf16_column(
1773 &runtime,
1774 src,
1775 &p("attn_v.weight"),
1776 contract.hidden_size,
1777 contract.kv_heads[layer] * contract.head_dim,
1778 f32_mirror,
1779 )?;
1780 let o = upload_step_bf16_row(
1781 &runtime,
1782 src,
1783 &p("attn_output.weight"),
1784 contract.query_heads[layer] * contract.head_dim,
1785 contract.hidden_size,
1786 f32_mirror,
1787 )?;
1788 let geometry = cfg.full_attention_geometry_at(layer as u32);
1789 let attention_placement =
1790 StepTpAttentionPlacement::resolve(runtime.native_p2p(), geometry.window);
1791 let attention = if attention_placement.is_rank_local() {
1792 let decode_input = if ep_device_arithmetic || crate::tp::step_tp_decode_v2_enabled()? {
1797 Some(std::sync::Mutex::new(
1798 runtime.allocate_replicated_device_rows(1, contract.hidden_size)?,
1799 ))
1800 } else {
1801 None
1802 };
1803 let gate_fused =
1806 crate::tp::step_tp_qkv_fused_enabled()? && src.find(&p("attn_gate.weight")).is_some();
1807 let gate_shards = if gate_fused && f32_mirror {
1808 Some(upload_step_tp_f32_row_shards(
1809 &runtime,
1810 src,
1811 &p("attn_gate.weight"),
1812 contract.query_heads[layer],
1813 contract.hidden_size,
1814 )?)
1815 } else {
1816 None
1817 };
1818 let gate_shards_bf16 = if gate_fused && !f32_mirror {
1819 Some(upload_step_tp_bf16_row_shards(
1820 &runtime,
1821 src,
1822 &p("attn_gate.weight"),
1823 contract.query_heads[layer],
1824 contract.hidden_size,
1825 )?)
1826 } else {
1827 None
1828 };
1829 Some(StepTpAttention {
1830 q_norm: upload_step_tp_f32_copies(
1831 &runtime,
1832 src,
1833 &p("attn_q_norm.weight"),
1834 contract.head_dim,
1835 )?,
1836 k_norm: upload_step_tp_f32_copies(
1837 &runtime,
1838 src,
1839 &p("attn_k_norm.weight"),
1840 contract.head_dim,
1841 )?,
1842 decode_input,
1843 gate_shards,
1844 gate_shards_bf16,
1845 })
1846 } else {
1847 None
1848 };
1849 if f32_mirror {
1850 eprintln!(
1851 "[step-tp-qkv] load layer={layer} devices={:?} projections=qkv \
1852 qkv_tensor_parallel=true attention_local=true kv_local=true output_local=true \
1853 transport={} native_p2p={} bf16_residency=f32-mirror \
1854 output=root-readback performance_claim=false",
1855 spec.devices,
1856 runtime.transport_label(),
1857 runtime.native_p2p(),
1858 );
1859 } else {
1860 eprintln!(
1861 "[step-tp-qkv] load layer={layer} devices={:?} projections=qkv \
1862 qkv_tensor_parallel=true attention_local=true kv_local=true output_local=true \
1863 transport={} native_p2p={} output=root-readback performance_claim=false",
1864 spec.devices,
1865 runtime.transport_label(),
1866 runtime.native_p2p(),
1867 );
1868 }
1869 eprintln!(
1870 "[step-tp-attn-plan] load layer={layer} devices={:?} \
1871 qkv_tensor_parallel=true attention_tensor_parallel={} kv_cache_distributed={} \
1872 attention_scope={} transport={} native_p2p={} replicated_decode_input_prepared={} \
1873 performance_claim=false",
1874 spec.devices,
1875 attention_placement.is_rank_local(),
1876 attention_placement.is_rank_local(),
1877 attention_placement.label(),
1878 runtime.transport_label(),
1879 runtime.native_p2p(),
1880 attention
1881 .as_ref()
1882 .is_some_and(|attention| attention.decode_input.is_some()),
1883 );
1884 if f32_mirror {
1885 eprintln!(
1886 "[step-tp-o] load layer={layer} devices={:?} projection=o \
1887 o_tensor_parallel=true attention_local=true kv_local=true \
1888 transport={} native_p2p={} reduction=global-tp8-block-order \
1889 bf16_residency=f32-mirror output=root-readback performance_claim=false",
1890 spec.devices,
1891 runtime.transport_label(),
1892 runtime.native_p2p(),
1893 );
1894 } else {
1895 eprintln!(
1896 "[step-tp-o] load layer={layer} devices={:?} projection=o \
1897 o_tensor_parallel=true attention_local=true kv_local=true \
1898 transport={} native_p2p={} reduction=global-tp8-block-order \
1899 output=root-readback performance_claim=false",
1900 spec.devices,
1901 runtime.transport_label(),
1902 runtime.native_p2p(),
1903 );
1904 }
1905 Ok(Some(StepTpQkv {
1906 runtime,
1907 q,
1908 k,
1909 v,
1910 o,
1911 attention,
1912 devices: spec.devices,
1913 layer,
1914 }))
1915}
1916
1917fn build_dev_exps(
1930 e: &Engine,
1931 resident: &mut ResidentPlan,
1932 il: usize,
1933 gate: &HostExps,
1934 up: &HostExps,
1935 down: &HostExps,
1936) -> Result<Option<crate::hybrid::DevExps>, Box<dyn std::error::Error>> {
1937 if !gate.is_uniform_layout() || !up.is_uniform_layout() || !down.is_uniform_layout() {
1940 return Ok(None);
1941 }
1942 let fp8_host = match (&gate.fp8_blk, &up.fp8_blk, &down.fp8_blk) {
1943 (None, None, None) => None,
1944 (Some(g), Some(u), Some(d)) => Some((g, u, d)),
1945 _ => {
1946 return Err("resident expert projections disagree on block-E4M3 scale carriage".into());
1947 }
1948 };
1949 let scale_bytes = fp8_host
1950 .map(|(g, u, d)| (g.scales.len() + u.scales.len() + d.scales.len()) * size_of::<f32>())
1951 .unwrap_or(0);
1952 let per_layer = gate.bytes.as_bytes().len()
1953 + up.bytes.as_bytes().len()
1954 + down.bytes.as_bytes().len()
1955 + scale_bytes;
1956 if gate.tiers.is_some() {
1957 return Ok(None); }
1959 let fits = resident.should_reside(e, il, per_layer);
1960 if !fits {
1961 return Ok(None);
1962 }
1963 use cudarc::driver::DevicePtr;
1964 let gu_il = std::env::var("MEMRA_MOE_GU_IL").as_deref() == Ok("1")
1965 && gate.out_f == up.out_f
1966 && gate.in_f == up.in_f
1967 && fp8_host.is_none();
1968 let n_expert = gate.n_expert;
1969 let (g, u) = if gu_il {
1970 let (rbg, rbu) = (gate.row_bytes, up.row_bytes);
1972 let n_rows = gate.out_f;
1973 let gb = gate.bytes.as_bytes();
1974 let ub = up.bytes.as_bytes();
1975 let mut il = vec![0u8; n_expert * n_rows * (rbg + rbu)];
1976 for ex in 0..n_expert {
1977 for o in 0..n_rows {
1978 let dst = (ex * n_rows + o) * (rbg + rbu);
1979 let sg = ex * gate.expert_stride + o * rbg;
1980 let su = ex * up.expert_stride + o * rbu;
1981 il[dst..dst + rbg].copy_from_slice(&gb[sg..sg + rbg]);
1982 il[dst + rbg..dst + rbg + rbu].copy_from_slice(&ub[su..su + rbu]);
1983 }
1984 }
1985 let ild = e.htod_bytes_padded(&il, 8)?;
1986 (ild, e.htod_bytes(&[0u8; 16])?)
1989 } else {
1990 (
1991 e.htod_bytes_padded(gate.bytes.as_bytes(), 8)?,
1992 e.htod_bytes_padded(up.bytes.as_bytes(), 8)?,
1993 )
1994 };
1995 let d = e.htod_bytes_padded(down.bytes.as_bytes(), 144)?;
2000 let fp8_blk = match fp8_host {
2001 Some((gate, up, down)) => {
2002 if e.fp8_blk_nan_count(&g)? != 0
2003 || e.fp8_blk_nan_count(&u)? != 0
2004 || e.fp8_blk_nan_count(&d)? != 0
2005 {
2006 return Err("native stacked block-E4M3 expert bank contains NaN codes".into());
2007 }
2008 Some(DevExpertFp8BlockScales {
2009 gate: DevExpertFp8ProjectionScales::upload(e, gate, n_expert)?,
2010 up: DevExpertFp8ProjectionScales::upload(e, up, n_expert)?,
2011 down: DevExpertFp8ProjectionScales::upload(e, down, n_expert)?,
2012 })
2013 }
2014 None => None,
2015 };
2016 let mut host = vec![0u64; 3 * n_expert];
2017 let (pg, pu, pd) = {
2018 let __s_e0 = e.stream();
2019 let (pg, _e0) = g.device_ptr(&__s_e0);
2020 let __s_e1 = e.stream();
2021 let (pu, _e1) = u.device_ptr(&__s_e1);
2022 let __s_e2 = e.stream();
2023 let (pd, _e2) = d.device_ptr(&__s_e2);
2024 (pg, pu, pd)
2025 };
2026 for ex in 0..n_expert {
2027 if gu_il {
2028 let stride = gate.out_f * (gate.row_bytes + up.row_bytes);
2029 host[ex] = pg + (ex * stride) as u64;
2030 host[n_expert + ex] = pg + (ex * stride + gate.row_bytes) as u64;
2031 } else {
2032 host[ex] = pg + (ex * gate.expert_stride) as u64;
2033 host[n_expert + ex] = pu + (ex * up.expert_stride) as u64;
2034 }
2035 host[2 * n_expert + ex] = pd + (ex * down.expert_stride) as u64;
2036 }
2037 if gu_il {
2038 eprintln!("[moe] gate/up dev slab INTERLEAVED (MEMRA_MOE_GU_IL)");
2039 }
2040 let ptr_row = e.htod_u64(&host)?;
2041 Ok(Some(crate::hybrid::DevExps {
2042 gate: g,
2043 up: u,
2044 down: d,
2045 ptr_row,
2046 gu_il,
2047 dev: e.ctx().ordinal(),
2048 fp8_blk,
2049 }))
2050}
2051
2052pub struct FullAttnLayer {
2053 pub wq: GpuTensor,
2054 pub wk: GpuTensor,
2055 pub wv: GpuTensor,
2056 pub wo: GpuTensor,
2057 pub q_norm: GpuTensor,
2058 pub k_norm: GpuTensor,
2059 pub attn_gate: Option<GpuTensor>,
2070 pub step_tp_qkv: Option<StepTpQkv>,
2074}
2075
2076pub struct StepTpQkv {
2077 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
2078 pub q: crate::tp::ResidentBf16ColumnParallel,
2079 pub k: crate::tp::ResidentBf16ColumnParallel,
2080 pub v: crate::tp::ResidentBf16ColumnParallel,
2081 pub o: crate::tp::ResidentStepBf16RowParallel,
2082 pub attention: Option<StepTpAttention>,
2083 pub devices: Vec<usize>,
2084 pub layer: usize,
2085}
2086
2087pub struct StepTpAttention {
2088 pub q_norm: Vec<CudaSlice<f32>>,
2089 pub k_norm: Vec<CudaSlice<f32>>,
2090 pub decode_input: Option<std::sync::Mutex<crate::tp::ResidentReplicatedDeviceRows>>,
2091 pub gate_shards: Option<Vec<CudaSlice<f32>>>,
2094 pub gate_shards_bf16: Option<Vec<CudaSlice<u8>>>,
2096}
2097
2098#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2099pub struct StepTpKvDeviceAdmission {
2100 pub device: usize,
2101 pub bytes: usize,
2102}
2103
2104#[derive(Clone, Copy, Debug)]
2108pub struct MlaGeom {
2109 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, }
2117
2118#[derive(Clone, Copy, Debug)]
2123pub struct MlaIndexerGeom {
2124 pub heads: usize, pub head_dim: usize, pub top_k: usize, pub pool: usize, pub always_select_tail: bool,
2129}
2130
2131impl MlaIndexerGeom {
2132 pub fn select_k(&self, n_pools: usize) -> usize {
2134 (self.top_k / self.pool).min(n_pools)
2135 }
2136
2137 pub fn index_width(&self, n_pools: usize) -> usize {
2139 self.select_k(n_pools) * self.pool
2140 + if self.always_select_tail {
2141 self.pool - 1
2142 } else {
2143 0
2144 }
2145 }
2146
2147 pub fn state_width(&self) -> usize {
2149 2 * self.head_dim
2150 }
2151}
2152
2153pub struct MlaIndexer {
2157 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,
2165}
2166
2167pub struct MlaAttnLayer {
2168 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,
2178 pub index: Option<MlaIndexer>,
2181 pub tp: Option<Box<crate::glm5_tp::Glm5TpMla>>,
2186}
2187
2188impl MlaAttnLayer {
2189 pub fn load(
2201 e: &Engine,
2202 src: &dyn TensorSource,
2203 il: u32,
2204 plan: &memra_gguf::model_plan::MlaAttentionPlan,
2205 ) -> Result<Self, Box<dyn std::error::Error>> {
2206 let memra_gguf::model_plan::MlaAttentionPlan::LatentKv {
2207 query_heads,
2208 q_lora_rank,
2209 kv_lora_rank,
2210 qk_head_dim,
2211 rope_head_dim,
2212 value_head_dim,
2213 sparse_index,
2214 ..
2215 } = plan
2216 else {
2217 return Err(format!(
2218 "native MLA loader has no compressed-KV implementation for block {il}"
2219 )
2220 .into());
2221 };
2222 let d_nope = qk_head_dim
2223 .checked_sub(*rope_head_dim)
2224 .ok_or("MLA rope head width exceeds total QK head width")?;
2225 let p = |s: &str| format!("blk.{il}.{s}");
2226 let geom = MlaGeom {
2227 n_head: *query_heads as usize,
2228 d_nope: d_nope as usize,
2229 d_rope: *rope_head_dim as usize,
2230 d_v: *value_head_dim as usize,
2231 kv_rank: *kv_lora_rank as usize,
2232 latent_dim: (*kv_lora_rank + *rope_head_dim) as usize,
2233 scale: 1.0 / (*qk_head_dim as f32).sqrt(),
2234 };
2235 let wq_a = load_t(e, src, &p("attn_q_a.weight"))?;
2236 let wq_b = load_t(e, src, &p("attn_q_b.weight"))?;
2237 let wkv_a = load_t(e, src, &p("attn_kv_a_mqa.weight"))?;
2238 let wk_b = load_t(e, src, &p("attn_k_b.weight"))?;
2239 let wv_b = load_t(e, src, &p("attn_v_b.weight"))?;
2240 let wo = load_t(e, src, &p("attn_output.weight"))?;
2241 for (w, tensor) in [(&wk_b, "attn_k_b"), (&wv_b, "attn_v_b")] {
2250 if !matches!(w, GpuTensor::Float { .. }) {
2251 return Err(format!(
2252 "blk.{il}.{tensor}.weight is not f32-resident. The MLA conversion-split \
2253 operands feed f32-only absorb/decompress kernels; the checkpoint source must \
2254 dequantize them (TensorTransform::SplitMlaKv) rather than hand the engine a \
2255 quantized plane"
2256 )
2257 .into());
2258 }
2259 }
2260 let n_head = wq_b.out_features() / (geom.d_nope + geom.d_rope);
2262 assert_eq!(
2263 wq_b.out_features(),
2264 n_head * (geom.d_nope + geom.d_rope),
2265 "wq_b out {} not a multiple of qk_head_dim {}",
2266 wq_b.out_features(),
2267 geom.d_nope + geom.d_rope
2268 );
2269 assert_eq!(
2270 wq_a.in_features(),
2271 wkv_a.in_features(),
2272 "q_a/kv_a hidden mismatch"
2273 );
2274 assert_eq!(
2275 wq_b.in_features(),
2276 *q_lora_rank as usize,
2277 "wq_b in != q_lora_rank"
2278 );
2279 assert_eq!(
2280 n_head, geom.n_head,
2281 "MLA checkpoint head count != ModelPlan"
2282 );
2283 assert_eq!(
2284 wkv_a.out_features(),
2285 geom.latent_dim,
2286 "wkv_a out != kv_lora_rank + rope"
2287 );
2288 assert_eq!(
2289 wk_b.ne(),
2290 &[geom.d_nope as u64, geom.kv_rank as u64, n_head as u64],
2291 "attn_k_b must be the TRANSPOSED (nope, kv_rank, head) conversion split"
2292 );
2293 assert_eq!(
2294 wv_b.ne(),
2295 &[geom.kv_rank as u64, geom.d_v as u64, n_head as u64],
2296 "attn_v_b must be the (kv_rank, v, head) conversion split"
2297 );
2298 assert_eq!(
2299 wo.in_features(),
2300 n_head * geom.d_v,
2301 "wo in != n_head * v_head_dim"
2302 );
2303 let index = Self::load_indexer(e, src, il, sparse_index, *q_lora_rank)?;
2304 Ok(MlaAttnLayer {
2305 wq_a,
2306 q_a_norm: load_t(e, src, &p("attn_q_a_norm.weight"))?,
2307 wq_b,
2308 wkv_a,
2309 kv_a_norm: load_t(e, src, &p("attn_kv_a_norm.weight"))?,
2310 wk_b,
2311 wv_b,
2312 wo,
2313 geom,
2314 index,
2315 tp: None,
2316 })
2317 }
2318
2319 fn load_indexer(
2331 e: &Engine,
2332 src: &dyn TensorSource,
2333 il: u32,
2334 sparse_index: &memra_gguf::model_plan::SparseIndexPlan,
2335 q_lora_rank: u32,
2336 ) -> Result<Option<MlaIndexer>, Box<dyn std::error::Error>> {
2337 let memra_gguf::model_plan::SparseIndexPlan::Own {
2338 heads,
2339 head_dim,
2340 top_k,
2341 kpool: Some(kpool),
2342 } = sparse_index
2343 else {
2344 return Ok(None);
2345 };
2346 let geom = MlaIndexerGeom {
2347 heads: *heads as usize,
2348 head_dim: *head_dim as usize,
2349 top_k: *top_k as usize,
2350 pool: kpool.pool as usize,
2351 always_select_tail: kpool.always_select_tail,
2352 };
2353 if geom.heads == 0 || geom.head_dim == 0 || geom.pool == 0 || geom.top_k < geom.pool {
2354 return Err(format!(
2355 "blk.{il}: SparseIndexPlan::Own declares an unusable k-pool indexer \
2356 (heads {}, head_dim {}, pool {}, top_k {}) — heads/head_dim/pool must be \
2357 positive and top_k must admit at least one pool",
2358 geom.heads, geom.head_dim, geom.pool, geom.top_k
2359 )
2360 .into());
2361 }
2362 let need = |suffix: &str| -> Result<GpuTensor, Box<dyn std::error::Error>> {
2367 let name = format!("blk.{il}.{suffix}");
2368 if !src.has(&name) {
2369 return Err(format!(
2370 "blk.{il}: the layer's ModelPlan declares a DSA k-pool indexer but the \
2371 checkpoint has no `{name}`. This layer MUST NOT fall back to dense \
2372 attention: dense and indexed attention are the same function only below \
2373 index_topk ({}), and glm5_next serves a 1,048,576-token context",
2374 geom.top_k
2375 )
2376 .into());
2377 }
2378 load_t(e, src, &name).map_err(|source| -> Box<dyn std::error::Error> {
2379 format!("blk.{il}: DSA k-pool indexer tensor `{name}` failed to load: {source}")
2380 .into()
2381 })
2382 };
2383 let wq_b = need("indexer.attn_q_b.weight")?;
2384 let wk = need("indexer.attn_k.weight")?;
2385 let k_norm_w = need("indexer.k_norm.weight")?;
2386 let k_norm_b = need("indexer.k_norm.bias")?;
2387 let weights_proj = need("indexer.proj.weight")?;
2388 let kpool_gate = need("indexer.kpool_gate.weight")?;
2389 let kpool_ape = need("indexer.kpool_ape.weight")?;
2390 for (w, name) in [
2393 (&k_norm_w, "indexer.k_norm.weight"),
2394 (&k_norm_b, "indexer.k_norm.bias"),
2395 (&kpool_ape, "indexer.kpool_ape.weight"),
2396 ] {
2397 if !matches!(w, GpuTensor::Float { .. }) {
2398 return Err(format!(
2399 "blk.{il}.{name} is not f32-resident. The indexer's LayerNorm affine and \
2400 k-pool positional embedding feed f32-only kernels"
2401 )
2402 .into());
2403 }
2404 }
2405 assert_eq!(
2406 wq_b.in_features(),
2407 q_lora_rank as usize,
2408 "blk.{il}.indexer.attn_q_b in != q_lora_rank"
2409 );
2410 assert_eq!(
2411 wq_b.out_features(),
2412 geom.heads * geom.head_dim,
2413 "blk.{il}.indexer.attn_q_b out != index heads * head_dim"
2414 );
2415 assert_eq!(
2416 wk.out_features(),
2417 geom.head_dim,
2418 "blk.{il}.indexer.attn_k out != index head_dim"
2419 );
2420 assert_eq!(
2421 weights_proj.out_features(),
2422 geom.heads,
2423 "blk.{il}.indexer.proj out != index heads"
2424 );
2425 assert_eq!(
2426 kpool_gate.out_features(),
2427 geom.head_dim,
2428 "blk.{il}.indexer.kpool_gate out != index head_dim"
2429 );
2430 assert_eq!(
2431 kpool_ape.float_data().len(),
2432 geom.pool * geom.head_dim,
2433 "blk.{il}.indexer.kpool_ape must hold pool * head_dim elements"
2434 );
2435 Ok(Some(MlaIndexer {
2436 wq_b,
2437 wk,
2438 k_norm_w,
2439 k_norm_b,
2440 weights_proj,
2441 kpool_gate,
2442 kpool_ape,
2443 geom,
2444 }))
2445 }
2446}
2447
2448#[track_caller]
2456pub(crate) fn mla_path_unimplemented(path: &str) -> ! {
2457 panic!(
2458 "Mixer::Mla has no {path} arm — the MLA forward is wired for the stateless forward, \
2459 the stateful prime and T=1 decode only (cu/mla_attn.cu, increment 4); this path needs \
2460 its own parity gate before it may run \
2461 (research/mla-bringup-20260801/DESIGN.md §4, increment 7)"
2462 )
2463}
2464
2465#[track_caller]
2471pub(crate) fn kda_path_unimplemented(path: &str) -> ! {
2472 panic!(
2473 "Mixer::Kda has no {path} arm — glm5_next KDA is wired for the stateless forward, the \
2474 stateful prime and T=1 decode only (crates/memra-engine/src/kda.rs); this path needs \
2475 its own parity gate before it may run"
2476 )
2477}
2478
2479pub struct LinearAttnLayer {
2480 pub geometry: memra_gguf::model_plan::GatedDeltaNetPlan,
2481 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, }
2491
2492#[allow(clippy::large_enum_variant)] pub enum Mixer {
2494 Full(FullAttnLayer),
2495 Linear(LinearAttnLayer),
2496 Mla(MlaAttnLayer),
2498 Kda(crate::kda::KdaAttnLayer),
2500}
2501
2502pub struct MoeWeights {
2509 pub gate_inp: GpuTensor, pub gate_inp_shexp: Option<GpuTensor>, pub exp_probs_b: Option<Vec<f32>>,
2515 pub exp_probs_b_dev: CudaSlice<f32>,
2516 pub active_experts: Option<Vec<bool>>,
2520 pub active_experts_dev: CudaSlice<u8>,
2521 pub gate_exps: HostExps, pub up_exps: HostExps, pub down_exps: HostExps, pub gate_shexp: Option<GpuTensor>,
2525 pub up_shexp: Option<GpuTensor>,
2526 pub down_shexp: Option<GpuTensor>,
2527 pub dev_exps: Option<DevExps>,
2534 pub step_ep: Option<StepEpExps>,
2538 pub step_tp: Option<StepTpExps>,
2542 pub glm5_ep: Option<crate::glm5_tp::Glm5EpExps>,
2547 pub dev_macros: cudarc::driver::CudaSlice<f32>,
2553 pub has_macros: bool,
2554 pub w4a16_bf16_activations: bool,
2557}
2558
2559#[allow(clippy::large_enum_variant)] pub enum StepEpExpertBank {
2562 E4m3(crate::tp::ResidentExpertParallel),
2563 Nvfp4(crate::tp::ResidentNvfp4ExpertParallel),
2564}
2565
2566impl StepEpExpertBank {
2567 pub fn e4m3(&self) -> Result<&crate::tp::ResidentExpertParallel, String> {
2571 match self {
2572 Self::E4m3(bank) => Ok(bank),
2573 Self::Nvfp4(_) => Err(
2574 "Step grouped expert program reached an NVFP4 bank; this path is qualified \
2575 for the E4M3 artifact only"
2576 .to_string(),
2577 ),
2578 }
2579 }
2580}
2581
2582pub struct StepEpExps {
2583 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
2584 pub experts: StepEpExpertBank,
2585 pub devices: Vec<usize>,
2586 pub configured_by_tp: bool,
2587 pub activation_limit: Option<f32>,
2588 pub nvfp4_device_routes: bool,
2590 pub grouped_decode: Option<std::sync::Mutex<StepEpGroupedDecode>>,
2593}
2594
2595pub struct StepEpGroupedDecode {
2596 pub(crate) projection: crate::tp::PreparedStepGroupedExpertParallelGate,
2597 pub(crate) combine: crate::tp::PreparedPeerWeightedRouteCombine,
2598}
2599
2600#[derive(Default)]
2601pub(crate) struct StepEpGroupedPrefill {
2602 pub(crate) state: Option<StepEpGroupedPrefillState>,
2603}
2604
2605pub(crate) struct StepEpGroupedPrefillState {
2606 pub(crate) devices: Vec<usize>,
2607 pub(crate) grouped: StepEpGroupedDecode,
2608}
2609
2610#[allow(clippy::large_enum_variant)] pub enum StepTpExpertBank {
2613 E4m3(crate::tp::ResidentTensorParallel),
2614 Nvfp4(crate::tp::ResidentNvfp4TensorParallel),
2615}
2616
2617pub struct StepTpExps {
2618 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
2619 pub experts: StepTpExpertBank,
2620 pub devices: Vec<usize>,
2621 pub activation_limit: Option<f32>,
2624}
2625
2626impl MoeWeights {
2627 #[inline]
2628 pub fn has_uniform_expert_layout(&self) -> bool {
2629 self.gate_exps.is_uniform_layout()
2630 && self.up_exps.is_uniform_layout()
2631 && self.down_exps.is_uniform_layout()
2632 }
2633
2634 #[inline]
2635 pub fn active_count(&self) -> usize {
2636 self.active_experts
2637 .as_ref()
2638 .map(|mask| mask.iter().filter(|&&active| active).count())
2639 .unwrap_or(self.gate_exps.n_expert)
2640 }
2641
2642 #[allow(clippy::too_many_arguments)]
2643 pub(crate) fn qmatvec_view(
2644 &self,
2645 e: &Engine,
2646 w: &CudaSlice<u8>,
2647 range: std::ops::Range<usize>,
2648 x: &cudarc::driver::CudaView<f32>,
2649 m: usize,
2650 in_f: usize,
2651 out_f: usize,
2652 qtype: i32,
2653 row_bytes: usize,
2654 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
2655 if self.w4a16_bf16_activations && qtype == crate::QT_NVFP4 {
2656 e.qmatvec_view_bf16_activation(w, range, x, m, in_f, out_f, qtype, row_bytes)
2657 } else {
2658 e.qmatvec_view(w, range, x, m, in_f, out_f, qtype, row_bytes)
2659 }
2660 }
2661}
2662
2663pub struct DevExps {
2666 pub gate: CudaSlice<u8>,
2667 pub up: CudaSlice<u8>,
2668 pub down: CudaSlice<u8>,
2669 pub ptr_row: CudaSlice<u64>,
2671 pub dev: usize,
2679 pub gu_il: bool,
2685 pub fp8_blk: Option<DevExpertFp8BlockScales>,
2689}
2690
2691pub struct DevExpertFp8BlockScales {
2692 pub gate: DevExpertFp8ProjectionScales,
2693 pub up: DevExpertFp8ProjectionScales,
2694 pub down: DevExpertFp8ProjectionScales,
2695}
2696
2697pub struct DevExpertFp8ProjectionScales {
2698 pub scales: CudaSlice<f32>,
2699 pub rows: usize,
2700 pub cols: usize,
2701 pub expert_stride: usize,
2702}
2703
2704impl DevExpertFp8ProjectionScales {
2705 fn validate(
2706 host: &crate::model::HostExpertFp8BlockScales,
2707 n_expert: usize,
2708 ) -> Result<(), String> {
2709 if host.expert_stride == 0 {
2710 return Err("block-E4M3 expert scale stride must be nonzero".into());
2711 }
2712 if host.rows * host.cols != host.expert_stride {
2713 return Err(format!(
2714 "block-E4M3 expert scale stride mismatch: {}x{} != {}",
2715 host.rows, host.cols, host.expert_stride
2716 ));
2717 }
2718 let want = n_expert
2719 .checked_mul(host.expert_stride)
2720 .ok_or("block-E4M3 expert scale slab length overflow")?;
2721 if host.scales.len() != want {
2722 return Err(format!(
2723 "block-E4M3 scale slab length mismatch: got {}, want {n_expert}x{}={want}",
2724 host.scales.len(),
2725 host.expert_stride
2726 ));
2727 }
2728 Ok(())
2729 }
2730
2731 fn upload(
2732 e: &Engine,
2733 host: &crate::model::HostExpertFp8BlockScales,
2734 n_expert: usize,
2735 ) -> Result<Self, Box<dyn std::error::Error>> {
2736 Self::validate(host, n_expert)?;
2737 Ok(Self {
2738 scales: e.htod(&host.scales)?,
2739 rows: host.rows,
2740 cols: host.cols,
2741 expert_stride: host.expert_stride,
2742 })
2743 }
2744}
2745
2746#[allow(clippy::large_enum_variant)] pub enum Ffn {
2749 Dense {
2750 ffn_gate: GpuTensor,
2751 ffn_up: GpuTensor,
2752 ffn_down: GpuTensor,
2753 },
2754 Moe(MoeWeights),
2755}
2756
2757pub struct HybridLayer {
2758 pub attn_norm: GpuTensor,
2759 pub post_attn_norm: GpuTensor, pub mixer: Mixer,
2761 pub ffn: Ffn,
2762 pub gemma4: Option<Gemma4LayerBits>,
2763 pub hyper: Option<crate::hyper::HyperLayer>,
2768}
2769
2770pub struct Gemma4LayerBits {
2774 pub ffn_norm: GpuTensor, pub post_ffw_norm: GpuTensor, pub moe_bits: Option<Gemma4MoeBits>,
2779 pub layer_scale: f32, pub e4b: Option<Gemma4E4bLayer>,
2782}
2783
2784pub struct Gemma4E4bLayer {
2789 pub inp_gate: GpuTensor, pub proj: GpuTensor, pub post_norm: GpuTensor, pub qkv_cat: Option<GpuTensor>,
2796 pub kv_share: Option<u32>,
2800}
2801
2802pub struct Gemma4E4bModel {
2806 pub tok_tbl_gpu: std::sync::OnceLock<CudaSlice<u8>>,
2809 pub tok_embd_bytes: Vec<u8>,
2810 pub tok_embd_qt: i32,
2811 pub tok_embd_row_bytes: usize,
2812 pub model_proj: GpuTensor, pub proj_norm: GpuTensor, pub n_epl: usize,
2815}
2816
2817pub struct Gemma4MoeBits {
2818 pub post_ffw_norm_1: GpuTensor, pub pre_ffw_norm_2: GpuTensor, pub post_ffw_norm_2: GpuTensor, pub shared_gate: GpuTensor,
2822 pub shared_up: GpuTensor,
2823 pub shared_down: GpuTensor,
2824 pub router_scale_pre: CudaSlice<f32>,
2829 pub per_expert_scale: Vec<f32>, pub per_expert_scale_d: CudaSlice<f32>, }
2832
2833fn load_mtp_head_maybe_nvfp4(
2846 e: &Engine,
2847 src: &dyn TensorSource,
2848 name: &str,
2849) -> Result<Option<GpuTensor>, Box<dyn std::error::Error>> {
2850 if !{
2851 static ENV: std::sync::OnceLock<Option<bool>> = std::sync::OnceLock::new();
2852 crate::step37_door(&ENV, "MEMRA_MTP_HEAD_NVFP4")
2853 } {
2854 return load_opt(e, src, name);
2855 }
2856 let Some(v) = src.find(name) else {
2857 return Ok(None);
2858 };
2859 if !matches!(v.ggml_type, GgmlType::BF16) || v.ne[0] % 64 != 0 {
2860 return load_opt(e, src, name);
2861 }
2862 let vals: Vec<f32> = v
2863 .bytes
2864 .chunks_exact(2)
2865 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
2866 .collect();
2867 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
2868 eprintln!(
2869 "[mtp-head] {name}: BF16 -> NVFP4 ({} MiB, was {} MiB)",
2870 blocks.len() >> 20,
2871 v.bytes.len() >> 20
2872 );
2873 Ok(Some(GpuTensor::from_quant_bytes(
2874 e,
2875 &blocks,
2876 GgmlType::NVFP4,
2877 v.ne[0],
2878 v.ne[1],
2879 1.0,
2880 )?))
2881}
2882
2883fn sha256_file_hex8(path: &std::path::Path) -> Result<String, Box<dyn std::error::Error>> {
2890 use sha2::{Digest, Sha256};
2891 let mut file = std::fs::File::open(path)?;
2892 let mut hasher = Sha256::new();
2893 std::io::copy(&mut file, &mut hasher)?;
2894 let digest = hasher.finalize();
2895 Ok(digest
2896 .iter()
2897 .take(4)
2898 .map(|byte| format!("{byte:02x}"))
2899 .collect())
2900}
2901
2902pub(crate) fn frspec_trim_own_head_name(n_trunk: usize) -> String {
2903 format!("blk.{n_trunk}.nextn.shared_head_head.weight")
2904}
2905
2906pub struct DflashTrimHead {
2917 pub head: GpuTensor,
2920 pub d2t: Vec<u32>,
2922}
2923
2924fn frspec_read_d2t(path: &str) -> Result<Vec<u32>, Box<dyn std::error::Error>> {
2929 Ok(if path.ends_with(".txt") {
2930 std::fs::read_to_string(path)?
2931 .lines()
2932 .filter_map(|l| l.trim().parse::<u32>().ok())
2933 .collect()
2934 } else {
2935 let tg = GgufFile::open(path)?;
2936 let d2t_t = tg
2937 .find("d2t")
2938 .expect("MEMRA_FRSPEC_TRIM file has no d2t tensor");
2939 let d2t_bytes = tg.tensor_data(d2t_t);
2940 match d2t_t.ggml_type {
2941 GgmlType::I32 => d2t_bytes
2942 .chunks_exact(4)
2943 .map(|c| i32::from_le_bytes(c.try_into().unwrap()) as u32)
2944 .collect(),
2945 GgmlType::I64 => d2t_bytes
2946 .chunks_exact(8)
2947 .map(|c| i64::from_le_bytes(c.try_into().unwrap()) as u32)
2948 .collect(),
2949 other => panic!("d2t must be I32/I64, got {other:?}"),
2950 }
2951 })
2952}
2953
2954#[allow(clippy::type_complexity)] fn frspec_gather_trimmed_head(
2963 e: &Engine,
2964 v: &memra_gguf::source::TensorView<'_>,
2965 d2t: &[u32],
2966 want_nvfp4_env: bool,
2967 macro_scale: f32,
2968) -> Result<(GpuTensor, Option<(usize, usize)>), Box<dyn std::error::Error>> {
2969 let out_f = v.ne[1] as usize;
2970 let row_bytes = v.bytes.len() / out_f;
2971 assert!(
2972 d2t.iter().all(|&t| (t as usize) < out_f),
2973 "d2t token id >= lm_head rows {out_f}"
2974 );
2975 let mut gathered = Vec::with_capacity(d2t.len() * row_bytes);
2976 for &t in d2t {
2977 let off = t as usize * row_bytes;
2978 gathered.extend_from_slice(&v.bytes[off..off + row_bytes]);
2979 }
2980 let want_nvfp4 =
2981 want_nvfp4_env && matches!(v.ggml_type, GgmlType::BF16) && v.ne[0].is_multiple_of(64);
2982 if want_nvfp4 {
2983 let in_f = v.ne[0] as usize;
2984 let vals: Vec<f32> = gathered
2985 .chunks_exact(2)
2986 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
2987 .collect();
2988 debug_assert_eq!(vals.len(), d2t.len() * in_f);
2989 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
2990 let sizes = (blocks.len(), gathered.len());
2991 let trimmed = GpuTensor::from_quant_bytes(
2992 e,
2993 &blocks,
2994 GgmlType::NVFP4,
2995 v.ne[0],
2996 d2t.len() as u64,
2997 1.0,
2998 )?;
2999 Ok((trimmed, Some(sizes)))
3000 } else {
3001 let trimmed = match v.ggml_type {
3002 GgmlType::BF16 => GpuTensor::FloatBf16 {
3003 data: e.htod_bytes(&gathered)?,
3004 ne: vec![v.ne[0], d2t.len() as u64],
3005 },
3006 GgmlType::F32 => GpuTensor::Float {
3007 data: e.htod(
3008 &gathered
3009 .chunks_exact(4)
3010 .map(|c| f32::from_le_bytes(c.try_into().unwrap()))
3011 .collect::<Vec<f32>>(),
3012 )?,
3013 ne: vec![v.ne[0], d2t.len() as u64],
3014 },
3015 _ => GpuTensor::from_quant_bytes(
3016 e,
3017 &gathered,
3018 v.ggml_type,
3019 v.ne[0],
3020 d2t.len() as u64,
3021 macro_scale,
3022 )?,
3023 };
3024 Ok((trimmed, None))
3025 }
3026}
3027
3028pub struct MtpHead {
3029 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>>,
3043 pub d2t_from_target_head: bool,
3047 pub geom: Option<DraftGeom>,
3053 pub step35: Option<Step35MtpGeom>,
3058}
3059
3060#[derive(Debug, Clone)]
3074pub struct Step35MtpGeom {
3075 pub il: u32,
3077 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>,
3088}
3089
3090impl Step35MtpGeom {
3091 pub fn from_plan(layer: &memra_gguf::model_plan::LayerPlan) -> Result<Self, String> {
3093 use memra_gguf::model_plan::{ActivationPlan, AttentionPlan};
3094
3095 let (attention, window) = match &layer.attention {
3096 AttentionPlan::Full(attention) => (attention, None),
3097 AttentionPlan::SlidingWindow { attention, window } => (attention, Some(*window)),
3098 other => {
3099 return Err(format!(
3100 "MTP block {} has unsupported tuned attention {other:?}",
3101 layer.index
3102 ));
3103 }
3104 };
3105 if attention.output_gate != memra_gguf::config::AttentionGateKind::SeparateHead {
3106 return Err(format!(
3107 "MTP block {} does not declare a separate attention gate",
3108 layer.index
3109 ));
3110 }
3111 let activation = match &layer.mlp {
3112 MlpPlan::Dense(dense) => &dense.activation,
3113 MlpPlan::Moe(moe) => &moe.activation,
3114 };
3115 let clamp_shexp = match activation {
3116 ActivationPlan::SwiGluClamped { limit } if *limit > 0.0 => Some(*limit),
3117 _ => None,
3118 };
3119 Ok(Step35MtpGeom {
3120 il: layer.index,
3121 n_head: attention.query_heads as usize,
3122 n_head_kv: attention.kv_heads as usize,
3123 n_rot: attention.rope.dimensions as usize,
3124 rope_base: attention.rope.base,
3125 swa: window.is_some(),
3126 window: window.unwrap_or(0) as usize,
3127 clamp_shexp,
3128 })
3129 }
3130}
3131
3132pub struct DraftGeom {
3134 pub d_inner: usize, pub n_head: usize, pub n_head_kv: usize,
3137 pub out_up: GpuTensor, }
3139
3140pub fn draft_head_tensor(has: impl Fn(&str) -> bool, n: u32) -> String {
3149 let own = format!("blk.{n}.nextn.shared_head_head.weight");
3150 if has(&own) {
3151 return own;
3152 }
3153 let legacy = format!("blk.{n}.nextn.shared_head.weight");
3156 if has(&legacy) {
3157 return legacy;
3158 }
3159 "output.weight".to_string()
3161}
3162
3163impl MtpHead {
3164 pub fn load_draft(
3171 e: &Engine,
3172 g: &GgufFile,
3173 main_cfg: &ModelConfig,
3174 ) -> Result<Self, Box<dyn std::error::Error>> {
3175 let src = GgufSource(g);
3176 let dcfg = src.try_config().map_err(std::io::Error::other)?;
3177 let draft_plan = match memra_gguf::model_packs::for_config(&dcfg) {
3178 Some(pack) => pack.compile_plan(&dcfg)?,
3179 None => memra_gguf::model_plan::ModelPlan::compile(&dcfg)?,
3180 };
3181 let main_plan = match memra_gguf::model_packs::for_config(main_cfg) {
3182 Some(pack) => pack.compile_plan(main_cfg)?,
3183 None => memra_gguf::model_plan::ModelPlan::compile(main_cfg)?,
3184 };
3185 if dcfg.nextn_predict_layers == 0 {
3190 return Err(format!(
3191 "draft GGUF has no nextn_predict_layers (arch {:?}) — not a NextN/MTP regime \
3192 draft; gemma assistant drafters attach via MEMRA_DRAFT, not '+draft'",
3193 g.arch()
3194 )
3195 .into());
3196 }
3197 let n = dcfg.n_layer - dcfg.nextn_predict_layers;
3198 let draft_block = draft_plan
3199 .mtp_blocks
3200 .iter()
3201 .find(|block| block.layer.index == n)
3202 .ok_or_else(|| format!("draft ModelPlan has no MTP block {n}"))?;
3203 let p = |s: &str| format!("blk.{n}.{s}");
3204
3205 let student = src.has(&p("nextn.out_up.weight"));
3209 assert_eq!(dcfg.n_embd, main_cfg.n_embd, "draft n_embd != model n_embd");
3210 assert_eq!(
3211 dcfg.head_dim_k, main_cfg.head_dim_k,
3212 "draft head_dim != model head_dim"
3213 );
3214 let main_sliding_gated = crate::plan_backend::decode_batch_program(&main_plan)
3221 == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
3222 let draft_sliding_gated = crate::plan_backend::decode_batch_program(&draft_plan)
3223 == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
3224 let step35 = match (main_sliding_gated, draft_sliding_gated) {
3225 (true, true) => {
3226 let g = Step35MtpGeom::from_plan(&draft_block.layer)?;
3227 let out_f = |t: &str| -> Option<usize> {
3229 src.find(&p(t))
3230 .and_then(|v| v.ne.get(1).copied())
3231 .map(|x| x as usize)
3232 };
3233 let hd = dcfg.head_dim_k as usize;
3234 let wq_out =
3235 out_f("attn_q.weight").ok_or("step35 draft block has no attn_q.weight")?;
3236 assert_eq!(
3237 wq_out,
3238 g.n_head * hd,
3239 "step35 draft blk.{n}: attn_q out {wq_out} != n_head({}) * head_dim({hd}) — \
3240 the draft file's head_count array disagrees with its own tensors",
3241 g.n_head
3242 );
3243 let wg_out = out_f("attn_gate.weight")
3246 .ok_or("step35 draft block has no attn_gate.weight (head-wise gate)")?;
3247 assert_eq!(
3248 wg_out, g.n_head,
3249 "step35 draft blk.{n}: attn_gate out {wg_out} != n_head({})",
3250 g.n_head
3251 );
3252 assert_eq!(
3256 g.n_head_kv, main_cfg.n_head_kv as usize,
3257 "step35 draft blk.{n} KV heads {} != trunk n_head_kv {} — the MTP scratch \
3258 rows are sized from the trunk cfg, so a differing draft KV width would \
3259 write past the row",
3260 g.n_head_kv, main_cfg.n_head_kv
3261 );
3262 eprintln!(
3263 "[mtp-draft] step35 MTP geometry blk.{n}: n_head={} n_head_kv={} n_rot={} \
3264 rope_base={:.0} swa={} window={}",
3265 g.n_head, g.n_head_kv, g.n_rot, g.rope_base, g.swa, g.window
3266 );
3267 Some(g)
3268 }
3269 (true, false) => {
3270 return Err(format!(
3271 "MEMRA_MTP_DRAFT operations are incompatible with the model's \
3272 sliding-gated-MoE program (draft arch {:?})",
3273 g.arch()
3274 )
3275 .into());
3276 }
3277 (false, true) => {
3278 return Err(
3279 "MEMRA_MTP_DRAFT requires sliding-gated-MoE operations but the model does not"
3280 .into(),
3281 );
3282 }
3283 (false, false) => None,
3284 };
3285 if step35.is_none() && !student {
3286 assert_eq!(dcfg.n_head, main_cfg.n_head, "draft n_head != model n_head");
3289 assert_eq!(
3290 dcfg.n_head_kv, main_cfg.n_head_kv,
3291 "draft n_head_kv != model n_head_kv"
3292 );
3293 }
3294
3295 let head_name = draft_head_tensor(|t| src.has(t), n);
3322 let head = load_t(e, &src, &head_name)?;
3323 let head_norm = match load_opt(e, &src, &p("nextn.shared_head_norm.weight"))? {
3324 Some(t) => Some(t),
3325 None => load_opt(e, &src, "output_norm.weight")?,
3326 };
3327
3328 let d2t: Option<Vec<u32>> = g.find("d2t").map(|t| {
3330 let bytes = g.tensor_data(t);
3331 match t.ggml_type {
3332 GgmlType::I32 => bytes
3333 .chunks_exact(4)
3334 .map(|c| i32::from_le_bytes(c.try_into().unwrap()) as u32)
3335 .collect(),
3336 GgmlType::I64 => bytes
3337 .chunks_exact(8)
3338 .map(|c| i64::from_le_bytes(c.try_into().unwrap()) as u32)
3339 .collect(),
3340 other => panic!("d2t must be I32/I64, got {other:?}"),
3341 }
3342 });
3343 if let Some(map) = &d2t {
3344 assert_eq!(
3345 map.len(),
3346 head.out_features(),
3347 "d2t len {} != draft head rows {}",
3348 map.len(),
3349 head.out_features()
3350 );
3351 let n_vocab = main_cfg.n_vocab as u64;
3352 assert!(
3353 map.iter().all(|&t| (t as u64) < n_vocab),
3354 "d2t contains token id >= model n_vocab {n_vocab}"
3355 );
3356 }
3357 let eh_proj = load_t(e, &src, &p("nextn.eh_proj.weight"))?;
3358 assert_eq!(
3361 eh_proj.in_features(),
3362 2 * main_cfg.n_embd as usize,
3363 "eh_proj in dim != 2*n_embd"
3364 );
3365 let geom = if student {
3366 let out_up = load_t(e, &src, &p("nextn.out_up.weight"))?;
3367 let d_inner = eh_proj.out_features();
3368 assert_eq!(
3369 out_up.out_features(),
3370 main_cfg.n_embd as usize,
3371 "out_up out dim != n_embd"
3372 );
3373 assert_eq!(
3374 out_up.in_features(),
3375 d_inner,
3376 "out_up in dim != eh_proj out dim (d_inner)"
3377 );
3378 assert!(
3379 dcfg.n_head >= 1 && dcfg.n_head_kv >= 1 && dcfg.n_head % dcfg.n_head_kv == 0,
3380 "student head counts malformed ({}/{})",
3381 dcfg.n_head,
3382 dcfg.n_head_kv
3383 );
3384 Some(DraftGeom {
3385 d_inner,
3386 n_head: dcfg.n_head as usize,
3387 n_head_kv: dcfg.n_head_kv as usize,
3388 out_up,
3389 })
3390 } else {
3391 None
3392 };
3393 let blk_prefix = format!("blk.{n}.");
3397 let head_src = head_name.strip_prefix(&blk_prefix).unwrap_or(&head_name);
3398 eprintln!(
3399 "[mtp-draft] external draft head: blk.{n}, source={}, head_vocab={}{}{}",
3400 head_src,
3401 head.out_features(),
3402 if d2t.is_some() {
3403 " (trimmed, d2t map)"
3404 } else {
3405 " (full)"
3406 },
3407 match &geom {
3408 Some(g) => format!(
3409 " (student d_inner={} heads={}/{})",
3410 g.d_inner, g.n_head, g.n_head_kv
3411 ),
3412 None => String::new(),
3413 }
3414 );
3415
3416 let mut resident = ResidentPlan::unsharded(e, &src, &dcfg);
3417 let mut step_runtimes = StepParallelRuntimeRegistry::default();
3418 Ok(MtpHead {
3419 enorm: load_t(e, &src, &p("nextn.enorm.weight"))?,
3420 hnorm: load_t(e, &src, &p("nextn.hnorm.weight"))?,
3421 eh_proj,
3422 attn_norm: load_t(e, &src, &p("attn_norm.weight"))?,
3423 post_attn_norm: load_opt(e, &src, &p("post_attention_norm.weight"))?
3424 .or(load_opt(e, &src, &p("ffn_norm.weight"))?)
3425 .expect("draft NextN block needs post_attention_norm or ffn_norm"),
3426 mixer: load_mixer_kind(
3427 e,
3428 &src,
3429 &dcfg,
3430 n,
3431 &draft_block.layer.attention,
3432 &mut step_runtimes,
3433 )?,
3434 ffn: load_ffn(
3435 e,
3436 &src,
3437 &dcfg,
3438 &draft_block.layer.mlp,
3439 n,
3440 None,
3441 &mut resident,
3442 &mut step_runtimes,
3443 )?,
3444 shared_head_norm: head_norm,
3445 shared_head_head: Some(head),
3446 d2t,
3447 d2t_from_target_head: false,
3448 geom,
3449 step35,
3450 })
3451 }
3452}
3453
3454pub struct GemmaAux {
3456 pub rope_freqs: Option<Vec<(usize, CudaSlice<f32>)>>,
3459 pub ones: Vec<(usize, CudaSlice<f32>)>,
3462 pub suppress_d: Option<(CudaSlice<i32>, usize)>,
3465 pub e4b: Option<Gemma4E4bModel>,
3467}
3468
3469impl GemmaAux {
3470 pub fn rope_freqs(&self, e: &Engine) -> Option<&CudaSlice<f32>> {
3471 self.rope_freqs.as_ref().map(|copies| {
3472 let dev = e.ctx().ordinal();
3473 &copies
3474 .iter()
3475 .find(|(d, _)| *d == dev)
3476 .unwrap_or_else(|| panic!("gemma4 rope_freqs has no local copy for device {dev}"))
3477 .1
3478 })
3479 }
3480
3481 pub fn ones(&self, e: &Engine) -> &CudaSlice<f32> {
3482 let dev = e.ctx().ordinal();
3483 &self
3484 .ones
3485 .iter()
3486 .find(|(d, _)| *d == dev)
3487 .unwrap_or_else(|| panic!("gemma4 ones has no local copy for device {dev}"))
3488 .1
3489 }
3490}
3491
3492pub struct Step35Aux {
3495 pub rope_freqs: Option<Vec<(usize, CudaSlice<f32>)>>,
3501}
3502
3503impl Step35Aux {
3504 pub fn rope_freqs(&self, e: &Engine) -> Option<&CudaSlice<f32>> {
3505 self.rope_freqs.as_ref().map(|copies| {
3506 let dev = e.ctx().ordinal();
3507 &copies
3508 .iter()
3509 .find(|(d, _)| *d == dev)
3510 .unwrap_or_else(|| panic!("step35 rope_freqs has no local copy for device {dev}"))
3511 .1
3512 })
3513 }
3514}
3515
3516pub struct HybridModel {
3517 pub cfg: ModelConfig,
3518 pub plan: memra_gguf::model_plan::ModelPlan,
3519 pub rewrite_qualifications: Option<memra_gguf::execution_manifest::RewriteQualifications>,
3520 pub embd: EmbedHost,
3521 pub output_norm: GpuTensor,
3522 pub output: GpuTensor,
3523 pub layers: Vec<HybridLayer>,
3524 pub mtp: Option<MtpHead>, pub mtp_extra: Vec<MtpHead>,
3528 pub dflash_trim: Option<DflashTrimHead>,
3532 pub embd_gpu: std::sync::OnceLock<cudarc::driver::CudaSlice<u8>>,
3535 pub gemma4_aux: Option<GemmaAux>,
3536 pub step35_aux: Option<Step35Aux>,
3538 pub prime_slabs: std::sync::Mutex<
3546 std::collections::HashMap<
3547 usize,
3548 std::sync::Arc<std::sync::Mutex<crate::hybrid_forward::PrimeSlabs>>,
3549 >,
3550 >,
3551 pub(crate) dspark_vgraphs: std::sync::Mutex<Option<crate::spec::DsparkVerifyGraphs>>,
3564 pub(crate) step_grouped_prefill: std::sync::Mutex<StepEpGroupedPrefill>,
3570 pub(crate) step35_token_graph:
3573 std::sync::Mutex<Option<crate::hybrid_forward::Step35TokenGraphState>>,
3574 pub hyper: Option<crate::hyper::HyperTopology>,
3579 pub hyper_head: Option<crate::hyper::HyperHead>,
3582 pub glm5_dflash: Option<crate::glm_spec::Glm5DflashDrafter>,
3589 pub(crate) draft_state_bytes: std::sync::atomic::AtomicUsize,
3598}
3599
3600impl HybridModel {
3601 pub fn install_rewrite_bundle(
3602 &mut self,
3603 bundle: &std::path::Path,
3604 ) -> Result<(), Box<dyn std::error::Error>> {
3605 self.rewrite_qualifications = Some(
3606 memra_gguf::execution_manifest::RewriteQualifications::load(bundle, &self.plan)
3607 .map_err(|error| format!("rewrite qualification: {error}"))?,
3608 );
3609 Ok(())
3610 }
3611
3612 pub fn rewrite_allowed(&self, surface: memra_gguf::execution_manifest::RewriteSurface) -> bool {
3613 self.rewrite_qualifications
3614 .as_ref()
3615 .is_none_or(|qualifications| qualifications.allows(surface))
3616 }
3617
3618 pub fn record_draft_state_bytes(&self, observed: usize) -> Option<usize> {
3623 use std::sync::atomic::Ordering;
3624 let prev = self
3625 .draft_state_bytes
3626 .fetch_max(observed, Ordering::Relaxed);
3627 (observed > prev).then_some(observed)
3628 }
3629
3630 pub fn draft_session_admission_bytes(&self) -> usize {
3636 self.draft_state_bytes
3637 .load(std::sync::atomic::Ordering::Relaxed)
3638 }
3639
3640 pub fn step_tp_unmaterialized_kv_bytes(
3646 &self,
3647 cache: Option<&crate::cache::Cache>,
3648 capacity: usize,
3649 ) -> Result<Vec<StepTpKvDeviceAdmission>, String> {
3650 if let Some(cache) = cache
3651 && cache.tp_kv.len() < self.layers.len()
3652 {
3653 return Err(format!(
3654 "Step TP admission cache has {} layers, model trunk has {}",
3655 cache.tp_kv.len(),
3656 self.layers.len()
3657 ));
3658 }
3659
3660 let mut by_device: HashMap<usize, usize> = HashMap::new();
3661 for (layer, weights) in self.layers.iter().enumerate() {
3662 let Mixer::Full(attention) = &weights.mixer else {
3663 continue;
3664 };
3665 let Some(tp) = attention
3666 .step_tp_qkv
3667 .as_ref()
3668 .filter(|tp| tp.attention.is_some())
3669 else {
3670 continue;
3671 };
3672 if cache.is_some_and(|cache| cache.tp_kv[layer].is_some()) {
3673 continue;
3674 }
3675 let geometry = self.cfg.full_attention_geometry_at(layer as u32);
3676 let shape = crate::cache::tp_kv_rank_allocation_shape(
3677 geometry.n_head_kv as usize * geometry.head_dim_k as usize,
3678 geometry.n_head_kv as usize * geometry.head_dim_v as usize,
3679 tp.devices.len(),
3680 )?;
3681 let physical_rows = geometry
3682 .window
3683 .map(|window| crate::cache::swa_ring_rows(window as usize, capacity))
3684 .unwrap_or(capacity);
3685 let bytes = shape.allocation_bytes(physical_rows);
3686 for &device in &tp.devices {
3687 let total = by_device.entry(device).or_default();
3688 *total = total.saturating_add(bytes);
3689 }
3690 }
3691
3692 let mut out: Vec<_> = by_device
3693 .into_iter()
3694 .map(|(device, bytes)| StepTpKvDeviceAdmission { device, bytes })
3695 .collect();
3696 out.sort_unstable_by_key(|charge| charge.device);
3697 Ok(out)
3698 }
3699
3700 pub fn step_tp_rank_engine(&self, device: usize) -> Option<&Engine> {
3702 self.layers.iter().find_map(|weights| {
3703 let Mixer::Full(attention) = &weights.mixer else {
3704 return None;
3705 };
3706 let tp = attention.step_tp_qkv.as_ref()?;
3707 let rank = tp
3708 .runtime
3709 .devices()
3710 .iter()
3711 .position(|&rank| rank == device)?;
3712 tp.runtime.rank_engine(rank)
3713 })
3714 }
3715
3716 pub(crate) fn step_tp_runtime_for_layer(
3717 &self,
3718 layer: usize,
3719 ) -> Option<&crate::tp::TpE4m3HostBounce> {
3720 let Mixer::Full(attention) = &self.layers.get(layer)?.mixer else {
3721 return None;
3722 };
3723 let tp = attention.step_tp_qkv.as_ref()?;
3724 tp.attention.as_ref()?;
3725 Some(tp.runtime.as_ref())
3726 }
3727
3728 pub fn decode_batch_program(&self) -> crate::plan_backend::DecodeBatchProgram {
3729 crate::plan_backend::decode_batch_program(&self.plan)
3730 }
3731
3732 pub fn uses_gemma_program(&self) -> bool {
3733 self.decode_batch_program() == crate::plan_backend::DecodeBatchProgram::Gemma
3734 }
3735
3736 pub fn uses_sliding_gated_moe_program(&self) -> bool {
3737 self.decode_batch_program() == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe
3738 }
3739
3740 pub fn has_plan_operation(&self, operation: memra_gguf::model_plan::OperationKind) -> bool {
3741 self.plan.trunk_operations().contains(&operation)
3742 }
3743
3744 pub fn load(e: &Engine, g: &GgufFile) -> Result<Self, Box<dyn std::error::Error>> {
3746 Self::load_from_source(e, &GgufSource(g))
3747 }
3748
3749 pub fn load_without_mtp(e: &Engine, g: &GgufFile) -> Result<Self, Box<dyn std::error::Error>> {
3752 Self::load_from_source_impl(e, &GgufSource(g), false)
3753 }
3754
3755 pub fn load_from_source(
3759 e: &Engine,
3760 src: &dyn TensorSource,
3761 ) -> Result<Self, Box<dyn std::error::Error>> {
3762 Self::load_from_source_impl(e, src, true)
3763 }
3764
3765 pub fn load_from_source_without_mtp(
3767 e: &Engine,
3768 src: &dyn TensorSource,
3769 ) -> Result<Self, Box<dyn std::error::Error>> {
3770 Self::load_from_source_impl(e, src, false)
3771 }
3772
3773 fn load_from_source_impl(
3774 e: &Engine,
3775 src: &dyn TensorSource,
3776 load_mtp: bool,
3777 ) -> Result<Self, Box<dyn std::error::Error>> {
3778 let cfg = src.try_config().map_err(std::io::Error::other)?;
3779 let plan = match memra_gguf::model_packs::for_config(&cfg) {
3780 Some(pack) => pack.compile_plan(&cfg)?,
3781 None => memra_gguf::model_plan::ModelPlan::compile(&cfg)?,
3782 };
3783 let auto_parallel = prepare_auto_parallel(src, &cfg, &plan)?;
3784 let batch_program = crate::plan_backend::decode_batch_program(&plan);
3785 let gemma_program = batch_program == crate::plan_backend::DecodeBatchProgram::Gemma;
3786 let sliding_gated_moe_program =
3787 batch_program == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
3788 if matches!(
3789 src.expert_activation_precision(),
3790 memra_gguf::source::ExpertActivationPrecision::Bf16
3791 ) {
3792 eprintln!(
3793 "[w4a16] artifact contract accepted: expert_weights=nvfp4 \
3794 expert_activations=bf16-rounded q8_expert_program=disabled"
3795 );
3796 }
3797 if sliding_gated_moe_program {
3802 crate::arm_step37_serving_defaults();
3803 }
3804 cfg.validate_attention_gate_layout()?;
3809 if cfg.sigmoid_router().is_some() {
3816 let host_oracle = std::env::var("MEMRA_SIG_ROUTER").as_deref() == Ok("0");
3817 match crate::sigrouter_contract::verify_host_expf() {
3818 Ok(()) => {}
3819 Err(e) if host_oracle => return Err(e.into()),
3820 Err(e) => eprintln!(
3821 "[sigrouter] WARN: host expf probe mismatch ({e}); device routing is \
3822 unaffected, but host-oracle replay/comparison cells are invalid on this host"
3823 ),
3824 }
3825 }
3826 if std::env::var("MEMRA_DRAFT").is_ok() && std::env::var("MEMRA_MMQ_SK").is_err() {
3835 let force = if cfg.n_embd >= 3500 { 0i8 } else { -1i8 };
3836 crate::MMQ_SK_FORCE.store(force, std::sync::atomic::Ordering::Relaxed);
3837 }
3838 crate::KV_FP8_FORCE.store(0, std::sync::atomic::Ordering::Relaxed);
3842
3843 let n_trunk = (cfg.n_layer - cfg.nextn_predict_layers) as usize;
3848 let mtp_skip_requested = load_mtp
3858 && match std::env::var("MEMRA_MTP_SKIP").ok().as_deref() {
3859 None | Some("") | Some("0") => false,
3860 Some("1") => true,
3861 Some(other) => {
3862 return Err(format!(
3863 "MEMRA_MTP_SKIP={other:?}: expected 1 (skip the embedded MTP block) or \
3864 0/unset (load it); refusing to guess"
3865 )
3866 .into());
3867 }
3868 };
3869 if mtp_skip_requested && std::env::var("MEMRA_MTP_DRAFT").is_ok_and(|p| !p.is_empty()) {
3870 return Err(
3871 "MEMRA_MTP_SKIP=1 together with MEMRA_MTP_DRAFT is contradictory: the skip \
3872 removes the MTP head to reclaim VRAM while MEMRA_MTP_DRAFT attaches an \
3873 external MTP head for MTP spec decode; unset one"
3874 .into(),
3875 );
3876 }
3877 if mtp_skip_requested && cfg.nextn_predict_layers > 0 {
3878 let prefixes: Vec<String> = (0..cfg.nextn_predict_layers)
3883 .map(|off| format!("blk.{}.", n_trunk as u32 + off))
3884 .collect();
3885 let skipped_bytes: Option<u64> = src.gguf().map(|g| {
3886 g.tensors
3887 .iter()
3888 .filter(|t| prefixes.iter().any(|p| t.name.starts_with(p.as_str())))
3889 .map(|t| t.n_bytes)
3890 .sum()
3891 });
3892 eprintln!(
3893 "[mtp-skip] MEMRA_MTP_SKIP=1: skipping {} embedded MTP/NextN block(s) \
3894 blk.{}..=blk.{} ({}); MTP spec decode is unavailable for this model \
3895 (dspark/DFlash2 drafting keeps its trimmed head via the MEMRA_FRSPEC_TRIM stub)",
3896 cfg.nextn_predict_layers,
3897 n_trunk,
3898 n_trunk as u32 + cfg.nextn_predict_layers - 1,
3899 match skipped_bytes {
3900 Some(b) => format!("~{} MiB of weights not loaded", b >> 20),
3901 None => "size unknown: non-GGUF source".to_string(),
3902 },
3903 );
3904 }
3905 let mtp_skip_trim_d2t: Option<Vec<u32>> = if mtp_skip_requested
3920 && cfg.nextn_predict_layers > 0
3921 && !crate::model::full_prec_enabled()
3922 {
3923 match std::env::var("MEMRA_FRSPEC_TRIM") {
3924 Ok(path) if !path.is_empty() => {
3925 let path = memra_gguf::hf::resolve_arg(&path)
3926 .map_err(|err| format!("MEMRA_FRSPEC_TRIM={path:?}: {err}"))?;
3927 let own_head_name = frspec_trim_own_head_name(n_trunk);
3928 if src.has(&own_head_name) {
3929 return Err(format!(
3930 "MEMRA_MTP_SKIP=1 with MEMRA_FRSPEC_TRIM: this artifact ships its \
3931 own MTP-block lm_head ({own_head_name}), so the trimmed draft rows \
3932 live in the block being skipped; gathering trunk rows instead is \
3933 the wrong-head bug (acceptance 0/248 receipt, \
3934 frspec_trim_own_head_name). Unset MEMRA_MTP_SKIP or \
3935 MEMRA_FRSPEC_TRIM"
3936 )
3937 .into());
3938 }
3939 if !src.has("output.weight") && !src.has("token_embd.weight") {
3940 return Err("MEMRA_MTP_SKIP=1 with MEMRA_FRSPEC_TRIM: model has no \
3941 output.weight (or tied token_embd.weight) to gather trimmed draft \
3942 rows from"
3943 .into());
3944 }
3945 let d2t = frspec_read_d2t(&path)?;
3946 if d2t.is_empty() {
3947 return Err(format!(
3948 "MEMRA_MTP_SKIP=1 with MEMRA_FRSPEC_TRIM={path}: the rank artifact \
3949 yields an EMPTY d2t list, so no stub draft head can be built; fix \
3950 the artifact or unset MEMRA_MTP_SKIP"
3951 )
3952 .into());
3953 }
3954 Some(d2t)
3955 }
3956 _ => None,
3957 }
3958 } else {
3959 None
3960 };
3961 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
3962 let pipeline = crate::plan_backend::PIPELINE
3963 .trunk_capabilities(&plan)
3964 .pipeline;
3965 let qualified_gemma_pp2 = gemma_program && fence.len() == 3;
3969 if !pipeline.supported && !qualified_gemma_pp2 {
3970 return Err(format!(
3971 "pipeline placement is unsupported for plan operations {:?}; blockers={:?}",
3972 plan.trunk_operations(),
3973 pipeline.blockers,
3974 )
3975 .into());
3976 }
3977 let illegal = illegal_pipeline_cuts(&fence, &plan.partition_boundaries);
3978 if !illegal.is_empty() {
3979 return Err(format!(
3980 "pipeline placement cuts {illegal:?} split outside ModelPlan legal boundaries {:?}",
3981 plan.partition_boundaries,
3982 )
3983 .into());
3984 }
3985 }
3986 crate::pp::init_model_transport(e, &cfg, n_trunk)?;
3987 let step_parallel =
3988 prepare_step_parallel_load(e, src, &cfg, n_trunk, auto_parallel.as_ref())?;
3989 let glm5_tp = if crate::glm5_tp::glm5_tp_armed() {
3993 use memra_gguf::model_plan::{AttentionPlan, MlpPlan};
3994 let moe = cfg.moe.as_ref().ok_or(
3995 "MEMRA_GLM5_TP requires a MoE model (glm5_next); this plan carries no MoE \
3996 metadata",
3997 )?;
3998 let mut layer_class = Vec::with_capacity(n_trunk);
3999 let mut layer_is_moe = Vec::with_capacity(n_trunk);
4000 let (mut kda_heads, mut kda_head_dim, mut mla_heads) = (0usize, 0usize, 0usize);
4001 for (il, lp) in plan.layers.iter().take(n_trunk).enumerate() {
4002 match &lp.attention {
4003 AttentionPlan::KimiDeltaNet(k) => {
4004 layer_class.push(crate::glm5_tp::Glm5LayerClass::Kda);
4005 kda_heads = k.num_heads as usize;
4006 kda_head_dim = k.head_dim as usize;
4007 }
4008 AttentionPlan::Mla(memra_gguf::model_plan::MlaAttentionPlan::LatentKv {
4009 query_heads,
4010 ..
4011 }) => {
4012 layer_class.push(crate::glm5_tp::Glm5LayerClass::Mla);
4013 mla_heads = *query_heads as usize;
4014 }
4015 other => {
4016 return Err(format!(
4017 "MEMRA_GLM5_TP requires a glm5_next-class plan (KDA/MLA mixers): \
4018 trunk layer {il} declares {other:?}"
4019 )
4020 .into());
4021 }
4022 }
4023 layer_is_moe.push(matches!(&lp.mlp, MlpPlan::Moe(_)));
4024 }
4025 let view = crate::glm5_tp::Glm5TpModelView {
4026 trunk_layers: n_trunk,
4027 layer_class,
4028 layer_is_moe,
4029 kda_heads,
4030 kda_head_dim,
4031 mla_heads,
4032 n_routed_experts: moe.expert_count as usize,
4033 top_k: moe.expert_used_count as usize,
4034 };
4035 crate::glm5_tp::prepare_glm5_tp_load(e, &view)?
4036 } else {
4037 let glm5_class = plan.layers.iter().take(n_trunk).any(|lp| {
4044 matches!(
4045 lp.attention,
4046 memra_gguf::model_plan::AttentionPlan::KimiDeltaNet(_)
4047 )
4048 });
4049 let ep_map_armed = crate::ep_map::ep_map_env()?;
4050 if let Some((flag, _)) = ep_map_armed
4051 && glm5_class
4052 {
4053 return Err(format!(
4054 "{flag} is set but MEMRA_GLM5_TP is off: the map cannot \
4055 engage, and a placement that silently reverts to the even split is \
4056 refused by name (unset one of the two)"
4057 )
4058 .into());
4059 }
4060 if glm5_class {
4065 for flag in ["MEMRA_GLM5_EP_DIET", "MEMRA_GLM5_EP_GROUPED_PRIME"] {
4066 if std::env::var(flag).as_deref() == Ok("1") {
4067 return Err(format!(
4068 "{flag}=1 is set but MEMRA_GLM5_TP is off: the EP dispatch \
4069 diet only exists inside the TP-2 EP walk and cannot engage \
4070 (unset one of the two)"
4071 )
4072 .into());
4073 }
4074 }
4075 }
4076 None
4077 };
4078 let embd = EmbedHost::from_source(src, "token_embd.weight");
4079 let e_head = crate::pp::layer_engine(e, n_trunk, n_trunk - 1)?;
4083 let output_norm = load_t(e_head, src, "output_norm.weight")?;
4084 let mut output = if src.has("output.weight") {
4086 load_t(e_head, src, "output.weight")?
4087 } else {
4088 load_t(e_head, src, "token_embd.weight")?
4089 };
4090 let mut resident = ResidentPlan::pp(e, src, &cfg, n_trunk)?;
4091 resident.exclude_distributed_expert_layers(
4092 step_parallel
4093 .ep_specs
4094 .iter()
4095 .map(|spec| spec.layer)
4096 .chain(step_parallel.tp_specs.iter().map(|spec| spec.layer)),
4097 );
4098 let mut step_runtimes = StepParallelRuntimeRegistry::with_config(step_parallel);
4099
4100 let gguf: Option<&GgufFile> = src.gguf();
4107 let mut spill: Option<crate::spill::SpillCtx> = if cfg
4110 .moe
4111 .as_ref()
4112 .is_some_and(|m| m.expert_count > 0)
4113 && crate::spill::disk_tier_enabled()
4114 && gguf.is_some()
4115 {
4116 let budget = crate::spill::MemBudget::probe(e)?;
4117 #[allow(clippy::unnecessary_unwrap)]
4118 let ctx = crate::spill::SpillCtx::open(gguf.unwrap(), &budget)?;
4120 eprintln!(
4121 "[spill] disk tier ON: free_vram={} MiB free_pinnable_ram={} MiB (MemAvailable*resolved_frac)",
4122 budget.free_vram >> 20,
4123 budget.free_pinnable_ram >> 20
4124 );
4125 Some(ctx)
4126 } else {
4127 None
4128 };
4129
4130 let hyper = crate::hyper::HyperTopology::from_plan(&plan)?;
4137 let hyper_head = match hyper.as_ref() {
4138 Some(topology) => {
4139 crate::hyper::HyperHead::load(e_head, src, topology, cfg.n_embd as usize)?
4140 }
4141 None => None,
4142 };
4143 let mut layers = Vec::with_capacity(n_trunk);
4144 for il in 0..n_trunk as u32 {
4145 let p = |s: &str| format!("blk.{il}.{s}");
4146 let layer_plan = plan
4147 .layers
4148 .get(il as usize)
4149 .ok_or_else(|| format!("ModelPlan has no trunk layer {il}"))?;
4150 let e = crate::pp::layer_engine(e, n_trunk, il as usize)?;
4154 layers.push(HybridLayer {
4156 attn_norm: load_t(e, src, &p("attn_norm.weight"))?,
4157 post_attn_norm: load_opt(e, src, &p("post_attention_norm.weight"))?
4158 .or(load_opt(e, src, &p("ffn_norm.weight"))?)
4159 .expect("need post_attention_norm or ffn_norm"),
4160 mixer: {
4161 let g4_shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
4165 let kv_from = n_trunk as u32 - g4_shared;
4166 if g4_shared > 0
4167 && il >= kv_from
4168 && !src.has(&format!("blk.{il}.attn_k.weight"))
4169 {
4170 let g4 = cfg.gemma4.as_ref().unwrap();
4171 let swa = g4.swa_pattern.get(il as usize).copied().unwrap_or(true);
4172 let tgt = kv_from - if swa { 2 } else { 1 };
4173 let tp = |s: &str| format!("blk.{tgt}.{s}");
4174 Mixer::Full(FullAttnLayer {
4175 wq: load_t(e, src, &p("attn_q.weight"))?,
4176 wk: load_t(e, src, &tp("attn_k.weight"))?,
4177 wv: load_t(e, src, &tp("attn_v.weight"))?,
4178 wo: load_t(e, src, &p("attn_output.weight"))?,
4179 q_norm: load_t(e, src, &p("attn_q_norm.weight"))?,
4180 k_norm: load_t(e, src, &tp("attn_k_norm.weight"))?,
4181 attn_gate: None, step_tp_qkv: None,
4183 })
4184 } else {
4185 load_mixer_kind(
4186 e,
4187 src,
4188 &cfg,
4189 il,
4190 &layer_plan.attention,
4191 &mut step_runtimes,
4192 )?
4193 }
4194 },
4195 ffn: load_ffn(
4196 e,
4197 src,
4198 &cfg,
4199 &layer_plan.mlp,
4200 il,
4201 spill.as_mut().map(|c| (gguf.unwrap(), c)),
4202 &mut resident,
4203 &mut step_runtimes,
4204 )?,
4205 gemma4: if gemma_program {
4206 let scalar = |n: &str| -> f32 {
4207 let t = src.find(&p(n)).unwrap_or_else(|| panic!("missing {n}"));
4208 memra_gguf::dequant::dequantize(t.ggml_type, &t.bytes, 1)[0]
4209 };
4210 let vecf = |n: &str| -> Vec<f32> {
4211 let t = src.find(&p(n)).unwrap_or_else(|| panic!("missing {n}"));
4212 memra_gguf::dequant::dequantize(
4213 t.ggml_type,
4214 &t.bytes,
4215 t.ne.iter().product::<u64>() as usize,
4216 )
4217 };
4218 let moe_bits = if src.find(&p("ffn_gate_inp.scale")).is_some() {
4219 Some(crate::hybrid::Gemma4MoeBits {
4220 post_ffw_norm_1: load_t(e, src, &p("post_ffw_norm_1.weight"))?,
4221 pre_ffw_norm_2: load_t(e, src, &p("pre_ffw_norm_2.weight"))?,
4222 post_ffw_norm_2: load_t(e, src, &p("post_ffw_norm_2.weight"))?,
4223 shared_gate: load_t(e, src, &p("ffn_gate.weight"))?,
4224 shared_up: load_t(e, src, &p("ffn_up.weight"))?,
4225 shared_down: load_t(e, src, &p("ffn_down.weight"))?,
4226 router_scale_pre: {
4227 let inv = 1.0 / (cfg.n_embd as f32).sqrt();
4228 let v: Vec<f32> =
4229 vecf("ffn_gate_inp.scale").iter().map(|x| x * inv).collect();
4230 e.htod(&v)?
4231 },
4232 per_expert_scale: vecf("ffn_down_exps.scale"),
4233 per_expert_scale_d: e.htod(&vecf("ffn_down_exps.scale"))?,
4234 })
4235 } else {
4236 None
4237 };
4238 let e4b = if src.has(&p("inp_gate.weight")) {
4240 let g4 = cfg.gemma4.as_ref().unwrap();
4241 let kv_from = n_trunk as u32 - g4.shared_kv_layers;
4242 let kv_share = if g4.shared_kv_layers > 0 && il >= kv_from {
4243 let swa = g4.swa_pattern.get(il as usize).copied().unwrap_or(true);
4244 Some(kv_from - if swa { 2 } else { 1 })
4245 } else {
4246 None
4247 };
4248 Some(crate::hybrid::Gemma4E4bLayer {
4249 inp_gate: load_t(e, src, &p("inp_gate.weight"))?,
4250 proj: load_t(e, src, &p("proj.weight"))?,
4251 post_norm: load_t(e, src, &p("post_norm.weight"))?,
4252 kv_share,
4253 qkv_cat: None, })
4255 } else {
4256 None
4257 };
4258 Some(Gemma4LayerBits {
4259 ffn_norm: load_t(e, src, &p("ffn_norm.weight"))?,
4260 post_ffw_norm: load_t(e, src, &p("post_ffw_norm.weight"))?,
4261 moe_bits,
4262 layer_scale: scalar("layer_output_scale.weight"),
4263 e4b,
4264 })
4265 } else {
4266 None
4267 },
4268 hyper: match hyper.as_ref() {
4269 Some(topology) => Some(crate::hyper::HyperLayer::load(
4270 e,
4271 src,
4272 il,
4273 topology,
4274 cfg.n_embd as usize,
4275 )?),
4276 None => None,
4277 },
4278 });
4279 if let Some(tp_plan) = &glm5_tp
4282 && tp_plan.layers.contains(&(il as usize))
4283 {
4284 let mut layer = layers.pop().expect("layer just pushed");
4285 layer.mixer = match layer.mixer {
4286 Mixer::Kda(la) => {
4287 Mixer::Kda(crate::glm5_tp::shard_kda_layer(e, &tp_plan.rt, la)?)
4288 }
4289 Mixer::Mla(la) => {
4290 Mixer::Mla(crate::glm5_tp::shard_mla_layer(e, &tp_plan.rt, la)?)
4291 }
4292 _ => {
4293 return Err(format!(
4294 "MEMRA_GLM5_TP selected layer {il}, whose loaded mixer is not \
4295 KDA/MLA — preflight and loader disagree (wiring bug)"
4296 )
4297 .into());
4298 }
4299 };
4300 if let Ffn::Moe(m) = &mut layer.ffn {
4301 let placement = match &tp_plan.ep_map {
4305 Some(map) => Some(
4306 map.layers
4307 .get(&(il as usize))
4308 .ok_or_else(|| {
4309 format!(
4310 "glm5-tp EP: preflight-validated map lost layer {il} \
4311 (wiring bug)"
4312 )
4313 })?
4314 .as_slice(),
4315 ),
4316 None => None,
4317 };
4318 crate::glm5_tp::arm_moe_ep(e, &tp_plan.rt, m, placement)?;
4319 }
4320 layers.push(layer);
4321 }
4322 }
4323
4324 let external_mtp_requested =
4328 load_mtp && std::env::var("MEMRA_MTP_DRAFT").is_ok_and(|path| !path.is_empty());
4329 let trim_mtp_requested = load_mtp
4330 && !crate::model::full_prec_enabled()
4331 && std::env::var("MEMRA_FRSPEC_TRIM").is_ok_and(|path| !path.is_empty());
4332 let _ = trim_mtp_requested;
4343 let glm5_mtp_requested =
4353 !cfg.arch.is_glm5_next() || std::env::var("MEMRA_GLM5_MTP").as_deref() == Ok("1");
4354 let embedded_head_count =
4357 if external_mtp_requested || !glm5_mtp_requested || mtp_skip_requested {
4358 0
4359 } else {
4360 cfg.nextn_predict_layers
4361 };
4362 if cfg.arch.is_glm5_next()
4363 && glm5_mtp_requested
4364 && !mtp_skip_requested
4365 && cfg.nextn_predict_layers > 0
4366 {
4367 eprintln!("[mtp-glm5] MEMRA_GLM5_MTP=1: loading the glm5_next NextN block");
4368 }
4369 let embedded_head_count = match std::env::var("MEMRA_MTP_HEADS")
4374 .ok()
4375 .and_then(|v| v.parse::<u32>().ok())
4376 .filter(|&n| n > 0)
4377 {
4378 Some(cap) if cap < embedded_head_count => {
4379 eprintln!(
4380 "[mtp-chain] MEMRA_MTP_HEADS={cap}: capping the embedded chain from \
4381 {embedded_head_count} heads (measurement knob)"
4382 );
4383 cap
4384 }
4385 _ => embedded_head_count,
4386 };
4387 let mut embedded_mtp = Vec::new();
4388 if load_mtp && embedded_head_count > 0 {
4389 for offset in 0..embedded_head_count {
4390 let n = n_trunk as u32 + offset;
4391 let e = crate::pp::layer_engine(e, n_trunk, n as usize)?;
4397 let p = |s: &str| format!("blk.{n}.{s}");
4398 let mtp_plan = plan
4399 .mtp_blocks
4400 .iter()
4401 .find(|block| block.layer.index == n)
4402 .ok_or_else(|| format!("ModelPlan has no embedded MTP block {n}"))?;
4403 if !src.has(&p("nextn.eh_proj.weight")) {
4404 if offset == 0 {
4405 break;
4406 }
4407 return Err(format!(
4408 "embedded MTP chain declares {} heads but blk.{n} has no \
4409 nextn.eh_proj.weight",
4410 cfg.nextn_predict_layers
4411 )
4412 .into());
4413 }
4414 embedded_mtp.push(MtpHead {
4415 enorm: load_t(e, src, &p("nextn.enorm.weight"))?,
4416 hnorm: load_t(e, src, &p("nextn.hnorm.weight"))?,
4417 eh_proj: load_t(e, src, &p("nextn.eh_proj.weight"))?,
4418 attn_norm: load_t(e, src, &p("attn_norm.weight"))?,
4419 post_attn_norm: load_opt(e, src, &p("post_attention_norm.weight"))?
4420 .or(load_opt(e, src, &p("ffn_norm.weight"))?)
4421 .expect("MTP block needs post_attention_norm or ffn_norm"),
4422 mixer: load_mixer_kind(
4423 e,
4424 src,
4425 &cfg,
4426 n,
4427 &mtp_plan.layer.attention,
4428 &mut step_runtimes,
4429 )?,
4430 ffn: load_ffn(
4431 e,
4432 src,
4433 &cfg,
4434 &mtp_plan.layer.mlp,
4435 n,
4436 spill.as_mut().map(|c| (gguf.unwrap(), c)),
4437 &mut resident,
4438 &mut step_runtimes,
4439 )?,
4440 shared_head_norm: load_opt(e, src, &p("nextn.shared_head_norm.weight"))?,
4441 shared_head_head: load_mtp_head_maybe_nvfp4(
4450 e,
4451 src,
4452 &p("nextn.shared_head_head.weight"),
4453 )?
4454 .or(load_opt(e, src, &p("nextn.shared_head.weight"))?),
4455 d2t: None,
4456 d2t_from_target_head: false,
4457 geom: None,
4458 step35: if sliding_gated_moe_program {
4459 Some(Step35MtpGeom::from_plan(&mtp_plan.layer)?)
4460 } else {
4461 None
4462 },
4463 });
4464 }
4465 }
4466 let mut embedded_mtp = embedded_mtp.into_iter();
4467 let mut mtp = embedded_mtp.next();
4468 let mut mtp_extra: Vec<MtpHead> = embedded_mtp.collect();
4469
4470 mtp = if load_mtp {
4474 match std::env::var("MEMRA_MTP_DRAFT") {
4475 Ok(path) if !path.is_empty() => {
4476 eprintln!("[mtp-draft] loading external MTP draft: {path}");
4477 let dg = GgufFile::open(&path)?;
4478 mtp_extra.clear();
4479 Some(MtpHead::load_draft(e, &dg, &cfg)?)
4480 }
4481 _ => mtp,
4482 }
4483 } else {
4484 None
4485 };
4486
4487 let trim_env = if load_mtp {
4498 std::env::var("MEMRA_FRSPEC_TRIM")
4499 } else {
4500 Err(std::env::VarError::NotPresent)
4501 };
4502 if crate::model::full_prec_enabled()
4503 && trim_env.as_deref().map(|p| !p.is_empty()).unwrap_or(false)
4504 {
4505 eprintln!(
4506 "[frspec-trim] DISABLED under MEMRA_FULL_PREC — using the natural full MTP head"
4507 );
4508 }
4509 mtp = match (
4510 if crate::model::full_prec_enabled() {
4511 Err(std::env::VarError::NotPresent)
4512 } else {
4513 trim_env
4514 },
4515 mtp,
4516 ) {
4517 (Ok(path), Some(mut head)) if !path.is_empty() => {
4518 let e = crate::pp::layer_engine(e, n_trunk, n_trunk)?;
4522 let path = memra_gguf::hf::resolve_arg(&path)
4526 .map_err(|err| format!("MEMRA_FRSPEC_TRIM={path:?}: {err}"))?;
4527 let d2t: Vec<u32> = frspec_read_d2t(&path)?;
4531 let own_head_name = frspec_trim_own_head_name(n_trunk);
4540 let own_head = src.find(&own_head_name);
4541 let from_own_head = own_head.is_some();
4542 let v = own_head
4543 .or_else(|| src.find("output.weight"))
4544 .or_else(|| src.find("token_embd.weight"))
4545 .expect("model has no output.weight for FR-Spec trim");
4546 let (trimmed, nvfp4_sizes) = frspec_gather_trimmed_head(
4566 e,
4567 &v,
4568 &d2t,
4569 std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1"),
4570 match src.find("output.scale") {
4572 Some(sv) => f32::from_le_bytes(sv.bytes[..4].try_into().unwrap()),
4573 None => 1.0,
4574 },
4575 )?;
4576 match nvfp4_sizes {
4577 Some((nvfp4_bytes, gathered_bytes)) => eprintln!(
4578 "[frspec-trim] self-trimmed head: {} rows of {} re-quantized BF16 -> NVFP4 \
4579 ({} MiB, was {} MiB)",
4580 d2t.len(),
4581 if from_own_head {
4582 own_head_name.as_str()
4583 } else {
4584 "main output.weight"
4585 },
4586 nvfp4_bytes >> 20,
4587 gathered_bytes >> 20,
4588 ),
4589 None => eprintln!(
4590 "[frspec-trim] self-trimmed head: {} rows of {} ({:?})",
4591 d2t.len(),
4592 if from_own_head {
4593 own_head_name.as_str()
4594 } else {
4595 "main output.weight"
4596 },
4597 v.ggml_type
4598 ),
4599 }
4600 head.shared_head_head = Some(trimmed);
4601 head.d2t = Some(d2t);
4602 head.d2t_from_target_head = !from_own_head;
4605 Some(head)
4606 }
4607 (_, m) => m,
4608 };
4609 let dflash_trim: Option<DflashTrimHead> = match mtp_skip_trim_d2t {
4620 Some(d2t) => {
4621 let v = src
4622 .find("output.weight")
4623 .or_else(|| src.find("token_embd.weight"))
4624 .ok_or("model has no output.weight for FR-Spec trim")?;
4625 let (head, nvfp4_sizes) = frspec_gather_trimmed_head(
4626 e,
4627 &v,
4628 &d2t,
4629 std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1"),
4630 match src.find("output.scale") {
4631 Some(sv) => f32::from_le_bytes(sv.bytes[..4].try_into().unwrap()),
4632 None => 1.0,
4633 },
4634 )?;
4635 eprintln!(
4636 "[mtp-skip] FR-Spec stub draft head built: {} rows of main output.weight \
4637 ({}); DFlash2 trim serves without the embedded MTP block",
4638 d2t.len(),
4639 match nvfp4_sizes {
4640 Some((nvfp4_bytes, gathered_bytes)) => format!(
4641 "re-quantized BF16 -> NVFP4, {} MiB, was {} MiB",
4642 nvfp4_bytes >> 20,
4643 gathered_bytes >> 20
4644 ),
4645 None => format!("{:?}", v.ggml_type),
4646 },
4647 );
4648 Some(DflashTrimHead { head, d2t })
4649 }
4650 None => None,
4651 };
4652 if let Some(d2t) = mtp.as_ref().and_then(|head| head.d2t.clone()) {
4665 let want_nvfp4_env = std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1");
4666 let mut kept = 0usize;
4667 let e = crate::pp::layer_engine(e, n_trunk, n_trunk)?;
4670 for (i, head) in mtp_extra.iter_mut().enumerate() {
4671 let name = frspec_trim_own_head_name(n_trunk + 1 + i);
4672 let Some(v) = src.find(&name) else { break };
4673 let out_f = v.ne[1] as usize;
4674 let row_bytes = v.bytes.len() / out_f;
4675 if d2t.iter().any(|&t| (t as usize) >= out_f) {
4676 break;
4677 }
4678 let mut gathered = Vec::with_capacity(d2t.len() * row_bytes);
4679 for &t in &d2t {
4680 let off = t as usize * row_bytes;
4681 gathered.extend_from_slice(&v.bytes[off..off + row_bytes]);
4682 }
4683 let want_nvfp4 =
4684 want_nvfp4_env && matches!(v.ggml_type, GgmlType::BF16) && v.ne[0] % 64 == 0;
4685 let trimmed = if want_nvfp4 {
4686 let vals: Vec<f32> = gathered
4687 .chunks_exact(2)
4688 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
4689 .collect();
4690 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
4691 GpuTensor::from_quant_bytes(
4692 e,
4693 &blocks,
4694 GgmlType::NVFP4,
4695 v.ne[0],
4696 d2t.len() as u64,
4697 1.0,
4698 )?
4699 } else {
4700 match v.ggml_type {
4701 GgmlType::BF16 => GpuTensor::FloatBf16 {
4702 data: e.htod_bytes(&gathered)?,
4703 ne: vec![v.ne[0], d2t.len() as u64],
4704 },
4705 GgmlType::F32 => GpuTensor::Float {
4706 data: e.htod(
4707 &gathered
4708 .chunks_exact(4)
4709 .map(|c| f32::from_le_bytes(c.try_into().unwrap()))
4710 .collect::<Vec<f32>>(),
4711 )?,
4712 ne: vec![v.ne[0], d2t.len() as u64],
4713 },
4714 _ => GpuTensor::from_quant_bytes(
4715 e,
4716 &gathered,
4717 v.ggml_type,
4718 v.ne[0],
4719 d2t.len() as u64,
4720 1.0,
4721 )?,
4722 }
4723 };
4724 head.shared_head_head = Some(trimmed);
4725 head.d2t = Some(d2t.clone());
4726 head.d2t_from_target_head = false;
4727 kept += 1;
4728 }
4729 let dropped = mtp_extra.len() - kept;
4730 mtp_extra.truncate(kept);
4731 eprintln!(
4732 "[frspec-trim] per-head trim: {kept} extra chain head(s) gathered from their own \
4733 blocks{}",
4734 if dropped > 0 {
4735 format!(" ({dropped} dropped: no own-head tensor)")
4736 } else {
4737 String::new()
4738 }
4739 );
4740 }
4741 if !mtp_extra.is_empty() {
4742 if plan.draft_source != memra_gguf::model_plan::DraftSourcePlan::Embedded
4743 || plan.mtp_blocks.len() != 1 + mtp_extra.len()
4744 || plan
4745 .mtp_blocks
4746 .iter()
4747 .any(|block| !matches!(block.layer.mlp, MlpPlan::Dense(_)))
4748 || mtp
4749 .iter()
4750 .chain(mtp_extra.iter())
4751 .any(|head| !matches!(head.ffn, Ffn::Dense { .. }))
4752 {
4753 return Err(
4754 "multi-head MTP requires embedded dense canonical blocks and matching loaded heads"
4755 .into(),
4756 );
4757 }
4758 eprintln!(
4759 "[mtp-draft] embedded chain: heads={} blocks={}..={} scratch=per-head",
4760 1 + mtp_extra.len(),
4761 n_trunk,
4762 n_trunk + mtp_extra.len()
4763 );
4764 }
4765
4766 let glm5_dflash = match std::env::var("MEMRA_GLM5_DFLASH") {
4774 Ok(spec) if !spec.is_empty() && cfg.arch.is_glm5_next() => {
4775 let dpath = memra_gguf::hf::resolve_arg(&spec)
4776 .map_err(|err| format!("MEMRA_GLM5_DFLASH={spec:?}: {err}"))?;
4777 let de = crate::pp::layer_engine(e, n_trunk, n_trunk)?;
4778 let dir = std::path::Path::new(&dpath);
4779 let draft = crate::dflash::DflashDraft::load(de, dir).map_err(|err| {
4780 format!("MEMRA_GLM5_DFLASH={dpath}: drafter load failed: {err}")
4781 })?;
4782 if draft.dflash2.is_none() {
4783 return Err(format!(
4784 "MEMRA_GLM5_DFLASH={dpath}: checkpoint is not a DFlash2DraftModel \
4785 (the glm5 draft source is the selector family only)"
4786 )
4787 .into());
4788 }
4789 if draft.cfg.hidden != cfg.n_embd as usize {
4790 return Err(format!(
4791 "MEMRA_GLM5_DFLASH={dpath}: drafter hidden {} != target n_embd {} \
4792 (the drafter consumes target features and the target's embed/lm_head)",
4793 draft.cfg.hidden, cfg.n_embd
4794 )
4795 .into());
4796 }
4797 if draft.cfg.target_layer_ids.is_empty()
4798 || draft.cfg.target_layer_ids.iter().any(|&t| t >= n_trunk)
4799 {
4800 return Err(format!(
4801 "MEMRA_GLM5_DFLASH={dpath}: target_layer_ids {:?} do not name valid \
4802 trunk layers (n_trunk {n_trunk})",
4803 draft.cfg.target_layer_ids
4804 )
4805 .into());
4806 }
4807 if draft.cfg.mask_token_id as usize >= output.out_features() {
4808 return Err(format!(
4809 "MEMRA_GLM5_DFLASH={dpath}: mask token {} outside the target vocab {}",
4810 draft.cfg.mask_token_id,
4811 output.out_features()
4812 )
4813 .into());
4814 }
4815 let sha8 = sha256_file_hex8(&dir.join("model.safetensors"))
4816 .map_err(|err| format!("MEMRA_GLM5_DFLASH={dpath}: sha256 pin: {err}"))?;
4817 Some(crate::glm_spec::Glm5DflashDrafter { draft, sha8 })
4818 }
4819 _ => None,
4820 };
4821
4822 if cfg.arch.is_glm5_next() && crate::glm_spec::glm5_spec_on() {
4830 match (glm5_dflash.as_ref(), mtp.as_ref()) {
4831 (Some(dr), head) => {
4832 let trim_note = match head.and_then(|h| h.d2t.as_ref()) {
4833 Some(map) => {
4834 format!("draft head TRIMMED to {} rows (FR-Spec d2t)", map.len())
4835 }
4836 None => "draft head FULL target vocab".to_string(),
4837 };
4838 eprintln!(
4839 "[glm5-spec] serve route ARMED: draft source = dflash2 @ {}; {trim_note}; \
4840 native MTP head {}",
4841 dr.sha8,
4842 if head.is_some() {
4843 "ALSO loaded (idle for drafting — dflash2 wins by selection)"
4844 } else {
4845 "NOT loaded (the q38 pattern: a full MoE trunk layer of VRAM saved)"
4846 }
4847 );
4848 }
4849 (None, Some(head)) => {
4850 match head.d2t.as_ref() {
4851 Some(map) => eprintln!(
4852 "[glm5-spec] serve route ARMED: MTP head loaded; draft head TRIMMED \
4853 to {} rows (FR-Spec d2t engaged)",
4854 map.len()
4855 ),
4856 None => eprintln!(
4857 "[glm5-spec] serve route ARMED: MTP head loaded; draft head FULL \
4858 target vocab (no FR-Spec trim)"
4859 ),
4860 }
4861 eprintln!("[glm5-spec] draft source = native-mtp");
4862 }
4863 (None, None) => eprintln!(
4864 "[glm5-spec] MEMRA_GLM5_SPEC=1 but no MTP head loaded \
4865 (set MEMRA_GLM5_MTP=1 or MEMRA_GLM5_DFLASH=<drafter>) — route stays \
4866 fail-closed, plain serving"
4867 ),
4868 }
4869 }
4870
4871 if let Some(ctx) = spill.as_ref() {
4872 eprintln!(
4873 "[spill] experts placed: {} pinned (Tier 1), {} mmap'd from disk (Tier 2, {} MiB)",
4874 ctx.n_pinned,
4875 ctx.n_mmap,
4876 ctx.mmap_bytes >> 20
4877 );
4878 }
4879
4880 if cfg.n_head_kv > 0 && cfg.n_head / cfg.n_head_kv > 8 {
4894 crate::FA_V4_MAX_DEFAULT.store(0, std::sync::atomic::Ordering::Relaxed);
4895 eprintln!(
4896 "[fa] v4 decode family disabled: gqa {} > fa_v4_smem capacity 8 (v3 lane serves)",
4897 cfg.n_head / cfg.n_head_kv
4898 );
4899 }
4900
4901 if gemma_program {
4902 crate::FA_VEC_MIN_DEFAULT.store(1, std::sync::atomic::Ordering::Relaxed);
4904 let real_moe = plan
4907 .trunk_operations()
4908 .contains(&memra_gguf::model_plan::OperationKind::MoeMlp);
4909 crate::FA_SPW_DEFAULT.store(
4910 if real_moe { 32 } else { 64 },
4911 std::sync::atomic::Ordering::Relaxed,
4912 );
4913 crate::FA_SP512_DEFAULT.store(
4915 if real_moe { 16 } else { 32 },
4916 std::sync::atomic::Ordering::Relaxed,
4917 );
4918 crate::FUSED_MR1_DEFAULT.store(!real_moe, std::sync::atomic::Ordering::Relaxed);
4928 crate::RMS_BLOCK_DEFAULT.store(1024, std::sync::atomic::Ordering::Relaxed);
4930 crate::FA_SP_GEMMA.store(true, std::sync::atomic::Ordering::Relaxed);
4932 }
4936 let force_embd_gpu = gemma_program;
4939 let gemma4_aux = if gemma_program {
4940 let rope_freqs = match src.find("rope_freqs.weight") {
4941 Some(t) => {
4942 let host = memra_gguf::dequant::dequantize(
4943 t.ggml_type,
4944 &t.bytes,
4945 t.ne.iter().product::<u64>() as usize,
4946 );
4947 let mut copies = Vec::new();
4948 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
4949 #[allow(clippy::needless_range_loop)]
4950 for s in 0..fence.len() - 1 {
4952 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
4953 let dev = owner.ctx().ordinal();
4954 if copies.iter().all(|(d, _)| *d != dev) {
4955 copies.push((dev, owner.htod(&host)?));
4956 }
4957 }
4958 } else {
4959 copies.push((e.ctx().ordinal(), e.htod(&host)?));
4960 }
4961 Some(copies)
4962 }
4963 None => {
4971 let g4 = cfg.gemma4.as_ref().unwrap();
4972 let n = (g4.rope_dims_global / 2) as usize;
4973 let keep =
4974 ((n as f32) * g4.partial_rotary_global.clamp(0.0, 1.0)).round() as usize;
4975 let host: Vec<f32> = (0..n)
4976 .map(|i| if i < keep { 1.0 } else { 1.0e30 })
4977 .collect();
4978 eprintln!(
4979 "[gemma4] rope_freqs.weight synthesized ({n} factors, first {keep} \
4980 rotate; source ships none — native checkpoint)"
4981 );
4982 let mut copies = Vec::new();
4983 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
4984 #[allow(clippy::needless_range_loop)]
4985 for s in 0..fence.len() - 1 {
4987 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
4988 let dev = owner.ctx().ordinal();
4989 if copies.iter().all(|(d, _)| *d != dev) {
4990 copies.push((dev, owner.htod(&host)?));
4991 }
4992 }
4993 } else {
4994 copies.push((e.ctx().ordinal(), e.htod(&host)?));
4995 }
4996 Some(copies)
4997 }
4998 };
4999 let e4b = match src.find("per_layer_token_embd.weight") {
5001 Some(t) => {
5002 let n_epl = cfg
5003 .gemma4
5004 .as_ref()
5005 .map(|g| g.n_embd_per_layer as usize)
5006 .unwrap_or(0);
5007 let row = t.ne[0] as usize; let row_bytes = t.bytes.len() / (t.ne[1] as usize);
5009 eprintln!(
5010 "[gemma4-e4b] per-layer-embed model detected (n_epl={n_epl}, row {row}) — \
5011 first-light forward (eager decode + prime); dc/graph/spec unwired \
5012 (HANDOVER-E4B.md)"
5013 );
5014 Some(crate::hybrid::Gemma4E4bModel {
5015 tok_tbl_gpu: std::sync::OnceLock::new(),
5016 tok_embd_bytes: t.bytes.to_vec(),
5017 tok_embd_qt: match t.ggml_type {
5018 memra_gguf::GgmlType::Q6_K => crate::QT_Q6_K,
5019 memra_gguf::GgmlType::Q8_0 => crate::QT_Q8_0,
5020 other => panic!("e4b per-layer tok embd: unhandled dtype {other:?}"),
5021 },
5022 tok_embd_row_bytes: row_bytes,
5023 model_proj: load_t(e, src, "per_layer_model_proj.weight")?,
5024 proj_norm: load_t(e, src, "per_layer_proj_norm.weight")?,
5025 n_epl,
5026 })
5027 }
5028 None => None,
5029 };
5030 let suppress_d = {
5031 let sup = &cfg.gemma4.as_ref().unwrap().suppress_tokens;
5032 if sup.is_empty() {
5033 None
5034 } else {
5035 let ids: Vec<i32> = sup.iter().map(|&x| x as i32).collect();
5036 eprintln!(
5037 "[gemma4] suppress_tokens: {} ids masked at sampling",
5038 ids.len()
5039 );
5040 Some((e.htod_i32(&ids)?, ids.len()))
5041 }
5042 };
5043 let ones_host = [1.0f32; 512];
5044 let mut ones = Vec::new();
5045 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
5046 #[allow(clippy::needless_range_loop)]
5047 for s in 0..fence.len() - 1 {
5049 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
5050 let dev = owner.ctx().ordinal();
5051 if ones.iter().all(|(d, _)| *d != dev) {
5052 ones.push((dev, owner.htod(&ones_host)?));
5053 }
5054 }
5055 } else {
5056 ones.push((e.ctx().ordinal(), e.htod(&ones_host)?));
5057 }
5058 Some(GemmaAux {
5059 rope_freqs,
5060 ones,
5061 suppress_d,
5062 e4b,
5063 })
5064 } else {
5065 None
5066 };
5067 let step35_aux = if sliding_gated_moe_program {
5071 let rope_freqs = match src.find("rope_freqs.weight") {
5072 Some(t) => {
5073 let host = memra_gguf::dequant::dequantize(
5074 t.ggml_type,
5075 &t.bytes,
5076 t.ne.iter().product::<u64>() as usize,
5077 );
5078 let mut copies = Vec::new();
5079 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
5080 #[allow(clippy::needless_range_loop)]
5081 for s in 0..fence.len() - 1 {
5083 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
5084 let dev = owner.ctx().ordinal();
5085 if copies.iter().all(|(d, _)| *d != dev) {
5086 copies.push((dev, owner.htod(&host)?));
5087 }
5088 }
5089 } else {
5090 copies.push((e.ctx().ordinal(), e.htod(&host)?));
5091 }
5092 Some(copies)
5093 }
5094 None => None,
5095 };
5096 Some(Step35Aux { rope_freqs })
5097 } else {
5098 None
5099 };
5100 let mut layers = layers;
5101 {
5108 let q8rp_on = match std::env::var("MEMRA_Q8RP").as_deref() {
5109 Ok("0") => false,
5110 Ok(_) => true,
5111 Err(_) => {
5119 cfg!(memra_hopper_mma) || {
5120 let q8b = |w: &crate::model::GpuTensor| -> usize {
5121 match w {
5122 crate::model::GpuTensor::Quant {
5123 bytes,
5124 qtype,
5125 row_bytes,
5126 ne,
5127 rp4: None,
5128 ..
5129 } if *qtype == crate::QT_Q8_0
5130 && ne.len() == 2
5131 && (ne[0] as usize).is_multiple_of(32)
5132 && *row_bytes == (ne[0] as usize / 32) * 34 =>
5133 {
5134 bytes.len()
5135 }
5136 _ => 0,
5137 }
5138 };
5139 let mut need = q8b(&output);
5140 for layer in layers.iter() {
5141 match &layer.mixer {
5142 Mixer::Full(fa) => {
5143 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
5144 need += q8b(w);
5145 }
5146 }
5147 Mixer::Linear(la) => {
5148 for w in [
5149 &la.wqkv,
5150 &la.wqkv_gate,
5151 &la.ssm_beta,
5152 &la.ssm_alpha,
5153 &la.ssm_out,
5154 ] {
5155 need += q8b(w);
5156 }
5157 }
5158 Mixer::Mla(_) => {}
5159 Mixer::Kda(_) => {} }
5161 if let Ffn::Dense {
5162 ffn_gate,
5163 ffn_up,
5164 ffn_down,
5165 } = &layer.ffn
5166 {
5167 for w in [ffn_gate, ffn_up, ffn_down] {
5168 need += q8b(w);
5169 }
5170 }
5171 }
5172 need > 0
5173 && e.ctx()
5174 .mem_get_info()
5175 .map(|(free, _)| free >= need + (8usize << 30))
5176 .unwrap_or(false)
5177 }
5178 }
5179 };
5180 let kqrp_on = crate::Engine::kqrp_enabled() || {
5190 std::env::var("MEMRA_KQRP").is_err() && {
5191 let kqb = |w: &crate::model::GpuTensor| -> usize {
5192 match w {
5193 crate::model::GpuTensor::Quant {
5194 bytes,
5195 qtype,
5196 row_bytes,
5197 ne,
5198 rp4: None,
5199 ..
5200 } if ne.len() == 2 && (ne[0] as usize).is_multiple_of(256) => {
5201 let sb = if *qtype == crate::QT_Q4_K {
5202 144
5203 } else if *qtype == crate::QT_Q6_K {
5204 210
5205 } else {
5206 return 0;
5207 };
5208 if *row_bytes == (ne[0] as usize / 256) * sb {
5209 bytes.len()
5210 } else {
5211 0
5212 }
5213 }
5214 _ => 0,
5215 }
5216 };
5217 let mut need = kqb(&output);
5218 for layer in layers.iter() {
5219 if let Mixer::Full(fa) = &layer.mixer {
5220 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
5221 need += kqb(w);
5222 }
5223 }
5224 if let Ffn::Dense {
5225 ffn_gate,
5226 ffn_up,
5227 ffn_down,
5228 } = &layer.ffn
5229 {
5230 for w in [ffn_gate, ffn_up, ffn_down] {
5231 need += kqb(w);
5232 }
5233 }
5234 }
5235 need > 0
5236 && e.ctx()
5237 .mem_get_info()
5238 .map(|(free, _)| free >= need + (8usize << 30))
5239 .unwrap_or(false)
5240 }
5241 };
5242 if q8rp_on || kqrp_on {
5243 let f16_model_ok = gemma_program
5250 || plan
5251 .trunk_operations()
5252 .contains(&memra_gguf::model_plan::OperationKind::MoeMlp)
5253 || std::env::var("MEMRA_PP_F16").as_deref() == Ok("1");
5254 let mut nmir = 0usize;
5255 let mut mir = |e_ref: &crate::Engine,
5259 w: &mut crate::model::GpuTensor|
5260 -> Result<(), Box<dyn std::error::Error>> {
5261 let before = matches!(w, crate::model::GpuTensor::Quant { rp4: Some(_), .. });
5262 if q8rp_on {
5263 e_ref.build_q8_rp4(w)?;
5264 }
5265 if kqrp_on {
5266 e_ref.build_q4k_rp4(w)?;
5267 e_ref.build_q6k_rp4(w)?;
5268 }
5269 let q6k = matches!(w, crate::model::GpuTensor::Quant { qtype, .. }
5274 if *qtype == crate::QT_Q6_K);
5275 if q8rp_on && crate::f16_ffi::pp_f16_enabled() && (f16_model_ok || q6k) {
5276 e_ref.build_q8_f16(w)?;
5277 }
5278 if !before && matches!(w, crate::model::GpuTensor::Quant { rp4: Some(_), .. }) {
5279 nmir += 1;
5280 }
5281 Ok(())
5282 };
5283 for (il, layer) in layers.iter_mut().enumerate() {
5284 let el = crate::pp::layer_engine(e, n_trunk, il)?;
5285 match &mut layer.mixer {
5286 Mixer::Full(fa) => {
5287 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
5288 mir(el, w)?;
5289 }
5290 }
5291 Mixer::Linear(la) => {
5292 for w in [
5293 &mut la.wqkv,
5294 &mut la.wqkv_gate,
5295 &mut la.ssm_beta,
5296 &mut la.ssm_alpha,
5297 &mut la.ssm_out,
5298 ] {
5299 mir(el, w)?;
5300 }
5301 }
5302 Mixer::Mla(_) => {}
5305 Mixer::Kda(_) => {} }
5307 if let Ffn::Dense {
5308 ffn_gate,
5309 ffn_up,
5310 ffn_down,
5311 } = &mut layer.ffn
5312 {
5313 for w in [ffn_gate, ffn_up, ffn_down] {
5314 mir(el, w)?;
5315 }
5316 }
5317 }
5318 mir(e_head, &mut output)?;
5319 if nmir > 0 {
5320 eprintln!("[q8rp] split-plane decode mirrors built: {nmir} tensors");
5321 }
5322 if q8rp_on && crate::f16_ffi::pp_f16_enabled() {
5337 for (want, tag) in [(crate::QT_Q4_K, "q4kf16"), (crate::QT_Q5_K, "q5kf16")] {
5338 let (mut n4, mut b4) = (0usize, 0usize);
5339 let mut mirk =
5340 |e_ref: &crate::Engine,
5341 w: &mut crate::model::GpuTensor|
5342 -> Result<(), Box<dyn std::error::Error>> {
5343 if matches!(w, crate::model::GpuTensor::Quant { qtype, f16: None, .. }
5344 if *qtype == want)
5345 {
5346 e_ref.build_q8_f16(w)?;
5347 if let crate::model::GpuTensor::Quant { f16: Some(m), .. } = w {
5348 n4 += 1;
5349 b4 += m.len();
5350 }
5351 }
5352 Ok(())
5353 };
5354 for (il, layer) in layers.iter_mut().enumerate() {
5355 let el = crate::pp::layer_engine(e, n_trunk, il)?;
5356 match &mut layer.mixer {
5357 Mixer::Full(fa) => {
5358 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
5359 mirk(el, w)?;
5360 }
5361 }
5362 Mixer::Linear(la) => {
5363 for w in [
5364 &mut la.wqkv,
5365 &mut la.wqkv_gate,
5366 &mut la.ssm_beta,
5367 &mut la.ssm_alpha,
5368 &mut la.ssm_out,
5369 ] {
5370 mirk(el, w)?;
5371 }
5372 }
5373 Mixer::Mla(_) => {} Mixer::Kda(_) => {} }
5376 if let Ffn::Dense {
5377 ffn_gate,
5378 ffn_up,
5379 ffn_down,
5380 } = &mut layer.ffn
5381 {
5382 for w in [ffn_gate, ffn_up, ffn_down] {
5383 mirk(el, w)?;
5384 }
5385 }
5386 }
5387 mirk(e_head, &mut output)?;
5388 if n4 > 0 {
5389 eprintln!(
5390 "[{tag}] prefill fp16 mirrors built: {n4} tensors \
5391 ({} MB)",
5392 b4 >> 20
5393 );
5394 }
5395 }
5396 }
5397 }
5398 }
5399 if gemma_program && crate::Engine::q4rp_enabled() {
5406 let mut nmir = 0usize;
5407 for (il, layer) in layers.iter_mut().enumerate() {
5408 let e = crate::pp::layer_engine(e, n_trunk, il)?;
5410 let is_moe26 = layer.gemma4.as_ref().is_some_and(|g| g.moe_bits.is_some());
5419 let is_e4b = layer.gemma4.as_ref().is_some_and(|g| g.e4b.is_some());
5420 if !(is_moe26 || is_e4b) {
5421 continue;
5422 }
5423 if let Mixer::Full(fa) = &mut layer.mixer {
5424 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
5425 e.build_q4_rp4(w)?;
5426 nmir += 1;
5427 }
5428 }
5429 if is_e4b {
5430 let own_kv = layer
5432 .gemma4
5433 .as_ref()
5434 .unwrap()
5435 .e4b
5436 .as_ref()
5437 .is_some_and(|e4| e4.kv_share.is_none());
5438 if own_kv
5439 && let Mixer::Full(fa) = &layer.mixer
5440 && let Some(mut cat) = e.build_q4_out_concat3(&fa.wq, &fa.wk, &fa.wv)?
5441 {
5442 e.build_q4_rp4(&mut cat)?;
5443 nmir += 1;
5444 layer.gemma4.as_mut().unwrap().e4b.as_mut().unwrap().qkv_cat = Some(cat);
5445 }
5446 if let Ffn::Dense {
5447 ffn_gate,
5448 ffn_up,
5449 ffn_down,
5450 } = &mut layer.ffn
5451 {
5452 for w in [ffn_gate, ffn_up, ffn_down] {
5453 e.build_q4_rp4(w)?;
5454 nmir += 1;
5455 }
5456 }
5457 let e4 = layer.gemma4.as_mut().unwrap().e4b.as_mut().unwrap();
5458 for w in [&mut e4.inp_gate, &mut e4.proj] {
5459 e.build_q4_rp4(w)?;
5460 nmir += 1;
5461 }
5462 }
5463 if let Some(mb) = layer.gemma4.as_mut().unwrap().moe_bits.as_mut() {
5464 for w in [&mut mb.shared_gate, &mut mb.shared_up, &mut mb.shared_down] {
5465 e.build_q4_rp4(w)?;
5466 nmir += 1;
5467 }
5468 }
5469 }
5470 if nmir > 0 {
5471 eprintln!("[q4rp] split-plane decode mirrors built: {nmir} trunk tensors");
5472 }
5473 let fast_on = std::env::var("MEMRA_FAST").as_deref() != Ok("0");
5480 if fast_on {
5481 let mut nswap = 0usize;
5482 let mut nf16 = 0usize;
5483 let q4f16_model_ok = matches!(cfg.n_embd, 3840 | 5376); if let Ok(v) = std::env::var("MEMRA_Q4F16")
5502 && v != "0"
5503 && v != "1"
5504 {
5505 return Err(format!(
5506 "MEMRA_Q4F16={v} is not 0 or 1 — this env selects the prefill \
5507 ARITHMETIC (fp16 mirrors vs int8 MMQ) and must never be guessed"
5508 )
5509 .into());
5510 }
5511 let f16_need = {
5512 let f16b = |w: &crate::model::GpuTensor| -> usize {
5513 match w {
5514 crate::model::GpuTensor::Quant {
5515 qtype,
5516 ne,
5517 f16: None,
5518 ..
5519 } if ne.len() == 2
5520 && matches!(
5521 *qtype,
5522 crate::QT_Q8_0
5523 | crate::QT_Q4_0
5524 | crate::QT_Q6_K
5525 | crate::QT_Q4_K
5526 | crate::QT_Q5_K
5527 ) =>
5528 {
5529 (ne[0] as usize) * (ne[1] as usize) * 2
5530 }
5531 _ => 0,
5532 }
5533 };
5534 let mut need = 0usize;
5535 for layer in layers.iter() {
5536 if layer.gemma4.as_ref().is_none_or(|g| g.moe_bits.is_some()) {
5537 continue;
5538 }
5539 if let Mixer::Full(fa) = &layer.mixer {
5540 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
5541 need += f16b(w);
5542 }
5543 }
5544 if let Ffn::Dense {
5545 ffn_gate,
5546 ffn_up,
5547 ffn_down,
5548 } = &layer.ffn
5549 {
5550 for w in [ffn_gate, ffn_up, ffn_down] {
5551 need += f16b(w);
5552 }
5553 }
5554 }
5555 need
5556 };
5557 let f16_free = e.ctx().mem_get_info().map(|(free, _)| free).unwrap_or(0);
5558 let f16_auto = q4f16_model_ok
5559 && std::env::var("MEMRA_Q4F16").is_err()
5560 && crate::f16_ffi::pp_f16_capacity_ok(f16_free, f16_need);
5561 let (f16_on, f16_why) = match std::env::var("MEMRA_Q4F16").as_deref() {
5567 Ok("1") => (true, "env MEMRA_Q4F16=1"),
5568 Ok("0") => (false, "env MEMRA_Q4F16=0"),
5569 _ if crate::f16_ffi::pp_f16_enabled() && q4f16_model_ok => {
5570 (true, "env MEMRA_PP_F16")
5571 }
5572 _ if f16_auto => (true, "capacity-keyed auto (UNPINNED)"),
5573 _ if !q4f16_model_ok => (false, "model geometry not eligible"),
5574 _ => (false, "capacity-keyed auto REFUSED (UNPINNED)"),
5575 };
5576 eprintln!(
5583 "[q4f16] prefill program = {} (reason: {}); free {} MiB, mirror mass {} MiB, \
5584 capacity threshold {} MiB (mass + 8192 headroom) — SELECTS PREFILL ARITHMETIC",
5585 if f16_on {
5586 "FP16 MIRRORS"
5587 } else {
5588 "INT8 MMQ (no f16 mirrors)"
5589 },
5590 f16_why,
5591 f16_free >> 20,
5592 f16_need >> 20,
5593 (f16_need + (8usize << 30)) >> 20,
5594 );
5595 for (il, layer) in layers.iter_mut().enumerate() {
5596 let e = crate::pp::layer_engine(e, n_trunk, il)?;
5598 let dense_gemma = layer.gemma4.as_ref().is_some_and(|g| g.moe_bits.is_none());
5599 if !dense_gemma {
5600 continue;
5601 }
5602 if let Mixer::Full(fa) = &mut layer.mixer {
5603 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
5604 if f16_on {
5605 e.build_q8_f16(w)?;
5606 if matches!(w, crate::model::GpuTensor::Quant { f16: Some(_), .. })
5607 {
5608 nf16 += 1;
5609 }
5610 }
5611 if e.build_q4_rp_swap(w)? {
5612 nswap += 1;
5613 }
5614 }
5615 }
5616 if let Ffn::Dense {
5617 ffn_gate,
5618 ffn_up,
5619 ffn_down,
5620 } = &mut layer.ffn
5621 {
5622 for w in [ffn_gate, ffn_up, ffn_down] {
5623 if f16_on {
5624 e.build_q8_f16(w)?;
5625 if matches!(w, crate::model::GpuTensor::Quant { f16: Some(_), .. })
5626 {
5627 nf16 += 1;
5628 }
5629 }
5630 if e.build_q4_rp_swap(w)? {
5631 nswap += 1;
5632 }
5633 }
5634 }
5635 }
5636 if nswap > 0 {
5637 eprintln!("[q4rp] split-plane IN-PLACE swap: {nswap} dense trunk tensors");
5638 }
5639 if nf16 > 0 {
5640 eprintln!("[q4f16] prefill fp16 mirrors built: {nf16} dense trunk tensors");
5641 }
5642 }
5643 }
5644 let model = HybridModel {
5645 cfg,
5646 plan,
5647 rewrite_qualifications: None,
5648 embd,
5649 output_norm,
5650 output,
5651 layers,
5652 mtp,
5653 mtp_extra,
5654 dflash_trim,
5655 embd_gpu: std::sync::OnceLock::new(),
5656 gemma4_aux,
5657 step35_aux,
5658 prime_slabs: std::sync::Mutex::new(std::collections::HashMap::new()),
5659 dspark_vgraphs: std::sync::Mutex::new(None),
5660 step_grouped_prefill: std::sync::Mutex::new(StepEpGroupedPrefill::default()),
5661 step35_token_graph: std::sync::Mutex::new(None),
5662 hyper,
5663 hyper_head,
5664 glm5_dflash,
5665 draft_state_bytes: std::sync::atomic::AtomicUsize::new(0),
5666 };
5667 e.configure_moe_cache_layout(model.moe_cache_block_sizes());
5668 if force_embd_gpu {
5669 let _ = model
5670 .embd_gpu
5671 .get_or_init(|| e.upload_u8(&model.embd.raw).expect("embed table upload"));
5672 }
5673 crate::pp::sync_stages_after_load(e, n_trunk)?;
5679 Ok(model)
5680 }
5681
5682 pub fn ensure_embed_resident(&self, e: &Engine) -> Result<(), Box<dyn std::error::Error>> {
5692 if std::env::var("MEMRA_EMBED_DEV").as_deref() == Ok("0") {
5693 return Ok(());
5694 }
5695 if self.embd_gpu.get().is_none() {
5696 let buf = e.upload_u8(&self.embd.raw)?;
5697 let _ = self.embd_gpu.set(buf); }
5699 Ok(())
5700 }
5701
5702 pub fn embed(
5703 &self,
5704 e: &Engine,
5705 tokens: &[u32],
5706 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
5707 let n_embd = self.cfg.n_embd as usize;
5708 if std::env::var("MEMRA_EMBED_DEV").as_deref() != Ok("0") {
5714 let tbl = self
5715 .embd_gpu
5716 .get_or_init(|| e.upload_u8(&self.embd.raw).expect("embed table upload"));
5717 let tok_d = e.htod_u32_v(tokens)?;
5718 let (qt, rb) = self.embd.qt_and_row_bytes(n_embd);
5719 return e.embed_gather_device_td(tbl, &tok_d, tokens.len(), n_embd, qt, rb);
5720 }
5721 let x = self.embd.gather(n_embd, tokens);
5722 e.htod(&x)
5723 }
5724}
5725
5726fn illegal_pipeline_cuts(fence: &[usize], legal_boundaries: &[usize]) -> Vec<usize> {
5727 fence
5728 .get(1..fence.len().saturating_sub(1))
5729 .unwrap_or_default()
5730 .iter()
5731 .copied()
5732 .filter(|cut| !legal_boundaries.contains(cut))
5733 .collect()
5734}
5735
5736#[cfg(test)]
5737mod pipeline_cut_tests {
5738 use super::illegal_pipeline_cuts;
5739
5740 #[test]
5741 fn manual_pipeline_cuts_cannot_bypass_model_plan_boundaries() {
5742 assert!(illegal_pipeline_cuts(&[0, 8, 16, 24], &[8, 16]).is_empty());
5743 assert_eq!(illegal_pipeline_cuts(&[0, 7, 16, 24], &[8, 16]), vec![7]);
5744 assert_eq!(
5745 illegal_pipeline_cuts(&[0, 7, 15, 24], &[8, 16]),
5746 vec![7, 15]
5747 );
5748 }
5749}
5750
5751#[cfg(test)]
5752mod auto_parallel_policy_tests {
5753 use super::{parse_auto_parallel_tp_attention, parse_auto_w4a16_bf16_mmv};
5754
5755 #[test]
5756 fn automatic_w4a16_bf16_residency_defaults_on_with_explicit_rollback() {
5757 assert!(parse_auto_w4a16_bf16_mmv(None).unwrap());
5758 assert!(!parse_auto_w4a16_bf16_mmv(Some("0")).unwrap());
5759 assert!(parse_auto_w4a16_bf16_mmv(Some("1")).unwrap());
5760 assert!(parse_auto_w4a16_bf16_mmv(Some("true")).is_err());
5761 assert!(parse_auto_w4a16_bf16_mmv(Some("")).is_err());
5762 }
5763
5764 #[test]
5765 fn automatic_tp_attention_is_strict_and_defaults_off() {
5766 assert!(!parse_auto_parallel_tp_attention(None).unwrap());
5767 assert!(!parse_auto_parallel_tp_attention(Some("")).unwrap());
5768 assert!(!parse_auto_parallel_tp_attention(Some("0")).unwrap());
5769 assert!(parse_auto_parallel_tp_attention(Some("1")).unwrap());
5770 assert!(parse_auto_parallel_tp_attention(Some("true")).is_err());
5771 assert!(parse_auto_parallel_tp_attention(Some("2")).is_err());
5772 }
5773}
5774
5775#[cfg(test)]
5776mod step_expert_selection_tests {
5777 use super::{
5778 StepExpertArtifact, StepExpertLayout, StepParallelLoadConfig, StepParallelRuntimeRegistry,
5779 StepTpAttentionPlacement, select_step_expert_layout,
5780 };
5781 use crate::tp::StepEpLayerSpec;
5782
5783 fn spec(layer: usize, ranks: usize) -> StepEpLayerSpec {
5784 StepEpLayerSpec {
5785 layer,
5786 devices: (0..ranks).collect(),
5787 }
5788 }
5789
5790 #[test]
5791 fn tp2_keeps_projection_sharded_experts() {
5792 let selection = select_step_expert_layout(24, &[], &[spec(24, 2)])
5793 .unwrap()
5794 .unwrap();
5795 assert_eq!(selection.layout, StepExpertLayout::TensorParallel);
5796 assert!(selection.configured_by_tp);
5797 }
5798
5799 #[test]
5800 fn tp4_and_tp8_use_expert_ownership_without_a_second_flag() {
5801 for ranks in [4, 8] {
5802 let selection = select_step_expert_layout(24, &[], &[spec(24, ranks)])
5803 .unwrap()
5804 .unwrap();
5805 assert_eq!(selection.layout, StepExpertLayout::ExpertParallel);
5806 assert!(selection.configured_by_tp);
5807 assert_eq!(selection.spec.devices.len(), ranks);
5808 }
5809 }
5810
5811 #[test]
5812 fn explicit_ep_remains_expert_parallel() {
5813 let selection = select_step_expert_layout(24, &[spec(24, 2)], &[])
5814 .unwrap()
5815 .unwrap();
5816 assert_eq!(selection.layout, StepExpertLayout::ExpertParallel);
5817 assert!(!selection.configured_by_tp);
5818 }
5819
5820 #[test]
5821 fn conflicting_ep_and_tp_assignments_fail_closed() {
5822 let error = select_step_expert_layout(24, &[spec(24, 4)], &[spec(24, 4)]).unwrap_err();
5823 assert!(error.contains("cannot enable MEMRA_STEP_EP and MEMRA_STEP_TP together"));
5824 }
5825
5826 #[test]
5827 fn runtime_registry_owns_one_immutable_load_snapshot() {
5828 let mut source_specs = vec![spec(24, 8)];
5829 let registry = StepParallelRuntimeRegistry::with_config(StepParallelLoadConfig {
5830 ep_specs: Vec::new(),
5831 tp_specs: source_specs.clone(),
5832 native_p2p: true,
5833 ep_device_arithmetic: true,
5834 f32_mirror: true,
5835 bulk_p2p: true,
5836 nvfp4_device_routes: true,
5837 auto_parallel: true,
5838 expert_artifact: StepExpertArtifact::default(),
5839 });
5840 source_specs[0].devices.clear();
5841
5842 let stored = registry.tp_spec(24).unwrap();
5843 assert_eq!(stored.devices, (0..8).collect::<Vec<_>>());
5844 assert!(registry.config.native_p2p);
5845 assert!(registry.config.ep_device_arithmetic);
5846 assert!(registry.config.f32_mirror);
5847 assert!(registry.config.bulk_p2p);
5848 assert!(registry.config.nvfp4_device_routes);
5849 assert!(registry.config.auto_parallel);
5850 assert_eq!(
5851 registry.expert_selection(24).unwrap().unwrap().layout,
5852 StepExpertLayout::ExpertParallel
5853 );
5854
5855 let standalone = StepParallelRuntimeRegistry::default();
5856 assert!(standalone.tp_spec(24).is_none());
5857 assert!(!standalone.config.native_p2p);
5858 assert!(!standalone.config.ep_device_arithmetic);
5859 assert!(!standalone.config.f32_mirror);
5860 assert!(!standalone.config.bulk_p2p);
5861 }
5862
5863 #[test]
5864 fn rank_local_attention_uses_bounded_swa_rings_only_with_native_p2p() {
5865 assert_eq!(
5866 StepTpAttentionPlacement::resolve(true, None),
5867 StepTpAttentionPlacement::RankLocalGlobal
5868 );
5869 assert_eq!(
5870 StepTpAttentionPlacement::resolve(true, Some(512)),
5871 StepTpAttentionPlacement::RankLocalSwa
5872 );
5873 assert_eq!(
5874 StepTpAttentionPlacement::resolve(false, None),
5875 StepTpAttentionPlacement::OwnerTransportFallback
5876 );
5877 assert_eq!(
5878 StepTpAttentionPlacement::resolve(false, Some(512)),
5879 StepTpAttentionPlacement::OwnerSwa
5880 );
5881 }
5882}
5883
5884#[cfg(test)]
5885mod residency_tests {
5886 use super::{DevExpertFp8ProjectionScales, ResidentPlan, residency_bytes_by_device};
5887 use crate::model::HostExpertFp8BlockScales;
5888 use std::collections::HashMap;
5889
5890 #[test]
5891 fn pp_residency_counts_only_each_devices_expert_slice() {
5892 let tensors = [
5893 ("blk.0.ffn_gate_exps.weight", 10usize),
5894 ("blk.0.ffn_up_exps.weight", 20),
5895 ("blk.1.ffn_down_exps.weight", 30),
5896 ("blk.2.ffn_gate_exps.weight", 40),
5897 ("blk.3.ffn_up_exps.weight", 50),
5898 ("blk.0.attn_q.weight", 7),
5899 ("output.weight", 11),
5900 ];
5901 let bytes = residency_bytes_by_device(tensors, &[0, 0, 1, 1], 0);
5902 assert_eq!(bytes.experts.get(&0), Some(&60));
5903 assert_eq!(bytes.experts.get(&1), Some(&90));
5904 assert_eq!(bytes.rest, 18);
5905 assert!(bytes.saw_experts);
5906 }
5907
5908 #[test]
5909 fn pp_residency_combines_stages_that_share_one_device() {
5910 let tensors = [
5911 ("blk.0.ffn_gate_exps.weight", 10usize),
5912 ("blk.1.ffn_gate_exps.weight", 20),
5913 ("blk.2.ffn_gate_exps.weight", 30),
5914 ("blk.3.ffn_gate_exps.weight", 40),
5915 ];
5916 let bytes = residency_bytes_by_device(tensors, &[0, 0, 0, 0], 0);
5917 assert_eq!(bytes.experts.get(&0), Some(&100));
5918 assert_eq!(bytes.experts.len(), 1);
5919 }
5920
5921 #[test]
5922 fn distributed_trunk_layers_do_not_poison_local_mtp_residency_estimates() {
5923 let mut plan = ResidentPlan {
5924 primary_device: 0,
5925 layer_devices: vec![0; 81],
5926 layer_counts: HashMap::from([(0, 81)]),
5927 exact_expert_bytes: None,
5928 trunk_bytes: 0,
5929 decisions: HashMap::new(),
5930 pp: false,
5931 };
5932 plan.exclude_distributed_expert_layers(1..80);
5933 assert_eq!(plan.layer_counts.get(&0), Some(&2));
5934 }
5935
5936 #[test]
5937 fn resident_fp8_scale_slab_must_match_every_expert() {
5938 let valid = HostExpertFp8BlockScales {
5939 scales: vec![1.0; 12],
5940 rows: 2,
5941 cols: 3,
5942 expert_stride: 6,
5943 };
5944 DevExpertFp8ProjectionScales::validate(&valid, 2).unwrap();
5945
5946 let short = HostExpertFp8BlockScales {
5947 scales: vec![1.0; 11],
5948 ..valid
5949 };
5950 assert_eq!(
5951 DevExpertFp8ProjectionScales::validate(&short, 2).unwrap_err(),
5952 "block-E4M3 scale slab length mismatch: got 11, want 2x6=12"
5953 );
5954 }
5955
5956 #[test]
5957 fn resident_fp8_scale_stride_must_match_its_grid() {
5958 let invalid = HostExpertFp8BlockScales {
5959 scales: vec![1.0; 8],
5960 rows: 2,
5961 cols: 2,
5962 expert_stride: 0,
5963 };
5964 assert_eq!(
5965 DevExpertFp8ProjectionScales::validate(&invalid, 2).unwrap_err(),
5966 "block-E4M3 expert scale stride must be nonzero"
5967 );
5968 }
5969}
5970
5971#[cfg(test)]
5972mod draft_head_tests {
5973 use super::{draft_head_tensor, frspec_trim_own_head_name};
5974
5975 const STEP37_DRAFTER: &[&str] = &[
5982 "output.weight",
5983 "output_norm.weight",
5984 "token_embd.weight",
5985 "blk.45.nextn.shared_head_norm.weight",
5986 "blk.45.nextn.shared_head_head.weight",
5987 "blk.46.nextn.shared_head_head.weight",
5988 "blk.47.nextn.shared_head_head.weight",
5989 ];
5990
5991 fn present(names: &'static [&'static str]) -> impl Fn(&str) -> bool {
5992 move |t: &str| names.contains(&t)
5993 }
5994
5995 #[test]
6003 fn step37_drafter_prefers_the_blocks_own_nextn_head_over_file_level_output() {
6004 assert_eq!(
6005 draft_head_tensor(present(STEP37_DRAFTER), 45),
6006 "blk.45.nextn.shared_head_head.weight"
6007 );
6008 }
6009
6010 #[test]
6014 fn each_nextn_block_selects_its_own_head() {
6015 for n in 45..=47u32 {
6016 assert_eq!(
6017 draft_head_tensor(present(STEP37_DRAFTER), n),
6018 format!("blk.{n}.nextn.shared_head_head.weight")
6019 );
6020 }
6021 }
6022
6023 #[test]
6027 fn draft_without_a_nextn_head_falls_back_to_file_level_output() {
6028 let fr_spec: &[&str] = &["output.weight", "output_norm.weight", "d2t.weight"];
6029 assert_eq!(draft_head_tensor(present(fr_spec), 45), "output.weight");
6030 }
6031
6032 #[test]
6037 fn legacy_shared_head_is_probed_but_loses_to_shared_head_head() {
6038 let legacy_only: &[&str] = &["output.weight", "blk.45.nextn.shared_head.weight"];
6039 assert_eq!(
6040 draft_head_tensor(present(legacy_only), 45),
6041 "blk.45.nextn.shared_head.weight"
6042 );
6043
6044 let both: &[&str] = &[
6045 "output.weight",
6046 "blk.45.nextn.shared_head.weight",
6047 "blk.45.nextn.shared_head_head.weight",
6048 ];
6049 assert_eq!(
6050 draft_head_tensor(present(both), 45),
6051 "blk.45.nextn.shared_head_head.weight"
6052 );
6053 }
6054
6055 #[test]
6059 fn a_different_blocks_nextn_head_is_never_borrowed() {
6060 let wrong_block: &[&str] = &[
6061 "output.weight",
6062 "blk.46.nextn.shared_head_head.weight",
6063 "blk.47.nextn.shared_head_head.weight",
6064 ];
6065 assert_eq!(draft_head_tensor(present(wrong_block), 45), "output.weight");
6066 }
6067
6068 #[test]
6073 fn frspec_trim_prefers_the_nextn_blocks_own_head_name() {
6074 assert_eq!(
6075 frspec_trim_own_head_name(45),
6076 "blk.45.nextn.shared_head_head.weight"
6077 );
6078 assert_eq!(
6080 frspec_trim_own_head_name(45),
6081 format!("blk.{}.nextn.shared_head_head.weight", 45)
6082 );
6083 assert_eq!(
6084 frspec_trim_own_head_name(40),
6085 "blk.40.nextn.shared_head_head.weight"
6086 );
6087 }
6088}