前言：

在多核異構架構成為工業SoC主流方案的當下，芯馳D9360憑借其Cortex-A55與Cortex-R5的協同設計，在需同時處理復雜應用與實時任務的場景中表現突出。本文將以眺望電子Core-D9360平臺為例，詳解如何利用RPMSG與VirtIO機制實現A核與R核間的可靠通信，并提供關鍵代碼實現與調試方法。

圖 1 Core-D9360 核心板

一、通信基礎：RPMSG框架

RPMSG（Remote Processor Messaging）是Linux內核中用于處理器間通信的框架，采用virtio + rpmsg + IPCC三層架構實現：

●virtio：提供虛擬化隊列管理

●rpmsg：實現消息封裝與路由

●IPCC：硬件中斷控制器

在Core-D9360核心板中，A55核運行Linux，R5核運行RTOS，兩者通過虛擬通道（VirtIO ring buffer）進行數據傳輸，支持雙向、異步、多通道通信。

圖 2 核間通訊架構

二、核心代碼實現

2.1添加例程驅動

RTOS系統的例程添加在SDK/source/ssdk/examples/kunlun/drivers下，分別創建Kconfig和rules.mk文件，其中Kconfig文件內容如下：

rules.mk文件內容如下：

返回上一層目錄下，在Kconfig下添加talowe驅動模塊的初始狀態和圖形界面類型，添加內容如下：

在rules.mk文件下，添加如下內容：

MODULES += $(LOCAL_DIR)/talowe: 如果配置選項 CONFIG_TALOWE_TEST為 "y"，則會將$(LOCAL_DIR)/talowe添加到變量MODULES中。當啟用了"talowe test"功能時，會將$(LOCAL_DIR)/talowe模塊添加到模塊列表中。

2.2R5核發送機制實現

將rpmsg目錄下的rpmsg_test.c文件復制到talowe.c文件中，該驅動描述如何通過RPMSG協議進行核間通信。我們可以在此程序基礎上加入自己的協議。

以下是R5核發送"Talowe:RtoA"字符串到A55核，并打印出接收到的信息的函數。

voidsend_receive_string(rpmsg_channel_tchannel,constchar*message){intmax_payload;structdcf_ccm_hdr*snd_pkg;structdcf_ccm_hdr*rcv_pkg;status_tret;
max_payload =rpmsg_channel_max_payload(channel);snd_pkg =osAlloc(max_payload);rcv_pkg =osAlloc(max_payload);ASSERT(snd_pkg && rcv_pkg);
memset(snd_pkg,0x0, max_payload);memset(rcv_pkg,0x0, max_payload);
snd_pkg->dmsg.msg_type = COMM_MSG_CCM_ECHO;snd_pkg->dmsg.opflags |= DCF_MSGF_TMS;snd_pkg->dmsg.msg_len =sizeof(structdcf_ccm_hdr) -sizeof(structdcf_message);
strncpy(snd_pkg->data, message, max_payload -sizeof(structdcf_ccm_hdr));
ret =rpmsg_channel_send(channel, RPMSG_ECHO_EPT_ADDR, snd_pkg, max_payload,3000);if(ret == RPMSG_SUCCESS) {printf("Sent: %s\n", message);}else{printf("Failed to send message\n");}intreceived_len;ret =rpmsg_channel_recv(channel, rcv_pkg, max_payload, &received_len,3000);if(ret == RPMSG_SUCCESS) {printf("Received: %s\n", rcv_pkg->data);}else{printf("Failed to receive message\n");}osFree(snd_pkg);osFree(rcv_pkg);}

將以上函數添加至talowe.c文件內，并在主函數rpmsg_test編寫調用該函數條件，如下：

constchar* message_to_send ="Talowe:RtoA";
staticintrpmsg_test(intargc,char*argv[]){ inttest_case =-1; inttype, rproc, times; ...... elseif(!strcmp(argv[0],"perf")) { test_case =3; type =atoi(argv[1]); rproc =atoi(argv[2]); times =atoi(argv[3]); } elseif(!strcmp(argv[0],"send")) {test_case =4;type =atoi(argv[1]);rproc =atoi(argv[2]);times =atoi(argv[3]); } else{ printf("Unknown cmd %s\n", argv[0]); gotoexit; } ...... ...... case3: do_rpmsg_perf_test(channel, times); break;
case4: send_receive_string(channel, message_to_send);break;
default: printf("Unknown case %d\n", test_case); break; }

加入串口打印提示：

staticvoid rpmsg_test_show(void){ printf("\nList rpmsg communicate with remote:\n"); printf("\tType\t\t\tRemote-Proc\n"); ...... ...... printf("\trpmsg_test perf \n"); printf("\trpmsg_test send \n");
printf("\ne.g: test ping rpmsg.virtio 10 times with secure, use command:\n"); printf("\trpmsg_test ping 0 1 10\n");}

編輯如下文件：

SDK/source/ssdk/ middleware/rpmsg_service/rpmsg_echo.c

staticintecho_channel_cb(void*data,intlen,unsignedlongsrc,void*arg){ rpmsg_channel_tchan = arg; structdcf_ccm_hdr*ccm_pkg = data;constchar*char_data = (char*)data; intret =0;
if(ccm_pkg->dmsg.msg_type == COMM_MSG_CCM_ECHO) { ccm_pkg->time[2] =timer_get_current_time(g_syscnt_timer) * g_sdrv_syscnt_dev.cnt_per_us; ret =rpmsg_channel_send(chan, src, data, len,1000); printf("Sending data (length %d): ", len);for(inti =8; i < len; i++) {printf("%c ", char_data[i]);}printf("\n");? ? }? ? else?if?(ccm_pkg->dmsg.msg_type == COMM_MSG_CCM_ACK) { ret =rpmsg_channel_send(chan, src, (char*)"ACK",4,1000); ssdk_printf(SSDK_INFO,"send ACK\n"); } else{ ssdk_printf(SSDK_WARNING,"echo_channel_cb: unknown %d bytes from addr %ld\n", len, src); }
if(ret !=0) { ssdk_printf(SSDK_WARNING,"echo_channel_cb: channel send failed\n"); }
returnret;}

通過以下指令打開圖形化配置界面，進入到Driver and Application Examples->Driver Example Support下，可以看到我們新加的talowe test Support驅動配置選項，需要關閉 RPMSG Example Application Support 選項，如下：

./tools/menuconfig.sh-bd9360_ref -pref -c secure

2.3A55核發送數據到R5核

A55核Linux系統通過echo_test命令與R5核進行通訊，以下是該命令源碼的文件位置：

自定義一個發送和接收字符串的函數，如下：

intsend_receive_string(intfd,constchar*send_str,intntimes,intseconds){inti =0, j =0;intsize, bytes_rcvd, bytes_sent;longelapse =0;err_cnt =0;i_payload = (struct_payload *)malloc(sizeof(struct_payload) + payload_max_size);r_payload = (struct_payload *)malloc(sizeof(struct_payload) + payload_max_size);if(i_payload ==0|| r_payload ==0) {printf("ERROR: Failed to allocate memory for payload.\n");return-1;}if(seconds)gettimeofday(&start_test,NULL);
i_payload->magic =0xA5;for(j =0; j < ntimes; j++) {i_payload->num = i;i_payload->size = size;strcpy(i_payload->data, send_str);if(verbose)printf("\r\n sending payload number");if(verbose)printf(" %d of size %lu\r\n", i_payload->num, (sizeof(struct_payload)) +strlen(i_payload->data));bytes_sent =write(fd, i_payload,sizeof(struct_payload) +strlen(i_payload->data));if(bytes_sent <=?0) {if?(verbose) {perror("\r\n Error sending data\n");break;}?elsefprintf(stderr,?"#");err_cnt++;continue;}printf("send string:%s\n",i_payload->data);r_payload->num =0;bytes_rcvd =read(fd, r_payload,sizeof(struct_payload) + payload_max_size);while(bytes_rcvd <=?0) {usleep(10000);bytes_rcvd =?read(fd, r_payload,?sizeof(struct?_payload) + payload_max_size);}r_payload->data[bytes_rcvd] ='\0';printf("receive string:%s\n",r_payload->data);
if(interval)sleep(interval);}free(i_payload);free(r_payload);return0;}

在main函數內添加新增函數條件，執行指令時實現A55核發送"Talowe:AtoR"字符串到R5核：

intmain(intargc,char*argv[]){ intopt; char*rpmsg_dev = RPMSG_DEVICE_NAME; ...... inttest_second =0; intsend_char =0; char*value_c;
if(strstr(argv[0],"property")) { do_property_test(argc, argv); return0; }...... while((opt =getopt(argc, argv,"anhiS")) !=-1) { switch(opt) { ...... case'v': verbose =1; break;
case'c':send_char =1;break;
case'b': benchmark =1; break; ...... default: printf("getopt return unsupported option: -%c\n", opt); usage("echo_test"); break; } } elseif(benchmark ==2) { iperf_test(fd, ntimes); }elseif(benchmark ==3) {
throughput_test(fd, ntimes, test_second);
}elseif( send_char ==1){constchar*message_to_send ="Talowe:AtoR";intresult =send_receive_string(fd, message_to_send,ntimes, test_second);if(result

在打印列表添加如下提示：

staticvoidusage(constchar*cmd){printf("This is s rpmsg echo test %s, MTU=%d\n", ECHO_TEST_VERSION, rpmsg_payload_test);...... printf("%s -v\t\t: Verbose mode\n", cmd); printf("%s -c [char]\t: send Talowe:AtoR\n", cmd); ...... printf("echo_test -d soc:ipcc@0.ipcc-echo.-1.30 -x -t 10 ; Throughput test in average (Ack'd) in 10 seconds (print in 1 second)\n");
exit(0);}

編輯如下文件：

SDK/source/linux/drivers/rpmsg/virtio_rpmsg_bus.c

staticintrpmsg_echo_cb(structrpmsg_device *rpdev,void*data,intlen, void*priv, u32 src){structdcf_message *msg = data;structvirtproc_info *vrp = priv;......switch(msg->msg_type) {caseCOMM_MSG_CCM_ECHO:dev_err(dev,"virtio send susses");/* Add timestamp in the time[2] */......break;caseCOMM_MSG_CCM_ACK:dev_err(dev,"rpmsg recv ACK");err = __send_offchannel_raw(vrp, RPMSG_ECHO_ADDR, src,"ACK",4,true);......break;default:/* No more action, just drop the packet */break;}return0;}

編輯如下文件：

SDK/source/linux/drivers/rpmsg/semidrive_ipcc.c

staticintrpmsg_ipcc_echo_cb(structrpmsg_device *rpdev,void*data,intlen, void*priv, u32 src){structrpmsg_ipcc_device *vrp = priv;structdevice *dev = vrp->dev;......switch(dmsg->msg_type) {caseCOMM_MSG_CCM_ECHO:dev_err(dev,"ipcc send susses");/* Add timestamp in the time[2] */ ......break;caseCOMM_MSG_CCM_ACK:dev_err(dev,"ipcc recv ACK");err = __send_offchannel_raw(vrp, RPMSG_ECHO_ADDR, src,"ACK",4,true); ......break;default:/* No more action, just drop the packet */break;}return0;}

內核需打開以下配置：

CONFIG_RPMSG=yCONFIG_RPMSG_CHAR=yCONFIG_RPMSG_VIRTIO=yCONFIG_RPMSG_SEMIDRIVE=y

三、燒寫測試驗證

3.1R5核向A55核發送信息

以下測試內容為R5核發送"Talowe:RtoA"字符串到A55核，A55核接收到字符串之后重新發送給R5核。

R5核：

rpmsg_testsend031

A55核：

3.2A55核向R5發送信息

以下測試內容為A55核發送"Talowe:AtoR"字符串到R5核，R5核接收到字符串之后重新發送給A55核。

A55核：

echo_test-d virtio0.rpmsg-echo.-1.30-c1

R5核：

四、結語

通過RPMsg在單芯片上實現了這種高效的核間通信機制，使得異構多核SoC能夠真正發揮"實時控制+高性能計算"的協同優勢，成為工業4.0、自動駕駛、AIoT等領域的核心技術底座。

廣州眺望電子科技有限公司推出Core-D9360核心板與EVM-D9載板外，還提供完整的SDK、編譯指南、測試手冊及技術文檔，覆蓋從環境搭建、代碼編譯、鏡像燒寫到功能驗證的全流程，助力開發者快速實現產品化。