From 93809eea1babcdfc0e0f9a30a0768538cd942ba9 Mon Sep 17 00:00:00 2001 From: boris Date: Sun, 13 Sep 2026 13:02:16 +0800 Subject: [PATCH] perf(engine): share immutable daily factor schemas and numeric buffers --- crates/fidc-core/src/data.rs | 9 +- crates/fidc-core/src/numeric_factors.rs | 259 +++++++++++++++--- .../src/numeric_factors/shared_rows.rs | 165 +++++++++++ 3 files changed, 388 insertions(+), 45 deletions(-) create mode 100644 crates/fidc-core/src/numeric_factors/shared_rows.rs diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index c519304..cfa3d8d 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -4389,13 +4389,7 @@ fn normalize_factor_snapshots( value, }); } - let already_normalized = snapshot.extra_factors.iter().all(|(field, value)| { - let trimmed = field.as_str().trim().trim_matches('"').trim_matches('\''); - !trimmed.is_empty() - && trimmed == field.as_str() - && trimmed.bytes().all(|byte| !byte.is_ascii_uppercase()) - && value.is_finite() - }); + let already_normalized = snapshot.extra_factors.has_normalized_finite_entries(); if already_normalized { return Ok(snapshot); } @@ -4465,6 +4459,7 @@ fn normalize_daily_snapshot_bundle( |row| row.symbol.as_str(), )?; sort_rows_by_symbol_if_needed(&mut bundle.market, |row| row.symbol.as_str()); + NumericFactorMap::share_rows(bundle.factors.iter_mut().map(|row| &mut row.extra_factors)); bundle.factors = normalize_factor_snapshots(bundle.factors)?; sort_rows_by_symbol_if_needed(&mut bundle.factors, |row| row.symbol.as_str()); sort_rows_by_symbol_if_needed(&mut bundle.candidates, |row| row.symbol.as_str()); diff --git a/crates/fidc-core/src/numeric_factors.rs b/crates/fidc-core/src/numeric_factors.rs index be44e83..f6cca55 100644 --- a/crates/fidc-core/src/numeric_factors.rs +++ b/crates/fidc-core/src/numeric_factors.rs @@ -8,10 +8,36 @@ use serde::de::{MapAccess, Visitor}; use serde::ser::SerializeMap; use serde::{Deserialize, Deserializer, Serialize, Serializer}; +mod shared_rows; +use shared_rows::{SharedIter, SharedRow}; + /// Sorted numeric fields stored contiguously, without a tree node per snapshot. -#[derive(Clone, Default, PartialEq)] +#[derive(Clone)] pub struct NumericFactorMap { - entries: Vec<(CompactString, f64)>, + storage: Storage, +} + +#[derive(Clone)] +enum Storage { + Owned(Vec<(CompactString, f64)>), + Shared(SharedRow), +} + +impl Default for NumericFactorMap { + fn default() -> Self { + Self::new() + } +} + +impl PartialEq for NumericFactorMap { + fn eq(&self, other: &Self) -> bool { + self.len() == other.len() && self.iter().eq(other.iter()) + } +} + +fn normalized_name(name: &str) -> bool { + let trimmed = name.trim().trim_matches('"').trim_matches('\''); + !trimmed.is_empty() && trimmed == name && !name.bytes().any(|byte| byte.is_ascii_uppercase()) } fn compact_key(key: Cow<'static, str>) -> CompactString { @@ -24,32 +50,70 @@ fn compact_key(key: Cow<'static, str>) -> CompactString { impl NumericFactorMap { pub const fn new() -> Self { Self { - entries: Vec::new(), + storage: Storage::Owned(Vec::new()), } } pub fn len(&self) -> usize { - self.entries.len() + match &self.storage { + Storage::Owned(entries) => entries.len(), + Storage::Shared(row) => row.len(), + } } pub fn is_empty(&self) -> bool { - self.entries.is_empty() + self.len() == 0 } pub fn clear(&mut self) { - self.entries.clear(); + match &mut self.storage { + Storage::Owned(entries) => entries.clear(), + Storage::Shared(_) => *self = Self::new(), + } + } + + /// Shares only immutable values; a subsequent mutation owns a private row. + pub fn share_rows<'a>(rows: impl IntoIterator) -> usize { + shared_rows::share(rows) + } + + pub(crate) fn has_normalized_finite_entries(&self) -> bool { + match &self.storage { + Storage::Owned(entries) => entries.iter().all(|(name, value)| { + normalized_name(name) && value.is_finite() + }), + Storage::Shared(row) => row.has_normalized_finite_entries(), + } + } + + fn owned_entries(&mut self) -> &mut Vec<(CompactString, f64)> { + if matches!(self.storage, Storage::Shared(_)) { + let entries = self.iter().map(|(key, value)| (key.clone(), *value)).collect(); + self.storage = Storage::Owned(entries); + } + match &mut self.storage { + Storage::Owned(entries) => entries, + Storage::Shared(_) => unreachable!("shared row was materialized"), + } } pub fn get(&self, key: &str) -> Option<&f64> { - self.entries - .binary_search_by(|(name, _)| name.as_str().cmp(key)) - .ok() - .map(|index| &self.entries[index].1) + match &self.storage { + Storage::Owned(entries) => entries + .binary_search_by(|(name, _)| name.as_str().cmp(key)) + .ok() + .map(|index| &entries[index].1), + Storage::Shared(row) => row.get(key), + } } pub fn get_mut(&mut self, key: &str) -> Option<&mut f64> { - self.entries + if !self.contains_key(key) { + return None; + } + let entries = self.owned_entries(); + entries .binary_search_by(|(name, _)| name.as_str().cmp(key)) .ok() - .map(|index| &mut self.entries[index].1) + .map(|index| &mut entries[index].1) } pub fn contains_key(&self, key: &str) -> bool { @@ -61,45 +125,51 @@ impl NumericFactorMap { } pub fn insert_compact(&mut self, key: CompactString, value: f64) -> Option { - if self - .entries + let entries = self.owned_entries(); + if entries .last() .is_none_or(|(last, _)| last.as_str() < key.as_str()) { - self.entries.push((key, value)); + entries.push((key, value)); return None; } - match self - .entries + match entries .binary_search_by(|(name, _)| name.as_str().cmp(key.as_str())) { - Ok(index) => Some(std::mem::replace(&mut self.entries[index].1, value)), + Ok(index) => Some(std::mem::replace(&mut entries[index].1, value)), Err(index) => { - self.entries.insert(index, (key, value)); + entries.insert(index, (key, value)); None } } } pub fn remove(&mut self, key: &str) -> Option { - self.entries + if !self.contains_key(key) { + return None; + } + let entries = self.owned_entries(); + entries .binary_search_by(|(name, _)| name.as_str().cmp(key)) .ok() - .map(|index| self.entries.remove(index).1) + .map(|index| entries.remove(index).1) } pub fn retain(&mut self, mut keep: impl FnMut(&CompactString, &mut f64) -> bool) { - self.entries.retain_mut(|(key, value)| keep(key, value)); + self.owned_entries().retain_mut(|(key, value)| keep(key, value)); } pub fn iter(&self) -> Iter<'_> { - Iter(self.entries.iter()) + match &self.storage { + Storage::Owned(entries) => Iter(IterStorage::Owned(entries.iter())), + Storage::Shared(row) => Iter(IterStorage::Shared(row.iter())), + } } pub fn keys(&self) -> impl DoubleEndedIterator + ExactSizeIterator { - self.entries.iter().map(|(key, _)| key) + self.iter().map(|(key, _)| key) } pub fn values(&self) -> impl DoubleEndedIterator + ExactSizeIterator { - self.entries.iter().map(|(_, value)| value) + self.iter().map(|(_, value)| value) } } @@ -116,19 +186,33 @@ impl Index<&str> for NumericFactorMap { } } -pub struct Iter<'a>(std::slice::Iter<'a, (CompactString, f64)>); +pub struct Iter<'a>(IterStorage<'a>); + +enum IterStorage<'a> { + Owned(std::slice::Iter<'a, (CompactString, f64)>), + Shared(SharedIter<'a>), +} impl<'a> Iterator for Iter<'a> { type Item = (&'a CompactString, &'a f64); fn next(&mut self) -> Option { - self.0.next().map(|(k, v)| (k, v)) + match &mut self.0 { + IterStorage::Owned(iter) => iter.next().map(|(key, value)| (key, value)), + IterStorage::Shared(iter) => iter.next(), + } } fn size_hint(&self) -> (usize, Option) { - self.0.size_hint() + match &self.0 { + IterStorage::Owned(iter) => iter.size_hint(), + IterStorage::Shared(iter) => iter.size_hint(), + } } } impl DoubleEndedIterator for Iter<'_> { fn next_back(&mut self) -> Option { - self.0.next_back().map(|(k, v)| (k, v)) + match &mut self.0 { + IterStorage::Owned(iter) => iter.next_back().map(|(key, value)| (key, value)), + IterStorage::Shared(iter) => iter.next_back(), + } } } impl ExactSizeIterator for Iter<'_> {} @@ -143,7 +227,11 @@ impl IntoIterator for NumericFactorMap { type Item = (CompactString, f64); type IntoIter = std::vec::IntoIter; fn into_iter(self) -> Self::IntoIter { - self.entries.into_iter() + match self.storage { + Storage::Owned(entries) => entries.into_iter(), + Storage::Shared(row) => row.iter().map(|(key, value)| (key.clone(), *value)) + .collect::>().into_iter(), + } } } @@ -167,7 +255,7 @@ impl FromIterator<(CompactString, f64)> for NumericFactorMap { false } }); - Self { entries } + Self { storage: Storage::Owned(entries) } } } impl Extend<(Cow<'static, str>, f64)> for NumericFactorMap { @@ -177,7 +265,7 @@ impl Extend<(Cow<'static, str>, f64)> for NumericFactorMap { } impl Extend<(CompactString, f64)> for NumericFactorMap { fn extend>(&mut self, iter: T) { - let mut incoming: Self = iter.into_iter().collect(); + let incoming: Self = iter.into_iter().collect(); if incoming.is_empty() { return; } @@ -185,15 +273,17 @@ impl Extend<(CompactString, f64)> for NumericFactorMap { *self = incoming; return; } - if self.entries.last().unwrap().0 < incoming.entries[0].0 { - self.entries.append(&mut incoming.entries); + let mut incoming = incoming.into_iter().collect::>(); + let entries = self.owned_entries(); + if entries.last().unwrap().0 < incoming[0].0 { + entries.append(&mut incoming); return; } // Merge sorted sets in linear time; wide factor batches must not shift // the existing vector once per field. Existing keys keep their identity. - let mut merged = Vec::with_capacity(self.len() + incoming.len()); - let mut old = std::mem::take(&mut self.entries).into_iter().peekable(); - let mut new = incoming.entries.into_iter().peekable(); + let mut merged = Vec::with_capacity(entries.len() + incoming.len()); + let mut old = std::mem::take(entries).into_iter().peekable(); + let mut new = incoming.into_iter().peekable(); while let (Some(left), Some(right)) = (old.peek(), new.peek()) { match left.0.cmp(&right.0) { std::cmp::Ordering::Less => merged.push(old.next().unwrap()), @@ -206,7 +296,7 @@ impl Extend<(CompactString, f64)> for NumericFactorMap { } merged.extend(old); merged.extend(new); - self.entries = merged; + *entries = merged; } } impl From<[(Cow<'static, str>, f64); N]> for NumericFactorMap { @@ -253,6 +343,99 @@ impl<'de> Deserialize<'de> for NumericFactorMap { mod tests { use super::*; + fn row_bits(row: &NumericFactorMap) -> Vec<(String, u64)> { + row.iter().map(|(key, value)| (key.to_string(), value.to_bits())).collect() + } + + #[test] + fn shared_rows_keep_missing_entries_names_and_float_bits() { + let mut rows = (0..128).map(|row| { + (0..12).filter(|field| (field + row) % 7 != 0).map(|field| { + let value = match (row, field) { + (1, 1) => -0.0, + (2, 1) => f64::from_bits(0x7ff8_0000_0000_0012), + (3, 1) => f64::INFINITY, + _ => row as f64 / 7.0 + field as f64, + }; + (Cow::Owned(format!("factor_{field:02}")), value) + }).collect::() + }).collect::>(); + rows[7].clear(); + rows[8].insert(Cow::Borrowed(" 'Quoted' "), 1.0); + let before = rows.iter().map(row_bits).collect::>(); + let serialized = rows.iter().map(|row| serde_json::to_string(row).unwrap()).collect::>(); + let normalized = rows.iter().map(NumericFactorMap::has_normalized_finite_entries).collect::>(); + assert_eq!(NumericFactorMap::share_rows(rows.iter_mut()), rows.len()); + for (index, row) in rows.iter().enumerate() { + assert!(matches!(row.storage, Storage::Shared(_))); + assert_eq!(row_bits(row), before[index]); + assert_eq!(serde_json::to_string(row).unwrap(), serialized[index]); + assert_eq!(row.has_normalized_finite_entries(), normalized[index]); + for field in 0..12 { + let name = format!("factor_{field:02}"); + assert_eq!(row.get(&name).map(|value| value.to_bits()), + before[index].iter().find(|(key, _)| key == &name).map(|(_, value)| *value)); + } + let mut remaining = before[index].clone(); + let mut iter = row.iter(); + while !remaining.is_empty() { + assert_eq!(iter.len(), remaining.len()); + let (expected, actual) = if remaining.len() % 2 == 0 { + (remaining.remove(0), iter.next()) + } else { + (remaining.pop().unwrap(), iter.next_back()) + }; + let (key, value) = actual.unwrap(); + assert_eq!((key.to_string(), value.to_bits()), expected); + } + assert_eq!(iter.len(), 0); + assert!(iter.next().is_none() && iter.next_back().is_none()); + assert_eq!(row.clone().into_iter().map(|(key, value)| (key.to_string(), value.to_bits())) + .collect::>(), before[index]); + } + assert_eq!(NumericFactorMap::share_rows(rows.iter_mut()), 0); + } + + #[test] + fn shared_row_mutations_are_private_and_keep_map_semantics() { + let mut rows = (0..64).map(|row| { + (0..8).map(|field| (Cow::Owned(format!("f{field}")), (row * 8 + field) as f64)) + .collect::() + }).collect::>(); + let originals = rows.clone(); + assert_eq!(NumericFactorMap::share_rows(rows.iter_mut()), 64); + assert_eq!(rows, originals); + let surviving = rows[0].clone(); + *rows[0].get_mut("f0").unwrap() = -99.0; + rows[1].insert(Cow::Borrowed("f1"), 101.0); + rows[2].remove("f2"); + rows[3].retain(|key, value| { *value *= 2.0; key.as_str() != "f3" }); + rows[4].extend([(Cow::Borrowed("f1"), 303.0), (Cow::Borrowed("new"), -0.0)]); + rows[5].clear(); + assert_eq!(surviving, originals[0]); + assert_eq!(rows[0]["f0"], -99.0); + assert_eq!(rows[1]["f1"], 101.0); + assert!(!rows[2].contains_key("f2")); + assert!(!rows[3].contains_key("f3")); + assert_eq!(rows[3]["f0"], originals[3]["f0"] * 2.0); + assert_eq!(rows[4]["new"].to_bits(), (-0.0_f64).to_bits()); + assert!(rows[5].is_empty()); + assert_eq!(&rows[6..], &originals[6..]); + drop(rows); + assert_eq!(surviving, originals[0]); + } + + #[test] + fn sparse_rows_do_not_expand_into_a_dense_matrix() { + let mut rows = (0..256).map(|row| NumericFactorMap::from([ + (Cow::Owned(format!("only_{row}")), row as f64), + ])).collect::>(); + let before = rows.clone(); + assert_eq!(NumericFactorMap::share_rows(rows.iter_mut()), 0); + assert_eq!(rows, before); + assert!(rows.iter().all(|row| matches!(row.storage, Storage::Owned(_)))); + } + #[test] fn compact_keys_inline_dynamic_names_and_keep_long_static_storage() { const LONG: &str = "a_long_static_factor_identifier_that_must_remain_borrowed"; diff --git a/crates/fidc-core/src/numeric_factors/shared_rows.rs b/crates/fidc-core/src/numeric_factors/shared_rows.rs new file mode 100644 index 0000000..253335f --- /dev/null +++ b/crates/fidc-core/src/numeric_factors/shared_rows.rs @@ -0,0 +1,165 @@ +use std::collections::BTreeSet; +use std::sync::Arc; + +use compact_str::CompactString; + +use super::{NumericFactorMap, Storage, normalized_name}; + +const MAX_SHARED_ROWS_BYTES: usize = 64 * 1024 * 1024; + +struct SharedRows { + fields: Vec, + values: Box<[f64]>, + present: Box<[u64]>, + lengths: Box<[usize]>, + normalized_finite: Box<[bool]>, +} + +#[derive(Clone)] +pub(super) struct SharedRow { + data: Arc, + index: usize, +} + +impl SharedRow { + pub(super) fn len(&self) -> usize { + self.data.lengths[self.index] + } + + pub(super) fn has_normalized_finite_entries(&self) -> bool { + self.data.normalized_finite[self.index] + } + + fn value_at(&self, field: usize) -> Option<&f64> { + let offset = self.index * self.data.fields.len() + field; + (self.data.present[offset / 64] & (1_u64 << (offset % 64)) != 0) + .then(|| &self.data.values[offset]) + } + + pub(super) fn get(&self, name: &str) -> Option<&f64> { + self.data.fields.binary_search_by(|field| field.as_str().cmp(name)) + .ok().and_then(|field| self.value_at(field)) + } + + pub(super) fn iter(&self) -> SharedIter<'_> { + SharedIter { row: self, front: 0, back: self.data.fields.len(), remaining: self.len() } + } +} + +pub(super) struct SharedIter<'a> { + row: &'a SharedRow, + front: usize, + back: usize, + remaining: usize, +} + +impl<'a> Iterator for SharedIter<'a> { + type Item = (&'a CompactString, &'a f64); + + fn next(&mut self) -> Option { + if self.remaining == 0 { + return None; + } + while self.front < self.back { + let field = self.front; + self.front += 1; + if let Some(value) = self.row.value_at(field) { + self.remaining -= 1; + return Some((&self.row.data.fields[field], value)); + } + } + None + } + + fn size_hint(&self) -> (usize, Option) { + (self.remaining, Some(self.remaining)) + } +} + +impl DoubleEndedIterator for SharedIter<'_> { + fn next_back(&mut self) -> Option { + if self.remaining == 0 { + return None; + } + while self.front < self.back { + self.back -= 1; + if let Some(value) = self.row.value_at(self.back) { + self.remaining -= 1; + return Some((&self.row.data.fields[self.back], value)); + } + } + None + } +} + +impl ExactSizeIterator for SharedIter<'_> {} + +pub(super) fn share<'a>(rows: impl IntoIterator) -> usize { + let rows = rows.into_iter().collect::>(); + if rows.len() < 2 || rows.iter().any(|row| matches!(row.storage, Storage::Shared(_))) { + return 0; + } + let Some(owned_bytes) = rows.iter().try_fold(0usize, |sum, row| { + let Storage::Owned(entries) = &row.storage else { return None }; + sum.checked_add(entries.capacity().checked_mul(std::mem::size_of::<(CompactString, f64)>())?) + }) else { return 0 }; + let fields = rows.iter().flat_map(|row| row.keys()).collect::>() + .into_iter().cloned().collect::>(); + let Some(cells) = rows.len().checked_mul(fields.len()) else { return 0 }; + if cells == 0 { + return 0; + } + let words = cells.div_ceil(64); + let Some(field_bytes) = fields.iter().try_fold(0usize, |size, field| { + size.checked_add(std::mem::size_of::())?.checked_add(field.len()) + }) else { return 0 }; + let Some(bytes) = cells.checked_mul(std::mem::size_of::()) + .and_then(|size| size.checked_add(words.checked_mul(std::mem::size_of::())?)) + .and_then(|size| size.checked_add(rows.len().checked_mul(std::mem::size_of::() + 1)?)) + .and_then(|size| size.checked_add(field_bytes)) + .and_then(|size| size.checked_add(std::mem::size_of::())) + else { return 0 }; + if bytes >= owned_bytes || bytes > MAX_SHARED_ROWS_BYTES { + return 0; + } + + let mut values = Vec::new(); + if values.try_reserve_exact(cells).is_err() { + return 0; + } + values.resize(cells, 0.0); + let mut present = vec![0_u64; words]; + let mut lengths = Vec::with_capacity(rows.len()); + let mut normalized_finite = Vec::with_capacity(rows.len()); + let canonical_fields = fields.iter().map(|name| normalized_name(name)).collect::>(); + for (index, row) in rows.iter().enumerate() { + lengths.push(row.len()); + let mut canonical_row = true; + let mut field = 0usize; + for (name, value) in row.iter() { + while &fields[field] < name { + field += 1; + } + debug_assert_eq!(&fields[field], name); + let offset = index * fields.len() + field; + values[offset] = *value; + present[offset / 64] |= 1_u64 << (offset % 64); + canonical_row &= canonical_fields[field] && value.is_finite(); + field += 1; + } + normalized_finite.push(canonical_row); + } + // Publish only after every input row has been copied without arithmetic. + let data = Arc::new(SharedRows { + fields, + values: values.into_boxed_slice(), + present: present.into_boxed_slice(), + lengths: lengths.into_boxed_slice(), + normalized_finite: normalized_finite.into_boxed_slice(), + }); + let count = rows.len(); + for (index, row) in rows.into_iter().enumerate() { + row.storage = Storage::Shared(SharedRow { data: Arc::clone(&data), index }); + } + count +}