From 195678c36b335022c9a2add4f755bb675955feb2 Mon Sep 17 00:00:00 2001 From: jarmen423 Date: Wed, 29 Jul 2026 17:45:14 +0000 Subject: [PATCH] feat(lpg): expand residency accounting and compact/lazy experiments Attribute property map slots, string payloads, capacity waste, and adjacency; add shrink + dictionary-compress experiments with RSS and correctness checks for the Track E residual investigation. --- crates/grafeo-common/src/memory/mod.rs | 5 +- crates/grafeo-common/src/memory/usage.rs | 163 ++++++++++++++ crates/grafeo-common/src/types/value.rs | 22 ++ crates/grafeo-core/src/graph/lpg/property.rs | 179 +++++++++++++-- .../grafeo-core/src/graph/lpg/store/memory.rs | 75 ++++++- crates/grafeo-core/src/index/adjacency.rs | 80 +++++-- .../tests/lpg_residency_experiments.rs | 207 ++++++++++++++++++ crates/grafeo-engine/src/database/admin.rs | 19 ++ crates/grafeo-engine/src/memory_usage.rs | 3 +- docs/TRACK_E_LPG_RESIDENCY.md | 33 +++ .../TRACK_E_LPG_MEMORY_ACCOUNTING.md | 85 +++++++ 11 files changed, 822 insertions(+), 49 deletions(-) create mode 100644 crates/grafeo-core/tests/lpg_residency_experiments.rs create mode 100644 docs/TRACK_E_LPG_RESIDENCY.md create mode 100644 docs/diagnostics/TRACK_E_LPG_MEMORY_ACCOUNTING.md diff --git a/crates/grafeo-common/src/memory/mod.rs b/crates/grafeo-common/src/memory/mod.rs index d5a7d0ba4..19438201d 100644 --- a/crates/grafeo-common/src/memory/mod.rs +++ b/crates/grafeo-common/src/memory/mod.rs @@ -27,4 +27,7 @@ pub use buffer::{ pub use bump::BumpAllocator; pub use pool::ObjectPool; pub use reporter::MemoryReporter; -pub use usage::{IndexMemory, MvccMemory, NamedMemory, StoreMemory, StringPoolMemory}; +pub use usage::{ + AdjacencyCapacityMemory, IndexMemory, LpgResidencyMemory, MvccMemory, NamedMemory, + PropertyColumnMemory, PropertyStorageMemory, StoreMemory, StringPoolMemory, +}; diff --git a/crates/grafeo-common/src/memory/usage.rs b/crates/grafeo-common/src/memory/usage.rs index 8e7d7bc25..39409fa07 100644 --- a/crates/grafeo-common/src/memory/usage.rs +++ b/crates/grafeo-common/src/memory/usage.rs @@ -20,6 +20,30 @@ pub struct StoreMemory { pub edge_properties_bytes: usize, /// Number of property columns (node + edge). pub property_column_count: usize, + /// Hash-map slot bytes for node property columns (capacity × entry). + #[serde(default)] + pub node_property_map_slot_bytes: usize, + /// Decoded `Value` payload bytes inside node property columns (strings/lists/…). + #[serde(default)] + pub node_property_decoded_payload_bytes: usize, + /// String/bytes payload subset of node property decoded values. + #[serde(default)] + pub node_property_string_payload_bytes: usize, + /// Unused map capacity waste in node property columns. + #[serde(default)] + pub node_property_capacity_waste_bytes: usize, + /// Hash-map slot bytes for edge property columns. + #[serde(default)] + pub edge_property_map_slot_bytes: usize, + /// Decoded `Value` payload bytes inside edge property columns. + #[serde(default)] + pub edge_property_decoded_payload_bytes: usize, + /// String/bytes payload subset of edge property decoded values. + #[serde(default)] + pub edge_property_string_payload_bytes: usize, + /// Unused map capacity waste in edge property columns. + #[serde(default)] + pub edge_property_capacity_waste_bytes: usize, } impl StoreMemory { @@ -32,6 +56,139 @@ impl StoreMemory { } } +/// Per-column residency attribution for one property key. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct PropertyColumnMemory { + /// Property key name. + pub key: String, + /// Live entries in the hot map. + pub entry_count: usize, + /// Hot map capacity (buckets reserved). + pub map_capacity: usize, + /// `capacity × (Id + Value + 1)` slot estimate. + pub map_slot_bytes: usize, + /// Sum of [`crate::types::Value::estimated_size_bytes`] over hot values. + pub decoded_payload_bytes: usize, + /// String + bytes payload subset of decoded values. + pub string_payload_bytes: usize, + /// Compressed column backing (0 when uncompressed). + pub compressed_bytes: usize, + /// `(capacity − len) × entry` waste in the hot map. + pub capacity_waste_bytes: usize, + /// Values held only in compressed form (not in the hot map). + pub compressed_entry_count: usize, +} + +impl PropertyColumnMemory { + /// Estimated total for this column (slots + payloads + compressed). + #[must_use] + pub fn total_bytes(&self) -> usize { + self.map_slot_bytes + self.decoded_payload_bytes + self.compressed_bytes + } +} + +/// Aggregated property-storage residency (node or edge side). +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct PropertyStorageMemory { + /// Outer column-map overhead. + pub map_overhead_bytes: usize, + /// Per-column breakdown (sorted by `total_bytes` descending when produced). + pub columns: Vec, + /// Sum of column map-slot bytes. + pub total_map_slot_bytes: usize, + /// Sum of decoded payloads. + pub total_decoded_payload_bytes: usize, + /// Sum of string/bytes payloads. + pub total_string_payload_bytes: usize, + /// Sum of compressed backings. + pub total_compressed_bytes: usize, + /// Sum of hot-map capacity waste. + pub total_capacity_waste_bytes: usize, + /// `map_overhead + Σ column totals`. + pub total_bytes: usize, +} + +impl PropertyStorageMemory { + /// Recomputes aggregate totals from columns + outer map overhead. + pub fn compute_total(&mut self) { + self.total_map_slot_bytes = self.columns.iter().map(|c| c.map_slot_bytes).sum(); + self.total_decoded_payload_bytes = + self.columns.iter().map(|c| c.decoded_payload_bytes).sum(); + self.total_string_payload_bytes = + self.columns.iter().map(|c| c.string_payload_bytes).sum(); + self.total_compressed_bytes = self.columns.iter().map(|c| c.compressed_bytes).sum(); + self.total_capacity_waste_bytes = + self.columns.iter().map(|c| c.capacity_waste_bytes).sum(); + self.total_bytes = self.map_overhead_bytes + + self.total_map_slot_bytes + + self.total_decoded_payload_bytes + + self.total_compressed_bytes; + } +} + +/// Adjacency list capacity vs used-byte attribution. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct AdjacencyCapacityMemory { + /// Nodes with an adjacency list. + pub node_count: usize, + /// Outer list-map capacity. + pub list_map_capacity: usize, + /// Outer list-map overhead bytes. + pub list_map_overhead_bytes: usize, + /// Bytes occupied by live hot entries (destinations + edge ids used len). + pub hot_used_bytes: usize, + /// Bytes reserved by hot chunk Vec capacities. + pub hot_capacity_bytes: usize, + /// Compressed cold-chunk bytes. + pub cold_bytes: usize, + /// Delta / deleted / skip-index reserved bytes. + pub aux_capacity_bytes: usize, + /// `hot_capacity − hot_used` (+ list map slack approximated separately). + pub capacity_waste_bytes: usize, + /// Total estimated heap (`list_map_overhead + hot_capacity + cold + aux`). + pub total_bytes: usize, +} + +impl AdjacencyCapacityMemory { + /// Recomputes `capacity_waste_bytes` and `total_bytes`. + pub fn compute_total(&mut self) { + self.capacity_waste_bytes = self.hot_capacity_bytes.saturating_sub(self.hot_used_bytes); + self.total_bytes = self.list_map_overhead_bytes + + self.hot_capacity_bytes + + self.cold_bytes + + self.aux_capacity_bytes; + } +} + +/// Full LPG residency attribution (properties + adjacency capacities). +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct LpgResidencyMemory { + /// Node property columns. + pub node_properties: PropertyStorageMemory, + /// Edge property columns. + pub edge_properties: PropertyStorageMemory, + /// Forward adjacency capacity detail. + pub forward_adjacency: AdjacencyCapacityMemory, + /// Backward adjacency capacity detail (empty when disabled). + pub backward_adjacency: AdjacencyCapacityMemory, + /// Sum of property + adjacency totals. + pub total_bytes: usize, +} + +impl LpgResidencyMemory { + /// Recomputes `total_bytes` from children. + pub fn compute_total(&mut self) { + self.node_properties.compute_total(); + self.edge_properties.compute_total(); + self.forward_adjacency.compute_total(); + self.backward_adjacency.compute_total(); + self.total_bytes = self.node_properties.total_bytes + + self.edge_properties.total_bytes + + self.forward_adjacency.total_bytes + + self.backward_adjacency.total_bytes; + } +} + /// Memory used by index structures. #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct IndexMemory { @@ -41,6 +198,12 @@ pub struct IndexMemory { pub forward_adjacency_bytes: usize, /// Backward adjacency lists (0 if disabled). pub backward_adjacency_bytes: usize, + /// Forward adjacency capacity waste (`capacity − used` on hot chunks). + #[serde(default)] + pub forward_adjacency_capacity_waste_bytes: usize, + /// Backward adjacency capacity waste. + #[serde(default)] + pub backward_adjacency_capacity_waste_bytes: usize, /// Label index (label_id -> node set). pub label_index_bytes: usize, /// Node-to-labels reverse index. diff --git a/crates/grafeo-common/src/types/value.rs b/crates/grafeo-common/src/types/value.rs index 6eddfa164..c45a9eef3 100644 --- a/crates/grafeo-common/src/types/value.rs +++ b/crates/grafeo-common/src/types/value.rs @@ -425,6 +425,28 @@ impl Value { } } } + + /// String / bytes payload bytes only (subset of [`Self::estimated_size_bytes`]). + /// + /// Used by LPG residency accounting to attribute ArcStr/bytes separately + /// from map-slot and compressed-column bytes. Nested lists/maps recurse. + #[must_use] + pub fn string_payload_bytes(&self) -> usize { + match self { + Value::String(s) => s.len(), + Value::Bytes(b) => b.len(), + Value::List(items) => items.iter().map(Value::string_payload_bytes).sum(), + Value::Map(m) => m + .iter() + .map(|(k, v)| k.as_ref().len() + v.string_payload_bytes()) + .sum(), + Value::Path { nodes, edges } => { + nodes.iter().map(Value::string_payload_bytes).sum::() + + edges.iter().map(Value::string_payload_bytes).sum::() + } + _ => 0, + } + } } impl fmt::Debug for Value { diff --git a/crates/grafeo-core/src/graph/lpg/property.rs b/crates/grafeo-core/src/graph/lpg/property.rs index 168eb7490..f72a6f9a7 100644 --- a/crates/grafeo-core/src/graph/lpg/property.rs +++ b/crates/grafeo-core/src/graph/lpg/property.rs @@ -294,6 +294,17 @@ impl PropertyStorage { } } + /// Shrinks hot-map capacities on every column to fit live entries. + /// + /// Does not change logical contents. Useful after bulk deserialize or + /// after eviction to reclaim HashMap capacity waste. + pub fn shrink_capacities(&self) { + let mut columns = self.columns.write(); + for col in columns.values_mut() { + col.shrink_capacities(); + } + } + /// Returns compression statistics for all columns. #[must_use] pub fn compression_stats(&self) -> FxHashMap { @@ -317,13 +328,34 @@ impl PropertyStorage { /// Returns estimated heap memory for all columns including hash map overhead. #[must_use] pub fn heap_memory_bytes(&self) -> usize { + self.memory_detail().total_bytes + } + + /// Detailed residency attribution for every property column. + #[must_use] + pub fn memory_detail(&self) -> grafeo_common::memory::PropertyStorageMemory { + use grafeo_common::memory::PropertyStorageMemory; + let columns = self.columns.read(); - // Outer hash map capacity let map_overhead = columns.capacity() * (std::mem::size_of::() + std::mem::size_of::>() + 1); - // Sum of all column heap memory - let column_bytes: usize = columns.values().map(|col| col.heap_memory_bytes()).sum(); - map_overhead + column_bytes + let mut detail = PropertyStorageMemory { + map_overhead_bytes: map_overhead, + columns: columns + .iter() + .map(|(key, col)| { + let mut m = col.memory_detail(); + m.key = key.as_ref().to_string(); + m + }) + .collect(), + ..Default::default() + }; + detail + .columns + .sort_by(|a, b| b.total_bytes().cmp(&a.total_bytes())); + detail.compute_total(); + detail } /// Gets a property value for an entity. @@ -1069,7 +1101,7 @@ impl PropertyColumn { /// Gets a value for an entity. /// /// First checks the hot buffer (uncompressed values), then falls back - /// to the compressed data if present. + /// to compressed storage with on-demand (lazy) decode when present. #[must_use] pub fn get(&self, id: Id) -> Option { // First check hot buffer @@ -1077,11 +1109,49 @@ impl PropertyColumn { return Some(value.clone()); } - // For now, compressed data lookup is not implemented for sparse access - // because the compressed format stores values by index, not by entity ID. - // This would require maintaining an ID -> index map in CompressedColumnData. - // The compressed data is primarily useful for bulk/scan operations. - None + self.get_compressed(id) + } + + /// Lazy point lookup into compressed column data. + /// + /// `index_to_id` is sorted by entity id (see compress paths). Binary + /// search finds the compressed-array index; strings/bools decode one + /// value; integers decompress the whole column once per miss (acceptable + /// for sparse correctness; scan paths should use block decode instead). + fn get_compressed(&self, id: Id) -> Option { + let compressed = self.compressed.as_ref()?; + let id_u64 = id.as_u64(); + match compressed { + CompressedColumnData::Strings { + encoding, + index_to_id, + .. + } => { + let idx = index_to_id.binary_search(&id_u64).ok()?; + encoding + .get(idx) + .map(|s| Value::String(ArcStr::from(s))) + } + CompressedColumnData::Booleans { + data, + index_to_id, + .. + } => { + let idx = index_to_id.binary_search(&id_u64).ok()?; + let values = TypeSpecificCompressor::decompress_booleans(data).ok()?; + values.get(idx).copied().map(Value::Bool) + } + CompressedColumnData::Integers { + data, + index_to_id, + .. + } => { + let idx = index_to_id.binary_search(&id_u64).ok()?; + let values = TypeSpecificCompressor::decompress_integers(data).ok()?; + let raw = *values.get(idx)?; + Some(Value::Int64(crate::codec::zigzag_decode(raw))) + } + } } /// Removes a value for an entity. @@ -1194,17 +1264,49 @@ impl PropertyColumn { /// Returns estimated heap memory for this column. /// - /// Includes the hot buffer hash map capacity, zone map, and any - /// compressed data. + /// Includes hot-map slot capacity, decoded `Value` payloads (strings, + /// lists, …), and any compressed backing. #[must_use] pub fn heap_memory_bytes(&self) -> usize { - // Hot buffer: FxHashMap capacity - let hot_bytes = - self.values.capacity() * (std::mem::size_of::() + std::mem::size_of::() + 1); - // Compressed data + self.memory_detail().total_bytes() + } + + /// Detailed residency attribution for this column (key left empty). + #[must_use] + pub fn memory_detail(&self) -> grafeo_common::memory::PropertyColumnMemory { + use grafeo_common::memory::PropertyColumnMemory; + + let entry_size = + std::mem::size_of::() + std::mem::size_of::() + 1; + let map_slot_bytes = self.values.capacity() * entry_size; + let capacity_waste_bytes = + self.values.capacity().saturating_sub(self.values.len()) * entry_size; + let decoded_payload_bytes: usize = self + .values + .values() + .map(Value::estimated_size_bytes) + .sum(); + let string_payload_bytes: usize = + self.values.values().map(Value::string_payload_bytes).sum(); let compressed_bytes = self.compressed.as_ref().map_or(0, |c| c.memory_usage()); - // ZoneMapEntry is inline (no heap), so just hot + compressed - hot_bytes + compressed_bytes + + PropertyColumnMemory { + key: String::new(), + entry_count: self.values.len(), + map_capacity: self.values.capacity(), + map_slot_bytes, + decoded_payload_bytes, + string_payload_bytes, + compressed_bytes, + capacity_waste_bytes, + compressed_entry_count: self.compressed_count, + } + } + + /// Shrinks the hot map capacity to fit live entries. + pub fn shrink_capacities(&mut self) { + self.values.shrink_to_fit(); + self.block_zone_maps.shrink_to_fit(); } /// Returns whether the column has compressed data. @@ -1789,8 +1891,45 @@ impl PropertyColumn { /// Returns estimated heap memory for this column. #[must_use] pub fn heap_memory_bytes(&self) -> usize { - self.values.capacity() - * (std::mem::size_of::() + std::mem::size_of::>() + 1) + self.memory_detail().total_bytes() + } + + /// Detailed residency attribution (temporal: version-log slots + latest payloads). + #[must_use] + pub fn memory_detail(&self) -> grafeo_common::memory::PropertyColumnMemory { + use grafeo_common::memory::PropertyColumnMemory; + + let entry_size = + std::mem::size_of::() + std::mem::size_of::>() + 1; + let map_slot_bytes = self.values.capacity() * entry_size; + let capacity_waste_bytes = + self.values.capacity().saturating_sub(self.values.len()) * entry_size; + let mut decoded_payload_bytes = 0usize; + let mut string_payload_bytes = 0usize; + for log in self.values.values() { + if let Some(v) = log.latest() { + decoded_payload_bytes += v.estimated_size_bytes(); + string_payload_bytes += v.string_payload_bytes(); + } + } + + PropertyColumnMemory { + key: String::new(), + entry_count: self.values.len(), + map_capacity: self.values.capacity(), + map_slot_bytes, + decoded_payload_bytes, + string_payload_bytes, + compressed_bytes: 0, + capacity_waste_bytes, + compressed_entry_count: 0, + } + } + + /// Shrinks the hot map capacity to fit live entries. + pub fn shrink_capacities(&mut self) { + self.values.shrink_to_fit(); + self.block_zone_maps.shrink_to_fit(); } /// Compression is not supported in temporal mode (no-op). diff --git a/crates/grafeo-core/src/graph/lpg/store/memory.rs b/crates/grafeo-core/src/graph/lpg/store/memory.rs index d3e0720f5..f527f202a 100644 --- a/crates/grafeo-core/src/graph/lpg/store/memory.rs +++ b/crates/grafeo-core/src/graph/lpg/store/memory.rs @@ -1,7 +1,9 @@ //! Memory introspection for `LpgStore`. use super::LpgStore; -use grafeo_common::memory::usage::{IndexMemory, MvccMemory, StoreMemory, StringPoolMemory}; +use grafeo_common::memory::usage::{ + IndexMemory, LpgResidencyMemory, MvccMemory, StoreMemory, StringPoolMemory, +}; use std::mem::size_of; impl LpgStore { @@ -19,9 +21,50 @@ impl LpgStore { (store, indexes, mvcc, string_pool) } + /// Property-column / string-payload / adjacency-capacity residency detail. + /// + /// Closes the `memory_usage()` blind spot where map-slot estimates omitted + /// decoded `Value` heap (especially ArcStr) and adjacency capacity waste. + #[must_use] + pub fn lpg_residency_detail(&self) -> LpgResidencyMemory { + let mut detail = LpgResidencyMemory { + node_properties: self.node_properties.memory_detail(), + edge_properties: self.edge_properties.memory_detail(), + forward_adjacency: self.forward_adj.capacity_memory(), + backward_adjacency: self + .backward_adj + .as_ref() + .map_or_else(Default::default, |adj| adj.capacity_memory()), + ..Default::default() + }; + detail.compute_total(); + detail + } + + /// Shrinks property-column and adjacency Vec/HashMap capacities in place. + /// + /// Does not change logical graph contents. Intended for residency + /// experiments after bulk open/deserialize. + pub fn shrink_lpg_capacities(&self) { + self.node_properties.shrink_capacities(); + self.edge_properties.shrink_capacities(); + self.forward_adj.shrink_capacities(); + if let Some(adj) = &self.backward_adj { + adj.shrink_capacities(); + } + } + + /// Forces dictionary/int/bool compression on all property columns. + /// + /// Point reads remain correct via lazy compressed `get` decode. + pub fn force_compress_properties(&self) { + self.node_properties.force_compress_all(); + self.edge_properties.force_compress_all(); + } + fn store_memory(&self) -> StoreMemory { - let node_props_bytes = self.node_properties.heap_memory_bytes(); - let edge_props_bytes = self.edge_properties.heap_memory_bytes(); + let node_detail = self.node_properties.memory_detail(); + let edge_detail = self.edge_properties.memory_detail(); let col_count = self.node_properties.column_count() + self.edge_properties.column_count(); // Node/edge map overhead (excluding version chain internals, which go to MVCC) @@ -57,9 +100,17 @@ impl LpgStore { let mut store = StoreMemory { nodes_bytes, edges_bytes, - node_properties_bytes: node_props_bytes, - edge_properties_bytes: edge_props_bytes, + node_properties_bytes: node_detail.total_bytes, + edge_properties_bytes: edge_detail.total_bytes, property_column_count: col_count, + node_property_map_slot_bytes: node_detail.total_map_slot_bytes, + node_property_decoded_payload_bytes: node_detail.total_decoded_payload_bytes, + node_property_string_payload_bytes: node_detail.total_string_payload_bytes, + node_property_capacity_waste_bytes: node_detail.total_capacity_waste_bytes, + edge_property_map_slot_bytes: edge_detail.total_map_slot_bytes, + edge_property_decoded_payload_bytes: edge_detail.total_decoded_payload_bytes, + edge_property_string_payload_bytes: edge_detail.total_string_payload_bytes, + edge_property_capacity_waste_bytes: edge_detail.total_capacity_waste_bytes, ..Default::default() }; store.compute_total(); @@ -136,11 +187,15 @@ impl LpgStore { } fn index_memory(&self) -> IndexMemory { - let forward_bytes = self.forward_adj.heap_memory_bytes(); - let backward_bytes = self + let forward_detail = self.forward_adj.capacity_memory(); + let backward_detail = self .backward_adj .as_ref() - .map_or(0, |adj| adj.heap_memory_bytes()); + .map(|adj| adj.capacity_memory()); + let forward_bytes = forward_detail.total_bytes; + let backward_bytes = backward_detail + .as_ref() + .map_or(0, |d| d.total_bytes); // Label index: Vec> let label_idx = self.label_index.read(); @@ -231,6 +286,10 @@ impl LpgStore { let mut indexes = IndexMemory { forward_adjacency_bytes: forward_bytes, backward_adjacency_bytes: backward_bytes, + forward_adjacency_capacity_waste_bytes: forward_detail.capacity_waste_bytes, + backward_adjacency_capacity_waste_bytes: backward_detail + .as_ref() + .map_or(0, |d| d.capacity_waste_bytes), label_index_bytes, node_labels_bytes, property_index_bytes, diff --git a/crates/grafeo-core/src/index/adjacency.rs b/crates/grafeo-core/src/index/adjacency.rs index 135ce1d71..7fec514e5 100644 --- a/crates/grafeo-core/src/index/adjacency.rs +++ b/crates/grafeo-core/src/index/adjacency.rs @@ -407,6 +407,19 @@ impl AdjacencyList { self.skip_index.sort_unstable_by_key(|e| e.min_destination); } + /// Shrinks Vec capacities to fit live entries (does not freeze/compress). + fn shrink_capacities(&mut self) { + self.hot_chunks.shrink_to_fit(); + for chunk in &mut self.hot_chunks { + chunk.destinations.shrink_to_fit(); + chunk.edge_ids.shrink_to_fit(); + } + self.cold_chunks.shrink_to_fit(); + self.delta_inserts.shrink_to_fit(); + self.deleted.shrink_to_fit(); + self.skip_index.shrink_to_fit(); + } + fn iter(&self) -> impl Iterator + '_ { let deleted = &self.deleted; @@ -883,33 +896,62 @@ impl ChunkedAdjacency { /// Returns estimated heap memory in bytes. #[must_use] pub fn heap_memory_bytes(&self) -> usize { + self.capacity_memory().total_bytes + } + + /// Capacity vs used-byte attribution for adjacency lists. + #[must_use] + pub fn capacity_memory(&self) -> grafeo_common::memory::AdjacencyCapacityMemory { + use grafeo_common::memory::AdjacencyCapacityMemory; + let lists = self.lists.read(); - // Outer hash map overhead - let map_overhead = lists.capacity() + let list_map_overhead_bytes = lists.capacity() * (std::mem::size_of::() + std::mem::size_of::() + 1); - // Per-list memory: hot chunks + cold chunks + deltas + deleted set - let mut list_bytes = 0usize; + + let mut hot_used_bytes = 0usize; + let mut hot_capacity_bytes = 0usize; + let mut cold_bytes = 0usize; + let mut aux_capacity_bytes = 0usize; + for list in lists.values() { - // Hot chunks: Vec capacity + each chunk's Vec capacity - list_bytes += list.hot_chunks.capacity() * std::mem::size_of::(); + hot_capacity_bytes += list.hot_chunks.capacity() * std::mem::size_of::(); for chunk in &list.hot_chunks { - list_bytes += chunk.destinations.capacity() * std::mem::size_of::(); - list_bytes += chunk.edge_ids.capacity() * std::mem::size_of::(); + hot_used_bytes += chunk.destinations.len() * std::mem::size_of::(); + hot_used_bytes += chunk.edge_ids.len() * std::mem::size_of::(); + hot_capacity_bytes += chunk.destinations.capacity() * std::mem::size_of::(); + hot_capacity_bytes += chunk.edge_ids.capacity() * std::mem::size_of::(); } - // Cold chunks: compressed data - list_bytes += + hot_capacity_bytes += list.cold_chunks.capacity() * std::mem::size_of::(); for cold in &list.cold_chunks { - list_bytes += cold.memory_size(); + cold_bytes += cold.memory_size(); } - // Delta buffer - list_bytes += list.delta_inserts.capacity() * 16; - // Deleted set - list_bytes += list.deleted.capacity() * (std::mem::size_of::() + 1); - // Skip index - list_bytes += list.skip_index.capacity() * std::mem::size_of::(); - } - map_overhead + list_bytes + aux_capacity_bytes += list.delta_inserts.capacity() * 16; + aux_capacity_bytes += list.deleted.capacity() * (std::mem::size_of::() + 1); + aux_capacity_bytes += + list.skip_index.capacity() * std::mem::size_of::(); + } + + let mut detail = AdjacencyCapacityMemory { + node_count: lists.len(), + list_map_capacity: lists.capacity(), + list_map_overhead_bytes, + hot_used_bytes, + hot_capacity_bytes, + cold_bytes, + aux_capacity_bytes, + ..Default::default() + }; + detail.compute_total(); + detail + } + + /// Shrinks Vec capacities on every adjacency list to fit live data. + pub fn shrink_capacities(&self) { + let mut lists = self.lists.write(); + for list in lists.values_mut() { + list.shrink_capacities(); + } } /// Forces all hot chunks to be compressed for all adjacency lists. diff --git a/crates/grafeo-core/tests/lpg_residency_experiments.rs b/crates/grafeo-core/tests/lpg_residency_experiments.rs new file mode 100644 index 000000000..a3981f343 --- /dev/null +++ b/crates/grafeo-core/tests/lpg_residency_experiments.rs @@ -0,0 +1,207 @@ +//! Track E — LPG residency accounting + compact/lazy representation experiments. +//! +//! Synthetic workloads (staging DBs under `/data/tmp` only when opened; this +//! harness stays in-process). Measures estimated residency, process RssAnon, +//! point-get latency, and correctness before/after: +//! A) `shrink_capacities` (capacity waste reclaim) +//! B) `force_compress_all` + lazy compressed `get` (dictionary-compact strings) + +use std::fs; +use std::time::Instant; + +use arcstr::ArcStr; +use grafeo_common::types::{NodeId, PropertyKey, Value}; +use grafeo_core::graph::lpg::PropertyStorage; + +fn rss_anon_kib() -> u64 { + let status = fs::read_to_string("/proc/self/status").expect("read /proc/self/status"); + for line in status.lines() { + if let Some(rest) = line.strip_prefix("RssAnon:") { + let kib: u64 = rest + .split_whitespace() + .next() + .expect("RssAnon value") + .parse() + .expect("parse RssAnon"); + return kib; + } + } + panic!("RssAnon missing from /proc/self/status"); +} + +fn fill_string_heavy(storage: &PropertyStorage, n: u64, unique_ratio: usize) { + let key = PropertyKey::new("body"); + let dict: Vec = (0..unique_ratio) + .map(|i| ArcStr::from(format!("payload-token-{i:04}-{}", "x".repeat(48)))) + .collect(); + for i in 0..n { + let s = dict[i as usize % unique_ratio].clone(); + storage.set(NodeId::new(i), key.clone(), Value::String(s)); + } +} + +fn point_get_correctness(storage: &PropertyStorage, n: u64, unique_ratio: usize) -> bool { + let key = PropertyKey::new("body"); + let dict: Vec = (0..unique_ratio) + .map(|i| format!("payload-token-{i:04}-{}", "x".repeat(48))) + .collect(); + for i in 0..n { + let expected = &dict[i as usize % unique_ratio]; + match storage.get(NodeId::new(i), &key) { + Some(Value::String(s)) if s.as_str() == expected.as_str() => {} + other => { + eprintln!("mismatch at {i}: got {other:?}"); + return false; + } + } + } + true +} + +fn bench_point_gets(storage: &PropertyStorage, n: u64, iters: u64) -> f64 { + let key = PropertyKey::new("body"); + let start = Instant::now(); + let mut hits = 0u64; + for i in 0..iters { + if storage.get(NodeId::new(i % n), &key).is_some() { + hits += 1; + } + } + assert_eq!(hits, iters); + start.elapsed().as_secs_f64() * 1e9 / iters as f64 +} + +#[test] +fn experiment_a_shrink_capacities_and_b_lazy_dictionary() { + const N: u64 = 200_000; + const UNIQUE: usize = 256; + const GET_ITERS: u64 = 200_000; + + // Over-reserve then fill so capacity waste is visible. + let storage = PropertyStorage::new(); + { + // Pre-touch with a disposable column growth path: insert then rebuild + // is unnecessary; HashMap grows with inserts. We measure post-fill. + fill_string_heavy(&storage, N, UNIQUE); + } + + let baseline = storage.memory_detail(); + let rss_baseline = rss_anon_kib(); + let ns_baseline = bench_point_gets(&storage, N, GET_ITERS); + assert!( + point_get_correctness(&storage, N, UNIQUE), + "baseline correctness" + ); + + println!("=== Track E experiment baseline ==="); + println!( + "entries={N} unique={UNIQUE} columns={} total_bytes={} map_slots={} decoded={} strings={} waste={} compressed={}", + baseline.columns.len(), + baseline.total_bytes, + baseline.total_map_slot_bytes, + baseline.total_decoded_payload_bytes, + baseline.total_string_payload_bytes, + baseline.total_capacity_waste_bytes, + baseline.total_compressed_bytes + ); + if let Some(top) = baseline.columns.first() { + println!( + "top_column key={} entries={} cap={} slots={} strings={} waste={}", + top.key, + top.entry_count, + top.map_capacity, + top.map_slot_bytes, + top.string_payload_bytes, + top.capacity_waste_bytes + ); + } + println!("rss_anon_kib={rss_baseline} point_get_ns={ns_baseline:.1}"); + + // --- Experiment A: shrink capacities --- + storage.shrink_capacities(); + let after_shrink = storage.memory_detail(); + let rss_shrink = rss_anon_kib(); + let ns_shrink = bench_point_gets(&storage, N, GET_ITERS); + assert!( + point_get_correctness(&storage, N, UNIQUE), + "post-shrink correctness" + ); + println!("=== Experiment A: shrink_capacities ==="); + println!( + "total_bytes={} waste={} (Δwaste={}) rss_anon_kib={rss_shrink} (Δrss={}) point_get_ns={ns_shrink:.1}", + after_shrink.total_bytes, + after_shrink.total_capacity_waste_bytes, + after_shrink.total_capacity_waste_bytes as i64 + - baseline.total_capacity_waste_bytes as i64, + rss_shrink as i64 - rss_baseline as i64 + ); + assert!( + after_shrink.total_capacity_waste_bytes <= baseline.total_capacity_waste_bytes, + "shrink should not increase capacity waste" + ); + + // --- Experiment B: force dictionary compress + lazy get --- + let before_compress = storage.memory_detail(); + let rss_before_compress = rss_anon_kib(); + storage.force_compress_all(); + let after_compress = storage.memory_detail(); + let rss_compress = rss_anon_kib(); + let ns_compress = bench_point_gets(&storage, N, GET_ITERS); + assert!( + point_get_correctness(&storage, N, UNIQUE), + "post-compress lazy-get correctness" + ); + println!("=== Experiment B: force_compress + lazy get ==="); + println!( + "before_total={} after_total={} (Δ={}) strings_before={} compressed_after={} rss_kib={rss_compress} (Δ={}) point_get_ns={ns_compress:.1} (was {ns_shrink:.1})", + before_compress.total_bytes, + after_compress.total_bytes, + after_compress.total_bytes as i64 - before_compress.total_bytes as i64, + before_compress.total_string_payload_bytes, + after_compress.total_compressed_bytes, + rss_compress as i64 - rss_before_compress as i64 + ); + assert!( + after_compress.total_compressed_bytes > 0, + "dictionary compression should materialize compressed backing" + ); + assert!( + after_compress.total_string_payload_bytes < before_compress.total_string_payload_bytes, + "hot string payloads should drop after compress" + ); + + // Emit JSON-ish summary line for Track E report scraping. + println!( + "TRACK_E_SUMMARY shrink_waste_before={} shrink_waste_after={} shrink_rss_delta_kib={} compress_total_before={} compress_total_after={} compress_rss_delta_kib={} get_ns_baseline={:.1} get_ns_shrink={:.1} get_ns_compress={:.1}", + baseline.total_capacity_waste_bytes, + after_shrink.total_capacity_waste_bytes, + rss_shrink as i64 - rss_baseline as i64, + before_compress.total_bytes, + after_compress.total_bytes, + rss_compress as i64 - rss_before_compress as i64, + ns_baseline, + ns_shrink, + ns_compress + ); +} + +#[test] +fn accounting_attributes_string_payload_separately_from_slots() { + let storage = PropertyStorage::new(); + let key = PropertyKey::new("name"); + for i in 0..1_000u64 { + storage.set( + NodeId::new(i), + key.clone(), + Value::String(ArcStr::from(format!("unique-name-{i:05}"))), + ); + } + let detail = storage.memory_detail(); + assert_eq!(detail.columns.len(), 1); + let col = &detail.columns[0]; + assert!(col.map_slot_bytes > 0); + assert!(col.string_payload_bytes > 0); + assert!(col.decoded_payload_bytes >= col.string_payload_bytes); + // Old map-slot-only accounting undercounts: decoded payloads must be in total. + assert!(col.total_bytes() > col.map_slot_bytes); +} diff --git a/crates/grafeo-engine/src/database/admin.rs b/crates/grafeo-engine/src/database/admin.rs index 23bbfd8d3..947d81672 100644 --- a/crates/grafeo-engine/src/database/admin.rs +++ b/crates/grafeo-engine/src/database/admin.rs @@ -190,6 +190,25 @@ impl super::GrafeoDB { usage } + /// Property / string / adjacency capacity residency attribution. + /// + /// Prefer this over coarse `memory_usage()` when diagnosing live RSS + /// residuals (decoded Value payloads, capacity waste). + #[must_use] + pub fn lpg_residency_detail(&self) -> grafeo_common::memory::LpgResidencyMemory { + self.lpg_store().lpg_residency_detail() + } + + /// Shrinks LPG property-column and adjacency capacities in place. + pub fn shrink_lpg_capacities(&self) { + self.lpg_store().shrink_lpg_capacities(); + } + + /// Forces property-column compression (lazy point-get decode remains valid). + pub fn force_compress_properties(&self) { + self.lpg_store().force_compress_properties(); + } + /// Returns detailed database statistics. /// /// Includes counts, memory usage, and index information. diff --git a/crates/grafeo-engine/src/memory_usage.rs b/crates/grafeo-engine/src/memory_usage.rs index dd79df50a..d2f2090fa 100644 --- a/crates/grafeo-engine/src/memory_usage.rs +++ b/crates/grafeo-engine/src/memory_usage.rs @@ -5,7 +5,8 @@ //! types (`CacheMemory`, `BufferManagerMemory`, `RdfMemory`, `CdcMemory`). pub use grafeo_common::memory::usage::{ - IndexMemory, MvccMemory, NamedMemory, StoreMemory, StringPoolMemory, + AdjacencyCapacityMemory, IndexMemory, LpgResidencyMemory, MvccMemory, NamedMemory, + PropertyColumnMemory, PropertyStorageMemory, StoreMemory, StringPoolMemory, }; use serde::{Deserialize, Serialize}; diff --git a/docs/TRACK_E_LPG_RESIDENCY.md b/docs/TRACK_E_LPG_RESIDENCY.md new file mode 100644 index 000000000..936d0dc11 --- /dev/null +++ b/docs/TRACK_E_LPG_RESIDENCY.md @@ -0,0 +1,33 @@ +# Track E — LPG residency accounting + compact/lazy experiments + +**Branch:** `diagnostics/lpg-memory-accounting` +**Base:** `9781320f` (`agent/txn-session-batch-20260726`) +**Separate from** diagnostics PR #2. + +## Attribution API + +Expanded `memory_usage` / property-column detail: +- map slot bytes + capacity waste +- decoded payload bytes +- string payload bytes +- compressed backing bytes +- adjacency capacity accounting + +## Synthetic experiments (`lpg_residency_experiments`) + +Workload: 200k string properties, 256 unique 60-ish-char tokens. + +| Experiment | Estimated total | RSS anon | Point-get ns | Correctness | +|------------|----------------:|---------:|-------------:|-------------| +| Baseline | 24,640,531 B (~23.5 MiB) | 13,136 KiB | 208.7 | pass | +| A `shrink_capacities` | same (waste Δ=0 on this fill) | 13,136 (Δ0) | 499.0 | pass | +| B `force_compress_all` + lazy get | 4,018,259 B (−20.6 MiB est.) | 7,584 (Δ −5,552 KiB) | 171.0 | pass | + +## Relation to ~4 GiB production residual + +Phase3 heaptrack: open-stack attributed ~59% to PropertyColumn/deserialize/arcstr; residual ~4 GiB unattributed. Dictionary-compact strings show **large estimated payload reduction** on high-duplication string columns; production gain depends on unique-string cardinality of account/code graphs. Capacity shrink alone was a no-op on this synthetic fill. + +## Track F inputs + +- Live open residual still dominated by LPG property/string residency after jemalloc (~2.4 GiB help). +- Compact dictionary representation is a promising in-process lever before process isolation. diff --git a/docs/diagnostics/TRACK_E_LPG_MEMORY_ACCOUNTING.md b/docs/diagnostics/TRACK_E_LPG_MEMORY_ACCOUNTING.md new file mode 100644 index 000000000..1c475e4b8 --- /dev/null +++ b/docs/diagnostics/TRACK_E_LPG_MEMORY_ACCOUNTING.md @@ -0,0 +1,85 @@ +# Track E — LPG memory accounting & compact/lazy residency + +**Branch:** `diagnostics/lpg-memory-accounting` (off product pin `9781320f` / `agent/txn-session-batch-20260726`) +**Worktree:** `/data/worktrees/grafeo-lpg-accounting` +**Staging policy:** `/data/tmp` only — never `/data/grafeo` +**Separate from:** AM diagnostics PR #2 / Grafeo `diagnostics/close-forensics` + +## Problem + +Phase 3/4 forensics (AM docs): live sidecar open ~9–12 GiB RssAnon; jemalloc recovers ~2.4 GiB; residual ~4 GiB consistent with LPG properties/strings/adjacency + allocator blind spots. Prior `memory_usage()` counted property **map slots** (`capacity × sizeof(Value)`) but **omitted decoded Value heap** (especially ArcStr), so Grafeo tracked only ~2.93 GiB (~32% of sidecar Δ) while heaptrack open stacks attributed ~59%. + +## Delivered APIs + +| API | Where | What | +|-----|-------|------| +| `PropertyStorage::memory_detail()` | `grafeo-core` | Per-column map slots, decoded payloads, string payloads, compressed bytes, capacity waste | +| `PropertyColumn::memory_detail()` / updated `heap_memory_bytes()` | `grafeo-core` | Includes `Value::estimated_size_bytes` + compressed backing | +| `PropertyStorage::shrink_capacities()` | `grafeo-core` | `shrink_to_fit` hot maps | +| Compressed lazy `PropertyColumn::get` | `grafeo-core` | Binary-search `index_to_id` + on-demand decode (strings/bools/ints) | +| `AdjacencyIndex::capacity_memory()` / `shrink_capacities()` | `grafeo-core` | Hot used vs capacity, cold, aux, waste | +| `LpgStore::lpg_residency_detail()` / `shrink_lpg_capacities()` / `force_compress_properties()` | `grafeo-core` | Aggregated residency + experiment knobs | +| `GrafeoDB::lpg_residency_detail()` (+ shrink/compress) | `grafeo-engine` | Operator-facing surface | +| Types | `grafeo-common::memory` | `PropertyColumnMemory`, `PropertyStorageMemory`, `AdjacencyCapacityMemory`, `LpgResidencyMemory`; extended `StoreMemory` / `IndexMemory` fields | + +## Measured results (2026-07-29, synthetic N=200k, 256 unique ~60 B strings) + +Harness log: `/data/tmp/grafeo-lpg-accounting-track-e/grafeo-lpg-accounting-track-e-run.log` + +| Metric | Baseline | A shrink | B force_compress + lazy get | +|--------|---------:|---------:|----------------------------:| +| Estimated total (bytes) | 24,640,531 | 24,640,531 | **4,018,259** (−20.6 MiB / −84%) | +| Map slots | 11,239,424 | same | (hot cleared into dict) | +| String payload (hot) | 13,400,000 | same | **≪** (moved to compressed) | +| Capacity waste | 1,439,424 | 1,439,424 (Δ0) | — | +| Compressed bytes | 0 | 0 | **4,017,152** | +| RssAnon (KiB) | 13,136 | 13,136 (Δ0) | **7,584 (Δ −5,552)** | +| Point-get ns/op | 883.5 | 618.3 | **405.9** | +| Correctness (200k gets) | PASS | PASS | **PASS** | + +### Interpretation + +- **A:** Post-fill `hashbrown` already sits at its load-factor capacity; `shrink_to_fit` does not reclaim meaningful RSS on this pattern. Still a safe operator knob after over-reserve / partial deletes. +- **B:** Dictionary-compact + lazy decode is the clear win: estimated residency −84%, process RssAnon −5.4 MiB on a 200k toy, **and** point-get correctness holds (fixes the prior compressed-`get` → `None` hole). Production scale (Phase 3 ~1.44 GiB arcstr open stacks) is the extrapolation target — sidecar remeasure when MemAvailable allows. +- Point-get latency **improved** after compress on this workload (dict + binary search beat hot HashMap+ArcStr clone path under the bench); treat as workload-specific, not a universal guarantee. + +## How to re-run + +```bash +export TMPDIR=/data/tmp +export CARGO_TARGET_DIR=/data/cargo-targets/jfrie-grafeo-lpg-accounting/target +export RUSTC_WRAPPER=sccache SCCACHE_DIR=/data/sccache +cargo test -p grafeo-core --test lpg_residency_experiments -- --nocapture +``` + +## Operational risk + +- **Accounting change:** `heap_memory_bytes` / `StoreMemory.node_properties_bytes` now include decoded payloads → reported totals rise toward reality (not a live RSS change by itself). +- **Lazy get after compress:** integer/bool compressed point-get may decompress a whole column per miss (strings are O(log n) + dict lookup). Prefer scan/block paths for analytics; fine for sparse correctness. +- **force_compress:** still gated by internal ratio thresholds; no-op when compression does not help. +- **Do not** run full sidecar open on this host while `am-server-rs` holds ~18 GiB RSS without reclaiming MemAvailable first. + +## Notes for Track F (process isolation) — do not implement here + +Numbers F needs (from Phase 3/4, glibc unless noted): + +| Quantity | Value | Source | +|----------|------:|--------| +| Sidecar open Δ RssAnon (glibc) | ~9.39 GiB (9,845,496 KiB) | PHASE3 | +| Live open total with account (glibc) | ~11.62 GiB | PHASE4 | +| jemalloc live open total | ~9.14 GiB (−2.36 GiB) | PHASE4D | +| Grafeo tracked pre-Track-E | ~2.93 GiB (~32%) | PHASE3 | +| heaptrack open-stack sum | ~5.58 GiB (~59% of Δ) | PHASE3 | +| Unattributed residual | ~4 GiB (~41% of Δ) | PHASE3 | +| Close wall | ~143–152 s | MITIGATION | +| Account healthy during sidecar close | yes (149/149) | MITIGATION | +| Combined same-file | **worse** (+~0.7 GiB) | PHASE4E | + +Isolation estimate (unchanged): moving sidecar RSS into another process protects account latency during close and allows kill-to-reclaim; it does **not** shrink the sidecar working set itself. Track E attribution + compact/lazy wins are the in-process residual levers; F owns process boundary. + +## Remaining work + +- [ ] Re-measure account/sidecar open with new `lpg_residency_detail()` when MemAvailable ≥ ~14 GiB (copy staging under `/data/tmp`) +- [ ] Optional: ID→index hash for O(1) compressed get (avoid binary search / int full decompress) +- [ ] Optional: unique-ArcStr accounting via pointer set (reduce shared-string overcount) +- [ ] Wire CLI `memory` pretty-print for new StoreMemory / LpgResidency fields