前言：

在多核異構(gòu)架構(gòu)成為工業(yè)SoC主流方案的當(dāng)下，芯馳D9360憑借其Cortex-A55與Cortex-R5的協(xié)同設(shè)計，在需同時處理復(fù)雜應(yīng)用與實(shí)時任務(wù)的場景中表現(xiàn)突出。本文將以眺望電子Core-D9360平臺為例，詳解如何利用RPMSG與VirtIO機(jī)制實(shí)現(xiàn)A核與R核間的可靠通信，并提供關(guān)鍵代碼實(shí)現(xiàn)與調(diào)試方法。

圖 1 Core-D9360 核心板

一、通信基礎(chǔ)：RPMSG框架

RPMSG（Remote Processor Messaging）是Linux內(nèi)核中用于處理器間通信的框架，采用virtio + rpmsg + IPCC三層架構(gòu)實(shí)現(xiàn)：

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

●rpmsg：實(shí)現(xiàn)消息封裝與路由

●IPCC：硬件中斷控制器

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

圖 2 核間通訊架構(gòu)

二、核心代碼實(shí)現(xiàn)

2.1添加例程驅(qū)動

RTOS系統(tǒng)的例程添加在SDK/source/ssdk/examples/kunlun/drivers下，分別創(chuàng)建Kconfig和rules.mk文件，其中Kconfig文件內(nèi)容如下：

rules.mk文件內(nèi)容如下：

返回上一層目錄下，在Kconfig下添加talowe驅(qū)動模塊的初始狀態(tài)和圖形界面類型，添加內(nèi)容如下：

在rules.mk文件下，添加如下內(nèi)容：

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

2.2R5核發(fā)送機(jī)制實(shí)現(xiàn)

將rpmsg目錄下的rpmsg_test.c文件復(fù)制到talowe.c文件中，該驅(qū)動描述如何通過RPMSG協(xié)議進(jìn)行核間通信。我們可以在此程序基礎(chǔ)上加入自己的協(xié)議。

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

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

將以上函數(shù)添加至talowe.c文件內(nèi)，并在主函數(shù)rpmsg_test編寫調(diào)用該函數(shù)條件，如下：

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

通過以下指令打開圖形化配置界面，進(jìn)入到Driver and Application Examples->Driver Example Support下，可以看到我們新加的talowe test Support驅(qū)動配置選項(xiàng)，需要關(guān)閉 RPMSG Example Application Support 選項(xiàng)，如下：

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

2.3A55核發(fā)送數(shù)據(jù)到R5核

A55核Linux系統(tǒng)通過echo_test命令與R5核進(jìn)行通訊，以下是該命令源碼的文件位置：

自定義一個發(fā)送和接收字符串的函數(shù)，如下：

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函數(shù)內(nèi)添加新增函數(shù)條件，執(zhí)行指令時實(shí)現(xiàn)A55核發(fā)送"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;}

內(nèi)核需打開以下配置：

CONFIG_RPMSG=yCONFIG_RPMSG_CHAR=yCONFIG_RPMSG_VIRTIO=yCONFIG_RPMSG_SEMIDRIVE=y

三、燒寫測試驗(yàn)證

3.1R5核向A55核發(fā)送信息

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

R5核：

rpmsg_testsend031

A55核：

3.2A55核向R5發(fā)送信息

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

A55核：

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

R5核：

四、結(jié)語

通過RPMsg在單芯片上實(shí)現(xiàn)了這種高效的核間通信機(jī)制，使得異構(gòu)多核SoC能夠真正發(fā)揮"實(shí)時控制+高性能計算"的協(xié)同優(yōu)勢，成為工業(yè)4.0、自動駕駛、AIoT等領(lǐng)域的核心技術(shù)底座。

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