1use crate::Engine;
6use crate::model::{EmbedHost, GpuTensor, HostExps};
7use cudarc::driver::CudaSlice;
8use memra_gguf::config::ModelConfig;
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 expert_artifact: StepExpertArtifact,
103}
104
105#[derive(Default)]
106pub(crate) struct StepParallelRuntimeRegistry {
107 config: StepParallelLoadConfig,
108 runtimes: HashMap<(Vec<usize>, bool, bool, bool), Arc<crate::tp::TpE4m3HostBounce>>,
109}
110
111#[derive(Clone, Copy, Debug, PartialEq, Eq)]
112enum StepExpertLayout {
113 TensorParallel,
114 ExpertParallel,
115}
116
117#[derive(Clone, Debug, PartialEq, Eq)]
118struct StepExpertSelection {
119 spec: crate::tp::StepEpLayerSpec,
120 layout: StepExpertLayout,
121 configured_by_tp: bool,
122}
123
124fn select_step_expert_layout(
125 layer: usize,
126 ep_specs: &[crate::tp::StepEpLayerSpec],
127 tp_specs: &[crate::tp::StepTpLayerSpec],
128) -> Result<Option<StepExpertSelection>, String> {
129 let ep = ep_specs.iter().find(|spec| spec.layer == layer);
130 let tp = tp_specs.iter().find(|spec| spec.layer == layer);
131 if ep.is_some() && tp.is_some() {
132 return Err(format!(
133 "Step layer {layer} cannot enable MEMRA_STEP_EP and MEMRA_STEP_TP together"
134 ));
135 }
136 Ok(match (ep, tp) {
137 (Some(spec), None) => Some(StepExpertSelection {
138 spec: spec.clone(),
139 layout: StepExpertLayout::ExpertParallel,
140 configured_by_tp: false,
141 }),
142 (None, Some(spec)) => Some(StepExpertSelection {
143 spec: spec.clone(),
144 layout: if spec.devices.len() > 2 {
145 StepExpertLayout::ExpertParallel
146 } else {
147 StepExpertLayout::TensorParallel
148 },
149 configured_by_tp: true,
150 }),
151 (None, None) => None,
152 (Some(_), Some(_)) => unreachable!(),
153 })
154}
155
156impl StepParallelRuntimeRegistry {
157 fn with_config(config: StepParallelLoadConfig) -> Self {
158 Self {
159 config,
160 runtimes: HashMap::new(),
161 }
162 }
163
164 fn tp_spec(&self, layer: usize) -> Option<&crate::tp::StepTpLayerSpec> {
165 self.config.tp_specs.iter().find(|spec| spec.layer == layer)
166 }
167
168 fn expert_selection(&self, layer: usize) -> Result<Option<StepExpertSelection>, String> {
169 select_step_expert_layout(layer, &self.config.ep_specs, &self.config.tp_specs)
170 }
171
172 fn runtime(
173 &mut self,
174 devices: &[usize],
175 native_p2p: bool,
176 ep_device_arithmetic: bool,
177 ) -> Result<Arc<crate::tp::TpE4m3HostBounce>, Box<dyn std::error::Error>> {
178 let bulk_p2p = self.config.bulk_p2p && native_p2p;
179 let key = (devices.to_vec(), native_p2p, ep_device_arithmetic, bulk_p2p);
180 if let Some(runtime) = self.runtimes.get(&key) {
181 return Ok(Arc::clone(runtime));
182 }
183 let runtime = Arc::new(crate::tp::TpE4m3HostBounce::new_configured(
184 devices,
185 native_p2p,
186 ep_device_arithmetic,
187 bulk_p2p,
188 )?);
189 let names = runtime.device_names()?;
190 if names
191 .iter()
192 .any(|name| !name.contains("RTX PRO 6000") || !name.contains("Blackwell"))
193 {
194 return Err(format!(
195 "Step distributed execution is qualified only on RTX PRO 6000 Blackwell, \
196 got {names:?}"
197 )
198 .into());
199 }
200 self.runtimes.insert(key, Arc::clone(&runtime));
201 Ok(runtime)
202 }
203}
204
205impl ResidentPlan {
206 fn from_layout(
207 src: &dyn TensorSource,
208 primary_device: usize,
209 layer_devices: Vec<usize>,
210 pp: bool,
211 ) -> Self {
212 let mut layer_counts = HashMap::new();
213 for &device in &layer_devices {
214 *layer_counts.entry(device).or_default() += 1;
215 }
216 let (exact_expert_bytes, trunk_bytes) = match src.gguf() {
217 Some(g) => {
218 let bytes = residency_bytes_by_device(
219 g.tensors
220 .iter()
221 .map(|t| (t.name.as_str(), t.n_bytes as usize)),
222 &layer_devices,
223 primary_device,
224 );
225 if bytes.saw_experts {
226 (Some(bytes.experts), bytes.rest)
227 } else {
228 (None, 0)
229 }
230 }
231 None => (None, 0),
232 };
233 Self {
234 primary_device,
235 layer_devices,
236 layer_counts,
237 exact_expert_bytes,
238 trunk_bytes,
239 decisions: HashMap::new(),
240 pp,
241 }
242 }
243
244 pub(crate) fn unsharded(e: &Engine, src: &dyn TensorSource, cfg: &ModelConfig) -> Self {
245 let device = e.ctx().ordinal();
246 Self::from_layout(src, device, vec![device; cfg.n_layer as usize], false)
247 }
248
249 pub(crate) fn pp(
250 e: &Engine,
251 src: &dyn TensorSource,
252 cfg: &ModelConfig,
253 n_trunk: usize,
254 ) -> Result<Self, Box<dyn std::error::Error>> {
255 let primary = e.ctx().ordinal();
256 let Some(_fence) = crate::pp::pp_cuts(n_trunk) else {
257 return Ok(Self::unsharded(e, src, cfg));
258 };
259 let mut layer_devices = vec![primary; cfg.n_layer as usize];
260 for (il, device) in layer_devices.iter_mut().take(n_trunk).enumerate() {
261 *device = crate::pp::layer_engine(e, n_trunk, il)?.ctx().ordinal();
262 }
263 Ok(Self::from_layout(src, primary, layer_devices, true))
264 }
265
266 fn should_reside(&mut self, e: &Engine, il: usize, per_layer: usize) -> bool {
267 let device = self
268 .layer_devices
269 .get(il)
270 .copied()
271 .unwrap_or(self.primary_device);
272 debug_assert_eq!(e.ctx().ordinal(), device);
273 if let Some(&decision) = self.decisions.get(&device) {
274 return decision;
275 }
276 if std::env::var("MEMRA_MOE_RESIDENT").as_deref() == Ok("0") {
277 self.decisions.insert(device, false);
278 return false;
279 }
280 let (free, _total) = match e.ctx().mem_get_info() {
281 Ok(v) => v,
282 Err(_) => {
283 self.decisions.insert(device, false);
284 return false;
285 }
286 };
287 let projected = self
288 .exact_expert_bytes
289 .as_ref()
290 .map(|bytes| bytes.get(&device).copied().unwrap_or(0))
291 .unwrap_or(per_layer * self.layer_counts.get(&device).copied().unwrap_or(1));
292 let budget = std::env::var("MEMRA_MOE_RESIDENT_GB")
293 .ok()
294 .and_then(|v| v.parse::<f64>().ok())
295 .map(|gb| (gb * 1e9) as usize)
296 .unwrap_or_else(|| {
297 let reserve = std::env::var("MEMRA_MOE_RESIDENT_HEADROOM_GB")
298 .ok()
299 .and_then(|v| v.parse::<f64>().ok())
300 .map(|gb| (gb * 1e9) as usize)
301 .unwrap_or(2_000_000_000);
302 (free as usize).saturating_sub(self.trunk_bytes + reserve)
303 });
304 let ok = projected <= budget;
305 eprintln!(
306 "[moe] resident-experts decision ({}dev{}): experts {:.2}GB + trunk {:.2}GB vs free {:.2}GB (expert budget {:.2}GB) -> {}",
307 if self.pp { "PP " } else { "" },
308 device,
309 projected as f64 / 1e9,
310 self.trunk_bytes as f64 / 1e9,
311 free as f64 / 1e9,
312 budget as f64 / 1e9,
313 if ok { "RESIDENT" } else { "SLRU cache" }
314 );
315 self.decisions.insert(device, ok);
316 ok
317 }
318}
319
320fn load_mixer_kind(
322 e: &Engine,
323 src: &dyn TensorSource,
324 cfg: &ModelConfig,
325 il: u32,
326 attention: &AttentionPlan,
327 step_runtimes: &mut StepParallelRuntimeRegistry,
328) -> Result<Mixer, Box<dyn std::error::Error>> {
329 let p = |s: &str| format!("blk.{il}.{s}");
330 Ok(match attention {
331 AttentionPlan::Mla(mla) => Mixer::Mla(MlaAttnLayer::load(e, src, il, mla)?),
332 AttentionPlan::Full(full)
333 | AttentionPlan::SlidingWindow {
334 attention: full, ..
335 } => {
336 Mixer::Full(FullAttnLayer {
337 wq: load_t(e, src, &p("attn_q.weight"))?,
338 wk: load_t(e, src, &p("attn_k.weight"))?,
339 wv: match load_opt(e, src, &p("attn_v.weight"))? {
344 Some(v) => v,
345 None => load_t(e, src, &p("attn_k.weight"))?,
346 },
347 wo: load_t(e, src, &p("attn_output.weight"))?,
348 q_norm: load_t(e, src, &p("attn_q_norm.weight"))?,
349 k_norm: load_t(e, src, &p("attn_k_norm.weight"))?,
350 attn_gate: if full.output_gate
354 == memra_gguf::config::AttentionGateKind::SeparateHead
355 {
356 Some(load_t(e, src, &p("attn_gate.weight"))?)
357 } else {
358 None
359 },
360 step_tp_qkv: build_step_tp_qkv(e, src, cfg, il as usize, step_runtimes)?,
361 })
362 }
363 AttentionPlan::GatedDeltaNet(geometry) => Mixer::Linear(LinearAttnLayer {
364 geometry: *geometry,
365 wqkv: load_t(e, src, &p("attn_qkv.weight"))?,
366 wqkv_gate: load_t(e, src, &p("attn_gate.weight"))?,
367 ssm_beta: load_t(e, src, &p("ssm_beta.weight"))?,
368 ssm_alpha: load_t(e, src, &p("ssm_alpha.weight"))?,
369 ssm_a: load_t(e, src, &p("ssm_a"))?,
370 ssm_dt: load_t(e, src, &p("ssm_dt.bias"))?,
371 ssm_conv1d: load_t(e, src, &p("ssm_conv1d.weight"))?,
372 ssm_norm: load_t(e, src, &p("ssm_norm.weight"))?,
373 ssm_out: load_t(e, src, &p("ssm_out.weight"))?,
374 }),
375 })
376}
377
378pub(crate) fn load_ffn(
385 e: &Engine,
386 src: &dyn TensorSource,
387 cfg: &ModelConfig,
388 mlp: &MlpPlan,
389 il: u32,
390 spill: Option<(&GgufFile, &mut crate::spill::SpillCtx)>,
391 resident: &mut ResidentPlan,
392 step_runtimes: &mut StepParallelRuntimeRegistry,
393) -> Result<Ffn, Box<dyn std::error::Error>> {
394 let p = |s: &str| format!("blk.{il}.{s}");
395 let artifact_dense = matches!(mlp, MlpPlan::Moe(_))
402 && !src.has(&p("ffn_gate_exps.weight"))
403 && !src.has(&p("ffn_gate_up_exps.weight"))
404 && src.has(&p("ffn_gate.weight"));
405 Ok(if artifact_dense {
406 Ffn::Dense {
407 ffn_gate: GpuTensor::load_from_source(e, src, &p("ffn_gate.weight"))?,
408 ffn_up: GpuTensor::load_from_source(e, src, &p("ffn_up.weight"))?,
409 ffn_down: GpuTensor::load_from_source(e, src, &p("ffn_down.weight"))?,
410 }
411 } else if let MlpPlan::Moe(moe) = mlp {
412 let n_expert = moe.expert_count as usize;
413 let (gate_exps, up_exps, down_exps) = match spill {
419 Some((g, ctx)) => (
420 HostExps::load_tiered(e, g, &p("ffn_gate_exps.weight"), ctx)?,
421 HostExps::load_tiered(e, g, &p("ffn_up_exps.weight"), ctx)?,
422 HostExps::load_tiered(e, g, &p("ffn_down_exps.weight"), ctx)?,
423 ),
424 None => {
425 let exps = |e: &Engine, n: &str| -> Result<HostExps, Box<dyn std::error::Error>> {
426 if src.has(n) {
427 HostExps::load_stacked_from_source(e, src, n)
428 } else {
429 HostExps::load_from_source(e, src, n, n_expert)
430 }
431 };
432 let fused = p("ffn_gate_up_exps.weight");
434 if !src.has(&p("ffn_gate_exps.weight")) && src.has(&fused) {
435 let ff = moe.expert_intermediate_size as usize;
436 (
437 HostExps::load_stacked_split_from_source(e, src, &fused, 0, ff)?,
438 HostExps::load_stacked_split_from_source(e, src, &fused, ff, 2 * ff)?,
439 exps(e, &p("ffn_down_exps.weight"))?,
440 )
441 } else {
442 (
443 exps(e, &p("ffn_gate_exps.weight"))?,
444 exps(e, &p("ffn_up_exps.weight"))?,
445 exps(e, &p("ffn_down_exps.weight"))?,
446 )
447 }
448 }
449 };
450 let (step_ep, step_tp) = build_step_distributed_exps(
451 e,
452 cfg,
453 src,
454 il as usize,
455 &gate_exps,
456 &up_exps,
457 &down_exps,
458 step_runtimes,
459 )?;
460 let dev_exps = if step_ep.is_some() || step_tp.is_some() {
466 None
467 } else {
468 build_dev_exps(e, resident, il as usize, &gate_exps, &up_exps, &down_exps)?
469 };
470 let mut macro_row = vec![1.0f32; 3 * n_expert];
472 for (slot, exps) in [(0usize, &gate_exps), (1, &up_exps), (2, &down_exps)] {
473 if let Some(ms) = exps.macros.as_ref() {
474 macro_row[slot * n_expert..(slot + 1) * n_expert].copy_from_slice(ms);
475 }
476 }
477 let has_macros = macro_row.iter().any(|&m| m != 1.0);
478 let dev_macros = e.htod(¯o_row)?;
479 let exp_probs_b = src
482 .find(&p("exp_probs_b.bias"))
483 .map(|v| memra_gguf::dequant::dequantize(v.ggml_type, &v.bytes, n_expert));
484 let active_experts = src.active_experts(il).map(<[bool]>::to_vec);
485 let route_bias = exp_probs_b.clone().unwrap_or_else(|| vec![0.0; n_expert]);
486 let active_row: Vec<u8> = active_experts
487 .as_ref()
488 .map(|mask| mask.iter().map(|&is_active| u8::from(is_active)).collect())
489 .unwrap_or_else(|| vec![1; n_expert]);
490 let exp_probs_b_dev = e.htod(&route_bias)?;
491 let active_experts_dev = e.htod_bytes(&active_row)?;
492 Ffn::Moe(MoeWeights {
493 gate_inp: load_t(e, src, &p("ffn_gate_inp.weight"))?,
494 gate_inp_shexp: load_opt(e, src, &p("ffn_gate_inp_shexp.weight"))?,
495 exp_probs_b,
496 exp_probs_b_dev,
497 active_experts,
498 active_experts_dev,
499 gate_exps,
500 up_exps,
501 down_exps,
502 gate_shexp: load_opt(e, src, &p("ffn_gate_shexp.weight"))?,
503 up_shexp: load_opt(e, src, &p("ffn_up_shexp.weight"))?,
504 down_shexp: load_opt(e, src, &p("ffn_down_shexp.weight"))?,
505 dev_exps,
506 step_ep,
507 step_tp,
508 dev_macros,
509 has_macros,
510 })
511 } else {
512 Ffn::Dense {
513 ffn_gate: load_t(e, src, &p("ffn_gate.weight"))?,
514 ffn_up: load_t(e, src, &p("ffn_up.weight"))?,
515 ffn_down: load_t(e, src, &p("ffn_down.weight"))?,
516 }
517 })
518}
519
520fn host_e4m3_bank(
521 exps: &HostExps,
522) -> Result<crate::tp::E4m3ExpertBank<'_>, Box<dyn std::error::Error>> {
523 if exps.qtype != crate::QT_F8_E4M3_BLK {
524 return Err(format!(
525 "Step EP requires native block-E4M3 expert banks, got qtype {}",
526 exps.qtype
527 )
528 .into());
529 }
530 let scales = exps
531 .fp8_blk
532 .as_ref()
533 .ok_or("Step EP native expert bank has no block-E4M3 scale plane")?;
534 Ok(crate::tp::E4m3ExpertBank {
535 codes: exps.bytes.as_bytes(),
536 scales: &scales.scales,
537 expert_count: exps.n_expert,
538 out_features: exps.out_f,
539 in_features: exps.in_f,
540 })
541}
542
543fn validate_step_expert_specs(
544 contract: &crate::parallel::ModelParallelContract,
545 flag: &str,
546 specs: &[crate::tp::StepEpLayerSpec],
547 allow_dense_attention_only: bool,
548) -> Result<(), Box<dyn std::error::Error>> {
549 for candidate in specs {
550 if candidate.layer >= contract.trunk_layers {
551 return Err(format!(
552 "{flag} layer {} is outside Step trunk layers 0..{}",
553 candidate.layer, contract.trunk_layers
554 )
555 .into());
556 }
557 if candidate.layer < contract.dense_prefix_layers {
558 if allow_dense_attention_only {
559 continue;
560 }
561 return Err(format!(
562 "{flag} layer {} is outside Step routed-expert layers {}..{}",
563 candidate.layer, contract.dense_prefix_layers, contract.trunk_layers
564 )
565 .into());
566 }
567 }
568 Ok(())
569}
570
571fn validate_step_expert_activation_layout(
572 cfg: &ModelConfig,
573 flag: &str,
574 selection: &StepExpertSelection,
575) -> Result<(), Box<dyn std::error::Error>> {
576 let _ = (cfg, flag, selection);
582 Ok(())
583}
584
585fn prepare_step_parallel_load(
586 e: &Engine,
587 src: &dyn TensorSource,
588 cfg: &ModelConfig,
589 trunk_layers: usize,
590) -> Result<StepParallelLoadConfig, Box<dyn std::error::Error>> {
591 let tp_specs = crate::tp::step_tp_layer_specs()?;
592 let ep_specs = crate::tp::step_ep_layer_specs()?;
593 let device_arithmetic = crate::tp::step_ep_device_arithmetic_enabled()?;
594 let f32_mirror = crate::tp::step_tp_f32_mirror_enabled()?;
595 let bulk_p2p = crate::tp::step_tp_bulk_p2p_enabled()?;
596 if tp_specs.is_empty() {
597 if device_arithmetic || f32_mirror || bulk_p2p {
598 return Err(
599 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1, MEMRA_STEP_TP_F32_MIRROR=1, or \
600 MEMRA_STEP_TP_BULK_P2P=1 requires MEMRA_STEP_TP; device arithmetic and bulk \
601 transport also require MEMRA_STEP_TP_NATIVE_P2P=1"
602 .into(),
603 );
604 }
605 let expert_artifact = if ep_specs.is_empty() {
608 StepExpertArtifact::default()
609 } else {
610 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
611 match crate::parallel::validate_step_fp8_checkpoint(src, &contract) {
612 Ok(_) => StepExpertArtifact::E4m3,
613 Err(fp8_error) => {
614 match crate::parallel::validate_step_nvfp4_checkpoint(src, &contract) {
615 Ok(_) => StepExpertArtifact::Nvfp4,
616 Err(nvfp4_error) => {
617 return Err(format!(
618 "Step checkpoint qualifies as neither native expert artifact \
619 class: [E4M3] {fp8_error} [NVFP4] {nvfp4_error}"
620 )
621 .into());
622 }
623 }
624 }
625 }
626 };
627 return Ok(StepParallelLoadConfig {
628 ep_specs,
629 expert_artifact,
630 ..StepParallelLoadConfig::default()
631 });
632 }
633 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
634 validate_step_expert_specs(&contract, "MEMRA_STEP_EP", &ep_specs, false)?;
635 validate_step_expert_specs(&contract, "MEMRA_STEP_TP", &tp_specs, true)?;
636 for spec in &tp_specs {
637 let selection = select_step_expert_layout(spec.layer, &ep_specs, &tp_specs)?
638 .ok_or("Step TP expert selection disappeared during preflight")?;
639 validate_step_expert_activation_layout(cfg, "MEMRA_STEP_TP", &selection)?;
640 }
641
642 let layer_owners = (0..trunk_layers)
643 .map(|layer| {
644 crate::pp::layer_engine(e, trunk_layers, layer).map(|engine| engine.ctx().ordinal())
645 })
646 .collect::<Result<Vec<_>, _>>()?;
647 let plan = contract.preflight_step_tp_specs(
648 tp_specs
649 .iter()
650 .map(|spec| (spec.layer, spec.devices.as_slice())),
651 &layer_owners,
652 )?;
653
654 for devices in &plan.runtime_groups {
655 let hardware = crate::parallel::detect_uniform_hardware(devices)?;
656 if !contract.hardware_targets.contains(&hardware) {
657 return Err(format!(
658 "{} has no qualified {hardware:?} TP contract for devices {devices:?}",
659 contract.variant
660 )
661 .into());
662 }
663 }
664
665 let native_p2p = crate::tp::step_tp_native_p2p_enabled()?;
666 if bulk_p2p && !native_p2p {
667 return Err("MEMRA_STEP_TP_BULK_P2P=1 requires MEMRA_STEP_TP_NATIVE_P2P=1".into());
668 }
669 if device_arithmetic
670 && (!ep_specs.is_empty()
671 || !native_p2p
672 || plan.expert_parallel_layers() == 0
673 || plan.tensor_parallel_expert_layers() != 0)
674 {
675 return Err(
676 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires native-P2P TP4/TP8 \
677 expert ownership for every selected routed-expert layer"
678 .into(),
679 );
680 }
681 let (qualified_experts, expert_artifact) =
685 match crate::parallel::validate_step_fp8_checkpoint(src, &contract) {
686 Ok(qualified) => (qualified, StepExpertArtifact::E4m3),
687 Err(fp8_error) => match crate::parallel::validate_step_nvfp4_checkpoint(src, &contract)
688 {
689 Ok(qualified) => (qualified, StepExpertArtifact::Nvfp4),
690 Err(nvfp4_error) => {
691 return Err(format!(
692 "Step checkpoint qualifies as neither native expert artifact class: \
693 [E4M3] {fp8_error} [NVFP4] {nvfp4_error}"
694 )
695 .into());
696 }
697 },
698 };
699 if expert_artifact == StepExpertArtifact::Nvfp4 {
700 if device_arithmetic {
701 return Err(
702 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 is qualified for the E4M3 expert artifact \
703 only; the NVFP4 expert program is host-canonical in this increment"
704 .into(),
705 );
706 }
707 if bulk_p2p {
712 return Err(
713 "MEMRA_STEP_TP_BULK_P2P=1 is qualified for the E4M3 expert artifact only; the \
714 NVFP4 bank transport increment has not landed"
715 .into(),
716 );
717 }
718 }
719
720 if f32_mirror {
721 eprintln!(
722 "[step-tp-preflight] layers={} full_trunk={} runtime_groups={} \
723 dense_attention_layers={} tensor_expert_layers={} expert_owner_layers={} \
724 qualified_fp8_expert_projection_slices={} owner_first=true \
725 hardware=rtx-pro-6000-blackwell \
726 native_p2p={} bulk_p2p={} device_arithmetic={} bf16_residency=f32-mirror \
727 weights_loaded=false performance_claim=false",
728 plan.layers.len(),
729 plan.full_trunk,
730 plan.runtime_groups.len(),
731 plan.dense_attention_layers(),
732 plan.tensor_parallel_expert_layers(),
733 plan.expert_parallel_layers(),
734 qualified_experts,
735 native_p2p,
736 bulk_p2p,
737 device_arithmetic,
738 );
739 } else {
740 eprintln!(
741 "[step-tp-preflight] layers={} full_trunk={} runtime_groups={} \
742 dense_attention_layers={} tensor_expert_layers={} expert_owner_layers={} \
743 qualified_fp8_expert_projection_slices={} owner_first=true \
744 hardware=rtx-pro-6000-blackwell \
745 native_p2p={} bulk_p2p={} device_arithmetic={} \
746 weights_loaded=false performance_claim=false",
747 plan.layers.len(),
748 plan.full_trunk,
749 plan.runtime_groups.len(),
750 plan.dense_attention_layers(),
751 plan.tensor_parallel_expert_layers(),
752 plan.expert_parallel_layers(),
753 qualified_experts,
754 native_p2p,
755 bulk_p2p,
756 device_arithmetic,
757 );
758 }
759 Ok(StepParallelLoadConfig {
760 ep_specs,
761 tp_specs,
762 native_p2p,
763 ep_device_arithmetic: device_arithmetic,
764 f32_mirror,
765 bulk_p2p,
766 expert_artifact,
767 })
768}
769
770fn step_nvfp4_native<'a>(
772 src: &'a dyn TensorSource,
773 layer: usize,
774 proj: &str,
775) -> Result<memra_gguf::source::Nvfp4StackedNative<'a>, Box<dyn std::error::Error>> {
776 let name = format!("blk.{layer}.ffn_{proj}_exps.weight");
777 src.find_nvfp4_stacked_native(&name)
778 .ok_or_else(|| format!("Step NVFP4 expert program is missing native bank {name}").into())
779}
780
781fn step_nvfp4_bank<'a>(
783 native: &'a memra_gguf::source::Nvfp4StackedNative<'a>,
784) -> crate::tp::Nvfp4ExpertBank<'a> {
785 crate::tp::Nvfp4ExpertBank {
786 codes: native.codes,
787 scales: native.scales,
788 macros: &native.macros,
789 expert_count: native.n_expert,
790 out_features: native.out_f,
791 in_features: native.in_f,
792 }
793}
794
795fn build_step_distributed_exps(
796 e: &Engine,
797 cfg: &ModelConfig,
798 src: &dyn TensorSource,
799 layer: usize,
800 gate: &HostExps,
801 up: &HostExps,
802 down: &HostExps,
803 step_runtimes: &mut StepParallelRuntimeRegistry,
804) -> Result<(Option<StepEpExps>, Option<StepTpExps>), Box<dyn std::error::Error>> {
805 let ep_device_arithmetic = step_runtimes.config.ep_device_arithmetic;
806 if step_runtimes.config.ep_specs.is_empty() && step_runtimes.config.tp_specs.is_empty() {
807 if ep_device_arithmetic {
808 return Err(
809 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires MEMRA_STEP_TP and \
810 MEMRA_STEP_TP_NATIVE_P2P=1"
811 .into(),
812 );
813 }
814 return Ok((None, None));
815 }
816 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
817 validate_step_expert_specs(
818 &contract,
819 "MEMRA_STEP_EP",
820 &step_runtimes.config.ep_specs,
821 false,
822 )?;
823 validate_step_expert_specs(
824 &contract,
825 "MEMRA_STEP_TP",
826 &step_runtimes.config.tp_specs,
827 true,
828 )?;
829 let Some(selection) = step_runtimes.expert_selection(layer)? else {
830 return Ok((None, None));
831 };
832 validate_step_expert_activation_layout(
833 cfg,
834 if selection.configured_by_tp {
835 "MEMRA_STEP_TP"
836 } else {
837 "MEMRA_STEP_EP"
838 },
839 &selection,
840 )?;
841 let activation_limit = cfg.clamp_exp_at(layer as u32);
842 let owner = e.ctx().ordinal();
843 if !selection.spec.devices.contains(&owner) {
844 let flag = if selection.configured_by_tp {
845 "MEMRA_STEP_TP"
846 } else {
847 "MEMRA_STEP_EP"
848 };
849 return Err(format!(
850 "{flag} layer {layer} owning PP device {owner} is absent from rank devices {:?}",
851 selection.spec.devices
852 )
853 .into());
854 }
855 let expert_parallel = selection.layout == StepExpertLayout::ExpertParallel;
856 if selection.configured_by_tp {
857 contract.plan(crate::parallel::TopologyRequest {
858 pipeline: 1,
859 tensor: selection.spec.devices.len(),
860 expert_parallel,
861 available_devices: selection.spec.devices.len(),
862 hardware: crate::parallel::HardwareTarget::RtxPro6000Blackwell,
863 })?;
864 }
865 let native_p2p = selection.configured_by_tp && step_runtimes.config.native_p2p;
866 if ep_device_arithmetic
867 && (!selection.configured_by_tp
868 || selection.layout != StepExpertLayout::ExpertParallel
869 || !native_p2p)
870 {
871 return Err(
872 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires a MEMRA_STEP_TP TP4/TP8 \
873 expert-owner layer and MEMRA_STEP_TP_NATIVE_P2P=1"
874 .into(),
875 );
876 }
877 let runtime =
878 step_runtimes.runtime(&selection.spec.devices, native_p2p, ep_device_arithmetic)?;
879 let expert_artifact = step_runtimes.config.expert_artifact;
880 match selection.layout {
881 StepExpertLayout::ExpertParallel => {
882 if expert_artifact == StepExpertArtifact::Nvfp4 {
883 let runtime = step_runtimes.runtime(
890 &selection.spec.devices,
891 step_runtimes.config.native_p2p,
892 false,
893 )?;
894 let gate_native = step_nvfp4_native(src, layer, "gate")?;
895 let up_native = step_nvfp4_native(src, layer, "up")?;
896 let down_native = step_nvfp4_native(src, layer, "down")?;
897 let experts = runtime.upload_expert_parallel_nvfp4(
898 step_nvfp4_bank(&gate_native),
899 step_nvfp4_bank(&up_native),
900 step_nvfp4_bank(&down_native),
901 )?;
902 eprintln!(
903 "[step-ep] layer={layer} devices={:?} experts={} artifact=nvfp4 \
904 expert_layout=expert-parallel expert_transport=host-bounce \
905 macro_fold=post-kernel-once native_p2p=false performance_claim=false",
906 selection.spec.devices, contract.expert_count
907 );
908 if let Some(limit) = activation_limit {
909 eprintln!(
910 "[step-ep-clamp] load layer={layer} routed_clamp={limit} \
911 formula=min-silu-times-clamped-up performance_claim=false"
912 );
913 }
914 return Ok((
915 Some(StepEpExps {
916 runtime,
917 experts: StepEpExpertBank::Nvfp4(experts),
918 devices: selection.spec.devices,
919 configured_by_tp: selection.configured_by_tp,
920 activation_limit,
921 grouped_decode: None,
922 }),
923 None,
924 ));
925 }
926 let experts = runtime.upload_expert_parallel(
927 host_e4m3_bank(gate)?,
928 host_e4m3_bank(up)?,
929 host_e4m3_bank(down)?,
930 )?;
931 let grouped_decode = if ep_device_arithmetic {
932 let tokens = 1;
933 let selected = (0..contract.experts_per_token).collect::<Vec<_>>();
934 let input = vec![0.0f32; contract.hidden_size];
935 let route_weights = vec![1.0f32; contract.experts_per_token];
936 let projection = runtime.prepare_step_grouped_expert_parallel_gate_with_capacity(
937 &experts,
938 &input,
939 tokens,
940 &selected,
941 activation_limit,
942 tokens,
943 )?;
944 let combine = runtime
945 .prepare_step_grouped_expert_parallel_combine(&projection, &route_weights)?;
946 Some(std::sync::Mutex::new(StepEpGroupedDecode {
947 projection,
948 combine,
949 }))
950 } else {
951 None
952 };
953 if selection.configured_by_tp {
954 eprintln!(
955 "[step-tp-ep] layer={layer} devices={:?} experts={} tp={} \
956 attention_layout=tensor-parallel expert_layout=expert-parallel \
957 expert_transport={} tp_transport={} native_p2p={} \
958 activation={} accumulation={} output={} \
959 grouped_decode_prepared={} grouped_decode_capacity=1 \
960 performance_claim=false",
961 selection.spec.devices,
962 contract.expert_count,
963 selection.spec.devices.len(),
964 runtime.transport_label(),
965 runtime.transport_label(),
966 runtime.native_p2p(),
967 runtime.expert_activation_label(),
968 runtime.expert_accumulation_label(),
969 runtime.expert_output_label(),
970 grouped_decode.is_some(),
971 );
972 } else {
973 eprintln!(
974 "[step-ep] layer={layer} devices={:?} experts={} \
975 expert_layout=expert-parallel expert_transport=host-bounce \
976 native_p2p=false performance_claim=false",
977 selection.spec.devices, contract.expert_count
978 );
979 }
980 if let Some(limit) = activation_limit {
981 eprintln!(
982 "[step-ep-clamp] load layer={layer} routed_clamp={limit} \
983 formula=min-silu-times-clamped-up performance_claim=false"
984 );
985 }
986 Ok((
987 Some(StepEpExps {
988 runtime,
989 experts: StepEpExpertBank::E4m3(experts),
990 devices: selection.spec.devices,
991 configured_by_tp: selection.configured_by_tp,
992 activation_limit,
993 grouped_decode,
994 }),
995 None,
996 ))
997 }
998 StepExpertLayout::TensorParallel => {
999 if activation_limit.is_some() && expert_artifact == StepExpertArtifact::E4m3 {
1000 return Err(format!(
1001 "layer {layer} uses the routed SwiGLU clamp and the E4M3 TP expert \
1002 program has no clamp arm; select EP for this layer (the NVFP4 TP \
1003 program carries the clamp)"
1004 )
1005 .into());
1006 }
1007 let experts = if expert_artifact == StepExpertArtifact::Nvfp4 {
1008 let gate_native = step_nvfp4_native(src, layer, "gate")?;
1009 let up_native = step_nvfp4_native(src, layer, "up")?;
1010 let down_native = step_nvfp4_native(src, layer, "down")?;
1011 StepTpExpertBank::Nvfp4(runtime.upload_tensor_parallel_nvfp4(
1012 step_nvfp4_bank(&gate_native),
1013 step_nvfp4_bank(&up_native),
1014 step_nvfp4_bank(&down_native),
1015 )?)
1016 } else {
1017 StepTpExpertBank::E4m3(runtime.upload_tensor_parallel(
1018 host_e4m3_bank(gate)?,
1019 host_e4m3_bank(up)?,
1020 host_e4m3_bank(down)?,
1021 )?)
1022 };
1023 eprintln!(
1024 "[step-tp] layer={layer} devices={:?} experts={} tp={} artifact={} \
1025 expert_layout=tensor-parallel transport={} native_p2p={} \
1026 performance_claim=false",
1027 selection.spec.devices,
1028 contract.expert_count,
1029 selection.spec.devices.len(),
1030 match expert_artifact {
1031 StepExpertArtifact::E4m3 => "e4m3",
1032 StepExpertArtifact::Nvfp4 => "nvfp4",
1033 },
1034 runtime.transport_label(),
1035 runtime.native_p2p(),
1036 );
1037 if let Some(limit) = activation_limit {
1038 eprintln!(
1039 "[step-tp-clamp] load layer={layer} routed_clamp={limit} \
1040 formula=min-silu-times-clamped-up performance_claim=false"
1041 );
1042 }
1043 Ok((
1044 None,
1045 Some(StepTpExps {
1046 runtime,
1047 experts,
1048 devices: selection.spec.devices,
1049 activation_limit,
1050 }),
1051 ))
1052 }
1053 }
1054}
1055
1056fn upload_step_bf16_column(
1057 runtime: &crate::tp::TpE4m3HostBounce,
1058 src: &dyn TensorSource,
1059 name: &str,
1060 expected_in: usize,
1061 expected_out: usize,
1062 f32_mirror: bool,
1063) -> Result<crate::tp::ResidentBf16ColumnParallel, Box<dyn std::error::Error>> {
1064 let tensor = src
1065 .find(name)
1066 .ok_or_else(|| format!("Step TP projection is missing {name}"))?;
1067 if tensor.ggml_type != GgmlType::BF16 {
1068 return Err(format!(
1069 "Step TP projection {name} must preserve checkpoint BF16 bytes, got {:?}",
1070 tensor.ggml_type
1071 )
1072 .into());
1073 }
1074 if tensor.ne.len() != 2 {
1075 return Err(format!(
1076 "Step TP projection {name} must be a 2-D matrix, got shape {:?}",
1077 tensor.ne
1078 )
1079 .into());
1080 }
1081 let matrix = crate::tp::Bf16Matrix {
1082 bytes: tensor.bytes.as_ref(),
1083 in_features: tensor.ne[0] as usize,
1084 out_features: tensor.ne[1] as usize,
1085 };
1086 matrix.validate()?;
1087 if matrix.in_features != expected_in || matrix.out_features != expected_out {
1088 return Err(format!(
1089 "Step TP projection {name} shape {}x{} != registered {expected_out}x{expected_in}",
1090 matrix.out_features, matrix.in_features
1091 )
1092 .into());
1093 }
1094 Ok(if f32_mirror {
1095 runtime.upload_step_bf16_column_parallel_f32_mirror(matrix)?
1096 } else {
1097 runtime.upload_step_bf16_column_parallel(matrix)?
1098 })
1099}
1100
1101fn upload_step_bf16_row(
1102 runtime: &crate::tp::TpE4m3HostBounce,
1103 src: &dyn TensorSource,
1104 name: &str,
1105 expected_in: usize,
1106 expected_out: usize,
1107 f32_mirror: bool,
1108) -> Result<crate::tp::ResidentStepBf16RowParallel, Box<dyn std::error::Error>> {
1109 let tensor = src
1110 .find(name)
1111 .ok_or_else(|| format!("Step TP projection is missing {name}"))?;
1112 if tensor.ggml_type != GgmlType::BF16 {
1113 return Err(format!(
1114 "Step TP projection {name} must preserve checkpoint BF16 bytes, got {:?}",
1115 tensor.ggml_type
1116 )
1117 .into());
1118 }
1119 if tensor.ne.len() != 2 {
1120 return Err(format!(
1121 "Step TP projection {name} must be a 2-D matrix, got shape {:?}",
1122 tensor.ne
1123 )
1124 .into());
1125 }
1126 let matrix = crate::tp::Bf16Matrix {
1127 bytes: tensor.bytes.as_ref(),
1128 in_features: tensor.ne[0] as usize,
1129 out_features: tensor.ne[1] as usize,
1130 };
1131 matrix.validate()?;
1132 if matrix.in_features != expected_in || matrix.out_features != expected_out {
1133 return Err(format!(
1134 "Step TP projection {name} shape {}x{} != registered {expected_out}x{expected_in}",
1135 matrix.out_features, matrix.in_features
1136 )
1137 .into());
1138 }
1139 Ok(if f32_mirror {
1140 runtime.upload_step_bf16_row_parallel_f32_mirror(matrix)?
1141 } else {
1142 runtime.upload_step_bf16_row_parallel(matrix)?
1143 })
1144}
1145
1146fn upload_step_tp_f32_copies(
1147 runtime: &crate::tp::TpE4m3HostBounce,
1148 src: &dyn TensorSource,
1149 name: &str,
1150 expected: usize,
1151) -> Result<Vec<CudaSlice<f32>>, Box<dyn std::error::Error>> {
1152 let tensor = src
1153 .find(name)
1154 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1155 let values = memra_gguf::dequant::dequantize(
1156 tensor.ggml_type,
1157 &tensor.bytes,
1158 tensor.ne.iter().product::<u64>() as usize,
1159 );
1160 if values.len() != expected || values.iter().any(|value| !value.is_finite()) {
1161 return Err(format!(
1162 "Step TP attention {name} has {} finite values, expected {expected}",
1163 values.len()
1164 )
1165 .into());
1166 }
1167 let mut copies = Vec::with_capacity(runtime.devices().len());
1168 for rank in 0..runtime.devices().len() {
1169 let engine = runtime
1170 .rank_engine(rank)
1171 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1172 let _main = engine.gpu.enter_main()?;
1173 copies.push(engine.htod(&values)?);
1174 }
1175 Ok(copies)
1176}
1177
1178fn upload_step_tp_f32_row_shards(
1183 runtime: &crate::tp::TpE4m3HostBounce,
1184 src: &dyn TensorSource,
1185 name: &str,
1186 rows: usize,
1187 cols: usize,
1188) -> Result<Vec<CudaSlice<f32>>, Box<dyn std::error::Error>> {
1189 let tensor = src
1190 .find(name)
1191 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1192 let values = memra_gguf::dequant::dequantize(
1193 tensor.ggml_type,
1194 &tensor.bytes,
1195 tensor.ne.iter().product::<u64>() as usize,
1196 );
1197 let world = runtime.devices().len();
1198 if values.len() != rows * cols || rows % world != 0 || values.iter().any(|v| !v.is_finite()) {
1199 return Err(format!(
1200 "Step TP attention {name} has {} finite values, expected {rows}x{cols} \
1201 (rows divisible by world {world})",
1202 values.len()
1203 )
1204 .into());
1205 }
1206 let local_rows = rows / world;
1207 let mut shards = Vec::with_capacity(world);
1208 for rank in 0..world {
1209 let engine = runtime
1210 .rank_engine(rank)
1211 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1212 let _main = engine.gpu.enter_main()?;
1213 shards
1214 .push(engine.htod(&values[rank * local_rows * cols..(rank + 1) * local_rows * cols])?);
1215 }
1216 Ok(shards)
1217}
1218
1219fn upload_step_tp_bf16_row_shards(
1221 runtime: &crate::tp::TpE4m3HostBounce,
1222 src: &dyn TensorSource,
1223 name: &str,
1224 rows: usize,
1225 cols: usize,
1226) -> Result<Vec<CudaSlice<u8>>, Box<dyn std::error::Error>> {
1227 let tensor = src
1228 .find(name)
1229 .ok_or_else(|| format!("Step TP attention is missing {name}"))?;
1230 if tensor.ggml_type != memra_gguf::GgmlType::BF16 || tensor.bytes.len() != rows * cols * 2 {
1231 return Err(format!(
1232 "Step TP attention {name} is not a bf16 [{rows}, {cols}] tensor ({} bytes, {:?})",
1233 tensor.bytes.len(),
1234 tensor.ggml_type
1235 )
1236 .into());
1237 }
1238 let world = runtime.devices().len();
1239 if rows % world != 0 {
1240 return Err(format!("{name} rows {rows} not divisible by world {world}").into());
1241 }
1242 let local = rows / world * cols * 2;
1243 let mut shards = Vec::with_capacity(world);
1244 for rank in 0..world {
1245 let engine = runtime
1246 .rank_engine(rank)
1247 .ok_or_else(|| format!("Step TP attention has no engine for rank {rank}"))?;
1248 let _main = engine.gpu.enter_main()?;
1249 shards.push(engine.htod_bytes(&tensor.bytes[rank * local..(rank + 1) * local])?);
1250 }
1251 Ok(shards)
1252}
1253
1254#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1255enum StepTpAttentionPlacement {
1256 RankLocalGlobal,
1257 RankLocalSwa,
1258 OwnerSwa,
1259 OwnerTransportFallback,
1260}
1261
1262impl StepTpAttentionPlacement {
1263 fn resolve(native_p2p: bool, window: Option<u32>) -> Self {
1264 match (native_p2p, window.is_some()) {
1265 (true, true) => Self::RankLocalSwa,
1266 (false, true) => Self::OwnerSwa,
1267 (true, false) => Self::RankLocalGlobal,
1268 (false, false) => Self::OwnerTransportFallback,
1269 }
1270 }
1271
1272 fn is_rank_local(self) -> bool {
1273 matches!(self, Self::RankLocalGlobal | Self::RankLocalSwa)
1274 }
1275
1276 fn label(self) -> &'static str {
1277 match self {
1278 Self::RankLocalGlobal => "rank-local-global",
1279 Self::RankLocalSwa => "rank-local-swa-ring",
1280 Self::OwnerSwa => "owner-swa",
1281 Self::OwnerTransportFallback => "owner-transport-fallback",
1282 }
1283 }
1284}
1285
1286fn build_step_tp_qkv(
1287 e: &Engine,
1288 src: &dyn TensorSource,
1289 cfg: &ModelConfig,
1290 layer: usize,
1291 step_runtimes: &mut StepParallelRuntimeRegistry,
1292) -> Result<Option<StepTpQkv>, Box<dyn std::error::Error>> {
1293 let Some(spec) = step_runtimes.tp_spec(layer).cloned() else {
1294 return Ok(None);
1295 };
1296 let contract = crate::parallel::ModelParallelContract::from_model(cfg)?;
1297 if layer >= contract.trunk_layers {
1298 return Err(format!(
1299 "MEMRA_STEP_TP layer {layer} is outside Step trunk layers 0..{}",
1300 contract.trunk_layers
1301 )
1302 .into());
1303 }
1304 let owner = e.ctx().ordinal();
1305 if spec.devices.first().copied() != Some(owner) {
1306 return Err(format!(
1307 "MEMRA_STEP_TP layer {layer} owning PP device {owner} must be the first QKV rank, \
1308 got {:?}",
1309 spec.devices
1310 )
1311 .into());
1312 }
1313 let plan = contract.plan(crate::parallel::TopologyRequest {
1314 pipeline: 1,
1315 tensor: spec.devices.len(),
1316 expert_parallel: spec.devices.len() > 2,
1317 available_devices: spec.devices.len(),
1318 hardware: crate::parallel::HardwareTarget::RtxPro6000Blackwell,
1319 })?;
1320 for rank in 0..spec.devices.len() {
1321 plan.query_head_range(layer, rank).ok_or_else(|| {
1322 format!("Step TP layer {layer} has no query-head range for rank {rank}")
1323 })?;
1324 plan.kv_head_range(layer, rank)
1325 .ok_or_else(|| format!("Step TP layer {layer} has no KV-head range for rank {rank}"))?;
1326 }
1327 let native_p2p = step_runtimes.config.native_p2p;
1328 let ep_device_arithmetic = step_runtimes.config.ep_device_arithmetic;
1329 let f32_mirror = step_runtimes.config.f32_mirror;
1330 if ep_device_arithmetic && (!native_p2p || !matches!(spec.devices.len(), 4 | 8)) {
1331 return Err(
1332 "MEMRA_STEP_EP_DEVICE_ARITHMETIC=1 requires a MEMRA_STEP_TP TP4/TP8 \
1333 expert-owner layer and MEMRA_STEP_TP_NATIVE_P2P=1"
1334 .into(),
1335 );
1336 }
1337 let runtime = step_runtimes.runtime(&spec.devices, native_p2p, ep_device_arithmetic)?;
1338 let p = |suffix: &str| format!("blk.{layer}.{suffix}");
1339 let q = upload_step_bf16_column(
1340 &runtime,
1341 src,
1342 &p("attn_q.weight"),
1343 contract.hidden_size,
1344 contract.query_heads[layer] * contract.head_dim,
1345 f32_mirror,
1346 )?;
1347 let k = upload_step_bf16_column(
1348 &runtime,
1349 src,
1350 &p("attn_k.weight"),
1351 contract.hidden_size,
1352 contract.kv_heads[layer] * contract.head_dim,
1353 f32_mirror,
1354 )?;
1355 let v = upload_step_bf16_column(
1356 &runtime,
1357 src,
1358 &p("attn_v.weight"),
1359 contract.hidden_size,
1360 contract.kv_heads[layer] * contract.head_dim,
1361 f32_mirror,
1362 )?;
1363 let o = upload_step_bf16_row(
1364 &runtime,
1365 src,
1366 &p("attn_output.weight"),
1367 contract.query_heads[layer] * contract.head_dim,
1368 contract.hidden_size,
1369 f32_mirror,
1370 )?;
1371 let geometry = cfg.full_attention_geometry_at(layer as u32);
1372 let attention_placement =
1373 StepTpAttentionPlacement::resolve(runtime.native_p2p(), geometry.window);
1374 let attention = if attention_placement.is_rank_local() {
1375 let decode_input = if ep_device_arithmetic || crate::tp::step_tp_decode_v2_enabled()? {
1380 Some(std::sync::Mutex::new(
1381 runtime.allocate_replicated_device_rows(1, contract.hidden_size)?,
1382 ))
1383 } else {
1384 None
1385 };
1386 let gate_fused =
1389 crate::tp::step_tp_qkv_fused_enabled()? && src.find(&p("attn_gate.weight")).is_some();
1390 let gate_shards = if gate_fused && f32_mirror {
1391 Some(upload_step_tp_f32_row_shards(
1392 &runtime,
1393 src,
1394 &p("attn_gate.weight"),
1395 contract.query_heads[layer],
1396 contract.hidden_size,
1397 )?)
1398 } else {
1399 None
1400 };
1401 let gate_shards_bf16 = if gate_fused && !f32_mirror {
1402 Some(upload_step_tp_bf16_row_shards(
1403 &runtime,
1404 src,
1405 &p("attn_gate.weight"),
1406 contract.query_heads[layer],
1407 contract.hidden_size,
1408 )?)
1409 } else {
1410 None
1411 };
1412 Some(StepTpAttention {
1413 q_norm: upload_step_tp_f32_copies(
1414 &runtime,
1415 src,
1416 &p("attn_q_norm.weight"),
1417 contract.head_dim,
1418 )?,
1419 k_norm: upload_step_tp_f32_copies(
1420 &runtime,
1421 src,
1422 &p("attn_k_norm.weight"),
1423 contract.head_dim,
1424 )?,
1425 decode_input,
1426 gate_shards,
1427 gate_shards_bf16,
1428 })
1429 } else {
1430 None
1431 };
1432 if f32_mirror {
1433 eprintln!(
1434 "[step-tp-qkv] load layer={layer} devices={:?} projections=qkv \
1435 qkv_tensor_parallel=true attention_local=true kv_local=true output_local=true \
1436 transport={} native_p2p={} bf16_residency=f32-mirror \
1437 output=root-readback performance_claim=false",
1438 spec.devices,
1439 runtime.transport_label(),
1440 runtime.native_p2p(),
1441 );
1442 } else {
1443 eprintln!(
1444 "[step-tp-qkv] load layer={layer} devices={:?} projections=qkv \
1445 qkv_tensor_parallel=true attention_local=true kv_local=true output_local=true \
1446 transport={} native_p2p={} output=root-readback performance_claim=false",
1447 spec.devices,
1448 runtime.transport_label(),
1449 runtime.native_p2p(),
1450 );
1451 }
1452 eprintln!(
1453 "[step-tp-attn-plan] load layer={layer} devices={:?} \
1454 qkv_tensor_parallel=true attention_tensor_parallel={} kv_cache_distributed={} \
1455 attention_scope={} transport={} native_p2p={} replicated_decode_input_prepared={} \
1456 performance_claim=false",
1457 spec.devices,
1458 attention_placement.is_rank_local(),
1459 attention_placement.is_rank_local(),
1460 attention_placement.label(),
1461 runtime.transport_label(),
1462 runtime.native_p2p(),
1463 attention
1464 .as_ref()
1465 .is_some_and(|attention| attention.decode_input.is_some()),
1466 );
1467 if f32_mirror {
1468 eprintln!(
1469 "[step-tp-o] load layer={layer} devices={:?} projection=o \
1470 o_tensor_parallel=true attention_local=true kv_local=true \
1471 transport={} native_p2p={} reduction=global-tp8-block-order \
1472 bf16_residency=f32-mirror output=root-readback performance_claim=false",
1473 spec.devices,
1474 runtime.transport_label(),
1475 runtime.native_p2p(),
1476 );
1477 } else {
1478 eprintln!(
1479 "[step-tp-o] load layer={layer} devices={:?} projection=o \
1480 o_tensor_parallel=true attention_local=true kv_local=true \
1481 transport={} native_p2p={} reduction=global-tp8-block-order \
1482 output=root-readback performance_claim=false",
1483 spec.devices,
1484 runtime.transport_label(),
1485 runtime.native_p2p(),
1486 );
1487 }
1488 Ok(Some(StepTpQkv {
1489 runtime,
1490 q,
1491 k,
1492 v,
1493 o,
1494 attention,
1495 devices: spec.devices,
1496 layer,
1497 }))
1498}
1499
1500fn build_dev_exps(
1513 e: &Engine,
1514 resident: &mut ResidentPlan,
1515 il: usize,
1516 gate: &HostExps,
1517 up: &HostExps,
1518 down: &HostExps,
1519) -> Result<Option<crate::hybrid::DevExps>, Box<dyn std::error::Error>> {
1520 if !gate.is_uniform_layout() || !up.is_uniform_layout() || !down.is_uniform_layout() {
1523 return Ok(None);
1524 }
1525 let fp8_host = match (&gate.fp8_blk, &up.fp8_blk, &down.fp8_blk) {
1526 (None, None, None) => None,
1527 (Some(g), Some(u), Some(d)) => Some((g, u, d)),
1528 _ => {
1529 return Err("resident expert projections disagree on block-E4M3 scale carriage".into());
1530 }
1531 };
1532 let scale_bytes = fp8_host
1533 .map(|(g, u, d)| (g.scales.len() + u.scales.len() + d.scales.len()) * size_of::<f32>())
1534 .unwrap_or(0);
1535 let per_layer = gate.bytes.as_bytes().len()
1536 + up.bytes.as_bytes().len()
1537 + down.bytes.as_bytes().len()
1538 + scale_bytes;
1539 if gate.tiers.is_some() {
1540 return Ok(None); }
1542 let fits = resident.should_reside(e, il, per_layer);
1543 if !fits {
1544 return Ok(None);
1545 }
1546 use cudarc::driver::DevicePtr;
1547 let gu_il = std::env::var("MEMRA_MOE_GU_IL").as_deref() == Ok("1")
1548 && gate.out_f == up.out_f
1549 && gate.in_f == up.in_f
1550 && fp8_host.is_none();
1551 let n_expert = gate.n_expert;
1552 let (g, u) = if gu_il {
1553 let (rbg, rbu) = (gate.row_bytes, up.row_bytes);
1555 let n_rows = gate.out_f;
1556 let gb = gate.bytes.as_bytes();
1557 let ub = up.bytes.as_bytes();
1558 let mut il = vec![0u8; n_expert * n_rows * (rbg + rbu)];
1559 for ex in 0..n_expert {
1560 for o in 0..n_rows {
1561 let dst = (ex * n_rows + o) * (rbg + rbu);
1562 let sg = ex * gate.expert_stride + o * rbg;
1563 let su = ex * up.expert_stride + o * rbu;
1564 il[dst..dst + rbg].copy_from_slice(&gb[sg..sg + rbg]);
1565 il[dst + rbg..dst + rbg + rbu].copy_from_slice(&ub[su..su + rbu]);
1566 }
1567 }
1568 let ild = e.htod_bytes_padded(&il, 8)?;
1569 (ild, e.htod_bytes(&[0u8; 16])?)
1572 } else {
1573 (
1574 e.htod_bytes_padded(gate.bytes.as_bytes(), 8)?,
1575 e.htod_bytes_padded(up.bytes.as_bytes(), 8)?,
1576 )
1577 };
1578 let d = e.htod_bytes_padded(down.bytes.as_bytes(), 144)?;
1583 let fp8_blk = match fp8_host {
1584 Some((gate, up, down)) => {
1585 if e.fp8_blk_nan_count(&g)? != 0
1586 || e.fp8_blk_nan_count(&u)? != 0
1587 || e.fp8_blk_nan_count(&d)? != 0
1588 {
1589 return Err("native stacked block-E4M3 expert bank contains NaN codes".into());
1590 }
1591 Some(DevExpertFp8BlockScales {
1592 gate: DevExpertFp8ProjectionScales::upload(e, gate, n_expert)?,
1593 up: DevExpertFp8ProjectionScales::upload(e, up, n_expert)?,
1594 down: DevExpertFp8ProjectionScales::upload(e, down, n_expert)?,
1595 })
1596 }
1597 None => None,
1598 };
1599 let mut host = vec![0u64; 3 * n_expert];
1600 let (pg, pu, pd) = {
1601 let __s_e0 = e.stream();
1602 let (pg, _e0) = g.device_ptr(&__s_e0);
1603 let __s_e1 = e.stream();
1604 let (pu, _e1) = u.device_ptr(&__s_e1);
1605 let __s_e2 = e.stream();
1606 let (pd, _e2) = d.device_ptr(&__s_e2);
1607 (pg as u64, pu as u64, pd as u64)
1608 };
1609 for ex in 0..n_expert {
1610 if gu_il {
1611 let stride = gate.out_f * (gate.row_bytes + up.row_bytes);
1612 host[ex] = pg + (ex * stride) as u64;
1613 host[n_expert + ex] = pg + (ex * stride + gate.row_bytes) as u64;
1614 } else {
1615 host[ex] = pg + (ex * gate.expert_stride) as u64;
1616 host[n_expert + ex] = pu + (ex * up.expert_stride) as u64;
1617 }
1618 host[2 * n_expert + ex] = pd + (ex * down.expert_stride) as u64;
1619 }
1620 if gu_il {
1621 eprintln!("[moe] gate/up dev slab INTERLEAVED (MEMRA_MOE_GU_IL)");
1622 }
1623 let ptr_row = e.htod_u64(&host)?;
1624 Ok(Some(crate::hybrid::DevExps {
1625 gate: g,
1626 up: u,
1627 down: d,
1628 ptr_row,
1629 gu_il,
1630 dev: e.ctx().ordinal(),
1631 fp8_blk,
1632 }))
1633}
1634
1635pub struct FullAttnLayer {
1636 pub wq: GpuTensor,
1637 pub wk: GpuTensor,
1638 pub wv: GpuTensor,
1639 pub wo: GpuTensor,
1640 pub q_norm: GpuTensor,
1641 pub k_norm: GpuTensor,
1642 pub attn_gate: Option<GpuTensor>,
1653 pub step_tp_qkv: Option<StepTpQkv>,
1657}
1658
1659pub struct StepTpQkv {
1660 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
1661 pub q: crate::tp::ResidentBf16ColumnParallel,
1662 pub k: crate::tp::ResidentBf16ColumnParallel,
1663 pub v: crate::tp::ResidentBf16ColumnParallel,
1664 pub o: crate::tp::ResidentStepBf16RowParallel,
1665 pub attention: Option<StepTpAttention>,
1666 pub devices: Vec<usize>,
1667 pub layer: usize,
1668}
1669
1670pub struct StepTpAttention {
1671 pub q_norm: Vec<CudaSlice<f32>>,
1672 pub k_norm: Vec<CudaSlice<f32>>,
1673 pub decode_input: Option<std::sync::Mutex<crate::tp::ResidentReplicatedDeviceRows>>,
1674 pub gate_shards: Option<Vec<CudaSlice<f32>>>,
1677 pub gate_shards_bf16: Option<Vec<CudaSlice<u8>>>,
1679}
1680
1681#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1682pub struct StepTpKvDeviceAdmission {
1683 pub device: usize,
1684 pub bytes: usize,
1685}
1686
1687#[derive(Clone, Copy, Debug)]
1691pub struct MlaGeom {
1692 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, }
1700
1701pub struct MlaAttnLayer {
1705 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,
1715}
1716
1717impl MlaAttnLayer {
1718 pub fn load(
1727 e: &Engine,
1728 src: &dyn TensorSource,
1729 il: u32,
1730 plan: &memra_gguf::model_plan::MlaAttentionPlan,
1731 ) -> Result<Self, Box<dyn std::error::Error>> {
1732 let memra_gguf::model_plan::MlaAttentionPlan::LatentKv {
1733 query_heads,
1734 q_lora_rank,
1735 kv_lora_rank,
1736 qk_head_dim,
1737 rope_head_dim,
1738 value_head_dim,
1739 ..
1740 } = plan
1741 else {
1742 return Err(format!(
1743 "native MLA loader has no compressed-KV implementation for block {il}"
1744 )
1745 .into());
1746 };
1747 let d_nope = qk_head_dim
1748 .checked_sub(*rope_head_dim)
1749 .ok_or("MLA rope head width exceeds total QK head width")?;
1750 let p = |s: &str| format!("blk.{il}.{s}");
1751 let geom = MlaGeom {
1752 n_head: *query_heads as usize,
1753 d_nope: d_nope as usize,
1754 d_rope: *rope_head_dim as usize,
1755 d_v: *value_head_dim as usize,
1756 kv_rank: *kv_lora_rank as usize,
1757 latent_dim: (*kv_lora_rank + *rope_head_dim) as usize,
1758 scale: 1.0 / (*qk_head_dim as f32).sqrt(),
1759 };
1760 let wq_a = load_t(e, src, &p("attn_q_a.weight"))?;
1761 let wq_b = load_t(e, src, &p("attn_q_b.weight"))?;
1762 let wkv_a = load_t(e, src, &p("attn_kv_a_mqa.weight"))?;
1763 let wk_b = load_t(e, src, &p("attn_k_b.weight"))?;
1764 let wv_b = load_t(e, src, &p("attn_v_b.weight"))?;
1765 let wo = load_t(e, src, &p("attn_output.weight"))?;
1766 let n_head = wq_b.out_features() / (geom.d_nope + geom.d_rope);
1768 assert_eq!(
1769 wq_b.out_features(),
1770 n_head * (geom.d_nope + geom.d_rope),
1771 "wq_b out {} not a multiple of qk_head_dim {}",
1772 wq_b.out_features(),
1773 geom.d_nope + geom.d_rope
1774 );
1775 assert_eq!(
1776 wq_a.in_features(),
1777 wkv_a.in_features(),
1778 "q_a/kv_a hidden mismatch"
1779 );
1780 assert_eq!(
1781 wq_b.in_features(),
1782 *q_lora_rank as usize,
1783 "wq_b in != q_lora_rank"
1784 );
1785 assert_eq!(
1786 n_head, geom.n_head,
1787 "MLA checkpoint head count != ModelPlan"
1788 );
1789 assert_eq!(
1790 wkv_a.out_features(),
1791 geom.latent_dim,
1792 "wkv_a out != kv_lora_rank + rope"
1793 );
1794 assert_eq!(
1795 wk_b.ne(),
1796 &[geom.d_nope as u64, geom.kv_rank as u64, n_head as u64],
1797 "attn_k_b must be the TRANSPOSED (nope, kv_rank, head) conversion split"
1798 );
1799 assert_eq!(
1800 wv_b.ne(),
1801 &[geom.kv_rank as u64, geom.d_v as u64, n_head as u64],
1802 "attn_v_b must be the (kv_rank, v, head) conversion split"
1803 );
1804 assert_eq!(
1805 wo.in_features(),
1806 n_head * geom.d_v,
1807 "wo in != n_head * v_head_dim"
1808 );
1809 Ok(MlaAttnLayer {
1810 wq_a,
1811 q_a_norm: load_t(e, src, &p("attn_q_a_norm.weight"))?,
1812 wq_b,
1813 wkv_a,
1814 kv_a_norm: load_t(e, src, &p("attn_kv_a_norm.weight"))?,
1815 wk_b,
1816 wv_b,
1817 wo,
1818 geom,
1819 })
1820 }
1821}
1822
1823#[track_caller]
1827pub(crate) fn mla_forward_unimplemented() -> ! {
1828 panic!(
1829 "Mixer::Mla has no forward arm yet — glm-dsa is loader-only in increment 2; \
1830 the CUDA forward lands in increment 4 (research/mla-bringup-20260801/DESIGN.md §4)"
1831 )
1832}
1833
1834pub struct LinearAttnLayer {
1835 pub geometry: memra_gguf::model_plan::GatedDeltaNetPlan,
1836 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, }
1846
1847pub enum Mixer {
1848 Full(FullAttnLayer),
1849 Linear(LinearAttnLayer),
1850 Mla(MlaAttnLayer),
1852}
1853
1854pub struct MoeWeights {
1861 pub gate_inp: GpuTensor, pub gate_inp_shexp: Option<GpuTensor>, pub exp_probs_b: Option<Vec<f32>>,
1867 pub exp_probs_b_dev: CudaSlice<f32>,
1868 pub active_experts: Option<Vec<bool>>,
1872 pub active_experts_dev: CudaSlice<u8>,
1873 pub gate_exps: HostExps, pub up_exps: HostExps, pub down_exps: HostExps, pub gate_shexp: Option<GpuTensor>,
1877 pub up_shexp: Option<GpuTensor>,
1878 pub down_shexp: Option<GpuTensor>,
1879 pub dev_exps: Option<DevExps>,
1886 pub step_ep: Option<StepEpExps>,
1890 pub step_tp: Option<StepTpExps>,
1894 pub dev_macros: cudarc::driver::CudaSlice<f32>,
1900 pub has_macros: bool,
1901}
1902
1903pub enum StepEpExpertBank {
1905 E4m3(crate::tp::ResidentExpertParallel),
1906 Nvfp4(crate::tp::ResidentNvfp4ExpertParallel),
1907}
1908
1909impl StepEpExpertBank {
1910 pub fn e4m3(&self) -> Result<&crate::tp::ResidentExpertParallel, String> {
1914 match self {
1915 Self::E4m3(bank) => Ok(bank),
1916 Self::Nvfp4(_) => Err(
1917 "Step grouped expert program reached an NVFP4 bank; this path is qualified \
1918 for the E4M3 artifact only"
1919 .to_string(),
1920 ),
1921 }
1922 }
1923}
1924
1925pub struct StepEpExps {
1926 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
1927 pub experts: StepEpExpertBank,
1928 pub devices: Vec<usize>,
1929 pub configured_by_tp: bool,
1930 pub activation_limit: Option<f32>,
1931 pub grouped_decode: Option<std::sync::Mutex<StepEpGroupedDecode>>,
1934}
1935
1936pub struct StepEpGroupedDecode {
1937 pub(crate) projection: crate::tp::PreparedStepGroupedExpertParallelGate,
1938 pub(crate) combine: crate::tp::PreparedPeerWeightedRouteCombine,
1939}
1940
1941#[derive(Default)]
1942pub(crate) struct StepEpGroupedPrefill {
1943 pub(crate) state: Option<StepEpGroupedPrefillState>,
1944}
1945
1946pub(crate) struct StepEpGroupedPrefillState {
1947 pub(crate) devices: Vec<usize>,
1948 pub(crate) grouped: StepEpGroupedDecode,
1949}
1950
1951pub enum StepTpExpertBank {
1953 E4m3(crate::tp::ResidentTensorParallel),
1954 Nvfp4(crate::tp::ResidentNvfp4TensorParallel),
1955}
1956
1957pub struct StepTpExps {
1958 pub runtime: Arc<crate::tp::TpE4m3HostBounce>,
1959 pub experts: StepTpExpertBank,
1960 pub devices: Vec<usize>,
1961 pub activation_limit: Option<f32>,
1964}
1965
1966impl MoeWeights {
1967 #[inline]
1968 pub fn has_uniform_expert_layout(&self) -> bool {
1969 self.gate_exps.is_uniform_layout()
1970 && self.up_exps.is_uniform_layout()
1971 && self.down_exps.is_uniform_layout()
1972 }
1973
1974 #[inline]
1975 pub fn active_count(&self) -> usize {
1976 self.active_experts
1977 .as_ref()
1978 .map(|mask| mask.iter().filter(|&&active| active).count())
1979 .unwrap_or(self.gate_exps.n_expert)
1980 }
1981}
1982
1983pub struct DevExps {
1986 pub gate: CudaSlice<u8>,
1987 pub up: CudaSlice<u8>,
1988 pub down: CudaSlice<u8>,
1989 pub ptr_row: CudaSlice<u64>,
1991 pub dev: usize,
1999 pub gu_il: bool,
2005 pub fp8_blk: Option<DevExpertFp8BlockScales>,
2009}
2010
2011pub struct DevExpertFp8BlockScales {
2012 pub gate: DevExpertFp8ProjectionScales,
2013 pub up: DevExpertFp8ProjectionScales,
2014 pub down: DevExpertFp8ProjectionScales,
2015}
2016
2017pub struct DevExpertFp8ProjectionScales {
2018 pub scales: CudaSlice<f32>,
2019 pub rows: usize,
2020 pub cols: usize,
2021 pub expert_stride: usize,
2022}
2023
2024impl DevExpertFp8ProjectionScales {
2025 fn validate(
2026 host: &crate::model::HostExpertFp8BlockScales,
2027 n_expert: usize,
2028 ) -> Result<(), String> {
2029 if host.expert_stride == 0 {
2030 return Err("block-E4M3 expert scale stride must be nonzero".into());
2031 }
2032 if host.rows * host.cols != host.expert_stride {
2033 return Err(format!(
2034 "block-E4M3 expert scale stride mismatch: {}x{} != {}",
2035 host.rows, host.cols, host.expert_stride
2036 ));
2037 }
2038 let want = n_expert
2039 .checked_mul(host.expert_stride)
2040 .ok_or("block-E4M3 expert scale slab length overflow")?;
2041 if host.scales.len() != want {
2042 return Err(format!(
2043 "block-E4M3 scale slab length mismatch: got {}, want {n_expert}x{}={want}",
2044 host.scales.len(),
2045 host.expert_stride
2046 ));
2047 }
2048 Ok(())
2049 }
2050
2051 fn upload(
2052 e: &Engine,
2053 host: &crate::model::HostExpertFp8BlockScales,
2054 n_expert: usize,
2055 ) -> Result<Self, Box<dyn std::error::Error>> {
2056 Self::validate(host, n_expert)?;
2057 Ok(Self {
2058 scales: e.htod(&host.scales)?,
2059 rows: host.rows,
2060 cols: host.cols,
2061 expert_stride: host.expert_stride,
2062 })
2063 }
2064}
2065
2066pub enum Ffn {
2068 Dense {
2069 ffn_gate: GpuTensor,
2070 ffn_up: GpuTensor,
2071 ffn_down: GpuTensor,
2072 },
2073 Moe(MoeWeights),
2074}
2075
2076pub struct HybridLayer {
2077 pub attn_norm: GpuTensor,
2078 pub post_attn_norm: GpuTensor, pub mixer: Mixer,
2080 pub ffn: Ffn,
2081 pub gemma4: Option<Gemma4LayerBits>,
2082}
2083
2084pub struct Gemma4LayerBits {
2088 pub ffn_norm: GpuTensor, pub post_ffw_norm: GpuTensor, pub moe_bits: Option<Gemma4MoeBits>,
2093 pub layer_scale: f32, pub e4b: Option<Gemma4E4bLayer>,
2096}
2097
2098pub struct Gemma4E4bLayer {
2103 pub inp_gate: GpuTensor, pub proj: GpuTensor, pub post_norm: GpuTensor, pub qkv_cat: Option<GpuTensor>,
2110 pub kv_share: Option<u32>,
2114}
2115
2116pub struct Gemma4E4bModel {
2120 pub tok_tbl_gpu: std::sync::OnceLock<CudaSlice<u8>>,
2123 pub tok_embd_bytes: Vec<u8>,
2124 pub tok_embd_qt: i32,
2125 pub tok_embd_row_bytes: usize,
2126 pub model_proj: GpuTensor, pub proj_norm: GpuTensor, pub n_epl: usize,
2129}
2130
2131pub struct Gemma4MoeBits {
2132 pub post_ffw_norm_1: GpuTensor, pub pre_ffw_norm_2: GpuTensor, pub post_ffw_norm_2: GpuTensor, pub shared_gate: GpuTensor,
2136 pub shared_up: GpuTensor,
2137 pub shared_down: GpuTensor,
2138 pub router_scale_pre: CudaSlice<f32>,
2143 pub per_expert_scale: Vec<f32>, pub per_expert_scale_d: CudaSlice<f32>, }
2146
2147fn load_mtp_head_maybe_nvfp4(
2160 e: &Engine,
2161 src: &dyn TensorSource,
2162 name: &str,
2163) -> Result<Option<GpuTensor>, Box<dyn std::error::Error>> {
2164 if !{
2165 static ENV: std::sync::OnceLock<Option<bool>> = std::sync::OnceLock::new();
2166 crate::step37_door(&ENV, "MEMRA_MTP_HEAD_NVFP4")
2167 } {
2168 return load_opt(e, src, name);
2169 }
2170 let Some(v) = src.find(name) else {
2171 return Ok(None);
2172 };
2173 if !matches!(v.ggml_type, GgmlType::BF16) || v.ne[0] % 64 != 0 {
2174 return load_opt(e, src, name);
2175 }
2176 let vals: Vec<f32> = v
2177 .bytes
2178 .chunks_exact(2)
2179 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
2180 .collect();
2181 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
2182 eprintln!(
2183 "[mtp-head] {name}: BF16 -> NVFP4 ({} MiB, was {} MiB)",
2184 blocks.len() >> 20,
2185 v.bytes.len() >> 20
2186 );
2187 Ok(Some(GpuTensor::from_quant_bytes(
2188 e,
2189 &blocks,
2190 GgmlType::NVFP4,
2191 v.ne[0],
2192 v.ne[1],
2193 1.0,
2194 )?))
2195}
2196
2197pub(crate) fn frspec_trim_own_head_name(n_trunk: usize) -> String {
2202 format!("blk.{n_trunk}.nextn.shared_head_head.weight")
2203}
2204
2205pub struct MtpHead {
2206 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>>,
2220 pub d2t_from_target_head: bool,
2224 pub geom: Option<DraftGeom>,
2230 pub step35: Option<Step35MtpGeom>,
2235}
2236
2237#[derive(Debug, Clone)]
2251pub struct Step35MtpGeom {
2252 pub il: u32,
2254 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>,
2265}
2266
2267impl Step35MtpGeom {
2268 pub fn from_plan(layer: &memra_gguf::model_plan::LayerPlan) -> Result<Self, String> {
2270 use memra_gguf::model_plan::{ActivationPlan, AttentionPlan};
2271
2272 let (attention, window) = match &layer.attention {
2273 AttentionPlan::Full(attention) => (attention, None),
2274 AttentionPlan::SlidingWindow { attention, window } => (attention, Some(*window)),
2275 other => {
2276 return Err(format!(
2277 "MTP block {} has unsupported tuned attention {other:?}",
2278 layer.index
2279 ));
2280 }
2281 };
2282 if attention.output_gate != memra_gguf::config::AttentionGateKind::SeparateHead {
2283 return Err(format!(
2284 "MTP block {} does not declare a separate attention gate",
2285 layer.index
2286 ));
2287 }
2288 let activation = match &layer.mlp {
2289 MlpPlan::Dense(dense) => &dense.activation,
2290 MlpPlan::Moe(moe) => &moe.activation,
2291 };
2292 let clamp_shexp = match activation {
2293 ActivationPlan::SwiGluClamped { limit } if *limit > 0.0 => Some(*limit),
2294 _ => None,
2295 };
2296 Ok(Step35MtpGeom {
2297 il: layer.index,
2298 n_head: attention.query_heads as usize,
2299 n_head_kv: attention.kv_heads as usize,
2300 n_rot: attention.rope.dimensions as usize,
2301 rope_base: attention.rope.base,
2302 swa: window.is_some(),
2303 window: window.unwrap_or(0) as usize,
2304 clamp_shexp,
2305 })
2306 }
2307}
2308
2309pub struct DraftGeom {
2311 pub d_inner: usize, pub n_head: usize, pub n_head_kv: usize,
2314 pub out_up: GpuTensor, }
2316
2317pub fn draft_head_tensor(has: impl Fn(&str) -> bool, n: u32) -> String {
2326 let own = format!("blk.{n}.nextn.shared_head_head.weight");
2327 if has(&own) {
2328 return own;
2329 }
2330 let legacy = format!("blk.{n}.nextn.shared_head.weight");
2333 if has(&legacy) {
2334 return legacy;
2335 }
2336 "output.weight".to_string()
2338}
2339
2340impl MtpHead {
2341 pub fn load_draft(
2348 e: &Engine,
2349 g: &GgufFile,
2350 main_cfg: &ModelConfig,
2351 ) -> Result<Self, Box<dyn std::error::Error>> {
2352 let src = GgufSource(g);
2353 let dcfg = src.try_config().map_err(std::io::Error::other)?;
2354 let draft_plan = match memra_gguf::model_packs::for_config(&dcfg) {
2355 Some(pack) => pack.compile_plan(&dcfg)?,
2356 None => memra_gguf::model_plan::ModelPlan::compile(&dcfg)?,
2357 };
2358 let main_plan = match memra_gguf::model_packs::for_config(main_cfg) {
2359 Some(pack) => pack.compile_plan(main_cfg)?,
2360 None => memra_gguf::model_plan::ModelPlan::compile(main_cfg)?,
2361 };
2362 if dcfg.nextn_predict_layers == 0 {
2367 return Err(format!(
2368 "draft GGUF has no nextn_predict_layers (arch {:?}) — not a NextN/MTP regime \
2369 draft; gemma assistant drafters attach via MEMRA_DRAFT, not '+draft'",
2370 g.arch()
2371 )
2372 .into());
2373 }
2374 let n = dcfg.n_layer - dcfg.nextn_predict_layers;
2375 let draft_block = draft_plan
2376 .mtp_blocks
2377 .iter()
2378 .find(|block| block.layer.index == n)
2379 .ok_or_else(|| format!("draft ModelPlan has no MTP block {n}"))?;
2380 let p = |s: &str| format!("blk.{n}.{s}");
2381
2382 let student = src.has(&p("nextn.out_up.weight"));
2386 assert_eq!(dcfg.n_embd, main_cfg.n_embd, "draft n_embd != model n_embd");
2387 assert_eq!(
2388 dcfg.head_dim_k, main_cfg.head_dim_k,
2389 "draft head_dim != model head_dim"
2390 );
2391 let main_sliding_gated = crate::plan_backend::decode_batch_program(&main_plan)
2398 == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
2399 let draft_sliding_gated = crate::plan_backend::decode_batch_program(&draft_plan)
2400 == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
2401 let step35 = match (main_sliding_gated, draft_sliding_gated) {
2402 (true, true) => {
2403 let g = Step35MtpGeom::from_plan(&draft_block.layer)?;
2404 let out_f = |t: &str| -> Option<usize> {
2406 src.find(&p(t))
2407 .and_then(|v| v.ne.get(1).copied())
2408 .map(|x| x as usize)
2409 };
2410 let hd = dcfg.head_dim_k as usize;
2411 let wq_out =
2412 out_f("attn_q.weight").ok_or("step35 draft block has no attn_q.weight")?;
2413 assert_eq!(
2414 wq_out,
2415 g.n_head * hd,
2416 "step35 draft blk.{n}: attn_q out {wq_out} != n_head({}) * head_dim({hd}) — \
2417 the draft file's head_count array disagrees with its own tensors",
2418 g.n_head
2419 );
2420 let wg_out = out_f("attn_gate.weight")
2423 .ok_or("step35 draft block has no attn_gate.weight (head-wise gate)")?;
2424 assert_eq!(
2425 wg_out, g.n_head,
2426 "step35 draft blk.{n}: attn_gate out {wg_out} != n_head({})",
2427 g.n_head
2428 );
2429 assert_eq!(
2433 g.n_head_kv, main_cfg.n_head_kv as usize,
2434 "step35 draft blk.{n} KV heads {} != trunk n_head_kv {} — the MTP scratch \
2435 rows are sized from the trunk cfg, so a differing draft KV width would \
2436 write past the row",
2437 g.n_head_kv, main_cfg.n_head_kv
2438 );
2439 eprintln!(
2440 "[mtp-draft] step35 MTP geometry blk.{n}: n_head={} n_head_kv={} n_rot={} \
2441 rope_base={:.0} swa={} window={}",
2442 g.n_head, g.n_head_kv, g.n_rot, g.rope_base, g.swa, g.window
2443 );
2444 Some(g)
2445 }
2446 (true, false) => {
2447 return Err(format!(
2448 "MEMRA_MTP_DRAFT operations are incompatible with the model's \
2449 sliding-gated-MoE program (draft arch {:?})",
2450 g.arch()
2451 )
2452 .into());
2453 }
2454 (false, true) => {
2455 return Err(
2456 "MEMRA_MTP_DRAFT requires sliding-gated-MoE operations but the model does not"
2457 .into(),
2458 );
2459 }
2460 (false, false) => None,
2461 };
2462 if step35.is_none() && !student {
2463 assert_eq!(dcfg.n_head, main_cfg.n_head, "draft n_head != model n_head");
2466 assert_eq!(
2467 dcfg.n_head_kv, main_cfg.n_head_kv,
2468 "draft n_head_kv != model n_head_kv"
2469 );
2470 }
2471
2472 let head_name = draft_head_tensor(|t| src.has(t), n);
2499 let head = load_t(e, &src, &head_name)?;
2500 let head_norm = match load_opt(e, &src, &p("nextn.shared_head_norm.weight"))? {
2501 Some(t) => Some(t),
2502 None => load_opt(e, &src, "output_norm.weight")?,
2503 };
2504
2505 let d2t: Option<Vec<u32>> = g.find("d2t").map(|t| {
2507 let bytes = g.tensor_data(t);
2508 match t.ggml_type {
2509 GgmlType::I32 => bytes
2510 .chunks_exact(4)
2511 .map(|c| i32::from_le_bytes(c.try_into().unwrap()) as u32)
2512 .collect(),
2513 GgmlType::I64 => bytes
2514 .chunks_exact(8)
2515 .map(|c| i64::from_le_bytes(c.try_into().unwrap()) as u32)
2516 .collect(),
2517 other => panic!("d2t must be I32/I64, got {other:?}"),
2518 }
2519 });
2520 if let Some(map) = &d2t {
2521 assert_eq!(
2522 map.len(),
2523 head.out_features(),
2524 "d2t len {} != draft head rows {}",
2525 map.len(),
2526 head.out_features()
2527 );
2528 let n_vocab = main_cfg.n_vocab as u64;
2529 assert!(
2530 map.iter().all(|&t| (t as u64) < n_vocab),
2531 "d2t contains token id >= model n_vocab {n_vocab}"
2532 );
2533 }
2534 let eh_proj = load_t(e, &src, &p("nextn.eh_proj.weight"))?;
2535 assert_eq!(
2538 eh_proj.in_features(),
2539 2 * main_cfg.n_embd as usize,
2540 "eh_proj in dim != 2*n_embd"
2541 );
2542 let geom = if student {
2543 let out_up = load_t(e, &src, &p("nextn.out_up.weight"))?;
2544 let d_inner = eh_proj.out_features();
2545 assert_eq!(
2546 out_up.out_features(),
2547 main_cfg.n_embd as usize,
2548 "out_up out dim != n_embd"
2549 );
2550 assert_eq!(
2551 out_up.in_features(),
2552 d_inner,
2553 "out_up in dim != eh_proj out dim (d_inner)"
2554 );
2555 assert!(
2556 dcfg.n_head >= 1 && dcfg.n_head_kv >= 1 && dcfg.n_head % dcfg.n_head_kv == 0,
2557 "student head counts malformed ({}/{})",
2558 dcfg.n_head,
2559 dcfg.n_head_kv
2560 );
2561 Some(DraftGeom {
2562 d_inner,
2563 n_head: dcfg.n_head as usize,
2564 n_head_kv: dcfg.n_head_kv as usize,
2565 out_up,
2566 })
2567 } else {
2568 None
2569 };
2570 let blk_prefix = format!("blk.{n}.");
2574 let head_src = head_name.strip_prefix(&blk_prefix).unwrap_or(&head_name);
2575 eprintln!(
2576 "[mtp-draft] external draft head: blk.{n}, source={}, head_vocab={}{}{}",
2577 head_src,
2578 head.out_features(),
2579 if d2t.is_some() {
2580 " (trimmed, d2t map)"
2581 } else {
2582 " (full)"
2583 },
2584 match &geom {
2585 Some(g) => format!(
2586 " (student d_inner={} heads={}/{})",
2587 g.d_inner, g.n_head, g.n_head_kv
2588 ),
2589 None => String::new(),
2590 }
2591 );
2592
2593 let mut resident = ResidentPlan::unsharded(e, &src, &dcfg);
2594 let mut step_runtimes = StepParallelRuntimeRegistry::default();
2595 Ok(MtpHead {
2596 enorm: load_t(e, &src, &p("nextn.enorm.weight"))?,
2597 hnorm: load_t(e, &src, &p("nextn.hnorm.weight"))?,
2598 eh_proj,
2599 attn_norm: load_t(e, &src, &p("attn_norm.weight"))?,
2600 post_attn_norm: load_opt(e, &src, &p("post_attention_norm.weight"))?
2601 .or(load_opt(e, &src, &p("ffn_norm.weight"))?)
2602 .expect("draft NextN block needs post_attention_norm or ffn_norm"),
2603 mixer: load_mixer_kind(
2604 e,
2605 &src,
2606 &dcfg,
2607 n,
2608 &draft_block.layer.attention,
2609 &mut step_runtimes,
2610 )?,
2611 ffn: load_ffn(
2612 e,
2613 &src,
2614 &dcfg,
2615 &draft_block.layer.mlp,
2616 n,
2617 None,
2618 &mut resident,
2619 &mut step_runtimes,
2620 )?,
2621 shared_head_norm: head_norm,
2622 shared_head_head: Some(head),
2623 d2t,
2624 d2t_from_target_head: false,
2625 geom,
2626 step35,
2627 })
2628 }
2629}
2630
2631pub struct GemmaAux {
2633 pub rope_freqs: Option<Vec<(usize, CudaSlice<f32>)>>,
2636 pub ones: Vec<(usize, CudaSlice<f32>)>,
2639 pub suppress_d: Option<(CudaSlice<i32>, usize)>,
2642 pub e4b: Option<Gemma4E4bModel>,
2644}
2645
2646impl GemmaAux {
2647 pub fn rope_freqs(&self, e: &Engine) -> Option<&CudaSlice<f32>> {
2648 self.rope_freqs.as_ref().map(|copies| {
2649 let dev = e.ctx().ordinal();
2650 &copies
2651 .iter()
2652 .find(|(d, _)| *d == dev)
2653 .unwrap_or_else(|| panic!("gemma4 rope_freqs has no local copy for device {dev}"))
2654 .1
2655 })
2656 }
2657
2658 pub fn ones(&self, e: &Engine) -> &CudaSlice<f32> {
2659 let dev = e.ctx().ordinal();
2660 &self
2661 .ones
2662 .iter()
2663 .find(|(d, _)| *d == dev)
2664 .unwrap_or_else(|| panic!("gemma4 ones has no local copy for device {dev}"))
2665 .1
2666 }
2667}
2668
2669pub struct Step35Aux {
2672 pub rope_freqs: Option<Vec<(usize, CudaSlice<f32>)>>,
2678}
2679
2680impl Step35Aux {
2681 pub fn rope_freqs(&self, e: &Engine) -> Option<&CudaSlice<f32>> {
2682 self.rope_freqs.as_ref().map(|copies| {
2683 let dev = e.ctx().ordinal();
2684 &copies
2685 .iter()
2686 .find(|(d, _)| *d == dev)
2687 .unwrap_or_else(|| panic!("step35 rope_freqs has no local copy for device {dev}"))
2688 .1
2689 })
2690 }
2691}
2692
2693pub struct HybridModel {
2694 pub cfg: ModelConfig,
2695 pub plan: memra_gguf::model_plan::ModelPlan,
2696 pub rewrite_qualifications: Option<memra_gguf::execution_manifest::RewriteQualifications>,
2697 pub embd: EmbedHost,
2698 pub output_norm: GpuTensor,
2699 pub output: GpuTensor,
2700 pub layers: Vec<HybridLayer>,
2701 pub mtp: Option<MtpHead>, pub mtp_extra: Vec<MtpHead>,
2705 pub embd_gpu: std::sync::OnceLock<cudarc::driver::CudaSlice<u8>>,
2708 pub gemma4_aux: Option<GemmaAux>,
2709 pub step35_aux: Option<Step35Aux>,
2711 pub prime_slabs: std::sync::Mutex<
2719 std::collections::HashMap<
2720 usize,
2721 std::sync::Arc<std::sync::Mutex<crate::hybrid_forward::PrimeSlabs>>,
2722 >,
2723 >,
2724 pub(crate) dspark_vgraphs: std::sync::Mutex<Option<crate::spec::DsparkVerifyGraphs>>,
2737 pub(crate) step_grouped_prefill: std::sync::Mutex<StepEpGroupedPrefill>,
2743 pub(crate) step35_token_graph:
2746 std::sync::Mutex<Option<crate::hybrid_forward::Step35TokenGraphState>>,
2747}
2748
2749impl HybridModel {
2750 pub fn install_rewrite_bundle(
2751 &mut self,
2752 bundle: &std::path::Path,
2753 ) -> Result<(), Box<dyn std::error::Error>> {
2754 self.rewrite_qualifications = Some(
2755 memra_gguf::execution_manifest::RewriteQualifications::load(bundle, &self.plan)
2756 .map_err(|error| format!("rewrite qualification: {error}"))?,
2757 );
2758 Ok(())
2759 }
2760
2761 pub fn rewrite_allowed(&self, surface: memra_gguf::execution_manifest::RewriteSurface) -> bool {
2762 self.rewrite_qualifications
2763 .as_ref()
2764 .is_none_or(|qualifications| qualifications.allows(surface))
2765 }
2766
2767 pub fn step_tp_unmaterialized_kv_bytes(
2773 &self,
2774 cache: Option<&crate::cache::Cache>,
2775 capacity: usize,
2776 ) -> Result<Vec<StepTpKvDeviceAdmission>, String> {
2777 if let Some(cache) = cache
2778 && cache.tp_kv.len() < self.layers.len()
2779 {
2780 return Err(format!(
2781 "Step TP admission cache has {} layers, model trunk has {}",
2782 cache.tp_kv.len(),
2783 self.layers.len()
2784 ));
2785 }
2786
2787 let mut by_device: HashMap<usize, usize> = HashMap::new();
2788 for (layer, weights) in self.layers.iter().enumerate() {
2789 let Mixer::Full(attention) = &weights.mixer else {
2790 continue;
2791 };
2792 let Some(tp) = attention
2793 .step_tp_qkv
2794 .as_ref()
2795 .filter(|tp| tp.attention.is_some())
2796 else {
2797 continue;
2798 };
2799 if cache.is_some_and(|cache| cache.tp_kv[layer].is_some()) {
2800 continue;
2801 }
2802 let geometry = self.cfg.full_attention_geometry_at(layer as u32);
2803 let shape = crate::cache::tp_kv_rank_allocation_shape(
2804 geometry.n_head_kv as usize * geometry.head_dim_k as usize,
2805 geometry.n_head_kv as usize * geometry.head_dim_v as usize,
2806 tp.devices.len(),
2807 )?;
2808 let physical_rows = geometry
2809 .window
2810 .map(|window| crate::cache::swa_ring_rows(window as usize, capacity))
2811 .unwrap_or(capacity);
2812 let bytes = shape.allocation_bytes(physical_rows);
2813 for &device in &tp.devices {
2814 let total = by_device.entry(device).or_default();
2815 *total = total.saturating_add(bytes);
2816 }
2817 }
2818
2819 let mut out: Vec<_> = by_device
2820 .into_iter()
2821 .map(|(device, bytes)| StepTpKvDeviceAdmission { device, bytes })
2822 .collect();
2823 out.sort_unstable_by_key(|charge| charge.device);
2824 Ok(out)
2825 }
2826
2827 pub fn step_tp_rank_engine(&self, device: usize) -> Option<&Engine> {
2829 self.layers.iter().find_map(|weights| {
2830 let Mixer::Full(attention) = &weights.mixer else {
2831 return None;
2832 };
2833 let tp = attention.step_tp_qkv.as_ref()?;
2834 let rank = tp
2835 .runtime
2836 .devices()
2837 .iter()
2838 .position(|&rank| rank == device)?;
2839 tp.runtime.rank_engine(rank)
2840 })
2841 }
2842
2843 pub(crate) fn step_tp_runtime_for_layer(
2844 &self,
2845 layer: usize,
2846 ) -> Option<&crate::tp::TpE4m3HostBounce> {
2847 let Mixer::Full(attention) = &self.layers.get(layer)?.mixer else {
2848 return None;
2849 };
2850 let tp = attention.step_tp_qkv.as_ref()?;
2851 tp.attention.as_ref()?;
2852 Some(tp.runtime.as_ref())
2853 }
2854
2855 pub fn decode_batch_program(&self) -> crate::plan_backend::DecodeBatchProgram {
2856 crate::plan_backend::decode_batch_program(&self.plan)
2857 }
2858
2859 pub fn uses_gemma_program(&self) -> bool {
2860 self.decode_batch_program() == crate::plan_backend::DecodeBatchProgram::Gemma
2861 }
2862
2863 pub fn uses_sliding_gated_moe_program(&self) -> bool {
2864 self.decode_batch_program() == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe
2865 }
2866
2867 pub fn has_plan_operation(&self, operation: memra_gguf::model_plan::OperationKind) -> bool {
2868 self.plan.trunk_operations().contains(&operation)
2869 }
2870
2871 pub fn load(e: &Engine, g: &GgufFile) -> Result<Self, Box<dyn std::error::Error>> {
2873 Self::load_from_source(e, &GgufSource(g))
2874 }
2875
2876 pub fn load_without_mtp(e: &Engine, g: &GgufFile) -> Result<Self, Box<dyn std::error::Error>> {
2879 Self::load_from_source_impl(e, &GgufSource(g), false)
2880 }
2881
2882 pub fn load_from_source(
2886 e: &Engine,
2887 src: &dyn TensorSource,
2888 ) -> Result<Self, Box<dyn std::error::Error>> {
2889 Self::load_from_source_impl(e, src, true)
2890 }
2891
2892 pub fn load_from_source_without_mtp(
2894 e: &Engine,
2895 src: &dyn TensorSource,
2896 ) -> Result<Self, Box<dyn std::error::Error>> {
2897 Self::load_from_source_impl(e, src, false)
2898 }
2899
2900 fn load_from_source_impl(
2901 e: &Engine,
2902 src: &dyn TensorSource,
2903 load_mtp: bool,
2904 ) -> Result<Self, Box<dyn std::error::Error>> {
2905 let cfg = src.try_config().map_err(std::io::Error::other)?;
2906 let plan = match memra_gguf::model_packs::for_config(&cfg) {
2907 Some(pack) => pack.compile_plan(&cfg)?,
2908 None => memra_gguf::model_plan::ModelPlan::compile(&cfg)?,
2909 };
2910 let batch_program = crate::plan_backend::decode_batch_program(&plan);
2911 let gemma_program = batch_program == crate::plan_backend::DecodeBatchProgram::Gemma;
2912 let sliding_gated_moe_program =
2913 batch_program == crate::plan_backend::DecodeBatchProgram::SlidingGatedMoe;
2914 if sliding_gated_moe_program {
2919 crate::arm_step37_serving_defaults();
2920 }
2921 cfg.validate_attention_gate_layout()?;
2926 if cfg.sigmoid_router().is_some() {
2933 let host_oracle = std::env::var("MEMRA_SIG_ROUTER").as_deref() == Ok("0");
2934 match crate::sigrouter_contract::verify_host_expf() {
2935 Ok(()) => {}
2936 Err(e) if host_oracle => return Err(e.into()),
2937 Err(e) => eprintln!(
2938 "[sigrouter] WARN: host expf probe mismatch ({e}); device routing is \
2939 unaffected, but host-oracle replay/comparison cells are invalid on this host"
2940 ),
2941 }
2942 }
2943 if std::env::var("MEMRA_DRAFT").is_ok() && std::env::var("MEMRA_MMQ_SK").is_err() {
2952 let force = if cfg.n_embd >= 3500 { 0i8 } else { -1i8 };
2953 crate::MMQ_SK_FORCE.store(force, std::sync::atomic::Ordering::Relaxed);
2954 }
2955 crate::KV_FP8_FORCE.store(0, std::sync::atomic::Ordering::Relaxed);
2959
2960 let n_trunk = (cfg.n_layer - cfg.nextn_predict_layers) as usize;
2965 crate::pp::init_model_transport(e, &cfg, n_trunk)?;
2966 let step_parallel = prepare_step_parallel_load(e, src, &cfg, n_trunk)?;
2967 let embd = EmbedHost::from_source(src, "token_embd.weight");
2968 let e_head = crate::pp::layer_engine(e, n_trunk, n_trunk - 1)?;
2972 let output_norm = load_t(e_head, src, "output_norm.weight")?;
2973 let mut output = if src.has("output.weight") {
2975 load_t(e_head, src, "output.weight")?
2976 } else {
2977 load_t(e_head, src, "token_embd.weight")?
2978 };
2979 let mut resident = ResidentPlan::pp(e, src, &cfg, n_trunk)?;
2980 let mut step_runtimes = StepParallelRuntimeRegistry::with_config(step_parallel);
2981
2982 let gguf: Option<&GgufFile> = src.gguf();
2989 let mut spill: Option<crate::spill::SpillCtx> = if cfg
2992 .moe
2993 .as_ref()
2994 .is_some_and(|m| m.expert_count > 0)
2995 && crate::spill::disk_tier_enabled()
2996 && gguf.is_some()
2997 {
2998 let budget = crate::spill::MemBudget::probe(e)?;
2999 let ctx = crate::spill::SpillCtx::open(gguf.unwrap(), &budget)?;
3000 eprintln!(
3001 "[spill] disk tier ON: free_vram={} MiB free_pinnable_ram={} MiB (MemAvailable*resolved_frac)",
3002 budget.free_vram >> 20,
3003 budget.free_pinnable_ram >> 20
3004 );
3005 Some(ctx)
3006 } else {
3007 None
3008 };
3009
3010 let mut layers = Vec::with_capacity(n_trunk);
3013 for il in 0..n_trunk as u32 {
3014 let p = |s: &str| format!("blk.{il}.{s}");
3015 let layer_plan = plan
3016 .layers
3017 .get(il as usize)
3018 .ok_or_else(|| format!("ModelPlan has no trunk layer {il}"))?;
3019 let e = crate::pp::layer_engine(e, n_trunk, il as usize)?;
3023 layers.push(HybridLayer {
3025 attn_norm: load_t(e, src, &p("attn_norm.weight"))?,
3026 post_attn_norm: load_opt(e, src, &p("post_attention_norm.weight"))?
3027 .or(load_opt(e, src, &p("ffn_norm.weight"))?)
3028 .expect("need post_attention_norm or ffn_norm"),
3029 mixer: {
3030 let g4_shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
3034 let kv_from = n_trunk as u32 - g4_shared;
3035 if g4_shared > 0
3036 && il >= kv_from
3037 && !src.has(&format!("blk.{il}.attn_k.weight"))
3038 {
3039 let g4 = cfg.gemma4.as_ref().unwrap();
3040 let swa = g4.swa_pattern.get(il as usize).copied().unwrap_or(true);
3041 let tgt = kv_from - if swa { 2 } else { 1 };
3042 let tp = |s: &str| format!("blk.{tgt}.{s}");
3043 Mixer::Full(FullAttnLayer {
3044 wq: load_t(e, src, &p("attn_q.weight"))?,
3045 wk: load_t(e, src, &tp("attn_k.weight"))?,
3046 wv: load_t(e, src, &tp("attn_v.weight"))?,
3047 wo: load_t(e, src, &p("attn_output.weight"))?,
3048 q_norm: load_t(e, src, &p("attn_q_norm.weight"))?,
3049 k_norm: load_t(e, src, &tp("attn_k_norm.weight"))?,
3050 attn_gate: None, step_tp_qkv: None,
3052 })
3053 } else {
3054 load_mixer_kind(
3055 e,
3056 src,
3057 &cfg,
3058 il,
3059 &layer_plan.attention,
3060 &mut step_runtimes,
3061 )?
3062 }
3063 },
3064 ffn: load_ffn(
3065 e,
3066 src,
3067 &cfg,
3068 &layer_plan.mlp,
3069 il,
3070 spill.as_mut().map(|c| (gguf.unwrap(), c)),
3071 &mut resident,
3072 &mut step_runtimes,
3073 )?,
3074 gemma4: if gemma_program {
3075 let scalar = |n: &str| -> f32 {
3076 let t = src.find(&p(n)).unwrap_or_else(|| panic!("missing {n}"));
3077 memra_gguf::dequant::dequantize(t.ggml_type, &t.bytes, 1)[0]
3078 };
3079 let vecf = |n: &str| -> Vec<f32> {
3080 let t = src.find(&p(n)).unwrap_or_else(|| panic!("missing {n}"));
3081 memra_gguf::dequant::dequantize(
3082 t.ggml_type,
3083 &t.bytes,
3084 t.ne.iter().product::<u64>() as usize,
3085 )
3086 };
3087 let moe_bits = if src.find(&p("ffn_gate_inp.scale")).is_some() {
3088 Some(crate::hybrid::Gemma4MoeBits {
3089 post_ffw_norm_1: load_t(e, src, &p("post_ffw_norm_1.weight"))?,
3090 pre_ffw_norm_2: load_t(e, src, &p("pre_ffw_norm_2.weight"))?,
3091 post_ffw_norm_2: load_t(e, src, &p("post_ffw_norm_2.weight"))?,
3092 shared_gate: load_t(e, src, &p("ffn_gate.weight"))?,
3093 shared_up: load_t(e, src, &p("ffn_up.weight"))?,
3094 shared_down: load_t(e, src, &p("ffn_down.weight"))?,
3095 router_scale_pre: {
3096 let inv = 1.0 / (cfg.n_embd as f32).sqrt();
3097 let v: Vec<f32> =
3098 vecf("ffn_gate_inp.scale").iter().map(|x| x * inv).collect();
3099 e.htod(&v)?
3100 },
3101 per_expert_scale: vecf("ffn_down_exps.scale"),
3102 per_expert_scale_d: e.htod(&vecf("ffn_down_exps.scale"))?,
3103 })
3104 } else {
3105 None
3106 };
3107 let e4b = if src.has(&p("inp_gate.weight")) {
3109 let g4 = cfg.gemma4.as_ref().unwrap();
3110 let kv_from = n_trunk as u32 - g4.shared_kv_layers;
3111 let kv_share = if g4.shared_kv_layers > 0 && il >= kv_from {
3112 let swa = g4.swa_pattern.get(il as usize).copied().unwrap_or(true);
3113 Some(kv_from - if swa { 2 } else { 1 })
3114 } else {
3115 None
3116 };
3117 Some(crate::hybrid::Gemma4E4bLayer {
3118 inp_gate: load_t(e, src, &p("inp_gate.weight"))?,
3119 proj: load_t(e, src, &p("proj.weight"))?,
3120 post_norm: load_t(e, src, &p("post_norm.weight"))?,
3121 kv_share,
3122 qkv_cat: None, })
3124 } else {
3125 None
3126 };
3127 Some(Gemma4LayerBits {
3128 ffn_norm: load_t(e, src, &p("ffn_norm.weight"))?,
3129 post_ffw_norm: load_t(e, src, &p("post_ffw_norm.weight"))?,
3130 moe_bits,
3131 layer_scale: scalar("layer_output_scale.weight"),
3132 e4b,
3133 })
3134 } else {
3135 None
3136 },
3137 });
3138 }
3139
3140 let external_mtp_requested =
3144 load_mtp && std::env::var("MEMRA_MTP_DRAFT").is_ok_and(|path| !path.is_empty());
3145 let trim_mtp_requested = load_mtp
3146 && !crate::model::full_prec_enabled()
3147 && std::env::var("MEMRA_FRSPEC_TRIM").is_ok_and(|path| !path.is_empty());
3148 let _ = trim_mtp_requested;
3159 let embedded_head_count = if external_mtp_requested {
3160 0
3161 } else {
3162 cfg.nextn_predict_layers
3163 };
3164 let embedded_head_count = match std::env::var("MEMRA_MTP_HEADS")
3169 .ok()
3170 .and_then(|v| v.parse::<u32>().ok())
3171 .filter(|&n| n > 0)
3172 {
3173 Some(cap) if cap < embedded_head_count => {
3174 eprintln!(
3175 "[mtp-chain] MEMRA_MTP_HEADS={cap}: capping the embedded chain from \
3176 {embedded_head_count} heads (measurement knob)"
3177 );
3178 cap
3179 }
3180 _ => embedded_head_count,
3181 };
3182 let mut embedded_mtp = Vec::new();
3183 if load_mtp && embedded_head_count > 0 {
3184 for offset in 0..embedded_head_count {
3185 let n = n_trunk as u32 + offset;
3186 let p = |s: &str| format!("blk.{n}.{s}");
3187 let mtp_plan = plan
3188 .mtp_blocks
3189 .iter()
3190 .find(|block| block.layer.index == n)
3191 .ok_or_else(|| format!("ModelPlan has no embedded MTP block {n}"))?;
3192 if !src.has(&p("nextn.eh_proj.weight")) {
3193 if offset == 0 {
3194 break;
3195 }
3196 return Err(format!(
3197 "embedded MTP chain declares {} heads but blk.{n} has no \
3198 nextn.eh_proj.weight",
3199 cfg.nextn_predict_layers
3200 )
3201 .into());
3202 }
3203 embedded_mtp.push(MtpHead {
3204 enorm: load_t(e, src, &p("nextn.enorm.weight"))?,
3205 hnorm: load_t(e, src, &p("nextn.hnorm.weight"))?,
3206 eh_proj: load_t(e, src, &p("nextn.eh_proj.weight"))?,
3207 attn_norm: load_t(e, src, &p("attn_norm.weight"))?,
3208 post_attn_norm: load_opt(e, src, &p("post_attention_norm.weight"))?
3209 .or(load_opt(e, src, &p("ffn_norm.weight"))?)
3210 .expect("MTP block needs post_attention_norm or ffn_norm"),
3211 mixer: load_mixer_kind(
3212 e,
3213 src,
3214 &cfg,
3215 n,
3216 &mtp_plan.layer.attention,
3217 &mut step_runtimes,
3218 )?,
3219 ffn: load_ffn(
3220 e,
3221 src,
3222 &cfg,
3223 &mtp_plan.layer.mlp,
3224 n,
3225 spill.as_mut().map(|c| (gguf.unwrap(), c)),
3226 &mut resident,
3227 &mut step_runtimes,
3228 )?,
3229 shared_head_norm: load_opt(e, src, &p("nextn.shared_head_norm.weight"))?,
3230 shared_head_head: load_mtp_head_maybe_nvfp4(
3239 e,
3240 src,
3241 &p("nextn.shared_head_head.weight"),
3242 )?
3243 .or(load_opt(e, src, &p("nextn.shared_head.weight"))?),
3244 d2t: None,
3245 d2t_from_target_head: false,
3246 geom: None,
3247 step35: if sliding_gated_moe_program {
3248 Some(Step35MtpGeom::from_plan(&mtp_plan.layer)?)
3249 } else {
3250 None
3251 },
3252 });
3253 }
3254 }
3255 let mut embedded_mtp = embedded_mtp.into_iter();
3256 let mut mtp = embedded_mtp.next();
3257 let mut mtp_extra: Vec<MtpHead> = embedded_mtp.collect();
3258
3259 mtp = if load_mtp {
3263 match std::env::var("MEMRA_MTP_DRAFT") {
3264 Ok(path) if !path.is_empty() => {
3265 eprintln!("[mtp-draft] loading external MTP draft: {path}");
3266 let dg = GgufFile::open(&path)?;
3267 mtp_extra.clear();
3268 Some(MtpHead::load_draft(e, &dg, &cfg)?)
3269 }
3270 _ => mtp,
3271 }
3272 } else {
3273 None
3274 };
3275
3276 let trim_env = if load_mtp {
3287 std::env::var("MEMRA_FRSPEC_TRIM")
3288 } else {
3289 Err(std::env::VarError::NotPresent)
3290 };
3291 if crate::model::full_prec_enabled()
3292 && trim_env.as_deref().map(|p| !p.is_empty()).unwrap_or(false)
3293 {
3294 eprintln!(
3295 "[frspec-trim] DISABLED under MEMRA_FULL_PREC — using the natural full MTP head"
3296 );
3297 }
3298 mtp = match (
3299 if crate::model::full_prec_enabled() {
3300 Err(std::env::VarError::NotPresent)
3301 } else {
3302 trim_env
3303 },
3304 mtp,
3305 ) {
3306 (Ok(path), Some(mut head)) if !path.is_empty() => {
3307 let path = memra_gguf::hf::resolve_arg(&path)
3311 .map_err(|err| format!("MEMRA_FRSPEC_TRIM={path:?}: {err}"))?;
3312 let d2t: Vec<u32> = if path.ends_with(".txt") {
3316 std::fs::read_to_string(&path)?
3317 .lines()
3318 .filter_map(|l| l.trim().parse::<u32>().ok())
3319 .collect()
3320 } else {
3321 let tg = GgufFile::open(&path)?;
3322 let d2t_t = tg
3323 .find("d2t")
3324 .expect("MEMRA_FRSPEC_TRIM file has no d2t tensor");
3325 let d2t_bytes = tg.tensor_data(d2t_t);
3326 match d2t_t.ggml_type {
3327 GgmlType::I32 => d2t_bytes
3328 .chunks_exact(4)
3329 .map(|c| i32::from_le_bytes(c.try_into().unwrap()) as u32)
3330 .collect(),
3331 GgmlType::I64 => d2t_bytes
3332 .chunks_exact(8)
3333 .map(|c| i64::from_le_bytes(c.try_into().unwrap()) as u32)
3334 .collect(),
3335 other => panic!("d2t must be I32/I64, got {other:?}"),
3336 }
3337 };
3338 let own_head_name = frspec_trim_own_head_name(n_trunk);
3347 let own_head = src.find(&own_head_name);
3348 let from_own_head = own_head.is_some();
3349 let v = own_head
3350 .or_else(|| src.find("output.weight"))
3351 .or_else(|| src.find("token_embd.weight"))
3352 .expect("model has no output.weight for FR-Spec trim");
3353 let out_f = v.ne[1] as usize;
3354 let row_bytes = v.bytes.len() / out_f;
3355 assert!(
3356 d2t.iter().all(|&t| (t as usize) < out_f),
3357 "d2t token id >= lm_head rows {out_f}"
3358 );
3359 let mut gathered = Vec::with_capacity(d2t.len() * row_bytes);
3360 for &t in &d2t {
3361 let off = t as usize * row_bytes;
3362 gathered.extend_from_slice(&v.bytes[off..off + row_bytes]);
3363 }
3364 let want_nvfp4 = std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1")
3384 && matches!(v.ggml_type, GgmlType::BF16)
3385 && v.ne[0] % 64 == 0;
3386 if want_nvfp4 {
3387 let in_f = v.ne[0] as usize;
3388 let vals: Vec<f32> = gathered
3389 .chunks_exact(2)
3390 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
3391 .collect();
3392 debug_assert_eq!(vals.len(), d2t.len() * in_f);
3393 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
3394 let trimmed = GpuTensor::from_quant_bytes(
3395 e,
3396 &blocks,
3397 GgmlType::NVFP4,
3398 v.ne[0],
3399 d2t.len() as u64,
3400 1.0,
3401 )?;
3402 eprintln!(
3403 "[frspec-trim] self-trimmed head: {} rows of {} re-quantized BF16 -> NVFP4 \
3404 ({} MiB, was {} MiB)",
3405 d2t.len(),
3406 if from_own_head {
3407 own_head_name.as_str()
3408 } else {
3409 "main output.weight"
3410 },
3411 blocks.len() >> 20,
3412 gathered.len() >> 20,
3413 );
3414 head.shared_head_head = Some(trimmed);
3415 head.d2t = Some(d2t);
3416 head.d2t_from_target_head = !from_own_head;
3417 Some(head)
3418 } else {
3419 let trimmed = match v.ggml_type {
3420 GgmlType::BF16 => GpuTensor::FloatBf16 {
3421 data: e.htod_bytes(&gathered)?,
3422 ne: vec![v.ne[0], d2t.len() as u64],
3423 },
3424 GgmlType::F32 => GpuTensor::Float {
3425 data: e.htod(
3426 &gathered
3427 .chunks_exact(4)
3428 .map(|c| f32::from_le_bytes(c.try_into().unwrap()))
3429 .collect::<Vec<f32>>(),
3430 )?,
3431 ne: vec![v.ne[0], d2t.len() as u64],
3432 },
3433 _ => GpuTensor::from_quant_bytes(
3434 e,
3435 &gathered,
3436 v.ggml_type,
3437 v.ne[0],
3438 d2t.len() as u64,
3439 match src.find("output.scale") {
3441 Some(sv) => f32::from_le_bytes(sv.bytes[..4].try_into().unwrap()),
3442 None => 1.0,
3443 },
3444 )?,
3445 };
3446 eprintln!(
3447 "[frspec-trim] self-trimmed head: {} rows of {} ({:?})",
3448 d2t.len(),
3449 if from_own_head {
3450 own_head_name.as_str()
3451 } else {
3452 "main output.weight"
3453 },
3454 v.ggml_type
3455 );
3456 head.shared_head_head = Some(trimmed);
3457 head.d2t = Some(d2t);
3458 head.d2t_from_target_head = !from_own_head;
3461 Some(head)
3462 }
3463 }
3464 (_, m) => m,
3465 };
3466 if let Some(d2t) = mtp.as_ref().and_then(|head| head.d2t.clone()) {
3479 let want_nvfp4_env = std::env::var("MEMRA_FRSPEC_TRIM_NVFP4").as_deref() == Ok("1");
3480 let mut kept = 0usize;
3481 for (i, head) in mtp_extra.iter_mut().enumerate() {
3482 let name = frspec_trim_own_head_name(n_trunk + 1 + i);
3483 let Some(v) = src.find(&name) else { break };
3484 let out_f = v.ne[1] as usize;
3485 let row_bytes = v.bytes.len() / out_f;
3486 if d2t.iter().any(|&t| (t as usize) >= out_f) {
3487 break;
3488 }
3489 let mut gathered = Vec::with_capacity(d2t.len() * row_bytes);
3490 for &t in &d2t {
3491 let off = t as usize * row_bytes;
3492 gathered.extend_from_slice(&v.bytes[off..off + row_bytes]);
3493 }
3494 let want_nvfp4 =
3495 want_nvfp4_env && matches!(v.ggml_type, GgmlType::BF16) && v.ne[0] % 64 == 0;
3496 let trimmed = if want_nvfp4 {
3497 let vals: Vec<f32> = gathered
3498 .chunks_exact(2)
3499 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
3500 .collect();
3501 let blocks = memra_gguf::nvfp4_repack::f32_to_nvfp4(&vals);
3502 GpuTensor::from_quant_bytes(
3503 e,
3504 &blocks,
3505 GgmlType::NVFP4,
3506 v.ne[0],
3507 d2t.len() as u64,
3508 1.0,
3509 )?
3510 } else {
3511 match v.ggml_type {
3512 GgmlType::BF16 => GpuTensor::FloatBf16 {
3513 data: e.htod_bytes(&gathered)?,
3514 ne: vec![v.ne[0], d2t.len() as u64],
3515 },
3516 GgmlType::F32 => GpuTensor::Float {
3517 data: e.htod(
3518 &gathered
3519 .chunks_exact(4)
3520 .map(|c| f32::from_le_bytes(c.try_into().unwrap()))
3521 .collect::<Vec<f32>>(),
3522 )?,
3523 ne: vec![v.ne[0], d2t.len() as u64],
3524 },
3525 _ => GpuTensor::from_quant_bytes(
3526 e,
3527 &gathered,
3528 v.ggml_type,
3529 v.ne[0],
3530 d2t.len() as u64,
3531 1.0,
3532 )?,
3533 }
3534 };
3535 head.shared_head_head = Some(trimmed);
3536 head.d2t = Some(d2t.clone());
3537 head.d2t_from_target_head = false;
3538 kept += 1;
3539 }
3540 let dropped = mtp_extra.len() - kept;
3541 mtp_extra.truncate(kept);
3542 eprintln!(
3543 "[frspec-trim] per-head trim: {kept} extra chain head(s) gathered from their own \
3544 blocks{}",
3545 if dropped > 0 {
3546 format!(" ({dropped} dropped: no own-head tensor)")
3547 } else {
3548 String::new()
3549 }
3550 );
3551 }
3552 if !mtp_extra.is_empty() {
3553 if plan.draft_source != memra_gguf::model_plan::DraftSourcePlan::Embedded
3554 || plan.mtp_blocks.len() != 1 + mtp_extra.len()
3555 || plan
3556 .mtp_blocks
3557 .iter()
3558 .any(|block| !matches!(block.layer.mlp, MlpPlan::Dense(_)))
3559 || mtp
3560 .iter()
3561 .chain(mtp_extra.iter())
3562 .any(|head| !matches!(head.ffn, Ffn::Dense { .. }))
3563 {
3564 return Err(
3565 "multi-head MTP requires embedded dense canonical blocks and matching loaded heads"
3566 .into(),
3567 );
3568 }
3569 eprintln!(
3570 "[mtp-draft] embedded chain: heads={} blocks={}..={} scratch=per-head",
3571 1 + mtp_extra.len(),
3572 n_trunk,
3573 n_trunk + mtp_extra.len()
3574 );
3575 }
3576
3577 if let Some(ctx) = spill.as_ref() {
3578 eprintln!(
3579 "[spill] experts placed: {} pinned (Tier 1), {} mmap'd from disk (Tier 2, {} MiB)",
3580 ctx.n_pinned,
3581 ctx.n_mmap,
3582 ctx.mmap_bytes >> 20
3583 );
3584 }
3585
3586 if cfg.n_head_kv > 0 && cfg.n_head / cfg.n_head_kv > 8 {
3600 crate::FA_V4_MAX_DEFAULT.store(0, std::sync::atomic::Ordering::Relaxed);
3601 eprintln!(
3602 "[fa] v4 decode family disabled: gqa {} > fa_v4_smem capacity 8 (v3 lane serves)",
3603 cfg.n_head / cfg.n_head_kv
3604 );
3605 }
3606
3607 if gemma_program {
3608 crate::FA_VEC_MIN_DEFAULT.store(1, std::sync::atomic::Ordering::Relaxed);
3610 let real_moe = plan
3613 .trunk_operations()
3614 .contains(&memra_gguf::model_plan::OperationKind::MoeMlp);
3615 crate::FA_SPW_DEFAULT.store(
3616 if real_moe { 32 } else { 64 },
3617 std::sync::atomic::Ordering::Relaxed,
3618 );
3619 crate::FA_SP512_DEFAULT.store(
3621 if real_moe { 16 } else { 32 },
3622 std::sync::atomic::Ordering::Relaxed,
3623 );
3624 crate::FUSED_MR1_DEFAULT.store(!real_moe, std::sync::atomic::Ordering::Relaxed);
3634 crate::RMS_BLOCK_DEFAULT.store(1024, std::sync::atomic::Ordering::Relaxed);
3636 crate::FA_SP_GEMMA.store(true, std::sync::atomic::Ordering::Relaxed);
3638 }
3642 let force_embd_gpu = gemma_program;
3645 let gemma4_aux = if gemma_program {
3646 let rope_freqs = match src.find("rope_freqs.weight") {
3647 Some(t) => {
3648 let host = memra_gguf::dequant::dequantize(
3649 t.ggml_type,
3650 &t.bytes,
3651 t.ne.iter().product::<u64>() as usize,
3652 );
3653 let mut copies = Vec::new();
3654 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
3655 for s in 0..fence.len() - 1 {
3656 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
3657 let dev = owner.ctx().ordinal();
3658 if copies.iter().all(|(d, _)| *d != dev) {
3659 copies.push((dev, owner.htod(&host)?));
3660 }
3661 }
3662 } else {
3663 copies.push((e.ctx().ordinal(), e.htod(&host)?));
3664 }
3665 Some(copies)
3666 }
3667 None => {
3675 let g4 = cfg.gemma4.as_ref().unwrap();
3676 let n = (g4.rope_dims_global / 2) as usize;
3677 let keep =
3678 ((n as f32) * g4.partial_rotary_global.clamp(0.0, 1.0)).round() as usize;
3679 let host: Vec<f32> = (0..n)
3680 .map(|i| if i < keep { 1.0 } else { 1.0e30 })
3681 .collect();
3682 eprintln!(
3683 "[gemma4] rope_freqs.weight synthesized ({n} factors, first {keep} \
3684 rotate; source ships none — native checkpoint)"
3685 );
3686 let mut copies = Vec::new();
3687 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
3688 for s in 0..fence.len() - 1 {
3689 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
3690 let dev = owner.ctx().ordinal();
3691 if copies.iter().all(|(d, _)| *d != dev) {
3692 copies.push((dev, owner.htod(&host)?));
3693 }
3694 }
3695 } else {
3696 copies.push((e.ctx().ordinal(), e.htod(&host)?));
3697 }
3698 Some(copies)
3699 }
3700 };
3701 let e4b = match src.find("per_layer_token_embd.weight") {
3703 Some(t) => {
3704 let n_epl = cfg
3705 .gemma4
3706 .as_ref()
3707 .map(|g| g.n_embd_per_layer as usize)
3708 .unwrap_or(0);
3709 let row = t.ne[0] as usize; let row_bytes = t.bytes.len() / (t.ne[1] as usize);
3711 eprintln!(
3712 "[gemma4-e4b] per-layer-embed model detected (n_epl={n_epl}, row {row}) — \
3713 first-light forward (eager decode + prime); dc/graph/spec unwired \
3714 (HANDOVER-E4B.md)"
3715 );
3716 Some(crate::hybrid::Gemma4E4bModel {
3717 tok_tbl_gpu: std::sync::OnceLock::new(),
3718 tok_embd_bytes: t.bytes.to_vec(),
3719 tok_embd_qt: match t.ggml_type {
3720 memra_gguf::GgmlType::Q6_K => crate::QT_Q6_K,
3721 memra_gguf::GgmlType::Q8_0 => crate::QT_Q8_0,
3722 other => panic!("e4b per-layer tok embd: unhandled dtype {other:?}"),
3723 },
3724 tok_embd_row_bytes: row_bytes,
3725 model_proj: load_t(e, src, "per_layer_model_proj.weight")?,
3726 proj_norm: load_t(e, src, "per_layer_proj_norm.weight")?,
3727 n_epl,
3728 })
3729 }
3730 None => None,
3731 };
3732 let suppress_d = {
3733 let sup = &cfg.gemma4.as_ref().unwrap().suppress_tokens;
3734 if sup.is_empty() {
3735 None
3736 } else {
3737 let ids: Vec<i32> = sup.iter().map(|&x| x as i32).collect();
3738 eprintln!(
3739 "[gemma4] suppress_tokens: {} ids masked at sampling",
3740 ids.len()
3741 );
3742 Some((e.htod_i32(&ids)?, ids.len()))
3743 }
3744 };
3745 let ones_host = [1.0f32; 512];
3746 let mut ones = Vec::new();
3747 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
3748 for s in 0..fence.len() - 1 {
3749 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
3750 let dev = owner.ctx().ordinal();
3751 if ones.iter().all(|(d, _)| *d != dev) {
3752 ones.push((dev, owner.htod(&ones_host)?));
3753 }
3754 }
3755 } else {
3756 ones.push((e.ctx().ordinal(), e.htod(&ones_host)?));
3757 }
3758 Some(GemmaAux {
3759 rope_freqs,
3760 ones,
3761 suppress_d,
3762 e4b,
3763 })
3764 } else {
3765 None
3766 };
3767 let step35_aux = if sliding_gated_moe_program {
3771 let rope_freqs = match src.find("rope_freqs.weight") {
3772 Some(t) => {
3773 let host = memra_gguf::dequant::dequantize(
3774 t.ggml_type,
3775 &t.bytes,
3776 t.ne.iter().product::<u64>() as usize,
3777 );
3778 let mut copies = Vec::new();
3779 if let Some(fence) = crate::pp::pp_cuts(n_trunk) {
3780 for s in 0..fence.len() - 1 {
3781 let owner = crate::pp::layer_engine(e, n_trunk, fence[s])?;
3782 let dev = owner.ctx().ordinal();
3783 if copies.iter().all(|(d, _)| *d != dev) {
3784 copies.push((dev, owner.htod(&host)?));
3785 }
3786 }
3787 } else {
3788 copies.push((e.ctx().ordinal(), e.htod(&host)?));
3789 }
3790 Some(copies)
3791 }
3792 None => None,
3793 };
3794 Some(Step35Aux { rope_freqs })
3795 } else {
3796 None
3797 };
3798 let mut layers = layers;
3799 {
3806 let q8rp_on = match std::env::var("MEMRA_Q8RP").as_deref() {
3807 Ok("0") => false,
3808 Ok(_) => true,
3809 Err(_) => {
3817 cfg!(memra_hopper_mma) || {
3818 let q8b = |w: &crate::model::GpuTensor| -> usize {
3819 match w {
3820 crate::model::GpuTensor::Quant {
3821 bytes,
3822 qtype,
3823 row_bytes,
3824 ne,
3825 rp4: None,
3826 ..
3827 } if *qtype == crate::QT_Q8_0
3828 && ne.len() == 2
3829 && (ne[0] as usize) % 32 == 0
3830 && *row_bytes == (ne[0] as usize / 32) * 34 =>
3831 {
3832 bytes.len()
3833 }
3834 _ => 0,
3835 }
3836 };
3837 let mut need = q8b(&output);
3838 for layer in layers.iter() {
3839 match &layer.mixer {
3840 Mixer::Full(fa) => {
3841 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
3842 need += q8b(w);
3843 }
3844 }
3845 Mixer::Linear(la) => {
3846 for w in [
3847 &la.wqkv,
3848 &la.wqkv_gate,
3849 &la.ssm_beta,
3850 &la.ssm_alpha,
3851 &la.ssm_out,
3852 ] {
3853 need += q8b(w);
3854 }
3855 }
3856 Mixer::Mla(_) => {}
3857 }
3858 if let Ffn::Dense {
3859 ffn_gate,
3860 ffn_up,
3861 ffn_down,
3862 } = &layer.ffn
3863 {
3864 for w in [ffn_gate, ffn_up, ffn_down] {
3865 need += q8b(w);
3866 }
3867 }
3868 }
3869 need > 0
3870 && e.ctx()
3871 .mem_get_info()
3872 .map(|(free, _)| free >= need + (8usize << 30))
3873 .unwrap_or(false)
3874 }
3875 }
3876 };
3877 let kqrp_on = crate::Engine::kqrp_enabled() || {
3887 std::env::var("MEMRA_KQRP").is_err() && {
3888 let kqb = |w: &crate::model::GpuTensor| -> usize {
3889 match w {
3890 crate::model::GpuTensor::Quant {
3891 bytes,
3892 qtype,
3893 row_bytes,
3894 ne,
3895 rp4: None,
3896 ..
3897 } if ne.len() == 2 && (ne[0] as usize) % 256 == 0 => {
3898 let sb = if *qtype == crate::QT_Q4_K {
3899 144
3900 } else if *qtype == crate::QT_Q6_K {
3901 210
3902 } else {
3903 return 0;
3904 };
3905 if *row_bytes == (ne[0] as usize / 256) * sb {
3906 bytes.len()
3907 } else {
3908 0
3909 }
3910 }
3911 _ => 0,
3912 }
3913 };
3914 let mut need = kqb(&output);
3915 for layer in layers.iter() {
3916 if let Mixer::Full(fa) = &layer.mixer {
3917 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
3918 need += kqb(w);
3919 }
3920 }
3921 if let Ffn::Dense {
3922 ffn_gate,
3923 ffn_up,
3924 ffn_down,
3925 } = &layer.ffn
3926 {
3927 for w in [ffn_gate, ffn_up, ffn_down] {
3928 need += kqb(w);
3929 }
3930 }
3931 }
3932 need > 0
3933 && e.ctx()
3934 .mem_get_info()
3935 .map(|(free, _)| free >= need + (8usize << 30))
3936 .unwrap_or(false)
3937 }
3938 };
3939 if q8rp_on || kqrp_on {
3940 let f16_model_ok = gemma_program
3947 || plan
3948 .trunk_operations()
3949 .contains(&memra_gguf::model_plan::OperationKind::MoeMlp)
3950 || std::env::var("MEMRA_PP_F16").as_deref() == Ok("1");
3951 let mut nmir = 0usize;
3952 let mut mir = |e_ref: &crate::Engine,
3956 w: &mut crate::model::GpuTensor|
3957 -> Result<(), Box<dyn std::error::Error>> {
3958 let before = matches!(w, crate::model::GpuTensor::Quant { rp4: Some(_), .. });
3959 if q8rp_on {
3960 e_ref.build_q8_rp4(w)?;
3961 }
3962 if kqrp_on {
3963 e_ref.build_q4k_rp4(w)?;
3964 e_ref.build_q6k_rp4(w)?;
3965 }
3966 let q6k = matches!(w, crate::model::GpuTensor::Quant { qtype, .. }
3971 if *qtype == crate::QT_Q6_K);
3972 if q8rp_on && crate::f16_ffi::pp_f16_enabled() && (f16_model_ok || q6k) {
3973 e_ref.build_q8_f16(w)?;
3974 }
3975 if !before && matches!(w, crate::model::GpuTensor::Quant { rp4: Some(_), .. }) {
3976 nmir += 1;
3977 }
3978 Ok(())
3979 };
3980 for (il, layer) in layers.iter_mut().enumerate() {
3981 let el = crate::pp::layer_engine(e, n_trunk, il)?;
3982 match &mut layer.mixer {
3983 Mixer::Full(fa) => {
3984 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
3985 mir(el, w)?;
3986 }
3987 }
3988 Mixer::Linear(la) => {
3989 for w in [
3990 &mut la.wqkv,
3991 &mut la.wqkv_gate,
3992 &mut la.ssm_beta,
3993 &mut la.ssm_alpha,
3994 &mut la.ssm_out,
3995 ] {
3996 mir(el, w)?;
3997 }
3998 }
3999 Mixer::Mla(_) => {}
4002 }
4003 if let Ffn::Dense {
4004 ffn_gate,
4005 ffn_up,
4006 ffn_down,
4007 } = &mut layer.ffn
4008 {
4009 for w in [ffn_gate, ffn_up, ffn_down] {
4010 mir(el, w)?;
4011 }
4012 }
4013 }
4014 mir(e_head, &mut output)?;
4015 if nmir > 0 {
4016 eprintln!("[q8rp] split-plane decode mirrors built: {nmir} tensors");
4017 }
4018 if q8rp_on && crate::f16_ffi::pp_f16_enabled() {
4033 for (want, tag) in [(crate::QT_Q4_K, "q4kf16"), (crate::QT_Q5_K, "q5kf16")] {
4034 let (mut n4, mut b4) = (0usize, 0usize);
4035 let mut mirk =
4036 |e_ref: &crate::Engine,
4037 w: &mut crate::model::GpuTensor|
4038 -> Result<(), Box<dyn std::error::Error>> {
4039 if matches!(w, crate::model::GpuTensor::Quant { qtype, f16: None, .. }
4040 if *qtype == want)
4041 {
4042 e_ref.build_q8_f16(w)?;
4043 if let crate::model::GpuTensor::Quant { f16: Some(m), .. } = w {
4044 n4 += 1;
4045 b4 += m.len();
4046 }
4047 }
4048 Ok(())
4049 };
4050 for (il, layer) in layers.iter_mut().enumerate() {
4051 let el = crate::pp::layer_engine(e, n_trunk, il)?;
4052 match &mut layer.mixer {
4053 Mixer::Full(fa) => {
4054 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
4055 mirk(el, w)?;
4056 }
4057 }
4058 Mixer::Linear(la) => {
4059 for w in [
4060 &mut la.wqkv,
4061 &mut la.wqkv_gate,
4062 &mut la.ssm_beta,
4063 &mut la.ssm_alpha,
4064 &mut la.ssm_out,
4065 ] {
4066 mirk(el, w)?;
4067 }
4068 }
4069 Mixer::Mla(_) => {} }
4071 if let Ffn::Dense {
4072 ffn_gate,
4073 ffn_up,
4074 ffn_down,
4075 } = &mut layer.ffn
4076 {
4077 for w in [ffn_gate, ffn_up, ffn_down] {
4078 mirk(el, w)?;
4079 }
4080 }
4081 }
4082 mirk(e_head, &mut output)?;
4083 if n4 > 0 {
4084 eprintln!(
4085 "[{tag}] prefill fp16 mirrors built: {n4} tensors \
4086 ({} MB)",
4087 b4 >> 20
4088 );
4089 }
4090 }
4091 }
4092 }
4093 }
4094 if gemma_program && crate::Engine::q4rp_enabled() {
4101 let mut nmir = 0usize;
4102 for (il, layer) in layers.iter_mut().enumerate() {
4103 let e = crate::pp::layer_engine(e, n_trunk, il)?;
4105 let is_moe26 = layer.gemma4.as_ref().is_some_and(|g| g.moe_bits.is_some());
4114 let is_e4b = layer.gemma4.as_ref().is_some_and(|g| g.e4b.is_some());
4115 if !(is_moe26 || is_e4b) {
4116 continue;
4117 }
4118 if let Mixer::Full(fa) = &mut layer.mixer {
4119 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
4120 e.build_q4_rp4(w)?;
4121 nmir += 1;
4122 }
4123 }
4124 if is_e4b {
4125 let own_kv = layer
4127 .gemma4
4128 .as_ref()
4129 .unwrap()
4130 .e4b
4131 .as_ref()
4132 .is_some_and(|e4| e4.kv_share.is_none());
4133 if own_kv {
4134 if let Mixer::Full(fa) = &layer.mixer {
4135 if let Some(mut cat) = e.build_q4_out_concat3(&fa.wq, &fa.wk, &fa.wv)? {
4136 e.build_q4_rp4(&mut cat)?;
4137 nmir += 1;
4138 layer.gemma4.as_mut().unwrap().e4b.as_mut().unwrap().qkv_cat =
4139 Some(cat);
4140 }
4141 }
4142 }
4143 if let Ffn::Dense {
4144 ffn_gate,
4145 ffn_up,
4146 ffn_down,
4147 } = &mut layer.ffn
4148 {
4149 for w in [ffn_gate, ffn_up, ffn_down] {
4150 e.build_q4_rp4(w)?;
4151 nmir += 1;
4152 }
4153 }
4154 let e4 = layer.gemma4.as_mut().unwrap().e4b.as_mut().unwrap();
4155 for w in [&mut e4.inp_gate, &mut e4.proj] {
4156 e.build_q4_rp4(w)?;
4157 nmir += 1;
4158 }
4159 }
4160 if let Some(mb) = layer.gemma4.as_mut().unwrap().moe_bits.as_mut() {
4161 for w in [&mut mb.shared_gate, &mut mb.shared_up, &mut mb.shared_down] {
4162 e.build_q4_rp4(w)?;
4163 nmir += 1;
4164 }
4165 }
4166 }
4167 if nmir > 0 {
4168 eprintln!("[q4rp] split-plane decode mirrors built: {nmir} trunk tensors");
4169 }
4170 let fast_on = std::env::var("MEMRA_FAST").as_deref() != Ok("0");
4177 if fast_on {
4178 let mut nswap = 0usize;
4179 let mut nf16 = 0usize;
4180 let q4f16_model_ok = matches!(cfg.n_embd, 3840 | 5376); if let Ok(v) = std::env::var("MEMRA_Q4F16") {
4199 if v != "0" && v != "1" {
4200 return Err(format!(
4201 "MEMRA_Q4F16={v} is not 0 or 1 — this env selects the prefill \
4202 ARITHMETIC (fp16 mirrors vs int8 MMQ) and must never be guessed"
4203 )
4204 .into());
4205 }
4206 }
4207 let f16_need = {
4208 let f16b = |w: &crate::model::GpuTensor| -> usize {
4209 match w {
4210 crate::model::GpuTensor::Quant {
4211 qtype,
4212 ne,
4213 f16: None,
4214 ..
4215 } if ne.len() == 2
4216 && matches!(
4217 *qtype,
4218 crate::QT_Q8_0
4219 | crate::QT_Q4_0
4220 | crate::QT_Q6_K
4221 | crate::QT_Q4_K
4222 | crate::QT_Q5_K
4223 ) =>
4224 {
4225 (ne[0] as usize) * (ne[1] as usize) * 2
4226 }
4227 _ => 0,
4228 }
4229 };
4230 let mut need = 0usize;
4231 for layer in layers.iter() {
4232 if layer.gemma4.as_ref().is_none_or(|g| g.moe_bits.is_some()) {
4233 continue;
4234 }
4235 if let Mixer::Full(fa) = &layer.mixer {
4236 for w in [&fa.wq, &fa.wk, &fa.wv, &fa.wo] {
4237 need += f16b(w);
4238 }
4239 }
4240 if let Ffn::Dense {
4241 ffn_gate,
4242 ffn_up,
4243 ffn_down,
4244 } = &layer.ffn
4245 {
4246 for w in [ffn_gate, ffn_up, ffn_down] {
4247 need += f16b(w);
4248 }
4249 }
4250 }
4251 need
4252 };
4253 let f16_free = e.ctx().mem_get_info().map(|(free, _)| free).unwrap_or(0);
4254 let f16_auto = q4f16_model_ok
4255 && std::env::var("MEMRA_Q4F16").is_err()
4256 && crate::f16_ffi::pp_f16_capacity_ok(f16_free, f16_need);
4257 let (f16_on, f16_why) = match std::env::var("MEMRA_Q4F16").as_deref() {
4263 Ok("1") => (true, "env MEMRA_Q4F16=1"),
4264 Ok("0") => (false, "env MEMRA_Q4F16=0"),
4265 _ if crate::f16_ffi::pp_f16_enabled() && q4f16_model_ok => {
4266 (true, "env MEMRA_PP_F16")
4267 }
4268 _ if f16_auto => (true, "capacity-keyed auto (UNPINNED)"),
4269 _ if !q4f16_model_ok => (false, "model geometry not eligible"),
4270 _ => (false, "capacity-keyed auto REFUSED (UNPINNED)"),
4271 };
4272 eprintln!(
4279 "[q4f16] prefill program = {} (reason: {}); free {} MiB, mirror mass {} MiB, \
4280 capacity threshold {} MiB (mass + 8192 headroom) — SELECTS PREFILL ARITHMETIC",
4281 if f16_on {
4282 "FP16 MIRRORS"
4283 } else {
4284 "INT8 MMQ (no f16 mirrors)"
4285 },
4286 f16_why,
4287 f16_free >> 20,
4288 f16_need >> 20,
4289 (f16_need + (8usize << 30)) >> 20,
4290 );
4291 for (il, layer) in layers.iter_mut().enumerate() {
4292 let e = crate::pp::layer_engine(e, n_trunk, il)?;
4294 let dense_gemma = layer.gemma4.as_ref().is_some_and(|g| g.moe_bits.is_none());
4295 if !dense_gemma {
4296 continue;
4297 }
4298 if let Mixer::Full(fa) = &mut layer.mixer {
4299 for w in [&mut fa.wq, &mut fa.wk, &mut fa.wv, &mut fa.wo] {
4300 if f16_on {
4301 e.build_q8_f16(w)?;
4302 if matches!(w, crate::model::GpuTensor::Quant { f16: Some(_), .. })
4303 {
4304 nf16 += 1;
4305 }
4306 }
4307 if e.build_q4_rp_swap(w)? {
4308 nswap += 1;
4309 }
4310 }
4311 }
4312 if let Ffn::Dense {
4313 ffn_gate,
4314 ffn_up,
4315 ffn_down,
4316 } = &mut layer.ffn
4317 {
4318 for w in [ffn_gate, ffn_up, ffn_down] {
4319 if f16_on {
4320 e.build_q8_f16(w)?;
4321 if matches!(w, crate::model::GpuTensor::Quant { f16: Some(_), .. })
4322 {
4323 nf16 += 1;
4324 }
4325 }
4326 if e.build_q4_rp_swap(w)? {
4327 nswap += 1;
4328 }
4329 }
4330 }
4331 }
4332 if nswap > 0 {
4333 eprintln!("[q4rp] split-plane IN-PLACE swap: {nswap} dense trunk tensors");
4334 }
4335 if nf16 > 0 {
4336 eprintln!("[q4f16] prefill fp16 mirrors built: {nf16} dense trunk tensors");
4337 }
4338 }
4339 }
4340 let model = HybridModel {
4341 cfg,
4342 plan,
4343 rewrite_qualifications: None,
4344 embd,
4345 output_norm,
4346 output,
4347 layers,
4348 mtp,
4349 mtp_extra,
4350 embd_gpu: std::sync::OnceLock::new(),
4351 gemma4_aux,
4352 step35_aux,
4353 prime_slabs: std::sync::Mutex::new(std::collections::HashMap::new()),
4354 dspark_vgraphs: std::sync::Mutex::new(None),
4355 step_grouped_prefill: std::sync::Mutex::new(StepEpGroupedPrefill::default()),
4356 step35_token_graph: std::sync::Mutex::new(None),
4357 };
4358 e.configure_moe_cache_layout(model.moe_cache_block_sizes());
4359 if force_embd_gpu {
4360 let _ = model
4361 .embd_gpu
4362 .get_or_init(|| e.upload_u8(&model.embd.raw).expect("embed table upload"));
4363 }
4364 crate::pp::sync_stages_after_load(e, n_trunk)?;
4370 Ok(model)
4371 }
4372
4373 pub fn ensure_embed_resident(&self, e: &Engine) -> Result<(), Box<dyn std::error::Error>> {
4383 if std::env::var("MEMRA_EMBED_DEV").as_deref() == Ok("0") {
4384 return Ok(());
4385 }
4386 if self.embd_gpu.get().is_none() {
4387 let buf = e.upload_u8(&self.embd.raw)?;
4388 let _ = self.embd_gpu.set(buf); }
4390 Ok(())
4391 }
4392
4393 pub fn embed(
4394 &self,
4395 e: &Engine,
4396 tokens: &[u32],
4397 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
4398 let n_embd = self.cfg.n_embd as usize;
4399 if std::env::var("MEMRA_EMBED_DEV").as_deref() != Ok("0") {
4405 let tbl = self
4406 .embd_gpu
4407 .get_or_init(|| e.upload_u8(&self.embd.raw).expect("embed table upload"));
4408 let tok_d = e.htod_u32_v(tokens)?;
4409 let (qt, rb) = self.embd.qt_and_row_bytes(n_embd);
4410 return e.embed_gather_device_td(tbl, &tok_d, tokens.len(), n_embd, qt, rb);
4411 }
4412 let x = self.embd.gather(n_embd, tokens);
4413 Ok(e.htod(&x)?)
4414 }
4415}
4416
4417#[cfg(test)]
4418mod step_expert_selection_tests {
4419 use super::{
4420 StepExpertArtifact, StepExpertLayout, StepParallelLoadConfig, StepParallelRuntimeRegistry,
4421 StepTpAttentionPlacement, select_step_expert_layout,
4422 };
4423 use crate::tp::StepEpLayerSpec;
4424
4425 fn spec(layer: usize, ranks: usize) -> StepEpLayerSpec {
4426 StepEpLayerSpec {
4427 layer,
4428 devices: (0..ranks).collect(),
4429 }
4430 }
4431
4432 #[test]
4433 fn tp2_keeps_projection_sharded_experts() {
4434 let selection = select_step_expert_layout(24, &[], &[spec(24, 2)])
4435 .unwrap()
4436 .unwrap();
4437 assert_eq!(selection.layout, StepExpertLayout::TensorParallel);
4438 assert!(selection.configured_by_tp);
4439 }
4440
4441 #[test]
4442 fn tp4_and_tp8_use_expert_ownership_without_a_second_flag() {
4443 for ranks in [4, 8] {
4444 let selection = select_step_expert_layout(24, &[], &[spec(24, ranks)])
4445 .unwrap()
4446 .unwrap();
4447 assert_eq!(selection.layout, StepExpertLayout::ExpertParallel);
4448 assert!(selection.configured_by_tp);
4449 assert_eq!(selection.spec.devices.len(), ranks);
4450 }
4451 }
4452
4453 #[test]
4454 fn explicit_ep_remains_expert_parallel() {
4455 let selection = select_step_expert_layout(24, &[spec(24, 2)], &[])
4456 .unwrap()
4457 .unwrap();
4458 assert_eq!(selection.layout, StepExpertLayout::ExpertParallel);
4459 assert!(!selection.configured_by_tp);
4460 }
4461
4462 #[test]
4463 fn conflicting_ep_and_tp_assignments_fail_closed() {
4464 let error = select_step_expert_layout(24, &[spec(24, 4)], &[spec(24, 4)]).unwrap_err();
4465 assert!(error.contains("cannot enable MEMRA_STEP_EP and MEMRA_STEP_TP together"));
4466 }
4467
4468 #[test]
4469 fn runtime_registry_owns_one_immutable_load_snapshot() {
4470 let mut source_specs = vec![spec(24, 8)];
4471 let registry = StepParallelRuntimeRegistry::with_config(StepParallelLoadConfig {
4472 ep_specs: Vec::new(),
4473 tp_specs: source_specs.clone(),
4474 native_p2p: true,
4475 ep_device_arithmetic: true,
4476 f32_mirror: true,
4477 bulk_p2p: true,
4478 expert_artifact: StepExpertArtifact::default(),
4479 });
4480 source_specs[0].devices.clear();
4481
4482 let stored = registry.tp_spec(24).unwrap();
4483 assert_eq!(stored.devices, (0..8).collect::<Vec<_>>());
4484 assert!(registry.config.native_p2p);
4485 assert!(registry.config.ep_device_arithmetic);
4486 assert!(registry.config.f32_mirror);
4487 assert!(registry.config.bulk_p2p);
4488 assert_eq!(
4489 registry.expert_selection(24).unwrap().unwrap().layout,
4490 StepExpertLayout::ExpertParallel
4491 );
4492
4493 let standalone = StepParallelRuntimeRegistry::default();
4494 assert!(standalone.tp_spec(24).is_none());
4495 assert!(!standalone.config.native_p2p);
4496 assert!(!standalone.config.ep_device_arithmetic);
4497 assert!(!standalone.config.f32_mirror);
4498 assert!(!standalone.config.bulk_p2p);
4499 }
4500
4501 #[test]
4502 fn rank_local_attention_uses_bounded_swa_rings_only_with_native_p2p() {
4503 assert_eq!(
4504 StepTpAttentionPlacement::resolve(true, None),
4505 StepTpAttentionPlacement::RankLocalGlobal
4506 );
4507 assert_eq!(
4508 StepTpAttentionPlacement::resolve(true, Some(512)),
4509 StepTpAttentionPlacement::RankLocalSwa
4510 );
4511 assert_eq!(
4512 StepTpAttentionPlacement::resolve(false, None),
4513 StepTpAttentionPlacement::OwnerTransportFallback
4514 );
4515 assert_eq!(
4516 StepTpAttentionPlacement::resolve(false, Some(512)),
4517 StepTpAttentionPlacement::OwnerSwa
4518 );
4519 }
4520}
4521
4522#[cfg(test)]
4523mod residency_tests {
4524 use super::{DevExpertFp8ProjectionScales, residency_bytes_by_device};
4525 use crate::model::HostExpertFp8BlockScales;
4526
4527 #[test]
4528 fn pp_residency_counts_only_each_devices_expert_slice() {
4529 let tensors = [
4530 ("blk.0.ffn_gate_exps.weight", 10usize),
4531 ("blk.0.ffn_up_exps.weight", 20),
4532 ("blk.1.ffn_down_exps.weight", 30),
4533 ("blk.2.ffn_gate_exps.weight", 40),
4534 ("blk.3.ffn_up_exps.weight", 50),
4535 ("blk.0.attn_q.weight", 7),
4536 ("output.weight", 11),
4537 ];
4538 let bytes = residency_bytes_by_device(tensors, &[0, 0, 1, 1], 0);
4539 assert_eq!(bytes.experts.get(&0), Some(&60));
4540 assert_eq!(bytes.experts.get(&1), Some(&90));
4541 assert_eq!(bytes.rest, 18);
4542 assert!(bytes.saw_experts);
4543 }
4544
4545 #[test]
4546 fn pp_residency_combines_stages_that_share_one_device() {
4547 let tensors = [
4548 ("blk.0.ffn_gate_exps.weight", 10usize),
4549 ("blk.1.ffn_gate_exps.weight", 20),
4550 ("blk.2.ffn_gate_exps.weight", 30),
4551 ("blk.3.ffn_gate_exps.weight", 40),
4552 ];
4553 let bytes = residency_bytes_by_device(tensors, &[0, 0, 0, 0], 0);
4554 assert_eq!(bytes.experts.get(&0), Some(&100));
4555 assert_eq!(bytes.experts.len(), 1);
4556 }
4557
4558 #[test]
4559 fn resident_fp8_scale_slab_must_match_every_expert() {
4560 let valid = HostExpertFp8BlockScales {
4561 scales: vec![1.0; 12],
4562 rows: 2,
4563 cols: 3,
4564 expert_stride: 6,
4565 };
4566 DevExpertFp8ProjectionScales::validate(&valid, 2).unwrap();
4567
4568 let short = HostExpertFp8BlockScales {
4569 scales: vec![1.0; 11],
4570 ..valid
4571 };
4572 assert_eq!(
4573 DevExpertFp8ProjectionScales::validate(&short, 2).unwrap_err(),
4574 "block-E4M3 scale slab length mismatch: got 11, want 2x6=12"
4575 );
4576 }
4577
4578 #[test]
4579 fn resident_fp8_scale_stride_must_match_its_grid() {
4580 let invalid = HostExpertFp8BlockScales {
4581 scales: vec![1.0; 8],
4582 rows: 2,
4583 cols: 2,
4584 expert_stride: 0,
4585 };
4586 assert_eq!(
4587 DevExpertFp8ProjectionScales::validate(&invalid, 2).unwrap_err(),
4588 "block-E4M3 expert scale stride must be nonzero"
4589 );
4590 }
4591}
4592
4593#[cfg(test)]
4594mod draft_head_tests {
4595 use super::{draft_head_tensor, frspec_trim_own_head_name};
4596
4597 const STEP37_DRAFTER: &[&str] = &[
4604 "output.weight",
4605 "output_norm.weight",
4606 "token_embd.weight",
4607 "blk.45.nextn.shared_head_norm.weight",
4608 "blk.45.nextn.shared_head_head.weight",
4609 "blk.46.nextn.shared_head_head.weight",
4610 "blk.47.nextn.shared_head_head.weight",
4611 ];
4612
4613 fn present(names: &'static [&'static str]) -> impl Fn(&str) -> bool {
4614 move |t: &str| names.contains(&t)
4615 }
4616
4617 #[test]
4625 fn step37_drafter_prefers_the_blocks_own_nextn_head_over_file_level_output() {
4626 assert_eq!(
4627 draft_head_tensor(present(STEP37_DRAFTER), 45),
4628 "blk.45.nextn.shared_head_head.weight"
4629 );
4630 }
4631
4632 #[test]
4636 fn each_nextn_block_selects_its_own_head() {
4637 for n in 45..=47u32 {
4638 assert_eq!(
4639 draft_head_tensor(present(STEP37_DRAFTER), n),
4640 format!("blk.{n}.nextn.shared_head_head.weight")
4641 );
4642 }
4643 }
4644
4645 #[test]
4649 fn draft_without_a_nextn_head_falls_back_to_file_level_output() {
4650 let fr_spec: &[&str] = &["output.weight", "output_norm.weight", "d2t.weight"];
4651 assert_eq!(draft_head_tensor(present(fr_spec), 45), "output.weight");
4652 }
4653
4654 #[test]
4659 fn legacy_shared_head_is_probed_but_loses_to_shared_head_head() {
4660 let legacy_only: &[&str] = &["output.weight", "blk.45.nextn.shared_head.weight"];
4661 assert_eq!(
4662 draft_head_tensor(present(legacy_only), 45),
4663 "blk.45.nextn.shared_head.weight"
4664 );
4665
4666 let both: &[&str] = &[
4667 "output.weight",
4668 "blk.45.nextn.shared_head.weight",
4669 "blk.45.nextn.shared_head_head.weight",
4670 ];
4671 assert_eq!(
4672 draft_head_tensor(present(both), 45),
4673 "blk.45.nextn.shared_head_head.weight"
4674 );
4675 }
4676
4677 #[test]
4681 fn a_different_blocks_nextn_head_is_never_borrowed() {
4682 let wrong_block: &[&str] = &[
4683 "output.weight",
4684 "blk.46.nextn.shared_head_head.weight",
4685 "blk.47.nextn.shared_head_head.weight",
4686 ];
4687 assert_eq!(draft_head_tensor(present(wrong_block), 45), "output.weight");
4688 }
4689
4690 #[test]
4695 fn frspec_trim_prefers_the_nextn_blocks_own_head_name() {
4696 assert_eq!(
4697 frspec_trim_own_head_name(45),
4698 "blk.45.nextn.shared_head_head.weight"
4699 );
4700 assert_eq!(
4702 frspec_trim_own_head_name(45),
4703 format!("blk.{}.nextn.shared_head_head.weight", 45)
4704 );
4705 assert_eq!(
4706 frspec_trim_own_head_name(40),
4707 "blk.40.nextn.shared_head_head.weight"
4708 );
4709 }
4710}