Memory Management & Optimization

import gc
import sys
import weakref

# Check reference count
x = []
print(sys.getrefcount(x))

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

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


# Circular references are NOT handled by refcounting alone
class Node:
    pass

gc.disable()  # so only the explicit collect() below can free the cycle

a, b = Node(), Node()
a.other = b   # a -> b
b.other = a   # b -> a, the refcounts never reach zero
watch = weakref.ref(a)

del a, b
print("alive after del:", watch() is not None)

gc.collect()  # the cycle detector is what actually frees it
print("alive after gc.collect():", watch() is not None)

gc.enable()

import gc


class Node:
    pass


# Get current GC thresholds
print(gc.get_threshold())
# 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()

# Build a cycle, which is exactly what refcounting cannot free
a, b = Node(), Node()
a.other, b.other = b, a
del a, b

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

# Cycles Python refuses to free end up here, not in the collected count
print(gc.garbage)

# DEBUG_COLLECTABLE reports each cycle member as it is freed
gc.set_debug(gc.DEBUG_COLLECTABLE)
c, d = Node(), Node()
c.other, d.other = d, c
del c, d
gc.collect()

# Always reset the flags, or the noise follows you to interpreter exit
gc.set_debug(0)
gc.enable()

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())}")

# 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

import tracemalloc


# Without __slots__: every instance carries its own __dict__
class PointDict:
    def __init__(self, x, y):
        self.x = x
        self.y = y


# With __slots__: attributes live in fixed slots, no __dict__ at all
class PointSlots:
    __slots__ = ('x', 'y')

    def __init__(self, x, y):
        self.x = x
        self.y = y


# Memory savings scale with instance count, so measure at scale
N = 1_000_000
for cls in (PointDict, PointSlots):
    tracemalloc.start()
    points = [cls(i, i) for i in range(N)]
    used = tracemalloc.get_traced_memory()[0]
    tracemalloc.stop()
    print(f"{cls.__name__:10} {used / 1024 / 1024:6.1f} MB  ({used / N:.0f} bytes per instance)")
    del points

# Allow weak references with __slots__
import weakref


class User:
    # We save memory on 'name' and 'email', but explicitly pay for a
    # 1-word pointer so instances can still be weakly referenced
    __slots__ = ('name', 'email', '__weakref__')

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


class Session:
    # No '__weakref__' here, so this one cannot be tracked weakly
    __slots__ = ('token',)

    def __init__(self, token):
        self.token = token


u = User("Alice", "alice@example.com")
r = weakref.ref(u)
print(r() is u, r().email)

try:
    weakref.ref(Session("abc123"))
except TypeError as e:
    print(f"TypeError: {e}")

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


class B:
    __slots__ = ('b',)


try:
    class C(A, B):
        __slots__ = ('c',)
except TypeError as e:
    print(f"TypeError: {e}")


# ✅ THE FIX: the "Mix-in" Pattern
# To use inheritance, only ONE base class can have non-empty slots.
class Mixin:
    __slots__ = ()  # empty slots keep Mixin a compatible parent


class D(A, Mixin):
    __slots__ = ('d',)


d = D()
d.a, d.d = 1, 2
print(d.a, d.d)

# Still no per-instance __dict__, so the saving survives the inheritance
print(hasattr(d, '__dict__'))

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 the top memory-consuming lines. Report line numbers rather than the
# full traceback so the result does not depend on where you saved the file.
snapshot = tracemalloc.take_snapshot()
for stat in snapshot.statistics('lineno')[:3]:
    frame = stat.traceback[0]
    print(f"line {frame.lineno}: {stat.size / 1024:.1f} KiB in {stat.count} blocks")

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("\n--- 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("\nGenerating 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

# The classic culprit: a module-level cache nobody ever clears
_session_cache = []


class Session:
    pass


leaky_object = Session()
_session_cache.append(leaky_object)

# 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.