Performance Engineering: Profiling, __slots__, C Extensions & Free-Threaded Python

import cProfile
import pstats
import io

# ── cProfile: call-count + time per function ──
def slow_function():
    total = 0
    for i in range(100_000):
        total += i * i
    return total

# Method 1: enable/disable
pr = cProfile.Profile()
pr.enable()
result = slow_function()
pr.disable()

stream = io.StringIO()
ps = pstats.Stats(pr, stream=stream).sort_stats("cumulative")
ps.print_stats(10)   # top 10 functions by cumulative time
print(stream.getvalue())

# Method 2: context manager
with cProfile.Profile() as pr:
    slow_function()
pstats.Stats(pr).sort_stats("tottime").print_stats(5)

# Method 3: command line (most convenient)
# python -m cProfile -s cumulative my_script.py
# python -m cProfile -o output.prof my_script.py
# python -m pstats output.prof   → interactive browser

pip install snakeviz
python -m cProfile -o profile.prof my_script.py
snakeviz profile.prof   # opens a sunburst diagram in the browser

pip install line_profiler memory_profiler

# ── line_profiler: time per line ──
# Run: kernprof -l -v script.py  (from the command line)
# The @profile decorator is injected automatically by kernprof

@profile   # noqa — injected by kernprof
def process_data(data: list[int]) -> int:
    result = 0
    for item in data:          # line 1
        if item % 2 == 0:      # line 2
            result += item     # line 3
        else:
            result -= item     # line 4
    return result

data = list(range(1_000_000))
process_data(data)

# ── memory_profiler: memory per line ──
from memory_profiler import profile as mprofile

@mprofile
def build_large_dict(n: int) -> dict:
    d = {}
    for i in range(n):
        d[i] = f"value_{i}"
    return d

build_large_dict(100_000)
# Output shows memory increment per line

# ── tracemalloc: built-in allocation tracking ──
import tracemalloc

tracemalloc.start()
data = [{"id": i, "val": str(i)} for i in range(10_000)]
snapshot = tracemalloc.take_snapshot()
for stat in snapshot.statistics("lineno")[:5]:
    print(stat)
tracemalloc.stop()

# Record a flamegraph of a running process
py-spy record --pid $(pgrep -f my_server.py) -o flamegraph.svg --duration 30

import timeit

# ── 1. Cache globals as locals in hot loops ──
import math

def slow_sqrt_sum(n: int) -> float:
    total = 0.0
    for i in range(n):
        total += math.sqrt(i)   # LOAD_GLOBAL math, LOAD_ATTR sqrt every iteration
    return total

def fast_sqrt_sum(n: int) -> float:
    total = 0.0
    sqrt = math.sqrt            # cached as local — LOAD_FAST is faster than LOAD_GLOBAL
    for i in range(n):
        total += sqrt(i)
    return total

n = 1_000_000
print(timeit.timeit(lambda: slow_sqrt_sum(n), number=3))
print(timeit.timeit(lambda: fast_sqrt_sum(n), number=3))

# ── 2. __slots__ (covered in Lesson 45 but benchmarked here) ──
import sys

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

class WithSlots:
    __slots__ = ("x", "y", "z")
    def __init__(self, x, y, z): self.x, self.y, self.z = x, y, z

print(f"WithDict : {sys.getsizeof(WithDict(1,2,3))} B")
print(f"WithSlots: {sys.getsizeof(WithSlots(1,2,3))} B")

# ── 3. List vs set membership ──
needle = 999_999
large_list = list(range(1_000_000))
large_set  = set(large_list)

t_list = timeit.timeit(lambda: needle in large_list, number=100)
t_set  = timeit.timeit(lambda: needle in large_set,  number=100)
print(f"list: {t_list:.4f}s  set: {t_set:.6f}s  speedup: {t_list/t_set:.0f}x")

# ── 4. String joining ──
parts = ["a"] * 10_000

slow = timeit.timeit(lambda: "".join(str(p) for p in parts), number=1000)
fast = timeit.timeit(lambda: "".join(parts), number=1000)
print(f"generator join: {slow:.3f}s  list join: {fast:.3f}s")

# ── 5. functools.lru_cache for memoisation ──
from functools import lru_cache

@lru_cache(maxsize=None)
def fib(n: int) -> int:
    if n < 2: return n
    return fib(n - 1) + fib(n - 2)

print(fib(50))   # instantaneous; without cache: 2^50 calls

# ── 6. Avoid repeated attribute lookup ──
class Processor:
    def run_slow(self, data):
        result = []
        for x in data:
            result.append(x * 2)   # result.append looked up N times
        return result

    def run_fast(self, data):
        append = list.append       # cache the method
        result = []
        for x in data:
            append(result, x * 2)
        return result

import numpy as np
import timeit

N = 1_000_000

# ── Dot product ──
a_list = list(range(N))
b_list = list(range(N))
a_arr  = np.arange(N, dtype=np.float64)
b_arr  = np.arange(N, dtype=np.float64)

t_py  = timeit.timeit(lambda: sum(x*y for x, y in zip(a_list, b_list)), number=3)
t_np  = timeit.timeit(lambda: np.dot(a_arr, b_arr), number=3)
print(f"Python dot: {t_py:.3f}s  NumPy dot: {t_np:.5f}s  speedup: {t_py/t_np:.0f}x")

# ── Avoiding temporary arrays with in-place ops ──
rng  = np.random.default_rng(42)
data = rng.random((1000, 1000))

# Creates 3 intermediate arrays:
slow_result = (data * 2 + 1) ** 2

# No intermediate arrays — modify in-place:
fast_result = data.copy()
fast_result *= 2
fast_result += 1
fast_result **= 2

# ── numexpr for multi-array expressions ──
# pip install numexpr
try:
    import numexpr as ne
    expr = "(data * 2 + 1) ** 2"
    ne_result = ne.evaluate(expr)   # compiles to C, uses multiple threads
except ImportError:
    pass

# ── When NOT to use NumPy ──
# Small arrays (< ~100 elements): Python overhead > NumPy benefit
# Non-numeric data: strings, objects, mixed types
# Single-element operations: scalar math is faster in pure Python

/* fast_math.c — a simple C extension */
#define PY_SSIZE_T_CLEAN
#include <Python.h>
#include <math.h>

/* C function: sum of squares */
static PyObject *
sum_of_squares(PyObject *self, PyObject *args)
{
    long n;
    if (!PyArg_ParseTuple(args, "l", &n))   /* parse one long integer arg */
        return NULL;

    double total = 0.0;
    for (long i = 0; i < n; i++)
        total += (double)i * (double)i;

    return PyFloat_FromDouble(total);
}

/* Module method table */
static PyMethodDef FastMathMethods[] = {
    {"sum_of_squares", sum_of_squares, METH_VARARGS,
     "Compute sum of squares 0..n-1 (fast C implementation)."},
    {NULL, NULL, 0, NULL}   /* sentinel */
};

/* Module definition */
static struct PyModuleDef fastmathmodule = {
    PyModuleDef_HEAD_INIT,
    "fast_math",           /* module name */
    NULL,                  /* module docstring */
    -1,                    /* per-interpreter state (-1 = global) */
    FastMathMethods
};

/* Module init function — must be named PyInit_<modulename> */
PyMODINIT_FUNC
PyInit_fast_math(void)
{
    return PyModule_Create(&fastmathmodule);
}

# setup.py
from setuptools import setup, Extension

module = Extension(
    "fast_math",
    sources=["fast_math.c"],
    extra_compile_args=["-O3", "-march=native"],
)

setup(
    name="fast_math",
    ext_modules=[module],
)

python setup.py build_ext --inplace
python -c "import fast_math; print(fast_math.sum_of_squares(1_000_000))"

[build-system]
requires = ["setuptools", "wheel"]
build-backend = "setuptools.backends.legacy:build"

[project]
name = "fast_math"

pip install cython

# fast_sum.pyx — Cython source
# Type annotations compile to C stack variables (no PyObject overhead)
def sum_squares_py(int n) -> float:
    """Pure Python fallback — no type annotations."""
    total = 0.0
    for i in range(n):
        total += i * i
    return total

def sum_squares_typed(int n) -> double:
    """Typed Cython — compiles to efficient C loop."""
    cdef int i
    cdef double total = 0.0
    for i in range(n):
        total += <double>i * i
    return total

# ── Cython with NumPy arrays (typed memoryviews) ──
import numpy as np
cimport numpy as cnp

def dot_product(cnp.ndarray[double, ndim=1] a,
                cnp.ndarray[double, ndim=1] b) -> double:
    cdef int n = a.shape[0]
    cdef int i
    cdef double result = 0.0
    for i in range(n):
        result += a[i] * b[i]
    return result

# setup_cython.py
from setuptools import setup
from Cython.Build import cythonize
import numpy as np

setup(
    ext_modules=cythonize("fast_sum.pyx", annotate=True),
    include_dirs=[np.get_include()],
)

python setup_cython.py build_ext --inplace
python -c "import fast_sum; print(fast_sum.sum_squares_typed(1_000_000))"

# Generate annotation HTML to see Python vs C lines:
cython -a fast_sum.pyx

# Install free-threaded Python (e.g. via pyenv)
pyenv install 3.13t   # 't' suffix = free-threaded

# Verify GIL status
python -c "import sys; print(sys._is_gil_enabled())"   # False in free-threaded build

# Traditional GIL limitation — threads cannot run CPU-bound code in parallel
import threading, time

def cpu_work(n: int) -> int:
    total = 0
    for i in range(n): total += i
    return total

N = 5_000_000

# Single thread
t0 = time.perf_counter()
cpu_work(N); cpu_work(N)
print(f"Sequential:  {time.perf_counter()-t0:.3f}s")

# Two threads (with GIL: no speedup for CPU-bound)
t0 = time.perf_counter()
t1 = threading.Thread(target=cpu_work, args=(N,))
t2 = threading.Thread(target=cpu_work, args=(N,))
t1.start(); t2.start(); t1.join(); t2.join()
print(f"2 threads (GIL):    {time.perf_counter()-t0:.3f}s")  # roughly same as sequential

# In free-threaded Python 3.13 (no GIL):
# The two threads run on separate CPU cores simultaneously
# → ~2x speedup for CPU-bound workloads

import ctypes, os

# ── ctypes: load a shared library and call C functions ──
# Load libc (Linux/macOS)
libc = ctypes.CDLL("libc.so.6" if os.name != "nt" else "msvcrt.dll")

# Call strlen — tell ctypes the argument and return types
libc.strlen.argtypes = [ctypes.c_char_p]
libc.strlen.restype  = ctypes.c_size_t
print(libc.strlen(b"hello"))   # 5

# Create a C array and sum it
ArrayType = ctypes.c_double * 5
arr = ArrayType(1.0, 2.0, 3.0, 4.0, 5.0)
print(sum(arr))   # 15.0

# ── ctypes with a custom shared library ──
# Compile: gcc -O3 -shared -fPIC -o libfast.so fast_math.c
# lib = ctypes.CDLL("./libfast.so")
# lib.sum_of_squares.argtypes = [ctypes.c_long]
# lib.sum_of_squares.restype  = ctypes.c_double
# print(lib.sum_of_squares(1_000_000))

# ── cffi (C Foreign Function Interface) — more Pythonic than ctypes ──
# pip install cffi
from cffi import FFI

ffi = FFI()

# Declare the C function signature
ffi.cdef("""
    double sum_squares(long n);
    size_t strlen(const char *s);
""")

# Load the library (ABI mode — no compilation needed for existing .so)
lib = ffi.dlopen("libc.so.6")   # or your custom .so

# Call strlen
s = ffi.new("char[]", b"hello world")
print(lib.strlen(s))   # 11

# ── cffi in-line mode: embed C code directly in Python ──
ffi2 = FFI()
ffi2.cdef("int add(int a, int b);")

lib2 = ffi2.verify("""
    int add(int a, int b) { return a + b; }
""")
print(lib2.add(3, 4))   # 7  — compiled and loaded on the fly

pip install numba

from numba import njit, prange
import numpy as np
import timeit

# ── @njit: compile to native machine code (no Python overhead) ──
@njit
def sum_squares_numba(n: int) -> float:
    total = 0.0
    for i in range(n):
        total += i * i
    return total

# First call triggers compilation (~0.5s)
sum_squares_numba(10)

N = 1_000_000
t_py    = timeit.timeit(lambda: sum(i*i for i in range(N)), number=5)
t_numba = timeit.timeit(lambda: sum_squares_numba(N), number=5)
print(f"Python: {t_py:.3f}s   Numba: {t_numba:.5f}s   speedup: {t_py/t_numba:.0f}x")

# ── Parallel loops with prange ──
@njit(parallel=True)
def parallel_sum(arr: np.ndarray) -> float:
    total = 0.0
    for i in prange(len(arr)):   # prange → OpenMP parallel loop
        total += arr[i] * arr[i]
    return total

data = np.arange(10_000_000, dtype=np.float64)
print(parallel_sum(data))

# ── GPU acceleration with @cuda.jit ──
# from numba import cuda
# @cuda.jit
# def gpu_kernel(arr, out): ...   # runs on NVIDIA GPU

# ── What Numba can and cannot do ──
# ✅ Works with: NumPy arrays, numeric Python (int, float), loops, conditionals
# ❌ Fails with: arbitrary Python objects, dicts (mostly), strings, class instances

import dis, timeit

# ❌ Anti-pattern 1: Repeated attribute lookup in hot loops
class Processor:
    results = []
    def slow(self, data):
        for x in data:
            self.results.append(x)   # self.results AND .append looked up every iteration

    def fast(self, data):
        append = self.results.append  # cache the bound method
        for x in data:
            append(x)

# ❌ Anti-pattern 2: Building a string by concatenation
def slow_join(parts):
    result = ""
    for p in parts:
        result += p   # creates a NEW string object every iteration O(n²)
    return result

def fast_join(parts):
    return "".join(parts)   # single allocation

# ❌ Anti-pattern 3: Using a list where a deque is needed for popleft
from collections import deque
data = list(range(100_000))
dq   = deque(range(100_000))

t_list  = timeit.timeit(lambda: data.pop(0), number=1000)   # O(n) shift
t_deque = timeit.timeit(lambda: dq.popleft(), number=1000)  # O(1)
print(f"list.pop(0): {t_list:.4f}s  deque.popleft: {t_deque:.6f}s")

# ❌ Anti-pattern 4: Checking type with type() instead of isinstance()
def slow_check(x):
    return type(x) == int   # does not handle subclasses; slightly slower

def fast_check(x):
    return isinstance(x, int)   # handles subclasses; JIT-friendly

# ❌ Anti-pattern 5: Creating lambdas / partials in a tight loop
import functools
fn = functools.partial(pow, 2)   # create ONCE outside the loop
for _ in range(1_000_000):
    fn(10)   # reuse the same partial object