GIL与并行计算

概述

本章讲解GIL（全局解释器锁）机制和Cython的并行计算支持。GIL限制了Python多线程的CPU并行，但Cython可通过nogil和prange释放性能。

学习路径：GIL原理 → nogil声明 → OpenMP并行 → 多线程模式 → 多进程选择

核心概念：

• nogil：允许C级代码在无GIL状态下执行
• prange：并行化循环，自动处理约简
• 线程同步：互斥锁和约简变量

---

9.1 全局解释器锁（GIL）

GIL工作原理

功能说明：Python GIL限制同一时刻只有一个线程执行Python字节码。

# Python的GIL - 同时只有一个线程执行Python字节码
import threading

def cpu_task(n):
    result = 0
    for i in range(n):
        result += i
    return result

# 两个线程实际上无法并行执行CPU密集型任务
t1 = threading.Thread(target=cpu_task, args=(10000000,))
t2 = threading.Thread(target=cpu_task, args=(10000000,))
# 实际耗时 ≈ t1单独 + t2单独（而非并行）

GIL对性能的影响

功能说明：无类型声明的Cython代码仍受GIL限制。

# 无类型声明的代码仍受GIL限制
def slow_sum(n):
    total = 0
    for i in range(n):
        total += i
    return total
# 这个函数受GIL限制，无法真正并行

何时可以释放GIL

功能说明：纯C代码可以安全释放GIL，Python对象操作需要GIL。

# C级代码可以释放GIL
cdef void c_function() nogil:
    # 这段代码不需要Python对象操作
    # 可以安全地释放GIL
    pass

# Python对象操作需要GIL
cdef void python_operation() with gil:
    # 获取GIL后才能执行
    import os
    print(os.getcwd())

---

9.2 nogil声明

nogil语法

功能说明：声明函数在无GIL状态下执行，允许并行。

# 声明无GIL的函数
cdef void compute(int n) nogil:
    cdef int i
    cdef double total = 0.0
    for i in range(n):
        total += i * i
    # 注意：不能在此访问Python对象

# 调用
cdef void wrapper():
    cdef int n = 1000000
    with nogil:
        compute(n)

with gil块

功能说明：在nogil代码块中需要GIL时使用with gil。

# 获取GIL
cdef void need_gil() with gil:
    # 此时GIL被当前线程持有
    # 可以安全操作Python对象
    print("Hello from Python")
    result = [1, 2, 3]
    print(sum(result))

状态切换开销

功能说明：GIL切换有成本，频繁切换降低性能。

# GIL切换开销约50-100ns
# 频繁切换会降低性能

# 不好：频繁切换
cdef void bad_pattern(int n):
    cdef int i
    for i in range(n):
        with nogil:
            compute_single(i)  # 每元素切换一次
        # 需要Python对象操作
        save_result(i)

# 好：批量处理
cdef void good_pattern(int n):
    cdef int i
    # 大批量nogil计算
    with nogil:
        for i in range(n):
            compute_single(i)
    # 完成后统一处理结果
    for i in range(n):
        save_result(i)

最佳实践：最小化GIL切换次数，批量处理后统一交互。

---

9.3 OpenMP并行

prange并行循环

功能说明：prange自动并行化循环，配合reduction处理竞态。

# setup.py配置OpenMP
from setuptools import Extension
ext = Extension("parallel", ["parallel.pyx"],
                extra_compile_args=["-fopenmp"],
                extra_link_args=["-fopenmp"])

# Cython代码
from cython.parallel import prange

# 错误示例：竞态条件
cdef int sum_parallel(int n):
    cdef int total = 0
    cdef int i
    for i in prange(n, nogil=True):
        total += i  # 竞态条件！
    return total

# 正确用法：使用reduction
cdef int sum_with_reduction(int n):
    cdef int total = 0
    cdef int i
    for i in prange(n, nogil=True, reduction=total):
        total += i
    return total

输出示例：

>>> sum_with_reduction(1000000)
499999500000

并行化策略

功能说明：并行化无依赖的循环，实现线性加速。

# 简单并行
from cython.parallel import prange

cdef void parallel_square(double[:] arr) nogil:
    cdef int i
    cdef int n = arr.shape[0]
    for i in prange(n):
        arr[i] = arr[i] * arr[i]

# 指定线程数
cdef void parallel_process(double[:] arr, int num_threads) nogil:
    cdef int i
    for i in prange(arr.shape[0], num_threads=num_threads):
        arr[i] = process_single(arr[i])

约简变量

功能说明：reduction自动合并多个线程的结果。

# reduction示例：点积
cdef double dot_product(double[:] a, double[:] b) nogil:
    cdef double result = 0.0
    cdef int i
    cdef int n = a.shape[0]
    for i in prange(n, reduction=result):
        result += a[i] * b[i]
    return result

# 多个约简
cdef struct Result:
    double sum
    double sum_sq

cdef Result stats(double[:] arr) nogil:
    cdef Result r
    cdef int i
    r.sum = 0.0
    r.sum_sq = 0.0
    for i in prange(arr.shape[0], reduction=(r.sum, r.sum_sq)):
        r.sum += arr[i]
        r.sum_sq += arr[i] * arr[i]
    return r

输出示例：

>>> arr = np.array([1.0, 2.0, 3.0, 4.0])
>>> dot_product(arr, arr)
30.0

线程同步

功能说明：使用OpenMP互斥锁保护共享资源。

# 互斥锁
from cython.parallel import prange
cimport openmp

cdef int counter = 0
cdef openmp.omp_lock_t lock

openmp.omp_init_lock(&lock)

cdef void increment_many(int n):
    cdef int i
    for i in prange(n):
        openmp.omp_set_lock(&lock)
        counter += 1
        openmp.omp_unset_lock(&lock)

openmp.omp_destroy_lock(&lock)

常见坑：锁竞争严重时并行效率下降，尽量用reduction替代。

---

9.4 多线程模式

工作池模式

功能说明：预创建线程池，复用线程减少开销。

from cython.parallel import threadid
cimport openmp

cdef void worker(int worker_id, double[:] data) nogil:
    cdef int i
    cdef int n = data.shape[0]
    for i in range(n):
        if i % openmp.omp_get_max_threads() == worker_id:
            data[i] = process(data[i])

cdef void process_pool(double[:] arr, int num_threads):
    cdef int i
    for i in prange(num_threads, num_threads=num_threads):
        worker(i, arr)

生产者-消费者

功能说明：环形缓冲区实现生产者-消费者模式。

# 环形缓冲区示例
cdef int buffer_size = 100
cdef double[100] buffer
cdef int producer_idx = 0
cdef int consumer_idx = 0
cdef int count = 0

cdef void produce(double value) nogil:
    global producer_idx, count
    buffer[producer_idx] = value
    producer_idx = (producer_idx + 1) % buffer_size
    count += 1

cdef double consume() nogil:
    global consumer_idx, count
    if count == 0:
        return 0.0
    value = buffer[consumer_idx]
    consumer_idx = (consumer_idx + 1) % buffer_size
    count -= 1
    return value

---

9.5 多进程vs多线程

适用场景

场景	推荐方案	原因
CPU密集型	多进程	绕过GIL，每个进程独立
I/O密集型	多线程	共享内存简单
大内存数据	多进程	避免GIL锁定

进程间通信

功能说明：使用multiprocessing实现真正的并行。

# 使用multiprocessing
from multiprocessing import Process, Queue

def worker(queue):
    while True:
        data = queue.get()
        if data is None:
            break
        result = compute(data)
        queue.put(result)

# Cython中使用
cpdef void run_parallel(list tasks):
    cdef int i
    queue = Queue()
    processes = []
    for i in range(4):
        p = Process(target=worker, args=(queue,))
        p.start()
        processes.append(p)

    for task in tasks:
        queue.put(task)

    for i in range(4):
        queue.put(None)

    for p in processes:
        p.join()

共享内存

功能说明：multiprocessing.shared_memory实现进程间共享数据。

# 使用共享内存
from multiprocessing import Process
from multiprocessing.shared_memory import SharedMemory

# 创建共享内存
shm = SharedMemory(name='my_shm', create=True, size=1000)

# 连接共享内存
shm2 = SharedMemory(name='my_shm')

---

本章小结

GIL技术对比

技术	GIL	适用场景
纯C代码	nogil	CPU密集计算
prange	nogil	并行循环
Python线程	受限	I/O操作
multiprocessing	无GIL	多核并行

并行化原则

1. 用prange替代range实现循环并行
2. 用reduction处理竞态条件
3. 避免锁竞争，尽量用约简
4. 批量切换GIL，最小化开销

性能参考

方式	4核加速比
单线程	1x
prange无约简	~1x（竞态）
prange+reduction	~3.5x
multiprocessing	~3.8x

---

练习

1. 比较单线程和多线程prange的性能差异
2. 实现并行矩阵乘法（分块）
3. 使用reduction计算数组统计量（sum, sum_sq, max, min）
4. 实现工作池模式处理批量任务
5. 比较multiprocessing和prange的性能差异
6. 测量GIL切换开销（带nogil vs 不带）