深入芯驰D9360核间通信案例，RPMSG关键技术深度剖析

前言：

在多核异构架构成为工业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 < 0) {printf("Error sending/receiving string.\n");}    }    else {        if (test_second) {            ntimes = 0x7fffffff;        }......    return 0;}

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

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:[e-mail]-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、编译指南、测试手册及技术文档，覆盖从环境搭建、代码编译、镜像烧写到功能验证的全流程，助力开发者快速实现产品化。