use std::ops::Deref;
use std::{collections::BTreeMap, collections::BTreeSet, ops::Range};
use crate::{
ArchitectureStatePartitionError, ArchitectureStatePartitionPlan, ArchitectureStatePlacement,
ExecutionGraph, ExecutionGroupId, ExecutionUnitLayout, LayeredForwardState,
LayeredPartitionInput, LayeredPartitionOutput, ParameterGroupSpec,
PartitionedLayeredArchitecture, RuntimeState, StateLayout,
};
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd)]
#[non_exhaustive]
pub enum ParameterGroupOwner {
StaticRole(String),
StaticAnyOf(Vec<String>),
#[non_exhaustive]
ExecutionUnit {
group: ExecutionGroupId,
global_unit: usize,
},
}
impl ParameterGroupOwner {
pub fn static_role(role: impl Into<String>) -> Self {
Self::StaticRole(role.into())
}
pub fn static_any_of(roles: impl IntoIterator<Item = impl Into<String>>) -> Self {
Self::StaticAnyOf(roles.into_iter().map(Into::into).collect())
}
pub fn execution_unit(group: ExecutionGroupId, global_unit: usize) -> Self {
Self::ExecutionUnit { group, global_unit }
}
fn is_local<G, A>(&self, partition: &ArchitecturePartition<G, A>) -> bool {
match self {
Self::StaticRole(role) => partition.ownership().owns_static_role(role),
Self::StaticAnyOf(roles) => roles
.iter()
.any(|role| partition.ownership().owns_static_role(role)),
Self::ExecutionUnit { group, global_unit } => {
partition.owns_unit(group.as_str(), *global_unit)
}
}
}
fn is_local_partition_parts(
&self,
groups: &[PartitionGroup],
ownership: &PartitionOwnership,
) -> bool {
match self {
Self::StaticRole(role) => ownership.owns_static_role(role),
Self::StaticAnyOf(roles) => roles.iter().any(|role| ownership.owns_static_role(role)),
Self::ExecutionUnit { group, global_unit } => groups
.iter()
.any(|owned| owned.group() == group && owned.contains(*global_unit)),
}
}
fn static_storage_role(&self) -> Option<&str> {
match self {
Self::StaticRole(role) => Some(role),
Self::StaticAnyOf(roles) => roles.first().map(String::as_str),
Self::ExecutionUnit { .. } => None,
}
}
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct OwnedParameterGroupSpec {
owner: ParameterGroupOwner,
group: ParameterGroupSpec,
}
impl OwnedParameterGroupSpec {
pub fn new(owner: ParameterGroupOwner, group: ParameterGroupSpec) -> Self {
Self { owner, group }
}
pub const fn owner(&self) -> &ParameterGroupOwner {
&self.owner
}
pub const fn group(&self) -> &ParameterGroupSpec {
&self.group
}
pub fn into_group(self) -> ParameterGroupSpec {
self.group
}
}
impl Deref for OwnedParameterGroupSpec {
type Target = ParameterGroupSpec;
fn deref(&self) -> &Self::Target {
&self.group
}
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct ArchitectureParameterDescription {
graph: ExecutionGraph,
unit_layout: ExecutionUnitLayout,
groups: Vec<OwnedParameterGroupSpec>,
}
impl ArchitectureParameterDescription {
pub fn new(
graph: &ExecutionGraph,
layout: &ExecutionUnitLayout,
expected: impl IntoIterator<Item = ParameterGroupSpec>,
groups: impl IntoIterator<Item = OwnedParameterGroupSpec>,
) -> Result<Self, ArchitectureParameterError> {
validate_canonical_layout(graph, layout)
.map_err(|error| ArchitectureParameterError::InvalidLayout(error.to_string()))?;
let expected = parameter_targets(expected)?;
let groups = groups.into_iter().collect::<Vec<_>>();
let mut actual = BTreeMap::new();
for tagged in &groups {
match tagged.owner() {
ParameterGroupOwner::StaticRole(role) => {
if role.trim().is_empty() {
return Err(ArchitectureParameterError::EmptyStaticRole);
}
}
ParameterGroupOwner::StaticAnyOf(roles) => {
if roles.is_empty() || roles.iter().any(|role| role.trim().is_empty()) {
return Err(ArchitectureParameterError::EmptyStaticRole);
}
let unique = roles.iter().collect::<BTreeSet<_>>();
if unique.len() != roles.len() {
return Err(ArchitectureParameterError::DuplicateStaticRole);
}
}
ParameterGroupOwner::ExecutionUnit { group, global_unit } => {
let Some(group_index) = graph
.groups()
.iter()
.position(|candidate| candidate.id() == group.as_str())
else {
return Err(ArchitectureParameterError::UnknownExecutionGroup(
group.as_str().to_owned(),
));
};
let available = layout
.group_range(group_index)
.expect("validated canonical layout contains every group")
.len();
if *global_unit >= available {
return Err(ArchitectureParameterError::UnitOutOfRange {
group: group.as_str().to_owned(),
global_unit: *global_unit,
available,
});
}
}
}
for member in tagged.group().members() {
if let Some(previous) = actual.insert(member.target().to_owned(), tagged.owner()) {
return Err(ArchitectureParameterError::DuplicateOwnership {
target: member.target().to_owned(),
first: previous.clone(),
second: tagged.owner().clone(),
});
}
}
}
let actual_targets = actual.keys().cloned().collect::<BTreeSet<_>>();
let expected_targets = expected.keys().cloned().collect::<BTreeSet<_>>();
if let Some(target) = expected_targets.difference(&actual_targets).next() {
return Err(ArchitectureParameterError::MissingOwnership(target.clone()));
}
if let Some(target) = actual_targets.difference(&expected_targets).next() {
return Err(ArchitectureParameterError::UnexpectedOwnership(
target.clone(),
));
}
Ok(Self {
graph: graph.clone(),
unit_layout: layout.clone(),
groups,
})
}
pub const fn graph(&self) -> &ExecutionGraph {
&self.graph
}
pub const fn unit_layout(&self) -> &ExecutionUnitLayout {
&self.unit_layout
}
pub fn validate_architecture<B, S, M>(
&self,
architecture: &M,
) -> Result<(), ArchitecturePartitionError>
where
B: eredu_nn::NeuralBackend,
S: crate::RuntimeState<B>,
M: crate::LayeredArchitecture<B, S>,
M::Error: std::fmt::Display,
{
let (graph, unit_layout) = canonical_architecture_layout::<B, S, M>(architecture)?;
if graph != self.graph {
return Err(ArchitecturePartitionError::ArchitectureGraphMismatch);
}
if unit_layout != self.unit_layout {
return Err(ArchitecturePartitionError::ArchitectureUnitLayoutMismatch);
}
Ok(())
}
pub fn groups(&self) -> &[OwnedParameterGroupSpec] {
&self.groups
}
pub fn targets_for_role(&self, role: crate::ParameterRole) -> BTreeSet<String> {
self.groups
.iter()
.filter(|owned| owned.group().role() == role)
.flat_map(|owned| owned.group().members())
.map(|member| member.target().to_owned())
.collect()
}
pub fn select_owned<G, A>(
&self,
partition: &ArchitecturePartition<G, A>,
) -> Vec<OwnedParameterGroupSpec> {
self.groups
.iter()
.filter(|tagged| tagged.owner().is_local(partition))
.cloned()
.collect()
}
pub fn select_static_roles<'a, G, A>(
&'a self,
partition: &ArchitecturePartition<G, A>,
) -> Vec<&'a str> {
self.groups
.iter()
.filter(|tagged| tagged.owner().is_local(partition))
.filter_map(|tagged| tagged.owner().static_storage_role())
.collect::<BTreeSet<_>>()
.into_iter()
.collect()
}
}
fn parameter_targets(
groups: impl IntoIterator<Item = ParameterGroupSpec>,
) -> Result<BTreeMap<String, String>, ArchitectureParameterError> {
let mut targets = BTreeMap::new();
for group in groups {
for member in group.members() {
if let Some(previous) =
targets.insert(member.target().to_owned(), group.logical_name().to_owned())
{
return Err(ArchitectureParameterError::DuplicateExpectedTarget {
target: member.target().to_owned(),
first: previous,
second: group.logical_name().to_owned(),
});
}
}
}
Ok(targets)
}
#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
#[non_exhaustive]
pub enum ArchitectureParameterError {
#[error("invalid architecture parameter layout: {0}")]
InvalidLayout(String),
#[error("architecture parameter static role must not be empty")]
EmptyStaticRole,
#[error("architecture shared parameter owner repeats a static role")]
DuplicateStaticRole,
#[error("architecture parameter owner names unknown execution group {0:?}")]
UnknownExecutionGroup(String),
#[error("architecture parameter owner {group}:{global_unit} exceeds {available} units")]
UnitOutOfRange {
group: String,
global_unit: usize,
available: usize,
},
#[error("expected parameter target {target:?} appears in both {first:?} and {second:?}")]
DuplicateExpectedTarget {
target: String,
first: String,
second: String,
},
#[error("parameter target {target:?} is owned by both {first:?} and {second:?}")]
DuplicateOwnership {
target: String,
first: ParameterGroupOwner,
second: ParameterGroupOwner,
},
#[error("parameter target {0:?} has no architecture owner")]
MissingOwnership(String),
#[error("parameter target {0:?} is not present in the authoritative parameter groups")]
UnexpectedOwnership(String),
}
#[derive(Debug, Clone, Copy, Eq, Hash, PartialEq)]
#[non_exhaustive]
pub enum BoundaryTensorDtype {
Activation,
Uint32,
Int32,
}
#[derive(Debug, Clone, Copy, Eq, Hash, PartialEq)]
#[non_exhaustive]
pub enum PipelineActivationDtype {
Float16,
Bfloat16,
Float32,
}
#[derive(Debug, Clone, Copy, Eq, Hash, PartialEq)]
pub struct PipelineWireContract {
activation_dtype: PipelineActivationDtype,
}
impl PipelineWireContract {
pub const fn new(activation_dtype: PipelineActivationDtype) -> Self {
Self { activation_dtype }
}
pub const fn activation_dtype(self) -> PipelineActivationDtype {
self.activation_dtype
}
}
#[derive(Debug, Clone, Copy, Eq, Hash, PartialEq)]
#[non_exhaustive]
pub enum BoundaryTensorDimension {
Batch,
Sequence,
Fixed(i32),
}
#[derive(Debug, Clone, Eq, Hash, PartialEq)]
pub struct BoundaryTensorSpec {
role: String,
shape: Vec<BoundaryTensorDimension>,
dtype: BoundaryTensorDtype,
}
impl BoundaryTensorSpec {
pub fn new(
role: impl Into<String>,
shape: impl IntoIterator<Item = BoundaryTensorDimension>,
dtype: BoundaryTensorDtype,
) -> Self {
Self {
role: role.into(),
shape: shape.into_iter().collect(),
dtype,
}
}
pub fn primary_activation(hidden_size: i32) -> Self {
Self::new(
"hidden",
[
BoundaryTensorDimension::Batch,
BoundaryTensorDimension::Sequence,
BoundaryTensorDimension::Fixed(hidden_size),
],
BoundaryTensorDtype::Activation,
)
}
pub fn role(&self) -> &str {
&self.role
}
pub fn shape(&self) -> &[BoundaryTensorDimension] {
&self.shape
}
pub const fn dtype(&self) -> BoundaryTensorDtype {
self.dtype
}
}
#[derive(Debug, Clone, Eq, Hash, PartialEq)]
pub struct ResolvedBoundaryTensorSpec {
role: String,
shape: Vec<i32>,
dtype: BoundaryTensorDtype,
}
impl ResolvedBoundaryTensorSpec {
pub fn role(&self) -> &str {
&self.role
}
pub fn shape(&self) -> &[i32] {
&self.shape
}
pub const fn dtype(&self) -> BoundaryTensorDtype {
self.dtype
}
}
#[derive(Debug, Clone, Eq, Hash, PartialEq)]
pub struct BoundaryWireSchema {
identity: &'static str,
primary: BoundaryTensorSpec,
auxiliary: Vec<BoundaryTensorSpec>,
}
impl BoundaryWireSchema {
pub fn new(
identity: &'static str,
primary: BoundaryTensorSpec,
auxiliary: impl IntoIterator<Item = BoundaryTensorSpec>,
) -> Result<Self, ArchitectureBoundaryError> {
if identity.trim().is_empty() {
return Err(ArchitectureBoundaryError::EmptyIdentity);
}
if primary.dtype != BoundaryTensorDtype::Activation {
return Err(ArchitectureBoundaryError::InvalidPrimaryDtype { boundary: identity });
}
let auxiliary = auxiliary.into_iter().collect::<Vec<_>>();
let mut roles = BTreeSet::new();
for tensor in std::iter::once(&primary).chain(&auxiliary) {
if tensor.role.trim().is_empty() {
return Err(ArchitectureBoundaryError::EmptyTensorRole { boundary: identity });
}
if !roles.insert(tensor.role.as_str()) {
return Err(ArchitectureBoundaryError::DuplicateTensorRole {
boundary: identity,
role: tensor.role.clone(),
});
}
if tensor.shape.is_empty() {
return Err(ArchitectureBoundaryError::EmptyTensorShape {
boundary: identity,
role: tensor.role.clone(),
});
}
if tensor
.shape
.iter()
.any(|dimension| matches!(dimension, BoundaryTensorDimension::Fixed(value) if *value <= 0))
{
return Err(ArchitectureBoundaryError::InvalidTensorDimension {
boundary: identity,
role: tensor.role.clone(),
});
}
}
Ok(Self {
identity,
primary,
auxiliary,
})
}
pub const fn identity(&self) -> &'static str {
self.identity
}
pub const fn primary(&self) -> &BoundaryTensorSpec {
&self.primary
}
pub fn auxiliary(&self) -> &[BoundaryTensorSpec] {
&self.auxiliary
}
pub fn resolve(
&self,
batch_size: i32,
sequence_length: i32,
) -> Result<ResolvedBoundaryWireSchema, ArchitectureBoundaryError> {
self.resolve_each(
batch_size,
std::iter::repeat_n(sequence_length, 1 + self.auxiliary.len()),
)
}
pub fn resolve_each(
&self,
batch_size: i32,
sequence_lengths: impl IntoIterator<Item = i32>,
) -> Result<ResolvedBoundaryWireSchema, ArchitectureBoundaryError> {
let sequence_lengths = sequence_lengths.into_iter().collect::<Vec<_>>();
if sequence_lengths.len() != 1 + self.auxiliary.len() {
return Err(ArchitectureBoundaryError::TensorCount {
boundary: self.identity,
expected: 1 + self.auxiliary.len(),
actual: sequence_lengths.len(),
});
}
if batch_size <= 0 || sequence_lengths.iter().any(|sequence| *sequence <= 0) {
return Err(ArchitectureBoundaryError::InvalidInvocationGeometry {
boundary: self.identity,
batch_size,
sequence_length: sequence_lengths
.into_iter()
.find(|value| *value <= 0)
.unwrap_or(0),
});
}
let resolve = |tensor: &BoundaryTensorSpec, sequence_length| ResolvedBoundaryTensorSpec {
role: tensor.role.clone(),
shape: tensor
.shape
.iter()
.map(|dimension| match dimension {
BoundaryTensorDimension::Batch => batch_size,
BoundaryTensorDimension::Sequence => sequence_length,
BoundaryTensorDimension::Fixed(value) => *value,
})
.collect(),
dtype: tensor.dtype,
};
let mut sequences = sequence_lengths.into_iter();
Ok(ResolvedBoundaryWireSchema {
identity: self.identity,
primary: resolve(
&self.primary,
sequences.next().expect("validated primary sequence"),
),
auxiliary: self
.auxiliary
.iter()
.zip(sequences)
.map(|(tensor, sequence)| resolve(tensor, sequence))
.collect(),
})
}
}
#[derive(Debug, Clone, Eq, Hash, PartialEq)]
pub struct ResolvedBoundaryWireSchema {
identity: &'static str,
primary: ResolvedBoundaryTensorSpec,
auxiliary: Vec<ResolvedBoundaryTensorSpec>,
}
impl ResolvedBoundaryWireSchema {
pub const fn identity(&self) -> &'static str {
self.identity
}
pub const fn primary(&self) -> &ResolvedBoundaryTensorSpec {
&self.primary
}
pub fn auxiliary(&self) -> &[ResolvedBoundaryTensorSpec] {
&self.auxiliary
}
}
pub trait ArchitectureBoundary: Sized {
type Boundary<T>;
const IDENTITY: &'static str;
fn primary_tensor_spec(&self) -> BoundaryTensorSpec;
fn auxiliary_tensor_specs(&self) -> Vec<BoundaryTensorSpec>;
fn encode<T>(
&self,
boundary: Self::Boundary<T>,
) -> Result<Vec<ArchitectureBoundaryValue<T>>, ArchitectureBoundaryError>;
fn decode<T>(&self, tensors: Vec<T>) -> Result<Self::Boundary<T>, ArchitectureBoundaryError>;
fn wire_schema(&self) -> Result<BoundaryWireSchema, ArchitectureBoundaryError> {
BoundaryWireSchema::new(
Self::IDENTITY,
self.primary_tensor_spec(),
self.auxiliary_tensor_specs(),
)
}
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct ArchitectureBoundaryValue<T> {
role: String,
tensor: T,
}
impl<T> ArchitectureBoundaryValue<T> {
pub fn new(role: impl Into<String>, tensor: T) -> Result<Self, ArchitectureBoundaryError> {
let role = role.into();
if role.trim().is_empty() {
return Err(ArchitectureBoundaryError::EmptyTaggedTensorRole);
}
Ok(Self { role, tensor })
}
pub fn role(&self) -> &str {
&self.role
}
pub const fn tensor(&self) -> &T {
&self.tensor
}
pub fn into_parts(self) -> (String, T) {
(self.role, self.tensor)
}
}
#[derive(Debug, Clone, Copy, Default, Eq, PartialEq)]
pub struct NoAuxiliaryBoundary;
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub struct NoAuxiliaryBoundarySchema {
hidden_size: i32,
}
impl NoAuxiliaryBoundarySchema {
pub const fn new(hidden_size: i32) -> Self {
Self { hidden_size }
}
}
impl ArchitectureBoundary for NoAuxiliaryBoundarySchema {
type Boundary<T> = NoAuxiliaryBoundary;
const IDENTITY: &'static str = "none";
fn primary_tensor_spec(&self) -> BoundaryTensorSpec {
BoundaryTensorSpec::primary_activation(self.hidden_size)
}
fn auxiliary_tensor_specs(&self) -> Vec<BoundaryTensorSpec> {
Vec::new()
}
fn encode<T>(
&self,
_boundary: Self::Boundary<T>,
) -> Result<Vec<ArchitectureBoundaryValue<T>>, ArchitectureBoundaryError> {
Ok(Vec::new())
}
fn decode<T>(&self, tensors: Vec<T>) -> Result<Self::Boundary<T>, ArchitectureBoundaryError> {
validate_boundary_tensor_count(self, &tensors)?;
Ok(NoAuxiliaryBoundary)
}
}
pub fn validate_boundary_tensor_count<B, T>(
boundary: &B,
tensors: &[T],
) -> Result<(), ArchitectureBoundaryError>
where
B: ArchitectureBoundary,
{
let expected = boundary.wire_schema()?.auxiliary().len();
let actual = tensors.len();
if actual != expected {
return Err(ArchitectureBoundaryError::TensorCount {
boundary: B::IDENTITY,
expected,
actual,
});
}
Ok(())
}
#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
#[non_exhaustive]
pub enum ArchitectureBoundaryError {
#[error("architecture boundary identity must not be empty")]
EmptyIdentity,
#[error("architecture boundary value contains an empty tensor role")]
EmptyTaggedTensorRole,
#[error("architecture boundary {boundary:?} primary tensor must use activation dtype")]
InvalidPrimaryDtype {
boundary: &'static str,
},
#[error("architecture boundary {boundary:?} contains an empty tensor role")]
EmptyTensorRole {
boundary: &'static str,
},
#[error("architecture boundary {boundary:?} repeats tensor role {role:?}")]
DuplicateTensorRole {
boundary: &'static str,
role: String,
},
#[error("architecture boundary {boundary:?} tensor {role:?} has no dimensions")]
EmptyTensorShape {
boundary: &'static str,
role: String,
},
#[error("architecture boundary {boundary:?} tensor {role:?} has a non-positive dimension")]
InvalidTensorDimension {
boundary: &'static str,
role: String,
},
#[error(
"architecture boundary {boundary:?} requires positive invocation geometry, got batch {batch_size} and sequence {sequence_length}"
)]
InvalidInvocationGeometry {
boundary: &'static str,
batch_size: i32,
sequence_length: i32,
},
#[error(
"architecture boundary {boundary:?} expected {expected} tensors but received {actual}"
)]
TensorCount {
boundary: &'static str,
expected: usize,
actual: usize,
},
#[error("architecture boundary {boundary:?} is invalid: {detail}")]
Invalid {
boundary: &'static str,
detail: String,
},
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct PartitionOwnership {
input: bool,
output: bool,
static_roles: Vec<String>,
}
impl PartitionOwnership {
pub fn new(
input: bool,
output: bool,
static_roles: impl IntoIterator<Item = impl Into<String>>,
) -> Result<Self, ArchitecturePartitionError> {
let static_roles = static_roles.into_iter().map(Into::into).collect::<Vec<_>>();
let mut unique = BTreeSet::new();
for role in &static_roles {
if role.trim().is_empty() {
return Err(ArchitecturePartitionError::EmptyStaticRole);
}
if !unique.insert(role.clone()) {
return Err(ArchitecturePartitionError::DuplicateStaticRole(
role.clone(),
));
}
}
Ok(Self {
input,
output,
static_roles,
})
}
pub const fn owns_input(&self) -> bool {
self.input
}
pub const fn owns_output(&self) -> bool {
self.output
}
pub fn static_roles(&self) -> &[String] {
&self.static_roles
}
pub fn owns_static_role(&self, role: &str) -> bool {
self.static_roles.iter().any(|candidate| candidate == role)
}
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct PartitionState {
layout: StateLayout,
global_layers: Range<usize>,
}
impl PartitionState {
pub fn new(
layout: StateLayout,
global_layer_offset: usize,
) -> Result<Self, ArchitecturePartitionError> {
let end = global_layer_offset.checked_add(layout.len()).ok_or(
ArchitecturePartitionError::StateOffsetOverflow {
offset: global_layer_offset,
layers: layout.len(),
},
)?;
Ok(Self {
layout,
global_layers: global_layer_offset..end,
})
}
pub const fn layout(&self) -> &StateLayout {
&self.layout
}
pub const fn global_layer_offset(&self) -> usize {
self.global_layers.start
}
pub fn global_layers(&self) -> Range<usize> {
self.global_layers.clone()
}
pub fn prompt_cache_identity<B, M>(
&self,
architecture: &M,
topology: eredu_core::cache::PromptCacheTopology,
) -> Result<eredu_core::cache::PromptCacheModelIdentity, ArchitecturePartitionError>
where
B: eredu_nn::NeuralBackend,
M: crate::ArchitectureParameters<B>,
M::DefinitionError: std::fmt::Display,
{
architecture
.state_identity(self, topology)
.map_err(|error| ArchitecturePartitionError::ArchitectureState(error.to_string()))?
.prompt_cache_identity(self.layout())
.map_err(|error| ArchitecturePartitionError::PromptCacheIdentity(error.to_string()))
}
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct PartitionGroup {
group: ExecutionGroupId,
group_index: usize,
global_units: Range<usize>,
}
impl PartitionGroup {
pub const fn group(&self) -> &ExecutionGroupId {
&self.group
}
pub const fn group_index(&self) -> usize {
self.group_index
}
pub fn global_units(&self) -> Range<usize> {
self.global_units.clone()
}
pub fn contains(&self, global_unit: usize) -> bool {
self.global_units.contains(&global_unit)
}
}
#[derive(Debug, Clone)]
pub struct ArchitecturePartition<G, A> {
graph: ExecutionGraph,
unit_layout: ExecutionUnitLayout,
groups: Vec<PartitionGroup>,
ownership: PartitionOwnership,
state: Option<PartitionState>,
local_geometry: G,
boundary_schema: A,
parameter_bindings: Vec<OwnedParameterGroupSpec>,
}
impl<G, A> ArchitecturePartition<G, A> {
#[allow(clippy::too_many_arguments)]
pub fn from_architecture<B, S, M, N>(
architecture: &M,
group_ranges: impl IntoIterator<Item = (N, Range<usize>)>,
ownership: PartitionOwnership,
local_geometry: G,
boundary_schema: A,
parameters: &ArchitectureParameterDescription,
) -> Result<Self, ArchitecturePartitionError>
where
B: eredu_nn::NeuralBackend,
S: crate::RuntimeState<B>,
M: crate::LayeredArchitecture<B, S>,
M::Error: std::fmt::Display,
N: Into<String>,
A: ArchitectureBoundary,
{
let (graph, unit_layout) = canonical_architecture_layout::<B, S, M>(architecture)?;
boundary_schema.wire_schema()?;
if parameters.graph() != &graph {
return Err(ArchitecturePartitionError::ArchitectureGraphMismatch);
}
if parameters.unit_layout() != &unit_layout {
return Err(ArchitecturePartitionError::ArchitectureUnitLayoutMismatch);
}
let complete_state = architecture
.state_layout()
.map_err(|error| ArchitecturePartitionError::ArchitectureState(error.to_string()))?;
let plan = architecture.state_partition_plan(&complete_state);
let mut partition = Self::new(
graph,
unit_layout,
group_ranges,
ownership,
None,
local_geometry,
boundary_schema,
std::iter::empty(),
)?;
partition.state = partition
.resolve_state_partition(&complete_state, &plan)
.map_err(|error| ArchitecturePartitionError::ArchitectureState(error.to_string()))?;
partition.parameter_bindings = parameters.select_owned(&partition);
Ok(partition)
}
#[allow(clippy::too_many_arguments)]
pub fn from_description<N>(
parameters: &ArchitectureParameterDescription,
group_ranges: impl IntoIterator<Item = (N, Range<usize>)>,
ownership: PartitionOwnership,
complete_state: &StateLayout,
state_plan: &ArchitectureStatePartitionPlan,
local_geometry: G,
boundary_schema: A,
) -> Result<Self, ArchitecturePartitionError>
where
N: Into<String>,
A: ArchitectureBoundary,
{
let graph = parameters.graph().clone();
let unit_layout = parameters.unit_layout().clone();
validate_canonical_layout(&graph, &unit_layout)?;
boundary_schema.wire_schema()?;
let mut partition = Self::new(
graph,
unit_layout,
group_ranges,
ownership,
None,
local_geometry,
boundary_schema,
std::iter::empty(),
)?;
partition.state = partition
.resolve_state_partition(complete_state, state_plan)
.map_err(|error| ArchitecturePartitionError::ArchitectureState(error.to_string()))?;
partition.parameter_bindings = parameters.select_owned(&partition);
Ok(partition)
}
#[allow(clippy::too_many_arguments)]
fn new<S>(
graph: ExecutionGraph,
unit_layout: ExecutionUnitLayout,
group_ranges: impl IntoIterator<Item = (S, Range<usize>)>,
ownership: PartitionOwnership,
state: Option<PartitionState>,
local_geometry: G,
boundary_schema: A,
parameter_bindings: impl IntoIterator<Item = OwnedParameterGroupSpec>,
) -> Result<Self, ArchitecturePartitionError>
where
S: Into<String>,
{
validate_canonical_layout(&graph, &unit_layout)?;
let mut seen_groups = BTreeSet::new();
let mut groups = Vec::new();
for (group, global_units) in group_ranges {
let group = group.into();
let group_index = graph
.groups()
.iter()
.position(|candidate| candidate.id() == group)
.ok_or_else(|| ArchitecturePartitionError::UnknownGroup(group.clone()))?;
if !seen_groups.insert(group.clone()) {
return Err(ArchitecturePartitionError::DuplicateGroup(group));
}
if global_units.is_empty() {
return Err(ArchitecturePartitionError::EmptyGroupRange { group });
}
let available = unit_layout
.group_range(group_index)
.expect("canonical layout contains every graph group")
.len();
if global_units.end > available {
return Err(ArchitecturePartitionError::GroupRangeOutOfBounds {
group,
start: global_units.start,
end: global_units.end,
available,
});
}
groups.push(PartitionGroup {
group: unit_layout
.group_id(group_index)
.expect("canonical layout contains every graph group identity")
.clone(),
group_index,
global_units,
});
}
groups.sort_by_key(PartitionGroup::group_index);
let parameter_bindings = parameter_bindings.into_iter().collect::<Vec<_>>();
let mut targets = BTreeSet::new();
for binding in ¶meter_bindings {
if !binding
.owner()
.is_local_partition_parts(&groups, &ownership)
{
return Err(ArchitecturePartitionError::NonLocalParameterOwner(
binding.owner().clone(),
));
}
for member in binding.members() {
if !targets.insert(member.target().to_owned()) {
return Err(ArchitecturePartitionError::DuplicateParameterTarget(
member.target().to_owned(),
));
}
}
}
Ok(Self {
graph,
unit_layout,
groups,
ownership,
state,
local_geometry,
boundary_schema,
parameter_bindings,
})
}
pub const fn graph(&self) -> &ExecutionGraph {
&self.graph
}
pub const fn unit_layout(&self) -> &ExecutionUnitLayout {
&self.unit_layout
}
pub fn groups(&self) -> &[PartitionGroup] {
&self.groups
}
pub fn units(&self) -> impl Iterator<Item = crate::ExecutionUnitAddress> + '_ {
self.groups.iter().flat_map(move |owned| {
let group = owned.group_index;
let base = self
.unit_layout
.group_range(group)
.expect("partition group belongs to its canonical layout")
.start;
owned.global_units.clone().map(move |index| {
self.unit_layout
.address(base + index)
.expect("partition unit belongs to its canonical layout")
})
})
}
pub fn owns_unit(&self, group: &str, global_unit: usize) -> bool {
self.groups
.iter()
.any(|owned| owned.group.as_str() == group && owned.contains(global_unit))
}
pub const fn ownership(&self) -> &PartitionOwnership {
&self.ownership
}
pub const fn state(&self) -> Option<&PartitionState> {
self.state.as_ref()
}
pub fn prompt_cache_identity<B, M>(
&self,
architecture: &M,
topology: eredu_core::cache::PromptCacheTopology,
) -> Result<eredu_core::cache::PromptCacheModelIdentity, ArchitecturePartitionError>
where
B: eredu_nn::NeuralBackend,
M: crate::ArchitectureParameters<B>,
M::DefinitionError: std::fmt::Display,
{
let state = self
.state()
.ok_or(ArchitecturePartitionError::MissingArchitectureState)?;
state.prompt_cache_identity::<B, M>(architecture, topology)
}
pub fn resolve_state_partition(
&self,
complete: &StateLayout,
plan: &ArchitectureStatePartitionPlan,
) -> Result<Option<PartitionState>, ArchitectureStatePartitionError> {
if plan.rules().is_empty() {
return Err(ArchitectureStatePartitionError::EmptyPlan);
}
let mut rules = plan.rules().iter().collect::<Vec<_>>();
rules.sort_by_key(|rule| rule.layers().start);
let mut frontier = 0usize;
for rule in &rules {
let layers = rule.layers();
if layers.is_empty() {
return Err(ArchitectureStatePartitionError::EmptyRange {
start: layers.start,
end: layers.end,
});
}
if layers.end > complete.len() {
return Err(ArchitectureStatePartitionError::RangeOutOfBounds {
start: layers.start,
end: layers.end,
layers: complete.len(),
});
}
if layers.start < frontier {
return Err(ArchitectureStatePartitionError::OverlappingRange {
start: layers.start,
frontier,
});
}
if layers.start > frontier {
return Err(ArchitectureStatePartitionError::UnassignedLayer { layer: frontier });
}
if let ArchitectureStatePlacement::GroupUnits { group } = rule.placement() {
let units = self
.unit_layout
.group_range(group)
.ok_or(ArchitectureStatePartitionError::UnknownGroup { group })?
.len();
if layers.len() != units {
return Err(ArchitectureStatePartitionError::GroupLengthMismatch {
group,
start: layers.start,
end: layers.end,
units,
});
}
}
frontier = layers.end;
}
if frontier != complete.len() {
return Err(ArchitectureStatePartitionError::UnassignedLayer { layer: frontier });
}
let mut selected = Vec::new();
for rule in plan.rules() {
let layers = rule.layers();
match rule.placement() {
ArchitectureStatePlacement::GroupUnits { group } => {
if let Some(owned) = self
.groups
.iter()
.find(|owned| owned.group_index() == group)
{
let units = owned.global_units();
selected.push(layers.start + units.start..layers.start + units.end);
}
}
ArchitectureStatePlacement::OutputOwner if self.ownership.owns_output() => {
selected.push(layers);
}
ArchitectureStatePlacement::OutputOwner => {}
}
}
if selected.is_empty() {
return Ok(None);
}
selected.sort_by_key(|layers| layers.start);
let start = selected[0].start;
let mut end = selected[0].end;
for layers in selected.iter().skip(1) {
if layers.start != end {
return Err(ArchitectureStatePartitionError::DiscontiguousSelection {
frontier: end,
start: layers.start,
});
}
end = layers.end;
}
let layout = complete
.slice(start..end)
.map_err(|error| ArchitectureStatePartitionError::InvalidLayout(error.to_string()))?;
PartitionState::new(layout, start)
.map(Some)
.map_err(|error| ArchitectureStatePartitionError::InvalidLayout(error.to_string()))
}
pub const fn local_geometry(&self) -> &G {
&self.local_geometry
}
pub const fn boundary_schema(&self) -> &A {
&self.boundary_schema
}
pub fn boundary_schema_mut(&mut self) -> &mut A {
&mut self.boundary_schema
}
pub fn parameter_bindings(&self) -> &[OwnedParameterGroupSpec] {
&self.parameter_bindings
}
pub fn parameter_bindings_for_owner<'a>(
&'a self,
owner: &'a ParameterGroupOwner,
) -> impl Iterator<Item = &'a ParameterGroupSpec> + 'a {
self.parameter_bindings
.iter()
.filter(move |binding| binding.owner() == owner)
.map(OwnedParameterGroupSpec::group)
}
pub fn validate_architecture<B, S, M>(
&self,
architecture: &M,
) -> Result<(), ArchitecturePartitionError>
where
B: eredu_nn::NeuralBackend,
S: crate::RuntimeState<B>,
M: crate::LayeredArchitecture<B, S>,
M::Error: std::fmt::Display,
{
let (graph, unit_layout) = canonical_architecture_layout::<B, S, M>(architecture)?;
if graph != self.graph {
return Err(ArchitecturePartitionError::ArchitectureGraphMismatch);
}
if unit_layout != self.unit_layout {
return Err(ArchitecturePartitionError::ArchitectureUnitLayoutMismatch);
}
Ok(())
}
}
#[derive(Debug, Clone)]
pub struct LayeredPartitionDriver {
group: usize,
range: Range<usize>,
state_layout: Option<StateLayout>,
owns_input: bool,
owns_output: bool,
}
impl LayeredPartitionDriver {
pub fn new<G, A>(
partition: &ArchitecturePartition<G, A>,
group_index: usize,
storage_range: Range<usize>,
) -> Result<Self, LayeredPartitionError> {
Self::new_with_state_ownership(partition, group_index, storage_range, true)
}
pub fn new_with_state_ownership<G, A>(
partition: &ArchitecturePartition<G, A>,
group_index: usize,
storage_range: Range<usize>,
group_owns_state: bool,
) -> Result<Self, LayeredPartitionError> {
let group = partition
.groups()
.iter()
.find(|group| group.group_index() == group_index)
.ok_or(LayeredPartitionError::GroupNotOwned { group: group_index })?;
let range = group.global_units();
if storage_range != range {
return Err(LayeredPartitionError::StorageRange {
storage: storage_range,
partition: range,
});
}
if group_owns_state {
let state = partition
.state()
.ok_or(LayeredPartitionError::MissingState)?;
if state.global_layers().start > range.start || state.global_layers().end < range.end {
return Err(LayeredPartitionError::StateRange {
state: state.global_layers(),
partition: range,
});
}
}
Ok(Self {
group: group.group_index(),
range,
state_layout: group_owns_state
.then(|| partition.state().map(|state| state.layout().clone()))
.flatten(),
owns_input: partition.ownership().owns_input(),
owns_output: partition.ownership().owns_output(),
})
}
pub fn range(&self) -> Range<usize> {
self.range.clone()
}
pub const fn group_index(&self) -> usize {
self.group
}
pub fn state_layout(&self) -> &StateLayout {
self.state_layout
.as_ref()
.expect("state_layout requires a state-owning layered partition driver")
}
pub const fn optional_state_layout(&self) -> Option<&StateLayout> {
self.state_layout.as_ref()
}
pub const fn owns_input(&self) -> bool {
self.owns_input
}
pub const fn owns_output(&self) -> bool {
self.owns_output
}
pub fn input<'a, T, A>(
&self,
input: LayeredPartitionInput<'a, T, A>,
) -> Result<LayeredPartitionInput<'a, T, A>, LayeredPartitionError> {
match (&input, self.owns_input) {
(LayeredPartitionInput::Tokens(_), true)
| (LayeredPartitionInput::Hidden { .. }, _) => Ok(input),
(LayeredPartitionInput::Tokens(_), false) => {
Err(LayeredPartitionError::TokensOnNonInputOwner)
}
}
}
pub fn exchange_boundary<B>(
&self,
value: B::Tensor,
group: &B::Group,
executor: &B::Executor,
) -> Result<B::Tensor, B::CollectiveError>
where
B: crate::CollectiveBackend,
{
B::all_to_all(value, group, executor)
}
#[allow(
clippy::too_many_arguments,
clippy::type_complexity,
reason = "the result preserves the concrete architecture error without erased dispatch"
)]
pub fn begin<'a, B, S, M>(
&self,
architecture: &mut M,
input: LayeredPartitionInput<
'a,
B::Tensor,
<M::Boundary as ArchitectureBoundary>::Boundary<B::Tensor>,
>,
mask: Option<&B::Tensor>,
state: &mut S,
parallel: Option<&B::ParallelContext>,
context: &<B::Tensor as eredu_nn::Tensor>::Context,
) -> Result<
LayeredForwardState<B::Tensor, M::ForwardContext>,
LayeredPartitionBeginError<M::Error>,
>
where
B: eredu_nn::NeuralBackend,
S: RuntimeState<B>,
M: PartitionedLayeredArchitecture<B, S>,
M::Error: std::fmt::Display,
{
let expected = self
.state_layout
.as_ref()
.ok_or(LayeredPartitionBeginError::MissingState { group: self.group })?;
let mut forward = match parallel {
Some(parallel) => architecture
.begin_partition_parallel(input, mask, state, expected, 0, parallel, context),
None => architecture.begin_partition(input, mask, state, expected, 0, context),
}
.map_err(LayeredPartitionBeginError::Architecture)?;
forward.hidden = architecture
.enter_partition_group(
self.group,
&forward.hidden,
state,
&mut forward.context,
parallel,
context,
)
.map_err(LayeredPartitionBeginError::Architecture)?;
Ok(forward)
}
#[allow(
clippy::too_many_arguments,
clippy::type_complexity,
reason = "the signature exposes the backend and architecture boundary types explicitly"
)]
pub fn finish<B, S, M>(
&self,
architecture: &mut M,
hidden: &B::Tensor,
state: &mut S,
forward: &mut M::ForwardContext,
parallel: Option<&B::ParallelContext>,
context: &<B::Tensor as eredu_nn::Tensor>::Context,
) -> Result<
LayeredPartitionOutput<
B::Tensor,
<M::Boundary as ArchitectureBoundary>::Boundary<B::Tensor>,
>,
M::Error,
>
where
B: eredu_nn::NeuralBackend,
S: RuntimeState<B>,
M: PartitionedLayeredArchitecture<B, S>,
{
let hidden = architecture
.leave_partition_group(self.group, hidden, state, forward, parallel, context)?;
architecture.finish_partition(&hidden, state, forward, self.owns_output, parallel, context)
}
}
#[derive(Debug, thiserror::Error)]
pub enum LayeredPartitionBeginError<E>
where
E: std::fmt::Display,
{
#[error("stateless partition group {group} requires an architecture stateless entry strategy")]
MissingState {
group: usize,
},
#[error("partition architecture entry failed: {0}")]
Architecture(E),
}
#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
#[non_exhaustive]
pub enum LayeredPartitionError {
#[error("layered partition does not own execution group {group}")]
GroupNotOwned {
group: usize,
},
#[error("partition storage range {storage:?} disagrees with canonical range {partition:?}")]
StorageRange {
storage: Range<usize>,
partition: Range<usize>,
},
#[error("layered partition has no runtime state")]
MissingState,
#[error("partition state range {state:?} disagrees with canonical range {partition:?}")]
StateRange {
state: Range<usize>,
partition: Range<usize>,
},
#[error("non-input partition received token ids")]
TokensOnNonInputOwner,
}
fn canonical_architecture_layout<B, S, M>(
architecture: &M,
) -> Result<(ExecutionGraph, ExecutionUnitLayout), ArchitecturePartitionError>
where
B: eredu_nn::NeuralBackend,
S: crate::RuntimeState<B>,
M: crate::LayeredArchitecture<B, S>,
M::Error: std::fmt::Display,
{
let graph = architecture
.execution_graph()
.map_err(|error| ArchitecturePartitionError::ArchitectureTopology(error.to_string()))?;
let primary = architecture.primary_execution_group();
let primary_index = graph.group_index(primary).ok_or_else(|| {
ArchitecturePartitionError::ArchitectureTopology(format!(
"primary execution group {primary:?} is not present in the canonical graph"
))
})?;
let primary_transport = architecture.group_transport(primary_index);
if primary_transport.kind != crate::ArchitectureGroupKind::Decoder
|| primary_transport.placement != crate::ArchitectureGroupPlacement::Pipeline
{
return Err(ArchitecturePartitionError::ArchitectureTopology(format!(
"primary execution group {primary:?} must be a pipeline decoder"
)));
}
let mut declared_groups = BTreeSet::from([primary.to_owned()]);
for prediction in architecture.prediction_execution_groups() {
let prediction_index = graph.group_index(&prediction).ok_or_else(|| {
ArchitecturePartitionError::ArchitectureTopology(format!(
"prediction execution group {prediction:?} is not present in the canonical graph"
))
})?;
let prediction_transport = architecture.group_transport(prediction_index);
if prediction_transport.kind != crate::ArchitectureGroupKind::Prediction
|| prediction_transport.placement != crate::ArchitectureGroupPlacement::OutputOwner
{
return Err(ArchitecturePartitionError::ArchitectureTopology(format!(
"prediction execution group {prediction:?} must be an output-owner prediction"
)));
}
if !declared_groups.insert(prediction.clone()) {
return Err(ArchitecturePartitionError::ArchitectureTopology(format!(
"execution group {prediction:?} is declared as a primary or prediction group more than once"
)));
}
}
let mut counts = Vec::with_capacity(graph.groups().len());
let mut paths = BTreeSet::new();
for group in 0..graph.groups().len() {
let count = architecture
.group_unit_count(group)
.map_err(|error| ArchitecturePartitionError::ArchitectureTopology(error.to_string()))?;
counts.push(count);
for index in 0..count {
let path = architecture.unit_path(group, index).map_err(|error| {
ArchitecturePartitionError::ArchitectureTopology(error.to_string())
})?;
if path.trim().is_empty() {
return Err(ArchitecturePartitionError::EmptyArchitectureUnitPath { group, index });
}
if !paths.insert(path.clone()) {
return Err(ArchitecturePartitionError::DuplicateArchitectureUnitPath(
path,
));
}
}
}
let unit_layout = ExecutionUnitLayout::new(&graph, counts)
.map_err(|error| ArchitecturePartitionError::ArchitectureTopology(error.to_string()))?;
Ok((graph, unit_layout))
}
fn validate_canonical_layout(
graph: &ExecutionGraph,
layout: &ExecutionUnitLayout,
) -> Result<(), ArchitecturePartitionError> {
if graph.groups().len() != layout.group_count() {
return Err(ArchitecturePartitionError::LayoutGroupCountMismatch {
graph: graph.groups().len(),
layout: layout.group_count(),
});
}
for (index, group) in graph.groups().iter().enumerate() {
let layout_group = layout
.group_id(index)
.expect("matching group counts provide every layout identity");
if layout_group.as_str() != group.id() {
return Err(ArchitecturePartitionError::LayoutGroupMismatch {
index,
graph: group.id().to_owned(),
layout: layout_group.as_str().to_owned(),
});
}
}
Ok(())
}
#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
#[non_exhaustive]
pub enum ArchitecturePartitionError {
#[error("invalid architecture partition boundary: {0}")]
InvalidBoundary(#[from] ArchitectureBoundaryError),
#[error("neutral architecture topology is invalid: {0}")]
ArchitectureTopology(String),
#[error("neutral architecture state is invalid: {0}")]
ArchitectureState(String),
#[error("architecture partition owns no mutable state")]
MissingArchitectureState,
#[error("architecture prompt-cache identity is invalid: {0}")]
PromptCacheIdentity(String),
#[error("neutral architecture unit {group}:{index} has an empty path")]
EmptyArchitectureUnitPath {
group: usize,
index: usize,
},
#[error("neutral architecture repeats unit path {0:?}")]
DuplicateArchitectureUnitPath(String),
#[error("architecture partition dependency graph differs from the neutral architecture")]
ArchitectureGraphMismatch,
#[error("architecture partition unit layout differs from the neutral architecture")]
ArchitectureUnitLayoutMismatch,
#[error("execution graph contains {graph} groups but its unit layout contains {layout}")]
LayoutGroupCountMismatch {
graph: usize,
layout: usize,
},
#[error("execution group {index} is {graph:?} in the graph but {layout:?} in the unit layout")]
LayoutGroupMismatch {
index: usize,
graph: String,
layout: String,
},
#[error("architecture partition names unknown execution group {0:?}")]
UnknownGroup(String),
#[error("architecture partition repeats execution group {0:?}")]
DuplicateGroup(String),
#[error("architecture partition declares an empty unit range for group {group:?}")]
EmptyGroupRange {
group: String,
},
#[error(
"architecture partition range {start}..{end} for group {group:?} exceeds {available} units"
)]
GroupRangeOutOfBounds {
group: String,
start: usize,
end: usize,
available: usize,
},
#[error("architecture partition static role must not be empty")]
EmptyStaticRole,
#[error("architecture partition repeats static role {0:?}")]
DuplicateStaticRole(String),
#[error("state layer offset {offset} plus {layers} local layers overflowed usize")]
StateOffsetOverflow {
offset: usize,
layers: usize,
},
#[error("architecture partition repeats parameter target {0:?}")]
DuplicateParameterTarget(String),
#[error("architecture partition includes non-local parameter owner {0:?}")]
NonLocalParameterOwner(ParameterGroupOwner),
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{MemberSharding, ParameterMemberSpec, ParameterRole};
use eredu_core::{cache::LayerCachePolicy, LayerSchedule};
#[derive(Debug, Clone, Eq, PartialEq)]
struct Geometry(&'static str);
#[derive(Debug, Clone, Eq, PartialEq)]
struct Boundary {
route: usize,
}
#[derive(Debug, Clone, Eq, PartialEq)]
struct PairBoundary<T> {
tokens: T,
embedded: T,
}
#[derive(Debug, Clone, Copy)]
struct PairBoundarySchema;
impl ArchitectureBoundary for PairBoundarySchema {
type Boundary<T> = PairBoundary<T>;
const IDENTITY: &'static str = "fixture.target";
fn primary_tensor_spec(&self) -> BoundaryTensorSpec {
BoundaryTensorSpec::primary_activation(8)
}
fn auxiliary_tensor_specs(&self) -> Vec<BoundaryTensorSpec> {
vec![
BoundaryTensorSpec::new(
"tokens",
[
BoundaryTensorDimension::Batch,
BoundaryTensorDimension::Sequence,
],
BoundaryTensorDtype::Uint32,
),
BoundaryTensorSpec::new(
"embedded",
[
BoundaryTensorDimension::Batch,
BoundaryTensorDimension::Sequence,
BoundaryTensorDimension::Fixed(16),
],
BoundaryTensorDtype::Activation,
),
]
}
fn encode<T>(
&self,
boundary: Self::Boundary<T>,
) -> Result<Vec<ArchitectureBoundaryValue<T>>, ArchitectureBoundaryError> {
Ok(vec![
ArchitectureBoundaryValue::new("tokens", boundary.tokens)?,
ArchitectureBoundaryValue::new("embedded", boundary.embedded)?,
])
}
fn decode<T>(
&self,
mut tensors: Vec<T>,
) -> Result<Self::Boundary<T>, ArchitectureBoundaryError> {
validate_boundary_tensor_count(self, &tensors)?;
let embedded = tensors.pop().expect("validated embedded tensor");
let tokens = tensors.pop().expect("validated token tensor");
Ok(PairBoundary { tokens, embedded })
}
}
fn graph() -> ExecutionGraph {
ExecutionGraph::chain(["primary", "prediction"]).unwrap()
}
fn layout(graph: &ExecutionGraph) -> ExecutionUnitLayout {
ExecutionUnitLayout::new(graph, [4, 3]).unwrap()
}
fn state_layout(layers: usize) -> StateLayout {
StateLayout::new(
LayerSchedule::new(layers, vec![LayerCachePolicy::NoState; layers]).unwrap(),
)
.unwrap()
}
fn parameter(logical: &str, target: &str) -> ParameterGroupSpec {
ParameterGroupSpec::new(
logical,
ParameterRole::Replicated,
[ParameterMemberSpec::new(
target,
vec![2, 2],
MemberSharding::Replicated,
)],
)
.unwrap()
}
fn valid_partition() -> ArchitecturePartition<Geometry, Boundary> {
let graph = graph();
let layout = layout(&graph);
ArchitecturePartition::new(
graph,
layout,
[("prediction", 0..2), ("primary", 1..4)],
PartitionOwnership::new(true, false, ["embedding", "normalization"]).unwrap(),
Some(PartitionState::new(state_layout(2), 7).unwrap()),
Geometry("local"),
Boundary { route: 3 },
[
OwnedParameterGroupSpec::new(
ParameterGroupOwner::static_role("embedding"),
parameter("model.embed_tokens", "model.embed_tokens.weight"),
),
OwnedParameterGroupSpec::new(
ParameterGroupOwner::execution_unit(
ExecutionGroupId::new("primary").unwrap(),
1,
),
parameter("model.layers.1", "model.layers.1.weight"),
),
],
)
.unwrap()
}
fn state_plan_partition(
primary: Range<usize>,
ownership: PartitionOwnership,
) -> ArchitecturePartition<(), ()> {
let graph = graph();
ArchitecturePartition::new(
graph.clone(),
layout(&graph),
[("primary", primary)],
ownership,
None,
(),
(),
[],
)
.unwrap()
}
#[test]
fn architecture_state_plan_attaches_declared_tail_to_output_owner() {
let complete = state_layout(6);
let plan = ArchitectureStatePartitionPlan::new([
crate::ArchitectureStatePartitionRule::group_units(0, 0..4),
crate::ArchitectureStatePartitionRule::output_owner(4..6),
]);
let interior = state_plan_partition(
1..3,
PartitionOwnership::new(false, false, std::iter::empty::<&str>()).unwrap(),
);
let output = state_plan_partition(
3..4,
PartitionOwnership::new(false, true, std::iter::empty::<&str>()).unwrap(),
);
assert_eq!(
interior
.resolve_state_partition(&complete, &plan)
.unwrap()
.unwrap()
.global_layers(),
1..3
);
assert_eq!(
output
.resolve_state_partition(&complete, &plan)
.unwrap()
.unwrap()
.global_layers(),
3..6
);
}
#[test]
fn architecture_state_plan_rejects_noncontiguous_local_state() {
let complete = state_layout(6);
let plan = ArchitectureStatePartitionPlan::new([
crate::ArchitectureStatePartitionRule::output_owner(0..2),
crate::ArchitectureStatePartitionRule::group_units(0, 2..6),
]);
let output = state_plan_partition(
3..4,
PartitionOwnership::new(false, true, std::iter::empty::<&str>()).unwrap(),
);
assert_eq!(
output.resolve_state_partition(&complete, &plan),
Err(ArchitectureStatePartitionError::DiscontiguousSelection {
frontier: 2,
start: 5,
})
);
}
fn parameter_description(
expected: Vec<ParameterGroupSpec>,
groups: Vec<OwnedParameterGroupSpec>,
) -> Result<ArchitectureParameterDescription, ArchitectureParameterError> {
let graph = graph();
ArchitectureParameterDescription::new(&graph, &layout(&graph), expected, groups)
}
#[test]
fn description_driven_partition_selects_state_and_parameters_before_construction() {
let embedding = parameter("embedding", "model.embed_tokens.weight");
let layer = parameter("layer", "model.layers.1.weight");
let description = parameter_description(
vec![embedding.clone(), layer.clone()],
vec![
OwnedParameterGroupSpec::new(
ParameterGroupOwner::static_role("embedding"),
embedding,
),
OwnedParameterGroupSpec::new(
ParameterGroupOwner::execution_unit(
ExecutionGroupId::new("primary").unwrap(),
1,
),
layer,
),
],
)
.unwrap();
let ownership = PartitionOwnership::new(true, false, ["embedding"]).unwrap();
let state = state_layout(4);
let state_plan = ArchitectureStatePartitionPlan::new([
crate::ArchitectureStatePartitionRule::group_units(0, 0..4),
]);
let partition = ArchitecturePartition::from_description(
&description,
[("primary", 1..3)],
ownership,
&state,
&state_plan,
Geometry("selected-before-allocation"),
PairBoundarySchema,
)
.unwrap();
assert_eq!(partition.groups()[0].global_units(), 1..3);
assert_eq!(partition.state().unwrap().global_layers(), 1..3);
assert_eq!(
partition.local_geometry(),
&Geometry("selected-before-allocation")
);
assert_eq!(partition.parameter_bindings().len(), 2);
assert_eq!(
partition
.parameter_bindings()
.iter()
.flat_map(|group| group.members())
.map(ParameterMemberSpec::target)
.collect::<Vec<_>>(),
["model.embed_tokens.weight", "model.layers.1.weight"]
);
}
#[test]
fn parameter_description_selects_static_roles_and_canonical_units() {
let embedding = parameter("embedding", "model.embed_tokens.weight");
let layer = parameter("layer", "model.layers.1.weight");
let description = parameter_description(
vec![embedding.clone(), layer.clone()],
vec![
OwnedParameterGroupSpec::new(
ParameterGroupOwner::static_role("embedding"),
embedding,
),
OwnedParameterGroupSpec::new(
ParameterGroupOwner::execution_unit(
ExecutionGroupId::new("primary").unwrap(),
1,
),
layer,
),
],
)
.unwrap();
let partition = valid_partition();
assert_eq!(description.graph(), partition.graph());
assert_eq!(description.unit_layout(), partition.unit_layout());
let selected = description.select_owned(&partition);
assert_eq!(selected.len(), 2);
assert_eq!(selected[0].logical_name(), "embedding");
assert_eq!(selected[1].logical_name(), "layer");
assert_eq!(
selected[0].owner(),
&ParameterGroupOwner::static_role("embedding")
);
assert_eq!(
selected[1].owner(),
&ParameterGroupOwner::execution_unit(ExecutionGroupId::new("primary").unwrap(), 1,)
);
}
#[test]
fn parameter_description_selects_every_owned_target_for_a_role() {
let expert = ParameterGroupSpec::new(
"model.layers.1.expert_intermediate",
ParameterRole::ExpertIntermediate,
[
ParameterMemberSpec::new(
"model.layers.1.moe.packed.weight",
vec![4, 2],
MemberSharding::Replicated,
),
ParameterMemberSpec::new(
"model.layers.1.moe.packed.scales",
vec![4, 1],
MemberSharding::Replicated,
),
ParameterMemberSpec::new(
"model.layers.1.moe.alias.biases",
vec![4, 1],
MemberSharding::Replicated,
),
],
)
.unwrap();
let replicated = parameter("router", "model.layers.1.moe.router.weight");
let owner =
ParameterGroupOwner::execution_unit(ExecutionGroupId::new("primary").unwrap(), 1);
let description = parameter_description(
vec![expert.clone(), replicated.clone()],
vec![
OwnedParameterGroupSpec::new(owner.clone(), expert),
OwnedParameterGroupSpec::new(owner, replicated),
],
)
.unwrap();
assert_eq!(
description.targets_for_role(ParameterRole::ExpertIntermediate),
BTreeSet::from([
"model.layers.1.moe.alias.biases".to_owned(),
"model.layers.1.moe.packed.scales".to_owned(),
"model.layers.1.moe.packed.weight".to_owned(),
])
);
}
#[test]
fn parameter_description_selects_shared_static_owner_by_any_consumer() {
let embedding = parameter("embedding", "model.embed_tokens.weight");
let description = parameter_description(
vec![embedding.clone()],
vec![OwnedParameterGroupSpec::new(
ParameterGroupOwner::static_any_of(["output", "embedding"]),
embedding,
)],
)
.unwrap();
assert_eq!(description.select_owned(&valid_partition()).len(), 1);
let duplicate = parameter("embedding", "model.embed_tokens.weight");
assert_eq!(
parameter_description(
vec![duplicate.clone()],
vec![OwnedParameterGroupSpec::new(
ParameterGroupOwner::static_any_of(["embedding", "embedding"]),
duplicate,
)],
)
.unwrap_err(),
ArchitectureParameterError::DuplicateStaticRole,
);
}
#[test]
fn partition_rejects_parameter_owner_outside_local_unit_ranges() {
let graph = graph();
let error = ArchitecturePartition::new(
graph.clone(),
layout(&graph),
[("primary", 1..4)],
PartitionOwnership::new(false, false, ["embedding"]).unwrap(),
None,
(),
(),
[OwnedParameterGroupSpec::new(
ParameterGroupOwner::execution_unit(
ExecutionGroupId::new("prediction").unwrap(),
0,
),
parameter("prediction", "prediction.weight"),
)],
)
.unwrap_err();
assert!(matches!(
error,
ArchitecturePartitionError::NonLocalParameterOwner(
ParameterGroupOwner::ExecutionUnit { .. }
)
));
}
#[test]
fn parameter_description_rejects_missing_duplicate_and_out_of_range_ownership() {
let embedding = parameter("embedding", "model.embed_tokens.weight");
let layer = parameter("layer", "model.layers.1.weight");
assert_eq!(
parameter_description(
vec![embedding.clone(), layer.clone()],
vec![OwnedParameterGroupSpec::new(
ParameterGroupOwner::static_role("embedding"),
embedding.clone(),
)],
)
.unwrap_err(),
ArchitectureParameterError::MissingOwnership("model.layers.1.weight".into())
);
assert!(matches!(
parameter_description(
vec![embedding.clone()],
vec![
OwnedParameterGroupSpec::new(
ParameterGroupOwner::static_role("embedding"),
embedding.clone(),
),
OwnedParameterGroupSpec::new(
ParameterGroupOwner::static_role("output"),
embedding.clone(),
),
],
)
.unwrap_err(),
ArchitectureParameterError::DuplicateOwnership { .. }
));
assert_eq!(
parameter_description(
vec![layer.clone()],
vec![OwnedParameterGroupSpec::new(
ParameterGroupOwner::execution_unit(
ExecutionGroupId::new("prediction").unwrap(),
3,
),
layer,
)],
)
.unwrap_err(),
ArchitectureParameterError::UnitOutOfRange {
group: "prediction".into(),
global_unit: 3,
available: 3,
}
);
}
#[test]
fn retains_canonical_topology_ownership_and_typed_family_values() {
let mut partition = valid_partition();
assert_eq!(partition.graph().groups().len(), 2);
assert_eq!(partition.unit_layout().len(), 7);
assert_eq!(partition.groups()[0].group().as_str(), "primary");
assert_eq!(partition.groups()[0].group_index(), 0);
assert_eq!(partition.groups()[0].global_units(), 1..4);
assert!(partition.owns_unit("primary", 3));
assert!(!partition.owns_unit("primary", 0));
assert!(partition.ownership().owns_input());
assert!(!partition.ownership().owns_output());
assert!(partition.ownership().owns_static_role("embedding"));
assert_eq!(
partition
.units()
.map(|unit| (unit.group(), unit.index()))
.collect::<Vec<_>>(),
[(0, 1), (0, 2), (0, 3), (1, 0), (1, 1)]
);
assert_eq!(partition.state().unwrap().global_layers(), 7..9);
assert_eq!(partition.local_geometry(), &Geometry("local"));
partition.boundary_schema_mut().route = 5;
assert_eq!(partition.boundary_schema().route, 5);
assert_eq!(partition.parameter_bindings().len(), 2);
}
#[test]
fn typed_boundary_owns_roles_order_and_atomic_cardinality_validation() {
let boundary = PairBoundary {
tokens: 3,
embedded: 7,
};
let schema = PairBoundarySchema;
let values = schema.encode(boundary).unwrap();
assert_eq!(values[0].role(), "tokens");
assert_eq!(values[1].role(), "embedded");
let tensors = values
.into_iter()
.map(ArchitectureBoundaryValue::into_parts)
.map(|(_, tensor)| tensor)
.collect();
assert_eq!(
schema.decode(tensors).unwrap(),
PairBoundary {
tokens: 3,
embedded: 7
}
);
let resolved = schema.wire_schema().unwrap().resolve(2, 3).unwrap();
assert_eq!(resolved.primary().shape(), [2, 3, 8]);
assert_eq!(resolved.primary().dtype(), BoundaryTensorDtype::Activation);
assert_eq!(resolved.auxiliary()[0].shape(), [2, 3]);
assert_eq!(resolved.auxiliary()[0].dtype(), BoundaryTensorDtype::Uint32);
assert_eq!(resolved.auxiliary()[1].shape(), [2, 3, 16]);
assert_eq!(
resolved.auxiliary()[1].dtype(),
BoundaryTensorDtype::Activation
);
assert_eq!(
schema.decode(vec![3]).unwrap_err(),
ArchitectureBoundaryError::TensorCount {
boundary: "fixture.target",
expected: 2,
actual: 1,
}
);
}
#[test]
fn boundary_schema_rejects_role_and_geometry_drift_before_transport() {
let invalid_primary = BoundaryWireSchema::new(
"fixture.invalid",
BoundaryTensorSpec::new(
"hidden",
[BoundaryTensorDimension::Fixed(8)],
BoundaryTensorDtype::Uint32,
),
[],
)
.unwrap_err();
assert_eq!(
invalid_primary,
ArchitectureBoundaryError::InvalidPrimaryDtype {
boundary: "fixture.invalid",
}
);
let duplicate = BoundaryWireSchema::new(
"fixture.invalid",
BoundaryTensorSpec::primary_activation(8),
[
BoundaryTensorSpec::new(
"state",
[BoundaryTensorDimension::Fixed(1)],
BoundaryTensorDtype::Activation,
),
BoundaryTensorSpec::new(
"state",
[BoundaryTensorDimension::Fixed(2)],
BoundaryTensorDtype::Activation,
),
],
)
.unwrap_err();
assert_eq!(
duplicate,
ArchitectureBoundaryError::DuplicateTensorRole {
boundary: "fixture.invalid",
role: "state".into(),
}
);
let invalid = BoundaryWireSchema::new(
"fixture.invalid",
BoundaryTensorSpec::primary_activation(8),
[BoundaryTensorSpec::new(
"state",
[BoundaryTensorDimension::Fixed(0)],
BoundaryTensorDtype::Activation,
)],
)
.unwrap_err();
assert_eq!(
invalid,
ArchitectureBoundaryError::InvalidTensorDimension {
boundary: "fixture.invalid",
role: "state".into(),
}
);
}
#[test]
fn rejects_noncanonical_unknown_and_duplicate_groups() {
let graph = graph();
let mismatched_graph = ExecutionGraph::chain(["primary", "other"]).unwrap();
let error = ArchitecturePartition::new(
graph.clone(),
layout(&mismatched_graph),
[("primary", 0..1)],
PartitionOwnership::new(false, false, std::iter::empty::<String>()).unwrap(),
None,
(),
(),
std::iter::empty(),
)
.unwrap_err();
assert!(matches!(
error,
ArchitecturePartitionError::LayoutGroupMismatch { .. }
));
let error = ArchitecturePartition::new(
graph.clone(),
layout(&graph),
[("missing", 0..1)],
PartitionOwnership::new(false, false, std::iter::empty::<String>()).unwrap(),
None,
(),
(),
std::iter::empty(),
)
.unwrap_err();
assert_eq!(
error,
ArchitecturePartitionError::UnknownGroup("missing".into())
);
let error = ArchitecturePartition::new(
graph.clone(),
layout(&graph),
[("primary", 0..1), ("primary", 1..2)],
PartitionOwnership::new(false, false, std::iter::empty::<String>()).unwrap(),
None,
(),
(),
std::iter::empty(),
)
.unwrap_err();
assert_eq!(
error,
ArchitecturePartitionError::DuplicateGroup("primary".into())
);
}
#[test]
fn rejects_empty_and_out_of_bounds_group_ranges() {
let graph = graph();
let error = ArchitecturePartition::new(
graph.clone(),
layout(&graph),
[("primary", 2..2)],
PartitionOwnership::new(false, false, std::iter::empty::<String>()).unwrap(),
None,
(),
(),
std::iter::empty(),
)
.unwrap_err();
assert!(matches!(
error,
ArchitecturePartitionError::EmptyGroupRange { .. }
));
let error = ArchitecturePartition::new(
graph.clone(),
layout(&graph),
[("prediction", 1..4)],
PartitionOwnership::new(false, false, std::iter::empty::<String>()).unwrap(),
None,
(),
(),
std::iter::empty(),
)
.unwrap_err();
assert!(matches!(
error,
ArchitecturePartitionError::GroupRangeOutOfBounds { .. }
));
}
#[test]
fn rejects_state_offset_overflow() {
assert_eq!(
PartitionState::new(state_layout(2), usize::MAX).unwrap_err(),
ArchitecturePartitionError::StateOffsetOverflow {
offset: usize::MAX,
layers: 2,
}
);
}
#[test]
fn rejects_empty_static_roles_and_duplicate_parameter_targets() {
assert_eq!(
PartitionOwnership::new(false, false, [" "]).unwrap_err(),
ArchitecturePartitionError::EmptyStaticRole
);
let graph = graph();
let error = ArchitecturePartition::new(
graph.clone(),
layout(&graph),
[("primary", 0..1)],
PartitionOwnership::new(false, false, ["embedding", "normalization"]).unwrap(),
None,
(),
(),
[
OwnedParameterGroupSpec::new(
ParameterGroupOwner::static_role("embedding"),
parameter("first", "shared.weight"),
),
OwnedParameterGroupSpec::new(
ParameterGroupOwner::static_role("normalization"),
parameter("second", "shared.weight"),
),
],
)
.unwrap_err();
assert_eq!(
error,
ArchitecturePartitionError::DuplicateParameterTarget("shared.weight".into())
);
}
fn layered_partition(
storage_state: Range<usize>,
owns_input: bool,
) -> ArchitecturePartition<(), ()> {
let graph = ExecutionGraph::chain(["decoder"]).unwrap();
let layout = ExecutionUnitLayout::new(&graph, [4]).unwrap();
ArchitecturePartition::new(
graph,
layout,
[("decoder", 1..3)],
PartitionOwnership::new(owns_input, false, std::iter::empty::<String>()).unwrap(),
Some(
PartitionState::new(state_layout(storage_state.len()), storage_state.start)
.unwrap(),
),
(),
(),
std::iter::empty(),
)
.unwrap()
}
#[test]
fn layered_driver_rejects_storage_and_state_range_drift() {
let partition = layered_partition(1..3, true);
assert!(LayeredPartitionDriver::new(&partition, 0, 1..3).is_ok());
assert_eq!(
LayeredPartitionDriver::new(&partition, 0, 0..2).unwrap_err(),
LayeredPartitionError::StorageRange {
storage: 0..2,
partition: 1..3,
}
);
let partition = layered_partition(0..2, true);
assert_eq!(
LayeredPartitionDriver::new(&partition, 0, 1..3).unwrap_err(),
LayeredPartitionError::StateRange {
state: 0..2,
partition: 1..3,
}
);
}
#[test]
fn layered_driver_represents_stateless_root_without_borrowing_decoder_state() {
let graph = ExecutionGraph::chain(["vision", "decoder"]).unwrap();
let layout = ExecutionUnitLayout::new(&graph, [1, 2]).unwrap();
let partition = ArchitecturePartition::new(
graph,
layout,
[("vision", 0..1), ("decoder", 0..2)],
PartitionOwnership::new(true, true, std::iter::empty::<String>()).unwrap(),
Some(PartitionState::new(state_layout(1), 1).unwrap()),
(),
(),
std::iter::empty(),
)
.unwrap();
let vision =
LayeredPartitionDriver::new_with_state_ownership(&partition, 0, 0..1, false).unwrap();
assert!(vision.optional_state_layout().is_none());
assert_eq!(vision.group_index(), 0);
let without_state = ArchitecturePartition::new(
ExecutionGraph::chain(["vision"]).unwrap(),
ExecutionUnitLayout::new(&ExecutionGraph::chain(["vision"]).unwrap(), [1]).unwrap(),
[("vision", 0..1)],
PartitionOwnership::new(true, false, std::iter::empty::<String>()).unwrap(),
None,
(),
(),
std::iter::empty(),
)
.unwrap();
assert_eq!(
LayeredPartitionDriver::new(&without_state, 0, 0..1).unwrap_err(),
LayeredPartitionError::MissingState
);
assert!(
LayeredPartitionDriver::new_with_state_ownership(&without_state, 0, 0..1, false)
.unwrap()
.optional_state_layout()
.is_none()
);
}
#[test]
fn layered_driver_restricts_tokens_but_accepts_architecture_prepared_hidden() {
let input_owner =
LayeredPartitionDriver::new(&layered_partition(1..3, true), 0, 1..3).unwrap();
assert!(matches!(
input_owner.input(LayeredPartitionInput::<i32, NoAuxiliaryBoundary>::Tokens(
&7
)),
Ok(LayeredPartitionInput::Tokens(7))
));
assert!(matches!(
input_owner.input(LayeredPartitionInput::Hidden {
hidden: 7,
auxiliary: NoAuxiliaryBoundary,
}),
Ok(LayeredPartitionInput::Hidden {
hidden: 7,
auxiliary: NoAuxiliaryBoundary,
})
));
let hidden_owner =
LayeredPartitionDriver::new(&layered_partition(1..3, false), 0, 1..3).unwrap();
assert_eq!(
hidden_owner
.input(LayeredPartitionInput::<i32, NoAuxiliaryBoundary>::Tokens(
&7
))
.unwrap_err(),
LayeredPartitionError::TokensOnNonInputOwner
);
assert!(matches!(
hidden_owner.input(LayeredPartitionInput::Hidden {
hidden: 7,
auxiliary: NoAuxiliaryBoundary,
}),
Ok(LayeredPartitionInput::Hidden {
hidden: 7,
auxiliary: NoAuxiliaryBoundary,
})
));
}
}