Virtio Rpmsg Bus

---

Timeline

Timeline

2026-06-22

init

This article introduces the virtio-based rpmsg bus driver in the Linux kernel (virtio_rpmsg_bus), elaborates on its implementation as the virtio transport of the rpmsg bus, responsible for sending and receiving messages between the Linux main core and remote processors through vring, and abstracting remote services as rpmsg_device for upper-layer drivers to bind. The article analyzes in detail the three-layer structure of rpmsg, the communication model, and core data structures such as rpmsg_device、rpmsg_endpoint and virtproc_info, and explains key mechanisms such as endpoint creation and destruction, buffer management, send synchronization, and name service. Finally, it points out that rpmsg is actually an address-based message distribution mechanism.

---

linux 5.10.238

rpmsg.drawio

rpmsg structure

rpmsg has three layers:

┌────────────────────────────────────────┐
│ 业务层：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.cIn the middle and lower layers, its responsibilities are:

1. Take over the virtio rpmsg device
2. Use vring to send and receive messages
3. Create/destroy rpmsg_device based on name service

It producesrpmsg_device, but does not consume. The consumer isrpmsg_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.cis in the Linux kernel the virtio-based rpmsg bus driver. Its role is:

Enable the sending and receiving of rpmsg messages between the Linux main core and remote processors through the virtio vring mechanism, and abstract remote services as Linux’srpmsg_device, for upper-layer rpmsg drivers to bind and use.
This file itself is the virtio transport implementation of the rpmsg bus, it does not care about specific business protocols, such as audio, sensors, TEE, MCU control, etc.; it is only responsible for delivering messages to the corresponding endpoint.

Basic communication model

rpmsg

The communication unit of rpmsg is:

• rpmsg_devicerpmsg channel device, a communication channel, is a ‘device’
• rpmsg_driverthe driver that handles the business of this channel
• rpmsg_endpointThe actual sending/receiving endpoint on the channel (address + callback)

Relationship among the three:

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

rpmsg_device is the middle “device managed by the driver”. When creatingrpmsg_deviceit triggers the probe function for matching, and after successful matching, callsrpmsg_driver->probefunction

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 

wherestruct rpmsg_devicedefined as follows:

/**
 * 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_opsrepresents anrpmsg_deviceoperation of

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

A rpmsg_device is equivalent to a logical channel, for example:

• channel name: “rpmsg-demo”
• src: local endpoint address
• dst: remote endpoint address

Beforevirtio_rpmsg_bus.cHere, the channel is created as follows (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);

That is, the remote side announces a service “rpmsg-demo”, and Linux creates arpmsg_devicerepresenting this channel; or the local side actively creates a service “rpmsg-demo”, i.e., creates arpmsg_devicerepresenting the local channel. After the announce, the NS service notifies the remote side to also create the “rpmsg-demo” channel. The local and remoterpmsg_devicesrc and dst are opposite to each other, and they communicate through this channel.

Actuallyvirtio_rpmsg_bus.cThe abstract rpmsg channel descriptor in it is

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

that is, arpmsg_devicealso add the core data structurestruct virtproc_info *vrp, which is the key data structure for virtio to implement the rpmsg bus

struct rpmsg_endpoint

rpmsg_devicejust like a telephone line + an extension service, andrpmsg_endpointrepresents the person who actually answers the phone

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

The rpmsg_endpoint-related logic is mainly inrpmsg_core.camong which the most important isrpmsg_rx_cb_t cbandconst struct rpmsg_endpoint_ops *opsthese two members:

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

This function represents the callback function triggered when the ept receives a message

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

represents the operation function of rpmsg_endpoint.

Endpoint creation

The core function is:__rpmsg_create_ept(), it is the operation function of rpmsg_device

/* 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. Allocationstruct rpmsg_endpoint
2. Initialize reference count and callback lock
3. Record callback, private data, ops
4. Allocate a local address for the endpoint
5. Insertvrp->endpointsThis idr

Key logic:

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

• If the caller does not specify an address, allocate dynamically
• Dynamic addresses from1024start
• 0 ~ 1023Reserved for predefined services
  And the reserved address definitions:

#define RPMSG_RESERVED_ADDRESSES	(1024)

Endpoint addresses are allocated viaidr_alloc()allocation:

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

Endpoint destruction

Core function:__rpmsg_destroy_ept(), it is the operation function of rpmsg_endpoint

1. FromidrDelete endpoint from
2. Set callback toNULL
3. Decrement the reference count, release the endpoint if necessary

Key code:

/**
 * __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);
}

Pay special attention to concurrency here:

• Delete the idr: prevent new RX messages from finding this endpoint
• Setcb = NULL: Prevent in-flight RX calls from invoking the callback after the endpoint has been obtained
• kref: Prevent the endpoint from being freed while the RX path is using it

struct virtproc_info

Core data structures: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;
};

This is the private state of the entire virtio rpmsg device, attached tovdev->priv = vrpin, divided by function

• virtio basics:
  • vdevUnderlying virtio device
• TX/RX channels:
  • rvq,svqRX/TX virtqueue
• buffer management:
  • rbufs,sbufsRX/TX buffer virtual addresses
  • num_bufsTotal number of buffers
  • buf_sizeSingle buffer size
  • last_sbufTX allocation cursor
  • bufs_dmaBuffer DMA base address
• TX synchronization:
  • tx_lockProtects svq/sbufs/sleepers
  • sendqWait queue for waiting on TX buffers
  • sleepersWaiter count (controls the tx-complete interrupt enable/disable)
• endpoint management:
  • endpointsEndpoint idr (lookup by address)
  • endpoints_lockprotects the endpoint table
• name service：
  • ns_eptName service endpoint (addr 53)

struct virtio_device *vdev

Purpose: points to the underlying virtio device. It is the root of this rpmsg instance. Through it you can get:
- vdev->dev device node (dev_err/dev_for debugging)
- vdev->config virtio configuration operations (reset/del_vqs, etc.)
- vdev->priv points back to vrp itself
Set in probe:

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

In callback, it is used to retrieve vrp:

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

struct virtqueue *rvq, *svq

Purpose:

• rvq = receive virtqueue: remote -> Linux
• svq = send virtqueue: Linux -> remote
  These are the two channels for receiving and sending. In probe:

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

• Receiving messages:virtqueue_get_buf(rvq)/virtqueue_add_inbuf(rvq)
• Sending messages:virtqueue_add_outbuf(svq)/virtqueue_get_buf(svq)

void *rbufs, *sbufs

Purpose:

• rbufs: start of kernel virtual address for RX buffer area

• sbufs: start of kernel virtual address for TX buffer area

  In probe, a whole block of DMA memory is split into two halves:

vrp->rbufs = bufs_va;                     /* First half: RX */
vrp->sbufs = bufs_va + total_buf_space / 2; /* Second half: TX */

As shown in the figure

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

get_a_tx_buf()That is, get the TX buffer from sbufs by index:

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

unsigned int num_bufs

Purpose: total number of RX + TX buffers (each takes half). In probe, calculated based on vring size:

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 */

So:

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

Used in many placesnum_bufs / 2As a boundary, for exampleget_a_tx_buf()：

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

unsigned int buf_size

Purpose: number of bytes per buffer, currently fixed at 512 (MAX_RPMSG_BUF_SIZE）。

In the probe function

vrp->buf_size = MAX_RPMSG_BUF_SIZE;

It determines:

• Address stride of each buffer:sbufs + buf_size * i
• Maximum payload per message

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

int last_sbuf

Purpose: The ‘pioneer cursor’ of the TX buffer, recording how many have been initially allocated in order. Inget_a_tx_buf()used in:

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: There are still unused new TX buffers, take them in order
• Otherwise: all TX buffers have been used, so recycle and reuse from the used ring
  Once the limit is reached,last_sbufit stops growing, and thereafter relies entirely onvirtqueue_get_buf()recycling.

dma_addr_t bufs_dma

Purpose: The DMA base address of the entire buffer (address from the device/DMA perspective). In probe, it isdma_alloc_coherent()returned by:

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

rbufs/sbufsis the virtual address used by the CPU, and bufs_dma is the physical/bus address used for release and DMA mapping. In remove, use it to release:

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

struct mutex tx_lock

Purpose: Protects the shared state on the sending side: svq, sbufs, sleepers. Allows multiple senders to call concurrentlyrpmsg_send(). A mutex is used because sending may need to wake up a remote processor that is ‘napping’, and this process may sleep, so a sleepable mutex must be used.

struct idr endpoints

Purpose: The idr index table of all local endpoints, keyed by endpoint address. Used to quickly find an endpoint by address. This is the core data structure for message distribution. When creating an endpoint, an address is allocated and inserted:

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

When receiving a message, look up by dst:

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

So rpmsg is actuallyaddress-based dispatch, and it relies on this idr.

struct mutex endpoints_lock

Purpose: Protects concurrent access to the endpoints idr. It must be held when adding, deleting, querying, or modifying endpoints:

mutex_lock(&vrp->endpoints_lock);
ept = idr_find(&vrp->endpoints, ...);
if (ept)
      kref_get(&ept->refcount);   /* After finding it, first add a reference to prevent it from being released. */
mutex_unlock(&vrp->endpoints_lock);

Note that it and tx_lock are two different locks:
-tx_lockprotects the sending path
-endpoints_lockprotects the endpoint table

wait_queue_head_t sendq

Purpose: Senders waiting for a TX buffer sleep on this wait queue. When there is no TX buffer, rpmsg_send() sleeps and waits:

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

After the remote side consumes the TX buffer, the TX complete callback wakes it up:

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

atomic_t sleepers

Purpose: How many senders are currently waiting for a TX buffer (waiter count). It works with the dynamic on/off of the TX complete interrupt:

/* First waiter: enable tx-complete callback */
if (atomic_inc_return(&vrp->sleepers) == 1)
      virtqueue_enable_cb(vrp->svq);

/* Last waiter: disable tx-complete callback */
if (atomic_dec_and_test(&vrp->sleepers))
      virtqueue_disable_cb(vrp->svq);

Design purpose:

• No one waiting for TX buffer: disable tx-complete interrupt to save overhead
• Someone waiting for TX buffer: enable tx-complete interrupt, wake up immediately after remote returns buffer

(This is the 5.10 design; the Linux 7.1.1 version has removed sleepers due to the introduction of poll support.)

struct rpmsg_endpoint *ns_ept

Purpose: dedicated endpoint for name service (fixed address 53). It does not belong to any ordinary rpmsg channel; it is used inside the bus to handle remote service ‘create/destroy’ notifications. It is created in probe when the remote supports the 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);
}

Its callback isrpmsg_ns_cb()responsible for creating/destroying rpmsg_device according to NS messages. Destroyed separately on remove:

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

Note: rpdev is passed NULL when creating, because it is not bound to a specific channel.

Buffer design

The driver uses fixed-size buffers. Related definitions:

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

• Up to 512 buffers
• Half RX, half TX
• Each buffer is 512 bytes
• Maximum total memory 512 * 512 = 256 KiB

In other words:

总 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

Allocated in probe:

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

Then split:

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

Each message sent has a common header:

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

Meaning:

field	meaning
src	Source endpoint address
dst	Destination endpoint address
reserved	Reserved
len	payload length
flags	Message flags
data[]	Actual payload data

After Linux receives a message, it will, based on thedstaddress, find the local endpoint, then call the callback of that endpoint.

get_a_tx_buf()

Code:

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

This function is a very simple TX buffer allocator.

It has two phases:

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

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

last_sbufIt is only useful in the first phase.

Once:

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

Afterwards, it always goes:

virtqueue_get_buf(vrp->svq, &len);

In other words:

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

virtio API

API	Function
virtio_find_vqs()	Find/create virtqueue
virtqueue_add_inbuf()	Give the device a “writable” buffer
virtqueue_add_outbuf()	Give the device a “readable” buffer
virtqueue_add_sgs	Generic buffer addition function, including in and out
virtqueue_get_buf()	Retrieve the buffer processed by the device from the used ring
virtqueue_kick()	Notify the other side: the queue has a new buffer
virtqueue_kick_prepare()	Determine whether notification is needed
virtqueue_notify()	Actually send the notification
virtqueue_enable_cb()	Enable virtqueue callback/interrupt
virtqueue_disable_cb()	Disable virtqueue callback/interrupt to prevent interrupt storms during polling
virtqueue_get_vring_size()	Get ring size
virtio_has_feature()	Check virtio feature
virtio_device_ready()	Set DRIVER_OK, the device can start working

probe function

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.probeThe function is defined as:

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;

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

    vrp->vdev = vdev;
    
    // Initialize members of 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;
}

The analysis is as follows

Allocationvirtproc_info

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

Initialization:

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

Find virtqueue

The driver requires two virtqueues:

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

Then:

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

Two queues:

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

Meaning:

virtqueue	Purpose	callback
rvq/ input	Linux receives messages sent from remote	rpmsg_recv_done()
svq/ output	Linux sends messages to remote	rpmsg_xmit_done()

Calculate buffer count

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;

That is to say: if the vring is small, reduce the number of buffers according to the vring capacity, otherwise at most 512 buffers.

Allocate coherent DMA buffer

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

This buffer is shared DMA memory that can be accessed by both the main core and the virtio backend.

Put the RX buffer into the RX virtqueue in advance

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

This step is crucial:

Linux first places a batch of empty RX buffers into the available ring, so that the remote processor can later write messages into these buffers.

Disable TX complete interrupt by default

virtqueue_disable_cb(vrp->svq);

Because most of the time, the sender does not need to receive an interrupt every time a TX buffer is consumed by the remote end. Only when Linux has no TX buffer and the sender has to sleep and wait, does it temporarily enable the TX complete interrupt.

Create name service endpoint

If the remote end supports:

VIRTIO_RPMSG_F_NS

Then the created address is53Name service endpoint:

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

Address definition:

#define RPMSG_NS_ADDR			(53)

This endpoint is specifically for handling remote service creation/destruction notifications.

Device ready and notify remote end

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

The order here is very important:

1. RX buffer is ready (virtqueue_kick_prepare)
2. Set the virtio device to ready
3. Notify the remote end that it can start sending messages

Finally print:

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

remove function

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

process:

1. reset virtio device

vdev->config->reset(vdev);

First stop the device to avoid continuing to send and receive.

2. Delete all child rpmsg devices

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

Beforerpmsg_remove_device()Inside, call:

device_unregister(dev);

3. Destroy 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);

Send message(rpmsg_send_offchannel_raw())

All send APIs will eventually enterrpmsg_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;
}

The call chain is roughly:

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

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

Parameter check

First check src/dst:

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

Broadcast address is not allowed as the actual sending address.
Then check length:

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

Because a single buffer is fixed at 512 bytes, the payload cannot exceed512 - sizeof(struct rpmsg_hdr)

Get TX buffer

msg = get_a_tx_buf(vrp);

Its logic:

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

Meaning:

1. Initial stage: directly take a never-used buffer from the TX buffer pool
2. Subsequent stage: fromsvqreclaim TX buffers that have been consumed by the remote end from the used ring

What to do when there is no TX buffer?

If intrysend：

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

If it is a normalsend, then wait, up to 15 seconds:

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

If timeout:

return -ERESTARTSYS;

Before waiting, it will call:

rpmsg_upref_sleepers(vrp);

Purpose: If this is the first sender sleeping and waiting for a TX buffer, enable the TX complete interrupt.

virtqueue_enable_cb(vrp->svq);

After waiting ends, it will call:

rpmsg_downref_sleepers(vrp);

If there are no more waiters, disable the TX complete interrupt:

virtqueue_disable_cb(vrp->svq);

Fill in the rpmsg header and 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);

Note that here it usescpu_to_virtio16/32(), because virtio devices may have specific byte-order requirements.

Add to the TX virtqueue and kick the remote side

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

Then:

virtqueue_kick(vrp->svq);

Full meaning:
Linux fills the TX buffer → adds the buffer to the output virtqueue → kicks the remote processor → the remote side takes the message from the virtqueue

TX complete callback

Set in the probe function:

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

That is, the TX callback function isrpmsg_xmit_done

Note: Normally the TX complete interrupt is disabled; it is only enabled when a sender is sleeping and waiting for a 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);
}

When the remote side consumes the TX buffer, the virtio backend puts the buffer into the used ring and triggers the TX complete callback. Linux wakes up the sender thread waiting for a TX buffer.

TX sleepers mechanism: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);
}

Meaning: Before the sender thread goes to sleep because there is no TX buffer:sleepers++, if this is the first waiter, enable the TX complete callback
Correspondingly, there is alsorpmsg_downref_sleepers(): After the sender thread wakes up:sleepers--, if this is the last waiter, disable the TX complete callback

Purpose of this mechanism:

• No one waiting for TX buffer: disable TX complete interrupt to reduce overhead.
• Someone waiting for TX buffer: enable TX complete interrupt, and wake up the waiter as soon as the remote side returns the buffer.

Full path:

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

Receive message(rpmsg_recv_done)

Set in the probe function:

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

That is, the RX callback function isrpmsg_recv_done, the function is defined as follows:

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

Get used buffer from RX virtqueue

msg = virtqueue_get_buf(rvq, &len);

If a msg can be obtained, try to continue and loop through all available messages:

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

    msgs_received++;

    msg = virtqueue_get_buf(rvq, &len);
}

Single message processing: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;
}

On reception, it is not dispatched based on channel name, but based onmsg->dst, invrp->endpointslook up the local endpoint. So rpmsg at runtime is more like address-based messaging, and the channel name is mainly used for device discovery and driver matching.
Main steps:

1. Parse payload length

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

2. Check whether the message length is valid

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

Prevent out-of-bounds access caused by abnormal length passed from the remote side.

3. According todstLook up endpoint

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

This is the core of rpmsg dispatch:msg->dst == local endpoint address

4. Increase endpoint reference count

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

Avoid the endpoint being released during callback execution.

5. Invoke callback

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

The parameters passed to the upper-layer callback include:

Parameters	meaning
ept->rpdev	Corresponding rpmsg device
msg->data	payload
msg_len	payload length
ept->priv	endpoint private data
msg->src	remote source address

6. Put the RX buffer back into the virtqueue
   After processing, this buffer needs to be returned to the remote side for reuse:

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

Kick the remote

Finally, inrpmsg_recv_done()if a message was processed, kick the remote:

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

name service mechanism

Note: name service is an optional feature. Feature definition:

#define VIRTIO_RPMSG_F_NS	0

If the remote does not support this feature, dynamic service discovery is unavailable. In this case, the channel may need to be created statically.

rpmsg allows the remote to dynamically notify Linux: ‘I created a service’ or ‘I destroyed a service’. This is called name service. Related structure:

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

Fields:

field	meaning
name	Service name
addr	remote service address
flags	Create or destroy

flags：

RPMSG_NS_CREATE  = 0
RPMSG_NS_DESTROY = 1

NS endpoint

The name service uses a fixed address:

#define RPMSG_NS_ADDR			(53)

Created at probe:

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

Note hererpdevthe parameter isNULL, because the NS endpoint does not belong to any ordinary rpmsg channel, but is an internal endpoint used by the bus itself.

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

After receiving a remote NS message:

1. Check length
2. Ensure the endpoint is not bound to a real onerpdev
3. Fix the end of the name string
4. Constructrpmsg_channel_info
5. Create or destroy the channel based on flags

Create channel:

newch = rpmsg_create_channel(vrp, &chinfo);

Destroy channel:

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

Local service announce mechanism

This driver can not only receive remote NS, but also announce local services to the remote side.

Related functions:

virtio_rpmsg_announce_create()
virtio_rpmsg_announce_destroy()

When a local rpmsg channel that needs to be announced is created, the driver constructsrpmsg_ns_msg, and sends it to the remote NS address53(inrpmsg_core.c). Create notification:

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

Destroy notification:

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

rpmsg channel creation

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

It, based on the passed-instruct rpmsg_channel_info *chinfocreates arpmsg_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;
};

process:

1. Check whether the same channel already exists

tmp = rpmsg_find_device(dev, chinfo);

If it already exists, do not create it again.

2. Allocationvirtio_rpmsg_channel

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

This structure:

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

It is a virtio transport private channel, exposing itsrpmsg_device。

3. paddingrpmsg_device

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

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

then register:

ret = rpmsg_register_device(rpdev);

After registration, the rpmsg bus matches the id table of the upper-layer rpmsg driver, and then calls the corresponding driver’s probe.

Overall send/receive flow diagram

Linux sends messages to remote

上层 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 receives a message from the remote side

远端处理器写入 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

Remote side publishes a service

远端发送 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 lifecycle

RX buffer lifecycle:

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

Therefore, the RX buffer is cyclically reused.

TX buffer lifecycle

TX buffer lifecycle:

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

Reference documentation

[1]

AMPRelated：3 Linux rpmsgsubsystem(STM32MP157Solution)

[2]

Multi-core heterogeneous inter-core communication–ipcc

[3]

Remote Processor Messaging (RPMsg)