use crate::arrow::{TensorDtype, TensorMetadata};
use ipfrs_core::Cid;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use super::GradientError;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SparseGradient {
pub indices: Vec<usize>,
pub values: Vec<f32>,
pub shape: Vec<usize>,
pub metadata: TensorMetadata,
}
impl SparseGradient {
pub fn new(indices: Vec<usize>, values: Vec<f32>, shape: Vec<usize>) -> Self {
let metadata = TensorMetadata {
name: "sparse_gradient".to_string(),
shape: shape.clone(),
dtype: TensorDtype::Float32,
strides: None,
custom: HashMap::new(),
};
Self {
indices,
values,
shape,
metadata,
}
}
pub fn nnz(&self) -> usize {
self.indices.len()
}
pub fn total_elements(&self) -> usize {
self.shape.iter().product()
}
pub fn sparsity_ratio(&self) -> f32 {
1.0 - (self.nnz() as f32 / self.total_elements() as f32)
}
pub fn to_dense(&self) -> Vec<f32> {
let total = self.total_elements();
let mut dense = vec![0.0; total];
for (&idx, &val) in self.indices.iter().zip(&self.values) {
if idx < total {
dense[idx] = val;
}
}
dense
}
pub fn verify_shape(&self) -> Result<(), GradientError> {
let total = self.total_elements();
for &idx in &self.indices {
if idx >= total {
return Err(GradientError::InvalidGradient(format!(
"Index {} out of bounds for shape {:?}",
idx, self.shape
)));
}
}
if self.indices.len() != self.values.len() {
return Err(GradientError::InvalidGradient(format!(
"Indices length {} != values length {}",
self.indices.len(),
self.values.len()
)));
}
Ok(())
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QuantizedGradient {
pub quantized_values: Vec<i8>,
pub scale: f32,
pub min_val: f32,
pub shape: Vec<usize>,
pub metadata: TensorMetadata,
}
impl QuantizedGradient {
pub fn from_dense(values: &[f32], shape: Vec<usize>) -> Self {
let (quantized_values, scale, min_val) = Self::quantize_i8(values);
let metadata = TensorMetadata {
name: "quantized_gradient".to_string(),
shape: shape.clone(),
dtype: TensorDtype::Int8,
strides: None,
custom: HashMap::new(),
};
Self {
quantized_values,
scale,
min_val,
shape,
metadata,
}
}
fn quantize_i8(values: &[f32]) -> (Vec<i8>, f32, f32) {
if values.is_empty() {
return (Vec::new(), 1.0, 0.0);
}
let min_val = values.iter().copied().fold(f32::INFINITY, f32::min);
let max_val = values.iter().copied().fold(f32::NEG_INFINITY, f32::max);
let scale = if (max_val - min_val).abs() < 1e-8 {
1.0
} else {
(max_val - min_val) / 255.0
};
let quantized = values
.iter()
.map(|&v| {
let normalized = (v - min_val) / scale;
(normalized - 128.0).round().clamp(-128.0, 127.0) as i8
})
.collect();
(quantized, scale, min_val)
}
pub fn to_dense(&self) -> Vec<f32> {
self.quantized_values
.iter()
.map(|&q| {
let normalized = (q as f32) + 128.0;
normalized * self.scale + self.min_val
})
.collect()
}
pub fn compression_ratio(&self) -> f32 {
let original_size = self.quantized_values.len() * 4;
let compressed_size = self.quantized_values.len() + 8; original_size as f32 / compressed_size as f32
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum LayerGradient {
Dense { values: Vec<f32>, shape: Vec<usize> },
Sparse(SparseGradient),
Quantized(QuantizedGradient),
}
impl LayerGradient {
pub fn shape(&self) -> &[usize] {
match self {
LayerGradient::Dense { shape, .. } => shape,
LayerGradient::Sparse(sg) => &sg.shape,
LayerGradient::Quantized(qg) => &qg.shape,
}
}
pub fn to_dense(&self) -> Vec<f32> {
match self {
LayerGradient::Dense { values, .. } => values.clone(),
LayerGradient::Sparse(sg) => sg.to_dense(),
LayerGradient::Quantized(qg) => qg.to_dense(),
}
}
pub fn memory_size(&self) -> usize {
match self {
LayerGradient::Dense { values, .. } => values.len() * 4,
LayerGradient::Sparse(sg) => sg.indices.len() * 4 + sg.values.len() * 4,
LayerGradient::Quantized(qg) => qg.quantized_values.len() + 8,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GradientDelta {
#[serde(serialize_with = "crate::serialize_cid")]
#[serde(deserialize_with = "crate::deserialize_cid")]
pub base_model: Cid,
pub layer_gradients: HashMap<String, LayerGradient>,
pub checksum: u64,
pub timestamp: i64,
}
impl GradientDelta {
pub fn new(base_model: Cid) -> Self {
Self {
base_model,
layer_gradients: HashMap::new(),
checksum: 0,
timestamp: chrono::Utc::now().timestamp(),
}
}
pub fn add_dense_gradient(&mut self, layer_name: String, values: Vec<f32>, shape: Vec<usize>) {
self.layer_gradients
.insert(layer_name, LayerGradient::Dense { values, shape });
self.update_checksum();
}
pub fn add_sparse_gradient(&mut self, layer_name: String, gradient: SparseGradient) {
self.layer_gradients
.insert(layer_name, LayerGradient::Sparse(gradient));
self.update_checksum();
}
pub fn add_quantized_gradient(&mut self, layer_name: String, gradient: QuantizedGradient) {
self.layer_gradients
.insert(layer_name, LayerGradient::Quantized(gradient));
self.update_checksum();
}
fn update_checksum(&mut self) {
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
let mut hasher = DefaultHasher::new();
self.layer_gradients.len().hash(&mut hasher);
let mut sorted_layers: Vec<_> = self.layer_gradients.iter().collect();
sorted_layers.sort_by_key(|(name, _)| *name);
for (name, gradient) in sorted_layers {
name.hash(&mut hasher);
gradient.shape().hash(&mut hasher);
let dense = gradient.to_dense();
let sample_size = dense.len().min(100);
for &v in dense.iter().take(sample_size) {
v.to_bits().hash(&mut hasher);
}
}
self.checksum = hasher.finish();
}
pub fn verify_checksum(&self) -> Result<(), GradientError> {
let mut temp = self.clone();
temp.update_checksum();
if temp.checksum == self.checksum {
Ok(())
} else {
Err(GradientError::ChecksumFailed)
}
}
pub fn total_memory_size(&self) -> usize {
self.layer_gradients.values().map(|g| g.memory_size()).sum()
}
}
pub struct GradientCompressor;
impl GradientCompressor {
pub fn top_k(
values: &[f32],
shape: Vec<usize>,
k: usize,
) -> Result<SparseGradient, GradientError> {
if k == 0 || k > values.len() {
return Err(GradientError::InvalidCompressionRatio(
k as f32 / values.len() as f32,
));
}
let mut indexed_values: Vec<(usize, f32)> = values
.iter()
.enumerate()
.map(|(i, &v)| (i, v.abs()))
.collect();
indexed_values.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
indexed_values.truncate(k);
let mut indices = Vec::with_capacity(k);
let mut sparse_values = Vec::with_capacity(k);
for (idx, _) in indexed_values {
indices.push(idx);
sparse_values.push(values[idx]);
}
Ok(SparseGradient::new(indices, sparse_values, shape))
}
pub fn threshold(values: &[f32], shape: Vec<usize>, threshold: f32) -> SparseGradient {
let mut indices = Vec::new();
let mut sparse_values = Vec::new();
for (i, &v) in values.iter().enumerate() {
if v.abs() >= threshold {
indices.push(i);
sparse_values.push(v);
}
}
SparseGradient::new(indices, sparse_values, shape)
}
pub fn quantize(values: &[f32], shape: Vec<usize>) -> QuantizedGradient {
QuantizedGradient::from_dense(values, shape)
}
pub fn random_sparsification(
values: &[f32],
shape: Vec<usize>,
keep_ratio: f32,
) -> Result<SparseGradient, GradientError> {
use rand::RngExt;
if keep_ratio <= 0.0 || keep_ratio > 1.0 {
return Err(GradientError::InvalidCompressionRatio(keep_ratio));
}
let mut rng = rand::rng();
let mut indices = Vec::new();
let mut sparse_values = Vec::new();
for (i, &v) in values.iter().enumerate() {
if rng.random::<f32>() < keep_ratio {
indices.push(i);
sparse_values.push(v / keep_ratio); }
}
Ok(SparseGradient::new(indices, sparse_values, shape))
}
}
pub struct GradientAggregator;
impl GradientAggregator {
pub fn average(gradients: &[Vec<f32>]) -> Result<Vec<f32>, GradientError> {
if gradients.is_empty() {
return Err(GradientError::EmptyGradientSet);
}
let len = gradients[0].len();
for g in gradients.iter() {
if g.len() != len {
return Err(GradientError::ShapeMismatch {
expected: vec![len],
actual: vec![g.len()],
});
}
}
let mut result = vec![0.0; len];
let count = gradients.len() as f32;
for gradient in gradients {
for (i, &v) in gradient.iter().enumerate() {
result[i] += v / count;
}
}
Ok(result)
}
pub fn weighted_average(
gradients: &[Vec<f32>],
weights: &[f32],
) -> Result<Vec<f32>, GradientError> {
if gradients.is_empty() {
return Err(GradientError::EmptyGradientSet);
}
if gradients.len() != weights.len() {
return Err(GradientError::InvalidGradient(format!(
"Gradient count {} != weight count {}",
gradients.len(),
weights.len()
)));
}
let len = gradients[0].len();
for g in gradients.iter() {
if g.len() != len {
return Err(GradientError::ShapeMismatch {
expected: vec![len],
actual: vec![g.len()],
});
}
}
let weight_sum: f32 = weights.iter().sum();
if weight_sum == 0.0 {
return Err(GradientError::InvalidGradient(
"Sum of weights is zero".to_string(),
));
}
let mut result = vec![0.0; len];
for (gradient, &weight) in gradients.iter().zip(weights) {
let normalized_weight = weight / weight_sum;
for (i, &v) in gradient.iter().enumerate() {
result[i] += v * normalized_weight;
}
}
Ok(result)
}
pub fn apply_momentum(
current_gradient: &[f32],
previous_momentum: &[f32],
momentum_factor: f32,
) -> Result<Vec<f32>, GradientError> {
if current_gradient.len() != previous_momentum.len() {
return Err(GradientError::ShapeMismatch {
expected: vec![previous_momentum.len()],
actual: vec![current_gradient.len()],
});
}
let result = current_gradient
.iter()
.zip(previous_momentum)
.map(|(&g, &m)| momentum_factor * m + g)
.collect();
Ok(result)
}
}
pub struct GradientVerifier;
impl GradientVerifier {
pub fn verify_shape(gradient: &[f32], expected_shape: &[usize]) -> Result<(), GradientError> {
let expected_size: usize = expected_shape.iter().product();
if gradient.len() != expected_size {
return Err(GradientError::ShapeMismatch {
expected: expected_shape.to_vec(),
actual: vec![gradient.len()],
});
}
Ok(())
}
pub fn detect_outliers(gradient: &[f32], std_threshold: f32) -> Result<(), GradientError> {
if gradient.is_empty() {
return Ok(());
}
let mean = gradient.iter().sum::<f32>() / gradient.len() as f32;
let variance =
gradient.iter().map(|&v| (v - mean).powi(2)).sum::<f32>() / gradient.len() as f32;
let std_dev = variance.sqrt();
for (i, &v) in gradient.iter().enumerate() {
let z_score = (v - mean).abs() / std_dev;
if z_score > std_threshold {
return Err(GradientError::OutlierDetected { index: i, value: v });
}
}
Ok(())
}
pub fn verify_finite(gradient: &[f32]) -> Result<(), GradientError> {
for (i, &v) in gradient.iter().enumerate() {
if !v.is_finite() {
return Err(GradientError::InvalidGradient(format!(
"Non-finite value at index {}: {}",
i, v
)));
}
}
Ok(())
}
pub fn l2_norm(gradient: &[f32]) -> f32 {
gradient.iter().map(|&v| v * v).sum::<f32>().sqrt()
}
pub fn clip_by_norm(gradient: &mut [f32], max_norm: f32) {
let norm = Self::l2_norm(gradient);
if norm > max_norm {
let scale = max_norm / norm;
for v in gradient.iter_mut() {
*v *= scale;
}
}
}
}