目錄

說(shuō)明

初識(shí)rpmsg-lite

rpmsg-lite組件優(yōu)勢(shì)

rpmsg-lite工程架構(gòu)

rpmsg-lite通信流程（RTOS）

rpmsg-lite通信流程（MCMGR）

1 說(shuō)明

本文檔旨在說(shuō)明如何在RT-Thread(運(yùn)行于Cortex-M85核)和裸機(jī)程序(運(yùn)行于Cortex-M33核)之間使用rpmsg-lite進(jìn)行通信，并采用MCMGR組件替代rpmsg-lite原生的隊(duì)列機(jī)制實(shí)現(xiàn)雙核間同步管理。

硬件平臺(tái)：Renesas RA8P1( ARM Cortex-M85 & ARM Cortex-M33)

或復(fù)制鏈接購(gòu)買：https://item.taobao.com/item.htm?ft=t&id=987791181903

軟件環(huán)境

M85核：RT-Thread操作系統(tǒng)

M33核：裸機(jī)程序

通信組件：rpmsg-lite + MCMGR

2 初識(shí)rpmsg-lite

RPMsg-Lite 是一款輕量級(jí)開(kāi)源異構(gòu)處理器通信框架，適用于多核系統(tǒng)中的小型微控制器（MCU）。

它支持在非對(duì)稱多處理（AMP）配置中，例如運(yùn)行 Linux 的主處理器與運(yùn)行實(shí)時(shí)操作系統(tǒng)（如 RT-Thread）的協(xié)處理器之間進(jìn)行高效的消息傳遞。

RPMsg-Lite 針對(duì)資源受限環(huán)境設(shè)計(jì)，相比 OpenAMP 提供更簡(jiǎn)化的 API，支持零拷貝和靜態(tài)內(nèi)存選項(xiàng)，并確保與 RPMsg 協(xié)議兼容。通過(guò)使用共享內(nèi)存進(jìn)行數(shù)據(jù)交換，要求配置非緩存內(nèi)存，非常適合卸載 CPU 密集型任務(wù)。

3 rpmsg-lite組件優(yōu)勢(shì)

1）靜態(tài)API創(chuàng)建

相比較RPMsg，RPMsg-Lite采用靜態(tài)API創(chuàng)建方式，相比較動(dòng)態(tài)分配，靜態(tài)API創(chuàng)建的方式至少減少了5KB的代碼體量，也避免了動(dòng)態(tài)內(nèi)存管理開(kāi)銷。

2）基于共享內(nèi)存的數(shù)據(jù)零拷貝

相比較傳統(tǒng)驅(qū)動(dòng)外設(shè)來(lái)說(shuō)，都需要經(jīng)歷由應(yīng)用層->驅(qū)動(dòng)層->硬件緩沖區(qū)這么多次拷貝的過(guò)程，相對(duì)應(yīng)的也會(huì)帶來(lái)一定的延遲；而RPMsg-Lite則是數(shù)據(jù)零拷貝的這么一種實(shí)現(xiàn)方式，也就是應(yīng)用層->共享內(nèi)存方式，僅一次的內(nèi)存拷貝，將數(shù)據(jù)交換周期能夠最小降低至μs微秒級(jí)

（當(dāng)然這部分依賴于雙核硬件平臺(tái)的IPC機(jī)制，即處理器間通信）

3）硬中斷的高效觸發(fā)

對(duì)于傳統(tǒng)共享中斷來(lái)說(shuō)，都是需要輪詢所有關(guān)聯(lián)外設(shè)的狀態(tài)寄存器；而RPMsg-Lite直接使用 platform_notify 來(lái)主動(dòng)觸發(fā)核間中斷通知，同時(shí)這個(gè)觸發(fā)時(shí)機(jī)可以由用戶決定，比如說(shuō)累計(jì)n個(gè)消息后觸發(fā)中斷，以此來(lái)批量處理核間通信。

4 rpmsg-lite工程架構(gòu)

rpmsg-lite工程分為兩個(gè)版本：

RT-Thread + Freertos + RPMsg-Lite：該示例統(tǒng)一使用RPMsg-Lite結(jié)合RTOS的消息隊(duì)列來(lái)實(shí)現(xiàn)子組件queue

RT-Thread + Bera metal + RPMsg-Lite + MCMGR：為了方便裸機(jī)系統(tǒng)能夠?qū)崟r(shí)同步主次核之間的狀態(tài)，移植了官方的MCMGR組件（multiple core manager）

5 rpmsg-lite通信流程（RTOS）

需要說(shuō)明的是，裸機(jī)rpmsg-lite與RTOS版的rpmsg-lite的收發(fā)邏輯是不一樣的，這里分開(kāi)說(shuō)明

這里繪制了一張圖簡(jiǎn)單說(shuō)明下，首先明確上層留給用戶的rpmsg-lite api，最為關(guān)鍵的就是rpmsg-dev的初始化、消息的發(fā)送與接收：

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)

這一步會(huì)分別在master和remote初始化一個(gè) rpmsg-dev 實(shí)體，并且聲明共享內(nèi)存起始地址，這個(gè)過(guò)程中會(huì)綁定兩個(gè)virtualqueue的回調(diào)函數(shù)（收發(fā)關(guān)鍵）

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);

這里最終會(huì)傳遞給 virtqueue 的初始化創(chuàng)建，并作為 callback_fc 參數(shù)進(jìn)行綁定：

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

與此同時(shí)，virtqueue_notify() 函數(shù)會(huì)作為第五個(gè)參數(shù)綁定為通知函數(shù)，這一步對(duì)接到移植層的子 layer：enviroment layer，也就是 platform_notify()，用于觸發(fā)通知另外一個(gè)核心，一般使用核間中斷，基于此，就可以觸發(fā)另一個(gè)核心來(lái)處理接收數(shù)據(jù)的邏輯了：

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);}

注意：

補(bǔ)充一點(diǎn)，為了方便多核心間的通信，這里的virtualqueue默認(rèn)都是配置為兩個(gè)，并且雙核間的收發(fā)virtualqueue都是分別一一對(duì)應(yīng)的：

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

2) 發(fā)送數(shù)據(jù)

首先來(lái)看下 rpmsg-lite 的發(fā)送數(shù)據(jù)API：

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

這里的data是我們傳輸過(guò)來(lái)的數(shù)據(jù)，那么后面就是這么幾個(gè)步驟：

# 1.從共享內(nèi)存申請(qǐng)buffer（vq_tx_master）buffer = rpmsg_lite_dev->vq_ops->vq_tx_alloc(rpmsg_lite_dev->tvq, &buff_len, &idx);# 2.將共享地址傳遞給rpmsg_msg，并對(duì)該變量傳遞實(shí)際數(shù)據(jù)rpmsg_msg = (structrpmsg_std_msg *)buffer;env_memcpy(rpmsg_msg->data, data, size);# 3.將前面申請(qǐng)的這段buffer標(biāo)記為可用模式，并加入到vq_ring.avail中，一般這里應(yīng)該是用于緩存下的多核通信/* Enqueue buffer on virtqueue. */rpmsg_lite_dev->vq_ops->vq_tx(rpmsg_lite_dev->tvq, buffer, buff_len, idx);

那么到這里，我們想要傳遞的數(shù)據(jù)就已經(jīng)放進(jìn)去 virtualqueue了，那么接下來(lái)就是再發(fā)送一個(gè)通知給另外一個(gè)核心，通知它從共享隊(duì)列中讀取數(shù)據(jù)。對(duì)應(yīng)的流程如下：

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

3) 接收數(shù)據(jù)

首先明確接收數(shù)據(jù)的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)

深入進(jìn)去就是需要從消息隊(duì)列中拿出數(shù)據(jù)

可以觀察上圖中紅色箭頭所指示的流程，在接收到來(lái)自另一個(gè)核心的IPC通知后，會(huì)觸發(fā)進(jìn)入 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();}

此處會(huì)告知本核心的virtualqueue，并觸發(fā)對(duì)應(yīng)的回調(diào)函數(shù)（接收為rpmsg_lite_rx_callback），在這個(gè)回調(diào)函數(shù)中，會(huì)執(zhí)行到一個(gè) ept->rx_cb，這里也就是我們?cè)诔跏蓟痳pmsg-lite時(shí)綁定的endpoint回調(diào)函數(shù)，即 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 的消息隊(duì)列線程，進(jìn)而去處理消息。

6 rpmsg-lite通信流程（MCMGR）

MCMGR組件開(kāi)源地址：https://github.com/nxp-mcuxpresso/mcux-mcmgr

1）mcmgr雙核間同步

# M85# 首先注冊(cè)一個(gè)event回調(diào)函數(shù)，實(shí)時(shí)檢測(cè)來(lái)自次核的通知，也就是 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# 通過(guò)MCMGR_TriggerEvent觸發(fā)IPC中斷，發(fā)送通知給主核 /* 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>接收

接收邏輯關(guān)鍵取決于下面這個(gè)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)

通過(guò)綁定的回調(diào)函數(shù)rx_cb，再由mcmgr多核管理器統(tǒng)一注冊(cè)事件，并根據(jù)平臺(tái)層提供的platform_notify機(jī)制觸發(fā)env_isr，并觸發(fā)到mcmgr_ipc_callback

來(lái)執(zhí)行接收回調(diào)rx_cb，而這個(gè)觸發(fā)由rpmsg_lite_send提供觸發(fā)動(dòng)作（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>發(fā)送

發(fā)送邏輯關(guān)鍵取決于下面這個(gè)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是我們傳輸過(guò)來(lái)的數(shù)據(jù)，那么后面就是這么幾個(gè)步驟：# 1.從共享內(nèi)存申請(qǐng)buffer（vq_tx_master）buffer = rpmsg_lite_dev->vq_ops->vq_tx_alloc(rpmsg_lite_dev->tvq, &buff_len, &idx);# 2.將共享地址傳遞給rpmsg_msg，并對(duì)該變量傳遞實(shí)際數(shù)據(jù)rpmsg_msg = (structrpmsg_std_msg *)buffer;env_memcpy(rpmsg_msg->data, data, size);# 3.將前面申請(qǐng)的這段buffer標(biāo)記為可用模式，并加入到vq_ring.avail中，一般這里應(yīng)該是用于緩存下的多核通信/* Enqueue buffer on virtqueue. */rpmsg_lite_dev->vq_ops->vq_tx(rpmsg_lite_dev->tvq, buffer, buff_len, idx);# 最后執(zhí)行通知機(jī)制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 開(kāi)源倉(cāng)庫(kù)，歡迎撒個(gè)星(Star)支持，更期待你的代碼貢獻(xiàn)：https://github.com/RT-Thread/rt-thread

獲取硬件

RT-Thread 與瑞薩電子聯(lián)合推出 RA8P1 Titan Board，基于 1GHz Arm Cortex-M85 + 250MHz Cortex-M33 雙核架構(gòu)，集成 Ethos-U55 NPU ，實(shí)現(xiàn) 256 GOPS 的 AI 性能、超過(guò) 7300 CoreMarks 的突破性 CPU 性能和先進(jìn)的人工智能 (AI) 功能，可支持語(yǔ)音、視覺(jué)和實(shí)時(shí)分析 AI 場(chǎng)景!