Python内存管理之垃圾回收机制深入详解

// C语言中的手动内存管理
#include <stdlib.h>

int main() {
    int *arr = (int*)malloc(10 * sizeof(int));  // 手动分配内存
    if (arr == NULL) {
        return -1;  // 内存分配失败处理
    }
    
    // 使用内存...
    for (int i = 0; i < 10; i++) {
        arr[i] = i;
    }
    
    free(arr);  // 手动释放内存
    return 0;
}

# Python中的自动内存管理
def process_data():
    # 自动分配内存
    data = [i for i in range(1000000)]
    result = [x * 2 for x in data]
    
    # 不需要手动释放内存
    return result

# 函数结束后，不再使用的内存会被自动回收

# memory_architecture.py
import sys
import os
from typing import Dict, List, Any
import ctypes

class MemoryArchitecture:
    """Python内存架构分析"""
    
    def __init__(self):
        self.memory_layers = {
            "application_layer": {
                "description": "Python对象层 - 开发者直接接触的层面",
                "components": ["对象创建", "引用管理", "生命周期"],
                "responsibility": "对象的创建和引用管理"
            },
            "interpreter_layer": {
                "description": "Python解释器层 - CPython实现",
                "components": ["PyObject", "类型系统", "引用计数"],
                "responsibility": "对象表示和基础内存管理"
            },
            "memory_allocator_layer": {
                "description": "内存分配器层 - Python内存分配策略",
                "components": ["对象分配器", "小块内存分配", "内存池"],
                "responsibility": "高效的内存分配和回收"
            },
            "system_layer": {
                "description": "操作系统层 - 底层内存管理",
                "components": ["malloc/free", "虚拟内存", "物理内存"],
                "responsibility": "物理内存的分配和管理"
            }
        }
    
    def analyze_memory_usage(self):
        """分析当前内存使用情况"""
        import gc
        
        print("=== Python内存架构分析 ===")
        
        # 各层内存使用分析
        for layer, info in self.memory_layers.items():
            print(f"\n{layer.upper()}层:")
            print(f"  描述: {info['description']}")
            print(f"  组件: {', '.join(info['components'])}")
        
        # 当前内存统计
        print(f"\n当前内存统计:")
        print(f"  进程内存使用: {self._get_process_memory():.2f} MB")
        print(f"  Python对象数量: {len(gc.get_objects())}")
        print(f"  垃圾回收器跟踪对象: {len(gc.get_tracked_objects())}")
    
    def _get_process_memory(self):
        """获取进程内存使用"""
        import psutil
        process = psutil.Process(os.getpid())
        return process.memory_info().rss / 1024 / 1024  # MB

# 使用示例
architecture = MemoryArchitecture()
architecture.analyze_memory_usage()

# object_representation.py
import sys
import struct
from dataclasses import dataclass
from typing import Any

class ObjectMemoryLayout:
    """Python对象内存布局分析"""
    
    @staticmethod
    def analyze_object(obj: Any) -> Dict[str, Any]:
        """分析对象的内存布局"""
        obj_type = type(obj)
        obj_id = id(obj)
        obj_size = sys.getsizeof(obj)
        
        # 获取对象的引用计数（仅CPython有效）
        ref_count = ObjectMemoryLayout._get_ref_count(obj)
        
        return {
            "type": obj_type.__name__,
            "id": obj_id,
            "size": obj_size,
            "ref_count": ref_count,
            "memory_address": hex(obj_id)
        }
    
    @staticmethod
    def _get_ref_count(obj: Any) -> int:
        """获取对象的引用计数"""
        # 注意：这仅适用于CPython实现
        return ctypes.c_long.from_address(id(obj)).value
    
    @staticmethod
    def compare_objects(*objects: Any) -> List[Dict[str, Any]]:
        """比较多个对象的内存特性"""
        results = []
        for obj in objects:
            analysis = ObjectMemoryLayout.analyze_object(obj)
            results.append(analysis)
        return results
    
    @staticmethod
    def demonstrate_memory_layout():
        """演示不同对象的内存布局"""
        print("=== Python对象内存布局演示 ===")
        
        # 创建不同类型的对象
        objects = [
            42,                    # 整数
            3.14159,              # 浮点数
            "Hello, World!",      # 字符串
            [1, 2, 3, 4, 5],      # 列表
            {"key": "value"},     # 字典
            (1, 2, 3),            # 元组
            {1, 2, 3}             # 集合
        ]
        
        results = ObjectMemoryLayout.compare_objects(*objects)
        
        for result in results:
            print(f"\n{result['type']}:")
            print(f"  内存地址: {result['memory_address']}")
            print(f"  大小: {result['size']} 字节")
            print(f"  引用计数: {result['ref_count']}")

# PyObject结构模拟（概念性）
class PyObject:
    """模拟CPython中PyObject的基本结构"""
    
    def __init__(self, obj_type, value):
        self.ob_refcnt = 1  # 引用计数
        self.ob_type = obj_type  # 类型指针
        self.ob_value = value  # 实际值
        
    def __repr__(self):
        return f"PyObject(type={self.ob_type}, refcnt={self.ob_refcnt}, value={self.ob_value})"

# 使用示例
if __name__ == "__main__":
    ObjectMemoryLayout.demonstrate_memory_layout()
    
    # 演示PyObject概念
    print("\n=== PyObject概念演示 ===")
    int_obj = PyObject("int", 42)
    str_obj = PyObject("str", "hello")
    
    print(f"整数对象: {int_obj}")
    print(f"字符串对象: {str_obj}")

# reference_counting.py
import sys
import ctypes
from typing import List, Dict, Any

class ReferenceCountingDemo:
    """引用计数机制演示"""
    
    def __init__(self):
        self.reference_events = []
    
    def track_references(self, obj: Any, description: str) -> None:
        """跟踪对象的引用变化"""
        current_count = self._get_ref_count(obj)
        event = {
            "description": description,
            "ref_count": current_count,
            "object_id": id(obj),
            "object_type": type(obj).__name__
        }
        self.reference_events.append(event)
        
        print(f"{description}: 引用计数 = {current_count}")
    
    def _get_ref_count(self, obj: Any) -> int:
        """安全地获取引用计数"""
        try:
            # 注意：这仅适用于CPython
            return ctypes.c_long.from_address(id(obj)).value
        except:
            # 对于其他Python实现，返回估计值
            return -1
    
    def demonstrate_basic_reference_counting(self):
        """演示基础引用计数"""
        print("=== 基础引用计数演示 ===")
        
        # 创建新对象
        my_list = [1, 2, 3]
        self.track_references(my_list, "创建列表")
        
        # 增加引用
        list_ref = my_list
        self.track_references(my_list, "创建另一个引用")
        
        # 在数据结构中引用
        container = [my_list]
        self.track_references(my_list, "添加到另一个列表")
        
        # 减少引用
        del list_ref
        self.track_references(my_list, "删除一个引用")
        
        # 从数据结构中移除
        container.clear()
        self.track_references(my_list, "从容器中移除")
        
        # 最后删除原始引用
        del my_list
    
    def demonstrate_function_references(self):
        """演示函数中的引用计数"""
        print("\n=== 函数中的引用计数 ===")
        
        def process_data(data):
            self.track_references(data, "函数参数接收")
            result = [x * 2 for x in data]
            self.track_references(data, "函数内部使用")
            return result
        
        data = [1, 2, 3, 4, 5]
        self.track_references(data, "函数调用前")
        
        result = process_data(data)
        self.track_references(data, "函数返回后")
        
        return data, result
    
    def analyze_reference_cycles(self):
        """分析循环引用"""
        print("\n=== 循环引用分析 ===")
        
        # 创建循环引用
        class Node:
            def __init__(self, value):
                self.value = value
                self.next = None
        
        # 创建两个节点并形成循环引用
        node1 = Node(1)
        node2 = Node(2)
        
        self.track_references(node1, "创建node1")
        self.track_references(node2, "创建node2")
        
        # 形成循环引用
        node1.next = node2
        node2.next = node1
        
        self.track_references(node1, "形成循环引用后 - node1")
        self.track_references(node2, "形成循环引用后 - node2")
        
        # 删除外部引用
        del node1
        del node2
        
        print("注意：虽然删除了外部引用，但由于循环引用，对象不会被立即释放")

# 引用计数数学原理
class ReferenceCountingTheory:
    """引用计数的数学原理"""
    
    @staticmethod
    def calculate_memory_lifetime(ref_count_history: List[int]) -> float:
        """
        计算对象的内存生命周期
        基于引用计数的变化模式
        """
        if not ref_count_history:
            return 0.0
        
        # 简单的生命周期估算：基于引用计数变化的频率和幅度
        changes = 0
        total_change_magnitude = 0
        
        for i in range(1, len(ref_count_history)):
            change = abs(ref_count_history[i] - ref_count_history[i-1])
            if change > 0:
                changes += 1
                total_change_magnitude += change
        
        if changes == 0:
            return float('inf')  # 引用计数不变，对象长期存在
        
        # 平均变化幅度越大，生命周期可能越短
        avg_change = total_change_magnitude / changes
        estimated_lifetime = 100.0 / avg_change  # 简化模型
        
        return estimated_lifetime
    
    @staticmethod
    def demonstrate_reference_counting_formula():
        """演示引用计数的数学公式"""
        print("\n=== 引用计数数学原理 ===")
        
        # 引用计数的基本公式
        formula = """
        引用计数变化公式：
        
        RC_{t+1} = RC_t + Δ_ref
        
        其中：
        - RC_t: 时间t时的引用计数
        - Δ_ref: 引用变化量
            Δ_ref = 新引用数量 - 消失引用数量
        
        对象释放条件：
        RC_t = 0 ⇒ 对象被立即释放
        """
        print(formula)
        
        # 示例计算
        ref_count_history = [1, 2, 3, 2, 1, 0]  # 典型的引用计数变化
        lifetime = ReferenceCountingTheory.calculate_memory_lifetime(ref_count_history)
        
        print(f"示例引用计数历史: {ref_count_history}")
        print(f"估算的对象生命周期: {lifetime:.2f}")

# 使用示例
if __name__ == "__main__":
    demo = ReferenceCountingDemo()
    demo.demonstrate_basic_reference_counting()
    demo.demonstrate_function_references()
    demo.analyze_reference_cycles()
    
    ReferenceCountingTheory.demonstrate_reference_counting_formula()

# reference_counting_analysis.py
from dataclasses import dataclass
from typing import List, Dict
import time

@dataclass
class ReferenceCountingMetrics:
    """引用计数性能指标"""
    objects_created: int
    objects_destroyed: int
    memory_usage_mb: float
    collection_time_ms: float

class ReferenceCountingAnalysis:
    """引用计数机制深度分析"""
    
    def __init__(self):
        self.metrics_history: List[ReferenceCountingMetrics] = []
    
    def analyze_advantages(self):
        """分析引用计数的优势"""
        advantages = {
            "immediate_reclamation": {
                "description": "立即回收 - 引用计数为0时立即释放内存",
                "benefit": "减少内存占用，提高内存利用率",
                "example": "局部变量在函数结束时立即释放"
            },
            "predictable_timing": {
                "description": "可预测的回收时机",
                "benefit": "避免Stop-the-World暂停",
                "example": "内存释放均匀分布在程序执行过程中"
            },
            "low_latency": {
                "description": "低延迟 - 不需要复杂的垃圾回收周期",
                "benefit": "适合实时性要求高的应用",
                "example": "GUI应用、游戏等"
            },
            "cache_friendly": {
                "description": "缓存友好 - 对象在不再使用时立即释放",
                "benefit": "提高缓存命中率",
                "example": "临时对象不会长时间占用缓存"
            }
        }
        
        print("=== 引用计数优势分析 ===")
        for adv_key, adv_info in advantages.items():
            print(f"\n{adv_info['description']}:")
            print(f"  好处: {adv_info['benefit']}")
            print(f"  示例: {adv_info['example']}")
    
    def analyze_limitations(self):
        """分析引用计数的局限性"""
        limitations = {
            "circular_references": {
                "description": "循环引用问题 - 无法回收形成循环引用的对象",
                "impact": "内存泄漏",
                "example": "两个对象相互引用，但没有外部引用"
            },
            "performance_overhead": {
                "description": "性能开销 - 每次引用操作都需要更新计数",
                "impact": "降低程序执行速度",
                "example": "函数调用、赋值操作都有额外开销"
            },
            "memory_fragmentation": {
                "description": "内存碎片 - 频繁分配释放导致内存碎片",
                "impact": "降低内存使用效率",
                "example": "大量小对象的创建和销毁"
            },
            "atomic_operations": {
                "description": "原子操作开销 - 多线程环境需要原子操作",
                "impact": "并发性能下降",
                "example": "多线程同时修改引用计数"
            }
        }
        
        print("\n=== 引用计数局限性分析 ===")
        for lim_key, lim_info in limitations.items():
            print(f"\n{lim_info['description']}:")
            print(f"  影响: {lim_info['impact']}")
            print(f"  示例: {lim_info['example']}")
    
    def performance_benchmark(self):
        """性能基准测试"""
        print("\n=== 引用计数性能测试 ===")
        
        import gc
        gc.disable()  # 暂时禁用其他GC机制
        
        start_time = time.time()
        start_memory = self._get_memory_usage()
        
        # 创建大量临时对象
        objects_created = 0
        for i in range(100000):
            # 创建临时对象，依赖引用计数进行回收
            temp_list = [i for i in range(100)]
            temp_dict = {str(i): i for i in range(50)}
            objects_created += 2
            
            # 立即失去引用，应该被立即回收
            del temp_list
            del temp_dict
        
        end_time = time.time()
        end_memory = self._get_memory_usage()
        
        gc.enable()
        
        execution_time = (end_time - start_time) * 1000  # 毫秒
        memory_used = end_memory - start_memory
        
        metrics = ReferenceCountingMetrics(
            objects_created=objects_created,
            objects_destroyed=objects_created,  # 理论上应该全部被销毁
            memory_usage_mb=memory_used,
            collection_time_ms=execution_time
        )
        
        self.metrics_history.append(metrics)
        
        print(f"创建对象数量: {metrics.objects_created}")
        print(f"执行时间: {metrics.collection_time_ms:.2f} ms")
        print(f"内存使用变化: {metrics.memory_usage_mb:.2f} MB")
        print(f"平均每个对象处理时间: {metrics.collection_time_ms/metrics.objects_created:.4f} ms")
    
    def _get_memory_usage(self):
        """获取内存使用量"""
        import psutil
        import os
        process = psutil.Process(os.getpid())
        return process.memory_info().rss / 1024 / 1024  # MB

# 循环引用问题深度分析
class CircularReferenceAnalyzer:
    """循环引用问题分析器"""
    
    def demonstrate_circular_reference_problem(self):
        """演示循环引用问题"""
        print("\n=== 循环引用问题演示 ===")
        
        class Person:
            def __init__(self, name):
                self.name = name
                self.friends = []
            
            def add_friend(self, friend):
                self.friends.append(friend)
                friend.friends.append(self)  # 相互引用
        
        # 创建循环引用
        alice = Person("Alice")
        bob = Person("Bob")
        
        print(f"创建Alice: {id(alice)}")
        print(f"创建Bob: {id(bob)}")
        
        # 形成循环引用
        alice.add_friend(bob)
        
        print("形成循环引用: Alice ↔ Bob")
        
        # 删除外部引用
        del alice
        del bob
        
        print("删除外部引用后，由于循环引用，对象无法被引用计数机制回收")
    
    def analyze_circular_reference_patterns(self):
        """分析常见的循环引用模式"""
        patterns = {
            "bidirectional_relationship": {
                "description": "双向关系 - 两个对象相互引用",
                "example": "父子节点相互引用",
                "solution": "使用弱引用(weakref)"
            },
            "self_reference": {
                "description": "自引用 - 对象引用自身",
                "example": "对象在属性中引用自己",
                "solution": "避免自引用或使用弱引用"
            },
            "container_reference": {
                "description": "容器引用 - 对象被容器引用同时又引用容器",
                "example": "对象在列表中，同时又持有该列表的引用",
                "solution": "谨慎设计数据结构"
            },
            "complex_cycle": {
                "description": "复杂循环 - 多个对象形成引用环",
                "example": "A→B→C→A 的引用链",
                "solution": "需要分代垃圾回收来处理"
            }
        }
        
        print("\n=== 循环引用模式分析 ===")
        for pattern_key, pattern_info in patterns.items():
            print(f"\n{pattern_info['description']}:")
            print(f"  示例: {pattern_info['example']}")
            print(f"  解决方案: {pattern_info['solution']}")

# 使用示例
if __name__ == "__main__":
    analysis = ReferenceCountingAnalysis()
    analysis.analyze_advantages()
    analysis.analyze_limitations()
    analysis.performance_benchmark()
    
    circular_analyzer = CircularReferenceAnalyzer()
    circular_analyzer.demonstrate_circular_reference_problem()
    circular_analyzer.analyze_circular_reference_patterns()

# generational_gc.py
import gc
import time
from dataclasses import dataclass
from typing import List, Dict, Any
import weakref

@dataclass
class GenerationStats:
    """分代统计信息"""
    generation: int
    object_count: int
    collection_count: int
    last_collection_time: float

class GenerationalGCAnalyzer:
    """分代垃圾回收分析器"""
    
    def __init__(self):
        self.gc_stats = {}
        self.setup_gc_monitoring()
    
    def setup_gc_monitoring(self):
        """设置GC监控"""
        # 启用调试功能
        gc.set_debug(gc.DEBUG_STATS)
    
    def analyze_generations(self):
        """分析分代垃圾回收机制"""
        print("=== 分代垃圾回收分析 ===")
        
        # 获取GC统计信息
        stats = gc.get_stats()
        
        print("\n分代假设原理:")
        print("1. 年轻代假设: 大多数对象很快变得不可达")
        print("2. 老年代假设: 存活时间越长的对象，越可能继续存活")
        print("3. 代间提升: 存活足够久的对象会被提升到老一代")
        
        print(f"\n当前GC统计:")
        for gen_stats in stats:
            print(f"  第{gen_stats['generation']}代:")
            print(f"    回收次数: {gen_stats['collected']}")
            print(f"    存活对象: {gen_stats['alive']}")
            print(f"    不可回收对象: {gen_stats['uncollectable']}")
    
    def demonstrate_generational_behavior(self):
        """演示分代行为"""
        print("\n=== 分代行为演示 ===")
        
        # 创建不同生命周期的对象
        short_lived_objects = self._create_short_lived_objects()
        long_lived_objects = self._create_long_lived_objects()
        
        print("创建短期存活对象和长期存活对象...")
        
        # 强制进行垃圾回收并观察行为
        for generation in range(3):
            print(f"\n--- 强制第{generation}代GC ---")
            collected = gc.collect(generation)
            print(f"回收对象数量: {collected}")
            
            # 获取当前代统计
            current_stats = gc.get_count()
            print(f"当前代计数: {current_stats}")
    
    def _create_short_lived_objects(self) -> List[Any]:
        """创建短期存活对象"""
        objects = []
        for i in range(1000):
            # 创建对象但立即失去引用（模拟短期存活）
            temp = [j for j in range(10)]
            objects.append(temp)
        return objects[:100]  # 只保留少量引用
    
    def _create_long_lived_objects(self) -> List[Any]:
        """创建长期存活对象"""
        long_lived = []
        # 创建一些会长期存活的对象
        for i in range(100):
            obj = {"id": i, "data": "长期存活数据"}
            long_lived.append(obj)
        return long_lived
    
    def analyze_gc_thresholds(self):
        """分析GC触发阈值"""
        print("\n=== GC触发阈值分析 ===")
        
        # 获取当前GC阈值
        thresholds = gc.get_threshold()
        
        print("各代GC触发阈值:")
        for i, threshold in enumerate(thresholds):
            print(f"  第{i}代: {threshold}")
        
        print("\n阈值含义:")
        print("  第0代: 当分配的对象数量达到此阈值时，触发第0代GC")
        print("  第1代: 当第0代GC执行次数达到此阈值时，触发第1代GC") 
        print("  第2代: 当第1代GC执行次数达到此阈值时，触发第2代GC")
        
        # 当前对象计数
        current_count = gc.get_count()
        print(f"\n当前对象计数: {current_count}")
        print(f"距离下一次GC: {thresholds[0] - current_count[0]} 个对象")

class GCPerformanceAnalyzer:
    """GC性能分析器"""
    
    def __init__(self):
        self.performance_data = []
    
    def measure_gc_performance(self, object_count: int = 10000):
        """测量GC性能"""
        print(f"\n=== GC性能测试 ({object_count}个对象) ===")
        
        # 禁用GC进行基准测试
        gc.disable()
        base_time = self._create_and_destroy_objects(object_count)
        
        # 启用GC进行测试
        gc.enable()
        gc_time = self._create_and_destroy_objects(object_count)
        
        print(f"无GC时间: {base_time:.4f} 秒")
        print(f"有GC时间: {gc_time:.4f} 秒")
        print(f"GC开销: {gc_time - base_time:.4f} 秒")
        print(f"相对开销: {(gc_time - base_time) / base_time * 100:.2f}%")
    
    def _create_and_destroy_objects(self, count: int) -> float:
        """创建和销毁对象并测量时间"""
        import time
        
        start_time = time.time()
        
        objects = []
        for i in range(count):
            # 创建复杂对象
            obj = {
                'id': i,
                'data': [j for j in range(10)],
                'nested': {'key': 'value' * (i % 10)}
            }
            objects.append(obj)
        
        # 模拟对象使用
        for obj in objects:
            _ = obj['id'] + len(obj['data'])
        
        # 销毁对象（通过失去引用）
        del objects
        
        end_time = time.time()
        return end_time - start_time
    
    def analyze_memory_pressure_impact(self):
        """分析内存压力对GC的影响"""
        print("\n=== 内存压力对GC的影响 ===")
        
        memory_pressures = [1000, 5000, 10000, 50000]
        
        for pressure in memory_pressures:
            print(f"\n内存压力: {pressure} 个对象")
            
            # 测量不同内存压力下的GC性能
            start_time = time.time()
            
            # 创建内存压力
            large_objects = []
            for i in range(pressure):
                large_list = [j for j in range(100)]
                large_objects.append(large_list)
            
            # 执行GC并测量时间
            gc_start = time.time()
            collected = gc.collect()
            gc_time = time.time() - gc_start
            
            # 清理
            del large_objects
            
            total_time = time.time() - start_time
            
            print(f"  GC回收对象: {collected}")
            print(f"  GC执行时间: {gc_time:.4f} 秒")
            print(f"  总执行时间: {total_time:.4f} 秒")

# 使用示例
if __name__ == "__main__":
    generational_analyzer = GenerationalGCAnalyzer()
    generational_analyzer.analyze_generations()
    generational_analyzer.demonstrate_generational_behavior()
    generational_analyzer.analyze_gc_thresholds()
    
    performance_analyzer = GCPerformanceAnalyzer()
    performance_analyzer.measure_gc_performance(5000)
    performance_analyzer.analyze_memory_pressure_impact()

# mark_sweep_algorithm.py
from typing import Set, List, Dict, Any
from enum import Enum
import time

class ObjectColor(Enum):
    """对象标记颜色（三色标记法）"""
    WHITE = 0  # 未访问，可能垃圾
    GRAY = 1   # 正在处理，已访问但引用未处理完
    BLACK = 2  # 已处理，存活对象

class GCNode:
    """垃圾回收节点（模拟对象）"""
    
    def __init__(self, obj_id: int, size: int = 1):
        self.obj_id = obj_id
        self.size = size
        self.references: List['GCNode'] = []
        self.color = ObjectColor.WHITE
        self.generation = 0
    
    def add_reference(self, node: 'GCNode'):
        """添加引用"""
        self.references.append(node)
    
    def __repr__(self):
        return f"GCNode({self.obj_id}, color={self.color.name}, gen={self.generation})"

class MarkSweepCollector:
    """标记-清除垃圾回收器模拟"""
    
    def __init__(self):
        self.roots: Set[GCNode] = set()  # 根对象集合
        self.all_objects: Dict[int, GCNode] = {}  # 所有对象
        self.object_counter = 0
        
        # 统计信息
        self.stats = {
            'collections': 0,
            'objects_collected': 0,
            'memory_reclaimed': 0,
            'collection_times': []
        }
    
    def allocate_object(self, size: int = 1) -> GCNode:
        """分配新对象"""
        self.object_counter += 1
        obj = GCNode(self.object_counter, size)
        self.all_objects[obj.obj_id] = obj
        return obj
    
    def add_root(self, node: GCNode):
        """添加根对象"""
        self.roots.add(node)
    
    def mark_phase(self):
        """标记阶段 - 标记所有从根对象可达的对象"""
        # 重置所有对象为白色
        for obj in self.all_objects.values():
            obj.color = ObjectColor.WHITE
        
        # 从根对象开始标记
        gray_set: Set[GCNode] = set()
        
        # 根对象标记为灰色
        for root in self.roots:
            root.color = ObjectColor.GRAY
            gray_set.add(root)
        
        # 处理灰色对象
        while gray_set:
            current = gray_set.pop()
            
            # 标记当前对象为黑色
            current.color = ObjectColor.BLACK
            
            # 处理所有引用
            for referenced in current.references:
                if referenced.color == ObjectColor.WHITE:
                    referenced.color = ObjectColor.GRAY
                    gray_set.add(referenced)
    
    def sweep_phase(self) -> List[GCNode]:
        """清除阶段 - 回收所有白色对象"""
        collected_objects = []
        remaining_objects = {}
        
        for obj_id, obj in self.all_objects.items():
            if obj.color == ObjectColor.WHITE:
                # 白色对象是垃圾，进行回收
                collected_objects.append(obj)
                self.stats['objects_collected'] += 1
                self.stats['memory_reclaimed'] += obj.size
            else:
                # 黑色对象存活，保留并提升代际
                obj.generation = min(obj.generation + 1, 2)
                remaining_objects[obj_id] = obj
        
        self.all_objects = remaining_objects
        return collected_objects
    
    def collect_garbage(self) -> List[GCNode]:
        """执行垃圾回收"""
        start_time = time.time()
        
        print("开始垃圾回收...")
        print(f"回收前对象数量: {len(self.all_objects)}")
        
        # 标记阶段
        self.mark_phase()
        
        # 清除阶段
        collected = self.sweep_phase()
        
        # 更新统计
        self.stats['collections'] += 1
        collection_time = time.time() - start_time
        self.stats['collection_times'].append(collection_time)
        
        print(f"回收后对象数量: {len(self.all_objects)}")
        print(f"回收对象数量: {len(collected)}")
        print(f"回收时间: {collection_time:.4f} 秒")
        
        return collected
    
    def demonstrate_algorithm(self):
        """演示标记-清除算法"""
        print("=== 标记-清除算法演示 ===")
        
        # 创建对象图
        root1 = self.allocate_object()
        root2 = self.allocate_object()
        
        obj3 = self.allocate_object()
        obj4 = self.allocate_object()
        obj5 = self.allocate_object()  # 这个对象将形成循环引用但不可达
        
        # 建立引用关系
        root1.add_reference(obj3)
        root2.add_reference(obj4)
        obj3.add_reference(obj4)
        
        # 创建循环引用但不可达的对象
        obj5.add_reference(obj5)  # 自引用
        
        # 设置根对象
        self.add_root(root1)
        self.add_root(root2)
        
        print("\n对象图结构:")
        print(f"根对象: {root1.obj_id}, {root2.obj_id}")
        print(f"可达对象: {obj3.obj_id} ← root1, {obj4.obj_id} ← root2 & obj3")
        print(f"不可达对象: {obj5.obj_id} (自引用)")
        
        # 执行垃圾回收
        collected = self.collect_garbage()
        
        print(f"\n回收的对象: {[obj.obj_id for obj in collected]}")
        
        # 显示存活对象
        print(f"存活对象: {list(self.all_objects.keys())}")

class GenerationalCollector(MarkSweepCollector):
    """分代垃圾回收器"""
    
    def __init__(self):
        super().__init__()
        self.generations = [set(), set(), set()]  # 三代对象集合
        self.collection_thresholds = [700, 10, 10]  # 各代回收阈值
        self.allocation_count = 0
    
    def allocate_object(self, size: int = 1) -> GCNode:
        """分配对象到年轻代"""
        obj = super().allocate_object(size)
        self.generations[0].add(obj)
        self.allocation_count += 1
        
        # 检查是否需要年轻代GC
        if self.allocation_count >= self.collection_thresholds[0]:
            self.collect_generation(0)
        
        return obj
    
    def collect_generation(self, generation: int):
        """回收指定代的对象"""
        print(f"\n--- 执行第{generation}代GC ---")
        
        if generation == 0:
            # 年轻代GC：只处理第0代
            self._collect_young()
        else:
            # 老年代GC：处理指定代及所有更年轻的代
            self._collect_old(generation)
    
    def _collect_young(self):
        """年轻代回收"""
        # 临时将年轻代对象作为根
        old_roots = self.roots.copy()
        self.roots.update(self.generations[1])  # 老年代对象作为根
        self.roots.update(self.generations[2])  # 老老年代对象作为根
        
        # 执行标记-清除
        collected = super().collect_garbage()
        
        # 提升存活对象到下一代
        self._promote_survivors()
        
        # 恢复根集合
        self.roots = old_roots
        
        # 重置分配计数
        self.allocation_count = 0
    
    def _promote_survivors(self):
        """提升存活对象到下一代"""
        promoted = set()
        for obj in self.generations[0]:
            if obj in self.all_objects.values():  # 对象仍然存活
                new_gen = min(obj.generation + 1, 2)
                self.generations[new_gen].add(obj)
                promoted.add(obj)
        
        # 从年轻代移除已提升的对象
        self.generations[0] = self.generations[0] - promoted
    
    def _collect_old(self, generation: int):
        """老年代回收"""
        # 收集指定代及所有更年轻的代
        for gen in range(generation + 1):
            # 将这些代的对象临时作为根
            for g in range(gen + 1, 3):
                self.roots.update(self.generations[g])
        
        # 执行标记-清除
        collected = super().collect_garbage()
        
        # 重新组织分代
        self._reorganize_generations()

# 使用示例
if __name__ == "__main__":
    print("=== 标记-清除算法演示 ===")
    basic_collector = MarkSweepCollector()
    basic_collector.demonstrate_algorithm()
    
    print("\n" + "="*50 + "\n")
    
    print("=== 分代垃圾回收演示 ===")
    gen_collector = GenerationalCollector()
    
    # 模拟对象分配模式
    for i in range(1000):
        obj = gen_collector.allocate_object()
        if i % 100 == 0:
            # 偶尔创建长期存活的对象
            gen_collector.add_root(obj)

# weak_references.py
import weakref
import gc
from typing import List, Dict, Any
from dataclasses import dataclass

class WeakReferenceDemo:
    """弱引用演示"""
    
    def demonstrate_basic_weakref(self):
        """演示基础弱引用"""
        print("=== 基础弱引用演示 ===")
        
        class Data:
            def __init__(self, value):
                self.value = value
                print(f"创建Data对象: {self.value}")
            
            def __del__(self):
                print(f"销毁Data对象: {self.value}")
        
        # 创建普通引用
        data = Data("important_data")
        strong_ref = data
        
        # 创建弱引用
        weak_ref = weakref.ref(data)
        
        print(f"原始对象: {data}")
        print(f"强引用: {strong_ref}")
        print(f"弱引用: {weak_ref}")
        print(f"通过弱引用访问: {weak_ref()}")
        
        # 删除强引用
        del data
        del strong_ref
        
        # 强制垃圾回收
        gc.collect()
        
        print(f"回收后弱引用: {weak_ref()}")
    
    def demonstrate_weak_value_dictionary(self):
        """演示弱值字典"""
        print("\n=== 弱值字典演示 ===")
        
        # 创建弱值字典
        cache = weakref.WeakValueDictionary()
        
        class ExpensiveObject:
            def __init__(self, key):
                self.key = key
                self.data = "昂贵的计算结果"
                print(f"创建昂贵对象: {self.key}")
            
            def __del__(self):
                print(f"销毁昂贵对象: {self.key}")
        
        # 向缓存添加对象
        obj1 = ExpensiveObject("key1")
        obj2 = ExpensiveObject("key2")
        
        cache["key1"] = obj1
        cache["key2"] = obj2
        
        print(f"缓存内容: {list(cache.keys())}")
        print(f"获取key1: {cache.get('key1')}")
        
        # 删除对象的强引用
        del obj1
        gc.collect()
        
        print(f"回收后缓存内容: {list(cache.keys())}")
        print(f"获取key1: {cache.get('key1')}")
    
    def demonstrate_weak_set(self):
        """演示弱引用集合"""
        print("\n=== 弱引用集合演示 ===")
        
        observer_set = weakref.WeakSet()
        
        class Observer:
            def __init__(self, name):
                self.name = name
            
            def update(self):
                print(f"Observer {self.name} 收到更新")
            
            def __repr__(self):
                return f"Observer({self.name})"
        
        # 创建观察者
        obs1 = Observer("A")
        obs2 = Observer("B")
        obs3 = Observer("C")
        
        # 添加到弱引用集合
        observer_set.add(obs1)
        observer_set.add(obs2)
        observer_set.add(obs3)
        
        print(f"观察者集合: {list(observer_set)}")
        
        # 删除一些观察者
        del obs2
        gc.collect()
        
        print(f"回收后观察者集合: {list(observer_set)}")
    
    def solve_circular_reference(self):
        """使用弱引用解决循环引用问题"""
        print("\n=== 使用弱引用解决循环引用 ===")
        
        class TreeNode:
            def __init__(self, value):
                self.value = value
                self._parent = None
                self.children = []
                print(f"创建节点: {self.value}")
            
            @property
            def parent(self):
                return self._parent() if self._parent else None
            
            @parent.setter
            def parent(self, node):
                if node is None:
                    self._parent = None
                else:
                    self._parent = weakref.ref(node)
            
            def add_child(self, child):
                self.children.append(child)
                child.parent = self
            
            def __del__(self):
                print(f"销毁节点: {self.value}")
        
        # 创建树结构（可能产生循环引用）
        root = TreeNode("root")
        child1 = TreeNode("child1")
        child2 = TreeNode("child2")
        
        root.add_child(child1)
        root.add_child(child2)
        
        print(f"根节点的子节点: {[child.value for child in root.children]}")
        print(f"子节点1的父节点: {child1.parent.value if child1.parent else None}")
        
        # 删除根节点引用
        del root
        gc.collect()
        
        print("注意：由于使用弱引用，循环引用被正确打破")

class CacheManager:
    """基于弱引用的缓存管理器"""
    
    def __init__(self, max_size: int = 100):
        self.cache = weakref.WeakValueDictionary()
        self.max_size = max_size
        self.access_count = 0
        self.hit_count = 0
    
    def get(self, key: Any) -> Any:
        """从缓存获取值"""
        self.access_count += 1
        
        value = self.cache.get(key)
        if value is not None:
            self.hit_count += 1
        
        return value
    
    def set(self, key: Any, value: Any):
        """设置缓存值"""
        if len(self.cache) >= self.max_size:
            self._evict_oldest()
        
        self.cache[key] = value
    
    def _evict_oldest(self):
        """驱逐最老的缓存项"""
        # WeakValueDictionary会自动清理，这里只是演示
        print("缓存达到最大大小，等待自动清理...")
    
    def get_stats(self) -> Dict[str, Any]:
        """获取缓存统计"""
        hit_rate = self.hit_count / self.access_count if self.access_count > 0 else 0
        
        return {
            'cache_size': len(self.cache),
            'access_count': self.access_count,
            'hit_count': self.hit_count,
            'hit_rate': hit_rate,
            'max_size': self.max_size
        }

# 使用示例
if __name__ == "__main__":
    demo = WeakReferenceDemo()
    demo.demonstrate_basic_weakref()
    demo.demonstrate_weak_value_dictionary()
    demo.demonstrate_weak_set()
    demo.solve_circular_reference()
    
    print("\n=== 缓存管理器演示 ===")
    cache = CacheManager(max_size=5)
    
    # 模拟缓存使用
    for i in range(10):
        key = f"key_{i}"
        value = f"value_{i}"
        cache.set(key, value)
        
        # 偶尔访问之前的键
        if i % 3 == 0 and i > 0:
            cached_value = cache.get(f"key_{i-1}")
            print(f"访问 key_{i-1}: {cached_value}")
    
    stats = cache.get_stats()
    print(f"\n缓存统计: {stats}")

# complete_gc_system.py
import gc
import time
from typing import Dict, List, Any
from dataclasses import dataclass
from enum import Enum
import threading

class GCStrategy(Enum):
    """垃圾回收策略"""
    REFERENCE_COUNTING = "reference_counting"
    GENERATIONAL_GC = "generational_gc"
    MANUAL_GC = "manual_gc"
    DISABLED_GC = "disabled_gc"

@dataclass
class GCProfile:
    """GC配置档案"""
    name: str
    strategy: GCStrategy
    thresholds: tuple
    enabled: bool
    debug: bool

class CompleteGCSystem:
    """完整的垃圾回收系统"""
    
    def __init__(self):
        self.profiles: Dict[str, GCProfile] = {}
        self.current_profile: str = "balanced"
        self.performance_stats: Dict[str, List[float]] = {
            'collection_times': [],
            'memory_usage': [],
            'object_counts': []
        }
        
        self._setup_default_profiles()
    
    def _setup_default_profiles(self):
        """设置默认配置档案"""
        self.profiles = {
            "performance": GCProfile(
                name="performance",
                strategy=GCStrategy.DISABLED_GC,
                thresholds=(0, 0, 0),
                enabled=False,
                debug=False
            ),
            "balanced": GCProfile(
                name="balanced", 
                strategy=GCStrategy.GENERATIONAL_GC,
                thresholds=(700, 10, 10),
                enabled=True,
                debug=False
            ),
            "aggressive": GCProfile(
                name="aggressive",
                strategy=GCStrategy.GENERATIONAL_GC, 
                thresholds=(300, 5, 5),
                enabled=True,
                debug=False
            ),
            "debug": GCProfile(
                name="debug",
                strategy=GCStrategy.GENERATIONAL_GC,
                thresholds=(100, 2, 2),
                enabled=True,
                debug=True
            )
        }
    
    def set_profile(self, profile_name: str):
        """设置GC配置"""
        if profile_name not in self.profiles:
            raise ValueError(f"未知的GC配置: {profile_name}")
        
        profile = self.profiles[profile_name]
        self.current_profile = profile_name
        
        # 应用配置
        gc.set_threshold(*profile.thresholds)
        gc.enable() if profile.enabled else gc.disable()
        gc.set_debug(gc.DEBUG_STATS if profile.debug else 0)
        
        print(f"切换到GC配置: {profile_name}")
        print(f"  策略: {profile.strategy.value}")
        print(f"  阈值: {profile.thresholds}")
        print(f"  启用: {profile.enabled}")
        print(f"  调试: {profile.debug}")
    
    def monitor_gc_performance(self, duration: int = 30):
        """监控GC性能"""
        print(f"开始GC性能监控 ({duration}秒)...")
        
        start_time = time.time()
        monitoring_thread = threading.Thread(
            target=self._monitoring_worker,
            args=(duration,)
        )
        monitoring_thread.daemon = True
        monitoring_thread.start()
        
        # 模拟工作负载
        self._generate_workload(duration)
        
        monitoring_thread.join()
        self._generate_performance_report()
    
    def _monitoring_worker(self, duration: int):
        """监控工作线程"""
        end_time = time.time() + duration
        
        while time.time() < end_time:
            # 收集性能数据
            current_time = time.time()
            
            # 内存使用
            memory_usage = self._get_memory_usage()
            
            # 对象计数
            object_count = len(gc.get_objects())
            
            # 记录数据
            self.performance_stats['memory_usage'].append(memory_usage)
            self.performance_stats['object_counts'].append(object_count)
            
            time.sleep(1)  # 每秒采样一次
    
    def _generate_workload(self, duration: int):
        """生成工作负载"""
        print("生成模拟工作负载...")
        
        end_time = time.time() + duration
        objects_created = 0
        
        while time.time() < end_time:
            # 创建各种对象模拟真实工作负载
            self._create_temporary_objects()
            self._create_long_lived_objects()
            self._create_circular_references()
            
            objects_created += 100
            time.sleep(0.1)  # 控制负载强度
        
        print(f"工作负载完成，创建了约 {objects_created} 个对象")
    
    def _create_temporary_objects(self):
        """创建临时对象"""
        # 短期存活的对象
        for i in range(50):
            temp_list = [j for j in range(100)]
            temp_dict = {f"key_{j}": j for j in range(50)}
            # 对象会很快超出作用域并被回收
    
    def _create_long_lived_objects(self):
        """创建长期存活对象"""
        if not hasattr(self, 'long_lived_objects'):
            self.long_lived_objects = []
        
        # 一些长期存活的对象
        for i in range(10):
            persistent_obj = {"id": i, "data": "长期数据" * 100}
            self.long_lived_objects.append(persistent_obj)
    
    def _create_circular_references(self):
        """创建循环引用"""
        # 偶尔创建一些循环引用
        class Node:
            def __init__(self, id):
                self.id = id
                self.partner = None
        
        node1 = Node(1)
        node2 = Node(2)
        
        # 形成循环引用
        node1.partner = node2
        node2.partner = node1
        
        # 不保存引用，让GC来处理
    
    def _get_memory_usage(self) -> float:
        """获取内存使用量"""
        import psutil
        import os
        process = psutil.Process(os.getpid())
        return process.memory_info().rss / 1024 / 1024  # MB
    
    def _generate_performance_report(self):
        """生成性能报告"""
        print("\n" + "="*50)
        print("GC性能报告")
        print("="*50)
        
        if not self.performance_stats['memory_usage']:
            print("没有收集到性能数据")
            return
        
        # 内存使用分析
        memory_data = self.performance_stats['memory_usage']
        avg_memory = sum(memory_data) / len(memory_data)
        max_memory = max(memory_data)
        min_memory = min(memory_data)
        
        print(f"内存使用分析:")
        print(f"  平均: {avg_memory:.2f} MB")
        print(f"  最大: {max_memory:.2f} MB") 
        print(f"  最小: {min_memory:.2f} MB")
        print(f"  波动: {max_memory - min_memory:.2f} MB")
        
        # 对象数量分析
        object_data = self.performance_stats['object_counts']
        avg_objects = sum(object_data) / len(object_data)
        
        print(f"\n对象数量分析:")
        print(f"  平均对象数: {avg_objects:.0f}")
        
        # GC统计
        gc_stats = gc.get_stats()
        print(f"\nGC统计:")
        for gen_stats in gc_stats:
            print(f"  第{gen_stats['generation']}代:")
            print(f"    回收次数: {gen_stats['collected']}")
            print(f"    存活对象: {gen_stats['alive']}")

class MemoryOptimizer:
    """内存优化工具"""
    
    @staticmethod
    def optimize_memory_usage():
        """优化内存使用"""
        print("=== 内存优化建议 ===")
        
        suggestions = [
            "1. 使用生成器代替列表处理大数据集",
            "2. 及时删除不再需要的大对象",
            "3. 使用__slots__减少对象内存开销", 
            "4. 避免不必要的对象创建",
            "5. 使用适当的数据结构",
            "6. 定期调用gc.collect()在关键点",
            "7. 使用弱引用打破循环引用",
            "8. 监控内存使用并设置警报"
        ]
        
        for suggestion in suggestions:
            print(suggestion)
    
    @staticmethod
    def demonstrate_memory_optimization():
        """演示内存优化技术"""
        print("\n=== 内存优化演示 ===")
        
        # 演示生成器的内存优势
        print("1. 生成器 vs 列表:")
        
        # 列表方法（占用大量内存）
        def get_numbers_list(n):
            return [i for i in range(n)]
        
        # 生成器方法（内存高效）
        def get_numbers_generator(n):
            for i in range(n):
                yield i
        
        # 测试内存使用
        import sys
        
        list_size = sys.getsizeof(get_numbers_list(1000000))
        gen_size = sys.getsizeof(get_numbers_generator(1000000))
        
        print(f"  列表大小: {list_size / 1024 / 1024:.2f} MB")
        print(f"  生成器大小: {gen_size} 字节")
        print(f"  内存节省: {(list_size - gen_size) / list_size * 100:.1f}%")
        
        # 演示__slots__的内存优势
        print("\n2. __slots__ 内存优化:")
        
        class RegularClass:
            def __init__(self, x, y):
                self.x = x
                self.y = y
        
        class SlotsClass:
            __slots__ = ['x', 'y']
            def __init__(self, x, y):
                self.x = x
                self.y = y
        
        regular_obj = RegularClass(1, 2)
        slots_obj = SlotsClass(1, 2)
        
        regular_size = sys.getsizeof(regular_obj) + sys.getsizeof(regular_obj.__dict__)
        slots_size = sys.getsizeof(slots_obj)
        
        print(f"  普通类大小: {regular_size} 字节")
        print(f"  slots类大小: {slots_size} 字节") 
        print(f"  内存节省: {(regular_size - slots_size) / regular_size * 100:.1f}%")

# 使用示例
if __name__ == "__main__":
    # 完整GC系统演示
    gc_system = CompleteGCSystem()
    
    # 测试不同配置
    for profile_name in ["performance", "balanced", "aggressive"]:
        print(f"\n{'='*60}")
        print(f"测试配置: {profile_name}")
        print('='*60)
        
        gc_system.set_profile(profile_name)
        gc_system.monitor_gc_performance(duration=10)
    
    # 内存优化演示
    MemoryOptimizer.optimize_memory_usage()
    MemoryOptimizer.demonstrate_memory_optimization()