Day 07 · 同步原语 + IPC + Week 01 综合复盘

1. 为什么需要同步：竞争条件的本质

来看一段你以为"显然没问题"的 Python 代码：

import threading

counter = 0

def add():
    global counter
    for _ in range(1_000_000):
        counter += 1

t1 = threading.Thread(target=add)
t2 = threading.Thread(target=add)
t1.start(); t2.start()
t1.join();  t2.join()
print(counter)   # 你期望 2_000_000，但很多次跑只会得到 1_374_812、1_902_447……

问题出在 counter += 1。这一行在 CPython 字节码里其实是三步：

线程 A 读到 counter=100，准备写回 101；恰好此刻线程 B 也读到 100，也准备写回 101。最终 counter=101——这次更新丢了一次。这就是竞争条件（race condition）：执行结果依赖于两个或更多线程的相对调度顺序。

很多人会问："Python 不是有 GIL 吗？"GIL 只保证一个字节码指令内部不被打断，不保证多个字节码组成的复合操作是原子的。+= 是 4 条字节码，所以可以被打断。这也是为什么 list.append()（C 实现里只有一次原子操作）在 GIL 下安全，但 list[i] += 1（要 LOAD_SUBSCR + BINARY_ADD + STORE_SUBSCR）不安全。

所有同步原语的目标只有一个：把一段「不希望被打断」的代码——临界区（critical section）——保护起来，让它在任何时刻最多被一个线程执行。这种性质叫互斥（mutual exclusion）。

2. 同步原语全家福

2.1 Mutex（互斥锁）

最朴素的工具。同一时刻只有一个线程能"持有"锁，其余线程阻塞等待。在 Linux 里，pthread mutex 由 futex（fast userspace mutex）实现：没人争抢时整个加锁解锁在用户态完成，零 syscall；一旦有人需要排队，才陷入内核挂起。这个"快路径在用户态、慢路径在内核态"的设计是现代锁性能的核心。

import threading
lock = threading.Lock()
counter = 0

def add():
    global counter
    for _ in range(1_000_000):
        with lock:           # 进入临界区
            counter += 1     # 现在是原子的

2.2 Spinlock（自旋锁）

不阻塞，原地"自旋"忙等：

while (!atomic_compare_exchange(&lock, 0, 1)) {
    /* 什么都不干，CPU 空转 */
}

看起来很傻，但极短的临界区（比如十几条指令）下，自旋反而比 mutex 快——因为 mutex 的休眠/唤醒至少要两次上下文切换（几微秒），而 spinlock 可能只浪费几十纳秒。Linux 内核里中断处理、调度器自身用的全是 spinlock，不能在中断上下文里"睡觉"。用户空间几乎不需要 spinlock，除非你在写 lock-free 数据结构。

2.3 RWLock（读写锁）

读操作可以并发，写操作必须独占。适合"读多写少"场景。注意陷阱：

• 写者饥饿：如果读者不断进来，写者可能永远拿不到锁。多数实现（如 pthread）默认读者优先，要显式选写者优先策略。
• 升级死锁：持读锁的同时想升级到写锁，几乎所有 RWLock 都不支持，必须先释放读锁再争抢写锁，期间可能被别人插队。

2.4 Semaphore（信号量）

本质是一个非负计数器，提供两个原子操作：P/wait（计数 -1，若 <0 则阻塞）和 V/post（计数 +1，若有人等就唤醒）。Mutex 是 semaphore 的特例（初始值 1）。Semaphore 真正的杀手锏是限流——比如限制最多 8 个线程并发访问数据库：

sema = threading.Semaphore(8)
def query():
    with sema:
        run_db_query()   # 最多 8 个并发

2.5 Condition Variable（条件变量）

解决"等待某个条件成立"的问题。必须配合 mutex 使用。经典生产者-消费者模板：

cond = threading.Condition()
queue = []

def producer():
    with cond:
        queue.append(item)
        cond.notify()         # 唤醒一个 consumer

def consumer():
    with cond:
        while not queue:      # 注意必须是 while 不是 if（防止虚假唤醒）
            cond.wait()       # 自动释放锁并睡，被唤醒后重新拿锁
        item = queue.pop(0)

为什么是 while 不是 if？POSIX 允许 cond.wait() 出现"虚假唤醒"（spurious wakeup）：没人 notify，它也可能自己醒。即使没有虚假唤醒，被唤醒到拿到锁之间，状态也可能被其他 consumer 改掉。所以醒来必须重新检查条件。

对比小表

3. 原子操作与 CAS：硬件提供的最小同步单元

所有上层同步原语的地基都是原子操作（atomic operation）——一条 CPU 指令完成读+改+写，过程中不会被其他核心看到中间状态。最重要的原子原语是 CAS（Compare-And-Swap）：

// 伪代码：原子地执行下面整块
bool CAS(int *ptr, int expected, int new_val) {
    if (*ptr == expected) {
        *ptr = new_val;
        return true;
    }
    return false;
}

在 x86 上对应 LOCK CMPXCHG 指令；ARM 上对应 LDREX/STREX 对（Load-Exclusive / Store-Exclusive）。所有 mutex/spinlock 都是用 CAS 实现的最外层入口。

CAS 还撑起了无锁数据结构（lock-free data structure）。例子：原子计数器

do {
    old = counter;
    new = old + 1;
} while (!CAS(&counter, old, new));

不需要锁，多线程竞争激烈时第二次循环就可能成功。代价是 ABA 问题：在你 CAS 之间，值从 A 变 B 又变回 A，CAS 会"误判"。一般用版本号或 tagged pointer 解决。

在 Python 里，标准库的 queue.Queue 用的是 Lock + Condition；高性能场景可以用 multiprocessing.Value（共享内存 + ctypes 原子操作）或第三方 atomics 库。

4. 死锁的四个必要条件（Coffman 条件）

1971 年 Coffman 证明：死锁发生当且仅当下面四个条件同时成立——

1. 互斥（Mutual Exclusion）：资源同时只能被一个线程持有。
2. 持有并等待（Hold and Wait）：线程持有资源 R1，同时等待 R2。
3. 不可剥夺（No Preemption）：资源不能被强制收回，只能由持有者主动释放。
4. 循环等待（Circular Wait）：存在线程环 T1→T2→...→Tn→T1，每个等待下一个持有的资源。

破解任意一条即可。工程上最常用的是破解条件 4：所有线程按全局固定顺序加锁。比如转账：

def transfer(a, b, amount):
    # 不管 a 转 b 还是 b 转 a，都按 id 小的先加锁
    first, second = (a, b) if id(a) < id(b) else (b, a)
    with first.lock:
        with second.lock:
            a.balance -= amount
            b.balance += amount

另一个常用招是给锁加超时（破解条件 3 的变种）：拿不到就回滚释放已持有的锁，回退一段随机时间重试，类似以太网 CSMA/CD 的 backoff。Python 的 lock.acquire(timeout=...) 就是干这个。

5. IPC 六大件

线程同步解决的是"同一进程内部的协作"，IPC（Inter-Process Communication）解决的是"跨进程的协作"。Linux 提供六大件，理解它们的差异比记住 API 重要：

5.1 匿名管道——shell 管道的本质

你每天用的 cat foo | grep bar | wc -l 背后就是匿名管道。shell 先 pipe() 拿到一对 fd（读端 + 写端），fork() 三个子进程，每个子进程用 dup2() 把对应 fd 重定向到 stdin/stdout，再 exec()。

# 用 Python 演示
import os
r, w = os.pipe()
if os.fork() == 0:
    os.close(r)
    os.write(w, b"hello from child")
    os._exit(0)
os.close(w)
print(os.read(r, 1024))   # b'hello from child'

5.2 共享内存——最快也最危险

两个进程 mmap 同一块物理内存，从此对它的读写就是普通内存访问，不走内核、不复制。这是大数据 IPC 的唯一可选项。代价：没有任何同步保证，你必须自己用 semaphore/mutex 配套保护，否则就是 race condition。

from multiprocessing import shared_memory
import numpy as np

# 进程 A 创建
shm = shared_memory.SharedMemory(create=True, size=4*1024*1024, name="kv_cache")
arr = np.ndarray((1024, 1024), dtype=np.float32, buffer=shm.buf)

# 进程 B 直接打开同一块
shm2 = shared_memory.SharedMemory(name="kv_cache")
arr2 = np.ndarray((1024, 1024), dtype=np.float32, buffer=shm2.buf)
# arr2 修改，arr 立即看到

5.3 Unix Domain Socket（UDS）

语法和网络 socket 一模一样，但在同一台机器上走的是内核内存拷贝，比 TCP loopback 还快（因为省了协议栈）。Docker daemon (/var/run/docker.sock)、systemd、Kubernetes kubelet 都用 UDS 暴露 API。UDS 还可以传文件描述符——这是 nginx master-worker、container runtime 移交 socket 的核心技巧。

6. 信号 signal——异步事件的"轻量级中断"

信号是 OS 给进程的异步通知。它打断进程当前执行，跳到信号处理函数，完事再回到原代码。你按 Ctrl-C 触发的就是 SIGINT。

常用信号速查

信号的最大陷阱：信号处理函数是 async-signal-safe 的。你不能在里面调 printf、malloc、获取任何锁——会死锁。安全的做法是只写一个 flag，主循环检查 flag 后再做事，或者用 signalfd/self-pipe trick 把信号转成可读事件，融入主 event loop。

import signal, sys

stop = False
def handler(signum, frame):
    global stop
    stop = True              # 只设标志，不做事
signal.signal(signal.SIGTERM, handler)
signal.signal(signal.SIGINT, handler)

while not stop:
    do_one_iteration()
graceful_shutdown()

K8s 滚动更新就靠这个：kubelet 先发 SIGTERM 给容器进程，等 terminationGracePeriodSeconds（默认 30s）让你优雅退出，超时不退再发 SIGKILL。这也是为什么生产服务必须捕获 SIGTERM 做收尾——否则正在处理的请求会被硬切断。

7. 案例研读：vLLM 怎么用共享内存协调多 GPU

把今天学的东西串起来。vLLM 在 tensor-parallel 模式下，要把一个 LLM 模型切到 N 张 GPU 上，每张 GPU 由一个独立 Python 进程驱动（受 GIL 限制，多进程而非多线程）。这些进程之间有大量协调需求：

具体用到的机制：

• multiprocessing.shared_memory——主进程把请求 batch 写入共享内存，worker 直接读，避免 pickle 序列化几十 MB 的 token tensor。
• 信号量 / Event——主进程通知 worker"新 batch 到了"，worker 处理完通知主进程"结果好了"。
• Unix Domain Socket——OpenAI 兼容 API server 与 engine 进程之间用 UDS 通信，比 TCP 更快也更安全（不监听网络端口）。
• SIGTERM 处理——优雅停机时主进程发 SIGTERM 给 worker，worker 先完成当前 token 的 forward 再退出，避免 GPU 显存悬挂。
• spinlock 思想——continuous batching 的调度循环就是一个紧密 spinning 的事件循环，不睡眠以最小化 TTFT（time to first token）。

如果你能看懂 vLLM 的 vllm/executor/multiproc_worker_utils.py，你就已经把今天的内容打通到工业级实践了。

📚 Week 01 综合复盘

# 进程
ps -ef                    # 所有进程全格式
ps -eLf                   # 看线程
pstree -p                 # 树状关系
top / htop                # 实时
pidstat -t -p <pid> 1     # 单进程明细

# 调度与性能
uptime                    # load average
mpstat -P ALL 1           # 每核 CPU
perf stat -e cache-misses,branches ./prog
perf top                  # 热点函数
taskset -c 0-3 ./prog     # CPU 亲和性

# 跟踪
strace -f -e openat python a.py
ltrace -e malloc ./prog
py-spy dump --pid <pid>

# 权限
ls -l / stat /etc/shadow
getcap -r /usr/bin 2>/dev/null
sudo -l
find / -perm -4000 2>/dev/null   # 全 SUID 文件

# IPC
ipcs -m / -s / -q           # System V IPC
ls -la /dev/shm/            # POSIX shm
lsof -U                     # Unix sockets

# 信号
kill -l                     # 信号列表
kill -SIGTERM <pid>
cat /proc/<pid>/status | grep Sig    # 看进程信号 mask

今日小练习（3 道）

Week 02 预告：虚拟内存（OS 的另一座大山）