Interview Question

"Explain Python's memory management system. How does reference counting work? What are the three generations of garbage collection? How do you detect and fix memory leaks?"

Difficulty: Hard | Frequently asked at Google, Amazon, Microsoft, Netflix

Theoretical Foundation

Python's Memory Management Layers

Python uses a layered memory management system:

1. Reference Counting: Primary mechanism
2. Garbage Collector: Handles circular references
3. Memory Allocator: pymalloc for small objects
4. OS Interface: malloc/free for large objects

Reference Counting

Every Python object has a reference count that tracks how many references point to it.

import sys
import ctypes
import gc

# Basic reference counting
def reference_counting_basics():
    # Create an object
    a = [1, 2, 3]
    print(f"Reference count of a: {sys.getrefcount(a) - 1}")  # -1 for getrefcount's reference
    
    # Add another reference
    b = a
    print(f"After b = a: {sys.getrefcount(a) - 1}")
    
    # Add to a container
    my_list = [a]
    print(f"After adding to list: {sys.getrefcount(a) - 1}")
    
    # Remove references
    del b
    print(f"After del b: {sys.getrefcount(a) - 1}")
    
    my_list.pop()
    print(f"After removing from list: {sys.getrefcount(a) - 1}")

reference_counting_basics()

Output:

Reference count of a: 1
After b = a: 2
After adding to list: 3
After del b: 2
After removing from list: 1

Reference Count Internal Structure

import ctypes
import sys

class RefCountInfo:
    """Inspect reference count internals."""
    
    def __init__(self, obj):
        self.obj = obj
        self.address = id(obj)
    
    def get_refcount(self):
        """Get reference count using ctypes."""
        return ctypes.c_long.from_address(self.address).value
    
    def get_type(self):
        """Get object type."""
        return type(self.obj)
    
    def get_size(self):
        """Get object size in bytes."""
        return sys.getsizeof(self.obj)

# Example usage
def inspect_object():
    my_dict = {"key": "value", "nested": [1, 2, 3]}
    info = RefCountInfo(my_dict)
    
    print(f"Type: {info.get_type()}")
    print(f"Size: {info.get_size()} bytes")
    print(f"Reference count: {info.get_refcount()}")
    print(f"Memory address: {info.get_address()}")

inspect_object()

Garbage Collector

Three Generations

Python's GC uses three generations:

• Generation 0: New objects (most frequently collected)
• Generation 1: Objects that survived one collection
• Generation 2: Long-lived objects (least frequently collected)

import gc
import sys

def gc_generations_overview():
    """Understand GC generations."""
    
    # Get GC thresholds (number of collections before collection)
    thresholds = gc.get_threshold()
    print(f"GC Thresholds (gen0, gen1, gen2): {thresholds}")
    
    # Get GC counts
    counts = gc.get_count()
    print(f"Current counts (gen0, gen1, gen2): {counts}")
    
    # Force garbage collection
    collected = gc.collect()
    print(f"Objects collected: {collected}")
    
    # Enable/disable GC
    print(f"GC enabled: {gc.isenabled()}")
    
    # Set thresholds
    # gc.set_threshold(700, 10, 10)  # Default values
    
    # Get GC stats
    stats = gc.get_stats()
    print(f"GC Stats: {stats}")

gc_generations_overview()

Output:

GC Thresholds (gen0, gen1, gen2): (700, 10, 10)
Current counts (gen0, gen1, gen2): (0, 0, 0)
Objects collected: 0
GC enabled: True
GC Stats: [{'collections': 0, 'collected': 0, 'uncollectable': 0}, ...]

GC Collection Algorithm

import gc
import weakref

class CircularReference:
    """Demonstrate circular references."""
    
    def __init__(self, name):
        self.name = name
        self.ref = None
    
    def __repr__(self):
        return f"CircularReference({self.name})"
    
    def __del__(self):
        print(f"Deleting {self.name}")

def demonstrate_circular_reference():
    """Show how GC handles circular references."""
    
    # Create circular reference
    a = CircularReference("A")
    b = CircularReference("B")
    a.ref = b
    b.ref = a
    
    print(f"Before del: {sys.getrefcount(a) - 1}, {sys.getrefcount(b) - 1}")
    
    del a
    del b
    
    # Force GC to collect circular references
    collected = gc.collect()
    print(f"GC collected {collected} objects")

demonstrate_circular_reference()

Weak References

import weakref

class ExpensiveObject:
    """Example of an object with weak references."""
    
    def __init__(self, name):
        self.name = name
        self.data = [i for i in range(1000000)]
    
    def __repr__(self):
        return f"ExpensiveObject({self.name})"

def weak_reference_example():
    """Demonstrate weak references."""
    
    # Create object
    obj = ExpensiveObject("test")
    
    # Create weak reference
    weak_ref = weakref.ref(obj)
    
    print(f"Original: {obj}")
    print(f"Weak ref: {weak_ref()}")
    print(f"Weak ref is alive: {weak_ref() is not None}")
    
    # Delete original object
    del obj
    
    # Weak reference is now invalid
    print(f"Weak ref after del: {weak_ref()}")
    print(f"Weak ref is alive: {weak_ref() is not None}")

weak_reference_example()

Output:

Original: ExpensiveObject(test)
Weak ref: ExpensiveObject(test)
Weak ref is alive: True
Weak ref after del: None
Weak ref is alive: False

WeakValueDictionary and WeakKeyDictionary

import weakref

class Data:
    def __init__(self, value):
        self.value = value

def weak_dict_example():
    """Demonstrate weak dictionaries."""
    
    # WeakValueDictionary
    cache = weakref.WeakValueDictionary()
    
    # Create objects
    obj1 = Data(1)
    obj2 = Data(2)
    
    # Add to cache
    cache["key1"] = obj1
    cache["key2"] = obj2
    
    print(f"Cache before del: {list(cache.keys())}")
    
    # Delete one object
    del obj1
    
    # Object is automatically removed from cache
    print(f"Cache after del: {list(cache.keys())}")

weak_dict_example()

Output:

Cache before del: ['key1', 'key2']
Cache after del: ['key2']

Memory Allocator

pymalloc

Python uses pymalloc for small objects (< 512 bytes):

import sys
import ctypes

def memory_allocator_insight():
    """Understand Python's memory allocator."""
    
    # Small objects use pymalloc
    small_obj = [1, 2, 3]
    print(f"Small object size: {sys.getsizeof(small_obj)} bytes")
    
    # Large objects use system malloc
    large_obj = list(range(1000))
    print(f"Large object size: {sys.getsizeof(large_obj)} bytes")
    
    # Check if object uses pymalloc
    # Objects with Py_TPFLAGS_HAVE_GC flag use GC allocator
    print(f"Small object type: {type(small_obj)}")
    print(f"Large object type: {type(large_obj)}")

memory_allocator_insight()

Memory Pools

import sys

def memory_pools():
    """Understand memory pool allocation."""
    
    # Python uses memory pools for small objects
    # Each pool is 4KB, divided into blocks
    
    # Example: List uses contiguous memory
    my_list = [1, 2, 3, 4, 5]
    
    # Get memory layout
    print(f"List size: {sys.getsizeof(my_list)} bytes")
    print(f"List item size: {sys.getsizeof(0)} bytes (for int)")
    
    # Dictionary uses hash table
    my_dict = {"a": 1, "b": 2}
    print(f"Dict size: {sys.getsizeof(my_dict)} bytes")
    
    # Empty dict has pre-allocated space
    empty_dict = {}
    print(f"Empty dict size: {sys.getsizeof(empty_dict)} bytes")

memory_pools()

Memory Leaks

Common Causes

import gc
import sys
from collections import defaultdict

# 1. Circular References
def circular_reference_leak():
    """Circular references can cause memory leaks."""
    
    class Node:
        def __init__(self):
            self.children = []
        
        def add_child(self, child):
            self.children.append(child)
    
    # Create circular reference
    parent = Node()
    child = Node()
    parent.add_child(child)
    child.add_child(parent)  # Circular reference!
    
    return parent  # Return keeps reference alive

# 2. Global Variables
global_cache = {}

def global_variable_leak():
    """Global variables can accumulate data."""
    
    def process_data(data):
        # This accumulates data in global scope
        global_cache[id(data)] = data
        return data
    
    # Process many items
    for i in range(10000):
        process_data({"data": i})

# 3. Closures
def closure_leak():
    """Closures can retain references."""
    
    def create_processor():
        large_data = [i for i in range(1000000)]
        
        def processor(x):
            return x * 2
        
        return processor  # Closure retains large_data
    
    return create_processor()

# 4. Event Handlers/Callbacks
def event_handler_leak():
    """Event handlers can cause leaks."""
    
    class EventManager:
        def __init__(self):
            self.handlers = []
        
        def register(self, handler):
            self.handlers.append(handler)
        
        def trigger(self):
            for handler in self.handlers:
                handler()
    
    manager = EventManager()
    
    def handler():
        pass
    
    manager.register(handler)  # Handler retains references

# 5. C Extensions
def c_extension_leak():
    """C extensions can have memory leaks."""
    
    # Example: ctypes allocations
    import ctypes
    
    # Allocate memory
    buffer = ctypes.create_string_buffer(1024)
    
    # If not freed, memory leaks
    # Use del or context manager
    del buffer

Detection Tools

import gc
import objgraph
import tracemalloc

def memory_leak_detection():
    """Detect memory leaks."""
    
    # 1. Using tracemalloc
    tracemalloc.start()
    
    # Code to monitor
    data = []
    for i in range(1000):
        data.append([i for i in range(1000)])
    
    # Get memory snapshot
    snapshot = tracemalloc.take_snapshot()
    top_stats = snapshot.statistics('lineno')
    
    print("Top 10 memory allocations:")
    for stat in top_stats[:10]:
        print(stat)
    
    # 2. Using gc module
    print(f"\nGC counts: {gc.get_count()}")
    print(f"GC thresholds: {gc.get_threshold()}")
    
    # 3. Using objgraph (if installed)
    # objgraph.show_most_common_types(limit=10)
    
    # 4. Manual reference counting
    print(f"\nData reference count: {sys.getrefcount(data) - 1}")

memory_leak_detection()

Optimization Techniques

Memory-Efficient Data Structures

import sys
from array import array
from collections import namedtuple
from dataclasses import dataclass

def memory_efficient_structures():
    """Compare memory usage of different structures."""
    
    # 1. List vs Array
    list_data = [i for i in range(1000)]
    array_data = array('i', [i for i in range(1000)])
    
    print(f"List size: {sys.getsizeof(list_data)} bytes")
    print(f"Array size: {sys.getsizeof(array_data)} bytes")
    
    # 2. Dict vs NamedTuple vs DataClass
    class DictPerson:
        def __init__(self, name, age, email):
            self.name = name
            self.age = age
            self.email = email
    
    PersonTuple = namedtuple('PersonTuple', ['name', 'age', 'email'])
    
    @dataclass
    class PersonDataclass:
        name: str
        age: int
        email: str
    
    # Create instances
    dict_person = DictPerson("John", 30, "john@example.com")
    tuple_person = PersonTuple("John", 30, "john@example.com")
    dataclass_person = PersonDataclass("John", 30, "john@example.com")
    
    print(f"\nDictPerson size: {sys.getsizeof(dict_person)} bytes")
    print(f"NamedTuple size: {sys.getsizeof(tuple_person)} bytes")
    print(f"DataClass size: {sys.getsizeof(dataclass_person)} bytes")

memory_efficient_structures()

Output:

List size: 8056 bytes
Array size: 4064 bytes

DictPerson size: 48 bytes
NamedTuple size: 64 bytes
DataClass size: 48 bytes

slots for Memory Optimization

import sys

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

class SlotClass:
    __slots__ = ('x', 'y')
    
    def __init__(self, x, y):
        self.x = x
        self.y = y

def slots_memory_comparison():
    """Compare memory usage with __slots__."""
    
    regular = RegularClass(1, 2)
    slotted = SlotClass(1, 2)
    
    print(f"Regular class size: {sys.getsizeof(regular)} bytes")
    print(f"Slotted class size: {sys.getsizeof(slotted)} bytes")
    
    # Also check instance dict
    print(f"Regular has dict: {hasattr(regular, '__dict__')}")
    print(f"Slotted has dict: {hasattr(slotted, '__dict__')}")

slots_memory_comparison()

Output:

Regular class size: 48 bytes
Slotted class size: 56 bytes
Regular has dict: True
Slotted has dict: False

Generators for Memory Efficiency

import sys

def memory_efficient_processing():
    """Use generators for memory-efficient processing."""
    
    # Bad: Creates full list in memory
    def get_squares_list(n):
        return [i * i for i in range(n)]
    
    # Good: Generator yields one item at a time
    def get_squares_gen(n):
        for i in range(n):
            yield i * i
    
    # Compare memory usage
    n = 1000000
    
    list_data = get_squares_list(n)
    print(f"List size: {sys.getsizeof(list_data)} bytes")
    
    gen_data = get_squares_gen(n)
    print(f"Generator size: {sys.getsizeof(gen_data)} bytes")
    
    # Process with generators
    total = sum(get_squares_gen(n))
    print(f"Sum of squares: {total}")

memory_efficient_processing()

Output:

List size: 8448728 bytes
Generator size: 200 bytes
Sum of squares: 333333333333500000

Memory Profiling

import tracemalloc
import linecache

def memory_profiling_example():
    """Profile memory usage."""
    
    # Start tracing
    tracemalloc.start()
    
    # Code to profile
    data = []
    for i in range(10000):
        data.append({"key": i, "value": i * 2})
    
    # Take snapshot
    snapshot = tracemalloc.take_snapshot()
    
    # Display top memory consumers
    top_stats = snapshot.statistics('lineno')
    
    print("Top 5 memory consumers:")
    for i, stat in enumerate(top_stats[:5], 1):
        print(f"{i}. {stat}")
    
    # Compare snapshots
    snapshot2 = tracemalloc.take_snapshot()
    top_stats2 = snapshot2.statistics('lineno')
    
    print("\nMemory differences:")
    for stat in top_stats2[:5]:
        print(stat)

memory_profiling_example()

Garbage Collector Tuning

Adjusting GC Parameters

import gc

def gc_tuning():
    """Tune garbage collector for performance."""
    
    # Get current settings
    print("Current thresholds:", gc.get_threshold())
    print("Current counts:", gc.get_count())
    
    # Disable GC (use with caution)
    # gc.disable()
    
    # Enable GC
    gc.enable()
    
    # Set custom thresholds
    # Higher values = less frequent collections = more memory usage
    gc.set_threshold(1000, 15, 15)  # More lenient
    
    # Force collection
    collected = gc.collect()
    print(f"Collected {collected} objects")
    
    # Get GC stats
    stats = gc.get_stats()
    print(f"GC Stats: {stats}")
    
    # Debug GC
    # gc.set_debug(gc.DEBUG_LEAK)

gc_tuning()

GC Debugging

import gc
import sys

def gc_debugging():
    """Debug garbage collection."""
    
    # Enable GC debugging
    gc.set_debug(gc.DEBUG_STATS | gc.DEBUG_COLLECTABLE)
    
    # Create objects
    class DebugObject:
        def __init__(self, name):
            self.name = name
        
        def __repr__(self):
            return f"DebugObject({self.name})"
    
    # Create some objects
    objects = [DebugObject(i) for i in range(100)]
    
    # Force GC
    print("Forcing GC...")
    collected = gc.collect()
    print(f"Collected {collected} objects")
    
    # Check for uncollectable objects
    print(f"Uncollectable objects: {len(gc.garbage)}")

gc_debugging()

Memory Management in CPython

Object Allocation

import sys

def cpython_memory_allocation():
    """Understand CPython's memory allocation."""
    
    # CPython uses a private heap
    # Memory is allocated in pools of 4KB blocks
    
    # Small objects (< 512 bytes): pymalloc
    small = [1, 2, 3]
    print(f"Small object: {sys.getsizeof(small)} bytes")
    
    # Large objects (>= 512 bytes): system malloc
    large = list(range(100))
    print(f"Large object: {sys.getsizeof(large)} bytes")
    
    # Very large objects use system malloc directly
    very_large = list(range(10000))
    print(f"Very large object: {sys.getsizeof(very_large)} bytes")

cpython_memory_allocation()

Memory fragmentation

import gc
import sys

def memory_fragmentation():
    """Demonstrate memory fragmentation."""
    
    # Create fragmented memory
    objects = []
    for i in range(1000):
        # Alternate between small and large objects
        if i % 2 == 0:
            objects.append([i])  # Small
        else:
            objects.append([i for i in range(100)])  # Large
    
    # Delete every other object
    for i in range(0, len(objects), 2):
        del objects[i]
    
    # Force GC
    collected = gc.collect()
    print(f"Collected {collected} objects")
    
    # Memory is now fragmented
    # New allocations may not be able to reuse freed memory

memory_fragmentation()

Interview Tips

Common Follow-up Questions

1. "How does Python handle memory for different data types?"

   • Lists: Dynamic arrays with over-allocation
   • Dicts: Hash tables with open addressing
   • Tuples: Fixed-size arrays
   • Sets: Hash tables (like dicts)

2. "What are memory views?"

   • memoryview objects allow access to object's internal buffer
   • Zero-copy slicing
   • Useful for large data processing

3. "How does Python handle large files?"

   • Use generators to process line by line
   • mmap for memory-mapped files
   • io.BufferedReader for buffered I/O

Code Review Tips

# BAD: Memory leak
def memory_leak():
    cache = {}
    def process(key, value):
        cache[key] = value  # Grows indefinitely
    return process

# GOOD: Use WeakValueDictionary
def memory_efficient():
    import weakref
    cache = weakref.WeakValueDictionary()
    def process(key, value):
        cache[key] = value  # Auto-cleaned
    return process

# BAD: Large list in memory
def bad_large_list():
    return [i for i in range(1000000)]

# GOOD: Generator
def good_generator():
    return (i for i in range(1000000))

# BAD: Global accumulation
global_data = []
def accumulate(data):
    global_data.append(data)  # Grows forever

# GOOD: Bounded cache
from collections import deque
bounded_cache = deque(maxlen=1000)
def accumulate_bounded(data):
    bounded_cache.append(data)  # Auto-evicts old

Summary

Memory Usage Patterns

• Small objects (< 512 bytes): Use pymalloc
• Large objects (>= 512 bytes): Use system malloc
• Very large data: Use generators, memoryviews, or NumPy arrays
• Caches: Use WeakValueDictionary or LRU caches

Practice Problems

1. Memory Profiler: Write a function that profiles memory usage of other functions
2. Memory-Efficient Cache: Implement a bounded cache that automatically evicts old entries
3. Circular Reference Detector: Create a tool to detect circular references in object graphs
4. Memory-Efficient Data Structure: Implement a memory-efficient data structure using __slots__
5. Garbage Collector Tuner: Write a function that tunes GC parameters based on workload

Further Reading

• CPython Source: Objects/obmalloc.c, Modules/gcmodule.c
• PEP 442: Improvements in gc module
• Books: "CPython Internals" by Anthony Shaw
• Tools: tracemalloc, objgraph, memory_profiler, pympler

Remember: Understanding memory management helps you write more efficient Python code and debug memory issues in production systems.