Virtio Rpmsg Bus

---

时间轴

时间轴

2026-06-22

init

本文介绍了Linux内核中基于virtio的rpmsg总线驱动（virtio_rpmsg_bus），阐述了其作为rpmsg总线的virtio传输实现，负责在Linux主核与远端处理器之间通过vring收发消息，并将远端服务抽象为rpmsg_device供上层驱动绑定。文章详细分析了rpmsg的三层结构、通信模型，以及核心数据结构如rpmsg_device、rpmsg_endpoint和virtproc_info，并说明了endpoint的创建与销毁、缓冲区管理、发送同步和名字服务等关键机制。最后指出rpmsg实际是基于地址的消息分发机制。

---

linux 5.10.238

rpmsg.drawio

rpmsg 结构

rpmsg 有三层：

┌────────────────────────────────────────┐
│ 业务层：rpmsg_driver                     │  ← 写业务的地方
│ （rpmsg_tty / rpmsg_chrdev / 自定义）    │
├────────────────────────────────────────┤
│ rpmsg 总线 / core：rpmsg_core.c         │  ← 匹配 device 和 driver
│ rpmsg_device / rpmsg_endpoint          │
├────────────────────────────────────────┤
│ transport 后端：virtio_rpmsg_bus.c      │  ← 本博客重点分析的代码
│ （virtio_driver，搬运消息、造 channel）   │
├────────────────────────────────────────┤
│ virtio 总线 / vring                     │
└────────────────────────────────────────┘

virtio_rpmsg_bus.c在中下层，它的职责是：

1. 接管 virtio rpmsg 设备
2. 用 vring 收发消息
3. 根据 name service 创建/销毁 rpmsg_device

它生产rpmsg_device，但不消费。消费者是rpmsg_driver。

module_init/module_exit

static struct virtio_device_id id_table[] = {
    { VIRTIO_ID_RPMSG, VIRTIO_DEV_ANY_ID },
    { 0 },
};

static unsigned int features[] = {
    VIRTIO_RPMSG_F_NS,
};

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)
{
    int ret;

    ret = register_virtio_driver(&virtio_ipc_driver);
    if (ret)
        pr_err("failed to register virtio driver: %d\n", ret);

    return ret;
}
subsys_initcall(rpmsg_init);

static void __exit rpmsg_fini(void)
{
    unregister_virtio_driver(&virtio_ipc_driver);
}
module_exit(rpmsg_fini);

MODULE_DEVICE_TABLE(virtio, id_table);
MODULE_DESCRIPTION("Virtio-based remote processor messaging bus");
MODULE_LICENSE("GPL v2");

virtio_rpmsg_bus.c是 Linux 内核里 基于 virtio 的 rpmsg 总线驱动。它的作用是：

让 Linux 主核和远端处理器之间，通过 virtio vring 机制收发 rpmsg 消息，并把远端服务抽象成 Linux 的rpmsg_device，供上层 rpmsg driver 绑定使用。
这个文件本身是 rpmsg bus 的 virtio transport 实现，它不关心具体业务协议，比如音频、传感器、TEE、MCU 控制等；它只负责把消息送到对应 endpoint。

基本通信模型

rpmsg

rpmsg 的通信单元是：

• rpmsg_devicerpmsg channel device, 一条通信通道（channel），是“设备”
• rpmsg_driver处理这条通道业务的驱动
• rpmsg_endpoint通道上的实际收发端点（地址 + callback）

三者之间关系:

rpmsg_driver  ←—— 匹配 ——→  rpmsg_device
                                 |
                                 | 持有
                                 v
                            rpmsg_endpoint

rpmsg_device 就是中间那个“被驱动管理的设备”。当创建rpmsg_device时会触发 probe 函数进行匹配，匹配成功后调用rpmsg_driver->probe函数

struct rpmsg_device

struct rpmsg_device
   ├── u32 src
   ├── u32 dst
   ├── bool announce
   ├── struct device dev
   ├── struct rpmsg_device_id id;
   ├── const char *driver_override;
   ├── const struct rpmsg_device_ops *ops;
   └── struct rpmsg_endpoint *ept
               ├── u32 addr
               ├── void *priv
               ├── rpmsg_rx_cb_t cb
               ├── struct mutex cb_lock
               ├── struct kref refcount
               ├── struct rpmsg_device *rpdev
               └── const struct rpmsg_endpoint_ops *ops 

其中struct rpmsg_device定义如下：

/**
 * rpmsg_device - device that belong to the rpmsg bus
 * @dev: the device struct
 * @id: device id (used to match between rpmsg drivers and devices)
 * @driver_override: driver name to force a match; do not set directly,
 *                   because core frees it; use driver_set_override() to
 *                   set or clear it.
 * @src: local address
 * @dst: destination address
 * @ept: the rpmsg endpoint of this channel
 * @announce: if set, rpmsg will announce the creation/removal of this channel
 */
struct rpmsg_device {
    struct device dev;
    struct rpmsg_device_id id;
    const char *driver_override;
    u32 src;
    u32 dst;
    struct rpmsg_endpoint *ept;
    bool announce;

    const struct rpmsg_device_ops *ops;
};

struct rpmsg_device_ops表示一个rpmsg_device的 operation

/**
* struct rpmsg_device_ops - indirection table for the rpmsg_device operations
* @create_ept:		create backend-specific endpoint, required
* @announce_create:	announce presence of new channel, optional
* @announce_destroy:	announce destruction of channel, optional
*
* Indirection table for the operations that a rpmsg backend should implement.
* @announce_create and @announce_destroy are optional as the backend might
* advertise new channels implicitly by creating the endpoints.
*/
struct rpmsg_device_ops {
    struct rpmsg_endpoint *(*create_ept)(struct rpmsg_device *rpdev,
                        rpmsg_rx_cb_t cb, void *priv,
                        struct rpmsg_channel_info chinfo);

    int (*announce_create)(struct rpmsg_device *ept);
    int (*announce_destroy)(struct rpmsg_device *ept);
};

一个 rpmsg_device 等价于一条逻辑通道，比如：

• channel name: “rpmsg-demo”
• src: 本地 endpoint 地址
• dst: 远端 endpoint 地址

在virtio_rpmsg_bus.c里，channel 是这样创建的（rpmsg_create_channel()）：

rpdev->src = chinfo->src;
rpdev->dst = chinfo->dst;
rpdev->ops = &virtio_rpmsg_ops;
strncpy(rpdev->id.name, chinfo->name, RPMSG_NAME_SIZE);

也就是说：远端 announce 一个服务 “rpmsg-demo”，Linux 创建一个rpmsg_device代表这条 channel；或者本地主动创建一个服务 “rpmsg-demo”，即创建一个rpmsg_device代表本地 channel。announce 之后，由 NS 服务通知远端也创建 “rpmsg-demo” 的 channel。本地和远端的rpmsg_device的 src 和 dst 彼此相反，通过这条通道进行通信。

实际上virtio_rpmsg_bus.c中抽象的rpmsg channel descriptor为

/**
 * struct virtio_rpmsg_channel - rpmsg channel descriptor
 * @rpdev: the rpmsg channel device
 * @vrp: the virtio remote processor device this channel belongs to
 *
 * This structure stores the channel that links the rpmsg device to the virtio
 * remote processor device.
 */
struct virtio_rpmsg_channel {
    struct rpmsg_device rpdev;

    struct virtproc_info *vrp;
};

即一个rpmsg_device还要加上核心数据结构struct virtproc_info *vrp，这是 virtio 实现 rpmsg bus 的关键数据结构

struct rpmsg_endpoint

rpmsg_device就像一根电话线 + 一个分机号服务，而rpmsg_endpoint代表实际接听电话的人

/**
 * struct rpmsg_endpoint - binds a local rpmsg address to its user
 * @rpdev: rpmsg channel device
 * @refcount: when this drops to zero, the ept is deallocated
 * @cb: rx callback handler
 * @cb_lock: must be taken before accessing/changing @cb
 * @addr: local rpmsg address
 * @priv: private data for the driver's use
 *
 * In essence, an rpmsg endpoint represents a listener on the rpmsg bus, as
 * it binds an rpmsg address with an rx callback handler.
 *
 * Simple rpmsg drivers shouldn't use this struct directly, because
 * things just work: every rpmsg driver provides an rx callback upon
 * registering to the bus, and that callback is then bound to its rpmsg
 * address when the driver is probed. When relevant inbound messages arrive
 * (i.e. messages which their dst address equals to the src address of
 * the rpmsg channel), the driver's handler is invoked to process it.
 *
 * More complicated drivers though, that do need to allocate additional rpmsg
 * addresses, and bind them to different rx callbacks, must explicitly
 * create additional endpoints by themselves (see rpmsg_create_ept()).
 */
struct rpmsg_endpoint {
    struct rpmsg_device *rpdev;
    struct kref refcount;
    rpmsg_rx_cb_t cb;
    struct mutex cb_lock;
    u32 addr;
    void *priv;

    const struct rpmsg_endpoint_ops *ops;
};

rpmsg_endpoint相关逻辑主要在rpmsg_core.c中，其中最重要的是rpmsg_rx_cb_t cb和const struct rpmsg_endpoint_ops *ops这两个成员：

typedef int (*rpmsg_rx_cb_t)(struct rpmsg_device *, void *, int, void *, u32);

该函数表示该ept收到消息后触发的回调函数

/**
 * struct rpmsg_endpoint_ops - indirection table for rpmsg_endpoint operations
 * @destroy_ept:	see @rpmsg_destroy_ept(), required
 * @send:		see @rpmsg_send(), required
 * @sendto:		see @rpmsg_sendto(), optional
 * @send_offchannel:	see @rpmsg_send_offchannel(), optional
 * @trysend:		see @rpmsg_trysend(), required
 * @trysendto:		see @rpmsg_trysendto(), optional
 * @trysend_offchannel:	see @rpmsg_trysend_offchannel(), optional
 * @poll:		see @rpmsg_poll(), optional
 *
 * Indirection table for the operations that a rpmsg backend should implement.
 * In addition to @destroy_ept, the backend must at least implement @send and
 * @trysend, while the variants sending data off-channel are optional.
 */
struct rpmsg_endpoint_ops {
    void (*destroy_ept)(struct rpmsg_endpoint *ept);

    int (*send)(struct rpmsg_endpoint *ept, void *data, int len);
    int (*sendto)(struct rpmsg_endpoint *ept, void *data, int len, u32 dst);
    int (*send_offchannel)(struct rpmsg_endpoint *ept, u32 src, u32 dst,
                  void *data, int len);

    int (*trysend)(struct rpmsg_endpoint *ept, void *data, int len);
    int (*trysendto)(struct rpmsg_endpoint *ept, void *data, int len, u32 dst);
    int (*trysend_offchannel)(struct rpmsg_endpoint *ept, u32 src, u32 dst,
                 void *data, int len);
    __poll_t (*poll)(struct rpmsg_endpoint *ept, struct file *filp,
                 poll_table *wait);
};

表示 rpmsg_endpoint 的 operation 函数。

endpoint 创建

核心函数是：__rpmsg_create_ept(),它是rpmsg_device的operation函数

/* for more info, see below documentation of 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;
    struct device *dev = rpdev ? &rpdev->dev : &vrp->vdev->dev;

    ept = kzalloc(sizeof(*ept), GFP_KERNEL);
    if (!ept)
        return NULL;

    kref_init(&ept->refcount);
    mutex_init(&ept->cb_lock);

    ept->rpdev = rpdev;
    ept->cb = cb;
    ept->priv = priv;
    ept->ops = &virtio_endpoint_ops;

    /* do we need to allocate a local address ? */
    if (addr == RPMSG_ADDR_ANY) {
        id_min = RPMSG_RESERVED_ADDRESSES;
        id_max = 0;
    } else {
        id_min = addr;
        id_max = addr + 1;
    }

    mutex_lock(&vrp->endpoints_lock);

    /* bind the endpoint to an rpmsg address (and allocate one if needed) */
    id = idr_alloc(&vrp->endpoints, ept, id_min, id_max, GFP_KERNEL);
    if (id < 0) {
        dev_err(dev, "idr_alloc failed: %d\n", id);
        goto free_ept;
    }
    ept->addr = id;

    mutex_unlock(&vrp->endpoints_lock);

    return ept;

free_ept:
    mutex_unlock(&vrp->endpoints_lock);
    kref_put(&ept->refcount, __ept_release);
    return NULL;
}

static struct rpmsg_endpoint *virtio_rpmsg_create_ept(struct rpmsg_device *rpdev,
                              rpmsg_rx_cb_t cb,
                              void *priv,
                              struct rpmsg_channel_info chinfo)
{
    struct virtio_rpmsg_channel *vch = to_virtio_rpmsg_channel(rpdev);

    return __rpmsg_create_ept(vch->vrp, rpdev, cb, priv, chinfo.src);
}

1. 分配struct rpmsg_endpoint
2. 初始化引用计数和 callback lock
3. 记录 callback、private data、ops
4. 给 endpoint 分配本地地址
5. 插入vrp->endpoints这个 idr

关键逻辑：

if (addr == RPMSG_ADDR_ANY) {
    id_min = RPMSG_RESERVED_ADDRESSES;
    id_max = 0;
} else {
    id_min = addr;
    id_max = addr + 1;
}

• 如果调用方不指定地址，则动态分配
• 动态地址从1024开始
• 0 ~ 1023保留给预定义服务
  而保留地址定义：

#define RPMSG_RESERVED_ADDRESSES	(1024)

endpoint 地址通过idr_alloc()分配：

id = idr_alloc(&vrp->endpoints, ept, id_min, id_max, GFP_KERNEL);

endpoint 销毁

核心函数：__rpmsg_destroy_ept()，它是rpmsg_endpoint的operation函数

1. 从idr中删除 endpoint
2. 把 callback 设置成NULL
3. 减引用计数，必要时释放 endpoint

关键代码：

/**
 * __rpmsg_destroy_ept() - destroy an existing rpmsg endpoint
 * @vrp: virtproc which owns this ept
 * @ept: endpoing to destroy
 *
 * An internal function which destroy an ept without assuming it is
 * bound to an rpmsg channel. This is needed for handling the internal
 * name service endpoint, which isn't bound to an rpmsg channel.
 * See also __rpmsg_create_ept().
 */
static void
__rpmsg_destroy_ept(struct virtproc_info *vrp, struct rpmsg_endpoint *ept)
{
    /* make sure new inbound messages can't find this ept anymore */
    mutex_lock(&vrp->endpoints_lock);
    idr_remove(&vrp->endpoints, ept->addr);
    mutex_unlock(&vrp->endpoints_lock);

    /* make sure in-flight inbound messages won't invoke cb anymore */
    mutex_lock(&ept->cb_lock);
    ept->cb = NULL;
    mutex_unlock(&ept->cb_lock);

    kref_put(&ept->refcount, __ept_release);
}

这里特别注意并发：

• 删除 idr：防止新的 RX 消息找到这个 endpoint
• 设置cb = NULL：防止已经拿到 endpoint 的 in-flight RX 调用 callback
• kref：防止 RX 路径正在使用 endpoint 时被释放

struct virtproc_info

核心数据结构：struct virtproc_info

struct virtproc_info {
    struct virtio_device *vdev;
    struct virtqueue *rvq, *svq;
    void *rbufs, *sbufs;
    unsigned int num_bufs;
    unsigned int buf_size;
    int last_sbuf;
    dma_addr_t bufs_dma;
    struct mutex tx_lock;
    struct idr endpoints;
    struct mutex endpoints_lock;
    wait_queue_head_t sendq;
    atomic_t sleepers;
    struct rpmsg_endpoint *ns_ept;
};

这是整个 virtio rpmsg 设备的私有状态，挂在vdev->priv = vrp中, 按功能划分

• virtio 基础：
  • vdev底层 virtio 设备
• 收发通道：
  • rvq,svqRX/TX virtqueue
• buffer 管理：
  • rbufs,sbufsRX/TX buffer 虚拟地址
  • num_bufs总 buffer 数
  • buf_size单 buffer 大小
  • last_sbufTX 开荒游标
  • bufs_dmabuffer DMA 基址
• 发送同步：
  • tx_lock保护 svq/sbufs/sleepers
  • sendq等待 TX buffer 的等待队列
  • sleepers等待者计数（控制 tx-complete 中断开关）
• endpoint 管理：
  • endpointsendpoint idr（按地址查找）
  • endpoints_lock保护 endpoint 表
• name service：
  • ns_ept名字服务 endpoint（addr 53）

struct virtio_device *vdev

作用：指向底层的 virtio 设备。它是这个 rpmsg 实例的根。通过它可以拿到：
- vdev->dev 设备节点（dev_err/dev_dbg 用）
- vdev->config virtio 配置操作（reset/del_vqs 等）
- vdev->priv 反向指回 vrp 自己
probe 里设置：

vrp->vdev = vdev;
...
vdev->priv = vrp;

callback 里就靠它取回 vrp：

struct virtproc_info *vrp = rvq->vdev->priv;

struct virtqueue *rvq, *svq

作用：

• rvq = receive virtqueue：远端 -> Linux
• svq = send virtqueue： Linux -> 远端
  这是收发的两条通道。probe 里：

vrp->rvq = vqs[0];   /* input  */
vrp->svq = vqs[1];   /* output */

• 接收消息：virtqueue_get_buf(rvq)/virtqueue_add_inbuf(rvq)
• 发送消息：virtqueue_add_outbuf(svq)/virtqueue_get_buf(svq)

void *rbufs, *sbufs

作用：

• rbufs：RX buffer 区域的内核虚拟地址起点

• sbufs：TX buffer 区域的内核虚拟地址起点

  probe 里把一整块 DMA 内存切两半：

vrp->rbufs = bufs_va;                     /* 前半 RX */
vrp->sbufs = bufs_va + total_buf_space / 2; /* 后半 TX */

如图

bufs_va
  ├── rbufs：RX[0] RX[1] ...
  └── sbufs：TX[0] TX[1] ...

get_a_tx_buf()就是从 sbufs 上按索引取 TX buffer：

ret = vrp->sbufs + vrp->buf_size * vrp->last_sbuf++;

unsigned int num_bufs

作用：RX + TX 的 buffer 总数（两者各占一半）。probe 里根据 vring 大小计算：

if (virtqueue_get_vring_size(vrp->rvq) < MAX_RPMSG_NUM_BUFS / 2)
      vrp->num_bufs = virtqueue_get_vring_size(vrp->rvq) * 2;
else
      vrp->num_bufs = MAX_RPMSG_NUM_BUFS;  /* 512 */

所以：

RX buffer 数 = num_bufs / 2
TX buffer 数 = num_bufs / 2

多处都用num_bufs / 2作为边界，例如get_a_tx_buf()：

if (vrp->last_sbuf < vrp->num_bufs / 2)

unsigned int buf_size

作用：单个 buffer 的字节数，当前固定 512（MAX_RPMSG_BUF_SIZE）。

probe函数中

vrp->buf_size = MAX_RPMSG_BUF_SIZE;

它决定：

• 每个 buffer 的地址步长：sbufs + buf_size * i
• 单条消息 payload 上限

if (len > vrp->buf_size - sizeof(struct rpmsg_hdr))
      return -EMSGSIZE;

int last_sbuf

作用：TX buffer 的“开荒游标”，记录初次顺序分配到第几个。在get_a_tx_buf()里用：

if (vrp->last_sbuf < vrp->num_bufs / 2)
      ret = vrp->sbufs + vrp->buf_size * vrp->last_sbuf++;
else
      ret = virtqueue_get_buf(vrp->svq, &len);

• last_sbuf < num_bufs/2：还有没用过的新 TX buffer，按序拿
• 否则：所有 TX buffer 都用过，改从 used ring 回收复用
  一旦达到上限，last_sbuf就固定不再增长，后续全靠virtqueue_get_buf()回收。

dma_addr_t bufs_dma

作用：整块 buffer 的 DMA 基地址（设备/DMA 视角的地址）。probe 里由dma_alloc_coherent()返回：

bufs_va = dma_alloc_coherent(vdev->dev.parent,
                           total_buf_space, &vrp->bufs_dma, GFP_KERNEL);

rbufs/sbufs是 CPU 用的虚拟地址，bufs_dma 是释放和 DMA 映射时用的物理/总线地址。remove 时用它释放：

dma_free_coherent(vdev->dev.parent, total_buf_space,
                vrp->rbufs, vrp->bufs_dma);

struct mutex tx_lock

作用：保护发送侧的共享状态：svq、sbufs、sleepers。允许多个发送者并发调用rpmsg_send()。用 mutex 是因为发送可能要唤醒在“打盹”的远端处理器，这个过程可能睡眠，所以必须用可睡眠的 mutex。

struct idr endpoints

作用：本地所有 endpoint 的 idr 索引表，key 是 endpoint 地址。用于快速按地址查找 endpoint。这是消息分发的核心数据结构。创建 endpoint 时分配地址并插入：

id = idr_alloc(&vrp->endpoints, ept, id_min, id_max, GFP_KERNEL);
ept->addr = id;

接收消息时按 dst 查找：

ept = idr_find(&vrp->endpoints, virtio32_to_cpu(vrp->vdev, msg->dst));

所以 rpmsg 实际是address-based dispatch，靠的就是这个 idr。

struct mutex endpoints_lock

作用：保护 endpoints idr 的并发访问。增删查改 endpoint 时都要持有：

mutex_lock(&vrp->endpoints_lock);
ept = idr_find(&vrp->endpoints, ...);
if (ept)
      kref_get(&ept->refcount);   /* 查到后先加引用，防止释放 */
mutex_unlock(&vrp->endpoints_lock);

注意它和 tx_lock 是两把不同的锁：
-tx_lock保护发送路径
-endpoints_lock保护 endpoint 表

wait_queue_head_t sendq

作用：等待 TX buffer 的发送者睡在这个等待队列上。没有 TX buffer 时，rpmsg_send() 睡眠等待：

wait_event_interruptible_timeout(vrp->sendq,
                      (msg = get_a_tx_buf(vrp)),
                      msecs_to_jiffies(15000));

远端消费完 TX buffer 后，TX complete callback 唤醒它：

static void rpmsg_xmit_done(struct virtqueue *svq)
{
      struct virtproc_info *vrp = svq->vdev->priv;
      wake_up_interruptible(&vrp->sendq);
}

atomic_t sleepers

作用：当前有多少个发送者正在等待 TX buffer（等待者计数）。它配合 TX complete 中断的动态开关：

/* 第一个等待者：打开 tx-complete callback */
if (atomic_inc_return(&vrp->sleepers) == 1)
      virtqueue_enable_cb(vrp->svq);

/* 最后一个等待者：关闭 tx-complete callback */
if (atomic_dec_and_test(&vrp->sleepers))
      virtqueue_disable_cb(vrp->svq);

设计目的：

• 没人等 TX buffer：关闭 tx-complete 中断，省开销
• 有人等 TX buffer：打开 tx-complete 中断，远端归还 buffer 后立即唤醒

（这是 5.10 的设计，Linux 7.1.1 版本的因为引入 poll 支持已去掉 sleepers。）

struct rpmsg_endpoint *ns_ept

作用：name service 专用 endpoint（固定地址 53）。它不属于任何普通 rpmsg channel，是 bus 内部用来处理远端服务“创建/销毁”通知的。probe 里在远端支持 NS feature 时创建：

if (virtio_has_feature(vdev, VIRTIO_RPMSG_F_NS)) {
      vrp->ns_ept = __rpmsg_create_ept(vrp, NULL, rpmsg_ns_cb,
                                       vrp, RPMSG_NS_ADDR);
}

它的 callback 是rpmsg_ns_cb()，负责按 NS 消息创建/销毁 rpmsg_device。remove 时单独销毁：

if (vrp->ns_ept)
      __rpmsg_destroy_ept(vrp, vrp->ns_ept);

注意创建时 rpdev 传 NULL，因为它不绑定具体 channel。

buffer 设计

驱动使用固定大小 buffer。相关定义：

#define MAX_RPMSG_NUM_BUFS	(512)
#define MAX_RPMSG_BUF_SIZE	(512)

• 最多 512 个 buffer
• 一半 RX，一半 TX
• 每个 buffer 512 字节
• 最大总内存 512 * 512 = 256 KiB

也就是说：

总 buffer 区域
+------------------------+  <-- vrp->rbufs = bufs_va = dma_alloc_coherent(vdev->dev.parent,
|                        |                    total_buf_space, &vrp->bufs_dma,GFP_KERNEL);
|   RX Buffer (前一半)   |
|                        |
+------------------------+  <-- vrp->sbufs = bufs_va + total_buf_space / 2;
|                        |
|   TX Buffer (后一半)   |
|                        |
+------------------------+  <-- vrp->rbufs + total_buf_space

在 probe 里分配:

bufs_va = dma_alloc_coherent(vdev->dev.parent,
                     total_buf_space, &vrp->bufs_dma,
                     GFP_KERNEL);

然后切分:

vrp->rbufs = bufs_va;
vrp->sbufs = bufs_va + total_buf_space / 2;

发送的每条消息都有一个公共头：

struct rpmsg_hdr {
    __virtio32 src;
    __virtio32 dst;
    __virtio32 reserved;
    __virtio16 len;
    __virtio16 flags;
    u8 data[];
} __packed;

含义：

字段	含义
src	源 endpoint 地址
dst	目标 endpoint 地址
reserved	保留
len	payload 长度
flags	消息标志
data[]	实际业务数据

Linux 收到消息后，会根据dst地址找到本地 endpoint，然后调用该 endpoint 的 callback。

get_a_tx_buf()

代码：

/* super simple buffer "allocator" that is just enough for now */
static void *get_a_tx_buf(struct virtproc_info *vrp)
{
    unsigned int len;
    void *ret;

    /* support multiple concurrent senders */
    mutex_lock(&vrp->tx_lock);

    /*
     * either pick the next unused tx buffer
     * (half of our buffers are used for sending messages)
     */
    if (vrp->last_sbuf < vrp->num_bufs / 2)
        ret = vrp->sbufs + vrp->buf_size * vrp->last_sbuf++;
    /* or recycle a used one */
    else
        ret = virtqueue_get_buf(vrp->svq, &len);

    mutex_unlock(&vrp->tx_lock);

    return ret;
}

这个函数是一个很简单的 TX buffer allocator。

它有两个阶段：

阶段 1：还有从未使用过的 TX buffer
    按数组顺序从 vrp->sbufs 里拿

阶段 2：所有 TX buffer 都至少用过一次
    从 svq used ring 回收远端已经读完的 TX buffer

last_sbuf只在第一阶段有用。

一旦：

vrp->last_sbuf == vrp->num_bufs / 2

后续就一直走：

virtqueue_get_buf(vrp->svq, &len);

也就是说：

last_sbuf = 一次性“开荒游标”
virtqueue_get_buf() = 后续复用 buffer 的来源

virtio API

API	作用
virtio_find_vqs()	找/创建 virtqueue
virtqueue_add_inbuf()	给设备一个“可写入”的 buffer
virtqueue_add_outbuf()	给设备一个“可读取”的 buffer
virtqueue_add_sgs	通用的缓冲区添加函数，包含 in 和 out
virtqueue_get_buf()	从 used ring 取回设备处理完的 buffer
virtqueue_kick()	通知对方：队列有新缓冲区
virtqueue_kick_prepare()	判断是否需要通知
virtqueue_notify()	真正发通知
virtqueue_enable_cb()	打开 virtqueue callback/中断
virtqueue_disable_cb()	关闭 virtqueue callback/中断，用于轮询期间防止中断风暴
virtqueue_get_vring_size()	获取 ring 大小
virtio_has_feature()	检查 virtio feature
virtio_device_ready()	设置 DRIVER_OK，设备可以开始工作

probe 函数

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

virtio_driver.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;
    void *bufs_va;
    int err = 0, i;
    size_t total_buf_space;
    bool notify;

    // 分配struct virtproc_info
    vrp = kzalloc(sizeof(*vrp), GFP_KERNEL);
    if (!vrp)
        return -ENOMEM;

    vrp->vdev = vdev;
    
    // 初始化struct virtproc_info的成员
    idr_init(&vrp->endpoints);
    mutex_init(&vrp->endpoints_lock);
    mutex_init(&vrp->tx_lock);
    init_waitqueue_head(&vrp->sendq);

    /* We expect two virtqueues, rx and tx (and in this order) */
    err = virtio_find_vqs(vdev, 2, vqs, vq_cbs, names, NULL);
    if (err)
        goto free_vrp;

    vrp->rvq = vqs[0];
    vrp->svq = vqs[1];

    /* we expect symmetric tx/rx vrings */
    WARN_ON(virtqueue_get_vring_size(vrp->rvq) !=
        virtqueue_get_vring_size(vrp->svq));

    /* we need less buffers if vrings are small */
    if (virtqueue_get_vring_size(vrp->rvq) < MAX_RPMSG_NUM_BUFS / 2)
        vrp->num_bufs = virtqueue_get_vring_size(vrp->rvq) * 2;
    else
        vrp->num_bufs = MAX_RPMSG_NUM_BUFS;

    vrp->buf_size = MAX_RPMSG_BUF_SIZE;

    total_buf_space = vrp->num_bufs * vrp->buf_size;

    /* allocate coherent memory for the buffers */
    bufs_va = dma_alloc_coherent(vdev->dev.parent,
                     total_buf_space, &vrp->bufs_dma,
                     GFP_KERNEL);
    if (!bufs_va) {
        err = -ENOMEM;
        goto vqs_del;
    }

    dev_dbg(&vdev->dev, "buffers: va %pK, dma %pad\n",
        bufs_va, &vrp->bufs_dma);

    /* half of the buffers is dedicated for RX */
    vrp->rbufs = bufs_va;

    /* and half is dedicated for TX */
    vrp->sbufs = bufs_va + total_buf_space / 2;

    /* set up the receive buffers */
    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 */
    }

    /* suppress "tx-complete" interrupts */
    virtqueue_disable_cb(vrp->svq);

    vdev->priv = vrp;

    /* if supported by the remote processor, enable the name service */
    if (virtio_has_feature(vdev, VIRTIO_RPMSG_F_NS)) {
        /* a dedicated endpoint handles the name service msgs */
        vrp->ns_ept = __rpmsg_create_ept(vrp, NULL, rpmsg_ns_cb,
                        vrp, RPMSG_NS_ADDR);
        if (!vrp->ns_ept) {
            dev_err(&vdev->dev, "failed to create the ns ept\n");
            err = -ENOMEM;
            goto free_coherent;
        }
    }

    /*
     * Prepare to kick but don't notify yet - we can't do this before
     * device is ready.
     */
    notify = virtqueue_kick_prepare(vrp->rvq);

    /* From this point on, we can notify and get callbacks. */
    virtio_device_ready(vdev);

    /* tell the remote processor it can start sending messages */
    /*
     * this might be concurrent with callbacks, but we are only
     * doing notify, not a full kick here, so that's ok.
     */
    if (notify)
        virtqueue_notify(vrp->rvq);

    dev_info(&vdev->dev, "rpmsg host is online\n");

    return 0;

free_coherent:
    dma_free_coherent(vdev->dev.parent, total_buf_space,
              bufs_va, vrp->bufs_dma);
vqs_del:
    vdev->config->del_vqs(vrp->vdev);
free_vrp:
    kfree(vrp);
    return err;
}

分析如下

分配virtproc_info

vrp = kzalloc(sizeof(*vrp), GFP_KERNEL);

初始化：

idr_init(&vrp->endpoints);
mutex_init(&vrp->endpoints_lock);
mutex_init(&vrp->tx_lock);
init_waitqueue_head(&vrp->sendq);

查找 virtqueue

驱动需要两个 virtqueue：

vq_callback_t *vq_cbs[] = { rpmsg_recv_done, rpmsg_xmit_done };
static const char * const names[] = { "input", "output" };

然后：

err = virtio_find_vqs(vdev, 2, vqs, vq_cbs, names, NULL);

两个 queue：

vrp->rvq = vqs[0];
vrp->svq = vqs[1];

含义：

virtqueue	用途	callback
rvq/ input	Linux 接收远端发来的消息	rpmsg_recv_done()
svq/ output	Linux 发送消息给远端	rpmsg_xmit_done()

计算 buffer 数量

if (virtqueue_get_vring_size(vrp->rvq) < MAX_RPMSG_NUM_BUFS / 2)
    vrp->num_bufs = virtqueue_get_vring_size(vrp->rvq) * 2;
else
    vrp->num_bufs = MAX_RPMSG_NUM_BUFS;

也就是说：如果 vring 小，就按 vring 能力减少 buffer 数，否则最多 512 个 buffer

分配 coherent DMA buffer

bufs_va = dma_alloc_coherent(vdev->dev.parent,
                 total_buf_space, &vrp->bufs_dma,
                 GFP_KERNEL);

这个 buffer 是主核和 virtio 后端都能访问的共享 DMA 内存。

预先把 RX buffer 放入 RX virtqueue

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

这一步很关键：

Linux 先把一批空 RX buffer 放到 available ring 中，远端处理器之后才能把消息写进这些 buffer。

默认关闭 TX complete 中断

virtqueue_disable_cb(vrp->svq);

因为大多数时候，发送方不需要每次 TX buffer 被远端消费都收到中断。只有当 Linux 没有 TX buffer、发送者要睡眠等待时，才临时打开 TX complete 中断。

创建 name service endpoint

如果远端支持：

VIRTIO_RPMSG_F_NS

则创建地址为53的 name service endpoint：

vrp->ns_ept = __rpmsg_create_ept(vrp, NULL, rpmsg_ns_cb,
                     vrp, RPMSG_NS_ADDR);

地址定义：

#define RPMSG_NS_ADDR			(53)

这个 endpoint 专门处理远端服务创建/销毁通知。

设备 ready 并通知远端

notify = virtqueue_kick_prepare(vrp->rvq);
virtio_device_ready(vdev);
if (notify)
    virtqueue_notify(vrp->rvq);

这里的顺序很重要：

1. RX buffer 已经准备好(virtqueue_kick_prepare)
2. virtio 设备置 ready
3. 通知远端可以开始发送消息

最后打印：

dev_info(&vdev->dev, "rpmsg host is online\n");

remove 函数

static int rpmsg_remove_device(struct device *dev, void *data)
{
    device_unregister(dev);

    return 0;
}

static void rpmsg_remove(struct virtio_device *vdev)
{
    struct virtproc_info *vrp = vdev->priv;
    size_t total_buf_space = vrp->num_bufs * vrp->buf_size;
    int ret;

    vdev->config->reset(vdev);

    ret = device_for_each_child(&vdev->dev, NULL, rpmsg_remove_device);
    if (ret)
        dev_warn(&vdev->dev, "can't remove rpmsg device: %d\n", ret);

    if (vrp->ns_ept)
        __rpmsg_destroy_ept(vrp, vrp->ns_ept);

    idr_destroy(&vrp->endpoints);

    vdev->config->del_vqs(vrp->vdev);

    dma_free_coherent(vdev->dev.parent, total_buf_space,
              vrp->rbufs, vrp->bufs_dma);

    kfree(vrp);
}

流程：

1. reset virtio device

vdev->config->reset(vdev);

先停止设备，避免继续收发。

2. 删除所有子 rpmsg device

device_for_each_child(&vdev->dev, NULL, rpmsg_remove_device);

在rpmsg_remove_device()里调用：

device_unregister(dev);

3. 销毁 NS endpoint

if (vrp->ns_ept)
    __rpmsg_destroy_ept(vrp, vrp->ns_ept);

4. endpoints、virtqueue、DMA buffer、vrp

idr_destroy(&vrp->endpoints);
vdev->config->del_vqs(vrp->vdev);
dma_free_coherent(...);
kfree(vrp);

发送消息(rpmsg_send_offchannel_raw())

所有发送 API 最终都会进入rpmsg_send_offchannel_raw()

/**
 * rpmsg_send_offchannel_raw() - send a message across to the remote processor
 * @rpdev: the rpmsg channel
 * @src: source address
 * @dst: destination address
 * @data: payload of message
 * @len: length of payload
 * @wait: indicates whether caller should block in case no TX buffers available
 *
 * This function is the base implementation for all of the rpmsg sending API.
 *
 * It will send @data of length @len to @dst, and say it's from @src. The
 * message will be sent to the remote processor which the @rpdev channel
 * belongs to.
 *
 * The message is sent using one of the TX buffers that are available for
 * communication with this remote processor.
 *
 * If @wait is true, the caller will be blocked until either a TX buffer is
 * available, or 15 seconds elapses (we don't want callers to
 * sleep indefinitely due to misbehaving remote processors), and in that
 * case -ERESTARTSYS is returned. The number '15' itself was picked
 * arbitrarily; there's little point in asking drivers to provide a timeout
 * value themselves.
 *
 * Otherwise, if @wait is false, and there are no TX buffers available,
 * the function will immediately fail, and -ENOMEM will be returned.
 *
 * Normally drivers shouldn't use this function directly; instead, drivers
 * should use the appropriate rpmsg_{try}send{to, _offchannel} API
 * (see include/linux/rpmsg.h).
 *
 * Returns 0 on success and an appropriate error value on failure.
 */
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;

    /* bcasting isn't allowed */
    if (src == RPMSG_ADDR_ANY || dst == RPMSG_ADDR_ANY) {
        dev_err(dev, "invalid addr (src 0x%x, dst 0x%x)\n", src, dst);
        return -EINVAL;
    }

    /*
     * We currently use fixed-sized buffers, and therefore the payload
     * length is limited.
     *
     * One of the possible improvements here is either to support
     * user-provided buffers (and then we can also support zero-copy
     * messaging), or to improve the buffer allocator, to support
     * variable-length buffer sizes.
     */
    if (len > vrp->buf_size - sizeof(struct rpmsg_hdr)) {
        dev_err(dev, "message is too big (%d)\n", len);
        return -EMSGSIZE;
    }

    /* grab a buffer */
    msg = get_a_tx_buf(vrp);
    if (!msg && !wait)
        return -ENOMEM;

    /* no free buffer ? wait for one (but bail after 15 seconds) */
    while (!msg) {
        /* enable "tx-complete" interrupts, if not already enabled */
        rpmsg_upref_sleepers(vrp);

        /*
         * sleep until a free buffer is available or 15 secs elapse.
         * the timeout period is not configurable because there's
         * little point in asking drivers to specify that.
         * if later this happens to be required, it'd be easy to add.
         */
        err = wait_event_interruptible_timeout(vrp->sendq,
                    (msg = get_a_tx_buf(vrp)),
                    msecs_to_jiffies(15000));

        /* disable "tx-complete" interrupts if we're the last sleeper */
        rpmsg_downref_sleepers(vrp);

        /* timeout ? */
        if (!err) {
            dev_err(dev, "timeout waiting for a tx buffer\n");
            return -ERESTARTSYS;
        }
    }

    msg->len = cpu_to_virtio16(vrp->vdev, len);
    msg->flags = 0;
    msg->src = cpu_to_virtio32(vrp->vdev, src);
    msg->dst = cpu_to_virtio32(vrp->vdev, dst);
    msg->reserved = 0;
    memcpy(msg->data, data, len);

    dev_dbg(dev, "TX From 0x%x, To 0x%x, Len %d, Flags %d, Reserved %d\n",
        src, dst, len, msg->flags, msg->reserved);
#if defined(CONFIG_DYNAMIC_DEBUG)
    dynamic_hex_dump("rpmsg_virtio TX: ", DUMP_PREFIX_NONE, 16, 1,
             msg, sizeof(*msg) + len, true);
#endif

    rpmsg_sg_init(&sg, msg, sizeof(*msg) + len);

    mutex_lock(&vrp->tx_lock);

    /* add message to the remote processor's virtqueue */
    err = virtqueue_add_outbuf(vrp->svq, &sg, 1, msg, GFP_KERNEL);
    if (err) {
        /*
         * need to reclaim the buffer here, otherwise it's lost
         * (memory won't leak, but rpmsg won't use it again for TX).
         * this will wait for a buffer management overhaul.
         */
        dev_err(dev, "virtqueue_add_outbuf failed: %d\n", err);
        goto out;
    }

    /* tell the remote processor it has a pending message to read */
    virtqueue_kick(vrp->svq);
out:
    mutex_unlock(&vrp->tx_lock);
    return err;
}

调用链大致是：

rpmsg_send()
  → ept->ops->send()
    → virtio_rpmsg_send()
      → rpmsg_send_offchannel_raw()

rpmsg_trysend()
  → virtio_rpmsg_trysend()
    → rpmsg_send_offchannel_raw(..., wait = false)

参数检查

首先检查 src/dst：

if (src == RPMSG_ADDR_ANY || dst == RPMSG_ADDR_ANY)
    return -EINVAL;

不允许广播地址作为实际发送地址。
然后检查长度：

if (len > vrp->buf_size - sizeof(struct rpmsg_hdr))
    return -EMSGSIZE;

因为单个 buffer 固定 512 字节，所以 payload 不能超过512 - sizeof(struct rpmsg_hdr)

获取 TX buffer

msg = get_a_tx_buf(vrp);

它的逻辑：

if (vrp->last_sbuf < vrp->num_bufs / 2)
    ret = vrp->sbufs + vrp->buf_size * vrp->last_sbuf++;
else
    ret = virtqueue_get_buf(vrp->svq, &len);

含义：

1. 初始阶段：直接从 TX buffer 池中拿一个从没用过的 buffer
2. 后续阶段：从svqused ring 中回收远端已经消费完的 TX buffer

没有 TX buffer 时怎么办？

如果是trysend：

if (!msg && !wait)
    return -ENOMEM;

如果是普通send，则等待，最多 15 秒：

err = wait_event_interruptible_timeout(vrp->sendq,
            (msg = get_a_tx_buf(vrp)),
            msecs_to_jiffies(15000));

如果超时：

return -ERESTARTSYS;

等待前会调用：

rpmsg_upref_sleepers(vrp);

作用： 如果这是第一个睡眠等待 TX buffer 的发送者,则打开 TX complete 中断

virtqueue_enable_cb(vrp->svq);

等待结束后调用：

rpmsg_downref_sleepers(vrp);

如果已经没有等待者，就关闭 TX complete 中断：

virtqueue_disable_cb(vrp->svq);

填充 rpmsg header 和 payload

msg->len = cpu_to_virtio16(vrp->vdev, len);
msg->flags = 0;
msg->src = cpu_to_virtio32(vrp->vdev, src);
msg->dst = cpu_to_virtio32(vrp->vdev, dst);
msg->reserved = 0;
memcpy(msg->data, data, len);

注意这里使用cpu_to_virtio16/32()，因为 virtio 设备可能有特定字节序要求。

加入 TX virtqueue 并 kick 远端

virtqueue_add_outbuf(vrp->svq, &sg, 1, msg, GFP_KERNEL);

然后：

virtqueue_kick(vrp->svq);

完整含义：
Linux 填好 TX buffer → 把 buffer 加入 output virtqueue → kick 远端处理器 → 远端从 virtqueue 取走消息

TX complete 回调

在 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];
    
    .....
    
    err = virtio_find_vqs(vdev, 2, vqs, vq_cbs, names, NULL);
    
    .....
}

即 TX 的回调函数为rpmsg_xmit_done

注意：平时 TX complete 中断是关闭的，只有有发送者睡眠等待 buffer 时才打开。

/*
 * This is invoked whenever the remote processor completed processing
 * a TX msg we just sent it, and the buffer is put back to the used ring.
 *
 * Normally, though, we suppress this "tx complete" interrupt in order to
 * avoid the incurred overhead.
 */
static void rpmsg_xmit_done(struct virtqueue *svq)
{
    struct virtproc_info *vrp = svq->vdev->priv;

    dev_dbg(&svq->vdev->dev, "%s\n", __func__);

    /* wake up potential senders that are waiting for a tx buffer */
    wake_up_interruptible(&vrp->sendq);
}

当远端消费了 TX buffer，virtio 后端把 buffer 放入 used ring，并触发 TX complete callback。Linux 唤醒等待 TX buffer 的发送线程。

TX sleepers 机制：rpmsg_upref_sleepers()

static void rpmsg_upref_sleepers(struct virtproc_info *vrp)
{
    /* support multiple concurrent senders */
    mutex_lock(&vrp->tx_lock);

    /* are we the first sleeping context waiting for tx buffers ? */
    if (atomic_inc_return(&vrp->sleepers) == 1)
        /* enable "tx-complete" interrupts before dozing off */
        virtqueue_enable_cb(vrp->svq);

    mutex_unlock(&vrp->tx_lock);
}

含义：发送线程因为没有 TX buffer 准备睡眠前：sleepers++, 如果这是第一个等待者，就打开 TX complete callback
对应还有rpmsg_downref_sleepers()：发送线程醒来后：sleepers--, 如果这是最后一个等待者，就关闭 TX complete callback

这套机制的目的：

• 没人等 TX buffer：关闭 TX complete 中断，减少开销
• 有人等 TX buffer：打开 TX complete 中断，远端归还 buffer 后马上唤醒等待者

完整链路：

TX buffer 用完
    ↓
rpmsg_send() 准备睡眠
    ↓
rpmsg_upref_sleepers()
    ↓
第一个 sleeper 打开 svq callback
    ↓
远端读完 TX buffer
    ↓
rpmsg_xmit_done()
    ↓
wake_up_interruptible(&vrp->sendq)
    ↓
发送线程醒来
    ↓
get_a_tx_buf() / virtqueue_get_buf(svq)
    ↓
拿到回收 buffer

接收消息(rpmsg_recv_done)

在 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];
    
    .....
    
    err = virtio_find_vqs(vdev, 2, vqs, vq_cbs, names, NULL);
    
    .....
}

即 RX 的回调函数为rpmsg_recv_done，函数定义如下：

rpmsg_recv_done

/* called when an rx buffer is used, and it's time to digest a message */
static void rpmsg_recv_done(struct virtqueue *rvq)
{
    struct virtproc_info *vrp = rvq->vdev->priv;
    struct device *dev = &rvq->vdev->dev;
    struct rpmsg_hdr *msg;
    unsigned int len, msgs_received = 0;
    int err;

    msg = virtqueue_get_buf(rvq, &len);
    if (!msg) {
        dev_err(dev, "uhm, incoming signal, but no used buffer ?\n");
        return;
    }

    while (msg) {
        err = rpmsg_recv_single(vrp, dev, msg, len);
        if (err)
            break;

        msgs_received++;

        msg = virtqueue_get_buf(rvq, &len);
    }

    dev_dbg(dev, "Received %u messages\n", msgs_received);

    /* tell the remote processor we added another available rx buffer */
    if (msgs_received)
        virtqueue_kick(vrp->rvq);
}

从 RX virtqueue 取 used buffer

msg = virtqueue_get_buf(rvq, &len);

如果能拿到msg, 尝试继续，循环处理所有可用消息：

while (msg) {
    err = rpmsg_recv_single(vrp, dev, msg, len);
    if (err)
        break;

    msgs_received++;

    msg = virtqueue_get_buf(rvq, &len);
}

单条消息处理：rpmsg_recv_single()

static int rpmsg_recv_single(struct virtproc_info *vrp, struct device *dev,
                 struct rpmsg_hdr *msg, unsigned int len)
{
    struct rpmsg_endpoint *ept;
    struct scatterlist sg;
    unsigned int msg_len = virtio16_to_cpu(vrp->vdev, msg->len);
    int err;

    dev_dbg(dev, "From: 0x%x, To: 0x%x, Len: %d, Flags: %d, Reserved: %d\n",
        virtio32_to_cpu(vrp->vdev, msg->src),
        virtio32_to_cpu(vrp->vdev, msg->dst), msg_len,
        virtio16_to_cpu(vrp->vdev, msg->flags),
        virtio32_to_cpu(vrp->vdev, msg->reserved));
#if defined(CONFIG_DYNAMIC_DEBUG)
    dynamic_hex_dump("rpmsg_virtio RX: ", DUMP_PREFIX_NONE, 16, 1,
             msg, sizeof(*msg) + msg_len, true);
#endif

    /*
     * We currently use fixed-sized buffers, so trivially sanitize
     * the reported payload length.
     */
    if (len > vrp->buf_size ||
        msg_len > (len - sizeof(struct rpmsg_hdr))) {
        dev_warn(dev, "inbound msg too big: (%d, %d)\n", len, msg_len);
        return -EINVAL;
    }

    /* use the dst addr to fetch the callback of the appropriate user */
    mutex_lock(&vrp->endpoints_lock);

    ept = idr_find(&vrp->endpoints, virtio32_to_cpu(vrp->vdev, msg->dst));

    /* let's make sure no one deallocates ept while we use it */
    if (ept)
        kref_get(&ept->refcount);

    mutex_unlock(&vrp->endpoints_lock);

    if (ept) {
        /* make sure ept->cb doesn't go away while we use it */
        mutex_lock(&ept->cb_lock);

        if (ept->cb)
            ept->cb(ept->rpdev, msg->data, msg_len, ept->priv,
                virtio32_to_cpu(vrp->vdev, msg->src));

        mutex_unlock(&ept->cb_lock);

        /* farewell, ept, we don't need you anymore */
        kref_put(&ept->refcount, __ept_release);
    } else
        dev_warn(dev, "msg received with no recipient\n");

    /* publish the real size of the buffer */
    rpmsg_sg_init(&sg, msg, vrp->buf_size);

    /* add the buffer back to the remote processor's virtqueue */
    err = virtqueue_add_inbuf(vrp->rvq, &sg, 1, msg, GFP_KERNEL);
    if (err < 0) {
        dev_err(dev, "failed to add a virtqueue buffer: %d\n", err);
        return err;
    }

    return 0;
}

接收时不是根据 channel name 分发，而是根据msg->dst，在vrp->endpoints里查本地 endpoint。所以 rpmsg 实际运行时更像 address-based messaging，channel name 主要用于设备发现和 driver 匹配。主要步骤：

1. 解析 payload 长度

msg_len = virtio16_to_cpu(vrp->vdev, msg->len);

2. 检查消息长度是否合法

if (len > vrp->buf_size ||
    msg_len > (len - sizeof(struct rpmsg_hdr))) {
    return -EINVAL;
}

防止远端传入异常长度导致越界。

3. 根据dst查找 endpoint

ept = idr_find(&vrp->endpoints, virtio32_to_cpu(vrp->vdev, msg->dst));

这是 rpmsg 分发的核心：msg->dst == 本地 endpoint 地址

4. 增加 endpoint 引用计数

if (ept)
    kref_get(&ept->refcount);

避免 callback 执行期间 endpoint 被释放。

5. 调用 callback

if (ept->cb)
    ept->cb(ept->rpdev, msg->data, msg_len, ept->priv,
        virtio32_to_cpu(vrp->vdev, msg->src));

传给上层 callback 的参数包括：

参数	含义
ept->rpdev	对应 rpmsg device
msg->data	payload
msg_len	payload 长度
ept->priv	endpoint 私有数据
msg->src	远端源地址

6. 把 RX buffer 放回 virtqueue
   处理完之后，这个 buffer 需要重新给远端用：

rpmsg_sg_init(&sg, msg, vrp->buf_size);
err = virtqueue_add_inbuf(vrp->rvq, &sg, 1, msg, GFP_KERNEL);

kick 远端

最后，在rpmsg_recv_done()里如果处理过消息，就 kick 远端：

if (msgs_received)
    virtqueue_kick(vrp->rvq);

name service 机制

注意：name service 是可选 feature。feature 定义：

#define VIRTIO_RPMSG_F_NS	0

如果远端不支持这个 feature，动态服务发现就不可用。此时 channel 可能需要通过静态方式创建。

rpmsg 支持远端动态通知 Linux：我创建了一个服务或者我销毁了一个服务，这叫 name service。相关结构：

struct rpmsg_ns_msg {
    char name[RPMSG_NAME_SIZE];
    __virtio32 addr;
    __virtio32 flags;
} __packed;

字段：

字段	含义
name	服务名称
addr	远端服务地址
flags	创建或销毁

flags：

RPMSG_NS_CREATE  = 0
RPMSG_NS_DESTROY = 1

NS endpoint

name service 使用固定地址：

#define RPMSG_NS_ADDR			(53)

probe 时创建：

vrp->ns_ept = __rpmsg_create_ept(vrp, NULL, rpmsg_ns_cb,
                     vrp, RPMSG_NS_ADDR);

注意这里rpdev参数是NULL，因为 NS endpoint 不属于某个普通 rpmsg channel，而是 bus 自己用的内部 endpoint。

NS callback：rpmsg_ns_cb()

/* invoked when a name service announcement arrives */
static int rpmsg_ns_cb(struct rpmsg_device *rpdev, void *data, int len,
               void *priv, u32 src)
{
    struct rpmsg_ns_msg *msg = data;
    struct rpmsg_device *newch;
    struct rpmsg_channel_info chinfo;
    struct virtproc_info *vrp = priv;
    struct device *dev = &vrp->vdev->dev;
    int ret;

#if defined(CONFIG_DYNAMIC_DEBUG)
    dynamic_hex_dump("NS announcement: ", DUMP_PREFIX_NONE, 16, 1,
             data, len, true);
#endif

    if (len != sizeof(*msg)) {
        dev_err(dev, "malformed ns msg (%d)\n", len);
        return -EINVAL;
    }

    /*
     * the name service ept does _not_ belong to a real rpmsg channel,
     * and is handled by the rpmsg bus itself.
     * for sanity reasons, make sure a valid rpdev has _not_ sneaked
     * in somehow.
     */
    if (rpdev) {
        dev_err(dev, "anomaly: ns ept has an rpdev handle\n");
        return -EINVAL;
    }

    /* don't trust the remote processor for null terminating the name */
    msg->name[RPMSG_NAME_SIZE - 1] = '\0';

    strncpy(chinfo.name, msg->name, sizeof(chinfo.name));
    chinfo.src = RPMSG_ADDR_ANY;
    chinfo.dst = virtio32_to_cpu(vrp->vdev, msg->addr);

    dev_info(dev, "%sing channel %s addr 0x%x\n",
         virtio32_to_cpu(vrp->vdev, msg->flags) & RPMSG_NS_DESTROY ?
         "destroy" : "creat", msg->name, chinfo.dst);

    if (virtio32_to_cpu(vrp->vdev, msg->flags) & RPMSG_NS_DESTROY) {
        ret = rpmsg_unregister_device(&vrp->vdev->dev, &chinfo);
        if (ret)
            dev_err(dev, "rpmsg_destroy_channel failed: %d\n", ret);
    } else {
        newch = rpmsg_create_channel(vrp, &chinfo);
        if (!newch)
            dev_err(dev, "rpmsg_create_channel failed\n");
    }

    return 0;
}

收到远端 NS 消息后：

1. 检查长度
2. 确保 endpoint 不绑定真实rpdev
3. 修正 name 字符串结尾
4. 构造rpmsg_channel_info
5. 根据 flags 创建或销毁 channel

创建 channel：

newch = rpmsg_create_channel(vrp, &chinfo);

销毁 channel：

ret = rpmsg_unregister_device(&vrp->vdev->dev, &chinfo);

本地服务 announce 机制

这个驱动不仅能接收远端 NS，也能向远端 announce 本地服务。

相关函数：

virtio_rpmsg_announce_create()
virtio_rpmsg_announce_destroy()

当本地创建一个需要 announce 的 rpmsg channel 时，驱动会构造rpmsg_ns_msg，发到远端的 NS 地址53(在rpmsg_core.c中)。 创建通知：

nsm.flags = cpu_to_virtio32(vrp->vdev, RPMSG_NS_CREATE);
err = rpmsg_sendto(rpdev->ept, &nsm, sizeof(nsm), RPMSG_NS_ADDR);

销毁通知：

nsm.flags = cpu_to_virtio32(vrp->vdev, RPMSG_NS_DESTROY);
err = rpmsg_sendto(rpdev->ept, &nsm, sizeof(nsm), RPMSG_NS_ADDR);

rpmsg channel 创建

/*
 * create an rpmsg channel using its name and address info.
 * this function will be used to create both static and dynamic
 * channels.
 */
static struct rpmsg_device *rpmsg_create_channel(struct virtproc_info *vrp,
                         struct rpmsg_channel_info *chinfo)
{
    struct virtio_rpmsg_channel *vch;
    struct rpmsg_device *rpdev;
    struct device *tmp, *dev = &vrp->vdev->dev;
    int ret;

    /* make sure a similar channel doesn't already exist */
    tmp = rpmsg_find_device(dev, chinfo);
    if (tmp) {
        /* decrement the matched device's refcount back */
        put_device(tmp);
        dev_err(dev, "channel %s:%x:%x already exist\n",
                chinfo->name, chinfo->src, chinfo->dst);
        return NULL;
    }

    vch = kzalloc(sizeof(*vch), GFP_KERNEL);
    if (!vch)
        return NULL;

    /* Link the channel to our vrp */
    vch->vrp = vrp;

    /* Assign public information to the rpmsg_device */
    rpdev = &vch->rpdev;
    rpdev->src = chinfo->src;
    rpdev->dst = chinfo->dst;
    rpdev->ops = &virtio_rpmsg_ops;

    /*
     * rpmsg server channels has predefined local address (for now),
     * and their existence needs to be announced remotely
     */
    rpdev->announce = rpdev->src != RPMSG_ADDR_ANY;

    strncpy(rpdev->id.name, chinfo->name, RPMSG_NAME_SIZE);

    rpdev->dev.parent = &vrp->vdev->dev;
    rpdev->dev.release = virtio_rpmsg_release_device;
    ret = rpmsg_register_device(rpdev);
    if (ret)
        return NULL;

    return rpdev;
}

它根据传入的struct rpmsg_channel_info *chinfo创建一个rpmsg_device。

/**
 * struct rpmsg_channel_info - channel info representation
 * @name: name of service
 * @src: local address
 * @dst: destination address
 */
struct rpmsg_channel_info {
    char name[RPMSG_NAME_SIZE];
    u32 src;
    u32 dst;
};

流程：

1. 检查是否已有相同 channel

tmp = rpmsg_find_device(dev, chinfo);

如果已经存在，则不重复创建。

2. 分配virtio_rpmsg_channel

vch = kzalloc(sizeof(*vch), GFP_KERNEL);

这个结构：

struct virtio_rpmsg_channel {
    struct rpmsg_device rpdev;
    struct virtproc_info *vrp;
};

它是 virtio transport 私有 channel，对外暴露其中的rpmsg_device。

3. 填充rpmsg_device

rpdev->src = chinfo->src;
rpdev->dst = chinfo->dst;
rpdev->ops = &virtio_rpmsg_ops;

strncpy(rpdev->id.name, chinfo->name, RPMSG_NAME_SIZE);

然后注册：

ret = rpmsg_register_device(rpdev);

注册后，rpmsg bus 会匹配上层 rpmsg driver 的 id table，然后调用对应 driver 的 probe。

整体收发流程图

Linux 发送消息给远端

上层 rpmsg driver
    |
    | rpmsg_send()
    v
rpmsg core
    |
    v
virtio_rpmsg_send()
    |
    v
rpmsg_send_offchannel_raw()
    |
    | 1. 检查 src/dst/len
    | 2. 获取 TX buffer
    | 3. 填 rpmsg_hdr
    | 4. memcpy payload
    | 5. virtqueue_add_outbuf()
    | 6. virtqueue_kick()
    v
远端处理器从 svq 取消息

Linux 接收远端消息

远端处理器写入 RX buffer
    |
    | 通知 virtqueue
    v
rpmsg_recv_done()
    |
    v
virtqueue_get_buf()
    |
    v
rpmsg_recv_single()
    |
    | 1. 检查长度
    | 2. 用 msg->dst 查 endpoint
    | 3. 调 endpoint callback
    | 4. 把 RX buffer 重新 add_inbuf()
    v
上层 rpmsg driver 收到 callback

远端发布服务

远端发送 NS 消息到 addr 53
    |
    v
rpmsg_recv_done()
    |
    v
rpmsg_recv_single()
    |
    v
endpoint 53 的 callback
    |
    v
rpmsg_ns_cb()
    |
    | RPMSG_NS_CREATE
    v
rpmsg_create_channel()
    |
    v
rpmsg_register_device()
    |
    v
匹配上层 rpmsg driver

RX buffer 生命周期

RX buffer 生命周期：

probe 时 add_inbuf
    ↓
远端写入消息
    ↓
Linux virtqueue_get_buf
    ↓
调用 callback
    ↓
Linux 重新 add_inbuf
    ↓
远端再次使用

所以 RX buffer 是循环复用的。

TX buffer 生命周期

TX buffer 生命周期：

Linux 从 sbufs 初始池拿 buffer
    ↓
填消息
    ↓
virtqueue_add_outbuf
    ↓
远端消费
    ↓
buffer 进入 used ring
    ↓
Linux virtqueue_get_buf 回收
    ↓
再次发送

参考文档

[1]

AMP相关：3 Linux rpmsg子系统(STM32MP157方案)

[2]

多核异构核间通信–ipcc

[3]

Remote Processor Messaging (RPMsg)