use eredu_nn::{
DistributedNeuralBackend, GroupSelection, GroupedGatedProductOperator, GroupedNeuralBackend,
GroupedRelu2Operator, Tensor, TensorParallelGroupedOutput,
};
use crate::ExpertPass;
use crate::{
observe_and_intervene, ActivationObserver, ParameterBankAccess, ParameterBankKey,
ReplicatedTextMaterializationTask, ReplicatedTextParameterOwner, RoutingObservation,
WeightLoweringKind,
};
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct AddressableBankParameter {
binding_name: String,
task: ReplicatedTextMaterializationTask,
recipe: eredu_checkpoint::recipe::DerivedWeightRecipe,
source_output: eredu_checkpoint::recipe::RecipeMetadata,
selected_bytes: u64,
quantization_companions: Option<crate::QuantizationCompanionBindings>,
}
impl AddressableBankParameter {
pub fn new(
binding_name: impl Into<String>,
task: ReplicatedTextMaterializationTask,
recipe: eredu_checkpoint::recipe::DerivedWeightRecipe,
source_output: eredu_checkpoint::recipe::RecipeMetadata,
selected_bytes: u64,
quantization_companions: Option<crate::QuantizationCompanionBindings>,
) -> Result<Self, AddressableBankMemberError> {
let binding_name = binding_name.into();
if binding_name.trim().is_empty() {
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "addressable binding name is empty".into(),
});
}
task.source_recipe()
.map_err(|error| AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: error.to_string(),
})?;
let descriptor = task.lowering_descriptor();
if descriptor.source() != task.source_encoding()
|| descriptor.executable() != task.executable()
|| descriptor.physical_shape() != task.physical_shape()
|| descriptor.logical_shape() != task.logical_shape()
{
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "selected source, executable, or lowering descriptor drifted".into(),
});
}
let declared_sources = task
.sources()
.iter()
.map(String::as_str)
.collect::<std::collections::BTreeSet<_>>();
let recipe_sources = recipe
.source_keys()
.into_iter()
.collect::<std::collections::BTreeSet<_>>();
if recipe_sources.is_empty() || !recipe_sources.is_subset(&declared_sources) {
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "member recipe consumes sources outside the selected task".into(),
});
}
if source_output.byte_len() == 0 {
return Err(AddressableBankMemberError::ZeroSourceBytes {
parameter: task.name().to_owned(),
});
}
if selected_bytes == 0 {
return Err(AddressableBankMemberError::ZeroSelectedBytes {
parameter: task.name().to_owned(),
});
}
let transforms = matches!(
task.lowering(),
WeightLoweringKind::Transform | WeightLoweringKind::DerivedTransform
);
if !transforms && quantization_companions.is_some() {
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "non-transform lowering declared local transform companions".into(),
});
}
if transforms
&& quantization_companions.is_none()
&& source_output.dtype() != &eredu_checkpoint::recipe::RecipeDtype::F4
{
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "floating transform omitted its local output companions".into(),
});
}
if transforms
&& source_output.dtype() != &eredu_checkpoint::recipe::RecipeDtype::F4
&& task.output_companions().is_empty()
{
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "floating transform omitted its exact selected output companions".into(),
});
}
if let Some(companions) = quantization_companions.as_ref() {
let declared_roles = task
.output_companions()
.iter()
.map(|companion| companion.role())
.collect::<std::collections::BTreeSet<_>>();
let mut bound_roles =
std::collections::BTreeSet::from([eredu_nn::LinearCompanionRole::Scale]);
if companions.affine_bias().is_some() {
bound_roles.insert(eredu_nn::LinearCompanionRole::AffineBias);
}
if declared_roles != bound_roles {
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "selected quantization companion roles differ from exact outputs"
.into(),
});
}
for companion in task.output_companions() {
let local = match companion.role() {
eredu_nn::LinearCompanionRole::Scale => companions.scale(),
eredu_nn::LinearCompanionRole::AffineBias => companions
.affine_bias()
.expect("validated affine-bias role has one binding"),
};
if companion.name() != local && !companion.name().ends_with(&format!(".{local}")) {
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: format!(
"local companion {local:?} differs from selected output {:?}",
companion.name()
),
});
}
let owner_matches = match (task.owner(), companion.owner()) {
(
ReplicatedTextParameterOwner::ExecutionUnit { group, unit },
crate::ParameterGroupOwner::ExecutionUnit {
group: companion_group,
global_unit,
},
) => group == companion_group.as_str() && unit == global_unit,
(
ReplicatedTextParameterOwner::StaticRole(role),
crate::ParameterGroupOwner::StaticRole(companion_role),
) => role == companion_role,
_ => false,
};
if !owner_matches {
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: format!(
"selected companion {:?} has a different owner",
companion.name()
),
});
}
}
}
let expected = selected_addressable_parameter_bytes(&task, &source_output)?;
if selected_bytes != expected {
return Err(AddressableBankMemberError::SelectedByteMismatch {
parameter: task.name().to_owned(),
expected,
actual: selected_bytes,
});
}
Ok(Self {
binding_name,
task,
recipe,
source_output,
selected_bytes,
quantization_companions,
})
}
pub fn binding_name(&self) -> &str {
&self.binding_name
}
pub const fn task(&self) -> &ReplicatedTextMaterializationTask {
&self.task
}
pub const fn recipe(&self) -> &eredu_checkpoint::recipe::DerivedWeightRecipe {
&self.recipe
}
pub const fn source_output(&self) -> &eredu_checkpoint::recipe::RecipeMetadata {
&self.source_output
}
pub const fn source_bytes(&self) -> u64 {
self.source_output.byte_len()
}
pub const fn selected_bytes(&self) -> u64 {
self.selected_bytes
}
pub const fn quantization_companions(&self) -> Option<&crate::QuantizationCompanionBindings> {
self.quantization_companions.as_ref()
}
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct AddressableBankMember {
key: ParameterBankKey,
placement: AddressableBankMemberPlacement,
parameters: Vec<AddressableBankParameter>,
source_bytes: u64,
selected_bytes: u64,
}
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
#[non_exhaustive]
pub enum AddressableBankDistribution {
Replicated,
ExpertParallel,
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct AddressableBankMemberPlacement {
owner_group: crate::ExecutionGroupId,
owner_unit: usize,
unit_path: String,
distribution: AddressableBankDistribution,
owner_rank: Option<usize>,
}
impl AddressableBankMemberPlacement {
pub fn new(
owner_group: crate::ExecutionGroupId,
owner_unit: usize,
unit_path: impl Into<String>,
distribution: AddressableBankDistribution,
) -> Result<Self, AddressableBankMemberError> {
let unit_path = unit_path.into();
if unit_path.trim().is_empty() {
return Err(AddressableBankMemberError::InvalidPlacement(
"addressable member unit path is empty".into(),
));
}
Ok(Self {
owner_group,
owner_unit,
unit_path,
distribution,
owner_rank: None,
})
}
pub fn with_owner_rank(mut self, owner_rank: usize) -> Self {
self.owner_rank = Some(owner_rank);
self
}
pub const fn owner_group(&self) -> &crate::ExecutionGroupId {
&self.owner_group
}
pub const fn owner_unit(&self) -> usize {
self.owner_unit
}
pub fn unit_path(&self) -> &str {
&self.unit_path
}
pub const fn distribution(&self) -> AddressableBankDistribution {
self.distribution
}
pub const fn owner_rank(&self) -> Option<usize> {
self.owner_rank
}
}
impl AddressableBankMember {
pub fn new(
key: ParameterBankKey,
placement: AddressableBankMemberPlacement,
parameters: impl IntoIterator<Item = AddressableBankParameter>,
) -> Result<Self, AddressableBankMemberError> {
let parameters = parameters.into_iter().collect::<Vec<_>>();
if parameters.is_empty() {
return Err(AddressableBankMemberError::EmptyMember { key });
}
if placement.owner_unit() != key.unit() {
return Err(AddressableBankMemberError::InvalidPlacement(format!(
"addressable member unit {} differs from placement unit {}",
key.unit(),
placement.owner_unit()
)));
}
let mut bindings = std::collections::BTreeSet::new();
let mut targets = std::collections::BTreeSet::new();
let mut source_bytes = 0u64;
let mut selected_bytes = 0u64;
for parameter in ¶meters {
if !bindings.insert(parameter.binding_name())
|| !targets.insert(parameter.task().name())
{
return Err(AddressableBankMemberError::DuplicateParameter { key });
}
if !matches!(
parameter.task().owner(),
ReplicatedTextParameterOwner::ExecutionUnit { group, unit }
if *unit == placement.owner_unit()
&& group == placement.owner_group().as_str()
) {
return Err(AddressableBankMemberError::InvalidParameter {
parameter: parameter.task().name().to_owned(),
detail: "selected task has a non-bank owner".into(),
});
}
source_bytes = source_bytes
.checked_add(parameter.source_bytes())
.ok_or(AddressableBankMemberError::SourceByteOverflow { key })?;
selected_bytes = selected_bytes
.checked_add(parameter.selected_bytes())
.ok_or(AddressableBankMemberError::SelectedByteOverflow { key })?;
}
Ok(Self {
key,
placement,
parameters,
source_bytes,
selected_bytes,
})
}
pub const fn key(&self) -> ParameterBankKey {
self.key
}
pub const fn placement(&self) -> &AddressableBankMemberPlacement {
&self.placement
}
pub fn with_owner_rank(mut self, owner_rank: usize) -> Self {
self.placement = self.placement.with_owner_rank(owner_rank);
self
}
pub fn parameters(&self) -> &[AddressableBankParameter] {
&self.parameters
}
pub const fn source_bytes(&self) -> u64 {
self.source_bytes
}
pub const fn selected_bytes(&self) -> u64 {
self.selected_bytes
}
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct AddressableBindingTransform {
quantization: eredu_checkpoint::WeightQuantization,
companion_dtype: eredu_checkpoint::recipe::RecipeDtype,
}
impl AddressableBindingTransform {
pub const fn quantization(&self) -> eredu_checkpoint::WeightQuantization {
self.quantization
}
pub const fn companion_dtype(&self) -> &eredu_checkpoint::recipe::RecipeDtype {
&self.companion_dtype
}
}
#[derive(Debug, Clone)]
pub struct AddressableBankBindingPlan {
key: ParameterBankKey,
bindings: Vec<crate::WeightBinding>,
transformations: std::collections::BTreeMap<String, AddressableBindingTransform>,
selected_bytes: u64,
placement: AddressableBankMemberPlacement,
}
impl AddressableBankBindingPlan {
pub const fn key(&self) -> ParameterBankKey {
self.key
}
pub fn bindings(&self) -> &[crate::WeightBinding] {
&self.bindings
}
pub const fn transformations(
&self,
) -> &std::collections::BTreeMap<String, AddressableBindingTransform> {
&self.transformations
}
pub const fn selected_bytes(&self) -> u64 {
self.selected_bytes
}
pub const fn placement(&self) -> &AddressableBankMemberPlacement {
&self.placement
}
#[allow(clippy::type_complexity)]
pub fn into_parts(
self,
) -> (
ParameterBankKey,
Vec<crate::WeightBinding>,
std::collections::BTreeMap<String, AddressableBindingTransform>,
u64,
AddressableBankMemberPlacement,
) {
(
self.key,
self.bindings,
self.transformations,
self.selected_bytes,
self.placement,
)
}
}
pub fn plan_addressable_bank_bindings<L, E>(
members: &[AddressableBankMember],
source: &dyn eredu_checkpoint::store::CheckpointSource,
mut lower_mxfp4: L,
) -> Result<Vec<AddressableBankBindingPlan>, AddressableBankMemberError>
where
L: FnMut(
&ReplicatedTextMaterializationTask,
eredu_checkpoint::recipe::DerivedWeightRecipe,
&dyn eredu_checkpoint::store::CheckpointSource,
) -> Result<eredu_checkpoint::recipe::DerivedWeightRecipe, E>,
E: std::fmt::Display,
{
let mut plans = Vec::with_capacity(members.len());
for member in members {
let mut bindings = Vec::with_capacity(member.parameters().len());
let mut transformations = std::collections::BTreeMap::new();
for parameter in member.parameters() {
let task = parameter.task();
let declared = task
.sources()
.iter()
.map(String::as_str)
.collect::<std::collections::BTreeSet<_>>();
let physical = task
.physical_sources()
.iter()
.map(|item| item.catalog_key())
.collect::<std::collections::BTreeSet<_>>();
if declared != physical || physical.len() != task.physical_sources().len() {
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "selected physical provenance does not exactly cover task sources"
.into(),
});
}
for admitted in task.physical_sources() {
let actual = source
.source_provenance(admitted.catalog_key())
.map_err(|error| AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: error.to_string(),
})?;
let metadata = source
.source_metadata(admitted.catalog_key())
.map_err(|error| AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: error.to_string(),
})?;
if actual.catalog_key != admitted.catalog_key()
|| actual.physical_tensor != admitted.tensor()
|| actual.output != admitted.output()
|| actual.backing_shard.as_deref() != Some(admitted.shard())
|| actual.source_encoding != *admitted.source_encoding()
|| metadata.encoded_byte_len != admitted.encoded_byte_len()
{
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: format!(
"source {:?} differs from admitted provenance",
admitted.catalog_key()
),
});
}
}
let mut recipe = parameter.recipe().clone();
let inferred = recipe.infer(source).map_err(|error| {
AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: error.to_string(),
}
})?;
if &inferred != parameter.source_output() {
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "member-local recipe output drifted".into(),
});
}
if task.executable() == eredu_checkpoint::LinearFormat::MxFp4
&& inferred.dtype() == &eredu_checkpoint::recipe::RecipeDtype::F4
&& parameter.quantization_companions().is_none()
{
recipe = lower_mxfp4(task, recipe, source).map_err(|error| {
AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: error.to_string(),
}
})?;
}
let metadata = recipe.infer(source).map_err(|error| {
AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: error.to_string(),
}
})?;
let mut binding = crate::WeightBinding::from_recipe(
parameter.binding_name(),
recipe,
metadata.byte_len(),
)
.and_then(|binding| binding.with_logical_target(task.name()))
.map_err(|error| AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: error.to_string(),
})?;
if let Some(companions) = parameter.quantization_companions() {
let quantization = task.executable().weight_quantization().ok_or_else(|| {
AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "transformed task has no packed format".into(),
}
})?;
transformations.insert(
parameter.binding_name().to_owned(),
AddressableBindingTransform {
quantization,
companion_dtype: parameter.source_output().dtype().clone(),
},
);
binding = binding
.with_quantization_companions(
companions.scale(),
companions.affine_bias().map(str::to_owned),
)
.map_err(|error| AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: error.to_string(),
})?;
}
bindings.push(binding);
}
crate::WeightBindingPlan::new(&bindings).map_err(|error| {
AddressableBankMemberError::InvalidParameter {
parameter: format!("{:?}", member.key()),
detail: error.to_string(),
}
})?;
plans.push(AddressableBankBindingPlan {
key: member.key(),
bindings,
transformations,
selected_bytes: member.selected_bytes(),
placement: member.placement().clone(),
});
}
Ok(plans)
}
pub fn selected_addressable_parameter_bytes(
task: &ReplicatedTextMaterializationTask,
metadata: &eredu_checkpoint::recipe::RecipeMetadata,
) -> Result<u64, AddressableBankMemberError> {
if !matches!(
task.lowering(),
WeightLoweringKind::Transform | WeightLoweringKind::DerivedTransform
) {
return Ok(metadata.byte_len());
}
let quantization = task.executable().weight_quantization().ok_or_else(|| {
AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "transform lowering has no packed executable format".into(),
}
})?;
if matches!(
quantization,
eredu_checkpoint::WeightQuantization::GgufIQuant { .. }
) {
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "load-time transform selected checkpoint-native GGUF encoding".into(),
});
}
if task.lowering_descriptor().packed_axis() != metadata.shape().len().checked_sub(1) {
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "transform packed axis is not the final logical matrix axis".into(),
});
}
let shape = metadata.shape();
let (&columns, row_shape) =
shape
.split_last()
.ok_or_else(|| AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "transform target is not a matrix".into(),
})?;
let rows = row_shape
.iter()
.try_fold(1u64, |total, dimension| {
total.checked_mul(*dimension as u64)
})
.ok_or_else(|| AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "transform row geometry overflowed".into(),
})?;
let group = usize::try_from(quantization.group_size()).map_err(|_| {
AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "transform group size is invalid".into(),
}
})?;
if group == 0 || !columns.is_multiple_of(group) || !columns.is_multiple_of(32) {
return Err(AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "transform geometry is incompatible with its packed format".into(),
});
}
let groups = (columns / group) as u64;
let packed = (columns as u64)
.checked_mul(quantization.bits() as u64)
.and_then(|bits| bits.checked_div(8))
.ok_or_else(|| AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "packed row byte geometry overflowed".into(),
})?;
let scalar_bytes = metadata.dtype().bit_width().map_err(|error| {
AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: error.to_string(),
}
})? / 8;
let companion = if matches!(quantization, eredu_checkpoint::WeightQuantization::MxFp4) {
groups
} else {
groups.checked_mul(scalar_bytes).ok_or_else(|| {
AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "scale byte geometry overflowed".into(),
}
})?
};
let bias = if quantization.has_biases() {
companion
} else {
0
};
rows.checked_mul(
packed
.checked_add(companion)
.and_then(|bytes| bytes.checked_add(bias))
.ok_or_else(|| AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "selected row byte geometry overflowed".into(),
})?,
)
.ok_or_else(|| AddressableBankMemberError::InvalidParameter {
parameter: task.name().to_owned(),
detail: "selected byte geometry overflowed".into(),
})
}
#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
pub enum AddressableBankMemberError {
#[error("invalid addressable bank member placement: {0}")]
InvalidPlacement(String),
#[error("addressable bank member {key:?} is empty")]
EmptyMember {
key: ParameterBankKey,
},
#[error("addressable bank member {key:?} repeats a parameter")]
DuplicateParameter {
key: ParameterBankKey,
},
#[error("invalid addressable bank parameter {parameter:?}: {detail}")]
InvalidParameter {
parameter: String,
detail: String,
},
#[error("addressable bank parameter {parameter:?} source byte geometry is zero")]
ZeroSourceBytes {
parameter: String,
},
#[error("addressable bank member {key:?} source byte geometry overflowed")]
SourceByteOverflow {
key: ParameterBankKey,
},
#[error("addressable bank member {key:?} selected byte geometry overflowed")]
SelectedByteOverflow {
key: ParameterBankKey,
},
#[error("addressable bank parameter {parameter:?} selected byte geometry is zero")]
ZeroSelectedBytes {
parameter: String,
},
#[error("addressable bank parameter {parameter:?} selected bytes differ: expected {expected}, got {actual}")]
SelectedByteMismatch {
parameter: String,
expected: u64,
actual: u64,
},
}
pub trait IndexedMovement<B>
where
B: GroupedNeuralBackend,
{
type Error;
fn index_demands(
&mut self,
indices: &B::Tensor,
upper_bound: usize,
context: &<B::Tensor as Tensor>::Context,
) -> Result<Vec<(usize, u64)>, Self::Error>;
fn remap_indices(
&mut self,
indices: &B::Tensor,
mapping: &[(usize, usize)],
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Tensor, Self::Error>;
fn select_rows(
&mut self,
value: &B::Tensor,
start: usize,
end: usize,
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Tensor, Self::Error>;
fn concatenate_rows(
&mut self,
values: &[B::Tensor],
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Tensor, Self::Error>;
}
pub trait ExpertRouteTensorMovement<T> {
type Error;
fn shape(&self, value: &T) -> Vec<usize>;
fn gather_rows(&mut self, value: &T, rows: &[usize]) -> Result<T, Self::Error>;
fn gather_route_values(
&mut self,
value: &T,
flattened_routes: &[usize],
) -> Result<T, Self::Error>;
fn scatter_add_rows(
&mut self,
value: T,
destination_rows: &[usize],
output_rows: usize,
) -> Result<T, Self::Error>;
}
pub trait ExpertRouteExchange<T> {
type Error;
fn exchange_tensor(
&mut self,
counts: &crate::CommunicationPeerCounts,
value: T,
) -> Result<T, Self::Error>;
fn exchange_indices(
&mut self,
counts: &crate::CommunicationPeerCounts,
values: Vec<usize>,
) -> Result<Vec<usize>, Self::Error>;
}
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
#[non_exhaustive]
pub enum ExpertRouteCombination {
CoefficientWeightedSum,
}
pub struct AddressableExpertRouteRequest<'a, T> {
pub unit: usize,
pub input: &'a T,
pub global_experts: &'a [usize],
pub owner_local_experts: &'a [usize],
pub selected_scores: &'a T,
pub coefficients: &'a T,
pub pass: ExpertPass,
pub access: ParameterBankAccess,
pub combination: ExpertRouteCombination,
}
impl<T> AddressableExpertRouteRequest<'_, T> {
pub fn addressable_bank_key(&self, row: usize) -> Option<ParameterBankKey> {
self.global_experts
.get(row)
.copied()
.map(|global| ParameterBankKey::new(self.unit, global))
}
pub fn owner_local_execution_id(&self, row: usize) -> Option<usize> {
self.owner_local_experts.get(row).copied()
}
}
pub trait AddressableExpertRouteProvider<T> {
type Error;
fn execute_addressable_routes(
&mut self,
request: AddressableExpertRouteRequest<'_, T>,
) -> Result<T, Self::Error>;
fn execute_addressable_routes_tensor_parallel(
&mut self,
request: AddressableExpertRouteRequest<'_, T>,
) -> Result<RoutedExpertTensorParallelOutput<T>, Self::Error> {
self.execute_addressable_routes(request)
.map(RoutedExpertTensorParallelOutput::Complete)
}
}
#[derive(Debug, Clone, Copy)]
pub struct ParameterBankAcquisition<'a> {
entries: &'a [(ParameterBankKey, u64)],
access: ParameterBankAccess,
}
impl<'a> ParameterBankAcquisition<'a> {
pub const fn new(entries: &'a [(ParameterBankKey, u64)], access: ParameterBankAccess) -> Self {
Self { entries, access }
}
pub const fn entries(&self) -> &'a [(ParameterBankKey, u64)] {
self.entries
}
pub const fn access(&self) -> ParameterBankAccess {
self.access
}
}
pub trait AddressableGroupedBank<B>
where
B: GroupedNeuralBackend,
{
type Acquisition;
type Report;
type Error;
fn member_bytes(&self, key: ParameterBankKey) -> Option<u64>;
fn acquire(
&mut self,
request: ParameterBankAcquisition<'_>,
context: &<B::Tensor as Tensor>::Context,
) -> Result<Self::Acquisition, Self::Error>;
fn gated_product_groups(
&mut self,
acquisition: &Self::Acquisition,
spec: &eredu_nn::GroupedGatedProductSpec,
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::GatedProductGroups, Self::Error>;
fn relu2_groups(
&mut self,
acquisition: &Self::Acquisition,
spec: &eredu_nn::GroupedRelu2Spec,
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Relu2Groups, Self::Error>;
fn complete(
&mut self,
acquisition: Self::Acquisition,
output: &B::Tensor,
context: &<B::Tensor as Tensor>::Context,
) -> Result<(), Self::Error>;
fn report(&self) -> Result<Self::Report, Self::Error>;
}
pub trait AddressableGatedProductBank<B>
where
B: GroupedNeuralBackend,
{
type Error;
fn acquire(
&mut self,
key: ParameterBankKey,
spec: &eredu_nn::GroupedGatedProductSpec,
context: &<B::Tensor as Tensor>::Context,
) -> Result<&mut B::GatedProductGroups, Self::Error>;
}
pub struct RoutedExpertRequest<'a, T> {
pub layer: usize,
pub input: &'a T,
pub routes: &'a GroupSelection<T>,
pub pass: ExpertPass,
}
impl<T> RoutedExpertRequest<'_, T> {
pub const fn parameter_bank_access(&self) -> ParameterBankAccess {
self.pass.parameter_bank_access()
}
}
pub enum RoutedExpertTensorParallelOutput<T> {
Complete(T),
Partial(TensorParallelGroupedOutput<T>),
}
pub fn reduce_tensor_parallel_expert_output<B>(
output: TensorParallelGroupedOutput<B::Tensor>,
parallel: &B::ParallelContext,
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Tensor, eredu_nn::Error>
where
B: GroupedNeuralBackend + DistributedNeuralBackend,
{
let reduced = B::sum_parallel(output.reducible().clone(), parallel, context)?;
match output.post_reduce().cloned() {
Some(bias) => reduced.add(&bias, context),
None => Ok(reduced),
}
}
pub fn combine_tensor_parallel_expert_outputs<B>(
left: TensorParallelGroupedOutput<B::Tensor>,
right: TensorParallelGroupedOutput<B::Tensor>,
context: &<B::Tensor as Tensor>::Context,
) -> Result<TensorParallelGroupedOutput<B::Tensor>, eredu_nn::Error>
where
B: GroupedNeuralBackend,
{
let post_reduce = match (left.post_reduce().cloned(), right.post_reduce().cloned()) {
(Some(left), Some(right)) => Some(left.add(&right, context)?),
(Some(bias), None) | (None, Some(bias)) => Some(bias),
(None, None) => None,
};
Ok(TensorParallelGroupedOutput::new(
left.reducible().add(right.reducible(), context)?,
post_reduce,
))
}
pub fn combine_routed_expert_tensor_parallel<B>(
left: RoutedExpertTensorParallelOutput<B::Tensor>,
right: RoutedExpertTensorParallelOutput<B::Tensor>,
context: &<B::Tensor as Tensor>::Context,
) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, eredu_nn::Error>
where
B: GroupedNeuralBackend,
{
match (left, right) {
(
RoutedExpertTensorParallelOutput::Complete(left),
RoutedExpertTensorParallelOutput::Complete(right),
) => Ok(RoutedExpertTensorParallelOutput::Complete(
left.add(&right, context)?,
)),
(
RoutedExpertTensorParallelOutput::Partial(left),
RoutedExpertTensorParallelOutput::Partial(right),
) => combine_tensor_parallel_expert_outputs::<B>(left, right, context)
.map(RoutedExpertTensorParallelOutput::Partial),
_ => Err(eredu_nn::Error::backend(
"provider mixed complete and rank-local expert outputs in one block",
)),
}
}
pub fn reduce_routed_expert_tensor_parallel<B>(
output: RoutedExpertTensorParallelOutput<B::Tensor>,
parallel: &B::ParallelContext,
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Tensor, eredu_nn::Error>
where
B: GroupedNeuralBackend + DistributedNeuralBackend,
{
match output {
RoutedExpertTensorParallelOutput::Complete(output) => Ok(output),
RoutedExpertTensorParallelOutput::Partial(output) => {
reduce_tensor_parallel_expert_output::<B>(output, parallel, context)
}
}
}
pub trait RoutedExpertProvider<B>
where
B: GroupedNeuralBackend,
{
type Error;
fn forward_grouped(
&mut self,
resident_bank: &mut B::GatedProductGroups,
request: RoutedExpertRequest<'_, B::Tensor>,
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Tensor, Self::Error>;
fn forward_compact_grouped(
&mut self,
resident_bank: &mut B::GatedProductGroups,
request: RoutedExpertRequest<'_, B::Tensor>,
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Tensor, Self::Error> {
self.forward_grouped(resident_bank, request, context)
}
fn forward_relu2_routed(
&mut self,
resident_bank: &mut B::Relu2Groups,
request: RoutedExpertRequest<'_, B::Tensor>,
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Tensor, Self::Error>;
}
pub trait TensorParallelRoutedExpertProvider<B>: RoutedExpertProvider<B>
where
B: GroupedNeuralBackend,
{
fn forward_grouped_tensor_parallel(
&mut self,
resident_bank: &mut B::GatedProductGroups,
request: RoutedExpertRequest<'_, B::Tensor>,
partitions: usize,
context: &<B::Tensor as Tensor>::Context,
) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error>;
fn forward_compact_grouped_tensor_parallel(
&mut self,
resident_bank: &mut B::GatedProductGroups,
request: RoutedExpertRequest<'_, B::Tensor>,
partitions: usize,
context: &<B::Tensor as Tensor>::Context,
) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
self.forward_grouped_tensor_parallel(resident_bank, request, partitions, context)
}
fn forward_relu2_routed_tensor_parallel(
&mut self,
resident_bank: &mut B::Relu2Groups,
request: RoutedExpertRequest<'_, B::Tensor>,
partitions: usize,
context: &<B::Tensor as Tensor>::Context,
) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error>;
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RoutedObservationPoint {
path: String,
expert_count: i32,
}
impl RoutedObservationPoint {
pub fn new(path: impl Into<String>, expert_count: i32) -> Self {
Self {
path: path.into(),
expert_count,
}
}
pub fn path(&self) -> &str {
&self.path
}
pub const fn expert_count(&self) -> i32 {
self.expert_count
}
}
#[derive(Debug)]
pub enum ObservedExpertProviderError<P, O> {
Provider(P),
Observer(O),
}
impl<P, O> std::fmt::Display for ObservedExpertProviderError<P, O>
where
P: std::fmt::Display,
O: std::fmt::Display,
{
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Provider(error) => write!(formatter, "routed expert provider failed: {error}"),
Self::Observer(error) => write!(formatter, "routed expert observer failed: {error}"),
}
}
}
impl<P, O> std::error::Error for ObservedExpertProviderError<P, O>
where
P: std::error::Error + 'static,
O: std::error::Error + 'static,
{
}
pub struct ObservedExpertProvider<'a, P, O: ?Sized, E> {
provider: &'a mut P,
observer: &'a mut O,
point: RoutedObservationPoint,
error: std::marker::PhantomData<fn() -> E>,
}
impl<'a, P, O: ?Sized, E> ObservedExpertProvider<'a, P, O, E> {
pub fn new(provider: &'a mut P, observer: &'a mut O, point: RoutedObservationPoint) -> Self {
Self {
provider,
observer,
point,
error: std::marker::PhantomData,
}
}
fn observe<T, ObservationError>(
&mut self,
routes: &eredu_nn::GroupSelection<T>,
output: &T,
) -> Result<T, ObservationError>
where
T: Clone,
O: ActivationObserver<T, ObservationError>,
{
self.observer.observe_routing(RoutingObservation {
path: self.point.path(),
selected_experts: routes.group_indices(),
selected_scores: routes.selected_scores(),
coefficients: routes.coefficients(),
routed_output: output,
local_routed_output: None,
reduced_routed_output: None,
shared_output: None,
combined_output: None,
expert_count: self.point.expert_count(),
})?;
observe_and_intervene(
self.observer,
&format!("{}.output", self.point.path()),
output,
)
}
}
impl<B, P, O, E> RoutedExpertProvider<B> for ObservedExpertProvider<'_, P, O, E>
where
B: GroupedNeuralBackend,
P: RoutedExpertProvider<B>,
O: ActivationObserver<B::Tensor, E> + ?Sized,
{
type Error = ObservedExpertProviderError<P::Error, E>;
fn forward_grouped(
&mut self,
resident_bank: &mut B::GatedProductGroups,
request: RoutedExpertRequest<'_, B::Tensor>,
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Tensor, Self::Error> {
let routes = request.routes;
let output = self
.provider
.forward_grouped(resident_bank, request, context)
.map_err(ObservedExpertProviderError::Provider)?;
self.observe(routes, &output)
.map_err(ObservedExpertProviderError::Observer)
}
fn forward_relu2_routed(
&mut self,
resident_bank: &mut B::Relu2Groups,
request: RoutedExpertRequest<'_, B::Tensor>,
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Tensor, Self::Error> {
let routes = request.routes;
let output = self
.provider
.forward_relu2_routed(resident_bank, request, context)
.map_err(ObservedExpertProviderError::Provider)?;
self.observe(routes, &output)
.map_err(ObservedExpertProviderError::Observer)
}
}
impl<B, P, O, E> TensorParallelRoutedExpertProvider<B> for ObservedExpertProvider<'_, P, O, E>
where
B: GroupedNeuralBackend,
P: TensorParallelRoutedExpertProvider<B>,
O: ActivationObserver<B::Tensor, E> + ?Sized,
{
fn forward_grouped_tensor_parallel(
&mut self,
resident_bank: &mut B::GatedProductGroups,
request: RoutedExpertRequest<'_, B::Tensor>,
partitions: usize,
context: &<B::Tensor as Tensor>::Context,
) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
self.provider
.forward_grouped_tensor_parallel(resident_bank, request, partitions, context)
.map_err(ObservedExpertProviderError::Provider)
}
fn forward_relu2_routed_tensor_parallel(
&mut self,
resident_bank: &mut B::Relu2Groups,
request: RoutedExpertRequest<'_, B::Tensor>,
partitions: usize,
context: &<B::Tensor as Tensor>::Context,
) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
self.provider
.forward_relu2_routed_tensor_parallel(resident_bank, request, partitions, context)
.map_err(ObservedExpertProviderError::Provider)
}
}
#[derive(Debug, Default, Clone, Copy)]
pub struct ResidentExpertProvider;
impl<B> RoutedExpertProvider<B> for ResidentExpertProvider
where
B: GroupedNeuralBackend,
{
type Error = eredu_nn::Error;
fn forward_grouped(
&mut self,
resident_bank: &mut B::GatedProductGroups,
request: RoutedExpertRequest<'_, B::Tensor>,
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Tensor, Self::Error> {
resident_bank.forward_grouped(request.input, request.routes, context)
}
fn forward_relu2_routed(
&mut self,
resident_bank: &mut B::Relu2Groups,
request: RoutedExpertRequest<'_, B::Tensor>,
context: &<B::Tensor as Tensor>::Context,
) -> Result<B::Tensor, Self::Error> {
resident_bank.forward_grouped(request.input, request.routes, context)
}
}
impl<B> TensorParallelRoutedExpertProvider<B> for ResidentExpertProvider
where
B: eredu_nn::TensorParallelGroupedNeuralBackend,
{
fn forward_grouped_tensor_parallel(
&mut self,
resident_bank: &mut B::GatedProductGroups,
request: RoutedExpertRequest<'_, B::Tensor>,
partitions: usize,
context: &<B::Tensor as Tensor>::Context,
) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
B::gated_product_groups_tensor_parallel(
resident_bank,
request.input,
request.routes,
partitions,
context,
)
.map(RoutedExpertTensorParallelOutput::Partial)
}
fn forward_relu2_routed_tensor_parallel(
&mut self,
resident_bank: &mut B::Relu2Groups,
request: RoutedExpertRequest<'_, B::Tensor>,
partitions: usize,
context: &<B::Tensor as Tensor>::Context,
) -> Result<RoutedExpertTensorParallelOutput<B::Tensor>, Self::Error> {
B::relu2_groups_tensor_parallel(
resident_bank,
request.input,
request.routes,
partitions,
context,
)
.map(RoutedExpertTensorParallelOutput::Partial)
}
}