深入理解 Python GIL：从原理到实践

import sys

a = []
print(sys.getrefcount(a))   # 输出 2：a 本身持有 1 个引用，getrefcount 调用时临时持有 1 个

b = a
print(sys.getrefcount(a))   # 输出 3：b 也引用了同一个列表

del b
print(sys.getrefcount(a))   # 输出 2：b 被删除，引用数减 1

import threading
import time


def count_primes(n: int) -> int:
    """暴力判断 [2, n) 内有多少个质数，纯 Python 循环，受 GIL 严格限制。"""
    count = 0
    for num in range(2, n):
        is_prime = True
        for divisor in range(2, int(num**0.5) + 1):
            if num % divisor == 0:
                is_prime = False
                break
        if is_prime:
            count += 1
    return count


N = 50_000
REPEAT = 4

# 单线程：顺序执行 4 次
t0 = time.perf_counter()
for _ in range(REPEAT):
    count_primes(N)
single_time = time.perf_counter() - t0

# 多线程：4 个线程同时执行
threads = [threading.Thread(target=count_primes, args=(N,)) for _ in range(REPEAT)]
t0 = time.perf_counter()
for t in threads:
    t.start()
for t in threads:
    t.join()
multi_time = time.perf_counter() - t0

print(f"单线程耗时：{single_time:.2f}s")
print(f"多线程耗时：{multi_time:.2f}s（4 个线程）")
print(f"加速比：{single_time / multi_time:.2f}x（理想应为 4x）")

import urllib.request
import threading
import time


def fetch_url(url: str):
    """发起一次 HTTP GET 请求，等待返回。IO 等待期间 GIL 会被释放。"""
    with urllib.request.urlopen(url, timeout=10) as resp:
        resp.read()


URL = "https://httpbin.org/delay/1"
REPEAT = 4

# 单线程：顺序发 4 次请求，约 4 秒
t0 = time.perf_counter()
for _ in range(REPEAT):
    fetch_url(URL)
single_time = time.perf_counter() - t0

# 多线程：4 个线程并发发请求，约 1 秒
threads = [threading.Thread(target=fetch_url, args=(URL,)) for _ in range(REPEAT)]
t0 = time.perf_counter()
for t in threads:
    t.start()
for t in threads:
    t.join()
multi_time = time.perf_counter() - t0

print(f"单线程耗时：{single_time:.2f}s")
print(f"多线程耗时：{multi_time:.2f}s（4 个线程）")
print(f"加速比：{single_time / multi_time:.2f}x")

import numpy as np

# 陷阱：object dtype 数组不是真正的向量化
arr = np.array([1, "hello", 3.14], dtype=object)  # object dtype
result = np.vectorize(lambda x: str(x))(arr)       # 逐元素调用 Python 函数，受 GIL 限制

// NumPy 源码中典型的 GIL 释放模式
Py_BEGIN_ALLOW_THREADS   // 释放 GIL，其他 Python 线程可以运行
/* ... 纯 C 计算，不操作任何 Python 对象 ... */
Py_END_ALLOW_THREADS     // 重新获取 GIL

import threading
import time
import numpy as np

N = 5_000_000
THREADS = 4


def python_sum(arr: list) -> float:
    """用纯 Python 循环累加，每一步都在执行 Python 字节码，GIL 全程持有。"""
    total = 0.0
    for x in arr:
        total += x
    return total


def numpy_sum(arr: np.ndarray) -> float:
    """调用 np.sum()，计算在 C 层执行，进入时释放 GIL，多线程可并行。"""
    return np.sum(arr)


def run_single(func, arg):
    t0 = time.perf_counter()
    for _ in range(THREADS):
        func(arg)
    return time.perf_counter() - t0


def run_multi(func, arg):
    threads = [threading.Thread(target=func, args=(arg,)) for _ in range(THREADS)]
    t0 = time.perf_counter()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return time.perf_counter() - t0


py_data = list(range(N))
np_data = np.arange(N, dtype=np.float64)

py_single = run_single(python_sum, py_data)
py_multi  = run_multi(python_sum, py_data)

np_single = run_single(numpy_sum, np_data)
np_multi  = run_multi(numpy_sum, np_data)

import numpy as np

# ❌ 陷阱：np.vectorize 是 for 循环的语法糖，受 GIL 限制
result = np.vectorize(lambda x: x ** 2)(arr)

# ✅ 正确：直接用 NumPy 广播，在 C 层执行
result = arr ** 2

import threading
import time
import numpy as np

N = 5_000_000
THREADS = 4


def python_sum(arr: list) -> float:
    total = 0.0
    for x in arr:
        total += x
    return total


def numpy_sum(arr: np.ndarray) -> float:
    return np.sum(arr)


def run_single(func, arg):
    t0 = time.perf_counter()
    for _ in range(THREADS):
        func(arg)
    return time.perf_counter() - t0


def run_multi(func, arg):
    threads = [threading.Thread(target=func, args=(arg,)) for _ in range(THREADS)]
    t0 = time.perf_counter()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return time.perf_counter() - t0


if __name__ == "__main__":
    py_data = list(range(N))
    np_data = np.arange(N, dtype=np.float64)

    py_single = run_single(python_sum, py_data)
    py_multi  = run_multi(python_sum, py_data)
    print(f"Python 循环 - 单线程：{py_single:.2f}s，多线程：{py_multi:.2f}s，加速比：{py_single/py_multi:.2f}x")

    np_single = run_single(numpy_sum, np_data)
    np_multi  = run_multi(numpy_sum, np_data)
    print(f"NumPy sum  - 单线程：{np_single:.2f}s，多线程：{np_multi:.2f}s，加速比：{np_single/np_multi:.2f}x")

import threading
import multiprocessing
import time


def max_collatz_steps(limit: int) -> int:
    """计算 [1, limit) 中 Collatz 序列最长的步数。纯 Python 数值运算。"""
    def steps(n):
        count = 0
        while n != 1:
            n = n // 2 if n % 2 == 0 else 3 * n + 1
            count += 1
        return count
    return max(steps(n) for n in range(1, limit))


LIMIT = 200_000
WORKERS = 4


def run_single():
    t0 = time.perf_counter()
    for _ in range(WORKERS):
        max_collatz_steps(LIMIT)
    return time.perf_counter() - t0


def run_threads():
    threads = [threading.Thread(target=max_collatz_steps, args=(LIMIT,))
               for _ in range(WORKERS)]
    t0 = time.perf_counter()
    for t in threads: t.start()
    for t in threads: t.join()
    return time.perf_counter() - t0


def run_processes():
    with multiprocessing.Pool(processes=WORKERS) as pool:
        t0 = time.perf_counter()
        pool.starmap(max_collatz_steps, [(LIMIT,)] * WORKERS)
        elapsed = time.perf_counter() - t0
    return elapsed


if __name__ == "__main__":
    t_single  = run_single()
    t_thread  = run_threads()
    t_process = run_processes()

    print(f"单线程：{t_single:.2f}s")
    print(f"多线程：{t_thread:.2f}s，加速比：{t_single/t_thread:.2f}x")
    print(f"多进程：{t_process:.2f}s，加速比：{t_single/t_process:.2f}x")

import asyncio
import aiohttp
import time


async def fetch(session: aiohttp.ClientSession, url: str) -> int:
    """协程：发起 HTTP 请求，await 时主动释放控制权给事件循环。"""
    async with session.get(url) as resp:
        content = await resp.read()
        return len(content)


async def main():
    URL = "https://httpbin.org/delay/1"
    N   = 10   # 并发请求数

    async with aiohttp.ClientSession() as session:
        t0 = time.perf_counter()
        tasks = [fetch(session, URL) for _ in range(N)]
        results = await asyncio.gather(*tasks)   # 并发等待所有请求
        elapsed = time.perf_counter() - t0

    print(f"并发 {N} 个请求（每个约 1 秒）：耗时 {elapsed:.2f}s")
    print(f"加速比：{N / elapsed:.1f}x（理想为 {N}x）")


if __name__ == "__main__":
    asyncio.run(main())

import os, time, shutil, tempfile
import numpy as np
import torch
from torch.utils.data import Dataset, DataLoader


def prepare_dataset_on_disk(size, image_hw, data_dir):
    """将随机图像保存为独立 .npy 文件，模拟真实磁盘数据集。"""
    os.makedirs(data_dir, exist_ok=True)
    for i in range(size):
        img = np.random.randint(0, 256, (3, image_hw, image_hw), dtype=np.uint8)
        np.save(os.path.join(data_dir, f"{i:05d}.npy"), img)


class DiskImageDataset(Dataset):
    """
    每次 __getitem__ 从磁盘读取一个 .npy 文件，再做 CPU 预处理。
    这才是 num_workers 有效的前提：
      - 磁盘读取是 IO 操作（GIL 在系统调用时会释放）
      - numpy 解码 + 数据增强是 CPU 计算（多进程可并行）
    """
    def __init__(self, data_dir):
        self.files = sorted([
            os.path.join(data_dir, f)
            for f in os.listdir(data_dir) if f.endswith(".npy")
        ])

    def __len__(self):
        return len(self.files)

    def __getitem__(self, idx):
        img = np.load(self.files[idx]).astype(np.float32)   # 磁盘读取
        if np.random.rand() > 0.5:
            img = img[:, :, ::-1].copy()
        img = img / 255.0
        mean = np.array([0.485, 0.456, 0.406], dtype=np.float32).reshape(3, 1, 1)
        std  = np.array([0.229, 0.224, 0.225], dtype=np.float32).reshape(3, 1, 1)
        img = (img - mean) / std
        img = np.clip(img * 1.1 + np.random.normal(0, 0.01, img.shape).astype(np.float32), -3, 3)
        return torch.from_numpy(img.copy()), idx % 10


def benchmark_dataloader(num_workers, dataset, batch_size=32):
    loader = DataLoader(
        dataset, batch_size=batch_size, num_workers=num_workers,
        shuffle=True, pin_memory=False,
        prefetch_factor=2 if num_workers > 0 else None,
        persistent_workers=True if num_workers > 0 else False,
    )
    loader_iter = iter(loader)
    next(loader_iter)   # 预热：让 worker 进程完成 fork + import
    t0 = time.perf_counter()
    for batch_imgs, _ in loader_iter:
        _ = batch_imgs.mean()
    return time.perf_counter() - t0


if __name__ == "__main__":
    data_dir = os.path.join(tempfile.gettempdir(), "gil_fake_images")
    try:
        prepare_dataset_on_disk(1000, 128, data_dir)
        dataset = DiskImageDataset(data_dir)
        for nw in [0, 1, 2, 4]:
            t = benchmark_dataloader(nw, dataset)
            print(f"num_workers={nw}：{t:.2f}s")
    finally:
        shutil.rmtree(data_dir)

import time, threading, concurrent.futures
import numpy as np
import torch
import torch.nn as nn


class Encoder(nn.Module):
    def __init__(self, input_dim=2048, hidden_dim=2048, embed_dim=256):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(input_dim, hidden_dim), nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim), nn.ReLU(),
            nn.Linear(hidden_dim, embed_dim),
        )

    def forward(self, x):
        return self.net(x)


INPUT_DIM  = 2048    # 模拟 ResNet 最后一层特征展平
EMBED_DIM  = 256
N_SAMPLES  = 50_000  # 足够大，才能体现向量化优势
BATCH_SIZE = 256
WORKERS    = 4


def preprocess_one(raw: np.ndarray) -> torch.Tensor:
    """对单个样本做预处理：L2 归一化 + 随机 dropout + clip。"""
    feat = raw.astype(np.float32)
    norm = np.linalg.norm(feat) + 1e-8
    feat = feat / norm
    mask = (np.random.rand(len(feat)) > 0.1).astype(np.float32)
    feat = feat * mask
    feat = np.clip(feat, -5.0, 5.0)
    return torch.from_numpy(feat)


def extract_serial(model, raw_data):
    """串行逐样本预处理：Python for 循环，GIL 全程持有。"""
    model.eval()
    embeddings = []
    t0 = time.perf_counter()
    with torch.no_grad():
        for i in range(0, N_SAMPLES, BATCH_SIZE):
            chunk = raw_data[i:i+BATCH_SIZE]
            batch = torch.stack([preprocess_one(x) for x in chunk])
            embeddings.append(model(batch))
    return time.perf_counter() - t0, torch.cat(embeddings)


def extract_vectorized(model, raw_data):
    """
    NumPy 向量化批量预处理：无任何 Python 循环。
    将 preprocess_one 的所有操作改写为对整个矩阵的广播运算。
    """
    model.eval()
    t0 = time.perf_counter()

    feat = raw_data.astype(np.float32)

    # L2 归一化（axis=1 广播，操作整个矩阵）
    norms = np.linalg.norm(feat, axis=1, keepdims=True) + 1e-8
    feat = feat / norms

    # 向量化 feature dropout（一次生成整个 mask 矩阵）
    mask = (np.random.rand(N_SAMPLES, INPUT_DIM) > 0.1).astype(np.float32)
    feat = feat * mask

    feat = np.clip(feat, -5.0, 5.0)

    all_tensors = torch.from_numpy(feat)
    embeddings = []
    with torch.no_grad():
        for i in range(0, N_SAMPLES, BATCH_SIZE):
            embeddings.append(model(all_tensors[i:i+BATCH_SIZE]))

    return time.perf_counter() - t0, torch.cat(embeddings)

import time
import threading
import numpy as np
from joblib import Parallel, delayed

N_STOCKS = 500
N_DAYS   = 252
WORKERS  = 4


def generate_price_data(n_stocks=N_STOCKS, n_days=N_DAYS) -> np.ndarray:
    returns = np.random.normal(0.0005, 0.02, (n_stocks, n_days))
    return 100 * np.exp(np.cumsum(returns, axis=1))


def compute_factors_for_one_stock(prices: np.ndarray) -> dict:
    """纯 Python 循环实现 5 个技术因子（动量、波动率、RSI 等）。"""
    n = len(prices)
    mom_20 = [prices[i] / prices[i-20] - 1 for i in range(20, n)]
    mom_60 = [prices[i] / prices[i-60] - 1 for i in range(60, n)]
    returns = [prices[i] / prices[i-1] - 1 for i in range(1, n)]
    vol_20 = []
    for i in range(20, len(returns)):
        w = returns[i-20:i]
        mean = sum(w) / 20
        vol_20.append((sum((x-mean)**2 for x in w) / 20) ** 0.5)
    return {
        "mom_20": np.mean(mom_20) if mom_20 else 0,
        "mom_60": np.mean(mom_60) if mom_60 else 0,
        "vol_20": np.mean(vol_20) if vol_20 else 0,
    }


def run_vectorized(prices: np.ndarray) -> float:
    """NumPy/Pandas 矩阵化计算所有股票的所有因子。"""
    import pandas as pd
    t0 = time.perf_counter()
    returns = prices[:, 1:] / prices[:, :-1] - 1
    mom_20 = (prices[:, 20:] / prices[:, :-20] - 1).mean(axis=1)
    mom_60 = (prices[:, 60:] / prices[:, :-60] - 1).mean(axis=1)
    df_returns = pd.DataFrame(returns.T)
    vol_20 = df_returns.rolling(20).std().mean().values
    return time.perf_counter() - t0


if __name__ == "__main__":
    prices = generate_price_data()

    t1 = time.perf_counter()
    [compute_factors_for_one_stock(prices[i]) for i in range(N_STOCKS)]
    t1 = time.perf_counter() - t1

    t2 = time.perf_counter()
    results = [None] * N_STOCKS
    def worker(i): results[i] = compute_factors_for_one_stock(prices[i])
    threads = [threading.Thread(target=worker, args=(i,)) for i in range(N_STOCKS)]
    for t in threads: t.start()
    for t in threads: t.join()
    t2 = time.perf_counter() - t2

    t3 = time.perf_counter()
    Parallel(n_jobs=WORKERS)(delayed(compute_factors_for_one_stock)(prices[i]) for i in range(N_STOCKS))
    t3 = time.perf_counter() - t3

    t4 = run_vectorized(prices)
    print(f"串行：{t1:.2f}s | 多线程：{t2:.2f}s | joblib：{t3:.2f}s | 向量化：{t4:.2f}s")

import time
import math
import concurrent.futures
import numpy as np

S0, K, T, r, sigma = 100.0, 105.0, 1.0, 0.05, 0.2
N, M, WORKERS = 252, 100_000, 4


def simulate_one_path_python(seed: int) -> float:
    rng = np.random.default_rng(seed)
    dt = T / N
    S = S0
    for _ in range(N):
        z = rng.standard_normal()
        S *= math.exp((r - 0.5 * sigma**2) * dt + sigma * math.sqrt(dt) * z)
    return S


def run_serial():
    t0 = time.perf_counter()
    payoffs = [max(simulate_one_path_python(i) - K, 0) for i in range(M)]
    price = math.exp(-r * T) * sum(payoffs) / M
    return time.perf_counter() - t0, price


def run_vectorized():
    """精确终价公式：S_T = S0 * exp((r - 0.5σ²)T + σ√T·Z)，Z~N(0,1)"""
    t0 = time.perf_counter()
    Z = np.random.standard_normal(M)
    S_T = S0 * np.exp((r - 0.5 * sigma**2) * T + sigma * math.sqrt(T) * Z)
    payoffs = np.maximum(S_T - K, 0.0)
    price = math.exp(-r * T) * payoffs.mean()
    return time.perf_counter() - t0, price

import time, threading, concurrent.futures
import numpy as np
import pandas as pd

N_DAYS   = 5000
N_STOCKS = 100
THREADS  = 4


def event_driven_backtest(df: pd.DataFrame) -> dict:
    """逐根 K 线执行双均线策略，纯 Python 循环。"""
    close = df["close"].values
    n = len(close)
    cash, shares, nav = 10000.0, 0, []

    for i in range(20, n):
        ma5  = sum(close[i-5:i])  / 5    # Python sum()，每步都是字节码
        ma20 = sum(close[i-20:i]) / 20
        prev_ma5  = sum(close[i-6:i-1])  / 5
        prev_ma20 = sum(close[i-21:i-1]) / 20
        price = close[i]
        if ma5 > ma20 and prev_ma5 <= prev_ma20 and cash >= price:
            n_buy = int(cash / price)
            shares += n_buy; cash -= n_buy * price
        elif ma5 < ma20 and prev_ma5 >= prev_ma20 and shares > 0:
            cash += shares * price; shares = 0
        nav.append(cash + shares * price)

    return {"total_return": nav[-1] / 10000.0 - 1 if nav else 0}


def vectorized_backtest(df: pd.DataFrame) -> dict:
    """用 Pandas rolling 计算均线，NumPy 向量化生成信号。"""
    close = df["close"]
    ma5, ma20 = close.rolling(5).mean(), close.rolling(20).mean()
    cross_up   = (ma5 > ma20) & (ma5.shift(1) <= ma20.shift(1))
    cross_down = (ma5 < ma20) & (ma5.shift(1) >= ma20.shift(1))
    signal = pd.Series(np.nan, index=df.index)
    signal[cross_up] = 1; signal[cross_down] = 0
    signal = signal.ffill().fillna(0)
    strategy_return = (signal.shift(1) * close.pct_change()).fillna(0)
    total_return = (1 + strategy_return).prod() - 1
    return {"total_return": float(total_return)}


def run_all_serial(backtest_fn, all_data):
    t0 = time.perf_counter()
    results = [backtest_fn(df) for df in all_data]
    return time.perf_counter() - t0, results


def run_all_threaded(backtest_fn, all_data):
    """使用 ThreadPoolExecutor（THREADS 个 worker）并行回测所有股票。"""
    t0 = time.perf_counter()
    with concurrent.futures.ThreadPoolExecutor(max_workers=THREADS) as executor:
        results = list(executor.map(backtest_fn, all_data))
    return time.perf_counter() - t0, results

from numba import jit

# ❌ 不释放 GIL：单线程快，多线程仍串行
@jit(nopython=True, nogil=False)
def ema_numba(close, alpha): ...

# ✅ 释放 GIL：单线程一样快，多线程真正并行
@jit(nopython=True, nogil=True)
def ema_numba_nogil(close, alpha): ...

import threading
import time
import numpy as np
from numba import jit

N = 1_000_000
PERIOD = 20
THREADS = 4
alpha = 2.0 / (PERIOD + 1)


def ema_python(close: np.ndarray, alpha: float) -> np.ndarray:
    """纯 Python 递推，每步都是 Python 字节码，受 GIL 严格限制。"""
    result = np.empty(len(close))
    result[0] = close[0]
    for i in range(1, len(close)):
        result[i] = alpha * close[i] + (1 - alpha) * result[i - 1]
    return result


@jit(nopython=True, nogil=True)
def ema_numba_nogil(close: np.ndarray, alpha: float) -> np.ndarray:
    """
    nogil=True：函数执行期间释放 GIL。
    nopython=True 确保函数内部不操作任何 Python 对象，释放 GIL 才安全。
    多个线程可以真正并行地各自执行此函数。
    """
    result = np.empty(len(close))
    result[0] = close[0]
    for i in range(1, len(close)):
        result[i] = alpha * close[i] + (1 - alpha) * result[i - 1]
    return result


def run_single(func, data, repeat=THREADS):
    t0 = time.perf_counter()
    for _ in range(repeat):
        func(data, alpha)
    return time.perf_counter() - t0


def run_multi(func, data, n_threads=THREADS):
    threads = [threading.Thread(target=func, args=(data, alpha))
               for _ in range(n_threads)]
    t0 = time.perf_counter()
    for t in threads: t.start()
    for t in threads: t.join()
    return time.perf_counter() - t0


if __name__ == "__main__":
    close = np.random.rand(N).astype(np.float64) * 100 + 50

    # 预热：触发 JIT 编译（首次调用有延迟，不计入计时）
    ema_numba_nogil(close[:100], alpha)

    t_py_s = run_single(ema_python, close)
    t_nb_s = run_single(ema_numba_nogil, close)
    t_nb_m = run_multi(ema_numba_nogil, close)

    print(f"纯 Python 单线程：{t_py_s:.3f}s")
    print(f"Numba nogil 单线程：{t_nb_s:.3f}s（vs Python: {t_py_s/t_nb_s:.1f}x）")
    print(f"Numba nogil 多线程：{t_nb_m:.3f}s（多线程加速比: {t_nb_s*THREADS/t_nb_m:.2f}x）")

# ema_cython.pyx
import numpy as np
cimport numpy as np


def fast_ema_nogil(np.ndarray[double, ndim=1] close, int period):
    """
    在 with nogil: 块中执行递推，显式释放 GIL。
    多线程场景下，多个线程可以同时执行此函数。

    注意：with nogil 块内不能有任何 Python 对象操作，
    否则 Cython 编译时会报错，这是一个编译期安全检查。
    """
    cdef double alpha = 2.0 / (period + 1)
    cdef double[:] result = np.empty(len(close), dtype=np.float64)
    cdef double[:] c = close   # 先拿到 memoryview，避免 nogil 块内访问 Python 对象
    cdef int i
    cdef int n = len(close)

    result[0] = c[0]

    # with nogil 块：释放 GIL，允许其他 Python 线程运行
    # 块内只能操作 C 变量和 memoryview，不能有任何 Python 对象
    with nogil:
        for i in range(1, n):
            result[i] = alpha * c[i] + (1 - alpha) * result[i - 1]

    return np.asarray(result)

# 编译：python setup_cython.py build_ext --inplace
from setuptools import setup
from Cython.Build import cythonize
import numpy as np

setup(
    ext_modules=cythonize(
        "ema_cython.pyx",
        compiler_directives={
            "language_level": "3",
            "boundscheck": False,   # 关闭越界检查，提升性能（生产慎用）
            "wraparound": False,    # 关闭负索引支持
        },
    ),
    include_dirs=[np.get_include()],
)

#include <pybind11/pybind11.h>
#include <pybind11/numpy.h>

namespace py = pybind11;

py::array_t<double> fast_ema_nogil(py::array_t<double> close, int period) {
    auto buf = close.request();
    double* ptr = static_cast<double*>(buf.ptr);
    int n = static_cast<int>(buf.shape[0]);
    double alpha = 2.0 / (period + 1);

    py::array_t<double> result(n);
    auto res_buf = result.request();
    double* res = static_cast<double*>(res_buf.ptr);

    // 在释放 GIL 之前，确保所有 Python 对象操作已完成
    // gil_scoped_release：RAII 风格，构造时释放 GIL，析构时重新获取
    {
        py::gil_scoped_release release;

        // GIL 已释放：这里只操作原始 C 指针，不涉及任何 Python 对象
        // 多个线程可以同时执行到这里，真正并行
        res[0] = ptr[0];
        for (int i = 1; i < n; ++i) {
            res[i] = alpha * ptr[i] + (1.0 - alpha) * res[i - 1];
        }

    }   // GIL 在这里自动重新获取

    return result;
}

PYBIND11_MODULE(ema_cpp, m) {
    m.def("fast_ema_nogil", &fast_ema_nogil,
          py::arg("close"), py::arg("period"),
          "EMA 递推（释放 GIL，多线程可并行）");
}

import threading, time
import numpy as np
import ema_cpp   # 编译自 ema_cpp.cpp

N = 1_000_000
PERIOD = 20
THREADS = 4


def run_multi(func, data, period, n_threads=THREADS):
    threads = [threading.Thread(target=func, args=(data, period))
               for _ in range(n_threads)]
    t0 = time.perf_counter()
    for t in threads: t.start()
    for t in threads: t.join()
    return time.perf_counter() - t0


close = np.random.rand(N).astype(np.float64) * 100 + 50

t_pb_m = run_multi(ema_cpp.fast_ema_nogil, close, PERIOD)
print(f"pybind11 nogil 多线程（{THREADS} 线程）：{t_pb_m:.3f}s")

import time, threading
import numpy as np
from numba import jit
import ema_cython, ema_cpp

N, PERIOD, THREADS = 1_000_000, 20, 4


@jit(nopython=True, nogil=True)
def ema_numba(close, period):
    a = 2.0 / (period + 1)
    result = np.empty(len(close))
    result[0] = close[0]
    for i in range(1, len(close)):
        result[i] = a * close[i] + (1 - a) * result[i - 1]
    return result


def ema_python(close, period):
    a = 2.0 / (period + 1)
    result = np.empty(len(close))
    result[0] = close[0]
    for i in range(1, len(close)):
        result[i] = a * close[i] + (1 - a) * result[i - 1]
    return result


def bench(func, data, period, repeat=5):
    """多次运行取最小值，减少系统噪声。"""
    return min(
        (lambda: (t := time.perf_counter(), func(data, period), time.perf_counter() - t)[2])()
        for _ in range(repeat)
    )


def bench_multi(func, data, period, n_threads=THREADS):
    threads = [threading.Thread(target=func, args=(data, period))
               for _ in range(n_threads)]
    t0 = time.perf_counter()
    for t in threads: t.start()
    for t in threads: t.join()
    return time.perf_counter() - t0

# 检查当前 Python 是否为 no-GIL 构建
python -c "import sys; print(sys._is_gil_enabled())"
# 有 GIL 时输出 True，no-GIL 构建输出 False