Memory Management & Optimization

import sys

# Check reference count
x = []
print(sys.getrefcount(x))  # 2 (x itself + temp in function)

# Multiple references
y = x
z = [x]
print(sys.getrefcount(x))  # 4

# References decrease when deleted or go out of scope
del y
print(sys.getrefcount(x))  # 3

# Circular references are NOT handled by refcounting alone
a = []
b = [a]
a.append(b)  # a and b reference each other
del a, b  # Refcounts are still > 0, memory leak!

import gc

# Get current GC thresholds
print(gc.get_threshold())  # (700, 10, 10)
# Generation 0 collects after 700 allocations
# Generation 1 collects after 10 gen-0 collections
# Generation 2 collects after 10 gen-1 collections

# Disable automatic collection (be careful!)
gc.disable()

# Manually trigger collection
collected = gc.collect()
print(f"Collected {collected} objects")

# Check for uncollectable objects (cycles with __del__)
print(gc.garbage)

# Re-enable
gc.enable()

# Debug memory issues
gc.set_debug(gc.DEBUG_LEAK)

import weakref


class User:
    def __init__(self, user_id):
        self.user_id = user_id

    @staticmethod
    def load_from_db(user_id):
        print(f"--- Fetching User {user_id} from Database ---")
        return User(user_id)

    def __repr__(self):
        return f"User(id={self.user_id})"


# 1. Implementation of a Memory-Efficient Cache
class UserCache:
    def __init__(self):
        # Entries are removed when no strong references to the 'User' exist
        self._cache = weakref.WeakValueDictionary()

    def get_user(self, user_id):
        if user_id not in self._cache:
            user = User.load_from_db(user_id)
            self._cache[user_id] = user
        return self._cache[user_id]


# --- Testing the Cache ---
cache = UserCache()

# Create a strong reference
user_1 = cache.get_user(1)
print(f"Cache keys: {list(cache._cache.keys())}")  # → [1]

# Remove the strong reference
del user_1
print(f"Cache keys after deletion: {list(cache._cache.keys())}")  # → []


# The entry vanished from the cache automatically!

# 2. Callback Demo
class SomeClass:
    pass


def on_delete(ref):
    print("Notification: The tracked object has been garbage collected!")


obj = SomeClass()
# Create weak reference with a callback
weak_link = weakref.ref(obj, on_delete)

print("Deleting strong reference...")
del obj
# Note: In some environments, you might need gc.collect() to trigger this immediately

# Without __slots__ (default)
class Point:
    def __init__(self, x, y):
        self.x = x
        self.y = y
        # Attributes stored in __dict__ (dynamic, flexible, memory-heavy)

# With __slots__ (optimized)
class Point:
    __slots__ = ('x', 'y')
    
    def __init__(self, x, y):
        self.x = x
        self.y = y
        # No __dict__, attributes stored in fixed slots

# Memory savings scale with instance count
points = [Point(i, i) for i in range(1_000_000)]
# Could save ~200MB depending on Python version

# Allow weak references with __slots__
import weakref

class User:
    # We save memory on 'name' and 'email',
    # but explicitly pay for a 1-word pointer to allow weakrefs
    __slots__ = ('name', 'email', '__weakref__')

    def __init__(self, name, email):
        self.name = name
        self.email = email

u = User("Alice", "alice@example.com")
r = weakref.ref(u)
print(r())  # Works! Without '__weakref__' in slots, this would raise a TypeError.

# ❌ THE CONFLICT
class A: __slots__ = ('a',)
class B: __slots__ = ('b',)

class C(A, B):
    __slots__ = ('c',)
# Output: TypeError: multiple bases have instance lay-out conflict

# ✅ THE FIX: The "Mix-in" Pattern
# To use inheritance, only ONE base class can have non-empty slots.
class A:
    __slots__ = ('a',)

class B:
    __slots__ = () # Empty slots allow B to be a compatible parent

class C(A, B):
    __slots__ = ('b', 'c') # This works! C inherits 'a' and adds 'b','c'

import tracemalloc

# Start tracking allocations
tracemalloc.start()

# Your code here
data = [i for i in range(1_000_000)]

# Get memory usage
current, peak = tracemalloc.get_traced_memory()
print(f"Current: {current / 1024 / 1024:.2f} MB")
print(f"Peak: {peak / 1024 / 1024:.2f} MB")

# Get top memory-consuming lines
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

for stat in top_stats[:10]:
    print(stat)

tracemalloc.stop()

import objgraph
import gc
import random

# Mock class to demonstrate tracking
class LeakCandidate:
    def __init__(self):
        self.data = [random.random() for _ in range(1000)]

# 1. Basic Memory Audit
# Manually trigger GC to ensure we are looking at 'true' survivors
gc.collect()

print("--- Most Common Objects in Memory ---")
objgraph.show_most_common_types(limit=10)

# 2. Detecting Leaks (The 'Growth' check)
# In a real app, you would call this at different points in time
print("
--- Objects Created Since Last Growth Check ---")
objgraph.show_growth(limit=3)

# 3. Reference Inspection
# Create a cycle to see how objgraph visualizes it
a = LeakCandidate()
b = LeakCandidate()
a.internal_cycle = b
b.internal_cycle = a

# Generate the graph (requires Graphviz installed on the system)
print("
Generating reference graph 'refs.png'...")
objgraph.show_refs([a], filename='refs.png')

# 4. Finding the Culprit
# 'backrefs' shows what is PREVENTING an object from being collected
print("Generating back-reference graph 'backrefs.png'...")
objgraph.show_backrefs([a], filename='backrefs.png')

# Find why an object isn't being collected
import objgraph

# This generates a .dot file and converts to PNG
# It shows the 'chain of command' keeping the object alive
objgraph.show_backrefs([leaky_object], filename='leak_path.png')

# Pro Tip: Use this in production heat-checks to see
# if your caches are growing unexpectedly.