最新资讯

目录

1 文档目的

2 相关概念

2.1 术语

2.2 RPMsg相关概念

3 RPMsg核间通信软硬件模块框架

3.1 硬件原理

3.2 软件框架

4 使用RPMsg进行核间通信

4.1 RPMsg通信建立

4.1.1 使用名称服务建立通信

4.1.2 不用名称服务

4.2 RPMsg应用过程

4.3 应用层示例

5 RPMsg内核驱动源码解析

5.1 RPMsg框架源码

5.1.1 源代码与功能

5.1.2 源码分析

5.2 virtio介绍

5.2.1 virtio源码及virtio_queue原理介绍

5.2.2 源码分析

5.3 remoteproc介绍

5.3.1 remoteproc子系统

5.3.2 remoteproc源码解析

5.4 mailbox 框架

5.4.1 mailbox源码介绍

5.4.2 mailbox框架源码解析

6 收发数据流程总结

6.1 接收数据过程调用栈

6.2 发送数据过程调用栈

6.3 收发数据buffer处理过程

6.3.1  J6缓冲区处理流程

6.3.2 stm32-mp157缓冲区处理流程

6.3.3 缓冲区处理

7 附录

1 文档目的

该文档目的介绍RPMsg基础概念知识、RPMsg整体使用流程，rpmsg、virtio、remotepro、mailbox软件框架、硬件原理相关介绍和RPMsg适配新平台的开发流程，以帮助开发者更好理解RPMsg核间通信框架，进行核间通信开发。

2 相关概念

2.1 术语

RPMsg: remote processor messages，Linux下用于核间通信顶层框架，面向驱动开发者。

Virtio: virtual IO，linux平台下一种IO半虚拟化框架。

Remoteproc: Remote Processor Framework，用于管理异构远程处理器设备，既是一个硬件模块，Linux也实现了对应的remoteproc框架，remoteproc框架允许不同平台或架构控制远程处理器。

Mailbox：Linux一种软件框架，通过消息队列和中断驱动信号处理多处理器间的通讯，也有人把用于核间通信模块称为mailbox，软件上是个框架，硬件也是个IP。

MHU：mailbox and handshake uint，是个硬件模块，包含mailbox功能和握手功能，用来确认双方收发情况，减少丢包等。

IPCC：功能上跟mailbox一样，用于核间通信的硬件IP模块。

2.2 RPMsg相关概念

RPMsg-Lite 组件： Remote Processor Messaging（RPMsg）协议的一个轻量级实现。

通道：用于数据传输，通信前需要建立通道，跟实际硬件通道（mailbox/MHU/IPCC）不是同一个概念（后面详细解析），是RPMsg的一样虚拟概念。

端点：用于核间通信的具体控制点，利用通道进行数据通信。

3 RPMsg核间通信软硬件模块框架

3.1 硬件原理

RPMsg硬件上依赖mailbox或者MHU（mailbox and handshake uint）和共享内存。基于原理是利用mailbox/MHU模块，触发接收端对应通道中断，接收端收到中断，从共享内存中取出发送放放在共享内存中的数据，如下图：

3.2 软件框架

RPMsg是在基于虚拟化框架virtio上层实现软件框架，RPMsg总线是一种基于virtio的消息总线。Linux下的RPMsg核间通信，涉及多个Linux内核多个软件框架，包括virtio、remoteproc和mailbox，下图大致描述了各个软件框架在核间通信的关系：

APP通过调用RPMsg提供的设备来进行核间通信，RPMsg driver向RPMsg bus总线注册driver，同时由virtio和RPMsg的中间适配层RPMsg virtio layer向注册对应device，匹配对应的driver，同时该层又向下注册了virtio driver，将virtio和RPMsg关联起来。Virtio的下层是remoteproc框架，该框架向上注册了virtio device，以和virtio driver匹配向上提供device硬件相关接口（这里硬件接口，包括remoteproc这个IP的接口和下层mailbox接口）。Mailbox框架向外提供client，remoteproce框架同时注册上层了一个mailbox client以和mailbox交互，mailbox向下同时提供了controller，让具体的硬件模块提供硬件操作接口。

此外，有些软件框架可以脱离其他层单独使用，例如mailbox，驱动开发者可自我注册client生成设备提供APP层直接进行核间通信，remoteproc通过sysfs暴露接口给APP层，用户可以直接控制远端处理器的状态（启动、停止、固件加载等）。而virtio则是一个与硬件无关的框架，可用于其他外设作为数据管理使用。

最新Linux内核，GLINK和SMD也向RPMsg bus注册了device，这两个协议是用于高通平台，本文不展述。

4 使用RPMsg进行核间通信

本节从实际使用角度，描述user空间，APP如何使用RPMsg进行核间通信。

4.1 RPMsg通信建立

4.1.1 使用名称服务建立通信

RPMsg使用前，有时需要建立一定条件，即在使用远端服务时，需要依赖远端发送服务到本地，本地建立服务（即是设备名称）后，user空间app才可打开设备进行通信，如下图是通信建立的大致流程：

SOC在使用前，需要先收到远端服务请求，在本地建立RPMsg通道、端点和设备等信息，再反馈初始化信息给远端，双方才可进行通信。建立后，APP可主动发送数据，被动收数据或者主动沦陷，最后主动销毁通道。

在Linux内核代码，服务建立是kernel/drivers/rpmsg/rpmsg_ns.c文件完成，RPMsg device 层向上注册了ns的device，后续章节将深入代码解析该流程。

4.1.2 不用名称服务

最新的内核代码，也能在不使用名称服务情况下，直接建立端点进行通信，甚至脱离具体RPMsg通道建立端点来通信，这种情况下，需要双方手动登记通信的端点地址，如下图：

SOC侧在驱动加载时，直接建立通道和端点信息，之后进行正常通信，此时需要手动和远端登记好地址。

最新内核代码中，drivers/rpmsg/rpmsg_ctrl.c和drivers/rpmsg/rpmsg_char.c都是在不适用名称服务情况下直接进行通信，同时许多其他利用RPMsg框架进行核间通信的模块，都直接使用该方式，例如camera的核间通信等。

4.2 RPMsg应用过程

利用RPMsg核间通信过程：建立通道和端点同时绑定本地地址和远端地址，发送时带上本地地址，远端发送数据到本地时，也需要携带远端地址才能被本地端点使用，上送数据给应用层。

比较典型的应用，是一个通道两端各一个端点进行通信：

还可以一个通道多个端点

或者多通道多端点

4.3 应用层示例

在底层准备好后，应用层就能打开底层提供的设备进行核间通信。以/dev/rpmsg_ctrl0设备为例子，就是4.1.2提到的drivers/rpmsg/rpmsg_ctrl.c生成的设备，该RPMsg驱动支持生成多端点进行通信或者多通道，下面应用程序是使用流程：

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include "sstar_rpmsg.h"
int main(void)
{
    struct ss_rpmsg_endpoint_info info;
    char buffer[512];
    char data[512];
    int ret;
    char devPath[256];
    int fd, eptFd;
    unsigned int index = 0x0;
    memset(&info, 0, sizeof(info));
    info.src = EPT_ADDR_MACRO(EPT_TYPE_CUSTOMER, 1);
    info.dst = EPT_ADDR_MACRO(EPT_TYPE_CUSTOMER, 2);
    snprintf(info.name, sizeof(info.name), "demo");
    info.mode = RPMSG_MODE_RISCV;
    info.target_id = 0;
    fd = open("/dev/rpmsg_ctrl0", O_RDWR);
    if (fd < 0)
    {
        perror("open");
        return 0;
    }
    if (ioctl(fd, SS_RPMSG_CREATE_EPT_IOCTL, &info) < 0)//创建通道时携带端点地址
    {
        perror("ioctl");
        return 0;
    }
    sleep(2);
    snprintf(devPath, sizeof(devPath),  "/dev/rpmsg%d", info.id);//打开创建端点时生成的设备
    eptFd = open(devPath, O_RDWR);
    if (eptFd < 0)
    {
        fprintf(stderr, "Failed to open endpoint!
");
        return 0;
    }
    while (1)
    {
        snprintf(buffer, sizeof(buffer), "hello,world:0x%x
", index++);
        ret = write(eptFd, buffer, strlen(buffer) + 1);//写数据
        memset(data, 0, sizeof(data));
        ret = read(eptFd, data, sizeof(data));//读数据
        if (ret > 0)
            printf("read ept:%d, %s
", ret, data);
        else
            printf("read ept error:%d
", ret);
    }
    return 0;
}

使用由rpmsg_ctrl.c注册生产的设备来使用rpmsg进行核间通信；先用 /dev/rpmsg_ctrl0 设备通过ioctrl生成端点 SS_RPMSG_CREATE_EPT_IOCT ，此时驱动将为进程生成 /dev/rpmsg X 设备用来核间通信使用，后续可以使用常用的文件操作方式访问该设备进行核间通信。

• "/dev/rpmg_ctrl0" to instantiate /dev/rpmsg0 sysfs device interface,
• "/dev/rpmg0" sysfs device interface for communication.

在后面章节，将介绍分析驱动如何提供接口给应用层。

5 RPMsg内核驱动源码解析

RPMsg内核驱动框架涉及到RPMsg框架、virtio框架、remoteproc子系统和mailbox子系统，下面逐一介绍这些框架。

5.1 RPMsg框架源码

5.1.1 源代码与功能

RPMsg源码位于Linux源码目录kernel/drivers/rpmsg/下，下图是一个RPMsg大致软件功能框图，主要涉及源文件为：

kernel/drivers/rpmsg/rpmsg_core.c rpmsg核心代码

kernel/drivers/rpmsg/virtio_rpmsg_bus.c virtio与rpmsg适配层

kernel/drivers/rpmsg/rpmsg_ctrl.c驱动应用

kernel/drivers/rpmsg/rpmsg_ns.c驱动应用

kernel/drivers/rpmsg/rpmsg_char.c驱动应用

rpmsg_core.c创建了一个bus总线，驱动应用层（rpmsg_ctrl.c、rpmsg_ns.c、rpmsg_char.c）向下注册rpmsg_driver，同时注册字符设备暴露接口给应用层，virtio_rpmsg_bus.c文件向下注册了virtio driver，向上则注册rpmsg device，一旦和上层驱动应用的driver匹配，上层驱动应用即可通过device提供的接口访问到device层，即virtio_rpmsg_bus.c这一层，这是rpmsg bus主要功能，同时也支持多个device和driver的接入。

5.1.2 源码分析

先看bus总线的初始化,drivers/rpmsg/rpmsg_core.c文件里，内存加载时，会先调用bus_register注册一条总线，同时提供了device和driver的注册接口

static struct bus_type rpmsg_bus = {
    .name       = "rpmsg",
    .match      = rpmsg_dev_match,
    .dev_groups = rpmsg_dev_groups,
    .uevent     = rpmsg_uevent,
    .probe      = rpmsg_dev_probe,
    .remove     = rpmsg_dev_remove,
};
static int __init rpmsg_init(void)
{
    .....
    ret = bus_register(&rpmsg_bus);
    ....
    return ret;
}
postcore_initcall(rpmsg_init);

int __register_rpmsg_driver(struct rpmsg_driver *rpdrv, struct module *owner)
{
    rpdrv->drv.bus = &rpmsg_bus;
    rpdrv->drv.owner = owner;
    return driver_register(&rpdrv->drv);
}
EXPORT_SYMBOL(__register_rpmsg_driver);
int rpmsg_register_device(struct rpmsg_device *rpdev)
{
    return rpmsg_register_device_override(rpdev, NULL);
}
EXPORT_SYMBOL(rpmsg_register_device);

当匹配到driver和device时，将调用rpmsg_dev_probe并调用driver的probe函数传入device对象。除了对总线的初始化，该文件提供的对外接口，最终都是调用传入device的回调。

Device的注册在drivers/rpmsg/rpmsg_core.c，该文件最开始注册了virtio（后续virtio章节分析），当virtio的driver和device匹配时，rpmsg_probe被调用，该函数则是根据上层驱动需求创建rpmsg device，具体可由驱动开发者实现。

static struct virtio_driver virtio_ipc_driver = {
    .feature_table  = features,
    .feature_table_size = ARRAY_SIZE(features),
    .driver.name    = KBUILD_MODNAME,
    .driver.owner   = THIS_MODULE,
    .id_table   = id_table,
    .probe      = rpmsg_probe,
    .remove     = rpmsg_remove,
};
static int __init rpmsg_init(void)
{
...
    ret = register_virtio_driver(&virtio_ipc_driver);
  ...
}
subsys_initcall(rpmsg_init);

static int rpmsg_probe(struct virtio_device *vdev)
{
...
rpdev_ctrl = rpmsg_virtio_add_ctrl_dev(vdev);//rpmsg_ctrl设备
err = rpmsg_ns_register_device(rpdev_ns);//名称服务设备
}

该文件另外最大的功能，是提供device以及ept端点的回调给上层。

接下来分析最上层驱动应用，以4.3章节应用程序用的/dev/rpmsg_ctrl0设备来分析rpmsg驱动源码。该设备是由drivers/rpmsg/rpmsg_ctrl.c生成的。

在rpmsg_ctrl.c里，驱动初始化时调用rpmsg bus提供的driver注册接口注册了rpmsg driver，名为rpmsg_ctrl：

static struct rpmsg_driver rpmsg_ctrldev_driver = {
    .probe = rpmsg_ctrldev_probe,
    .remove = rpmsg_ctrldev_remove,
    .drv = {
        .name = "rpmsg_ctrl",
    },
};
ret = register_rpmsg_driver(&rpmsg_ctrldev_driver);

在drivers/rpmsg/virtio_rpmsg_bus.c，同样注册了一个名为rpmsg_ctrl的device：

rpmsg_probe
rpmsg_virtio_add_ctrl_dev
rpmsg_ctrldev_register_device
rpmsg_register_device_override(rpdev, "rpmsg_ctrl")（drivers/rpmsg/rpmsg_internal.h）
dev->bus = &rpmsg_bus;ret = device_add(dev);（kernel/drivers/rpmsg/rpmsg_core.c）

当driver和device匹配后，platform调用driver的probe，添加一个rpmsg_ctrlx设备给上层调用：

static const struct file_operations rpmsg_ctrldev_fops = {
    .owner = THIS_MODULE,
    .open = rpmsg_ctrldev_open,
    .release = rpmsg_ctrldev_release,
    .unlocked_ioctl = rpmsg_ctrldev_ioctl,
    .compat_ioctl = compat_ptr_ioctl,
};
static int rpmsg_ctrldev_probe(struct rpmsg_device *rpdev)
{
     cdev_init(&ctrldev->cdev, &rpmsg_ctrldev_fops);
 ...
    dev_set_name(&ctrldev->dev, "rpmsg_ctrl%d", ret);
  ...
ret = cdev_device_add(&ctrldev->cdev, &ctrldev->dev);
  ...
    return ret;
}

上层调用ioctrl创建ept时，会新创建一个rpmsgx设备，调用栈如下：

rpmsg_ctrldev_ioctl
rpmsg_chrdev_eptdev_create（drivers/rpmsg/rpmsg_char.c）
rpmsg_chrdev_eptdev_alloc->cdev_init(&eptdev->cdev, &rpmsg_eptdev_fops);
rpmsg_chrdev_eptdev_add
{
...
dev_set_name(dev, "rpmsg%d", ret);
ret = cdev_device_add(&eptdev->cdev, &eptdev->dev);
...
}
其中文件操作集rpmsg_eptdev_fops定义如下：
static const struct file_operations rpmsg_eptdev_fops = {
    .owner = THIS_MODULE,
    .open = rpmsg_eptdev_open,
    .release = rpmsg_eptdev_release,
    .read_iter = rpmsg_eptdev_read_iter,
    .write_iter = rpmsg_eptdev_write_iter,
    .poll = rpmsg_eptdev_poll,
    .unlocked_ioctl = rpmsg_eptdev_ioctl,
    .compat_ioctl = compat_ptr_ioctl,
};

应用层在打开rpmsg0进行读写

写：
rpmsg_eptdev_write_iter
rpmsg_sendto:ept->ops->sendto(ept, data, len, dst);//调用端点绑定的回调
端点ept的来源
rpmsg_eptdev_open
{
ept = rpmsg_create_ept(rpdev, rpmsg_ept_cb, eptdev, eptdev->chinfo);
eptdev->ept = ept;
}
而在rpmsg_create_ept: rpdev->ops->create_ept(rpdev, cb, priv, chinfo)（drivers/rpmsg/rpmsg_core.c）调用的是device的ops，该device之前以分析在kernel/drivers/rpmsg/virtio_rpmsg_bus.c生成
static struct rpmsg_device *rpmsg_virtio_add_ctrl_dev(struct virtio_device *vdev)
{
struct rpmsg_device *rpdev_ctrl;
    rpdev_ctrl->ops = &virtio_rpmsg_ops;
}
static const struct rpmsg_device_ops virtio_rpmsg_ops = {
    .create_channel = virtio_rpmsg_create_channel,
    .release_channel = virtio_rpmsg_release_channel,
    .create_ept = virtio_rpmsg_create_ept,
    .announce_create = virtio_rpmsg_announce_create,
    .announce_destroy = virtio_rpmsg_announce_destroy,
};
virtio_rpmsg_create_ept
static struct rpmsg_endpoint *__rpmsg_create_ept(struct virtproc_info *vrp,
                         struct rpmsg_device *rpdev,
                         rpmsg_rx_cb_t cb,
                         void *priv, u32 addr)
{
    int id_min, id_max, id;
    struct rpmsg_endpoint *ept;
ept->ops = &virtio_endpoint_ops;
...
return ept;
}
static const struct rpmsg_endpoint_ops virtio_endpoint_ops = {
    .destroy_ept = virtio_rpmsg_destroy_ept,
    .send = virtio_rpmsg_send,
    .sendto = virtio_rpmsg_sendto,
    .send_offchannel = virtio_rpmsg_send_offchannel,
    .trysend = virtio_rpmsg_trysend,
    .trysendto = virtio_rpmsg_trysendto,
    .trysend_offchannel = virtio_rpmsg_trysend_offchannel,
    .get_mtu = virtio_rpmsg_get_mtu,
};
回到最开始的写函数
rpmsg_sendto:ept->ops->sendto(ept, data, len, dst);
即此时写调用的是drivers/rpmsg/virtio_rpmsg_bus.c下的virtio_rpmsg_sendto
其他send的函数同理

读
rpmsg_eptdev_read_iter
{
skb = skb_dequeue(&eptdev->queue);
}
static int rpmsg_ept_cb(struct rpmsg_device *rpdev, void *buf, int len,
            void *priv, u32 addr)
{
skb_queue_tail(&eptdev->queue, skb);
}
ept = rpmsg_create_ept(rpdev, rpmsg_ept_cb, eptdev, eptdev->chinfo);
根据上述写的分析，这个rpmsg_create_ept最终调用drivers/rpmsg/virtio_rpmsg_bus.c下的
static struct rpmsg_endpoint *__rpmsg_create_ept(struct virtproc_info *vrp,
                         struct rpmsg_device *rpdev,
                         rpmsg_rx_cb_t cb,
                         void *priv, u32 addr)
{
ept->cb = cb;
}
该回调被绑定到端点上，等待其他地方调用。
该cb回调由以下流程调用
rpmsg_recv_done（注册到下层virtio，由下层调用，后面章节分析）
rpmsg_recv_single
{
ept->cb(ept->rpdev, msg->data, msg_len, ept->priv, __rpmsg32_to_cpu(little_endian, msg->src));
}

以上便是rpmsg所有的代码功能，驱动开发者接触相对较多的是该层代码。

5.2 virtio介绍

5.2.1 virtio源码及virtio_queue原理介绍

Virtio源码位于Linux内核路径kernel/drivers/virtio/下，virtio在此最重要的功能就是其对缓冲区的管理：vring机制。

drivers/virtio/virtio_ring.c：vring提供相关操作接口

drivers/virtio/virtio.c：逻辑核心，包括bus总线初始化等

如下图是virtio对vring的管理方式，virtio对上对下分别区分guest和host的概念。Avail ring指guest用于发送数据的缓冲区，use ring指guest用于接收数据的缓冲区，两者都是用与记录当前可用数据在desc buffer的索引值，dsec buffer则记录了所有缓冲区的物理地址。

Guest负责对vring缓冲区初始化，初始化完成后，当guest要发送数据时，将在avail记录发送缓冲区索引值，随后通过下层device通知下层发送。当host有数据时，将数据拷贝（大部分硬件已完成）到可用的use ring缓冲区，被回调上层注册下来的接口，将数据上送给上层。

5.2.2 源码分析

先看drivers/virtio/virtio.c，文件同样注册bus，同时提供device和driver的注册接口。

static int virtio_init(void)
{
    if (bus_register(&virtio_bus) != 0)
        panic("virtio bus registration failed");
    return 0;
}
core_initcall(virtio_init);
-------------------------
static struct bus_type virtio_bus = {
    .name  = "virtio",
    .match = virtio_dev_match,
    .dev_groups = virtio_dev_groups,
    .uevent = virtio_uevent,
    .probe = virtio_dev_probe,
    .remove = virtio_dev_remove,
};
int register_virtio_driver(struct virtio_driver *driver)
{
    /* Catch this early. */
    BUG_ON(driver->feature_table_size && !driver->feature_table);
    driver->driver.bus = &virtio_bus;
    return driver_register(&driver->driver);
}
int register_virtio_device(struct virtio_device *dev)
{
    err = device_add(&dev->dev);
    if (err)
        goto out_of_node_put;
    return 0;
...
    return err;
}

当device和driver匹配成功，将调用probe函数以及driver的probe，传入device设备对象：

static int virtio_dev_probe(struct device *_d)
{
...
        err = drv->probe(dev);
    if (err)
        goto err;
...
}

该文件除了提供基础的bus初始化，其他都是无关紧要接口。

drivers/virtio/virtio_ring.c则是对vring管理提供了对外接口，包括接收数据通知上层，以及下发发送的通知回调等。

virtqueue_add_outbuf向vring添加一个可用的发送buffer
virtqueue_add_inbuf向vring添加一个可用的接收buffer
virtqueue_kick_prepare 通知前或者初始化前的准备（主要是内存资源初始化相关）
virtqueue_kick->virtqueue_notify 通知发送接口，当guest已经准备好数据后，调用该接口通知下层发送数据
vring_interrupt host接收到数据，调用该接口通知上层，该接口最终调用上层注册的rpmsg_recv_done函数
vring_new_virtqueue添加一个/多个virtqueue，初始化时调用

来看virtio跟上下层实例交互，先看guest注册virtio driver，签名分析过virtio driver的注册是在drivers/rpmsg/virtio_rpmsg_bus.c里。

static int __init rpmsg_init(void)
{
...
    ret = register_virtio_driver(&virtio_ipc_driver);
  ...
}

当匹配成功后，driver的probe被调用，在probe函数有几个比较重要的初始化：

static int rpmsg_probe(struct virtio_device *vdev)
{
    vq_callback_t *vq_cbs[] = { rpmsg_recv_done, rpmsg_xmit_done };
    static const char * const names[] = { "input", "output" };
    struct virtqueue *vqs[2];
    struct virtproc_info *vrp;
    struct virtio_rpmsg_channel *vch = NULL;
    struct rpmsg_device *rpdev_ns, *rpdev_ctrl;
/* We expect two virtqueues, rx and tx (and in this order) */
//根据名称创建两个virtio_queue，调用device提供的find_vqs回调
    err = virtio_find_vqs(vdev, 2, vqs, vq_cbs, names, NULL);
    if (err)
        goto free_vrp;
    vrp->rvq = vqs[0];
vrp->svq = vqs[1];
...
/* set up the receive buffers */
//根据当前buf数量，配置到vring中，这里配置的一般是mailbox使用的共享内存
    for (i = 0; i < vrp->num_bufs / 2; i++) {
        struct scatterlist sg;
        void *cpu_addr = vrp->rbufs + i * vrp->buf_size;

        rpmsg_sg_init(&sg, cpu_addr, vrp->buf_size);
        err = virtqueue_add_inbuf(vrp->rvq, &sg, 1, cpu_addr,
                      GFP_KERNEL);
        WARN_ON(err); /* sanity check; this can't really happen */
    }
...
}

在driver里，上层rpmsg bus发送数据

static int rpmsg_send_offchannel_raw(struct rpmsg_device *rpdev,
                     u32 src, u32 dst,
                     void *data, int len, bool wait)
{
    struct virtio_rpmsg_channel *vch = to_virtio_rpmsg_channel(rpdev);
    struct virtproc_info *vrp = vch->vrp;
    struct device *dev = &rpdev->dev;
    struct scatterlist sg;
    struct rpmsg_hdr *msg;
int err;
...
    /* grab a buffer */
    msg = get_a_tx_buf(vrp);
    if (!msg && !wait)
        return -ENOMEM;
    ...
    rpmsg_sg_init(&sg, msg, sizeof(*msg) + len);
    mutex_lock(&vrp->tx_lock);
/* add message to the remote processor's virtqueue */
//将buffer添加vring中
    err = virtqueue_add_outbuf(vrp->svq, &sg, 1, msg, GFP_KERNEL);
   ....
/* tell the remote processor it has a pending message to read */
//通知下层进行数据发送
    virtqueue_kick(vrp->svq);
out:
    mutex_unlock(&vrp->tx_lock);
    return err;
}

接下来是device的注册过程，device的注册由virtio与remoteproc的适配层实现drivers/remoteproc/remoteproc_virtio.c

hobot_rproc_vdsp_resume（hobot-drivers/remoteproc/hobot_remoteproc.c）
hobot_remoteproc_boot_vdsp
rproc_boot
rproc_fw_boot
rproc_start
static int rproc_start_subdevices(struct rproc *rproc)
{
    struct rproc_subdev *subdev;
...
list_for_each_entry(subdev, &rproc->subdevs, node) {
        if (subdev->start) {
            ret = subdev->start(subdev);
            if (ret)
                goto unroll_registration;
        }
}
...
}
------------------
drivers/remoteproc/remoteproc_virtio.c
static int rproc_virtio_probe(struct platform_device *pdev)
{
rvdev->subdev.start = rproc_vdev_do_start;
rproc_add_subdev(rproc, &rvdev->subdev);
}
rproc_add_virtio_dev：ret = register_virtio_device(vdev);

5.3 remoteproc介绍

5.3.1 remoteproc子系统

remoteproc源码在drivers/remoteproc/目录下，主要用来管理远程处理器启动、停止、固件加载等，围绕这些功能，remoteproc提供了不同的接口。

drivers/remoteproc/remoteproc_virtio.c：virtio和remoteproc适配层

hobot-drivers/remoteproc/hobot_remoteproc.c：厂商实现，mailbox和remoteproc适配层。

kernel/drivers/remoteproc/remoteproc_core.c：remoteproc核心代码

如下是remoteproc子系统的大致功能图，remoteproc_core.c是整个代码核心，向上提供virtio device注册进行virtio bus，和virtio层关联，提供remoteproc层的接口，同时初始化自己成为一个mailbox client，和下层mailbox通信关联，调用mailbox框架下的硬件接口。

5.3.2 remoteproc源码解析

关于remoteproc如何生成注册一个virtio device，在virito章节已介绍，在向上层注册完device后，上层driver即可访问到remoteproe层的接口，例如这里的config操作集，上层可回调操作集接口：

static int rproc_add_virtio_dev(struct rproc_vdev *rvdev, int id)
{
vdev->config = &rproc_virtio_config_ops,
}
static const struct virtio_config_ops rproc_virtio_config_ops = {
    .get_features   = rproc_virtio_get_features,
    .finalize_features = rproc_virtio_finalize_features,
    .find_vqs   = rproc_virtio_find_vqs,
    .del_vqs    = rproc_virtio_del_vqs,
    .reset      = rproc_virtio_reset,
    .set_status = rproc_virtio_set_status,
    .get_status = rproc_virtio_get_status,
    .get        = rproc_virtio_get,
    .set        = rproc_virtio_set,
};

以这里的rproc_virtio_find_vqs为例，前面分析到virtio driver的probe函数里调用函数virtio_find_vqs，该函数的实现是回调device里config操作集，即这里的rproc_virtio_find_vqs

include/linux/virtio_config.h
static inline
int virtio_find_vqs(struct virtio_device *vdev, unsigned nvqs,
            struct virtqueue *vqs[], vq_callback_t *callbacks[],
            const char * const names[],
            struct irq_affinity *desc)
{
    return vdev->config->find_vqs(vdev, nvqs, vqs, callbacks, names, NULL, desc);
}

rproc_virtio_find_vqs则是初始化virtio_queue队列，同时绑定notify通知函数到vring中，上层发送时，将回调该函数通知到remoteproc层，而remoteproc层则调用下层mailbox硬件接口进行远端通知。

/* kick the remote processor, and let it know which virtqueue to poke at */
static bool rproc_virtio_notify(struct virtqueue *vq)
{
    struct rproc_vring *rvring = vq->priv;
    struct rproc *rproc = rvring->rvdev->rproc;
    int notifyid = rvring->notifyid;

    dev_dbg(&rproc->dev, "kicking vq index: %d
", notifyid);
    rproc->ops->kick(rproc, notifyid);
    return true;
}
static struct virtqueue *rp_find_vq(struct virtio_device *vdev,
                    unsigned int id,
                    void (*callback)(struct virtqueue *vq),
                    const char *name, bool ctx)
{
    ....
    /*
     * Create the new vq, and tell virtio we're not interested in
     * the 'weak' smp barriers, since we're talking with a real device.
     */
    vq = vring_new_virtqueue(id, num, rvring->align, vdev, false, ctx,
                 addr, rproc_virtio_notify, callback, name);
    if (!vq) {
        dev_err(dev, "vring_new_virtqueue %s failed
", name);
        rproc_free_vring(rvring);
        return ERR_PTR(-ENOMEM);
    }
    ...
}

rproc->ops->kick(rproc, notifyid) 的回调注册如下：

static struct rproc_ops hobot_vdsp_rproc_ops = {
    .start = hobot_vdsp_rproc_start,
    .stop = hobot_vdsp_rproc_stop,
.kick = hobot_vdsp_rproc_kick,
...
};
hobot_remoteproc_probe
localrproc = rproc_alloc(&pdev->dev, dev_name(&pdev->dev), &hobot_vdsp_rproc_ops, NULL, sizeof(struct hobot_rproc_vdsp_pdata));
rproc_alloc_ops
static int rproc_alloc_ops(struct rproc *rproc, const struct rproc_ops *ops)
{
//申请内存后拷贝ops内容，即外面传入的回调
    rproc->ops = kmemdup(ops, sizeof(*ops), GFP_KERNEL);
    if (!rproc->ops)
        return -ENOMEM;
    /* Default to rproc_coredump if no coredump function is specified */
    if (!rproc->ops->coredump)
        rproc->ops->coredump = rproc_coredump;
    if (rproc->ops->load)
        return 0;
    /* Default to ELF loader if no load function is specified */
    rproc->ops->load = rproc_elf_load_segments;
    rproc->ops->parse_fw = rproc_elf_load_rsc_table;
    rproc->ops->find_loaded_rsc_table = rproc_elf_find_loaded_rsc_table;
    rproc->ops->sanity_check = rproc_elf_sanity_check;
    rproc->ops->get_boot_addr = rproc_elf_get_boot_addr;
    return 0;
}

最终调用hobot_vdsp_rproc_kick函数

static void hobot_vdsp_rproc_kick(struct rproc *localrproc, int32_t vqid)
{
    struct hobot_rproc_vdsp_pdata *pdata = localrproc->priv;
    pdata->ipc_ops.trigger_interrupt(pdata);
}
static void vdsp_trigger_interrupt(void *priv)
{...
    if (pdata->vdsp_is_stop_flag != 1) {
        ret = mbox_send_message(pdata->mchan_rpmsg, tmp_data);
        if (ret < 0) {
            (void)dev_err(&pdata->hbrproc->dev, "[vdsp%d] mailbox notify failed: %d
",
                        pdata->device_index, ret);
        }
    } ...
}

其中mbox_send_message是mailbox提供的接口，最终回调到mailbox层

最后来看下在remoteproc这层是如何最为一个mailbox client使用的，在hobot-drivers/remoteproc/hobot_remoteproc.c，对实际物理通道进行回调绑定：

hobot_remoteproc_probe
irq_init
static int32_t irq_init(struct platform_device *pdev)
{
    //coverity[misra_c_2012_rule_11_5_violation:SUPPRESS], ## violation reason SYSSW_V_11.5_01
    struct hobot_rproc_vdsp_pdata *pdata = (struct hobot_rproc_vdsp_pdata *)(
    pdev->dev.driver_data);
    int32_t dsp_id = pdata->device_index;
    char str_vdsp_rpmsg_wq[VDSP_PATH_SIZE] = {0};
    char str_vdsp_dump_wq[VDSP_PATH_SIZE] = {0};
    char str_vdsp_async_wq[VDSP_PATH_SIZE] = {0};
    int32_t ret = 0;

    pdata->mclient_rpmsg.dev = &(pdev->dev);
    pdata->mclient_rpmsg.tx_block = (bool)true;
    pdata->mclient_rpmsg.tx_tout = MBOX_TIMEOUT;
    pdata->mclient_rpmsg.knows_txdone = (bool)false;
    pdata->mclient_rpmsg.rx_callback = hobot_rpmsg_handler;
    pdata->mclient_rpmsg.tx_prepare = NULL;
    pdata->mclient_rpmsg.tx_done = NULL;
    pdata->mchan_rpmsg = mbox_request_channel(&pdata->mclient_rpmsg, 0);
    if (IS_ERR(pdata->mchan_rpmsg)) {
        dev_err(&pdev->dev, "vdsp%d get mailbox channel:rpmsg failed
", dsp_id);
        return -EFAULT;
    }
    ...
    pdata->mclient_boot.dev = &(pdev->dev);
    pdata->mclient_boot.tx_block = (bool)true;
    pdata->mclient_boot.tx_tout = MBOX_TIMEOUT;
    pdata->mclient_boot.knows_txdone = (bool)false;
    pdata->mclient_boot.rx_callback = hobot_boot_handler;
    pdata->mclient_boot.tx_prepare = NULL;
    pdata->mclient_boot.tx_done = NULL;
    pdata->mchan_boot = mbox_request_channel(&pdata->mclient_boot, 1);
    if (IS_ERR(pdata->mchan_boot)) {
        mbox_free_channel(pdata->mchan_rpmsg);
        dev_err(&pdev->dev, "vdsp%d get mailbox channel:boot failed
", dsp_id);
        return -EFAULT;
}
。。。
       pdata->mclient_log.dev = &(pdev->dev);
    pdata->mclient_log.tx_block = (bool)true;
    pdata->mclient_log.tx_tout = MBOX_TIMEOUT;
    pdata->mclient_log.knows_txdone = (bool)false;
    pdata->mclient_log.rx_callback = hobot_log_handler;
    pdata->mclient_log.tx_prepare = NULL;
    pdata->mclient_log.tx_done = NULL;
    pdata->mchan_log = mbox_request_channel(&pdata->mclient_log, 2);
    if (IS_ERR(pdata->mchan_log)) {
        mbox_free_channel(pdata->mchan_rpmsg);
        mbox_free_channel(pdata->mchan_boot);
        dev_err(&pdev->dev, "vdsp%d get mailbox channel:log failed
", dsp_id);
        return -EFAULT;
    }
...
}

5.4 mailbox 框架

5.4.1 mailbox源码介绍

mailbox源码在Linux源码下的kernel/drivers/mailbox/

kernel/drivers/mailbox/mailbox.c: mailbox框架核心代码

hobot-drivers/mailbox/mbox/os/kernel/mbox_os.c：mailbox 硬件模块代码，注册实现controller。

mailbox为外层提供了controller和client两者，controller为mailbox框架提供硬件操作接口，client作为上层使用方，将通过controller来调用硬件操作相关接口，如下是mailbox软件框架图：

Mailbox内部维护了controller控制队列，外层通过接口注册的controller将被放到该队列中，这意味者，mailbox 框架支持多个硬件mailbox同时接入，用户作为client时，将绑定每个硬件chan的回调，下层收到chan数据时，将根据chan的编号，调用不同注册回调，在数据接收上，当mailbox硬件模块收到数据时进入中断，此时调用mbox_chan_received_data通知mailbox框架，mailbox框架将进行上层回调直到将数据上送给上层。在发送数据时，上层将调用mailbox的mbox_send_message通知mailbox框架发送中断给远端，实际该接口将回调controller注册的回调硬件接口。

5.4.2 mailbox框架源码解析

Mailbox的client端在上节已经分析了，来看下mailbox controller是如何注册的，mailbox的controller由硬件驱动模块来注册hobot-drivers/mailbox/mbox/os/kernel/mbox_os.c：

static int32_t hb_pl320_probe(struct platform_device *pdev)
{
    int32_t err;
    struct device *dev = &pdev->dev;
    struct pl320_mbox *mbox;
    struct hb_pl320_dev *mbox_pm = NULL;
   
    mbox->dev = dev;
    mbox->controller.dev = dev;
    mbox->controller.chans = mbox->chan;
    mbox->controller.num_chans = MAX_CHAN_PER_IPCM;
    mbox->controller.ops = &pl320_chan_ops;
    mbox->controller.of_xlate = pl320_chan_xlate;
    mbox->controller.txdone_irq = true;
    mbox->controller.txdone_poll = false;
    mbox_pm = (struct hb_pl320_dev *)dev_get_drvdata(&pdev->dev);
mbox_pm->mbox = mbox;
...
    err = devm_mbox_controller_register(dev, &mbox->controller);
    if (err) {
        dev_err(dev, "Failed to register mailboxes %d
", err);
        iounmap((void *)mbox->base);
        return err;
}
...
    dev_info(dev, "PL320 Mailbox registered
");
    return 0;
}

struct mbox_chan_ops pl320_chan_ops = {
    .send_data = pl320_send_data,
    .startup = pl320_startup,
    .shutdown = pl320_shutdown,
    .flush = pl320_flush,
};

同时提供了硬件操作回调集：pl320_chan_ops

以发送数据为例，前面分析到在remoproce发送数据回调的kick，最终调用到

vdsp_trigger_interrupt -> mbox_send_message

mbox_send_message定义在drivers/mailbox/mailbox.c

int mbox_send_message(struct mbox_chan *chan, void *mssg)
{
    int t;
    if (!chan || !chan->cl)
        return -EINVAL;
    t = add_to_rbuf(chan, mssg);
    if (t < 0) {
        dev_err(chan->mbox->dev, "Try increasing MBOX_TX_QUEUE_LEN
");
        return t;
    }
    msg_submit(chan);
...
return t;
}
static void msg_submit(struct mbox_chan *chan)
{
    ....
    data = chan->msg_data[idx];
    if (chan->cl->tx_prepare)
        chan->cl->tx_prepare(chan->cl, data);
    /* Try to submit a message to the MBOX controller */
    err = chan->mbox->ops->send_data(chan, data);//回调controller的回调，即硬件接口
    if (!err) {
        chan->active_req = data;
        chan->msg_count--;
    }
...
}

根据前面分析的controller注册生成，可见此时调到mailbox提供的硬件操作接口。

数据发送过程中，mailbox先注册硬件中断和绑定中断服务函数hobot-drivers/mailbox/mbox/os/kernel/mbox_os.c

hb_pl320_probe
static int32_t get_mbox_dev_resource(struct platform_device *pdev, struct pl320_mbox *mbox)
{
...
for (i = 0; i < mbox->num_mbox_chan; i++){
        mchan = pchan + i;
        mchan->irq = (uint32_t)platform_get_irq(pdev, (uint32_t)i);
        mchan->parent = mbox;
        mbox->chan[i + offset].con_priv = (void *)mchan;
...
        err = devm_request_threaded_irq(dev, mchan->irq,
                        hb_mbox_handler,
                        hb_mbox_softirq,
                        IRQF_ONESHOT,
                        dev_name(dev),
                        (void *)&mbox->chan[i + offset]);
        if (err) {
...
            return err;
        }
        spin_lock_init(&mchan->irq_spinlock);
        spin_lock_init(&mchan->flush_spinlock);
        mutex_init(&mchan->client_lock);
    }
}
static irqreturn_t hb_mbox_softirq(int irq, void *arg)
{
...
(void)mbox_chan_received_data(chan, mchan->rx_data);
...
}

在中断服务函数里，调用mailbox框架接口 mbox_chan_received_data ，通知上层接收数据

void mbox_chan_received_data(struct mbox_chan *chan, void *mssg)
{
    /* No buffering the received data */
    if (chan->cl->rx_callback)
        chan->cl->rx_callback(chan->cl, mssg);
}

回调client的rx_callback，就是前面分析的绑定通道回调过程。

6 收发数据流程总结

接下来整体过一遍收发数据的过程。

6.1 接收数据过程调用栈

接收数据过程，以J6平台代码流程为例，从硬件接收->上层接收逐层分析。

hb_pl320_probe
get_mbox_dev_resource 注册中断服务hb_mbox_softirq
hb_mbox_softirq
mbox_chan_received_data 回调client 的rx_callback->hobot_rpmsg_handler
hobot_rpmsg_handler 通知线程completion_notify_rpmsg
hobot_remoteproc_rpmsg_work线程被唤醒
hobot_remoteproc_rpmsg_work_cb
rproc_vq_interrupt
vring_interrupt 最终调用struct virtqueue vq 里的callback回调，该回调在初始化struct virtqueue时传入
rpmsg_recv_done（rpmsg_probe->virtio_find_vqs->(vdev->config->find_vqs) rproc_virtio_find_vqs->rp_find_vq
->vring_new_virtqueue->__vring_new_virtqueue-> vq->vq.callback = callback）
rpmsg_recv_single 调用端点绑定的回调ept->cb，该回调在端点初始化时传入
rpmsg_ept_cb（rpmsg_eptdev_open->rpmsg_create_ept时传入了该回调）
最终将buf放入队列skb_queue_tail(&eptdev->queue, skb);
上层调用read时，rpmsg_eptdev_read_iter被调用
skb = skb_dequeue(&eptdev->queue);//冲skb队列取数据

下图是stm32-157平台下的接收数据过程

6.2 发送数据过程调用栈

数据发送过程，同样先以J6平台为例，这次从上层开始往下分析

应用层read
rpmsg_eptdev_write_iter
rpmsg_sendto 最终调用端点绑定的send ept->ops->sendto，由端点初始化是传入
virtio_rpmsg_sendto （rpmsg_eptdev_open->rpmsg_create_ept->virtio_rpmsg_create_ept->__rpmsg_create_ept->virtio_endpoint_ops->virtio_rpmsg_sendto）
rpmsg_send_offchannel_raw
virtqueue_add_outbuf
virtqueue_kick
virtqueue_notify 调用vq->notify回调，该回调在初始化struct virtqueue时传入
rproc_virtio_notify（rpmsg_probe->virtio_find_vqs->(vdev->config->find_vqs) rproc_virtio_find_vqs->rp_find_vq
->vring_new_virtqueue->传入rproc_virtio_notify），该函数最终调用struct rproc *rproc的ops下kick回调。
hobot_vdsp_rproc_kick（hobot_remoteproc_probe->rproc_alloc传入hobot_vdsp_rproc_ops->rproc_alloc_ops->rproc->ops = kmemdup(ops, sizeof(*ops), GFP_KERNEL);）
trigger_interrupt:vdsp_trigger_interrupt
mbox_send_message
msg_submit:chan->mbox->ops->send_data
pl320_chan_ops：pl320_send_data 写寄存器，触发对端中断通知对端

下面是stm32-157的写调用栈流程：

6.3 收发数据buffer处理过程

6.3.1  J6缓冲区处理流程

1. 共享缓冲区的获取以及virtio_queue 创建

queue的初始化kernel/drivers/rpmsg/virtio_rpmsg_bus.c的probe发起的
rpmsg_probe -> virtio_find_vqs 该函数回调到virtio device的config集回调rproc_virtio_find_vqs
rproc_virtio_find_vqs ->rp_find_vq->vring_new_virtqueue，这个vring_new_virtqueue传入了共享内存地址，用来建立vring buffer缓冲区：
static struct virtqueue *rp_find_vq(struct virtio_device *vdev,
                    unsigned int id,
                    void (*callback)(struct virtqueue *vq),
                    const char *name, bool ctx)
{
struct rproc_vdev *rvdev = vdev_to_rvdev(vdev);
void *addr;
...
rvring = &rvdev->vring[id];
    num = rvring->num;
#if defined(CONFIG_HOBOT_REMOTEPROC) || defined(CONFIG_HOBOT_REMOTEPROC_MODULE)
    if (rproc->fix_map_mode) {
        addr = rvring->va;
        memset_io(addr, 0, size);
    } else {
        addr = mem->va;
        memset(addr, 0, size);
    }
#else
    addr = mem->va;
    memset(addr, 0, size);
#endif
...
vq = vring_new_virtqueue(id, num, rvring->align, vdev, false, ctx,
                 addr, rproc_virtio_notify, callback, name);
//从上面流程代码，推出addr来自vdev指向的rvdev->vring[id]
//vdev是struct virtio_device，virtio device由下层注册
}
Virtio Device  注册流程参考接下来是device的注册过程
在rproc_virtio_probe中实现了virtio device指向struct rproc_vdev，以及vdev里vring数据来源
static int rproc_virtio_probe(struct platform_device *pdev)
{
    struct device *dev = &pdev->dev;
    struct rproc_vdev_data *rvdev_data = dev->platform_data;
    struct rproc_vdev *rvdev;
    struct rproc *rproc = container_of(dev->parent, struct rproc, dev);
    struct fw_rsc_vdev *rsc;
    int i, ret;

    if (!rvdev_data)
        return -EINVAL;
    rvdev = devm_kzalloc(dev, sizeof(*rvdev), GFP_KERNEL);
    if (!rvdev)
        return -ENOMEM;
    rvdev->id = rvdev_data->id;
    rvdev->rproc = rproc;
rvdev->index = rvdev_data->index;
...
rsc = rvdev_data->rsc;// rsc数据来自platform device私有数据
...
platform_set_drvdata(pdev, rvdev);
    rvdev->pdev = pdev;
...
    /* parse the vrings */
    for (i = 0; i < rsc->num_of_vrings; i++) {
        ret = rproc_parse_vring(rvdev, rsc, i);
        if (ret)
            return ret;
    }
}
再看rproc_parse_vring,可知vring数据来自rsc
int
rproc_parse_vring(struct rproc_vdev *rvdev, struct fw_rsc_vdev *rsc, int i)
{
    struct rproc *rproc = rvdev->rproc;
    struct device *dev = &rproc->dev;
    struct fw_rsc_vdev_vring *vring = &rsc->vring[i];
    struct rproc_vring *rvring = &rvdev->vring[i];
  ....
    rvring->num = vring->num;
    rvring->align = vring->align;
    rvring->rvdev = rvdev;
#if defined(CONFIG_HOBOT_REMOTEPROC) || defined(CONFIG_HOBOT_REMOTEPROC_MODULE)
    if (rproc->fix_map_mode) {
        rvring->vring_size = vring->vring_size;
        rvring->da = vring->da;
        if (i == 0)
            rvdev->buffer_phy_addr = vring->da +
            rsc->num_of_vrings * vring->vring_size;
    }
    else {
        rvdev->buffer_phy_addr = 0;
    }
#endif
    return 0;
}
从上面分析知，vring的缓冲区地址，来自virtio device的platform_data平台数据里的rsc
分析platfrom里rsc数据由来
static int rproc_fw_boot(struct rproc *rproc, const struct firmware *fw)
{
    struct device *dev = &rproc->dev;
    const char *name = rproc->firmware;
    ...
    /* Load resource table, core dump segment list etc from the firmware */
    ret = rproc_parse_fw(rproc, fw);//解析固件头信息，这里包含了resource_table，记录了共享内存地址
    ....
/* handle fw resources which are required to boot rproc */
//调用rproc_loading_handlers所有回调，其中rproc_handle_vdev出入了上述分析platfrom里rsc
    ret = rproc_handle_resources(rproc, rproc_loading_handlers);
    if (ret) {
        dev_err(dev, "Failed to process resources: %d
", ret);
        goto clean_up_resources;
    }
    ....
    return ret;
}
//下面函数传入了platform的private数据，该函数被上面函数调用
static int rproc_handle_vdev(struct rproc *rproc, void *ptr,
                 int offset, int avail)
{
    struct fw_rsc_vdev *rsc = ptr;
    ...
    rvdev_data.id = rsc->id;
    rvdev_data.index = rproc->nb_vdev++;
    rvdev_data.rsc_offset = offset;
    rvdev_data.rsc = rsc;
    /*
     * When there is more than one remote processor, rproc->nb_vdev number is
     * same for each separate instances of "rproc". If rvdev_data.index is used
     * as device id, then we get duplication in sysfs, so need to use
     * PLATFORM_DEVID_AUTO to auto select device id.
     */
    pdev = platform_device_register_data(dev, "rproc-virtio", PLATFORM_DEVID_AUTO, &rvdev_data,
                         sizeof(rvdev_data));
    ...
    return 0;
}
/* handle firmware resource entries before booting the remote processor */
static int rproc_handle_resources(struct rproc *rproc,
                  rproc_handle_resource_t handlers[RSC_LAST])
{
    struct device *dev = &rproc->dev;
    rproc_handle_resource_t handler;
    int ret = 0, i;
    if (!rproc->table_ptr)
        return 0;
    for (i = 0; i < rproc->table_ptr->num; i++) {
        int offset = rproc->table_ptr->offset[i];
        struct fw_rsc_hdr *hdr = (void *)rproc->table_ptr + offset;//rsc的数据最终来自rporc->table_ptr.
        int avail = rproc->table_sz - offset - sizeof(*hdr);
        void *rsc = (void *)hdr + sizeof(*hdr);
...
        if (hdr->type >= RSC_VENDOR_START &&
            hdr->type <= RSC_VENDOR_END) {
            ...
        handler = handlers[hdr->type];
        if (!handler)
            continue;
        ret = handler(rproc, rsc, offset + sizeof(*hdr), avail);
        if (ret)
            break;
    }
    return ret;
}
从上面函数知道了rsc数据最终来自table_ptr，该数据由固件解析函数获取
rproc_parse_fw->rproc_elf_load_rsc_table

1. 下面流程图是整个buffer数据流程图，在获取到共享内存，初始化到vring后，各层可利用vring，进行在RPMsg驱动层之前的无拷贝动作。

6.3.2 stm32-mp157缓冲区处理流程

下图流程图，总结了stm32-mp157的缓冲区处理流程，相对于j6平台，对共享内存的处理简单了很多，直接从设备树读取预留内存后，配置初始化到vring中，其他流程均和j6平台一致。

6.3.3 缓冲区处理

在RPMsg框架中，virito的vring缓冲区，单个最大默认是512K，如果分配的内存大于512K，virtio_queue会将其分割成几个缓冲区块，记录在table里，这意味着用户每次单次都不能超过512K。下图是实际处理过程示意图，实际过程中，virtio driver（virtio_rpmsg_bus.c）会创建两个virtio_queue，一个用于发送，一个用于接收。

7 附录

Linux RPMsg framework overview - stm32mpu

J6 document

kernel/drivers/mailbox/mailbox-test.c

kernel/tools/virtio/virtio_test.c

kernel/tools/virtio/vringh_test.c