IPC共享内存调试

2025-10-18

字数统计: 6.6k | 阅读时长≈ 24 分钟

背景

由于 RPMsg 一次只能传递 496 字节的信息(在Linux系统中，RPMsg的最大大小包括16字节的头部，因此消息的有效载荷大小为512 - 16 = 496字节)，而CAN升级过程中需要传输的数据量是很大的，这样会很影响升级速率。因此，考虑在共享内存区域中传递大量数据，这样会更加高效。

参考 RPMsg_char zerocopy 示例，了解如何定义一个共享内存区域来在 Linux 和远程核心之间传递数据，然后使用 RPMsg 作为信号机制，在共享内存区域准备好读取时通知另一个核心。

调试

RPMsg_char zerocopy 这个仓库，展示了如何使用 rpmsg_charAPI 在 Linux 主机和远程 M4F 或 R5F MCU 端点之间传递共享内存数据。

仓库地址：rpmsg/rpmsg_char_zerocopy

master分支：适用 Linux 内核 6.6 或更高版本（SDK 10.x）。
ti-linux-6.1 分支：适用 Linux 内核 5.10（SDK 8.x）或 Linux 内核 6.1（SDK 9.x）。
linux目录：基于通用 Linux 的 rpmsg_char示例。
rtos目录：包含了针对 AM64x/R5F、AM62x/M4F 和 AM62Ax/C71/R5F 的 FreeRTOS 部分示例。

查看Linux 内核版本，可以在终端或命令行界面中使用以下命令：

uname -r

Demo流程说明

Linux端用指定的数据填充共享内存缓冲区。在访问 mmapped区域前后使用DMA_BUF_IOCTL_SYNCioctl进行非常基础的缓存一致性管理。
将缓冲区的物理地址、缓冲区大小和填充的固定数据通过rpmsg发送到远程端点(MCU)，并开始等待响应。
远程MCU将验证缓冲区数据，然后反转数据，用新数据填充缓冲区，并发送响应。
当Linux收到来自远程端点的消息时，验证与远程端共享的内存中的模式。

注意：填充共享内存缓冲区，也就是往共享内存写数据后，对端是不知道应用程序什么时候写的。所以在这里，rpmsg其实起到了通知的作用，实际上是产生了邮箱或中断通知，否则就需要轮询检查是否有新的数据。

这个示例与 ipc_rpmsg_echo_linux 示例类似，有几个关键的不同点：

创建了一个单一的 rpmsg 端点，用于与用户空间 Linux rpmsg_char 应用程序通信。
通过 rpmsg 通道交换包含共享内存描述符的⼆进制消息。共享数据位于共享内存中，不通过rpmsg 通道交换。
不与其他 CPU 交换任何消息。

API说明

Typedef说明

typedef void(* RPMessage_RecvCallback) (RPMessage_Object *obj, void *arg, void
*data, uint16_t dataLen, uint16_t remoteCoreId, uint16_t remoteEndPt)
`RPMessage_RecvCallback` 是一个回调函数类型定义，它在从指定的本地端点接收到任何CPU发
送的消息时被调用。

这个回调函数可以在 `RPMessage_construct` 期间选择性注册。

注意：
回调函数中必须处理所有消息内容。当回调函数返回时，消息缓冲区会被释放回发送者。
如果需要延迟处理消息内容，则需要复制消息内容。

参数：
`obj` [in] 使用 `RPMessage_construct` 创建的RPMessage端点对象。
`arg` [in] 用户在 `RPMessage_construct` 期间指定的参数。
`data` [in] 指向消息的指针。
`dataLen` [in] 消息的⻓度。
`remoteCoreId` [in] 发送方的核ID。
`remoteEndPt` [in] 发送方的端点。

typedef void(* RPMessage_RecvNotifyCallback) (RPMessage_Object *obj, void
*arg)
`RPMessage_RecvNotifyCallback` 是一个回调函数类型定义，它在从指定的本地端点接收到任何
CPU发送的消息时被调用。

这个回调函数可以在 `RPMessage_construct` 期间选择性注册。

注意：
与 `RPMessage_RecvCallback` 不同，这个回调函数仅通知有一条或多条消息待读取，
但消息本身不会被驱动程序读取，除非在该回调函数中或之后的某个任务中调用了
`RPMessage_recv`。
如果设置了 `RPMessage_RecvCallback`，则不会使用 `RPMessage_RecvNotifyCallback` 回
调函数。

参数：
`obj` [in] 使用 `RPMessage_construct` 创建的RPMessage端点对象。
`arg` [in] 用户在 `RPMessage_construct` 期间指定的参数。

typedef void(* RPMessage_ControlEndPtCallback) (void *arg, uint16_t
remoteCoreId, uint16_t remoteEndPt, const char *remoteServiceName)
`RPMessage_ControlEndPtCallback` 是一个回调函数类型定义，在控制端点上接收到通知消息
时调用的回调。

这个回调函数可以在 `RPMessage_init` 期间选择性注册。

注意：
与 `RPMessage_RecvCallback` 类似，回调函数中必须处理所有消息内容。
当回调函数返回时，消息缓冲区会被释放回发送者。如果需要延迟处理消息内容，则需要复制消息内
容。

参数：
`arg` [in] 用户在 `RPMessage_init` 期间指定的参数。
`remoteCoreId` [in] 发送方的核ID。
`remoteEndPt` [in] 在控制端点上宣布服务的发送者的端点。
`remoteServiceName` [in] 被宣布的远程服务的名称。

函数说明

void RPMessage_Params_init(RPMessage_Params * params)

`RPMessage_Params_init()` 函数用于为 `RPMessage_Params` 结构体设置默认值。
参数：
`params` [out] 需要被初始化为默认值的结构体。

void RPMessage_CreateParams_init(RPMessage_CreateParams * params)

`RPMessage_CreateParams_init()` 函数用于为 `RPMessage_CreateParams` 结构体设置默
认值。 
参数：
`params` [out] 需要被初始化为默认值的结构体。

int32_t RPMessage_init(const RPMessage_Params * params)

`RPMessage_init()` 函数用于初始化 RPMessage 模块。
参数：
`params` [in] 初始化参数。

void RPMessage_deInit(void)

`RPMessage_deInit()` 函数用于反初始化 RPMessage 模块。

int32_t RPMessage_waitForLinuxReady(uint32_t timeout)

`RPMessage_waitForLinuxReady()` 函数用于等待 Linux 端的 RPMessage 准备就绪。
在启用了 Linux 端的 RPMessage 之前，不应该向 Linux 发送消息，直到这个函数返回成功。

注意：
当在 RTOS/非RTOS 核心之间使用 RPMessage 时，不需要调用这个 API。

参数：
`timeout` [in] 超时时间，以系统时钟周期为单位。
返回值：
`SystemP_SUCCESS`，表示 Linux 端的 RPMessage 已经初始化完成。

void RPMessage_controlEndPtCallback(RPMessage_ControlEndPtCallback
controlEndPtCallback,
void * controlEndPtCallbackArgs
)

`RPMessage_controlEndPtCallback()` 函数用于设置一个回调，当在控制端点接收到控制消息
时调用。
参数：
`controlEndPtCallback` [in] 用户定义的回调函数，当控制消息到达时将被调用。
`controlEndPtCallbackArgs` [in] 传递给用户控制端点回调的参数。

int32_t RPMessage_construct(RPMessage_Object * obj,
const RPMessage_CreateParams * createParams
)

`RPMessage_construct()` 函数用于创建一个 RPMessage 对象，以便在指定的端点接收消息。

注意：
- 每个新创建的对象必须有一个唯一的本地端点。
- 本地端点必须小于 `RPMESSAGE_MAX_LOCAL_ENDPT`。
- 用户必须选择一个值，不支持使用 ANY 值。
- 如果在 `RPMessage_CreateParams` 中注册了回调，则不能使用 `RPMessage_recv`。

参数：
`obj` [out] 创建的对象。
`createParams` [in] 创建对象时的参数。
返回值：
`SystemP_SUCCESS` 表示成功，否则表示失败。

void RPMessage_destruct(RPMessage_Object * obj)

`RPMessage_destruct()` 函数用于删除之前创建的 RPMessage 对象。
参数：
`obj` [in] 要删除的对象。

void RPMessage_unblock(RPMessage_Object * obj)

`RPMessage_unblock()` 函数用于解除对输入对象的 `RPMessage_recv` 调用的阻塞，如果它
正在等待消息并且用户想要退出该任务。
参数：
`obj` [in] 要解除阻塞的对象。

uint16_t RPMessage_getLocalEndPt(const RPMessage_Object * obj)

`RPMessage_getLocalEndPt()` 函数用于返回一个 `RPMessage_Object` 的本地端点。
返回的值将与之前创建该对象时使用的值相同。
参数：
`obj` [in] 对象。
返回值：
输入对象的本地端点。

int32_t RPMessage_announce(uint16_t remoteProcId,
uint16_t localEndPt,
const char * name
)

`RPMessage_announce()` 函数用于向远程核心宣告一个本地端点，该端点上创建了一个服务。

注意：
- 宣告端点是可选的，IPC RPmessage不会以任何方式内部使用它。
- 用户必须逐个向所有感兴趣的远程核心宣告。没有向所有核心宣告的选项。
- 为了处理宣告消息，确保用户在 `RPMessage_init` 期间通过 `RPMessage_Params` 注册了
用户处理程序。
- 由最终用户决定如何使用回调来发出信号或等待直到远程服务被宣告。

参数：
`remoteProcId` [in] 要宣告的远程核心。
`localEndPt` [in] 正在宣告的服务的本地端点。
`name` [in] 正在宣告的服务的名称。
返回值：
如果宣告消息已发送，则返回 `SystemP_SUCCESS`，否则表示失败。

int32_t RPMessage_send(void * data,
uint16_t dataLen,
uint16_t remoteCoreId,
uint16_t remoteEndPt,
uint16_t localEndPt,
uint32_t timeout
)

`RPMessage_send()` 函数用于向指定远程端点的远程核心发送消息。

注意：
- `dataLen` 必须小于等于RPMessage_Params::vringMsgSize - 16字节，以留出内部头部的
空间。
- 为了让远程核心能够接收到消息，远程核心上要有一个与 `remoteEndPt` 相同值的端点。
- `localEndPt` 不是必需的，但是这个值在远程核心上是可用的，并且可以用作回复端点。使用
`RPMessage_getLocalEndPt` 来设置监听回复的 RPMessage 对象的本地端点。
- 当超时时间设置为 0 时，如果不可用的传输缓冲区，将⽴即返回 `SystemP_TIMEOUT`。否则，
它将等待指定的超时时间，直到有可用的传输缓冲区。

参数：
`data` [in] 要发送的消息数据的指针。
`dataLen` [in] 要发送的消息数据的大小。
`remoteCoreId` [in] 消息被发送到的远程核心ID。
`remoteEndPt` [in] 消息被发送到的远程核心端点ID。
`localEndPt` [in] 发送消息的本地端点。
`timeout` [in] 等待时间，以系统时钟周期为单位。
返回值：
`SystemP_SUCCESS`，当发送消息成功时返回。
`SystemP_TIMEOUT`，由于没有可用的传输缓冲区且超时发生，消息未被发送。

int32_t RPMessage_recv(RPMessage_Object * obj,
void * data,
uint16_t * dataLen,
uint16_t * remoteCoreId,
uint32_t * remoteEndPt,
uint32_t timeout
)

`RPMessage_recv()` 函数是一个阻塞式 API，直到在指定的本地端点接收到来自任何CPU的消
息。

注意：
- 本地端点在 `RPMessage_construct` 期间指定。
- 如果注册了回调，则不应使用此 API。
- 用户传递的 `dataLen` 包含用户消息缓冲区的大小，即指向 `data` 的缓冲区大小。如果接收
到的消息大小超过了 `*dataLen`，则会被截断。如果接收到的消息大小小于等于 `*dataLen`，则
所有接收到的字节都会被复制到 `data` 中，并且 `*dataLen` 表示 `data` 中有效字节的大
小。

参数：
`obj` [in] 使用 `RPMessage_construct` 创建的RPMessage端点对象。
`data` [in] 指向接收到的消息内容的指针。
`dataLen` [in] 用户消息缓冲区的⻓度，以字节为单位。
[out] 接收到的消息的大小，以字节为单位。
`remoteCoreId` [out] 发送方的核ID。
`remoteEndPt` [out] 发送方的端点。
`timeout` [in] 阻塞等待消息接收的时间，以系统时钟周期为单位。
返回值：
- `SystemP_SUCCESS`，接收到新消息，所有输出参数都是有效的。
- `SystemP_TIMEOUT`，由于超时，API被解除阻塞，输出参数不应被使用。

API for Task

void TaskP_Params_init(TaskP_Params * params)

`TaskP_Params_init()` 函数用于为 `TaskP_Params` 结构体设置默认值。
强烈建议在设置 `TaskP_Params` 中的值之前调用此函数。
参数：
`params` [out] 要设置为默认值的参数结构体。

int32_t TaskP_construct(TaskP_Object * obj,
TaskP_Params * params
)

`TaskP_construct()` 函数用于创建一个任务对象。
参数：
`obj` [out] 创建的对象。
`params` [in] 创建任务的参数。
返回值：
如果成功，返回 `SystemP_SUCCESS`。
如果出错，返回 `SystemP_FAILURE`。

共享内存的使用

使用共享内存的通讯方式，数据是不通过 rpmsg 通道交换的，rpmsg主要起通知作用，并传递共享内存描述的相关信息。这一条通知消息，可以和其他rpmsg的处理区分开，也可以不区分，主要有几种方式：

不新增 RPMessage 对象，和其他rpmsg一起处理，那么需要增加自定义交互协议，来交换包含共享内存描述的相关信息；
调用 RPMessage_construct创建一个新的 RPMessage 对象，和其他rpmsg分开处理。根据输入参数区分不同的RPMessage对象，做不同的处理。

注意：每个新创建的对象必须有一个唯一的本地端点。

假如之前已经用了14端点，共享内存的RPMessage对象可以用16端点。这就需要在项目中添加多个端点。

如何在MCU项目中添加多个RPMsg端点？

TI Resource Explorer

实际上就是应用了下面的补丁：

Linux_RPMsg_Echo-add-additional-endpoints.patch

问题

ASSERT

在之前代码的基础上，把demo的代码移植过去，增加一个任务后，出现了断言，如下图:

在 queue.c文件 xQueueSemaphoreTake函数1582行，configASSERT断言 QueueHandle_t xQueue是否为空。

出现断言打印，说明传了NULL。全局搜索没有直接调用 xQueueSemaphoreTake的地方，用的是xSemaphoreTake这个宏。

1 2	#define xSemaphoreTake( xSemaphore, xBlockTime ) xQueueSemaphoreTake( ( xSemaphore ), ( xBlockTime ) )

使用的地方比较多，所以下一步要定位，是新增加了哪个函数调用，导致该问题。

最终定位是新加的任务，调用 RPMessage_recv函数出现的这个问题。

问题在于RPMessage_recv使用了，未经过RPMessage_construct创建的RPMessage对象。

代码分析

详细分析见注释。

#define IPC_RPMESSAGE_ENDPT_CHRDEV_PING (14U)
#define MCU_IPC_RECV_OBJ (1)
/*
* 在初始化的阶段，RPMessage_construct函数会创建一个RPMessage对象gIpcRecvMsgObject[1]，
* 用来在IPC_RPMESSAGE_ENDPT_CHRDEV_PING端点接收消息。
* 然后会调用SemaphoreP_constructBinary创建一个⼆进制信号量对象。
*/
RPMessage_CreateParams_init(&createParams);
createParams.localEndPt = IPC_RPMESSAGE_ENDPT_CHRDEV_PING;
status = RPMessage_construct(&gIpcRecvMsgObject[MCU_IPC_RECV_OBJ],
&createParams);

/* Create the tasks which will handle the ping service */
TaskP_Params_init(&taskParams);
taskParams.name = "RPMESSAGE_CHAR_ZEROCOPY";
taskParams.stackSize = IPC_RPMESSAGE_TASK_STACK_SIZE;
taskParams.stack = gIpcTaskStack[0];
taskParams.priority = IPC_RPMESSAFE_TASK_PRI;
/* we use the same task function for echo but pass the appropiate rpmsg
handle to it, to echo messages */
taskParams.args = &gIpcRecvMsgObject[0];//gIpcRecvMsgObject[0]作为参数传递给
ipc_recv_task_main
taskParams.taskMain = ipc_recv_task_main;
status = TaskP_construct(&gIpcTask[0], &taskParams);
DebugP_assert(status == SystemP_SUCCESS);

void ipc_recv_task_main(void *args)
{
    int32_t status;
    struct ipc_buf ibuf = {0};
    uint16_t recvMsgSize, remoteCoreId;
    uint32_t remoteCoreEndPt;
    RPMessage_Object *pRpmsgObj = (RPMessage_Object *)args;

    DebugP_log("[IPC RPMSG ZEROCOPY] Remote Core waiting for messages at endpoint %d ... !!!\r\n",RPMessage_getLocalEndPt(pRpmsgObj));

    /* wait for messages forever in a loop */
    while(1)
    {
        /* Set 'recvMsgSize' to size of recv buffer, after return`recvMsgSize`
        * contains actual size of valid data in recv buffer.
        */
        recvMsgSize = sizeof(ibuf);
        /*
        * 前面在初始化的时候，创建的RPMessage对象是gIpcRecvMsgObject[1]，
        * 而gIpcRecvMsgObject[0]并没有做创建处理，对应的信号量是NULL。
        * 那么，在任务里面，gIpcRecvMsgObject[0]作为输入参数传给RPMessage_recv之后，
        * 会调用RPMessage_getEndPtMsg。当消息队列为空时，调用SemaphoreP_pend等待一个信号量对象
        * 或锁定一个互斥锁，最终调用xSemaphoreTake，检测信号量为NULL，于是出现了断言。
        */
        status = RPMessage_recv(pRpmsgObj,
                    &ibuf, &recvMsgSize,
                    &remoteCoreId, &remoteCoreEndPt,
                    SystemP_WAIT_FOREVER);
        DebugP_assert(status==SystemP_SUCCESS);
        ...
    }
}

MPU

MCU无法访问SOC发过来的地址，具体表现为，对地址进行读/写操作，MCU会挂掉。

1	IPC message: buffer address 0xXXXX, size 1M, pattern 0xaaaa5555

可能原因有：

内存权限和属性：R5F核可能没有足够的权限访问该内存区域，或者内存属性（如执行、写入、读取等）可能没有正确配置。检查共享内存区域的权限和属性设置。
内存对齐：如果共享内存的地址或大小没有正确对齐，可能会导致访问错误。

实际排查是因为没有访问权限，MPU里面没有对相应内存区域进行设置，需要在syscfg设置如下：

核间同步

两个CPU核同时访问同一个共享内存区域，如何实现同步，实现资源互斥访问是一个需要重点考虑的问题。

核间通信（IPC）的两种解决方案-CSDN博客

第一种通信的同步方案：

使用virtIO来实现，详见RPMSG-VirtIO。前面多次提到一次只能传递 496 字节的信息。

第⼆种通信的同步方案：

方案工作流程

R5F核写入共享内存：
- R5F核准备好数据后，将数据写入到共享内存的指定位置。
- R5F核通过RPMsg发送一个消息给A53核，通知它数据已经写入完成。
- 这个消息可以通过核间中断来触发A53核的处理，解除其阻塞状态。
- 可以将多个数据包批量写入共享内存，然后通过RPMsg通知一次，减少中断的频率，提升性能。
A53核读取数据：
- A53核在收到中断后从阻塞状态恢复，读取共享内存中R5F核写入的数据。
- 数据处理完后，A53核发送RPMsg通知R5F核操作完成。

同样，A53也可以通过类似的方式（共享内存写入 -> RPMsg 通知）将数据写给R5F核。

优点

同步精确：通过核间中断来通知对方核，可以做到精确的同步，这种方式减少了忙等(不像自旋锁那样忙等)。
同步机制简单：不需要复杂的锁来控制共享内存的访问权限。因为在每个核被通知后才会进行读写操作，避免了数据竞争。
减少CPU开销：通过阻塞等待和通知机制，避免了轮询式的资源检查，大大减少了CPU资源的浪费。
减少中断频率：通过批量写入共享内存，能够减少核间中断的频率，降低处理器之间的通信开销。RPMsg只在批量数据处理完后触发一次通知，从而提高效率。

需要注意的点

共享内存的一致性：需要手动清除缓存，确保两个核访问的是同样的数据。
RPMsg 通知的延迟：核间中断和RPMsg通常比较高效，但仍需要关注实际场景中的通知延迟，尤其是在高频率通信场景下。

优化思考

双缓冲机制：使用双缓冲，保证读写操作互不影响。例如，当R5F核写入时，A53核可以读取另一块缓冲区，减少读写等待的时间。
异步处理机制：如果接收方的处理速度较慢，可能会影响整个数据传输的吞吐量。在这种情况下，可以考虑让接收方异步处理接收到的数据，而不阻塞发送方的后续传输。当然，异步机制需要更复杂的同步控制。

说明

一个核心不同的task，访问同一个共享内存区域，也要考虑资源访问问题。

共享内存划分

假设共享内存传输的数据，都用同样一块地址空间。因为升级数据不是写一次就通知soc，是写满之后才通知，那么在升级数据传输过程中，如果有其他功能，需要使用共享内存来传输数据，会导致之前升级包的数据被覆盖。考虑到这种场景(实际可能不会这么操作)，可以对共享内存划分不同的区域。

物理地址说明

上面接口获取的是soc app申请的内存的物理地址，而不能简单认为是一个首地址，它可以不是首地址。如果soc app上电初始化之后，去申请内存，并且永远只申请一次(要考虑申请的大小，是申请一整个区域，还是其他)，那么由于内存空间都是空闲的，此时的物理地址就是首地址。或者可以每次申请并在使用完后释放，那么下次再去申请，应该也还是首地址。
soc app是把申请的内存的fd，通过 mmap函数把一个文件映射到进程的地址空间中，这样就可以直接通过内存操作来访问文件内容，也就是操作的就是app申请的空间。那么调用dmabuf_get_phys，获取到申请的内存的物理地址之后，发给mcu，这样两边访问的就是同一块地方。

TI官方资料

AM62Ax MCU+ SDK: Understanding inter-processor communication (IPC)

3.7. IPC for AM62ax — Processor SDK AM62Ax Documentation

8.5. Developing IPC applications — Processor SDK RTOS Automotive

打赏

版权声明： 本博客所有文章除特别声明外，著作权归作者所有。转载请注明出处！