166 lines
4.8 KiB
Python
166 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Real-time memory monitor for debugging expanded player memory leaks.
|
|
This module provides functions to track memory usage in real-time.
|
|
"""
|
|
|
|
import psutil
|
|
import os
|
|
import gc
|
|
import threading
|
|
import time
|
|
from loguru import logger
|
|
|
|
|
|
class MemoryMonitor:
|
|
"""Real-time memory monitoring for debugging memory leaks."""
|
|
|
|
def __init__(self):
|
|
self.process = psutil.Process(os.getpid())
|
|
self.baseline_memory = None
|
|
self.last_memory = None
|
|
self.monitoring = False
|
|
self.monitor_thread = None
|
|
|
|
def get_memory_usage(self):
|
|
"""Get current memory usage in MB."""
|
|
return self.process.memory_info().rss / 1024 / 1024
|
|
|
|
def get_memory_details(self):
|
|
"""Get detailed memory information."""
|
|
memory_info = self.process.memory_info()
|
|
memory_percent = self.process.memory_percent()
|
|
|
|
return {
|
|
"rss_mb": memory_info.rss / 1024 / 1024,
|
|
"vms_mb": memory_info.vms / 1024 / 1024,
|
|
"percent": memory_percent,
|
|
"num_threads": self.process.num_threads(),
|
|
}
|
|
|
|
def set_baseline(self, label="Baseline"):
|
|
"""Set the baseline memory usage."""
|
|
self.baseline_memory = self.get_memory_usage()
|
|
logger.info(f"🎯 {label} memory: {self.baseline_memory:.1f} MB")
|
|
return self.baseline_memory
|
|
|
|
def log_memory_change(self, label="Memory Check", force_gc=True):
|
|
"""Log current memory usage and change from baseline."""
|
|
if force_gc:
|
|
gc.collect()
|
|
|
|
current_memory = self.get_memory_usage()
|
|
details = self.get_memory_details()
|
|
|
|
if self.baseline_memory:
|
|
delta = current_memory - self.baseline_memory
|
|
logger.info(
|
|
f"📊 {label}: {current_memory:.1f} MB (Δ: {delta:+.1f} MB) | Threads: {
|
|
details['num_threads']
|
|
}"
|
|
)
|
|
else:
|
|
logger.info(
|
|
f"📊 {label}: {current_memory:.1f} MB | Threads: {
|
|
details['num_threads']
|
|
}"
|
|
)
|
|
|
|
self.last_memory = current_memory
|
|
return current_memory
|
|
|
|
def log_memory_spike(self, threshold_mb=10):
|
|
"""Log if there's a significant memory increase."""
|
|
if self.last_memory:
|
|
current = self.get_memory_usage()
|
|
increase = current - self.last_memory
|
|
if increase > threshold_mb:
|
|
logger.warning(
|
|
f"🚨 MEMORY SPIKE: +{increase:.1f} MB (from {self.last_memory:.1f} to {current:.1f} MB)"
|
|
)
|
|
return True
|
|
return False
|
|
|
|
def start_continuous_monitoring(self, interval_seconds=2):
|
|
"""Start continuous background monitoring."""
|
|
if self.monitoring:
|
|
return
|
|
|
|
self.monitoring = True
|
|
|
|
def monitor_loop():
|
|
while self.monitoring:
|
|
try:
|
|
self.log_memory_change("Continuous Monitor", force_gc=False)
|
|
time.sleep(interval_seconds)
|
|
except Exception as e:
|
|
logger.error(f"Memory monitor error: {e}")
|
|
break
|
|
|
|
self.monitor_thread = threading.Thread(target=monitor_loop, daemon=True)
|
|
self.monitor_thread.start()
|
|
logger.info(
|
|
f"🔍 Started continuous memory monitoring (every {interval_seconds}s)"
|
|
)
|
|
|
|
def stop_continuous_monitoring(self):
|
|
"""Stop continuous monitoring."""
|
|
if self.monitoring:
|
|
self.monitoring = False
|
|
logger.info("🛑 Stopped continuous memory monitoring")
|
|
|
|
|
|
# Global memory monitor instance
|
|
memory_monitor = MemoryMonitor()
|
|
|
|
# Convenience functions for easy use
|
|
|
|
|
|
def set_memory_baseline(label="Baseline"):
|
|
"""Set memory baseline."""
|
|
return memory_monitor.set_baseline(label)
|
|
|
|
|
|
def log_memory(label="Memory Check"):
|
|
"""Log current memory usage."""
|
|
return memory_monitor.log_memory_change(label)
|
|
|
|
|
|
def start_memory_monitoring():
|
|
"""Start continuous memory monitoring."""
|
|
memory_monitor.start_continuous_monitoring()
|
|
|
|
|
|
def stop_memory_monitoring():
|
|
"""Stop continuous memory monitoring."""
|
|
memory_monitor.stop_continuous_monitoring()
|
|
|
|
|
|
def check_memory_spike():
|
|
"""Check for memory spike."""
|
|
return memory_monitor.log_memory_spike()
|
|
|
|
|
|
# Test function
|
|
|
|
|
|
def test_memory_monitor():
|
|
"""Test the memory monitor."""
|
|
print("Testing Memory Monitor...")
|
|
monitor = MemoryMonitor()
|
|
|
|
monitor.set_baseline("Test Start")
|
|
|
|
# Simulate some memory usage
|
|
big_list = [i for i in range(100000)]
|
|
monitor.log_memory_change("After creating big list")
|
|
|
|
del big_list
|
|
monitor.log_memory_change("After deleting big list")
|
|
|
|
print("Memory monitor test complete!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_memory_monitor()
|