[i=s] 本帖最后由 HonestQiao 于 2026-5-5 19:38 编辑 [/i]
在 双核独立运行 中,M33核 和 M55核 已经能够各自独立闪烁 LED。本文在此基础上,进一步实现双核间的数据通信:由 M33核 每秒发送一个递增的计数器,M55核 接收并打印到串口。
通信框架选用 rpmsg-lite,它是 NXP 开源的轻量级 RPMsg 实现,专门面向 MCU 级别的异构多核场景,占用资源少、移植简单。PSoC E84 的 M33 与 M55 之间通过 共享内存(Shared Memory) 交换数据,rpmsg-lite 负责管理 vring 描述符环和 buffer 池,屏蔽底层细节。
一、rpmsg-lite 简介与共享内存布局
rpmsg-lite 的核心概念:
| 概念 |
说明 |
| Master |
负责初始化共享内存中的 vring、desc 链表和 buffer 池。本例中由M33 担任。 |
| Remote |
等待 Master 初始化完成后 attach 到已有 vring。本例中由M55 担任。 |
| Endpoint |
通信端点,通过 src/dst 地址路由消息。两端使用相同的 endpoint address(本例为 30)。 |
| Virtqueue |
虚拟队列,基于 virtio ring 实现,分为 tx(tvq)和 rx(rvq)。 |
在 PSoC E84 上,M33 与 M55 的共享内存区域由 BSP 预先分配:
共享内存基地址:0x261C0000
共享内存大小 :0x10000(64KB)
rpmsg-lite 会在该区域内自动放置两个 vring(rvq + tvq)以及 RL_BUFFER_COUNT 个 payload buffer。为了安全起见,我们在共享内存的末端(0x261CF000)放置一个启动握手标志(sentinel),用于解决 M55 和 M33 时序同步问题。
二、工程准备
在 双核独立运行 创建的两个工程基础上,确保各自已经添加了 rpmsg-lite 软件包:
- 打开 RT-Thread Settings
- 在 软件包 中搜索并添加
rpmsg-lite
- 保存配置,系统会自动下载并集成到
packages/rpmsg-lite/ 目录
两个工程都需要添加 rpmsg-lite。

三、rpmsg-lite 通信机制
在动手写代码之前,先理解 rpmsg-lite 在两核之间的数据流转路径。这样遇到问题时,才能定位是"发不出去"还是"收不到"。
正常通信路径(理想情况)
发送侧(M33):
application: rpmsg_lite_send(data)
-> core: rpmsg_lite_format_message (写入共享内存 vring)
-> virtqueue_kick
-> vq_ring_notify_host
-> virtqueue_notify
-> platform_notify (触发 IPC 中断 -> M55)
接收侧(M55):
IPC ISR (收到中断)
-> env_isr
-> virtqueue_notification (处理 rvq)
-> rpmsg_lite_rx_callback
-> ept->rx_cb (用户回调: rpmsg_rx_callback)
-> RL_RELEASE (释放 buffer 回空闲池)
本教程的实际路径(Polling workaround)
PSoC E84 的 platform_notify() 依赖 IPC 中断,而实际测试中发现这条路径不可靠(M55 经常收不到中断)。因此我们在两侧都启动轮询线程,用主动调用 virtqueue_notification() 的方式替代被动等待中断:
[M33 Polling 线程] 每 20ms: virtqueue_notification(rvq) + virtqueue_notification(tvq)
|
v
驱动 rvq 回调(处理 M55 的回复) + tvq 回调(更新 link_state)
[M55 Polling 线程] 每 20ms: virtqueue_notification(rvq)
|
v
驱动 rvq 回调 -> rpmsg_lite_rx_callback -> ept->rx_cb
启动时序与握手
这是另一个关键点。正常 RPMsg 流程假设 Master 和 Remote 先后启动,中间有足够时间让 master_init 完成。但 PSoC E84 上:
- M55 内核被 M33 释放复位后,启动速度极快
- M55 的
main() 可能在 M33 的 rpmsg_lite_master_init() 之前就运行到轮询线程
- 此时共享内存还是上电随机值,
virtqueue_get_available_buffer() 会读到垃圾 avail->idx,越界访问 desc[],触发 hard fault
解决:在 master_init 全部完成后,向共享内存末端写入一个 magic sentinel;M55 在调用 remote_init 之前先 spin 等待这个 magic。这就是"启动握手"机制。
整体通信架构图
下面这张图展示了本文最终实现的双核通信全貌,包含 Master/Remote 角色、共享内存布局、Polling 线程的位置,以及启动握手的时序关系:

四、M33 端:Master(发送方)
关键设计
M33 作为 Master,承担以下职责:
- 初始化共享内存:调用
rpmsg_lite_master_init() 零化 vring、建立 desc 空闲链表、填充 tx buffer 池。
- 启动握手:在
master_init 和 create_ept 全部成功后,向 0x261CF000 写入 magic(0xC0DEDA7A),告知 M55"内存已就绪"。
- Polling 线程:由于 PSoC E84 的 IPC 中断通知路径存在可靠性问题(M33 与 M55 之间的 IPC notify 有时无法送达),额外启动一个轮询线程,每隔 20ms 主动调用
virtqueue_notification() 驱动 rvq 和 tvq 的回调,确保接收和发送状态都能及时更新。
- 发送循环:每秒发送一次
uint32_t counter,使用 200ms 超时(而非 RL_DONT_BLOCK),在 tx buffer 暂时耗尽时自动等待 M55 回收。
完整代码
#include <rtthread.h>
#include <rtdevice.h>
#include <board.h>
#include "rpmsg_lite.h"
#include "virtqueue.h"
#define LED_PIN_B GET_PIN(16, 5)
#define RPMSG_LITE_SHMEM_BASE ((void *)0x261C0000)
#define RPMSG_LITE_SHMEM_SIZE (0x10000)
#define RPMSG_LITE_LINK_ID RL_PLATFORM_PSE84_M33_M55_LINK_ID
#define RPMSG_LITE_EPT_ADDR (30U)
/* Boot-handshake sentinel placed near the end of the shared region (well
* outside the area used by rpmsg-lite vrings + payload buffers). The master
* writes RPMSG_HANDSHAKE_MAGIC ONLY after master_init has zeroed the vrings,
* built the desc free-list and filled the tx-buffer pool. The remote side
* spin-waits on this magic before doing anything that touches the rings,
* otherwise it would read garbage out of un-initialized shared memory and
* crash inside virtqueue_get_available_buffer (UNALIGNED on desc[garbage]).
*/
#define RPMSG_HANDSHAKE_ADDR ((volatile uint32_t *)(0x261C0000 + 0xF000))
#define RPMSG_HANDSHAKE_MAGIC (0xC0DEDA7AU)
/* Send timeout (ms) - waits a short while if the master tx buffer pool is
* temporarily exhausted (i.e. the remote has not yet recycled buffers).
* Combined with the remote-side rx polling workaround this keeps the link
* flowing without depending on the IPC notify interrupt path. */
#define RPMSG_SEND_TIMEOUT_MS (200U)
/* Polling interval used as a workaround when the IPC notify interrupt is
* not delivered reliably between cores. On the master side we poll tvq so
* that any consumed-buffer notification path or remote replies are handled
* even if the IPC interrupt doesn't fire. */
#define RPMSG_POLL_INTERVAL_MS (20U)
static struct rpmsg_lite_instance *rpmsg_lite_dev = RL_NULL;
static struct rpmsg_lite_endpoint *rpmsg_ept = RL_NULL;
/* Master-side polling thread: drives both virtqueue callbacks directly so
* that the link works without depending on cross-core IPC interrupts. The
* tvq callback (rpmsg_lite_tx_callback) keeps link_state up to date; the
* rvq callback (rpmsg_lite_rx_callback) processes any messages from the
* remote side. */
static void rpmsg_poll_thread(void *parameter)
{
rt_kprintf("[M33] vq polling thread started, interval=%u ms\r\n",
RPMSG_POLL_INTERVAL_MS);
while (1)
{
if (rpmsg_lite_dev != RL_NULL)
{
if (rpmsg_lite_dev->rvq != RL_NULL)
{
virtqueue_notification(rpmsg_lite_dev->rvq);
}
if (rpmsg_lite_dev->tvq != RL_NULL)
{
virtqueue_notification(rpmsg_lite_dev->tvq);
}
}
rt_thread_mdelay(RPMSG_POLL_INTERVAL_MS);
}
}
static void rpmsg_master_thread(void *parameter)
{
uint32_t counter = 0;
int32_t ret;
rt_thread_t poll_tid;
rt_kprintf("[M33] Initializing rpmsg-lite master...\r\n");
/* Pre-clear the handshake sentinel BEFORE master_init runs, so the
* remote side cannot mistake stale shared memory for a "ready" signal
* after a warm reset of M33 only. */
*RPMSG_HANDSHAKE_ADDR = 0U;
__DSB();
rpmsg_lite_dev = rpmsg_lite_master_init(RPMSG_LITE_SHMEM_BASE, RPMSG_LITE_SHMEM_SIZE,
RPMSG_LITE_LINK_ID, RL_NO_FLAGS);
if (rpmsg_lite_dev == RL_NULL)
{
rt_kprintf("[M33] rpmsg_lite_master_init failed!\r\n");
return;
}
rt_kprintf("[M33] rpmsg-lite master initialized.\r\n");
rpmsg_ept = rpmsg_lite_create_ept(rpmsg_lite_dev, RPMSG_LITE_EPT_ADDR,
RL_NULL, RL_NULL);
if (rpmsg_ept == RL_NULL)
{
rt_kprintf("[M33] rpmsg_lite_create_ept failed!\r\n");
return;
}
rt_kprintf("[M33] rpmsg-lite endpoint created, addr=%d.\r\n", RPMSG_LITE_EPT_ADDR);
/* Publish the handshake magic AFTER vrings + tx pool are fully ready.
* The DSB before the store ensures all prior writes to the shared
* vrings are globally visible before the magic becomes visible. */
__DSB();
*RPMSG_HANDSHAKE_ADDR = RPMSG_HANDSHAKE_MAGIC;
__DSB();
rt_kprintf("[M33] handshake magic 0x%08x published at %p.\r\n",
(unsigned int)RPMSG_HANDSHAKE_MAGIC,
(void *)RPMSG_HANDSHAKE_ADDR);
/* Start the polling workaround */
poll_tid = rt_thread_create("rpmsg_p", rpmsg_poll_thread, RT_NULL,
1024, 9, 5);
if (poll_tid != RT_NULL)
{
rt_thread_startup(poll_tid);
}
else
{
rt_kprintf("[M33] Failed to create polling thread!\r\n");
}
/* Wait for remote side link up */
while (!rpmsg_lite_is_link_up(rpmsg_lite_dev))
{
rt_thread_mdelay(10);
}
rt_kprintf("[M33] rpmsg-lite link is up.\r\n");
while (1)
{
ret = rpmsg_lite_send(rpmsg_lite_dev, rpmsg_ept, RPMSG_LITE_EPT_ADDR,
(char *)&counter, sizeof(counter),
RPMSG_SEND_TIMEOUT_MS);
if (ret == RL_SUCCESS)
{
rt_kprintf("[M33] sent counter = %u\r\n", counter);
counter++;
}
else
{
rt_kprintf("[M33] send failed, ret=%d\r\n", ret);
}
rt_thread_mdelay(1000);
}
}
int main(void)
{
rt_thread_t tid;
rt_kprintf("Hello RT-Thread\r\n");
rt_kprintf("This core is cortex-m33\r\n");
rt_pin_mode(LED_PIN_B, PIN_MODE_OUTPUT);
tid = rt_thread_create("rpmsg_m", rpmsg_master_thread, RT_NULL,
2048, 10, 10);
if (tid != RT_NULL)
{
rt_thread_startup(tid);
}
else
{
rt_kprintf("[M33] Failed to create rpmsg thread!\r\n");
}
while (1)
{
rt_kprintf("log from cortex-m33\r\n");
rt_pin_write(LED_PIN_B, PIN_LOW);
rt_thread_mdelay(1000);
rt_pin_write(LED_PIN_B, PIN_HIGH);
rt_thread_mdelay(1000);
}
return 0;
}
五、M55 端:Remote(接收方)
关键设计
M55 作为 Remote,承担以下职责:
- 等待握手:在调用
rpmsg_lite_remote_init() 之前,先 spin 读取 0x261CF000 的 sentinel,直到看到 magic 值。这是为了等待 M33 完成 master_init 对共享内存的零化和初始化。
- 初始化与创建端点:看到握手信号后,再安全地调用
rpmsg_lite_remote_init() 和 rpmsg_lite_create_ept()。
- Rx Polling 线程:同样因为 IPC 中断不可靠,启动轮询线程每隔 20ms 调用
virtqueue_notification(rpmsg_lite_dev->rvq),主动检查是否有新消息到达。
- 接收回调:
rpmsg_rx_callback 中解析 payload(uint32_t counter)并打印。
完整代码
#include <rtthread.h>
#include <rtdevice.h>
#include <board.h>
#include "rpmsg_lite.h"
#include "virtqueue.h"
#define LED_PIN_G GET_PIN(16, 6)
#define RPMSG_LITE_SHMEM_BASE ((void *)0x261C0000)
#define RPMSG_LITE_LINK_ID RL_PLATFORM_PSE84_M33_M55_LINK_ID
#define RPMSG_LITE_EPT_ADDR (30U)
/* Boot-handshake sentinel - MUST match the master side. M55 boots/runs
* faster than the M33 NS image, so without this guard the remote would
* call rpmsg_lite_remote_init / start polling against a vring that has
* never been zeroed by the master, read garbage out of avail->idx, and
* hard-fault inside virtqueue_get_available_buffer when desc[garbage]
* is dereferenced (UNALIGNED, r0/r4 == garbage value). */
#define RPMSG_HANDSHAKE_ADDR ((volatile uint32_t *)(0x261C0000 + 0xF000))
#define RPMSG_HANDSHAKE_MAGIC (0xC0DEDA7AU)
#define RPMSG_HANDSHAKE_POLL_MS (50U)
#define RPMSG_HANDSHAKE_REPORT_MS (1000U)
/* Polling interval used as a workaround when the IPC notify interrupt is
* not delivered reliably between cores (e.g. when SRF or platform access
* permissions block the rpmsg_platform notify path). 20 ms keeps latency
* low while remaining cheap. */
#define RPMSG_POLL_INTERVAL_MS (20U)
static struct rpmsg_lite_instance *rpmsg_lite_dev = RL_NULL;
static struct rpmsg_lite_endpoint *rpmsg_ept = RL_NULL;
static int32_t rpmsg_rx_callback(void *payload, uint32_t payload_len,
uint32_t src, void *priv)
{
if (payload_len == sizeof(uint32_t))
{
uint32_t counter = *(uint32_t *)payload;
rt_kprintf("[M55] received counter = %u (from 0x%x)\r\n", counter, src);
}
else
{
rt_kprintf("[M55] received unknown data, len=%d\r\n", payload_len);
}
return RL_RELEASE;
}
/* Workaround thread: polls the receive virtqueue directly so that messages
* placed in the avail ring by the master are processed even if the IPC
* notification interrupt does not fire on this core. Calls into the same
* code path that the platform ISR would normally trigger (env_isr ->
* virtqueue_notification -> rpmsg_lite_rx_callback -> ept->rx_cb). */
static void rpmsg_rx_poll_thread(void *parameter)
{
rt_kprintf("[M55] rx polling thread started, interval=%u ms\r\n",
RPMSG_POLL_INTERVAL_MS);
while (1)
{
if ((rpmsg_lite_dev != RL_NULL) && (rpmsg_lite_dev->rvq != RL_NULL))
{
virtqueue_notification(rpmsg_lite_dev->rvq);
}
rt_thread_mdelay(RPMSG_POLL_INTERVAL_MS);
}
}
static void rpmsg_remote_thread(void *parameter)
{
rt_thread_t poll_tid;
uint32_t waited_ms = 0U;
uint32_t magic = 0U;
/* Synchronize with the master before doing anything that touches the
* shared rings. Spin (with cooperative sleep) until master_init has
* published RPMSG_HANDSHAKE_MAGIC. Without this guard we observed a
* hard fault (UNALIGNED, fault address == 0xeefa5626) in the rx-poll
* thread because virtqueue_get_available_buffer dereferenced
* desc[garbage_idx] before the master had zeroed the vring. */
rt_kprintf("[M55] waiting for master handshake at %p ...\r\n",
(void *)RPMSG_HANDSHAKE_ADDR);
for (;;)
{
__DSB();
magic = *RPMSG_HANDSHAKE_ADDR;
if (magic == RPMSG_HANDSHAKE_MAGIC)
{
break;
}
rt_thread_mdelay(RPMSG_HANDSHAKE_POLL_MS);
waited_ms += RPMSG_HANDSHAKE_POLL_MS;
if ((waited_ms % RPMSG_HANDSHAKE_REPORT_MS) == 0U)
{
rt_kprintf("[M55] still waiting for master, observed=0x%08x, "
"elapsed=%u ms\r\n",
(unsigned int)magic, (unsigned int)waited_ms);
}
}
rt_kprintf("[M55] master handshake observed after %u ms.\r\n",
(unsigned int)waited_ms);
rt_kprintf("[M55] Initializing rpmsg-lite remote...\r\n");
rpmsg_lite_dev = rpmsg_lite_remote_init(RPMSG_LITE_SHMEM_BASE,
RPMSG_LITE_LINK_ID, RL_NO_FLAGS);
if (rpmsg_lite_dev == RL_NULL)
{
rt_kprintf("[M55] rpmsg_lite_remote_init failed!\r\n");
return;
}
rt_kprintf("[M55] rpmsg-lite remote initialized.\r\n");
rpmsg_ept = rpmsg_lite_create_ept(rpmsg_lite_dev, RPMSG_LITE_EPT_ADDR,
rpmsg_rx_callback, RL_NULL);
if (rpmsg_ept == RL_NULL)
{
rt_kprintf("[M55] rpmsg_lite_create_ept failed!\r\n");
return;
}
rt_kprintf("[M55] rpmsg-lite endpoint created, addr=%d.\r\n", RPMSG_LITE_EPT_ADDR);
/* Start the rx polling workaround */
poll_tid = rt_thread_create("rpmsg_rx", rpmsg_rx_poll_thread, RT_NULL,
1024, 9, 5);
if (poll_tid != RT_NULL)
{
rt_thread_startup(poll_tid);
}
else
{
rt_kprintf("[M55] Failed to create rx polling thread!\r\n");
}
while (1)
{
rt_thread_mdelay(1000);
}
}
int main(void)
{
rt_thread_t tid;
rt_kprintf("Hello RT-Thread\r\n");
rt_kprintf("It's cortex-m55\r\n");
rt_pin_mode(LED_PIN_G, PIN_MODE_OUTPUT);
tid = rt_thread_create("rpmsg_r", rpmsg_remote_thread, RT_NULL,
2048, 10, 10);
if (tid != RT_NULL)
{
rt_thread_startup(tid);
}
else
{
rt_kprintf("[M55] Failed to create rpmsg thread!\r\n");
}
while (1)
{
rt_kprintf("log from cortex-m55\r\n");
rt_pin_write(LED_PIN_G, PIN_LOW);
rt_thread_mdelay(1000);
rt_pin_write(LED_PIN_G, PIN_HIGH);
rt_thread_mdelay(1000);
}
return 0;
}
六、编译与烧录
编译顺序与双核独立运行时相同:先 M33,后 M55。
- 编译 M33 工程,烧录
- 编译 M55 工程,烧录
- 按 RST 复位或重新上电
七、问题排查实录
在调试过程中,遇到了两个关键问题,记录如下,供读者参考。
问题 1:send failed, ret=-5001
现象:M33 成功初始化后,前两次 rpmsg_lite_send 成功,之后持续报 send failed, ret=-5001(即 RL_ERR_NO_MEM)。M55 端没有任何接收打印。
原因:
- M33 发送消息后,需要通过 IPC notify 通知 M55 有新数据
- PSoC E84 的 IPC 中断路径存在可靠性问题(可能与 SRF 配置或平台权限有关),M55 收不到中断
- M55 未触发
virtqueue_notification() -> 未调用 rpmsg_lite_rx_callback() -> buffer 未被消费和释放
- M33 的 tx buffer pool 只有
RL_BUFFER_COUNT(默认 2)个,两次发送后就耗尽
解决:在 M33 和 M55 两侧都增加 Polling 线程,绕过 IPC 中断,主动轮询 virtqueue:
/* M33 侧:轮询 rvq + tvq */
if (rpmsg_lite_dev->rvq != RL_NULL) virtqueue_notification(rpmsg_lite_dev->rvq);
if (rpmsg_lite_dev->tvq != RL_NULL) virtqueue_notification(rpmsg_lite_dev->tvq);
/* M55 侧:轮询 rvq */
if (rpmsg_lite_dev->rvq != RL_NULL) virtqueue_notification(rpmsg_lite_dev->rvq);
virtqueue_notification() 是 virtqueue.h 公开的 API,功能等价于触发一次该 vq 的 ISR 回调路径,可以在用户代码中安全调用。
问题 2:M55 hard fault(UNALIGNED,地址 0xeefa5626)
现象:M55 启动 rx polling 线程后,立即 hard fault。CFSR 显示 UNALIGNED,寄存器 r0/r4 的值都是 0xeefa5626(典型的未初始化内存垃圾值)。
根因:这是启动时序竞争(boot timing race)。
- M55 内核启动速度远快于 M33(NS)
- M55 的
rpmsg_remote_thread 在几百毫秒内就创建并启动了 polling 线程
- 但此时 M33 的
rpmsg_lite_master_init() 尚未执行,共享内存(0x261C0000 起)还是上电时的随机值
virtqueue_notification() -> virtqueue_get_available_buffer() 读取 avail->idx,得到一个垃圾值(如 0xeefa5626)
- 用该垃圾值索引
desc[],越界访问未初始化地址,触发 UNALIGNED hard fault
解决:在共享内存末端增加**启动握手(boot handshake)**机制。
M33 侧(Master):在 master_init 和 create_ept 全部成功之后,再向约定地址写入 magic:
*RPMSG_HANDSHAKE_ADDR = RPMSG_HANDSHAKE_MAGIC; /* 0xC0DEDA7A */
M55 侧(Remote):在调用 rpmsg_lite_remote_init() 之前,先 spin 等待这个 magic:
while (*RPMSG_HANDSHAKE_ADDR != RPMSG_HANDSHAKE_MAGIC)
{
rt_thread_mdelay(50);
}
这样确保 M55 第一次读取 vring 时,共享内存已经被 M33 完全初始化。
为什么不在 M55 侧直接 memset 共享内存? 因为 rpmsg_lite_master_init 除了零化内存,还要建立 desc 的 next 空闲链表、填充 used ring 等复杂结构。Remote 侧不具备这些初始化逻辑,也不能与 Master 同时写共享内存。
八、运行效果
如果一切正常,上电后串口输出应如下:
****************** PSOC Edge MCU: CM33 Secure Mode******************
PSRAM Cache is Enabled
PSRAM init successful
****************** PSOC Edge MCU: CM33 Secure Mode Exit******************
▒▒
\ | /
- RT - Thread Operating System
/ | \ 5.0.2 build May 2 2026 16:24:37
2006 - 2022 Copyright by RT-Thread team
Hello RT-Thread
This core is cortex-m33
log from cortex-m33
[M33] Initializing rpmsg-lite master...
[M33] rpmsg-lite master initialized.
[M33] rpmsg-lite endpoint created, addr=30.
[M33] handshake magic 0xc0deda7a published at 0x261cf000.
[M33] vq polling thread started, interval=20 ms
[M33] rpmsg-lite link is up.
[M33] sent counter = 0
[M55] master handshake observed after 50 ms.
[M55] Initializing rpmsg-lite remote...
[M55] rpmsg-lite remote initialized.
[M55] rpmsg-lite endpoint created, addr=30.
[M55] rx polling thread started, interval=20 ms
[M55] received counter = 0 (from 0x1e)
[M33] sent counter = 1
[M55] received counter = 1 (from 0x1e)
log from cortex-m55
log from cortex-m33
[M33] sent counter = 2
[M55] received counter = 2 (from 0x1e)
[M33] sent counter = 3
[M55] received counter = 3 (from 0x1e)
log from cortex-m55
log from cortex-m33
[M33] sent counter = 4
[M55] received counter = 4 (from 0x1e)
[M33] sent counter = 5
[M55] received counter = 5 (from 0x1e)

九、总结
本文在双核独立运行的基础上,成功实现了 M33 -> M55 的单向 rpmsg-lite 通信。核心经验如下:
| 要点 |
说明 |
| 启动顺序 |
Secure M33 -> M33(NS,使能 CM55 + master_init)-> M55(Remote) |
| IPC 不可靠 |
PSoC E84 的跨核 IPC notify 不够稳定,需用 polling 线程绕过 |
| 时序竞争 |
M55 启动快于 M33,必须用 sentinel 握手机制避免读取未初始化共享内存 |
| 只改用户层 |
所有 workaround 均放在 applications/main.c,不动 packages/rpmsg-lite/ |
| 超时发送 |
Master 侧发送使用 RPMSG_SEND_TIMEOUT_MS 而非 RL_DONT_BLOCK,配合 polling 自动恢复 |
后续可在此基础上扩展更复杂的双核交互,例如 M55 回复 ACK、使用 rpmsg 命名服务(rpmsg_ns)动态发现端点等。
十、参考资料
| 序号 |
资源 |
说明 |
| 1 |
RT-Thread 软件包 - rpmsg-lite |
官方软件包页面,包含完整的中文文档:组件优势、配置选项、API 用法 |
| 2 |
GitHub - flyingcys/rpmsg-lite |
NXP RPMsg-Lite 在 RT-Thread 生态中的 fork,源码与上游同步 |
| 3 |
RT-Thread 问答社区 - rpmsg-lite 多核通信原理分析 |
详尽的通信流程图解:初始化、发送、接收的完整调用链 |
| 4 |
微信公众号 - 基于 RT-Thread 的 RPMsg-Lite 异构多核通信原理分析 |
与资料 3 同源,便于手机端阅读 |
rpmsg-lite 核心知识点摘要
以下整理自上述参考资料,供快速查阅:
1. 设计动机
RPMsg-Lite 由 NXP 开发,相比 OpenAMP 的 RPMsg 实现,优势在于更小的代码体积和更简化的 API。对于资源受限的 Cortex-M 系列 MCU,这是更优选择:
| 组件/配置 |
Flash (B) |
RAM (B) |
| OpenAMP RPMsg |
5547 |
456 + 动态分配 |
| RPMsg-Lite / 动态 API |
3462 |
56 + 动态分配 |
| RPMsg-Lite / 静态 API(无 malloc) |
2926 |
352 |
2. 发送流程
rpmsg_lite_send
-> rpmsg_lite_format_message
-> virtqueue_kick
-> vq_ring_notify_host
-> virtqueue_notify
-> platform_notify (触发 IPC 中断通知对端)
对端收到 IPC 中断后:
env_isr
-> virtqueue_notification
-> rpmsg_lite_rx_callback
-> ept->rx_cb (用户注册的回调函数)
3. 接收流程(基于回调)
在 RTOS 环境下,推荐的方式是为每个 endpoint 注册独立的 rx_cb 回调。当对端发送数据时:
platform_notify() 触发本核 IPC 中断
- ISR 中调用
env_isr(),进而调用 virtqueue_notification()
virtqueue_notification() 触发 rpmsg_lite_rx_callback()
- 在
rpmsg_lite_rx_callback() 中遍历 endpoint 列表,找到匹配的 ept->rx_cb 并调用
- 用户在回调中处理接收到的数据,返回
RL_RELEASE 释放 buffer 或 RL_HOLD 持有 buffer
4. 关键配置选项
| 配置项 |
默认值 |
说明 |
RL_BUFFER_COUNT |
2 |
共享内存中 buffer 数量,必须是 2 的幂 |
RL_BUFFER_PAYLOAD_SIZE |
496 |
单个 buffer 有效载荷大小,必须是 2^n - 16 |
RL_API_HAS_ZEROCOPY |
1 |
启用零拷贝 API(rpmsg_lite_send_nocopy 等) |
RL_USE_STATIC_API |
0 |
静态 API 开关(禁用动态内存分配) |
RL_ALLOW_CONSUMED_BUFFERS_NOTIFICATION |
0 |
每次消费 buffer 后是否通知对端 |
5. 共享内存要求
- 必须配置为 Non-Cacheable(不可缓存) 内存
- 建议在链接器脚本中定义专用的共享内存段
- 必须确保应用的其他部分不会意外访问该区域
6. 关于 virtqueue_notification()
这是 virtqueue.h 中公开的 API(非内部函数),其本质是直接调用 vq->callback_fc(vq)。在本教程的 Polling workaround 中,正是利用这一特性,在用户线程中主动驱动 vq 回调,从而绕过不可靠的 IPC 中断路径。