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硬件模块。它的工作流程是这样的:

  1. Core0往Mailbox的某个寄存器写入数据
  2. 硬件自动产生一个中断信号给Core1
  3. Core1的中断服务程序读取Mailbox寄存器,获取消息
  4. 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 三种机制配合使用

在实际项目中,这三种机制不是孤立的。我总结了一个最佳实践:

  1. 共享内存:存放数据,按cache line对齐,分区管理
  2. Spinlock:保护共享内存的读写操作,临界区要短
  3. Mailbox:传递控制消息,触发对方处理数据

举个例子:Core0采集到传感器数据,先拿Spinlock,把数据写入共享内存,释放锁,然后发Mailbox告诉Core1「数据来了」。Core1收到中断后,拿Spinlock,读数据,释放锁,然后处理数据。整个过程清晰、高效、不会冲突。

最后说一句:核间通信没有银弹。共享内存快但不安全,Mailbox安全但有延迟,Spinlock简单但浪费CPU。你要根据实际场景选合适的组合。我在医疗设备项目中,通常用共享内存传大块数据(比如图像帧),用Mailbox传控制命令,用Spinlock保护关键标志位。这套组合拳打下来,基本没出过问题。

下一章我们讲双核任务分配与负载均衡,到时候会用到今天学的这些通信机制。先消化消化,有问题随时问。