graph_d 1.3.2

//! Advanced memory management for the graph database.
//!
//! This module provides custom allocators, memory pools, and efficient caching
//! to meet the ≤ 1GB memory target for 1M documents while maintaining high performance.
//!
//! # Architecture
//!
//! - **Arena Allocators**: Fast bump allocation for temporary objects
//! - **Memory Pools**: Specialized pools for nodes, relationships, and properties
//! - **Off-heap Storage**: Topology data stored outside the main heap
//! - **LRU Caching**: Intelligent caching with automatic eviction
//!
//! # Memory Efficiency Strategies
//!
//! 1. **Bump Allocation**: O(1) allocation for query processing
//! 2. **Pool Management**: Reuse of fixed-size blocks for graph elements
//! 3. **Cache Locality**: Hot data kept in fast-access caches
//! 4. **String Interning**: Deduplication of common property values

// Safe allocators that don't require unsafe code
pub mod allocator_safe;
pub mod batch;
pub mod cache;
pub mod interning;
pub mod profiler;

// INT-4: Re-export Batch types for public API (Satisfies: API completeness, RT-5, TN2)
pub use batch::{BatchContext, BatchManager, BatchMemoryTracker, BatchStats};

// Runtime memory profiling (Satisfies: RT-5, B3, GAP-6)
pub use profiler::{
    AllocationCategory, AllocationEvent, MemoryProfiler, MemorySnapshot, SharedMemoryProfiler,
};

// Unsafe allocators for maximum performance (optional)
#[cfg(feature = "unsafe-allocators")]
pub mod allocator;
#[cfg(feature = "unsafe-allocators")]
pub mod pool;

use crate::error::{GraphError, Result};
use crate::graph::Id;
use bumpalo::Bump;
use lru::LruCache;
use parking_lot::RwLock;
use std::num::NonZeroUsize;
use std::sync::Arc;

/// Handle for arena-allocated data.
///
/// This provides safe access to data allocated in the arena without
/// requiring lifetime management.
pub struct ArenaHandle<T> {
    value: T,
}

impl<T> ArenaHandle<T> {
    fn new(value: T) -> Self {
        Self { value }
    }

    /// Get a reference to the allocated value.
    pub fn get(&self) -> &T {
        &self.value
    }

    /// Get a mutable reference to the allocated value.
    pub fn get_mut(&mut self) -> &mut T {
        &mut self.value
    }
}

/// Memory management configuration for the graph database.
#[derive(Debug, Clone)]
pub struct MemoryConfig {
    /// Maximum memory budget in bytes (default: 1GB)
    pub max_memory_bytes: usize,
    /// Arena size for temporary allocations
    pub arena_size_bytes: usize,
    /// Node cache capacity
    pub node_cache_capacity: usize,
    /// Relationship cache capacity  
    pub relationship_cache_capacity: usize,
    /// Property cache capacity
    pub property_cache_capacity: usize,
    /// String interning table size
    pub string_table_capacity: usize,
}

impl Default for MemoryConfig {
    fn default() -> Self {
        Self {
            max_memory_bytes: 1024 * 1024 * 1024, // 1GB
            arena_size_bytes: 64 * 1024 * 1024,   // 64MB
            node_cache_capacity: 100_000,
            relationship_cache_capacity: 400_000,
            property_cache_capacity: 200_000,
            string_table_capacity: 50_000,
        }
    }
}

/// Central memory manager for the graph database.
/// Satisfies: RT-5 (memory usage within bounds)
/// Satisfies: B3 (memory efficiency goal)
#[allow(clippy::arc_with_non_send_sync)]
pub struct MemoryManager {
    config: MemoryConfig,
    /// Arena allocator for temporary allocations during query processing
    arena: Arc<RwLock<Bump>>,
    /// LRU cache for frequently accessed nodes
    node_cache: Arc<RwLock<LruCache<Id, Vec<u8>>>>,
    /// LRU cache for frequently accessed relationships
    relationship_cache: Arc<RwLock<LruCache<Id, Vec<u8>>>>,
    /// String interning table for memory deduplication
    string_interner: Arc<RwLock<interning::StringInterner>>,
    /// Memory usage tracking
    memory_stats: Arc<RwLock<MemoryStats>>,
    /// INT-3: Batch manager for efficient batch allocations
    /// Satisfies: RT-5, B3, TN2 (memory efficiency through batch operations)
    batch_manager: Arc<RwLock<batch::BatchManager>>,
}

/// Memory usage statistics for monitoring and optimization.
#[derive(Debug, Default)]
pub struct MemoryStats {
    /// Total bytes allocated
    pub total_allocated: usize,
    /// Bytes used by node cache
    pub node_cache_bytes: usize,
    /// Bytes used by relationship cache
    pub relationship_cache_bytes: usize,
    /// Bytes used by string interning
    pub string_table_bytes: usize,
    /// Arena memory usage
    pub arena_bytes: usize,
    /// Cache hit rate (0.0 - 1.0)
    pub cache_hit_rate: f64,
    /// Number of cache hits
    pub cache_hits: u64,
    /// Number of cache misses
    pub cache_misses: u64,
}

impl MemoryManager {
    /// Create a new memory manager with the given configuration.
    #[allow(clippy::arc_with_non_send_sync)]
    pub fn new(config: MemoryConfig) -> Result<Self> {
        let node_cache_cap = NonZeroUsize::new(config.node_cache_capacity)
            .ok_or_else(|| GraphError::Memory("Invalid node cache capacity".to_string()))?;
        let rel_cache_cap = NonZeroUsize::new(config.relationship_cache_capacity)
            .ok_or_else(|| GraphError::Memory("Invalid relationship cache capacity".to_string()))?;

        Ok(Self {
            arena: Arc::new(RwLock::new(Bump::with_capacity(config.arena_size_bytes))),
            node_cache: Arc::new(RwLock::new(LruCache::new(node_cache_cap))),
            relationship_cache: Arc::new(RwLock::new(LruCache::new(rel_cache_cap))),
            string_interner: Arc::new(RwLock::new(interning::StringInterner::new(
                config.string_table_capacity,
            ))),
            memory_stats: Arc::new(RwLock::new(MemoryStats::default())),
            // INT-3: Initialize batch manager for efficient batch operations
            // Satisfies: RT-5, B3, TN2 (memory efficiency through batch operations)
            batch_manager: Arc::new(RwLock::new(batch::BatchManager::default())),
            config,
        })
    }

    /// Create a new memory manager with default configuration.
    pub fn with_defaults() -> Result<Self> {
        Self::new(MemoryConfig::default())
    }

    /// Get a reference to the arena allocator for temporary allocations.
    pub fn arena(&self) -> &Arc<RwLock<Bump>> {
        &self.arena
    }

    /// Reset the arena allocator, freeing all temporary allocations.
    /// This should be called after query processing to reclaim memory.
    pub fn reset_arena(&self) {
        let mut arena = self.arena.write();
        arena.reset();

        // Update memory statistics
        let mut stats = self.memory_stats.write();
        stats.arena_bytes = 0;
    }

    /// Allocate memory in the arena for temporary use during query processing.
    /// Returns a handle that can be used to access the allocated data.
    pub fn arena_alloc<T>(&self, value: T) -> ArenaHandle<T> {
        // In a real implementation, this would need more sophisticated handling
        // For now, we'll use a simple approach that stores the value
        ArenaHandle::new(value)
    }

    /// Get a cached node by ID, returning None if not in cache.
    pub fn get_cached_node(&self, id: Id) -> Option<Vec<u8>> {
        let mut cache = self.node_cache.write();
        let result = cache.get(&id).cloned();

        // Update statistics
        let mut stats = self.memory_stats.write();
        if result.is_some() {
            stats.cache_hits += 1;
        } else {
            stats.cache_misses += 1;
        }
        stats.cache_hit_rate =
            stats.cache_hits as f64 / (stats.cache_hits + stats.cache_misses) as f64;

        result
    }

    /// Cache a node for fast future access.
    pub fn cache_node(&self, id: Id, data: Vec<u8>) {
        let data_size = data.len();
        let mut cache = self.node_cache.write();
        cache.put(id, data);

        // Update statistics
        let mut stats = self.memory_stats.write();
        stats.node_cache_bytes += data_size;
    }

    /// Get a cached relationship by ID, returning None if not in cache.
    pub fn get_cached_relationship(&self, id: Id) -> Option<Vec<u8>> {
        let mut cache = self.relationship_cache.write();
        let result = cache.get(&id).cloned();

        // Update statistics
        let mut stats = self.memory_stats.write();
        if result.is_some() {
            stats.cache_hits += 1;
        } else {
            stats.cache_misses += 1;
        }
        stats.cache_hit_rate =
            stats.cache_hits as f64 / (stats.cache_hits + stats.cache_misses) as f64;

        result
    }

    /// Cache a relationship for fast future access.
    pub fn cache_relationship(&self, id: Id, data: Vec<u8>) {
        let data_size = data.len();
        let mut cache = self.relationship_cache.write();
        cache.put(id, data);

        // Update statistics
        let mut stats = self.memory_stats.write();
        stats.relationship_cache_bytes += data_size;
    }

    /// Intern a string to reduce memory duplication.
    pub fn intern_string(&self, s: &str) -> u32 {
        let interner = self.string_interner.write();
        interner.intern(s)
    }

    /// Get an interned string by its ID.
    pub fn get_interned_string(&self, id: u32) -> Option<String> {
        let interner = self.string_interner.read();
        interner.get(id)
    }

    /// Get current memory usage statistics.
    pub fn memory_stats(&self) -> MemoryStats {
        let stats = self.memory_stats.read();
        MemoryStats {
            total_allocated: stats.total_allocated,
            node_cache_bytes: stats.node_cache_bytes,
            relationship_cache_bytes: stats.relationship_cache_bytes,
            string_table_bytes: stats.string_table_bytes,
            arena_bytes: stats.arena_bytes,
            cache_hit_rate: stats.cache_hit_rate,
            cache_hits: stats.cache_hits,
            cache_misses: stats.cache_misses,
        }
    }

    /// Check if we're approaching memory limits and trigger cleanup if needed.
    pub fn check_memory_pressure(&self) -> Result<()> {
        let stats = self.memory_stats.read();
        let total_used = stats.total_allocated;

        if total_used > (self.config.max_memory_bytes * 9) / 10 {
            // We're at 90% capacity, trigger aggressive cleanup
            drop(stats);
            self.cleanup_memory()?;
        }

        Ok(())
    }

    /// Perform memory cleanup to free unused resources.
    pub fn cleanup_memory(&self) -> Result<()> {
        // Reset arena to free temporary allocations
        self.reset_arena();

        // Clear half of each cache to free up space
        {
            let mut node_cache = self.node_cache.write();
            let current_size = node_cache.len();
            for _ in 0..(current_size / 2) {
                node_cache.pop_lru();
            }
        }

        {
            let mut rel_cache = self.relationship_cache.write();
            let current_size = rel_cache.len();
            for _ in 0..(current_size / 2) {
                rel_cache.pop_lru();
            }
        }

        // Update statistics
        let mut stats = self.memory_stats.write();
        stats.node_cache_bytes /= 2;
        stats.relationship_cache_bytes /= 2;
        stats.total_allocated =
            stats.node_cache_bytes + stats.relationship_cache_bytes + stats.string_table_bytes;

        Ok(())
    }

    /// Get memory configuration.
    pub fn config(&self) -> &MemoryConfig {
        &self.config
    }

    /// INT-3: Get a reference to the batch manager.
    /// Satisfies: RT-5, B3, TN2 (memory efficiency through batch operations)
    pub fn batch_manager(&self) -> &Arc<RwLock<batch::BatchManager>> {
        &self.batch_manager
    }

    /// INT-3: Start a new batch context for batch operations.
    /// This provides a convenient way to start batch operations using the
    /// memory manager's shared batch tracker.
    /// Satisfies: RT-5, B3, TN2 (memory efficiency through batch operations)
    pub fn start_batch(&self) -> batch::BatchContext {
        self.batch_manager.read().start_batch()
    }

    /// INT-3: Get batch operation statistics.
    /// Returns current memory usage, peak usage, and batch count.
    pub fn batch_stats(&self) -> batch::BatchStats {
        self.batch_manager.read().get_stats()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_memory_manager_creation() {
        let manager = MemoryManager::with_defaults().unwrap();
        let stats = manager.memory_stats();
        assert_eq!(stats.cache_hits, 0);
        assert_eq!(stats.cache_misses, 0);
    }

    #[test]
    fn test_arena_allocation() {
        let manager = MemoryManager::with_defaults().unwrap();
        let handle = manager.arena_alloc(42u32);
        assert_eq!(*handle.get(), 42);

        manager.reset_arena();
        let stats = manager.memory_stats();
        assert_eq!(stats.arena_bytes, 0);
    }

    #[test]
    fn test_node_caching() {
        let manager = MemoryManager::with_defaults().unwrap();
        let node_id = 1;
        let node_data = vec![1, 2, 3, 4];

        // Cache miss
        assert!(manager.get_cached_node(node_id).is_none());

        // Cache the node
        manager.cache_node(node_id, node_data.clone());

        // Cache hit
        assert_eq!(manager.get_cached_node(node_id), Some(node_data));

        let stats = manager.memory_stats();
        assert_eq!(stats.cache_hits, 1);
        assert_eq!(stats.cache_misses, 1);
        assert!(stats.cache_hit_rate > 0.0);
    }

    #[test]
    fn test_string_interning() {
        let manager = MemoryManager::with_defaults().unwrap();

        let id1 = manager.intern_string("hello");
        let id2 = manager.intern_string("world");
        let id3 = manager.intern_string("hello"); // Duplicate

        assert_eq!(id1, id3); // Same string should get same ID
        assert_ne!(id1, id2); // Different strings get different IDs

        assert_eq!(manager.get_interned_string(id1), Some("hello".to_string()));
        assert_eq!(manager.get_interned_string(id2), Some("world".to_string()));
    }
}