并发与并行

并发与并行

并发（concurrency）与并行（parallelism）研究的是程序如何同时做多件事。本文件涵盖并发与并行的区别、同步原语、经典并发问题、死锁、无锁数据结构、并行编程模型、异步编程以及扩展定律——这些概念是支撑多线程服务器、分布式训练以及每一个现代应用的基础。

• 单个 CPU 核一次只执行一条指令。但现代系统有 8、64 甚至成千上万个核（GPU）。即便在单核上，我们也想同时处理多个任务：一边下载文件、一边渲染界面、一边处理用户输入。并发和并行是管理多种活动的两种策略。

并发 vs 并行

• **并发（concurrency）**关注如何管理多个任务。任务通过交错推进：任务 A 跑一会儿，然后任务 B，再回到 A。在单核上，并发制造出同时执行的错觉。这些任务并不是真正同时进行的；它们轮流来。

• **并行（parallelism）**关注如何同时执行多个任务。有 n 个核，就能真正同时跑 n 个任务。并行需要多个硬件执行单元。

• 一个类比：并发是一个厨师在切菜和搅锅之间来回切换；并行是两个厨师，各干一件事。一个系统可以并发但不并行（单核、交错任务），可以并行但不并发（多核各自跑互不相干的程序），也可以两者兼具（多核跑着彼此交互的交错任务）。

• 在 ML 中，并发出现在数据加载上（把数据预处理与 GPU 计算重叠起来），而并行出现在分布式训练上（多个 GPU 同时计算梯度，第 6 章）。

同步原语

• 当多个线程共享数据时，**同步（synchronisation）**用来防止竞态条件。竞态条件是指结果取决于不可预测的线程执行顺序。

• 考虑两个线程都对一个共享计数器做自增：counter += 1。这其实是三步操作：（1）读 counter，（2）加 1，（3）写 counter。如果两个线程都读到同一个值（比如 5），都加 1，都写回 6，那么计数器最终是 6 而不是正确的 7。一次自增丢了。

• **互斥锁（mutex，mutual exclusion lock）保证同一时刻只有一个线程进入临界区。线程在进入临界区前获取（acquire）锁，退出后释放（release）**锁。其它任何试图获取已被持有锁的线程都会阻塞，直到锁被释放。

lock.acquire()
counter += 1      # 同一时刻只有一个线程在这里
lock.release()
​

• 互斥锁是正确的，但会引入争用（contention）：如果很多线程竞争同一把锁，它们就把时间花在等待上而不是计算上。这限制了可扩展性。极端情形下，所有线程都想要同一把锁，整个程序就被串行化了。

• **信号量（semaphore）**是互斥锁的推广。一个计数信号量维护一个计数器：wait() 把计数器减一（如果会变负就阻塞），signal() 把它加一。初始化为 1 的信号量行为像互斥锁。初始化为 n 的信号量允许至多 n 个线程同时进入临界区（对数据库连接这类资源池很有用）。

• **条件变量（condition variable）**让一个线程可以等到某个特定条件满足。线程释放锁、在条件变量上等待，当另一个线程通知该条件时被唤醒。这避免了忙等（busy-waiting，在循环里反复检查某个条件，浪费 CPU）。

• **管程（monitor）**把互斥锁、条件变量和共享数据打包成一个抽象。Java 的 synchronized 关键字和 Python 的 threading.Condition 实现了类似管程的语义。

• **读写锁（read-write locks）**区分读者（可以共享访问，因为读不改数据）和写者（需要独占访问）。多个读者可以同时持有锁，但一个写者会阻塞所有读者和其它写者。当读远多于写时（比如一个提供预测服务的缓存模型），这种锁是最优的。

经典并发问题

• 生产者-消费者（producer-consumer，有界缓冲）：生产者生成物品放进一个固定大小的缓冲区；消费者取出物品。挑战在于：缓冲区满时生产者要等，空时消费者要等，而且双方都要避免损坏缓冲区。

• 解决方案用两个信号量（一个数空槽，一个数满槽）加上一个保护缓冲区本身的互斥锁。这是大多数消息队列、日志系统和数据管道背后的模式。

• 读者-写者（readers-writers）：多个读者可以同时读，但写者需要独占访问。挑战在于公平性：如果读者源源不断到来，写者可能饥饿（永远得不到访问）。解决方案要么优先读者、要么优先写者，要么公平地交替。

• 哲学家就餐（dining philosophers）：五位哲学家围坐一桌，他们之间放着五把叉子。每个人需要两把叉子才能吃。如果五个人同时拿起左手的叉子，没人能拿到右手的叉子，于是所有人都饿死（死锁）。解决方案包括：原子地同时拿起两把叉子、引入不对称（一位哲学家先拿右手的叉子）、或用一个服务员（用信号量限制就餐人数不超过 4）。

死锁

• **死锁（deadlock）**发生在一组线程中，每个线程都在等另一个线程持有的资源，形成依赖环。谁都没法继续。

• 死锁的四个必要条件（必须同时成立）：

  1. 互斥（mutual exclusion）：资源一次只能被一个线程持有。
  2. 持有并等待（hold and wait）：线程在持有一个资源的同时等待另一个。
  3. 不可抢占（no preemption）：资源不能被强行从一个线程手中夺走。
  4. 循环等待（circular wait）：等待图中存在一个环。

• **死锁预防（deadlock prevention）**打破四个条件之一：

  • 消除循环等待：对资源施加一个全序。所有线程按相同顺序获取资源。如果每个线程总是先获取锁 A 再获取锁 B，环就不可能形成。
  • 消除持有并等待：要求线程一次性（原子地）申请所有资源。

• **死锁避免（deadlock avoidance）**动态决定是否授予某个资源请求会不会导致死锁。**银行家算法（Banker's algorithm）**维护每个线程的最大可能需求，只批准那些能让系统留在"安全状态"（所有线程最终都能完成的状态）的请求。该算法每次请求的复杂度是 O(n^2 m)（n 个线程，m 种资源类型），对大多数真实系统而言太贵。

• **死锁检测（deadlock detection）**让死锁发生，然后检测（在等待图中找环）并恢复（杀掉一个线程或回滚一个事务）。

• 在实践中，大多数系统对常见情形用预防（资源排序），对罕见情形用检测。数据库系统是经典例子：它检测事务之间的死锁，并中止其中一个以打破环。

无锁与无等待数据结构

• 锁会引入争用、优先级反转和死锁风险。无锁（lock-free）数据结构完全不用锁，而是借助硬件提供的原子操作（atomic operations）。

• 关键的原子操作是比较并交换（Compare-And-Swap，CAS）：原子地检查某个内存位置是否持有期望的值，若是则把它替换成新值。伪代码如下：

CAS(address, expected, new_value):
    if *address == expected:
        *address = new_value
        return true
    else:
        return false
​

• CAS 由单条硬件指令实现，所以即使没有锁它也是原子的。无锁算法把 CAS 用在重试循环里：读当前值、计算新值、尝试 CAS。如果在此期间另一个线程改了这个值，CAS 失败，线程重试。

• 无锁（lock-free）：至少有一个线程能在有限步内推进（不会死锁，但在争用下个别线程可能无限重试）。

• 无等待（wait-free）：每个线程都能在有界步数内推进（最强保证，但最难实现）。

• 无锁的栈、队列和哈希表被广泛用于高性能系统。Java 的 ConcurrentHashMap 和 Go 的原子操作都建立在 CAS 之上。

并行编程模型

• **共享内存（shared memory）**并行：所有线程访问同一个内存空间。同步由程序员负责。OpenMP 提供编译制导指令来并行化循环：

#pragma omp parallel for
for (int i = 0; i < n; i++) {
    result[i] = compute(data[i]);
}
​

• 编译器把循环的迭代分摊到可用的核上。OpenMP 对数据并行工作负载（对许多数据点做同样操作）很有效，广泛用于科学计算。

• **消息传递（message passing）**并行：每个进程有自己的内存。通信通过发送和接收消息完成。MPI（消息传递接口，Message Passing Interface）是跨节点分布式计算的标准：

MPI_Send(data, count, MPI_FLOAT, dest, tag, MPI_COMM_WORLD);
MPI_Recv(data, count, MPI_FLOAT, src, tag, MPI_COMM_WORLD, &status);
​

• MPI 能扩展到数千个节点，因为没有需要同步的共享状态。分布式深度学习（第 6 章）使用像 MPI_AllReduce（环形 all-reduce）这样的集合操作来在 GPU 之间同步梯度。

• GPU 并行遵循 SIMT（单指令多线程，Single Instruction, Multiple Threads）模型：成千上万个线程在不同数据上执行同一条指令。这非常适合矩阵运算（第 2 章），同样的乘加被应用到每个元素上。我们会在后续章节详细讲 GPU 编程。

异步与事件驱动编程

• 并非所有并发都需要线程。**异步（asynchronous）编程用一个事件循环（event loop）**在单线程里处理大量 I/O 密集型任务。

• 事件循环维护一个任务队列。当某个任务需要等 I/O（网络响应、读文件）时，它注册一个回调并交出控制权。事件循环去取下一个就绪的任务。当 I/O 完成时，回调被入队并最终执行。等待期间没有任何线程被阻塞。

• **协程（coroutines）**是可以挂起和恢复的函数。async/await 语法（Python、JavaScript、Rust）让协程看起来就像普通的顺序代码：

async def fetch_data(url):
    response = await http_get(url)  # 在此处挂起，事件循环去跑其它任务
    return process(response)         # 等到响应到来时恢复
​

• await 关键字挂起协程，把控制权交还给事件循环。当被等待的操作完成时，协程从它停下的地方恢复。这是协作式多任务：协程主动让出，不像抢占式多任务那样由 OS 强制切换线程。

• 异步非常适合有大量并发连接的 **I/O 密集型（I/O-bound）**工作负载（处理数千个客户端的 Web 服务器）。它不适合 **CPU 密集型（CPU-bound）**工作（单线程事件循环无法利用多核）。对 CPU 密集型工作，应当用线程或进程。

• Python 的**全局解释器锁（Global Interpreter Lock，GIL）**让线程无法实现真正的并行：同一时刻只能有一个线程执行 Python 字节码。这也是为什么 Python 用多进程（各自独立的进程，每个有自己的解释器）来实现 CPU 并行，用异步来实现 I/O 并发。GIL 正在 Python 3.13+ 中被移除（自由线程 Python），届时将启用真正的多线程并行。

扩展定律

• **阿姆达尔定律（Amdahl's law）**描述并行化一个程序所带来的理论加速比。如果程序中有 p 的比例可并行，其余 1 - p 是串行的：

• 其中 n 是处理器数。当 n \to \infty 时，最大加速比趋近于 \frac{1}{1-p}。如果程序有 95% 可并行，最大加速比就是 \frac{1}{0.05} = 20\times，无论你加多少核。串行部分就是瓶颈。

• 这对 ML 有深远影响：如果数据加载占训练时间的 10% 而且是串行的，那么加更多 GPU 最多只能把训练加速 10 倍。那 10% 的串行瓶颈限制了一切（这也是为什么高效的数据管道、把计算与 I/O 重叠起来很重要，第 6 章）。

• **古斯塔夫森定律（Gustafson's law）**提供了一个更乐观的视角。它不再固定问题规模去加处理器，而是固定总时间，问能多做多少工作。如果并行部分随问题规模而增长：

• 这对 n 是线性的。论点是：有了更多处理器，我们去求解更大的问题，而不是更快地求解同一个问题。在 ML 中，这对应于随 GPU 增多而增大 batch size（弱扩展），而不是保持 batch size 不变（强扩展）。

编程练习（使用 CoLab 或 notebook）

1. 演示竞态条件。两个线程不加同步地对一个共享计数器自增，观察丢失的更新。

import threading

counter = 0

def increment(n):
    global counter
    for _ in range(n):
        counter += 1  # 非原子操作：读、加、写

threads = [threading.Thread(target=increment, args=(100000,)) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()

print(f"Expected: {4 * 100000}")
print(f"Actual:   {counter}")
print(f"Lost updates: {4 * 100000 - counter}")
​

2. 用锁修复竞态条件，并测量开销。

import threading
import time

lock = threading.Lock()
counter = 0

def increment_locked(n):
    global counter
    for _ in range(n):
        with lock:
            counter += 1

start = time.time()
threads = [threading.Thread(target=increment_locked, args=(100000,)) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
elapsed = time.time() - start

print(f"Counter: {counter} (correct: {4 * 100000})")
print(f"Time with lock: {elapsed:.3f}s")
​

3. 可视化阿姆达尔定律。对不同并行比例，画出加速比随处理器数的变化。

import jax.numpy as jnp
import matplotlib.pyplot as plt

n_procs = jnp.arange(1, 65)

for p, color in [(0.5, "#e74c3c"), (0.9, "#f39c12"), (0.95, "#27ae60"), (0.99, "#3498db")]:
    speedup = 1 / ((1 - p) + p / n_procs)
    plt.plot(n_procs, speedup, color=color, linewidth=2, label=f"p={p}")
    # 最大加速比的水平线
    plt.axhline(1 / (1 - p), color=color, linestyle="--", alpha=0.3)

plt.xlabel("Number of processors")
plt.ylabel("Speedup")
plt.title("Amdahl's Law: Serial Fraction Limits Speedup")
plt.legend()
plt.grid(True)
plt.show()
​