基于RT-Thread的RPMsg-Lite异构多核通信原理分析 | 技术集结

目录

说明

初识rpmsg-lite

rpmsg-lite组件优势

rpmsg-lite工程架构

rpmsg-lite通信流程（RTOS）

rpmsg-lite通信流程（MCMGR）

1 说明

本文档旨在说明如何在RT-Thread(运行于Cortex-M85核)和裸机程序(运行于Cortex-M33核)之间使用rpmsg-lite进行通信，并采用MCMGR组件替代rpmsg-lite原生的队列机制实现双核间同步管理。

硬件平台：Renesas RA8P1( ARM Cortex-M85 & ARM Cortex-M33)

或复制链接购买：https://item.taobao.com/item.htm?ft=t&id=987791181903

软件环境

M85核：RT-Thread操作系统

M33核：裸机程序

通信组件：rpmsg-lite + MCMGR

2 初识rpmsg-lite

RPMsg-Lite 是一款轻量级开源异构处理器通信框架，适用于多核系统中的小型微控制器（MCU）。

它支持在非对称多处理（AMP）配置中，例如运行 Linux 的主处理器与运行实时操作系统（如 RT-Thread）的协处理器之间进行高效的消息传递。

RPMsg-Lite 针对资源受限环境设计，相比 OpenAMP 提供更简化的 API，支持零拷贝和静态内存选项，并确保与 RPMsg 协议兼容。通过使用共享内存进行数据交换，要求配置非缓存内存，非常适合卸载 CPU 密集型任务。

3 rpmsg-lite组件优势

1）静态API创建

相比较RPMsg，RPMsg-Lite采用静态API创建方式，相比较动态分配，静态API创建的方式至少减少了5KB的代码体量，也避免了动态内存管理开销。

2）基于共享内存的数据零拷贝

相比较传统驱动外设来说，都需要经历由应用层->驱动层->硬件缓冲区这么多次拷贝的过程，相对应的也会带来一定的延迟；而RPMsg-Lite则是数据零拷贝的这么一种实现方式，也就是应用层->共享内存方式，仅一次的内存拷贝，将数据交换周期能够最小降低至μs微秒级

（当然这部分依赖于双核硬件平台的IPC机制，即处理器间通信）

3）硬中断的高效触发

对于传统共享中断来说，都是需要轮询所有关联外设的状态寄存器；而RPMsg-Lite直接使用 platform_notify 来主动触发核间中断通知，同时这个触发时机可以由用户决定，比如说累计n个消息后触发中断，以此来批量处理核间通信。

4 rpmsg-lite工程架构

rpmsg-lite工程分为两个版本：

RT-Thread + Freertos + RPMsg-Lite：该示例统一使用RPMsg-Lite结合RTOS的消息队列来实现子组件queue

RT-Thread + Bera metal + RPMsg-Lite + MCMGR：为了方便裸机系统能够实时同步主次核之间的状态，移植了官方的MCMGR组件（multiple core manager）

5 rpmsg-lite通信流程（RTOS）

需要说明的是，裸机rpmsg-lite与RTOS版的rpmsg-lite的收发逻辑是不一样的，这里分开说明

这里绘制了一张图简单说明下，首先明确上层留给用户的rpmsg-lite api，最为关键的就是rpmsg-dev的初始化、消息的发送与接收：

1) 初始化

# masterstructrpmsg_lite_instance*rpmsg_lite_master_init(void*shmem_addr, size_tshmem_length, uint32_tlink_id, uint32_tinit_flags)# remotestructrpmsg_lite_instance*rpmsg_lite_remote_init(void*shmem_addr,uint32_tlink_id,uint32_tinit_flags)

这一步会分别在master和remote初始化一个 rpmsg-dev 实体，并且声明共享内存起始地址，这个过程中会绑定两个virtualqueue的回调函数（收发关键）

callback[0] = rpmsg_lite_rx_callback;

callback[1] = rpmsg_lite_tx_callback;

structrpmsg_lite_endpoint*rpmsg_lite_create_ept(structrpmsg_lite_instance *rpmsg_lite_dev, uint32_taddr, rl_ept_rx_cb_trx_cb, void*rx_cb_data);

这里最终会传递给 virtqueue 的初始化创建，并作为 callback_fc 参数进行绑定：

int32_tvirtqueue_create(uint16_tid, constchar *name, structvring_alloc_info *ring, void(*callback_fc)(structvirtqueue *vq), void(*notify_fc)(structvirtqueue *vq), structvirtqueue **v_queue)

与此同时，virtqueue_notify() 函数会作为第五个参数绑定为通知函数，这一步对接到移植层的子 layer：enviroment layer，也就是 platform_notify()，用于触发通知另外一个核心，一般使用核间中断，基于此，就可以触发另一个核心来处理接收数据的逻辑了：

voidplatform_notify(uint32_tvector_id){ env_lock_mutex(platform_lock); R_IPC_MessageSend(&g_ipc1_ctrl, (uint32_t)(RL_GET_Q_ID(vector_id))); env_unlock_mutex(platform_lock);}

注意：

补充一点，为了方便多核心间的通信，这里的virtualqueue默认都是配置为两个，并且双核间的收发virtualqueue都是分别一一对应的：

Master Remotevqx[0]:tx_vq -> rx_vqvqx[0]:rx_vq -> tx_vq

2) 发送数据

首先来看下 rpmsg-lite 的发送数据API：

int32_trpmsg_lite_send(structrpmsg_lite_instance *rpmsg_lite_dev, structrpmsg_lite_endpoint *ept, uint32_tdst, char*data, uint32_tsize, uintptr_ttimeout)

这里的data是我们传输过来的数据，那么后面就是这么几个步骤：

# 1.从共享内存申请buffer（vq_tx_master）buffer = rpmsg_lite_dev->vq_ops->vq_tx_alloc(rpmsg_lite_dev->tvq, &buff_len, &idx);# 2.将共享地址传递给rpmsg_msg，并对该变量传递实际数据rpmsg_msg = (structrpmsg_std_msg *)buffer;env_memcpy(rpmsg_msg->data, data, size);# 3.将前面申请的这段buffer标记为可用模式，并加入到vq_ring.avail中，一般这里应该是用于缓存下的多核通信/* Enqueue buffer on virtqueue. */rpmsg_lite_dev->vq_ops->vq_tx(rpmsg_lite_dev->tvq, buffer, buff_len, idx);

那么到这里，我们想要传递的数据就已经放进去 virtualqueue了，那么接下来就是再发送一个通知给另外一个核心，通知它从共享队列中读取数据。对应的流程如下：

>>> rpmsg_lite_send -> rpmsg_lite_format_message -> virtqueue_kick -> vq_ring_notify_host -> virtqueue_notify -> platform_notify

3) 接收数据

首先明确接收数据的api，在RTOS下我们使用的是 rpmsg_queue_recv()：

int32_trpmsg_queue_recv(structrpmsg_lite_instance *rpmsg_lite_dev, rpmsg_queue_handle q, uint32_t*src, char*data, uint32_tmaxlen, uint32_t*len, uintptr_ttimeout)

深入进去就是需要从消息队列中拿出数据

可以观察上图中红色箭头所指示的流程，在接收到来自另一个核心的IPC通知后，会触发进入 env_isr：

voidg_ipc0_callback(ipc_callback_args_t *p_args){ rt_interrupt_enter(); /* Check for message received event */ if(IPC_EVENT_MESSAGE_RECEIVED & p_args->event) { env_isr(p_args->message); } rt_interrupt_leave();}

此处会告知本核心的virtualqueue，并触发对应的回调函数（接收为rpmsg_lite_rx_callback），在这个回调函数中，会执行到一个 ept->rx_cb，这里也就是我们在初始化rpmsg-lite时绑定的endpoint回调函数，即 rpmsg_queue_rx_cb：

my_ept =rpmsg_lite_create_ept(my_rpmsg, LOCAL_EPT_ADDR, rpmsg_queue_rx_cb, my_queue);int32_trpmsg_queue_rx_cb(void*payload,uint32_tpayload_len,uint32_tsrc,void*priv){ rpmsg_queue_rx_cb_data_tmsg; RL_ASSERT(priv != RL_NULL); msg.data = payload; msg.len = payload_len; msg.src = src; /* if message is successfully added into queue then hold rpmsg buffer */ if(0!=env_put_queue(priv, &msg,0)) { /* hold the rx buffer */ returnRL_HOLD; } returnRL_RELEASE;}

接着就是激活 rpmsg-lite 的消息队列线程，进而去处理消息。

6 rpmsg-lite通信流程（MCMGR）

MCMGR组件开源地址：https://github.com/nxp-mcuxpresso/mcux-mcmgr

1）mcmgr双核间同步

# M85# 首先注册一个event回调函数，实时检测来自次核的通知，也就是 APP_RPMSG_READY_EVENT_DATA 和 APP_RPMSG_EP_READY_EVENT_DATA事件staticvoidRPMsgRemoteReadyEventHandler(uint16_t eventData,void*context){ uint16_t *data = (uint16_t *)context; *data = eventData;}/* Register the application event before starting the secondary core */ (void)MCMGR_RegisterEvent(kMCMGR_RemoteApplicationEvent, RPMsgRemoteReadyEventHandler, (void*)&RPMsgRemoteReadyEventData); while(APP_RPMSG_READY_EVENT_DATA != RPMsgRemoteReadyEventData) { }; /* Wait until the secondary core application signals the rpmsg remote endpoint has been created. */ while(APP_RPMSG_EP_READY_EVENT_DATA != RPMsgRemoteReadyEventData) { }--------------------------------------------------------------------------------------------------------------------------------# M33# 通过MCMGR_TriggerEvent触发IPC中断，发送通知给主核 /* Signal the other core we are ready by triggering the event and passing the APP_RPMSG_READY_EVENT_DATA */ (void)MCMGR_TriggerEvent(kMCMGR_RemoteApplicationEvent, APP_RPMSG_READY_EVENT_DATA); /* Signal the other core the endpoint has been created by triggering the event and passing the * APP_RPMSG_READY_EP_EVENT_DATA */ (void)MCMGR_TriggerEvent(kMCMGR_RemoteApplicationEvent, APP_RPMSG_EP_READY_EVENT_DATA);

2）mcmgr双核间通信

<1>接收

接收逻辑关键取决于下面这个API：

structrpmsg_lite_endpoint*rpmsg_lite_create_ept(structrpmsg_lite_instance *rpmsg_lite_dev, uint32_taddr, rl_ept_rx_cb_trx_cb, void*rx_cb_data, structrpmsg_lite_ept_static_context *ept_context)

通过绑定的回调函数rx_cb，再由mcmgr多核管理器统一注册事件，并根据平台层提供的platform_notify机制触发env_isr，并触发到mcmgr_ipc_callback

来执行接收回调rx_cb，而这个触发由rpmsg_lite_send提供触发动作（virtqueue_kick）

# 示例代码staticint32_tmy_ept_read_cb(void*payload,uint32_tpayload_len,uint32_tsrc,void*priv){ int32_t*has_received = priv; if(payload_len <= sizeof(THE_MESSAGE))    {        (void)rt_memcpy((void *)&msg, payload, payload_len);        *has_received = 1;    }    (void)rt_kprintf("Primary core received a msg\r\n");    (void)rt_kprintf("Message: Size=%x, DATA = %i\r\n", payload_len, msg.DATA);    return RL_RELEASE;}my_ept = rpmsg_lite_create_ept(my_rpmsg, LOCAL_EPT_ADDR, my_ept_read_cb, (void *)&has_received, &my_ept_context);

所以流程为：

mcmgr_ipc_callback ->mcmgr_event_handler ->env_isr ->virtqueue_notification ->rpmsg_lite_rx_callback(vq->callback_fc) ->my_ept_read_cb(ept->rx_cb)

<2>发送

发送逻辑关键取决于下面这个API：

int32_trpmsg_lite_send(structrpmsg_lite_instance *rpmsg_lite_dev, structrpmsg_lite_endpoint *ept, uint32_tdst, char*data, uint32_tsize, uintptr_ttimeout)

流程为

(void)rpmsg_lite_send(my_rpmsg, my_ept, REMOTE_EPT_ADDR, (char*)&msg,sizeof(THE_MESSAGE), RL_DONT_BLOCK);returnrpmsg_lite_format_message(rpmsg_lite_dev, ept->addr, dst, data, size, RL_NO_FLAGS, timeout);# 这里的data是我们传输过来的数据，那么后面就是这么几个步骤：# 1.从共享内存申请buffer（vq_tx_master）buffer = rpmsg_lite_dev->vq_ops->vq_tx_alloc(rpmsg_lite_dev->tvq, &buff_len, &idx);# 2.将共享地址传递给rpmsg_msg，并对该变量传递实际数据rpmsg_msg = (structrpmsg_std_msg *)buffer;env_memcpy(rpmsg_msg->data, data, size);# 3.将前面申请的这段buffer标记为可用模式，并加入到vq_ring.avail中，一般这里应该是用于缓存下的多核通信/* Enqueue buffer on virtqueue. */rpmsg_lite_dev->vq_ops->vq_tx(rpmsg_lite_dev->tvq, buffer, buff_len, idx);# 最后执行通知机制virtqueue_kick ->vq_ring_notify_host(vq->notify_fc(vq)) ->platform_nofity -> (void)MCMGR_TriggerEventForce(kMCMGR_RemoteRPMsgEvent, (uint16_t)RL_GET_Q_ID(vector_id)); ->trig ipc callback ->mcmgr_ipc_callback

RT-Thread Github 开源仓库，欢迎撒个星(Star)支持，更期待你的代码贡献：https://github.com/RT-Thread/rt-thread

获取硬件

RT-Thread 与瑞萨电子联合推出 RA8P1 Titan Board，基于 1GHz Arm Cortex-M85 + 250MHz Cortex-M33 双核架构，集成 Ethos-U55 NPU ，实现 256 GOPS 的 AI 性能、超过 7300 CoreMarks 的突破性 CPU 性能和先进的人工智能 (AI) 功能，可支持语音、视觉和实时分析 AI 场景!