[PSOC Edge] 【英飞凌 Edgi Talk评测】6. 双核通信 —— 基于 rpmsg-lite

[复制链接]
60|0
HonestQiao 发表于 2026-5-2 21:45 | 显示全部楼层 |阅读模式
双核, , , , rpmsg-lite,
[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 软件包:

  1. 打开 RT-Thread Settings
  2. 软件包 中搜索并添加 rpmsg-lite
  3. 保存配置,系统会自动下载并集成到 packages/rpmsg-lite/ 目录

两个工程都需要添加 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 线程的位置,以及启动握手的时序关系:

图片上传失败:rpmsg-lite双核通信架构图

四、M33 端:Master(发送方)

关键设计

M33 作为 Master,承担以下职责:

  1. 初始化共享内存:调用 rpmsg_lite_master_init() 零化 vring、建立 desc 空闲链表、填充 tx buffer 池。
  2. 启动握手:在 master_initcreate_ept 全部成功后,向 0x261CF000 写入 magic(0xC0DEDA7A),告知 M55"内存已就绪"。
  3. Polling 线程:由于 PSoC E84 的 IPC 中断通知路径存在可靠性问题(M33 与 M55 之间的 IPC notify 有时无法送达),额外启动一个轮询线程,每隔 20ms 主动调用 virtqueue_notification() 驱动 rvq 和 tvq 的回调,确保接收和发送状态都能及时更新。
  4. 发送循环:每秒发送一次 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,承担以下职责:

  1. 等待握手:在调用 rpmsg_lite_remote_init() 之前,先 spin 读取 0x261CF000 的 sentinel,直到看到 magic 值。这是为了等待 M33 完成 master_init 对共享内存的零化和初始化。
  2. 初始化与创建端点:看到握手信号后,再安全地调用 rpmsg_lite_remote_init()rpmsg_lite_create_ept()
  3. Rx Polling 线程:同样因为 IPC 中断不可靠,启动轮询线程每隔 20ms 调用 virtqueue_notification(rpmsg_lite_dev->rvq),主动检查是否有新消息到达。
  4. 接收回调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

  1. 编译 M33 工程,烧录
  2. 编译 M55 工程,烧录
  3. 按 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_initcreate_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 回调。当对端发送数据时:

  1. platform_notify() 触发本核 IPC 中断
  2. ISR 中调用 env_isr(),进而调用 virtqueue_notification()
  3. virtqueue_notification() 触发 rpmsg_lite_rx_callback()
  4. rpmsg_lite_rx_callback() 中遍历 endpoint 列表,找到匹配的 ept->rx_cb 并调用
  5. 用户在回调中处理接收到的数据,返回 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 中断路径。

您需要登录后才可以回帖 登录 | 注册

本版积分规则

52

主题

140

帖子

2

粉丝
快速回复 在线客服 返回列表 返回顶部
0