Cover image for Virtio Rpmsg Bus

Virtio Rpmsg Bus

Created:Updated:
Linux
, 
Words 8k
Views
Visitors
Timeline

Timeline

2026-06-22

init

This article introduces the implementation principles and core data structures of the virtio-based rpmsg bus driver in the Linux kernel. It discusses the basic communication model of rpmsg and provides a detailed analysis of rpmsg_device、rpmsg_endpoint creation and destruction logic, as well as the underlying transmission mechanisms in the virtproc_info private state, including virtqueue send/receive channels, buffer management, DMA mapping, and concurrency control.

linux 5.10.238

rpmsg.drawio
rpmsg.drawio

rpmsg structure

rpmsg has three layers:

1
2
3
4
5
6
7
8
9
10
11
12
┌────────────────────────────────────────┐
│ 业务层：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 virtio rpmsg devices
  2. Send and receive messages using vring
  3. Create/destroy rpmsg_device based on name service

It producesrpmsg_device, but does not consume. The consumer isrpmsg_driver。

module_init/module_exit

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
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 virtio-based rpmsg bus driver. Its role is:

To allow the Linux main core and remote processors to send and receive rpmsg messages via the virtio vring mechanism, and to abstract remote services into Linux’srpmsg_devicefor binding and use by upper-layer rpmsg drivers.
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 (channel), which is a ‘device’
  • rpmsg_driverThe driver that handles the business of this channel
  • rpmsg_endpointThe actual send/receive endpoint on the channel (address + callback)

The relationship among the three:

1
2
3
4
5
rpmsg_driver  ←—— 匹配 ——→  rpmsg_device
                                 |
                                 | 持有
                                 v
                            rpmsg_endpoint

rpmsg_device is the ‘driven-managed device’ in the middle. When creatingrpmsg_devicewill trigger the probe function for matching, and call upon successful matchrpmsg_driver->probefunction

struct rpmsg_device

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/**
 * 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 arpmsg_device's operation

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
/**
* 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

Invirtio_rpmsg_bus.cin, the channel is created as follows (rpmsg_create_channel()）：

1
2
3
4
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 end announces a service “rpmsg-demo”, Linux creates arpmsg_deviceto represent this channel; or the local end actively creates a service “rpmsg-demo”, that is, creates arpmsg_deviceto represent the local channel. After announcing, the NS service notifies the remote end to also create a channel for “rpmsg-demo”. The local and remoterpmsg_device's src and dst are opposite to each other, communicating through this channel.

Actuallyvirtio_rpmsg_bus.cthe abstracted rpmsg channel descriptor in is

1
2
3
4
5
6
7
8
9
10
11
12
13
/**
 * 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;
};

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

struct rpmsg_endpoint

rpmsg_deviceis like a telephone line + an extension service, whilerpmsg_endpointrepresents the person who actually answers the phone

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
/**
 * 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 logic related to rpmsg_endpoint is mainly inrpmsg_core.c, among which the most important isrpmsg_rx_cb_t cbandconst struct rpmsg_endpoint_ops *opsthese two members:

1
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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/**
 * 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(), which is the operation function of rpmsg_device

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
/* 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. Allocatestruct rpmsg_endpoint
  2. Initialize reference count and callback lock
  3. Record callback, private data, ops
  4. Assign local address to endpoint
  5. Insertvrp->endpointsThis idr

Key logic:

1
2
3
4
5
6
7
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 address starts from1024Start
  • 0 ~ 1023Reserved for predefined services
    And the reserved address definition:
1
#define RPMSG_RESERVED_ADDRESSES	(1024)

endpoint address is assigned viaidr_alloc()Assignment:

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

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

  1. fromidrDelete endpoint in
  2. Set callback toNULL
  3. Decrement reference count, release endpoint if necessary

Key code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
/**
 * __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 idr: prevent new RX messages from finding this endpoint
  • Setcb = NULL: prevent in-flight RX that has already obtained the endpoint from calling the callback
  • kref: prevent the endpoint from being released while the RX path is using it

struct virtproc_info

Core data structure:struct virtproc_info

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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 = vrp, divided by function

  • virtio basics:
    • vdevunderlying virtio device
  • TX/RX channels:
    • rvq,svqRX/TX virtqueue
  • buffer management:
    • rbufs,sbufsRX/TX buffer virtual address
    • num_bufstotal buffer count
    • buf_sizesingle buffer size
    • last_sbufTX frontier cursor
    • bufs_dmabuffer DMA base address
  • TX synchronization:
    • tx_lockprotects svq/sbufs/sleepers
    • sendqwait queue for waiting on TX buffer
    • sleeperswaiter count (controls tx-complete interrupt switch)
  • endpoint management:
    • endpointsendpoint idr (lookup by address)
    • endpoints_lockprotects the endpoint table
  • name service：
    • ns_eptname service endpoint (addr 53)

struct virtio_device *vdev

Function: 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 dbg)
- vdev->config virtio config operations (reset/del_vqs etc.)
- vdev->priv points back to vrp itself
set in probe:

1
2
3
vrp->vdev = vdev;
...
vdev->priv = vrp;

In the callback, it relies on this to retrieve vrp:

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

struct virtqueue *rvq, *svq

Function:

  • rvq = receive virtqueue: remote -> Linux
  • svq = send virtqueue: Linux -> remote
    These are the two channels for receiving and sending. In probe:
1
2
vrp->rvq = vqs[0];   /* input  */
vrp->svq = vqs[1];   /* output */
  • Receive message:virtqueue_get_buf(rvq)/virtqueue_add_inbuf(rvq)
  • Send message:virtqueue_add_outbuf(svq)/virtqueue_get_buf(svq)

void *rbufs, *sbufs

Function:

  • rbufs: starting kernel virtual address of the RX buffer region

  • sbufs: starting kernel virtual address of the TX buffer region

    In probe, a whole block of DMA memory is cut in half:

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

As shown in the figure

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

get_a_tx_buf()It means taking the TX buffer by index from sbufs:

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

unsigned int num_bufs

Function: total number of RX + TX buffers (each takes up half). Calculated in probe based on vring size:

1
2
3
4
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 */

Therefore:

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

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

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

unsigned int buf_size

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

In the probe function

1
vrp->buf_size = MAX_RPMSG_BUF_SIZE;

It determines:

  • Address stride of each buffer:sbufs + buf_size * i
  • Payload limit of a single message
1
2
if (len > vrp->buf_size - sizeof(struct rpmsg_hdr))
      return -EMSGSIZE;

int last_sbuf

Function: TX buffer’s ‘initial allocation cursor’, records which one is sequentially allocated for the first time. Inget_a_tx_buf()used in:

1
2
3
4
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 unused new TX buffers, take them in order
  • Otherwise: all TX buffers have been used, switch to recycling and reusing from the used ring
    Once the upper limit is reached,last_sbufit is fixed and no longer grows, subsequently relying entirely onvirtqueue_get_buf()recycling.

dma_addr_t bufs_dma

Function: DMA base address of the entire buffer (address from the device/DMA perspective). In probe, bydma_alloc_coherent()returns:

1
2
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, bufs_dma is the physical/bus address used for releasing and DMA mapping. It is used to release during remove:

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

struct mutex tx_lock

Function: 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 “dozing” remote processor, a process that may sleep, so a sleepable mutex must be used.

struct idr endpoints

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

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

When receiving a message, look up by dst:

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

So rpmsg is actuallyaddress-based dispatch, relying on this idr.

struct mutex endpoints_lock

Function: Protects concurrent access to the endpoints idr. Must be held when adding, deleting, querying, or modifying endpoints:

1
2
3
4
5
mutex_lock(&vrp->endpoints_lock);
ept = idr_find(&vrp->endpoints, ...);
if (ept)
      kref_get(&ept->refcount);   /* After finding it, increment the reference count first to prevent release */
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

Function: Senders waiting for TX buffers sleep on this wait queue. When there are no TX buffers, rpmsg_send() sleeps and waits:

1
2
3
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:

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

atomic_t sleepers

Function: How many senders are currently waiting for TX buffers (waiter count). It coordinates with the dynamic switching of the TX complete interrupt:

1
2
3
4
5
6
7
/* 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 design for 5.10; in Linux 7.1.1, sleepers were removed 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 regular rpmsg channel; it is used internally by the bus to handle remote service ‘create/destroy’ notifications. Created in probe when the remote supports the NS feature:

1
2
3
4
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 during remove:

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

Note that rpdev is passed as NULL during creation because it is not bound to a specific channel.

Buffer design

The driver uses fixed-size buffers. Related definitions:

1
2
#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:

1
2
3
4
5
6
7
8
9
10
总 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:

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

Then split:

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

Each sent message has a common header:

1
2
3
4
5
6
7
8
struct rpmsg_hdr {
    __virtio32 src;
    __virtio32 dst;
    __virtio32 reserved;
    __virtio16 len;
    __virtio16 flags;
    u8 data[];
} __packed;

Meaning:

FieldMeaning
srcSource endpoint address
dstDestination endpoint address
reservedReserved
lenPayload length
flagsMessage flags
data[]Actual payload data

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

get_a_tx_buf()

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/* 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);

4	/*
     * 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 stages:

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

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

last_sbufOnly useful in the first stage.

Once:

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

Then it always proceeds:

1
virtqueue_get_buf(vrp->svq, &len);

In other words:

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

virtio API

APIFunction
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_sgsThe generic buffer add function includes in and out
virtqueue_get_buf()Retrieve the buffer processed by the device from the used ring
virtqueue_kick()Notify the other party: 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, used 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

1
2
3
4
5
6
7
8
9
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:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
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 the 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

Allocatevirtproc_info

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

Initialization:

1
2
3
4
idr_init(&vrp->endpoints);
mutex_init(&vrp->endpoints_lock);
mutex_init(&vrp->tx_lock);
init_waitqueue_head(&vrp->sendq);

Find virtqueue

The driver needs two virtqueues:

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

Then:

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

Two queues:

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

Meaning:

virtqueueUsagecallback
rvq/ inputLinux receives messages from the remote endrpmsg_recv_done()
svq/ outputLinux sends a message to the remote endrpmsg_xmit_done()

Calculate the number of buffers

1
2
3
4
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;

In other words: if the vring is small, reduce the number of buffers according to the vring’s capacity; otherwise, use a maximum of 512 buffers

Allocate coherent DMA buffer

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

This buffer is shared DMA memory accessible by both the main core and the virtio backend.

Pre-load RX buffers into the RX virtqueue

1
2
3
4
5
6
7
8
9
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 critical:

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

1
virtqueue_disable_cb(vrp->svq);

Because most of the time, the sender does not need an interrupt every time a TX buffer is consumed by the remote end. The TX complete interrupt is only temporarily enabled when Linux has no TX buffers and the sender needs to sleep and wait.

Create name service endpoint

If the remote end supports it:

1
VIRTIO_RPMSG_F_NS

then create an endpoint with the address53as the name service endpoint:

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

Address definition:

1
#define RPMSG_NS_ADDR			(53)

This endpoint specifically handles remote service creation/destruction notifications.

Device ready and notify the remote end

1
2
3
4
notify = virtqueue_kick_prepare(vrp->rvq);
virtio_device_ready(vdev);
if (notify)
    virtqueue_notify(vrp->rvq);

The order here is important:

  1. RX buffers are ready (virtqueue_kick_prepare)
  2. virtio device set to ready
  3. Notify the remote end that it can start sending messages

Finally print:

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

remove function

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
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
1
vdev->config->reset(vdev);

First, stop the device to prevent further transmission and reception.

  1. Delete all child rpmsg devices
1
device_for_each_child(&vdev->dev, NULL, rpmsg_remove_device);

Inrpmsg_remove_device()Called within:

1
device_unregister(dev);
  1. Destroy NS endpoint
1
2
if (vrp->ns_ept)
    __rpmsg_destroy_ept(vrp, vrp->ns_ept);
  1. endpoints、virtqueue、DMA buffer、vrp
1
2
3
4
idr_destroy(&vrp->endpoints);
vdev->config->del_vqs(vrp->vdev);
dma_free_coherent(...);
kfree(vrp);

Send message(rpmsg_send_offchannel_raw())

All sending APIs will eventually enterrpmsg_send_offchannel_raw()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
/**
 * 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:

1
2
3
4
5
6
7
8
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:

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

Broadcast addresses are not allowed as actual sending addresses.
Then check the length:

1
2
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

1
msg = get_a_tx_buf(vrp);

Its logic:

1
2
3
4
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 phase: directly take an unused buffer from the TX buffer pool
  2. Subsequent phase: fromsvqReclaim TX buffers that have been consumed by the remote end from the used ring

What to do when there is no TX buffer?

Iftrysend：

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

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

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

If timeout:

1
return -ERESTARTSYS;

Before waiting, it will call:

1
rpmsg_upref_sleepers(vrp);

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

1
virtqueue_enable_cb(vrp->svq);

Called after waiting ends:

1
rpmsg_downref_sleepers(vrp);

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

1
virtqueue_disable_cb(vrp->svq);

Fill the rpmsg header and payload

1
2
3
4
5
6
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 the use ofcpu_to_virtio16/32(), because virtio devices may have specific endianness requirements.

Add to TX virtqueue and kick the remote end

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

Then:

1
virtqueue_kick(vrp->svq);

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

TX complete callback

Set in the probe function

1
2
3
4
5
6
7
8
9
10
11
12
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);
    
    .....
}

i.e., the tx callback function isrpmsg_xmit_done

Note: The TX complete interrupt is usually disabled; it is only enabled when a sender is sleeping and waiting for a buffer.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
/*
 * 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 end consumes the TX buffer, the virtio backend puts the buffer into the used ring and triggers the TX complete callback. Linux wakes up the sending thread waiting for the TX buffer.

TX sleepers mechanism:rpmsg_upref_sleepers()

1
2
3
4
5
6
7
8
9
10
11
12
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 sending thread prepares 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 sending thread wakes up:sleepers--, if this is the last waiter, disable the TX complete callback

The purpose of this mechanism:

  • No one waiting for TX buffer: disable the TX complete interrupt to reduce overhead
  • Someone waiting for TX buffer: enable the TX complete interrupt to immediately wake up the waiter after the remote end returns the buffer

Complete chain:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
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

1
2
3
4
5
6
7
8
9
10
11
12
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);
    
    .....
}

i.e., the rx callback function isrpmsg_recv_done, the function is defined as follows:

rpmsg_recv_done

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
/* 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);
}

Fetch used buffer from RX virtqueue

1
msg = virtqueue_get_buf(rvq, &len);

If a msg is obtained, try to continue, looping to process all available messages:

1
2
3
4
5
6
7
8
9
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()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
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;
}

Upon reception, distribution is not based on channel name, but on:msg->dst, invrp->endpointslook up the local endpoint. Therefore, rpmsg actually operates more like address-based messaging at runtime; the channel name is mainly used for device discovery and driver matching.
Main steps:

  1. Parse payload length
1
msg_len = virtio16_to_cpu(vrp->vdev, msg->len);
  1. Check if the message length is valid
1
2
3
4
if (len > vrp->buf_size ||
    msg_len > (len - sizeof(struct rpmsg_hdr))) {
    return -EINVAL;
}

Prevent out-of-bounds access caused by abnormal lengths passed from the remote end.

  1. According todstFind endpoint
1
ept = idr_find(&vrp->endpoints, virtio32_to_cpu(vrp->vdev, msg->dst));

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

  1. Increment endpoint reference count
1
2
if (ept)
    kref_get(&ept->refcount);

Prevent the endpoint from being released during callback execution.

  1. Invoke callback
1
2
3
if (ept->cb)
    ept->cb(ept->rpdev, msg->data, msg_len, ept->priv,
        virtio32_to_cpu(vrp->vdev, msg->src));

Parameters passed to the upper-layer callback include:

ParameterMeaning
ept->rpdevCorresponding rpmsg device
msg->datapayload
msg_lenPayload length
ept->privEndpoint private data
msg->srcRemote source address
  1. Put the RX buffer back into the virtqueue
    After processing, this buffer needs to be given back to the remote for use:
1
2
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 messages have been processed, kick the remote:

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

name service mechanism

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

1
#define VIRTIO_RPMSG_F_NS	0

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

rpmsg supports the remote dynamically notifying Linux: I have created a service or I have destroyed a service, this is called name service. Related structure:

1
2
3
4
5
struct rpmsg_ns_msg {
    char name[RPMSG_NAME_SIZE];
    __virtio32 addr;
    __virtio32 flags;
} __packed;

Fields:

FieldMeaning
nameService name
addrRemote service address
flagsCreate or destroy

flags：

1
2
RPMSG_NS_CREATE  = 0
RPMSG_NS_DESTROY = 1

NS endpoint

name service uses a fixed address:

1
#define RPMSG_NS_ADDR			(53)

Created during probe:

1
2
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 a regular rpmsg channel, but is an internal endpoint used by the bus itself.

NS callback：rpmsg_ns_cb()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
/* 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 the remote NS message:

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

Create channel:

1
newch = rpmsg_create_channel(vrp, &chinfo);

Destroy channel:

1
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 end.

Related functions:

1
2
virtio_rpmsg_announce_create()
virtio_rpmsg_announce_destroy()

When an rpmsg channel that needs to be announced is created locally, the driver constructsrpmsg_ns_msg, and sends it to the remote NS address53(inrpmsg_core.cin). Creation notification:

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

Destruction notification:

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

rpmsg channel creation

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
/*
 * 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。

1
2
3
4
5
6
7
8
9
10
11
/**
 * 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 if the same channel already exists
1
tmp = rpmsg_find_device(dev, chinfo);

If it already exists, do not create it again.

  1. Allocatevirtio_rpmsg_channel
1
vch = kzalloc(sizeof(*vch), GFP_KERNEL);

This structure:

1
2
3
4
struct virtio_rpmsg_channel {
    struct rpmsg_device rpdev;
    struct virtproc_info *vrp;
};

It is a virtio transport private channel, exposing itsrpmsg_device。

  1. Fillrpmsg_device
1
2
3
4
5
rpdev->src = chinfo->src;
rpdev->dst = chinfo->dst;
rpdev->ops = &virtio_rpmsg_ops;

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

Then register:

1
ret = rpmsg_register_device(rpdev);

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

Overall send/receive flow chart

Linux sends a message to the remote end

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
上层 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 end

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
远端处理器写入 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 end publishes service

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
远端发送 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:

1
2
3
4
5
6
7
8
9
10
11
probe 时 add_inbuf
    ↓
远端写入消息
    ↓
Linux virtqueue_get_buf
    ↓
调用 callback
    ↓
Linux 重新 add_inbuf
    ↓
远端再次使用

So the RX buffer is recycled.

TX buffer lifecycle

TX buffer lifecycle:

1
2
3
4
5
6
7
8
9
10
11
12
13
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)