Week 06: CPython 内部机制

线程 1 ──┐
         ├──► GIL ──► Python 解释器
线程 2 ──┘

a = []      # refcount = 1
b = a       # refcount = 2
del a       # refcount = 1
del b       # refcount = 0, 对象被释放

import sys

# 查看/设置切换间隔 (默认 5ms)
print(sys.getswitchinterval())  # 0.005
sys.setswitchinterval(0.01)     # 设为 10ms

# I/O 操作时释放 GIL
with open("file.txt") as f:
    # GIL 释放, 其他线程可以运行
    content = f.read()

# 方案 1: multiprocessing (多进程)
from multiprocessing import Pool

def cpu_task(x):
    return x ** 2

with Pool(4) as p:
    results = p.map(cpu_task, range(1000000))

# 方案 2: C 扩展 (NumPy, Cython)
import numpy as np
result = np.dot(matrix1, matrix2)  # 在 C 中执行, 释放 GIL

# Block 大小以 8 字节对齐
# Size Class 0:   8 bytes
# Size Class 1:  16 bytes
# Size Class 2:  24 bytes
# ...
# Size Class 63: 512 bytes

# 申请 28 字节 → 分配 32 字节 (Size Class 3)

// CPython 源码
typedef struct _object {
    Py_ssize_t ob_refcnt;    // 引用计数
    PyTypeObject *ob_type;   // 类型指针
} PyObject;

// 可变大小对象
typedef struct {
    PyObject ob_base;
    Py_ssize_t ob_size;      // 元素数量
} PyVarObject;

// CPython 源码
typedef struct _object {
    Py_ssize_t ob_refcnt;    // 引用计数
    PyTypeObject *ob_type;   // 类型指针
} PyObject;

import sys

a = [1, 2, 3]
print(sys.getrefcount(a))  # 2 (a + getrefcount 参数)

b = a
print(sys.getrefcount(a))  # 3

del b
print(sys.getrefcount(a))  # 2

a = []
b = []
a.append(b)  # a -> b
b.append(a)  # b -> a

del a, b
# 引用计数仍为 1, 但对象不可达

import gc

# 查看阈值
print(gc.get_threshold())  # (700, 10, 10)

# 手动触发 GC
gc.collect()

# 禁用 GC (谨慎)
gc.disable()

+-------------------+
|    操作系统       |
+-------------------+
        ↑
+-------------------+
|  Python 内存池    | ← 小对象 (≤ 512 bytes)
|  (pymalloc)       |
+-------------------+
        ↑
+-------------------+
|  对象分配器       | ← 具体类型 (int, list, ...)
+-------------------+

# 字符串驻留
a = "hello"
b = "hello"
print(a is b)  # True (驻留)

a = "hello world"
b = "hello world"
print(a is b)  # False (可能不驻留)

# 手动驻留
import sys
a = sys.intern("hello world")
b = sys.intern("hello world")
print(a is b)  # True

print(type(42))        # <class 'int'>
print(type(int))       # <class 'type'>
print(type(type))      # <class 'type'>

print(isinstance(int, type))    # True
print(isinstance(type, type))   # True

class MyClass:
    def __new__(cls, *args, **kwargs):
        """创建实例 (返回实例)"""
        print("__new__ called")
        instance = super().__new__(cls)
        return instance
    
    def __init__(self, value):
        """初始化实例"""
        print("__init__ called")
        self.value = value

obj = MyClass(10)
# __new__ called
# __init__ called

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

class PointWithSlots:
    __slots__ = ["x", "y"]
    
    def __init__(self, x, y):
        self.x = x
        self.y = y

import sys

p1 = PointWithDict(1, 2)
p2 = PointWithSlots(1, 2)

print(sys.getsizeof(p1.__dict__))  # 64+ bytes
# p2 没有 __dict__, 更节省内存

# 不能动态添加属性
# p2.z = 3  # AttributeError

源代码 (.py) → 字节码 (.pyc) → 虚拟机执行

import dis

def add(a, b):
    return a + b

dis.dis(add)

  2           0 LOAD_FAST                0 (a)
              2 LOAD_FAST                1 (b)
              4 BINARY_ADD
              6 RETURN_VALUE

def example():
    x = 1
    return x + 2

code = example.__code__

print(code.co_varnames)    # ('x',)
print(code.co_consts)      # (None, 1, 2)
print(code.co_code)        # 字节码 (bytes)
print(code.co_stacksize)   # 栈大小

import weakref

class MyClass:
    pass

obj = MyClass()
weak_ref = weakref.ref(obj)

print(weak_ref())  # <__main__.MyClass object at ...>

del obj
print(weak_ref())  # None (对象已被回收)

import weakref

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

cache = weakref.WeakValueDictionary()

def get_item(key):
    if key not in cache:
        cache[key] = CachedItem(key * 2)
    return cache[key]

item = get_item(10)
print(cache[10].value)  # 20

del item
# cache[10] 也被移除 (对象被回收)

import weakref

class Resource:
    def __init__(self, name):
        self.name = name

def cleanup(name):
    print(f"Cleaning up {name}")

r = Resource("MyResource")
weakref.finalize(r, cleanup, r.name)

del r
# Cleaning up MyResource

import cProfile

def slow_function():
    total = 0
    for i in range(1000000):
        total += i
    return total

cProfile.run("slow_function()")

import timeit

# 比较列表推导式 vs 循环
list_comp = timeit.timeit("[x**2 for x in range(1000)]", number=1000)
loop = timeit.timeit("""
result = []
for x in range(1000):
    result.append(x**2)
""", number=1000)

print(f"列表推导式: {list_comp:.4f}s")
print(f"循环: {loop:.4f}s")

# 慢
def slow():
    for i in range(1000000):
        len([1, 2, 3])  # 每次查找全局 len

# 快
def fast():
    local_len = len  # 局部变量
    for i in range(1000000):
        local_len([1, 2, 3])

# 占用大量内存
def squares_list(n):
    return [x**2 for x in range(n)]

# 惰性生成
def squares_gen(n):
    for x in range(n):
        yield x**2

# 慢
total = 0
for x in numbers:
    total += x

# 快
total = sum(numbers)

import tracemalloc

# 开始追踪
tracemalloc.start()

# 执行可能泄漏的代码
data = [list(range(1000)) for _ in range(1000)]

# 获取快照
snapshot = tracemalloc.take_snapshot()

# 显示 Top 10 内存分配位置
top_stats = snapshot.statistics("lineno")
for stat in top_stats[:10]:
    print(stat)

# 比较两个快照 (检测增量)
snapshot1 = tracemalloc.take_snapshot()
# ... 执行更多代码 ...
snapshot2 = tracemalloc.take_snapshot()

diff = snapshot2.compare_to(snapshot1, "lineno")
for stat in diff[:5]:
    print(stat)

# 停止追踪
tracemalloc.stop()

import tracemalloc
from functools import wraps

def memory_profile(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        tracemalloc.start()
        result = func(*args, **kwargs)
        current, peak = tracemalloc.get_traced_memory()
        tracemalloc.stop()
        print(f"{func.__name__}: current={current/1024:.1f}KB, peak={peak/1024:.1f}KB")
        return result
    return wrapper

@memory_profile
def process_data():
    return [x**2 for x in range(100000)]