并发与并行

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

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

并发 vs 并行

• 并发是关于管理多个任务。任务通过交错执行取得进展：任务A运行一会儿，然后任务B，再回到任务A。在单核上，并发制造了同时执行的假象。任务并非真正同时执行；它们轮流执行。

• 并行是关于同时执行多个任务。在 \(n\) 个核心上，\(n\) 个任务可以真正在同一时间运行。并行需要多个硬件执行单元。

• 比喻：并发是一个厨师在切菜和搅拌锅之间交替工作。并行是两个厨师，每人同时做一个任务。一个系统可以是并发但不并行（单核，任务交错），并行但不并发（多核运行独立的、无交互的程序），或者两者兼具（多核运行交错且有交互的任务）。

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

同步原语

• 当多个线程共享数据时，同步可以防止竞态条件。当结果依赖于不可预测的线程执行顺序时，就会发生竞态条件。

• 考虑两个线程同时增加一个共享计数器：counter += 1。这实际上是三个操作：(1) 读取计数器，(2) 加1，(3) 写回计数器。如果两个线程都读取到相同的值（比如5），都加1，都写回6，最终计数器变成6而不是正确的7。一次递增丢失了。

• 互斥锁确保一次只有一个线程访问临界区。线程在进入临界区前获取锁，之后释放锁。任何其他试图获取已被占用锁的线程都会阻塞，直到锁被释放。

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

• 互斥锁是正确的，但会引入争用：如果许多线程竞争同一个锁，它们花时间等待而不是计算。这限制了可扩展性。极端情况下，所有线程都争夺同一个锁，导致整个程序串行化。

• 信号量是对互斥锁的推广。计数信号量维护一个计数器：wait() 递减计数器（若将变成负数则阻塞），signal() 递增计数器。初始化为1的信号量行为类似于互斥锁。初始化为 \(n\) 的信号量允许最多 \(n\) 个线程同时进入临界区（对资源池如数据库连接很有用）。

• 条件变量让一个线程等待直到特定条件满足。线程释放一个锁，在条件变量上等待，当另一个线程发出条件信号时被唤醒。这避免了忙等待（在循环中反复检查条件，浪费CPU）。

• 监视器将互斥锁、条件变量和共享数据捆绑成一个单独的抽象。Java的 synchronized 关键字和Python的 threading.Condition 实现了类似监视器的语义。

• 读写锁区分读者（可以共享访问，因为读不修改数据）和写者（需要独占访问）。多个读者可以同时持有锁，但写者会阻塞所有读者和其他写者。当读操作远多于写操作时（例如，一个缓存了模型用于预测的服务），这是最优的。

经典并发问题

• 生产者-消费者（有界缓冲区）：生产者生成物品并将它们放入固定大小的缓冲区；消费者移除物品。挑战：缓冲区满时生产者必须等待，缓冲区空时消费者必须等待，并且两者都必须避免损坏缓冲区。

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

• 读者-写者：多个读者可以同时读取，但写者需要独占访问。挑战在于公平性：如果读者不断到来，写者可能会饥饿（永远得不到访问权）。解决方案可以优先考虑读者或写者，或者交替公平地进行。

• 哲学家就餐问题：五位哲学家围坐在一张桌子旁，他们之间有五把叉子。每位哲学家需要两把叉子才能吃饭。如果五个人同时拿起左边的叉子，没有人能拿起右边的叉子，所有人都会饿死（死锁）。解决方案包括：原子地拿起两把叉子，引入不对称（一位哲学家先拿起右边的叉子），或者使用服务员（限制就餐人数为4的信号量）。

死锁

• 死锁发生在一组线程中，每个线程都在等待该组中另一个线程持有的资源，形成一个依赖循环。谁也无法继续。

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

  1. 互斥：资源一次只能被一个线程持有。
  2. 持有并等待：一个线程在等待另一个资源的同时持有一个资源。
  3. 不可抢占：资源不能被强制从线程手中夺走。
  4. 循环等待：等待图中存在一个循环。

• 死锁预防破坏四个条件之一：

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

• 死锁避免动态决定授予资源请求是否可能导致死锁。银行家算法维护每个线程的最大可能需求，并且只授予那些使系统保持在“安全状态”（即所有线程最终都能完成的状态）的请求。该算法每次请求的复杂度为 \(O(n^2 m)\)（\(n\) 个线程，\(m\) 种资源类型），对大多数实际系统来说代价过高。

• 死锁检测允许死锁发生，然后检测它们（通过在等待图中寻找循环）并恢复（通过终止一个线程或回滚一个事务）。

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

无锁与免等待数据结构

• 锁会引入争用、优先级反转以及死锁风险。无锁数据结构完全避免使用锁，转而使用硬件提供的原子操作。

• 关键的原子操作是比较并交换（CAS）：原子地检查一个内存位置是否包含期望值，如果是，则将其替换为新值。伪代码：

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

• CAS 作为单条硬件指令实现，因此即使没有锁也是原子的。无锁算法在重试循环中使用 CAS：读取当前值，计算新值，尝试 CAS。如果另一个线程在此期间修改了该值，CAS 失败，线程重试。

• 无锁：至少有一个线程能在有限步数内取得进展（不可能死锁，但个别线程可能在争用下无限重试）。

• 免等待：每个线程都能在有界步数内取得进展（最强保证，但最难实现）。

• 无锁的栈、队列和哈希映射广泛应用于高性能系统。Java 的 ConcurrentHashMap 和 Go 的原子操作都是基于 CAS 构建的。

并行编程模型

• 共享内存并行：所有线程访问同一内存空间。同步是程序员的责任。OpenMP 提供编译器指令来并行化循环：

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

• 编译器将循环迭代分配到可用的核心上。OpenMP 对数据并行工作负载（对许多数据点执行相同操作）非常有效，广泛应用于科学计算。

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

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（单指令多线程）模型：数千个线程在不同的数据上执行相同的指令。这非常适合矩阵运算（第2章），其中对每个元素应用相同的乘加操作。我们将在后面的章节中详细介绍 GPU 编程。

异步与事件驱动编程

• 并非所有并发都需要线程。异步编程使用事件循环在单个线程上处理许多 I/O 密集型任务。

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

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

async def fetch_data(url):
    response = await http_get(url)  # 在这里挂起，事件循环运行其他任务
    return process(response)         # 当响应到达时恢复

• await 关键字挂起协程并将控制权返回给事件循环。当被等待的操作完成时，协程从它离开的地方恢复执行。这是协作式多任务：协程自愿让出，与操作系统强制切换线程的抢占式多任务不同。

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

• Python 的全局解释器锁（GIL） 阻止了线程的真正并行：一次只有一个线程可以执行 Python 字节码。这就是为什么 Python 使用多进程（独立的进程，每个都有自己的解释器）来实现 CPU 并行，而使用异步来实现 I/O 并发。GIL 将在 Python 3.13+ 中被移除（自由线程 Python），这将开启真正的多线程并行。

缩放定律

• 阿姆达尔定律描述了并行化程序的理论加速比。如果程序的 \(p\) 部分是可并行的，其余 \(1-p\) 部分是串行的：

\[\text{Speedup}(n) = \frac{1}{(1-p) + \frac{p}{n}}\]

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

• 这对机器学习有深远的影响：如果数据加载占训练时间的 10% 并且是串行的，那么添加更多 GPU 最多只能将训练速度提高 10 倍。那 10% 的串行瓶颈限制了所有努力（这就是为什么高效的数据管道和计算与 I/O 重叠如此重要，第6章）。

• 古斯塔夫森定律提供了一个更乐观的视角。它不固定问题规模并增加处理器，而是固定总时间并询问在相同时间内可以完成多少额外工作。如果并行部分随问题规模缩放：

\[\text{Speedup}(n) = 1 - p + p \cdot n\]

• 这是关于 \(n\) 线性的。论点是：有更多处理器时，我们解决更大的问题，而不是更快地解决相同的问题。在机器学习中，这对应于随 GPU 增加而增加批量大小（弱扩展），而不是保持批量大小固定（强扩展）。

编码任务（使用 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"期望值: {4 * 100000}")
   print(f"实际值:   {counter}")
   print(f"丢失的更新: {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} (正确值: {4 * 100000})")
   print(f"带锁的时间: {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("处理器数量")
   plt.ylabel("加速比")
   plt.title("阿姆达尔定律：串行部分限制加速比")
   plt.legend()
   plt.grid(True)
   plt.show()