1use eredu_nn::{
4 DistributedNeuralBackend, GroupSelection, GroupedGatedProductOperator, GroupedNeuralBackend,
5 GroupedRelu2Operator, Tensor, TensorParallelGroupedOutput,
6};
7
8use crate::ExpertPass;
9use crate::{
10 observe_and_intervene, ActivationObserver, ParameterBankAccess, ParameterBankKey,
11 ReplicatedTextMaterializationTask, ReplicatedTextParameterOwner, RoutingObservation,
12 WeightLoweringKind,
13};
14
15#[derive(Debug, Clone, Eq, PartialEq)]
17pub struct AddressableBankParameter {
18 binding_name: String,
19 task: ReplicatedTextMaterializationTask,
20 recipe: eredu_checkpoint::recipe::DerivedWeightRecipe,
21 source_output: eredu_checkpoint::recipe::RecipeMetadata,
22 selected_bytes: u64,
23 quantization_companions: Option<crate::QuantizationCompanionBindings>,
24}
25
26impl AddressableBankParameter {
27 pub fn new(
29 binding_name: impl Into<String>,
30 task: ReplicatedTextMaterializationTask,
31 recipe: eredu_checkpoint::recipe::DerivedWeightRecipe,
32 source_output: eredu_checkpoint::recipe::RecipeMetadata,
33 selected_bytes: u64,
34 quantization_companions: Option<crate::QuantizationCompanionBindings>,
35 ) -> Result<Self, AddressableBankMemberError> {
36 let binding_name = binding_name.into();
37 if binding_name.trim().is_empty() {
38 return Err(AddressableBankMemberError::InvalidParameter {
39 parameter: task.name().to_owned(),
40 detail: "addressable binding name is empty".into(),
41 });
42 }
43 task.source_recipe()
44 .map_err(|error| AddressableBankMemberError::InvalidParameter {
45 parameter: task.name().to_owned(),
46 detail: error.to_string(),
47 })?;
48 let descriptor = task.lowering_descriptor();
49 if descriptor.source() != task.source_encoding()
50 || descriptor.executable() != task.executable()
51 || descriptor.physical_shape() != task.physical_shape()
52 || descriptor.logical_shape() != task.logical_shape()
53 {
54 return Err(AddressableBankMemberError::InvalidParameter {
55 parameter: task.name().to_owned(),
56 detail: "selected source, executable, or lowering descriptor drifted".into(),
57 });
58 }
59 let declared_sources = task
60 .sources()
61 .iter()
62 .map(String::as_str)
63 .collect::<std::collections::BTreeSet<_>>();
64 let recipe_sources = recipe
65 .source_keys()
66 .into_iter()
67 .collect::<std::collections::BTreeSet<_>>();
68 if recipe_sources.is_empty() || !recipe_sources.is_subset(&declared_sources) {
69 return Err(AddressableBankMemberError::InvalidParameter {
70 parameter: task.name().to_owned(),
71 detail: "member recipe consumes sources outside the selected task".into(),
72 });
73 }
74 if source_output.byte_len() == 0 {
75 return Err(AddressableBankMemberError::ZeroSourceBytes {
76 parameter: task.name().to_owned(),
77 });
78 }
79 if selected_bytes == 0 {
80 return Err(AddressableBankMemberError::ZeroSelectedBytes {
81 parameter: task.name().to_owned(),
82 });
83 }
84 let transforms = matches!(
85 task.lowering(),
86 WeightLoweringKind::Transform | WeightLoweringKind::DerivedTransform
87 );
88 if !transforms && quantization_companions.is_some() {
89 return Err(AddressableBankMemberError::InvalidParameter {
90 parameter: task.name().to_owned(),
91 detail: "non-transform lowering declared local transform companions".into(),
92 });
93 }
94 if transforms
95 && quantization_companions.is_none()
96 && source_output.dtype() != &eredu_checkpoint::recipe::RecipeDtype::F4
97 {
98 return Err(AddressableBankMemberError::InvalidParameter {
99 parameter: task.name().to_owned(),
100 detail: "floating transform omitted its local output companions".into(),
101 });
102 }
103 if transforms
104 && source_output.dtype() != &eredu_checkpoint::recipe::RecipeDtype::F4
105 && task.output_companions().is_empty()
106 {
107 return Err(AddressableBankMemberError::InvalidParameter {
108 parameter: task.name().to_owned(),
109 detail: "floating transform omitted its exact selected output companions".into(),
110 });
111 }
112 if let Some(companions) = quantization_companions.as_ref() {
113 let declared_roles = task
114 .output_companions()
115 .iter()
116 .map(|companion| companion.role())
117 .collect::<std::collections::BTreeSet<_>>();
118 let mut bound_roles =
119 std::collections::BTreeSet::from([eredu_nn::LinearCompanionRole::Scale]);
120 if companions.affine_bias().is_some() {
121 bound_roles.insert(eredu_nn::LinearCompanionRole::AffineBias);
122 }
123 if declared_roles != bound_roles {
124 return Err(AddressableBankMemberError::InvalidParameter {
125 parameter: task.name().to_owned(),
126 detail: "selected quantization companion roles differ from exact outputs"
127 .into(),
128 });
129 }
130 for companion in task.output_companions() {
131 let local = match companion.role() {
132 eredu_nn::LinearCompanionRole::Scale => companions.scale(),
133 eredu_nn::LinearCompanionRole::AffineBias => companions
134 .affine_bias()
135 .expect("validated affine-bias role has one binding"),
136 };
137 if companion.name() != local && !companion.name().ends_with(&format!(".{local}")) {
138 return Err(AddressableBankMemberError::InvalidParameter {
139 parameter: task.name().to_owned(),
140 detail: format!(
141 "local companion {local:?} differs from selected output {:?}",
142 companion.name()
143 ),
144 });
145 }
146 let owner_matches = match (task.owner(), companion.owner()) {
147 (
148 ReplicatedTextParameterOwner::ExecutionUnit { group, unit },
149 crate::ParameterGroupOwner::ExecutionUnit {
150 group: companion_group,
151 global_unit,
152 },
153 ) => group == companion_group.as_str() && unit == global_unit,
154 (
155 ReplicatedTextParameterOwner::StaticRole(role),
156 crate::ParameterGroupOwner::StaticRole(companion_role),
157 ) => role == companion_role,
158 _ => false,
159 };
160 if !owner_matches {
161 return Err(AddressableBankMemberError::InvalidParameter {
162 parameter: task.name().to_owned(),
163 detail: format!(
164 "selected companion {:?} has a different owner",
165 companion.name()
166 ),
167 });
168 }
169 }
170 }
171 let expected = selected_addressable_parameter_bytes(&task, &source_output)?;
172 if selected_bytes != expected {
173 return Err(AddressableBankMemberError::SelectedByteMismatch {
174 parameter: task.name().to_owned(),
175 expected,
176 actual: selected_bytes,
177 });
178 }
179 Ok(Self {
180 binding_name,
181 task,
182 recipe,
183 source_output,
184 selected_bytes,
185 quantization_companions,
186 })
187 }
188
189 pub fn binding_name(&self) -> &str {
191 &self.binding_name
192 }
193
194 pub const fn task(&self) -> &ReplicatedTextMaterializationTask {
196 &self.task
197 }
198
199 pub const fn recipe(&self) -> &eredu_checkpoint::recipe::DerivedWeightRecipe {
201 &self.recipe
202 }
203
204 pub const fn source_output(&self) -> &eredu_checkpoint::recipe::RecipeMetadata {
206 &self.source_output
207 }
208
209 pub const fn source_bytes(&self) -> u64 {
211 self.source_output.byte_len()
212 }
213
214 pub const fn selected_bytes(&self) -> u64 {
216 self.selected_bytes
217 }
218
219 pub const fn quantization_companions(&self) -> Option<&crate::QuantizationCompanionBindings> {
221 self.quantization_companions.as_ref()
222 }
223}
224
225#[derive(Debug, Clone, Eq, PartialEq)]
227pub struct AddressableBankMember {
228 key: ParameterBankKey,
229 placement: AddressableBankMemberPlacement,
230 parameters: Vec<AddressableBankParameter>,
231 source_bytes: u64,
232 selected_bytes: u64,
233}
234
235#[derive(Debug, Clone, Copy, Eq, PartialEq)]
237#[non_exhaustive]
238pub enum AddressableBankDistribution {
239 Replicated,
241 ExpertParallel,
243}
244
245#[derive(Debug, Clone, Eq, PartialEq)]
247pub struct AddressableBankMemberPlacement {
248 owner_group: crate::ExecutionGroupId,
249 owner_unit: usize,
250 unit_path: String,
251 distribution: AddressableBankDistribution,
252 owner_rank: Option<usize>,
253}
254
255impl AddressableBankMemberPlacement {
256 pub fn new(
258 owner_group: crate::ExecutionGroupId,
259 owner_unit: usize,
260 unit_path: impl Into<String>,
261 distribution: AddressableBankDistribution,
262 ) -> Result<Self, AddressableBankMemberError> {
263 let unit_path = unit_path.into();
264 if unit_path.trim().is_empty() {
265 return Err(AddressableBankMemberError::InvalidPlacement(
266 "addressable member unit path is empty".into(),
267 ));
268 }
269 Ok(Self {
270 owner_group,
271 owner_unit,
272 unit_path,
273 distribution,
274 owner_rank: None,
275 })
276 }
277
278 pub fn with_owner_rank(mut self, owner_rank: usize) -> Self {
280 self.owner_rank = Some(owner_rank);
281 self
282 }
283
284 pub const fn owner_group(&self) -> &crate::ExecutionGroupId {
286 &self.owner_group
287 }
288 pub const fn owner_unit(&self) -> usize {
290 self.owner_unit
291 }
292 pub fn unit_path(&self) -> &str {
294 &self.unit_path
295 }
296 pub const fn distribution(&self) -> AddressableBankDistribution {
298 self.distribution
299 }
300 pub const fn owner_rank(&self) -> Option<usize> {
302 self.owner_rank
303 }
304}
305
306impl AddressableBankMember {
307 pub fn new(
309 key: ParameterBankKey,
310 placement: AddressableBankMemberPlacement,
311 parameters: impl IntoIterator<Item = AddressableBankParameter>,
312 ) -> Result<Self, AddressableBankMemberError> {
313 let parameters = parameters.into_iter().collect::<Vec<_>>();
314 if parameters.is_empty() {
315 return Err(AddressableBankMemberError::EmptyMember { key });
316 }
317 if placement.owner_unit() != key.unit() {
318 return Err(AddressableBankMemberError::InvalidPlacement(format!(
319 "addressable member unit {} differs from placement unit {}",
320 key.unit(),
321 placement.owner_unit()
322 )));
323 }
324 let mut bindings = std::collections::BTreeSet::new();
325 let mut targets = std::collections::BTreeSet::new();
326 let mut source_bytes = 0u64;
327 let mut selected_bytes = 0u64;
328 for parameter in ¶meters {
329 if !bindings.insert(parameter.binding_name())
330 || !targets.insert(parameter.task().name())
331 {
332 return Err(AddressableBankMemberError::DuplicateParameter { key });
333 }
334 if !matches!(
335 parameter.task().owner(),
336 ReplicatedTextParameterOwner::ExecutionUnit { group, unit }
337 if *unit == placement.owner_unit()
338 && group == placement.owner_group().as_str()
339 ) {
340 return Err(AddressableBankMemberError::InvalidParameter {
341 parameter: parameter.task().name().to_owned(),
342 detail: "selected task has a non-bank owner".into(),
343 });
344 }
345 source_bytes = source_bytes
346 .checked_add(parameter.source_bytes())
347 .ok_or(AddressableBankMemberError::SourceByteOverflow { key })?;
348 selected_bytes = selected_bytes
349 .checked_add(parameter.selected_bytes())
350 .ok_or(AddressableBankMemberError::SelectedByteOverflow { key })?;
351 }
352 Ok(Self {
353 key,
354 placement,
355 parameters,
356 source_bytes,
357 selected_bytes,
358 })
359 }
360
361 pub const fn key(&self) -> ParameterBankKey {
363 self.key
364 }
365
366 pub const fn placement(&self) -> &AddressableBankMemberPlacement {
368 &self.placement
369 }
370
371 pub fn with_owner_rank(mut self, owner_rank: usize) -> Self {
373 self.placement = self.placement.with_owner_rank(owner_rank);
374 self
375 }
376
377 pub fn parameters(&self) -> &[AddressableBankParameter] {
379 &self.parameters
380 }
381
382 pub const fn source_bytes(&self) -> u64 {
384 self.source_bytes
385 }
386
387 pub const fn selected_bytes(&self) -> u64 {
389 self.selected_bytes
390 }
391}
392
393#[derive(Debug, Clone, Eq, PartialEq)]
395pub struct AddressableBindingTransform {
396 quantization: eredu_checkpoint::WeightQuantization,
397 companion_dtype: eredu_checkpoint::recipe::RecipeDtype,
398}
399
400impl AddressableBindingTransform {
401 pub const fn quantization(&self) -> eredu_checkpoint::WeightQuantization {
403 self.quantization
404 }
405 pub const fn companion_dtype(&self) -> &eredu_checkpoint::recipe::RecipeDtype {
407 &self.companion_dtype
408 }
409}
410
411#[derive(Debug, Clone)]
413pub struct AddressableBankBindingPlan {
414 key: ParameterBankKey,
415 bindings: Vec<crate::WeightBinding>,
416 transformations: std::collections::BTreeMap<String, AddressableBindingTransform>,
417 selected_bytes: u64,
418 placement: AddressableBankMemberPlacement,
419}
420
421impl AddressableBankBindingPlan {
422 pub const fn key(&self) -> ParameterBankKey {
424 self.key
425 }
426 pub fn bindings(&self) -> &[crate::WeightBinding] {
428 &self.bindings
429 }
430 pub const fn transformations(
432 &self,
433 ) -> &std::collections::BTreeMap<String, AddressableBindingTransform> {
434 &self.transformations
435 }
436 pub const fn selected_bytes(&self) -> u64 {
438 self.selected_bytes
439 }
440 pub const fn placement(&self) -> &AddressableBankMemberPlacement {
442 &self.placement
443 }
444 #[allow(clippy::type_complexity)]
446 pub fn into_parts(
447 self,
448 ) -> (
449 ParameterBankKey,
450 Vec<crate::WeightBinding>,
451 std::collections::BTreeMap<String, AddressableBindingTransform>,
452 u64,
453 AddressableBankMemberPlacement,
454 ) {
455 (
456 self.key,
457 self.bindings,
458 self.transformations,
459 self.selected_bytes,
460 self.placement,
461 )
462 }
463}
464
465pub fn plan_addressable_bank_bindings<L, E>(
467 members: &[AddressableBankMember],
468 source: &dyn eredu_checkpoint::store::CheckpointSource,
469 mut lower_mxfp4: L,
470) -> Result<Vec<AddressableBankBindingPlan>, AddressableBankMemberError>
471where
472 L: FnMut(
473 &ReplicatedTextMaterializationTask,
474 eredu_checkpoint::recipe::DerivedWeightRecipe,
475 &dyn eredu_checkpoint::store::CheckpointSource,
476 ) -> Result<eredu_checkpoint::recipe::DerivedWeightRecipe, E>,
477 E: std::fmt::Display,
478{
479 let mut plans = Vec::with_capacity(members.len());
480 for member in members {
481 let mut bindings = Vec::with_capacity(member.parameters().len());
482 let mut transformations = std::collections::BTreeMap::new();
483 for parameter in member.parameters() {
484 let task = parameter.task();
485 let declared = task
486 .sources()
487 .iter()
488 .map(String::as_str)
489 .collect::<std::collections::BTreeSet<_>>();
490 let physical = task
491 .physical_sources()
492 .iter()
493 .map(|item| item.catalog_key())
494 .collect::<std::collections::BTreeSet<_>>();
495 if declared != physical || physical.len() != task.physical_sources().len() {
496 return Err(AddressableBankMemberError::InvalidParameter {
497 parameter: task.name().to_owned(),
498 detail: "selected physical provenance does not exactly cover task sources"
499 .into(),
500 });
501 }
502 for admitted in task.physical_sources() {
503 let actual = source
504 .source_provenance(admitted.catalog_key())
505 .map_err(|error| AddressableBankMemberError::InvalidParameter {
506 parameter: task.name().to_owned(),
507 detail: error.to_string(),
508 })?;
509 let metadata = source
510 .source_metadata(admitted.catalog_key())
511 .map_err(|error| AddressableBankMemberError::InvalidParameter {
512 parameter: task.name().to_owned(),
513 detail: error.to_string(),
514 })?;
515 if actual.catalog_key != admitted.catalog_key()
516 || actual.physical_tensor != admitted.tensor()
517 || actual.output != admitted.output()
518 || actual.backing_shard.as_deref() != Some(admitted.shard())
519 || actual.source_encoding != *admitted.source_encoding()
520 || metadata.encoded_byte_len != admitted.encoded_byte_len()
521 {
522 return Err(AddressableBankMemberError::InvalidParameter {
523 parameter: task.name().to_owned(),
524 detail: format!(
525 "source {:?} differs from admitted provenance",
526 admitted.catalog_key()
527 ),
528 });
529 }
530 }
531 let mut recipe = parameter.recipe().clone();
532 let inferred = recipe.infer(source).map_err(|error| {
533 AddressableBankMemberError::InvalidParameter {
534 parameter: task.name().to_owned(),
535 detail: error.to_string(),
536 }
537 })?;
538 if &inferred != parameter.source_output() {
539 return Err(AddressableBankMemberError::InvalidParameter {
540 parameter: task.name().to_owned(),
541 detail: "member-local recipe output drifted".into(),
542 });
543 }
544 if task.executable() == eredu_checkpoint::LinearFormat::MxFp4
545 && inferred.dtype() == &eredu_checkpoint::recipe::RecipeDtype::F4
546 && parameter.quantization_companions().is_none()
547 {
548 recipe = lower_mxfp4(task, recipe, source).map_err(|error| {
549 AddressableBankMemberError::InvalidParameter {
550 parameter: task.name().to_owned(),
551 detail: error.to_string(),
552 }
553 })?;
554 }
555 let metadata = recipe.infer(source).map_err(|error| {
556 AddressableBankMemberError::InvalidParameter {
557 parameter: task.name().to_owned(),
558 detail: error.to_string(),
559 }
560 })?;
561 let mut binding = crate::WeightBinding::from_recipe(
562 parameter.binding_name(),
563 recipe,
564 metadata.byte_len(),
565 )
566 .and_then(|binding| binding.with_logical_target(task.name()))
567 .map_err(|error| AddressableBankMemberError::InvalidParameter {
568 parameter: task.name().to_owned(),
569 detail: error.to_string(),
570 })?;
571 if let Some(companions) = parameter.quantization_companions() {
572 let quantization = task.executable().weight_quantization().ok_or_else(|| {
573 AddressableBankMemberError::InvalidParameter {
574 parameter: task.name().to_owned(),
575 detail: "transformed task has no packed format".into(),
576 }
577 })?;
578 transformations.insert(
579 parameter.binding_name().to_owned(),
580 AddressableBindingTransform {
581 quantization,
582 companion_dtype: parameter.source_output().dtype().clone(),
583 },
584 );
585 binding = binding
586 .with_quantization_companions(
587 companions.scale(),
588 companions.affine_bias().map(str::to_owned),
589 )
590 .map_err(|error| AddressableBankMemberError::InvalidParameter {
591 parameter: task.name().to_owned(),
592 detail: error.to_string(),
593 })?;
594 }
595 bindings.push(binding);
596 }
597 crate::WeightBindingPlan::new(&bindings).map_err(|error| {
598 AddressableBankMemberError::InvalidParameter {
599 parameter: format!("{:?}", member.key()),
600 detail: error.to_string(),
601 }
602 })?;
603 plans.push(AddressableBankBindingPlan {
604 key: member.key(),
605 bindings,
606 transformations,
607 selected_bytes: member.selected_bytes(),
608 placement: member.placement().clone(),
609 });
610 }
611 Ok(plans)
612}
613
614pub fn selected_addressable_parameter_bytes(
622 task: &ReplicatedTextMaterializationTask,
623 metadata: &eredu_checkpoint::recipe::RecipeMetadata,
624) -> Result<u64, AddressableBankMemberError> {
625 if !matches!(
626 task.lowering(),
627 WeightLoweringKind::Transform | WeightLoweringKind::DerivedTransform
628 ) {
629 return Ok(metadata.byte_len());
630 }
631 let quantization = task.executable().weight_quantization().ok_or_else(|| {
632 AddressableBankMemberError::InvalidParameter {
633 parameter: task.name().to_owned(),
634 detail: "transform lowering has no packed executable format".into(),
635 }
636 })?;
637 if matches!(
638 quantization,
639 eredu_checkpoint::WeightQuantization::GgufIQuant { .. }
640 ) {
641 return Err(AddressableBankMemberError::InvalidParameter {
642 parameter: task.name().to_owned(),
643 detail: "load-time transform selected checkpoint-native GGUF encoding".into(),
644 });
645 }
646 if task.lowering_descriptor().packed_axis() != metadata.shape().len().checked_sub(1) {
647 return Err(AddressableBankMemberError::InvalidParameter {
648 parameter: task.name().to_owned(),
649 detail: "transform packed axis is not the final logical matrix axis".into(),
650 });
651 }
652 let shape = metadata.shape();
653 let (&columns, row_shape) =
654 shape
655 .split_last()
656 .ok_or_else(|| AddressableBankMemberError::InvalidParameter {
657 parameter: task.name().to_owned(),
658 detail: "transform target is not a matrix".into(),
659 })?;
660 let rows = row_shape
661 .iter()
662 .try_fold(1u64, |total, dimension| {
663 total.checked_mul(*dimension as u64)
664 })
665 .ok_or_else(|| AddressableBankMemberError::InvalidParameter {
666 parameter: task.name().to_owned(),
667 detail: "transform row geometry overflowed".into(),
668 })?;
669 let group = usize::try_from(quantization.group_size()).map_err(|_| {
670 AddressableBankMemberError::InvalidParameter {
671 parameter: task.name().to_owned(),
672 detail: "transform group size is invalid".into(),
673 }
674 })?;
675 if group == 0 || !columns.is_multiple_of(group) || !columns.is_multiple_of(32) {
676 return Err(AddressableBankMemberError::InvalidParameter {
677 parameter: task.name().to_owned(),
678 detail: "transform geometry is incompatible with its packed format".into(),
679 });
680 }
681 let groups = (columns / group) as u64;
682 let packed = (columns as u64)
683 .checked_mul(quantization.bits() as u64)
684 .and_then(|bits| bits.checked_div(8))
685 .ok_or_else(|| AddressableBankMemberError::InvalidParameter {
686 parameter: task.name().to_owned(),
687 detail: "packed row byte geometry overflowed".into(),
688 })?;
689 let scalar_bytes = metadata.dtype().bit_width().map_err(|error| {
690 AddressableBankMemberError::InvalidParameter {
691 parameter: task.name().to_owned(),
692 detail: error.to_string(),
693 }
694 })? / 8;
695 let companion = if matches!(quantization, eredu_checkpoint::WeightQuantization::MxFp4) {
696 groups
697 } else {
698 groups.checked_mul(scalar_bytes).ok_or_else(|| {
699 AddressableBankMemberError::InvalidParameter {
700 parameter: task.name().to_owned(),
701 detail: "scale byte geometry overflowed".into(),
702 }
703 })?
704 };
705 let bias = if quantization.has_biases() {
706 companion
707 } else {
708 0
709 };
710 rows.checked_mul(
711 packed
712 .checked_add(companion)
713 .and_then(|bytes| bytes.checked_add(bias))
714 .ok_or_else(|| AddressableBankMemberError::InvalidParameter {
715 parameter: task.name().to_owned(),
716 detail: "selected row byte geometry overflowed".into(),
717 })?,
718 )
719 .ok_or_else(|| AddressableBankMemberError::InvalidParameter {
720 parameter: task.name().to_owned(),
721 detail: "selected byte geometry overflowed".into(),
722 })
723}
724
725#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
727pub enum AddressableBankMemberError {
728 #[error("invalid addressable bank member placement: {0}")]
730 InvalidPlacement(String),
731 #[error("addressable bank member {key:?} is empty")]
733 EmptyMember {
734 key: ParameterBankKey,
736 },
737 #[error("addressable bank member {key:?} repeats a parameter")]
739 DuplicateParameter {
740 key: ParameterBankKey,
742 },
743 #[error("invalid addressable bank parameter {parameter:?}: {detail}")]
745 InvalidParameter {
746 parameter: String,
748 detail: String,
750 },
751 #[error("addressable bank parameter {parameter:?} source byte geometry is zero")]
753 ZeroSourceBytes {
754 parameter: String,
756 },
757 #[error("addressable bank member {key:?} source byte geometry overflowed")]
759 SourceByteOverflow {
760 key: ParameterBankKey,
762 },
763 #[error("addressable bank member {key:?} selected byte geometry overflowed")]
765 SelectedByteOverflow {
766 key: ParameterBankKey,
768 },
769 #[error("addressable bank parameter {parameter:?} selected byte geometry is zero")]
771 ZeroSelectedBytes {
772 parameter: String,
774 },
775 #[error("addressable bank parameter {parameter:?} selected bytes differ: expected {expected}, got {actual}")]
777 SelectedByteMismatch {
778 parameter: String,
780 expected: u64,
782 actual: u64,
784 },
785}
786
787pub trait IndexedMovement<B>
792where
793 B: GroupedNeuralBackend,
794{
795 type Error;
797
798 fn index_demands(
800 &mut self,
801 indices: &B::Tensor,
802 upper_bound: usize,
803 context: &<B::Tensor as Tensor>::Context,
804 ) -> Result<Vec<(usize, u64)>, Self::Error>;
805
806 fn remap_indices(
808 &mut self,
809 indices: &B::Tensor,
810 mapping: &[(usize, usize)],
811 context: &<B::Tensor as Tensor>::Context,
812 ) -> Result<B::Tensor, Self::Error>;
813
814 fn select_rows(
816 &mut self,
817 value: &B::Tensor,
818 start: usize,
819 end: usize,
820 context: &<B::Tensor as Tensor>::Context,
821 ) -> Result<B::Tensor, Self::Error>;
822
823 fn concatenate_rows(
825 &mut self,
826 values: &[B::Tensor],
827 context: &<B::Tensor as Tensor>::Context,
828 ) -> Result<B::Tensor, Self::Error>;
829}
830
831pub trait ExpertRouteTensorMovement<T> {
837 type Error;
839
840 fn shape(&self, value: &T) -> Vec<usize>;
842
843 fn gather_rows(&mut self, value: &T, rows: &[usize]) -> Result<T, Self::Error>;
845
846 fn gather_route_values(
848 &mut self,
849 value: &T,
850 flattened_routes: &[usize],
851 ) -> Result<T, Self::Error>;
852
853 fn scatter_add_rows(
858 &mut self,
859 value: T,
860 destination_rows: &[usize],
861 output_rows: usize,
862 ) -> Result<T, Self::Error>;
863}
864
865pub trait ExpertRouteExchange<T> {
871 type Error;
873
874 fn exchange_tensor(
876 &mut self,
877 counts: &crate::CommunicationPeerCounts,
878 value: T,
879 ) -> Result<T, Self::Error>;
880
881 fn exchange_indices(
883 &mut self,
884 counts: &crate::CommunicationPeerCounts,
885 values: Vec<usize>,
886 ) -> Result<Vec<usize>, Self::Error>;
887}
888
889#[derive(Debug, Clone, Copy, Eq, PartialEq)]
891#[non_exhaustive]
892pub enum ExpertRouteCombination {
893 CoefficientWeightedSum,
895}
896
897pub struct AddressableExpertRouteRequest<'a, T> {
899 pub unit: usize,
901 pub input: &'a T,
903 pub global_experts: &'a [usize],
909 pub owner_local_experts: &'a [usize],
911 pub selected_scores: &'a T,
913 pub coefficients: &'a T,
915 pub pass: ExpertPass,
917 pub access: ParameterBankAccess,
919 pub combination: ExpertRouteCombination,
921}
922
923impl<T> AddressableExpertRouteRequest<'_, T> {
924 pub fn addressable_bank_key(&self, row: usize) -> Option<ParameterBankKey> {
929 self.global_experts
930 .get(row)
931 .copied()
932 .map(|global| ParameterBankKey::new(self.unit, global))
933 }
934
935 pub fn owner_local_execution_id(&self, row: usize) -> Option<usize> {
937 self.owner_local_experts.get(row).copied()
938 }
939}
940
941pub trait AddressableExpertRouteProvider<T> {
948 type Error;
950
951 fn execute_addressable_routes(
953 &mut self,
954 request: AddressableExpertRouteRequest<'_, T>,
955 ) -> Result<T, Self::Error>;
956
957 fn execute_addressable_routes_tensor_parallel(
966 &mut self,
967 request: AddressableExpertRouteRequest<'_, T>,
968 ) -> Result<RoutedExpertTensorParallelOutput<T>, Self::Error> {
969 self.execute_addressable_routes(request)
970 .map(RoutedExpertTensorParallelOutput::Complete)
971 }
972}
973
974#[derive(Debug, Clone, Copy)]
976pub struct ParameterBankAcquisition<'a> {
977 entries: &'a [(ParameterBankKey, u64)],
978 access: ParameterBankAccess,
979}
980
981impl<'a> ParameterBankAcquisition<'a> {
982 pub const fn new(entries: &'a [(ParameterBankKey, u64)], access: ParameterBankAccess) -> Self {
984 Self { entries, access }
985 }
986
987 pub const fn entries(&self) -> &'a [(ParameterBankKey, u64)] {
989 self.entries
990 }
991
992 pub const fn access(&self) -> ParameterBankAccess {
994 self.access
995 }
996}
997
998pub trait AddressableGroupedBank<B>
1004where
1005 B: GroupedNeuralBackend,
1006{
1007 type Acquisition;
1009 type Report;
1011 type Error;
1013
1014 fn member_bytes(&self, key: ParameterBankKey) -> Option<u64>;
1016
1017 fn acquire(
1019 &mut self,
1020 request: ParameterBankAcquisition<'_>,
1021 context: &<B::Tensor as Tensor>::Context,
1022 ) -> Result<Self::Acquisition, Self::Error>;
1023
1024 fn gated_product_groups(
1026 &mut self,
1027 acquisition: &Self::Acquisition,
1028 spec: &eredu_nn::GroupedGatedProductSpec,
1029 context: &<B::Tensor as Tensor>::Context,
1030 ) -> Result<B::GatedProductGroups, Self::Error>;
1031
1032 fn relu2_groups(
1034 &mut self,
1035 acquisition: &Self::Acquisition,
1036 spec: &eredu_nn::GroupedRelu2Spec,
1037 context: &<B::Tensor as Tensor>::Context,
1038 ) -> Result<B::Relu2Groups, Self::Error>;
1039
1040 fn complete(
1042 &mut self,
1043 acquisition: Self::Acquisition,
1044 output: &B::Tensor,
1045 context: &<B::Tensor as Tensor>::Context,
1046 ) -> Result<(), Self::Error>;
1047
1048 fn report(&self) -> Result<Self::Report, Self::Error>;
1050}
1051
1052pub trait AddressableGatedProductBank<B>
1054where
1055 B: GroupedNeuralBackend,
1056{
1057 type Error;
1059
1060 fn acquire(
1062 &mut self,
1063 key: ParameterBankKey,
1064 spec: &eredu_nn::GroupedGatedProductSpec,
1065 context: &<B::Tensor as Tensor>::Context,
1066 ) -> Result<&mut B::GatedProductGroups, Self::Error>;
1067}
1068
1069pub struct RoutedExpertRequest<'a, T> {
1071 pub layer: usize,
1073 pub input: &'a T,
1075 pub routes: &'a GroupSelection<T>,
1077 pub pass: ExpertPass,
1079}
1080
1081impl<T> RoutedExpertRequest<'_, T> {
1082 pub const fn parameter_bank_access(&self) -> ParameterBankAccess {
1085 self.pass.parameter_bank_access()
1086 }
1087}
1088
1089pub enum RoutedExpertTensorParallelOutput<T> {
1091 Complete(T),
1093 Partial(TensorParallelGroupedOutput<T>),
1095}
1096
1097pub fn reduce_tensor_parallel_expert_output<B>(
1099 output: TensorParallelGroupedOutput<B::Tensor>,
1100 parallel: &B::ParallelContext,
1101 context: &<B::Tensor as Tensor>::Context,
1102) -> Result<B::Tensor, eredu_nn::Error>
1103where
1104 B: GroupedNeuralBackend + DistributedNeuralBackend,
1105{
1106 let reduced = B::sum_parallel(output.reducible().clone(), parallel, context)?;
1107 match output.post_reduce().cloned() {
1108 Some(bias) => reduced.add(&bias, context),
1109 None => Ok(reduced),
1110 }
1111}
1112
1113pub fn combine_tensor_parallel_expert_outputs<B>(
1115 left: TensorParallelGroupedOutput<B::Tensor>,
1116 right: TensorParallelGroupedOutput<B::Tensor>,
1117 context: &<B::Tensor as Tensor>::Context,
1118) -> Result<TensorParallelGroupedOutput<B::Tensor>, eredu_nn::Error>
1119where
1120 B: GroupedNeuralBackend,
1121{
1122 let post_reduce = match (left.post_reduce().cloned(), right.post_reduce().cloned()) {
1123 (Some(left), Some(right)) => Some(left.add(&right, context)?),
1124 (Some(bias), None) | (None, Some(bias)) => Some(bias),
1125 (None, None) => None,
1126 };
1127 Ok(TensorParallelGroupedOutput::new(
1128 left.reducible().add(right.reducible(), context)?,
1129 post_reduce,
1130 ))
1131}
1132
1133pub fn combine_routed_expert_tensor_parallel<B>(
1135 left: RoutedExpertTensorParallelOutput<B::Tensor>,
1136 right: RoutedExpertTensorParallelOutput<B::Tensor>,
1137 context: &<B::Tensor as Tensor>::Context,
1138) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, eredu_nn::Error>
1139where
1140 B: GroupedNeuralBackend,
1141{
1142 match (left, right) {
1143 (
1144 RoutedExpertTensorParallelOutput::Complete(left),
1145 RoutedExpertTensorParallelOutput::Complete(right),
1146 ) => Ok(RoutedExpertTensorParallelOutput::Complete(
1147 left.add(&right, context)?,
1148 )),
1149 (
1150 RoutedExpertTensorParallelOutput::Partial(left),
1151 RoutedExpertTensorParallelOutput::Partial(right),
1152 ) => combine_tensor_parallel_expert_outputs::<B>(left, right, context)
1153 .map(RoutedExpertTensorParallelOutput::Partial),
1154 _ => Err(eredu_nn::Error::backend(
1155 "provider mixed complete and rank-local expert outputs in one block",
1156 )),
1157 }
1158}
1159
1160pub fn reduce_routed_expert_tensor_parallel<B>(
1162 output: RoutedExpertTensorParallelOutput<B::Tensor>,
1163 parallel: &B::ParallelContext,
1164 context: &<B::Tensor as Tensor>::Context,
1165) -> Result<B::Tensor, eredu_nn::Error>
1166where
1167 B: GroupedNeuralBackend + DistributedNeuralBackend,
1168{
1169 match output {
1170 RoutedExpertTensorParallelOutput::Complete(output) => Ok(output),
1171 RoutedExpertTensorParallelOutput::Partial(output) => {
1172 reduce_tensor_parallel_expert_output::<B>(output, parallel, context)
1173 }
1174 }
1175}
1176
1177pub trait RoutedExpertProvider<B>
1184where
1185 B: GroupedNeuralBackend,
1186{
1187 type Error;
1189
1190 fn forward_grouped(
1192 &mut self,
1193 resident_bank: &mut B::GatedProductGroups,
1194 request: RoutedExpertRequest<'_, B::Tensor>,
1195 context: &<B::Tensor as Tensor>::Context,
1196 ) -> Result<B::Tensor, Self::Error>;
1197
1198 fn forward_compact_grouped(
1204 &mut self,
1205 resident_bank: &mut B::GatedProductGroups,
1206 request: RoutedExpertRequest<'_, B::Tensor>,
1207 context: &<B::Tensor as Tensor>::Context,
1208 ) -> Result<B::Tensor, Self::Error> {
1209 self.forward_grouped(resident_bank, request, context)
1210 }
1211
1212 fn forward_relu2_routed(
1214 &mut self,
1215 resident_bank: &mut B::Relu2Groups,
1216 request: RoutedExpertRequest<'_, B::Tensor>,
1217 context: &<B::Tensor as Tensor>::Context,
1218 ) -> Result<B::Tensor, Self::Error>;
1219}
1220
1221pub trait TensorParallelRoutedExpertProvider<B>: RoutedExpertProvider<B>
1223where
1224 B: GroupedNeuralBackend,
1225{
1226 fn forward_grouped_tensor_parallel(
1228 &mut self,
1229 resident_bank: &mut B::GatedProductGroups,
1230 request: RoutedExpertRequest<'_, B::Tensor>,
1231 partitions: usize,
1232 context: &<B::Tensor as Tensor>::Context,
1233 ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error>;
1234
1235 fn forward_compact_grouped_tensor_parallel(
1238 &mut self,
1239 resident_bank: &mut B::GatedProductGroups,
1240 request: RoutedExpertRequest<'_, B::Tensor>,
1241 partitions: usize,
1242 context: &<B::Tensor as Tensor>::Context,
1243 ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
1244 self.forward_grouped_tensor_parallel(resident_bank, request, partitions, context)
1245 }
1246
1247 fn forward_relu2_routed_tensor_parallel(
1249 &mut self,
1250 resident_bank: &mut B::Relu2Groups,
1251 request: RoutedExpertRequest<'_, B::Tensor>,
1252 partitions: usize,
1253 context: &<B::Tensor as Tensor>::Context,
1254 ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error>;
1255}
1256
1257#[derive(Debug, Clone, Eq, PartialEq)]
1260pub struct RoutedObservationPoint {
1261 path: String,
1262 expert_count: i32,
1263}
1264
1265impl RoutedObservationPoint {
1266 pub fn new(path: impl Into<String>, expert_count: i32) -> Self {
1268 Self {
1269 path: path.into(),
1270 expert_count,
1271 }
1272 }
1273
1274 pub fn path(&self) -> &str {
1276 &self.path
1277 }
1278
1279 pub const fn expert_count(&self) -> i32 {
1281 self.expert_count
1282 }
1283}
1284
1285#[derive(Debug)]
1287pub enum ObservedExpertProviderError<P, O> {
1288 Provider(P),
1290 Observer(O),
1292}
1293
1294impl<P, O> std::fmt::Display for ObservedExpertProviderError<P, O>
1295where
1296 P: std::fmt::Display,
1297 O: std::fmt::Display,
1298{
1299 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1300 match self {
1301 Self::Provider(error) => write!(formatter, "routed expert provider failed: {error}"),
1302 Self::Observer(error) => write!(formatter, "routed expert observer failed: {error}"),
1303 }
1304 }
1305}
1306
1307impl<P, O> std::error::Error for ObservedExpertProviderError<P, O>
1308where
1309 P: std::error::Error + 'static,
1310 O: std::error::Error + 'static,
1311{
1312}
1313
1314pub struct ObservedExpertProvider<'a, P, O: ?Sized, E> {
1322 provider: &'a mut P,
1323 observer: &'a mut O,
1324 point: RoutedObservationPoint,
1325 error: std::marker::PhantomData<fn() -> E>,
1326}
1327
1328impl<'a, P, O: ?Sized, E> ObservedExpertProvider<'a, P, O, E> {
1329 pub fn new(provider: &'a mut P, observer: &'a mut O, point: RoutedObservationPoint) -> Self {
1331 Self {
1332 provider,
1333 observer,
1334 point,
1335 error: std::marker::PhantomData,
1336 }
1337 }
1338
1339 fn observe<T, ObservationError>(
1340 &mut self,
1341 routes: &eredu_nn::GroupSelection<T>,
1342 output: &T,
1343 ) -> Result<T, ObservationError>
1344 where
1345 T: Clone,
1346 O: ActivationObserver<T, ObservationError>,
1347 {
1348 self.observer.observe_routing(RoutingObservation {
1349 path: self.point.path(),
1350 selected_experts: routes.group_indices(),
1351 selected_scores: routes.selected_scores(),
1352 coefficients: routes.coefficients(),
1353 routed_output: output,
1354 local_routed_output: None,
1355 reduced_routed_output: None,
1356 shared_output: None,
1357 combined_output: None,
1358 expert_count: self.point.expert_count(),
1359 })?;
1360 observe_and_intervene(
1361 self.observer,
1362 &format!("{}.output", self.point.path()),
1363 output,
1364 )
1365 }
1366}
1367
1368impl<B, P, O, E> RoutedExpertProvider<B> for ObservedExpertProvider<'_, P, O, E>
1369where
1370 B: GroupedNeuralBackend,
1371 P: RoutedExpertProvider<B>,
1372 O: ActivationObserver<B::Tensor, E> + ?Sized,
1373{
1374 type Error = ObservedExpertProviderError<P::Error, E>;
1375
1376 fn forward_grouped(
1377 &mut self,
1378 resident_bank: &mut B::GatedProductGroups,
1379 request: RoutedExpertRequest<'_, B::Tensor>,
1380 context: &<B::Tensor as Tensor>::Context,
1381 ) -> Result<B::Tensor, Self::Error> {
1382 let routes = request.routes;
1383 let output = self
1384 .provider
1385 .forward_grouped(resident_bank, request, context)
1386 .map_err(ObservedExpertProviderError::Provider)?;
1387 self.observe(routes, &output)
1388 .map_err(ObservedExpertProviderError::Observer)
1389 }
1390
1391 fn forward_relu2_routed(
1392 &mut self,
1393 resident_bank: &mut B::Relu2Groups,
1394 request: RoutedExpertRequest<'_, B::Tensor>,
1395 context: &<B::Tensor as Tensor>::Context,
1396 ) -> Result<B::Tensor, Self::Error> {
1397 let routes = request.routes;
1398 let output = self
1399 .provider
1400 .forward_relu2_routed(resident_bank, request, context)
1401 .map_err(ObservedExpertProviderError::Provider)?;
1402 self.observe(routes, &output)
1403 .map_err(ObservedExpertProviderError::Observer)
1404 }
1405}
1406
1407impl<B, P, O, E> TensorParallelRoutedExpertProvider<B> for ObservedExpertProvider<'_, P, O, E>
1408where
1409 B: GroupedNeuralBackend,
1410 P: TensorParallelRoutedExpertProvider<B>,
1411 O: ActivationObserver<B::Tensor, E> + ?Sized,
1412{
1413 fn forward_grouped_tensor_parallel(
1414 &mut self,
1415 resident_bank: &mut B::GatedProductGroups,
1416 request: RoutedExpertRequest<'_, B::Tensor>,
1417 partitions: usize,
1418 context: &<B::Tensor as Tensor>::Context,
1419 ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
1420 self.provider
1421 .forward_grouped_tensor_parallel(resident_bank, request, partitions, context)
1422 .map_err(ObservedExpertProviderError::Provider)
1423 }
1424
1425 fn forward_relu2_routed_tensor_parallel(
1426 &mut self,
1427 resident_bank: &mut B::Relu2Groups,
1428 request: RoutedExpertRequest<'_, B::Tensor>,
1429 partitions: usize,
1430 context: &<B::Tensor as Tensor>::Context,
1431 ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
1432 self.provider
1433 .forward_relu2_routed_tensor_parallel(resident_bank, request, partitions, context)
1434 .map_err(ObservedExpertProviderError::Provider)
1435 }
1436}
1437
1438#[derive(Debug, Default, Clone, Copy)]
1440pub struct ResidentExpertProvider;
1441
1442impl<B> RoutedExpertProvider<B> for ResidentExpertProvider
1443where
1444 B: GroupedNeuralBackend,
1445{
1446 type Error = eredu_nn::Error;
1447
1448 fn forward_grouped(
1449 &mut self,
1450 resident_bank: &mut B::GatedProductGroups,
1451 request: RoutedExpertRequest<'_, B::Tensor>,
1452 context: &<B::Tensor as Tensor>::Context,
1453 ) -> Result<B::Tensor, Self::Error> {
1454 resident_bank.forward_grouped(request.input, request.routes, context)
1455 }
1456
1457 fn forward_relu2_routed(
1458 &mut self,
1459 resident_bank: &mut B::Relu2Groups,
1460 request: RoutedExpertRequest<'_, B::Tensor>,
1461 context: &<B::Tensor as Tensor>::Context,
1462 ) -> Result<B::Tensor, Self::Error> {
1463 resident_bank.forward_grouped(request.input, request.routes, context)
1464 }
1465}
1466
1467impl<B> TensorParallelRoutedExpertProvider<B> for ResidentExpertProvider
1468where
1469 B: eredu_nn::TensorParallelGroupedNeuralBackend,
1470{
1471 fn forward_grouped_tensor_parallel(
1472 &mut self,
1473 resident_bank: &mut B::GatedProductGroups,
1474 request: RoutedExpertRequest<'_, B::Tensor>,
1475 partitions: usize,
1476 context: &<B::Tensor as Tensor>::Context,
1477 ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
1478 B::gated_product_groups_tensor_parallel(
1479 resident_bank,
1480 request.input,
1481 request.routes,
1482 partitions,
1483 context,
1484 )
1485 .map(RoutedExpertTensorParallelOutput::Partial)
1486 }
1487
1488 fn forward_relu2_routed_tensor_parallel(
1489 &mut self,
1490 resident_bank: &mut B::Relu2Groups,
1491 request: RoutedExpertRequest<'_, B::Tensor>,
1492 partitions: usize,
1493 context: &<B::Tensor as Tensor>::Context,
1494 ) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
1495 B::relu2_groups_tensor_parallel(
1496 resident_bank,
1497 request.input,
1498 request.routes,
1499 partitions,
1500 context,
1501 )
1502 .map(RoutedExpertTensorParallelOutput::Partial)
1503 }
1504}