P12 · SharedMemoryConnector：跨进程 put/get 协议
thinker 的 hidden states 每 token 都要发给 talker 进程：用 POSIX 共享内存，没有额外元数据通道——两边只靠 key 推出名字，用文件锁表达“写完了”。
mine/p12_shm_connector.py
验收 ./check.sh p12
提示 A-20 · 参考解 C-20 solutions/p12_shm_connector.py
题面速览
- 在系统中的位置：stage 之间的大块数据（hidden states、embedding、codec）不走 orchestrator，而由 worker 直接写进共享内存，下游 stage 的后台线程去读。同机部署默认就是 SharedMemoryConnector。
- 协议：
key 例如 "req42_0_7"（请求 42、stage 0 发出的第 7 块）
name = 短哈希(key) # macOS 的 POSIX shm 名字最长 31 字节
lock = <lock_dir>/<name>.lock
put(key, data): get(key):
open(lock, O_CREAT); flock(EX) open(lock) → 不存在：None
创建 shm（若残留旧段，先 unlink） flock(EX | NB) → BlockingIOError：还在写，None
写 8 字节长度 + data 打开 shm → 不存在：None
flock(UN) 读出 → close → unlink（读者释放）→ 删除 lock 文件
- 8 字节长度头：shm 段大小会被向上取整到页，读者无法从段大小得知真实长度。
- resource tracker 陷阱（测试会复现）：Python < 3.13 中
SharedMemory(create=True)会把段登记到创建者的 resource tracker。写进程先退出时，它的 tracker 把仍在登记中的段当“泄漏”删掉——读者再去读数据已经没了。解决：写完后把所有权转移给读者（从 tracker 注销），由读者 unlink。 - 生产者清理：请求被 abort 时下游可能永远不来读。生产者记住自己写过的 key，
cleanup_prefix("req42_")删除其中尚未被消费的段。
验收要点：往返且只能消费一次 · 空 payload 与 4MB payload · 独立写进程先退出 1 秒后读者仍能读到 · 锁被他人持有时 get 返回 None · cleanup_prefix 只清理未被消费的同前缀 key。
直觉与类比
共享内存段就是一个快递柜格子，key 是取件码。快递员（写者）和收件人（读者）从没见过面，但用同一张表从取件码算出同一个格子号（shm_name(key)）。每个格子门上有锁（锁文件 + flock）：快递员先锁门再放东西，放完才开锁；收件人先试着锁门，锁不上（LOCK_NB → BlockingIOError）说明还在放，转身就走、待会再来。格子都是标准大号（一页），所以快递员在最前面贴纸条“实际 N 字节”（长度头）。收件人取走后亲手退格子（unlink）并摘掉门上的锁（删锁文件）。
每个 Python 进程都有一个物业清扫员（resource_tracker 子进程）。快递员租格子时物业把格子登记在他名下；快递员下班（进程退出）时清扫员一查：还没退的格子“肯定是忘了”——清空退掉，还在日志里抱怨 leaked shared_memory objects。可包裹收件人还没来取！解决：放好包裹后去物业把登记撤掉（resource_tracker.unregister），相当于过户给收件人。
框架视角 · T · SBWC · LIFE
P12 的参与者是进程，不是线程或 task：进程之间不共享 Python 对象，threading.Lock 毫无意义；能用的只有内核提供的东西——共享内存段、文件、flock——以及双方约定好的协议（名字怎么算、谁创建、谁释放）。
| P12 | |
|---|---|
| T | 写者进程、读者进程，外加容易被忘掉的第三方：每个进程自己的 resource_tracker 子进程。跨进程的同步原语只有 flock（建议性锁） |
| S | 内核里：段（写者创建、读者 unlink）、锁文件（写者 O_CREAT、读者删除；它的存在本身就是状态：“有一个未被消费的 key”）、flock 的持有者。进程本地：写者“我写过的 key”集合、tracker 的登记表 |
| B | 写者的 flock(LOCK_EX) 可以阻塞（最多等读者读完一块）；读者永不阻塞：LOCK_EX | LOCK_NB，拿不到就返回 None，由 P13 的 recv 线程稍后重试 |
| W | 写者：创建 + 写 + 交出所有权。读者：读 + 拷贝 + 释放。tracker：进程退出时清扫仍登记着的段——你没写、但一定会运行的 worker |
| C | “名字”就是通道：shm_name(key) / lock_path(key) 两边独立计算。唯一的带内元数据是 8 字节长度头 |
| L | 段：写者 create → 写 → close（不 unlink）→ 从 tracker 注销 → 读者 attach → 拷贝 → close → unlink。锁文件：写者创建 → 读者删除。abort：写者 cleanup_prefix 回收 |
| F | ① 读者能锁上门时，段要么不存在、要么已完整写好；② 每个段恰好被 unlink 一次；③ 锁文件存在 ⇔ 这个 key 写过且没被消费；④ get 的每条 return 路径都解锁并关闭 fd |
| E | 同名残留段（FileExistsError → 替换）；锁文件在而段不在；写者先退出；空与 4MB payload；前缀 "spk3_" 不能误伤 "spk30_…" |
易错点
- 用 Python 的
hash()算名字：不同进程有随机种子，两边算不出同一个名字；用 hashlib。 - 名字太长：macOS 上限 31 字节（含前导
/）。 - 只靠
shm.size推断长度：读出整页；空 payload 因 size=0 无法创建（ValueError: 'size' must be a positive number）。 - 先建段后拿锁：读者可能在两者之间拿到锁读到半成品。
- 写者忘了
resource_tracker.unregister：写进程退出后读者拿到 None。 data = shm.buf[8:8+n]后直接shm.close()：BufferError（memoryview 还在）；要先bytes(...)拷贝。- get 的某个
return None分支没解锁/关 fd：fd 泄漏、锁一直挂着。 - 锁文件在但段不在时抛异常或删锁文件：应返回 None 下次再来。
- cleanup_prefix 前缀匹配别的请求，或把已被消费的 key 也计数。
逐步图：一个 key 的生命周期（L = 锁文件，S = shm 段，T = 写者 tracker 中的登记）
L 无 · S 无 → get 返回 None（锁文件不存在）
L 有 · flock 无人持有 · S 无 → None（段不存在）——不要抛异常，也不要删锁文件
L 有 · flock 写者持有 · S 有（未写完）· T 登记 → 已注销 → None（BlockingIOError）
L 有 · flock 无 · S 有 · T 无 → 数据
L 有 · S 有（因为 T 里没有它）→ 数据
如果写者没有注销，这一行的 S 会变成？
无——写者的 tracker 在退出时把它 unlink 了，读者拿到 None。
L 无 · S 无（读者 unlink）→ 再 get 返回 None
三级提示
L1I/O 示例与关键问题I/O 示例 + 关键问题 + 细节（轻度剧透）
例 1：往返 + 只能消费一次
c = ShmConnector(namespace="hv"); k = "req7_0_0"
c.put(k, b"abc")
os.path.exists(c.lock_path(k)) -> True # put 结束后锁文件还在
c.get(k) -> b"abc"
c.get(k) -> None # 读者已 unlink 段并删了锁文件
c.get("never_written") -> None
c.shm_name(k) == c.shm_name(k) -> True # 确定性
len(c.shm_name(k)) -> 例如 23 # namespace + 短哈希；必须远小于 31
c.put(k, b"one"); c.put(k, b"two!") # 同一个 key 写两次（旧段未被读）
c.get(k) -> b"two!" # put 发现残留段 → 替换
例 2：锁被别人持有 = 还在写
c.put("job_0_3", b"xyz")
fd = os.open(c.lock_path("job_0_3"), os.O_RDWR); fcntl.flock(fd, fcntl.LOCK_EX)
c.get("job_0_3") -> None # 不阻塞，立即返回
fcntl.flock(fd, fcntl.LOCK_UN); os.close(fd)
c.get("job_0_3") -> b"xyz"
即使“别人”是同进程里另一次 os.open 得到的 fd，flock 也会冲突——flock 锁属于打开的文件描述（open file description），不属于进程。
例 3：cleanup_prefix
for i in range(3): c.put(f"rA_0_{i}", bytes([i]))
c.put("rAx_0_0", b"q")
c.get("rA_0_1") -> b"\x01" # 被消费了一个
c.cleanup_prefix("rA_") -> 2 # rA_0_0、rA_0_2；"rAx_0_0" 不以 "rA_" 开头
c.cleanup_prefix("rA_") -> 0
c.get("rA_0_0") -> None
c.get("rAx_0_0") -> b"q"
关键问题
- 读者只知道 key，怎么知道打开哪个段、哪个锁文件？这对 shm_name 提出了什么要求？
- 为什么要 8 字节长度头？只靠 shm.size，3 字节 payload 会读出多少？
put(k, b"")会怎样？ - get 返回 None 有三种原因，分别对应写者生命周期的哪个阶段？
- “先拿锁再创建段”和“先创建段再拿锁”有什么区别？
- 谁负责 unlink 段？为什么不是写者？写者退出时谁会“好心”替它 unlink？
- cleanup_prefix 怎么判断一个自己写过的 key 已经被消费了？
- 读者为什么要在 close() 之前就把数据
bytes(...)拷出来？
细节
SharedMemory(name=...)在 POSIX 上自动加/；通常用namespace + "_" + hashlib 前 N 位。- macOS 上
SharedMemory(create=True, size=5).size是 16384（一页）。 fcntl.flock(fd, LOCK_EX)阻塞；LOCK_EX | LOCK_NB拿不到抛 BlockingIOError。锁是建议性的。- 锁文件 = “存在一个未消费的 key”；lock_dir 为 None 时 Linux 用 /dev/shm，否则临时目录。
- resource tracker：创建（3.12 中附着也会）时登记，
unlink()会注销。读者“附着 → unlink”是平衡的；写者“创建 → 不 unlink”不平衡——所有权要显式转交。 - 残留段：
create=True抛 FileExistsError，要替换而不是失败或追加。
L2思路与常见坑思路拆分与常见坑（中度剧透）
- 状态只有：lock_dir、namespace、“我写过的 key”集合。
lock_path(key)由shm_name(key)派生。 - put 的顺序：
os.open(lock, O_CREAT|O_RDWR)→flock(LOCK_EX)→ 创建段（FileExistsError 时附着旧段、close、unlink，再创建）→ 注销 tracker → 写长度头和数据 →close()→ 记录 key → finally 里 LOCK_UN 并关 fd。锁必须在创建段之前拿到。 - get 的顺序：
os.open(lock, O_RDWR)（FileNotFoundError → None）→flock(LOCK_EX|LOCK_NB)（BlockingIOError → None）→ 附着段（FileNotFoundError → None）→ 读长度、拷贝 → close → unlink → 删锁文件 → finally 解锁关 fd。 - tracker 修复：
resource_tracker.unregister(shm._name, "shared_memory")（_name带/前缀）；try/except 包住兼容不同版本。 - cleanup_prefix：遍历记录的 key（先拍快照），前缀匹配的从记录中删除；锁文件还在（没人读过）就附着 + unlink 段、删锁文件，计数 +1；吞掉 FileNotFoundError。
L3详解：答案详解：答案、逐步追踪、骨架提示（重度剧透）
关键问题答案
- 两边各自用 shm_name(key) / lock_path(key) 推出名字：确定性（用 hashlib 不用 hash()）、短（≤ 30 左右）、只含安全字符（十六进制即可）。
- 只靠 shm.size 会读出整页（macOS 上 16384 字节）；空 payload 会因 size=0 无法创建。
- 锁文件不存在（未写/已读）；锁被持有（正在写）；段不存在（写者刚创建锁文件还没建段）。
- 先建段后拿锁：读者可能在两者之间拿到锁读出没写完的段；先锁后建段则不可能。
- 读者负责 unlink：只有读者知道“已经用完了”。写者退出时它的 resource tracker 会替它 unlink——这正是要阻止的。
- 读者成功 get 后会删锁文件，所以用
os.path.exists(lock_path(key))判断。 - memoryview 切片持有 buffer 导出，close 会抛 BufferError；而且 close/unlink 之后内存可能已无效。
思考：get 删锁文件时还持有 flock，这在“每个 key 只有一个消费者、key 不复用”的前提下是安全的；如果允许同一个 key 被反复写，会出现什么竞争？
面试官追问 & 对照源码
- 同卡场景用 CUDA IPC 零拷贝：生命周期和跨进程流同步怎么保证？
- 生产者如何在不等 ACK 的前提下回收自己的 bookkeeping？
对照源码：distributed/omni_connectors/connectors/shm_connector.py:195-364 · entrypoints/stage_utils.py:193-237