一次 AMP 死锁排查
症状
/user-test-sched 三明治计时测试中，StarryOS（hart 0）侧客户端 ioctl(AWAIT) 永久阻塞。死锁只在完整 RPC 流程（DELAY + ECHO 连续调用）中出现，单步调用正常。
rt-async（hart 1）侧正常完成所有消息处理。
架构背景
QEMU virt RISC-V 64 双核 AMP。hart 0 运行 StarryOS（Linux 兼容内核），hart 1 运行 rt-async（M-mode RTOS）。双核通过共享内存 ring buffer 通信，CLINT MSIP 用作 IPI 通知。
hart 0 (StarryOS) hart 1 (rt-async)
───────────────── ─────────────────
用户态程序
│ call(DELAY) + IPI ─────────────────→ 接收请求
│ call_poll(ECHO) + IPI ─────────────→ 处理 DELAY
│ ioctl(AWAIT) ← 阻塞等待 处理 ECHO
│ send response + IPI
│ ←── 唤醒, 读响应 ←───────────────────────┘根因：spin::Mutex 不关中断
代码状态
ioctl(AWAIT) 使用 PollSet 管理 waker，PollSet 内部用 spin::Mutex（不关中断）保护 waker 队列。IPI handler 中调用 PollSet::wake() 也要获取同一把锁。
死锁场景
Task (hart 0) IPI 中断 (hart 0)
─────────────────────────────────────────────────────────
IPC_POLLSET.register(waker)
→ self.0.lock() ✓ 获得锁
→ 正在写入 waker...
IPI 到达！
ipi_irq_handler()
→ IPC_POLLSET.wake()
→ self.0.lock() ☠ 死等同 hart 上，中断 handler 等待被中断的代码释放锁——锁永远不会被释放，因为中断抢占了对端。
为什么单步调用不死锁
单步调用只触发一次 IPI。register() 和 wake() 不太可能同时竞争同一把锁。连续 RPC（DELAY + ECHO）使 IPI 频率翻倍，竞争窗口显著增大。
修复：SpinNoIrq + 移除 IPC_PENDING
1. SpinNoIrq 替换 PollSet
kspin::SpinNoIrq 在 lock() 时自动关中断，drop 时恢复。消除了同 hart 死锁：
static IPC_WAKER: SpinNoIrq<Option<Waker>> = SpinNoIrq::new(None);
fn ipi_irq_handler() {
if let Some(waker) = IPC_WAKER.lock().take() {
waker.wake();
}
}2. 移除 IPC_PENDING，改为 ring buffer 直读
IPC_PENDING: AtomicBool 是 ring buffer read != write 的冗余拷贝。两个状态源意味着不一致的可能。移除后直接读 CH1 ring buffer 的 read/write 原子变量：
fn ch1_has_pending() -> bool {
let base = shm_vaddr();
let r = unsafe {
(*((base + amp::CH1_RING_READ_OFFSET) as *const AtomicUsize))
.load(Ordering::Acquire)
};
let w = unsafe {
(*((base + amp::CH1_RING_WRITE_OFFSET) as *const AtomicUsize))
.load(Ordering::Acquire)
};
r != w
}ioctl(AWAIT) 实现：
poll_fn(|cx| {
if ch1_has_pending() {
return Poll::Ready(0usize);
}
let mut guard = IPC_WAKER.lock();
// SpinNoIrq.lock() 已关 sstatus.SIE，register 和 re-check 原子
if ch1_has_pending() {
Poll::Ready(0usize)
} else {
*guard = Some(cx.waker().clone());
Poll::Pending
}
})ring buffer 是唯一真相源，spurious wakeup 安全——被唤醒后直接检查 ring buffer，有数据就处理，没有就继续等。
3. intercom.rs IPI 策略
process_elastic 中每个 Notify 响应立即回 IPI（通过 process_all 的 on_notify 回调）。竞争检查窗口内处理的消息同理：
fn send_notify_ipi() {
unsafe { chip_qemu_virt_rt::send_ipi_to_linux() };
}
pub fn process_elastic() -> usize {
shm.set_busy();
let mut total_count = 0;
loop {
let n = SERVER.process_all::<RtAsyncRpc, _, _>(
|msg| handle_non_rpc(msg),
|| send_notify_ipi(),
);
total_count += n;
if n > 0 { continue; }
// 弹性自旋...
break;
}
shm.clear_busy();
core::sync::atomic::fence(Ordering::SeqCst);
if SERVER.has_pending() || SERVER.has_urgent() {
let n = SERVER.process_all::<RtAsyncRpc, _, _>(
|msg| handle_non_rpc(msg),
|| send_notify_ipi(),
);
total_count += n;
}
total_count
}改动总结
| 维度 | 改动前 | 改动后 |
|---|---|---|
| 中间标志 | IPC_PENDING: AtomicBool | 无（直接读 ring buffer） |
| waker 存储 | PollSet（spin::Mutex, 64槽） | SpinNoIrq<Option<Waker>> |
| IRQ 安全性 | 无（spin::Mutex 不关中断） | SpinNoIrq 类型保证 |
| IPI 策略 | 逐通知回 IPI | 逐通知回 IPI（ring buffer 直读兜底） |
| spurious wakeup | 不安全（IPC_PENDING 可能误导） | 安全（ring buffer 是权威状态） |
三条经验
1. 中断上下文共享的锁，必须关中断
这是嵌入式的常识，但容易被忽略——spin::Mutex 的 API 和 std::sync::Mutex 一模一样，看不出差异。在 bare metal 环境下，"锁"的语义比标准库多一个维度：IRQ 安全性。同 hart 上，中断 handler 等待被中断的代码释放锁——锁永远不会被释放。
2. 消除中间状态
IPC_PENDING 是对 ring buffer read != write 的冗余拷贝。两个状态源意味着不一致的可能。消除后 ring buffer 是唯一真相源，逻辑更简单，出错的表面积更小。
3. 类型系统是防止并发 bug 的最好工具
SpinNoIrq 把关中断编码在 lock() 语义里——不可能漏。比手动 NoPreemptIrqSave 靠人记忆可靠得多。