3、核间通信机制:共享内存设计原则、Mailbox中断机制、Spinlock自旋锁实现
双核处理器,说白了就是两个大脑要一起干活。但问题来了——它们怎么交流?怎么知道对方在想什么?怎么保证不抢同一块数据?
我在项目中遇到过不少因为核间通信没做好导致的诡异bug。比如系统跑着跑着突然死机,或者两个核互相等待,谁也动不了。嗯,这些坑我踩过不少。今天就把核间通信的三大法宝——共享内存、Mailbox、Spinlock——掰开揉碎了讲清楚。
3.1 共享内存设计原则
共享内存是最直接的通信方式。两个核都能访问同一块物理内存,数据写进去,对方就能读到。听起来简单,但设计不好就是灾难。
核心原则:共享内存只放数据,不放控制信息。控制信息走Mailbox,数据走共享内存。这个原则我吃了两次亏才真正理解。
3.1.1 内存区域划分
我个人习惯把共享内存分成三个区域:
| 区域 | 用途 | 大小建议 |
|---|---|---|
| 控制区 | 状态标志、数据就绪标志 | 64字节 |
| 数据区 | 实际传输的数据包 | 根据业务定,建议4KB起步 |
| 调试区 | 日志、错误码、计数器 | 256字节 |
你想想看,如果控制信息和数据混在一起,一个核在写数据,另一个核在读控制标志,很容易读到半截数据。所以一定要分开。
3.1.2 数据一致性保证
共享内存最大的坑就是缓存一致性问题。ARM Cortex-A系列有L1 cache,两个核各自缓存了一份数据,你以为写进去了,对方读到的却是旧值。
我曾经在一个项目中,Core0写了一个标志位,Core1死活读不到。查了两天才发现是cache没刷。解决办法很简单:
// Core0 写入数据后,必须刷cache
void shared_memory_write(volatile uint32_t *addr, uint32_t value) {
*addr = value;
// 刷写该地址所在的cache line
__clean_dcache(addr);
// 发送Mailbox通知Core1
send_mailbox_interrupt(CORE1);
}
// Core1 读取数据前,必须无效化cache
uint32_t shared_memory_read(volatile uint32_t *addr) {
// 先无效化cache,确保读到的是最新值
__invalidate_dcache(addr);
return *addr;
}
注意:刷cache操作很耗时,一次大概几十到几百个时钟周期。不要每个变量都刷,建议按cache line大小(通常是32或64字节)对齐数据,一次刷一整块。
3.2 Mailbox中断机制
共享内存解决了数据存放的问题,但怎么通知对方「数据来了」?轮询?太浪费CPU了。这时候就需要Mailbox。
Mailbox说白了就是一个硬件邮箱。你往里面塞一封信(写一个寄存器),对方就会收到一个中断。硬件帮你做了这件事,不需要软件轮询。
3.2.1 Mailbox工作原理
大多数双核处理器(比如TI的OMAP系列、NXP的i.MX系列)都内置了Mailbox硬件模块。它的工作流程是这样的:
- Core0往Mailbox的某个寄存器写入数据
- 硬件自动产生一个中断信号给Core1
- Core1的中断服务程序读取Mailbox寄存器,获取消息
- Core1处理完数据后,可以同样方式回复Core0
我记得第一次用Mailbox时,以为它只能传一个整数值。后来才发现,Mailbox寄存器通常有多个通道,每个通道可以传32位数据。你可以用这32位作为消息类型,或者作为共享内存的指针。
3.2.2 中断服务程序设计
Mailbox中断服务程序要尽量短。为什么?因为中断上下文里不能做复杂操作,否则会影响实时性。
// Core1的Mailbox中断服务程序
void mailbox_isr(void) {
uint32_t msg;
// 读取Mailbox消息
msg = read_mailbox(MAILBOX_CHANNEL_0);
// 根据消息类型处理
switch (msg & 0xFF) { // 低8位是消息类型
case MSG_DATA_READY:
// 数据已写入共享内存,去处理
process_shared_data();
break;
case MSG_SYNC_REQUEST:
// 同步请求,设置标志位
g_sync_flag = 1;
break;
case MSG_ERROR:
// 错误通知
handle_error(msg);
break;
default:
// 未知消息,记录日志
log_error("Unknown mailbox msg: 0x%x", msg);
break;
}
// 清除Mailbox中断标志
clear_mailbox_interrupt(MAILBOX_CHANNEL_0);
}
小技巧:中断服务程序里只做「收到消息→设置标志→唤醒任务」这三步。真正的数据处理放到任务级去做。这样既保证了实时性,又不会阻塞其他中断。
3.3 Spinlock自旋锁实现
共享内存最怕什么?两个核同时写同一个变量。比如Core0在读数据,Core1在写数据,读到的数据就是半成品。这时候就需要锁机制。
Spinlock(自旋锁)是最简单的锁。一个核拿了锁,另一个核就一直在那里转圈等,直到锁被释放。说白了就是「你不走,我不走,我就在这死等」。
3.3.1 硬件原子操作
实现Spinlock的关键是原子操作。ARM架构提供了LDREX/STREX指令对,可以保证「读-改-写」的原子性。
// 自旋锁结构体
typedef struct {
volatile uint32_t lock;
} spinlock_t;
// 初始化锁
void spinlock_init(spinlock_t *lock) {
lock->lock = 0; // 0表示未锁定
}
// 获取锁(自旋等待)
void spinlock_acquire(spinlock_t *lock) {
uint32_t result;
do {
// 尝试原子性地将lock从0改为1
__asm__ volatile(
"ldrex %0, [%1]\n" // 读取lock值到result
"cmp %0, #0\n" // 检查是否为0
"strexeq %0, #1, [%1]\n" // 如果是0,写入1
: "=&r" (result)
: "r" (&lock->lock)
: "memory"
);
} while (result != 0); // 如果result不为0,说明没拿到锁,继续循环
// 内存屏障,确保后续操作在锁之后执行
__dmb();
}
// 释放锁
void spinlock_release(spinlock_t *lock) {
// 内存屏障,确保之前的操作在释放锁之前完成
__dmb();
lock->lock = 0;
}
避坑指南:我曾经在一个项目中,Spinlock的临界区里调用了printf函数。结果printf内部又用了同一个锁,直接死锁。记住:临界区里绝对不能调用可能引起阻塞或重入的函数。
3.3.2 使用场景与注意事项
Spinlock适合保护非常短的临界区。比如修改一个标志位、更新一个计数器。如果临界区操作超过几十微秒,建议用信号量或消息队列代替。
为什么?因为Spinlock是忙等。一个核拿着锁在干活,另一个核就在那里空转,浪费CPU。你想想看,如果临界区要执行1毫秒,另一个核就白转1毫秒,多亏啊。
| 场景 | 推荐机制 | 原因 |
|---|---|---|
| 保护短变量(< 10条指令) | Spinlock | 开销小,延迟低 |
| 保护中等长度操作 | Spinlock + 关中断 | 防止中断打断临界区 |
| 保护长操作(> 100μs) | 信号量或Mutex | 避免CPU空转 |
3.3.3 实际使用示例
下面是一个实际项目中用Spinlock保护共享缓冲区的例子:
// 共享缓冲区
static uint8_t g_shared_buf[1024];
static uint32_t g_buf_index = 0;
static spinlock_t g_buf_lock;
// Core0:写入数据
void core0_write_data(uint8_t *data, uint32_t len) {
spinlock_acquire(&g_buf_lock);
// 临界区开始
for (uint32_t i = 0; i < len; i++) {
if (g_buf_index < 1024) {
g_shared_buf[g_buf_index++] = data[i];
}
}
// 临界区结束
spinlock_release(&g_buf_lock);
// 通知Core1数据已更新
send_mailbox_interrupt(CORE1);
}
// Core1:读取数据
void core1_read_data(uint8_t *buf, uint32_t *len) {
spinlock_acquire(&g_buf_lock);
// 临界区开始
*len = g_buf_index;
for (uint32_t i = 0; i < *len; i++) {
buf[i] = g_shared_buf[i];
}
g_buf_index = 0;
// 临界区结束
spinlock_release(&g_buf_lock);
}
嗯,这里要注意:Spinlock配合Mailbox使用效果最好。Core0写完数据后释放锁,然后发Mailbox通知Core1。Core1收到中断后,先拿锁,再读数据。这样既保证了数据一致性,又避免了轮询浪费。
3.4 三种机制配合使用
在实际项目中,这三种机制不是孤立的。我总结了一个最佳实践:
- 共享内存:存放数据,按cache line对齐,分区管理
- Spinlock:保护共享内存的读写操作,临界区要短
- Mailbox:传递控制消息,触发对方处理数据
举个例子:Core0采集到传感器数据,先拿Spinlock,把数据写入共享内存,释放锁,然后发Mailbox告诉Core1「数据来了」。Core1收到中断后,拿Spinlock,读数据,释放锁,然后处理数据。整个过程清晰、高效、不会冲突。
最后说一句:核间通信没有银弹。共享内存快但不安全,Mailbox安全但有延迟,Spinlock简单但浪费CPU。你要根据实际场景选合适的组合。我在医疗设备项目中,通常用共享内存传大块数据(比如图像帧),用Mailbox传控制命令,用Spinlock保护关键标志位。这套组合拳打下来,基本没出过问题。
下一章我们讲双核任务分配与负载均衡,到时候会用到今天学的这些通信机制。先消化消化,有问题随时问。