Rpmsg Char Driver

---

Timeline

Timeline

2026-06-29

init

This article introduces the design and implementation of the Rpmsg character device driver in Linux 5.10.x. The driver registers with the Rpmsg bus and adopts a mode that separates the control device from the endpoint device: the control node creates endpoints through ioctl, and the endpoint device handles user-space message reading and writing. The article analyzes key data structures, the use of locks and wait queues, the probe/remove flow, the deferred endpoint creation mechanism, and details such as blocking/wakeup and memory copy in the read/write path, and also points out caveats such as buffer size.

---

The current code analysis is based on Linux 5.10.x.

rpmsg.drawio

rpmsg_char.cIt implementsrpmsg_driverand registers to the Rpmsg Bus, providing an interface for interaction with user space, and undertakesvirtio_rpmsg_bus.cregisteredrpmsg_device

module_init / module_exit

#define RPMSG_DEV_MAX	(MINORMASK + 1)

static dev_t rpmsg_major;
static struct class *rpmsg_class;

static int rpmsg_char_init(void)
{
	int ret;

	ret = alloc_chrdev_region(&rpmsg_major, 0, RPMSG_DEV_MAX, "rpmsg");
	if (ret < 0) {
		pr_err("rpmsg: failed to allocate char dev region\n");
		return ret;
	}

	rpmsg_class = class_create(THIS_MODULE, "rpmsg");
	if (IS_ERR(rpmsg_class)) {
		pr_err("failed to create rpmsg class\n");
		unregister_chrdev_region(rpmsg_major, RPMSG_DEV_MAX);
		return PTR_ERR(rpmsg_class);
	}

	ret = register_rpmsg_driver(&rpmsg_chrdev_driver);
	if (ret < 0) {
		pr_err("rpmsgchr: failed to register rpmsg driver\n");
		class_destroy(rpmsg_class);
		unregister_chrdev_region(rpmsg_major, RPMSG_DEV_MAX);
	}

	return ret;
}
postcore_initcall(rpmsg_char_init);

static void rpmsg_chrdev_exit(void)
{
	unregister_rpmsg_driver(&rpmsg_chrdev_driver);
	class_destroy(rpmsg_class);
	unregister_chrdev_region(rpmsg_major, RPMSG_DEV_MAX);
}
module_exit(rpmsg_chrdev_exit);

MODULE_ALIAS("rpmsg:rpmsg_chrdev");
MODULE_LICENSE("GPL v2");

rpmsg_char.cis a character device driver, with the hierarchy as follows:

rpmsg_char.drawio

Note that it usespostcore_initcall, compared tomodule_initearly, becauserpmsg_core.cthe bus also usespostcore_initcall, both are registered after the driver model is ready.

rpmsg_char_initIt does three things:

1. alloc_chrdev_region(&rpmsg_major, 0, RPMSG_DEV_MAX, "rpmsg") — dynamically allocate a major device number,RPMSG_DEV_MAX = MINORMASK+1(i.e., the entire minor device number space).
2. class_create(THIS_MODULE, "rpmsg")— create/sys/class/rpmsg, so thatcdev_device_addafterwards will automatically be/dev/under … generate a device node (depending onudev/devtmpfs）。
3. register_rpmsg_driver(&rpmsg_chrdev_driver)— register an rpmsg driver (not a platform driver! It is a driver attached to the rpmsg bus).

Andrpmsg_chrdev_driverAs follows:

static struct rpmsg_driver rpmsg_chrdev_driver = {
	.probe = rpmsg_chrdev_probe,
	.remove = rpmsg_chrdev_remove,
	.drv = {
		.name = "rpmsg_chrdev",
	},
};

Key Data Structures

Type	Device Node	struct	Who creates	Function
Control device	/dev/rpmsg_ctrl0	rpmsg_ctrldev	rpmsg_chrdev_probe	One per channel; the user “creates endpoints” on it via ioctl
Endpoint device	/dev/rpmsg0	rpmsg_eptdev	rpmsg_ctrldev_ioctl(CREATE)	One per endpoint; the user reads/writes/polls it to send and receive messages

This separation of “control device + data device” is a classic pattern in Linux:Control nodes are used to instantiate data nodes, and data nodes carry the actual I/O.。

struct rpmsg_ctrldev

/**
 * struct rpmsg_ctrldev - control device for instantiating endpoint devices
 * @rpdev:	underlying rpmsg device
 * @cdev:	cdev for the ctrl device
 * @dev:	device for the ctrl device
 */
struct rpmsg_ctrldev {
    struct rpmsg_device *rpdev;   // Underlying rpmsg channel device
    struct cdev cdev;              // Character device
    struct device dev;             // Embedded device, registered under rpmsg_class
};

struct rpmsg_eptdev

/**
 * struct rpmsg_eptdev - endpoint device context
 * @dev:	endpoint device
 * @cdev:	cdev for the endpoint device
 * @rpdev:	underlying rpmsg device
 * @chinfo:	info used to open the endpoint
 * @ept_lock:	synchronization of @ept modifications
 * @ept:	rpmsg endpoint reference, when open
 * @queue_lock:	synchronization of @queue operations
 * @queue:	incoming message queue
 * @readq:	wait object for incoming queue
 */
struct rpmsg_eptdev {
    struct device dev;
    struct cdev cdev;

    struct rpmsg_device *rpdev;          // Underlying channel
    struct rpmsg_channel_info chinfo;    // Channel information used at open time (name/src/dst)

    struct mutex ept_lock;               // Protects modification of the ept pointer
    struct rpmsg_endpoint *ept;          // The real endpoint created only at open time, can be NULL

    spinlock_t queue_lock;               // Protects the receive queue
    struct sk_buff_head queue;           // Receive message queue (one skb per message)
    wait_queue_head_t readq;             // Wait queue for blocking reads
};

Lock and wait queue:

• struct mutex ept_lock;: Protects the ept pointer. ept becomes NULL in three cases:release、destroy ioctl, parent deviceremove. The read/write path must first take this lock to confirm ept is still alive.
• spinlock_t queue_lock;: Protects the queue (skb list). The callbackrpmsg_ept_cbenqueues in interrupt/atomic context, read_iter dequeues in process context, so a spinlock is used here and in cb usespin_lock, in read usespin_lock_irqsave。
• wait_queue_head_t readq;：rpmsg_ept_cbafter enqueueingwake_up_interruptible，read_iterwhen the queue is emptywait_event_interruptible. This is a typical “producer-consumer + wait queue”.

rpmsg_driver

static struct rpmsg_driver rpmsg_chrdev_driver = {
	.probe = rpmsg_chrdev_probe,
	.remove = rpmsg_chrdev_remove,
	.drv = {
		.name = "rpmsg_chrdev",
	},
};

Note: no callback function is defined here

rpmsg_chrdev_probe()

#define RPMSG_DEV_MAX	(MINORMASK + 1)

static dev_t rpmsg_major;
static struct class *rpmsg_class;

static DEFINE_IDA(rpmsg_ctrl_ida);
static DEFINE_IDA(rpmsg_ept_ida);
static DEFINE_IDA(rpmsg_minor_ida);

static int rpmsg_chrdev_probe(struct rpmsg_device *rpdev)
{
	struct rpmsg_ctrldev *ctrldev;
	struct device *dev;
	int ret;

	ctrldev = kzalloc(sizeof(*ctrldev), GFP_KERNEL);
	if (!ctrldev)
		return -ENOMEM;

	ctrldev->rpdev = rpdev;

	dev = &ctrldev->dev;
	device_initialize(dev);
	dev->parent = &rpdev->dev;
	dev->class = rpmsg_class;

	cdev_init(&ctrldev->cdev, &rpmsg_ctrldev_fops);
	ctrldev->cdev.owner = THIS_MODULE;

	ret = ida_simple_get(&rpmsg_minor_ida, 0, RPMSG_DEV_MAX, GFP_KERNEL);
	if (ret < 0)
		goto free_ctrldev;
	dev->devt = MKDEV(MAJOR(rpmsg_major), ret);

	ret = ida_simple_get(&rpmsg_ctrl_ida, 0, 0, GFP_KERNEL);
	if (ret < 0)
		goto free_minor_ida;
	dev->id = ret;
	dev_set_name(&ctrldev->dev, "rpmsg_ctrl%d", ret);

	ret = cdev_device_add(&ctrldev->cdev, &ctrldev->dev);
	if (ret)
		goto free_ctrl_ida;

	/* We can now rely on the release function for cleanup */
	dev->release = rpmsg_ctrldev_release_device;

	dev_set_drvdata(&rpdev->dev, ctrldev);

	return ret;

free_ctrl_ida:
	ida_simple_remove(&rpmsg_ctrl_ida, dev->id);
free_minor_ida:
	ida_simple_remove(&rpmsg_minor_ida, MINOR(dev->devt));
free_ctrldev:
	put_device(dev);
	kfree(ctrldev);

	return ret;
}

When the rpmsg bus matches arpmsg_devicetorpmsg_chrdev_driveris calledrpmsg_chrdev_probe. Process:

1. kzalloc a ctrldev,ctrldev->rpdev = rpdev。
2. device_initialize+ setparent/class，cdev_init(&ctrldev->cdev, &rpmsg_ctrldev_fops)。
3. twiceida_simple_get: fromrpmsg_minor_idaget the minor number, fromrpmsg_ctrl_idaget the id (forrpmsg_ctrl%d）。
4. cdev_device_add— register cdev and device together,/dev/rpmsg_ctrlNappears.
5. Key point:dev->release = rpmsg_ctrldev_release_deviceis incdev_device_addonly assigned afterwards. The comment says “We can now rely on the release function for cleanup” — meaningcdev_device_addIf it fails before success, use the manual error label to release; after success, it is left to the device model’s release callback as a fallback (kfree when the last reference count drops to zero). This is a very standard refcount + release pattern in the Linux device model.
6. dev_set_drvdata(&rpdev->dev, ctrldev)— so that remove can retrieve ctrldev from rpdev.

rpmsg_chrdev_remove()

static void rpmsg_chrdev_remove(struct rpmsg_device *rpdev)
{
	struct rpmsg_ctrldev *ctrldev = dev_get_drvdata(&rpdev->dev);
	int ret;

	/* Destroy all endpoints */
	ret = device_for_each_child(&ctrldev->dev, NULL, rpmsg_eptdev_destroy);
	if (ret)
		dev_warn(&rpdev->dev, "failed to nuke endpoints: %d\n", ret);

	cdev_device_del(&ctrldev->cdev, &ctrldev->dev);
	put_device(&ctrldev->dev);
}

rpmsg_ctrldev control node

file_operations

static const struct file_operations rpmsg_ctrldev_fops = {
	.owner = THIS_MODULE,
	.open = rpmsg_ctrldev_open,
	.release = rpmsg_ctrldev_release,
	.unlocked_ioctl = rpmsg_ctrldev_ioctl,
	.compat_ioctl = compat_ptr_ioctl,
};

rpmsg_ctrldev_open()

static int rpmsg_ctrldev_open(struct inode *inode, struct file *filp)
{
	struct rpmsg_ctrldev *ctrldev = cdev_to_ctrldev(inode->i_cdev);

	get_device(&ctrldev->dev);
	filp->private_data = ctrldev;

	return 0;
}

rpmsg_ctrldev_release()

static int rpmsg_ctrldev_release(struct inode *inode, struct file *filp)
{
	struct rpmsg_ctrldev *ctrldev = cdev_to_ctrldev(inode->i_cdev);

	put_device(&ctrldev->dev);

	return 0;
}

in the probe function there isdev->release = rpmsg_ctrldev_release_device;, so actuallyreleaseThe function is

static void rpmsg_ctrldev_release_device(struct device *dev)
{
	struct rpmsg_ctrldev *ctrldev = dev_to_ctrldev(dev);

	ida_simple_remove(&rpmsg_ctrl_ida, dev->id);
	ida_simple_remove(&rpmsg_minor_ida, MINOR(dev->devt));
	kfree(ctrldev);
}

rpmsg_ctrldev_ioctl()

static long rpmsg_ctrldev_ioctl(struct file *fp, unsigned int cmd,
				unsigned long arg)
{
	struct rpmsg_ctrldev *ctrldev = fp->private_data;
	void __user *argp = (void __user *)arg;
	struct rpmsg_endpoint_info eptinfo;
	struct rpmsg_channel_info chinfo;

	if (cmd != RPMSG_CREATE_EPT_IOCTL)
		return -EINVAL;

	if (copy_from_user(&eptinfo, argp, sizeof(eptinfo)))
		return -EFAULT;

	memcpy(chinfo.name, eptinfo.name, RPMSG_NAME_SIZE);
	chinfo.name[RPMSG_NAME_SIZE-1] = '\0';
	chinfo.src = eptinfo.src;
	chinfo.dst = eptinfo.dst;

	return rpmsg_eptdev_create(ctrldev, chinfo);
};

use this function to create ept, callrpmsg_eptdev_create()

rpmsg_eptdev_create()

static int rpmsg_eptdev_create(struct rpmsg_ctrldev *ctrldev,
			       struct rpmsg_channel_info chinfo)
{
	struct rpmsg_device *rpdev = ctrldev->rpdev;
	struct rpmsg_eptdev *eptdev;
	struct device *dev;
	int ret;

	eptdev = kzalloc(sizeof(*eptdev), GFP_KERNEL);
	if (!eptdev)
		return -ENOMEM;

	dev = &eptdev->dev;
	eptdev->rpdev = rpdev;
	eptdev->chinfo = chinfo;

	mutex_init(&eptdev->ept_lock);
	spin_lock_init(&eptdev->queue_lock);
	skb_queue_head_init(&eptdev->queue);
	init_waitqueue_head(&eptdev->readq);

	device_initialize(dev);
	dev->class = rpmsg_class;
	dev->parent = &ctrldev->dev;
	dev->groups = rpmsg_eptdev_groups;
	dev_set_drvdata(dev, eptdev);

	cdev_init(&eptdev->cdev, &rpmsg_eptdev_fops);
	eptdev->cdev.owner = THIS_MODULE;

	ret = ida_simple_get(&rpmsg_minor_ida, 0, RPMSG_DEV_MAX, GFP_KERNEL);
	if (ret < 0)
		goto free_eptdev;
	dev->devt = MKDEV(MAJOR(rpmsg_major), ret);

	ret = ida_simple_get(&rpmsg_ept_ida, 0, 0, GFP_KERNEL);
	if (ret < 0)
		goto free_minor_ida;
	dev->id = ret;
	dev_set_name(dev, "rpmsg%d", ret);

	ret = cdev_device_add(&eptdev->cdev, &eptdev->dev);
	if (ret)
		goto free_ept_ida;

	/* We can now rely on the release function for cleanup */
	dev->release = rpmsg_eptdev_release_device;

	return ret;

free_ept_ida:
	ida_simple_remove(&rpmsg_ept_ida, dev->id);
free_minor_ida:
	ida_simple_remove(&rpmsg_minor_ida, MINOR(dev->devt));
free_eptdev:
	put_device(dev);
	kfree(eptdev);

	return ret;
}

1. kzalloc eptdev, fill in rpdev, chinfo.
2. Initialize lock, skb queue, wait queue head.
3. device_initialize + setclass/parent/groups/drvdata。groups = rpmsg_eptdev_groupsprovide name/src/dst read-only attributes to sysfs
4. cdev_init usesrpmsg_eptdev_fops。
5. ida twice_simple_get obtains minor and ept id,dev_set_name(dev, "rpmsg%d", ret)→/dev/rpmsgN。
6. cdev_device_add, and after success set dev->release.
7. Note: at this point,eptdev->ept == NULL! The real endpoint has not been created yet.The endpoint is created only at open. This is lazy creation, to avoid occupying an address without anyone using it.

The goto order in the error handling path is strictly the reverse of the order in which resources are acquired, which is the standard unwind style. Note free_put in eptdev_after device, then kfree — there is actually a detail here:device_initializeafterwards one should use put_device lets the release callback free it, but because release is set only after cdev_device_add succeeds, in this path it hasn’t been set yet, so put_device will not actually kfree, so manual kfree is needed. This is exactly the cost of the delayed assignment of release.

compat_ptr_ioctl()

#ifdef CONFIG_COMPAT
/**
 * compat_ptr_ioctl - generic implementation of .compat_ioctl file operation
 *
 * This is not normally called as a function, but instead set in struct
 * file_operations as
 *
 *     .compat_ioctl = compat_ptr_ioctl,
 *
 * On most architectures, the compat_ptr_ioctl() just passes all arguments
 * to the corresponding ->ioctl handler. The exception is arch/s390, where
 * compat_ptr() clears the top bit of a 32-bit pointer value, so user space
 * pointers to the second 2GB alias the first 2GB, as is the case for
 * native 32-bit s390 user space.
 *
 * The compat_ptr_ioctl() function must therefore be used only with ioctl
 * functions that either ignore the argument or pass a pointer to a
 * compatible data type.
 *
 * If any ioctl command handled by fops->unlocked_ioctl passes a plain
 * integer instead of a pointer, or any of the passed data types
 * is incompatible between 32-bit and 64-bit architectures, a proper
 * handler is required instead of compat_ptr_ioctl.
 */
long compat_ptr_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
{
	if (!file->f_op->unlocked_ioctl)
		return -ENOIOCTLCMD;

	return file->f_op->unlocked_ioctl(file, cmd, (unsigned long)compat_ptr(arg));
}
EXPORT_SYMBOL(compat_ptr_ioctl);
#endif

directly usefs/ioctl.cincompat_ptr_ioctlimplementation, 64-bit kernel compatible with 32-bit userspace

rpmsg_eptdev ept node

file_operations

static const struct file_operations rpmsg_eptdev_fops = {
	.owner = THIS_MODULE,
	.open = rpmsg_eptdev_open,
	.release = rpmsg_eptdev_release,
	.read_iter = rpmsg_eptdev_read_iter,
	.write_iter = rpmsg_eptdev_write_iter,
	.poll = rpmsg_eptdev_poll,
	.unlocked_ioctl = rpmsg_eptdev_ioctl,
	.compat_ioctl = compat_ptr_ioctl,
};

rpmsg_eptdev_open()

static int rpmsg_eptdev_open(struct inode *inode, struct file *filp)
{
	struct rpmsg_eptdev *eptdev = cdev_to_eptdev(inode->i_cdev);
	struct rpmsg_endpoint *ept;
	struct rpmsg_device *rpdev = eptdev->rpdev;
	struct device *dev = &eptdev->dev;

	get_device(dev);

	ept = rpmsg_create_ept(rpdev, rpmsg_ept_cb, eptdev, eptdev->chinfo);
	if (!ept) {
		dev_err(dev, "failed to open %s\n", eptdev->chinfo.name);
		put_device(dev);
		return -EINVAL;
	}

	eptdev->ept = ept;
	filp->private_data = eptdev;

	return 0;
}

• rpmsg_create_eptforwarded to core and then to the virtio backendvirtio_rpmsg_create_ept→__rpmsg_create_ept（virtio_rpmsg_bus.c). The latter allocates a local address in the idr (ifchinfo.src == RPMSG_ADDR_ANYthen dynamically allocates starting from 1024), and the callbackrpmsg_ept_cbandpriv=eptdevbound to that address.
• After the callback is registered, messages sent from the remote with dst equal to this local address will triggerrpmsg_ept_cb (see backendrpmsg_recv_singleInmsg->dstlooks up the idr and callsept->cb）。
• get_device/ open failureput_device: ensures the device will not be released while the file is open.

And the callback of ept is as follows:

static int rpmsg_ept_cb(struct rpmsg_device *rpdev, void *buf, int len,
			void *priv, u32 addr)
{
	struct rpmsg_eptdev *eptdev = priv;
	struct sk_buff *skb;

	skb = alloc_skb(len, GFP_ATOMIC);
	if (!skb)
		return -ENOMEM;

	skb_put_data(skb, buf, len);

	spin_lock(&eptdev->queue_lock);
	skb_queue_tail(&eptdev->queue, skb);
	spin_unlock(&eptdev->queue_lock);

	/* wake up any blocking processes, waiting for new data */
	wake_up_interruptible(&eptdev->readq);

	return 0;
}

• The backend calls ept->cb in the virtqueue callback (softirq context), so here you cannot use mutex or sleep; use GFP_ATOMIC and spinlocks.
• A memory copy is done here (buf → skb). The backend gets the vring buffer with zero-copy when receiving, but in order to hand the data to userspace and not let the buffer be held by the user for a long time (the backend needs to return the buffer to rvq as soon as possible), a copy is made into skb here.rpmsg_recv_singleimmediately after calling cbvirtqueue_add_inbufreturn the original buffer.

rpmsg_eptdev_release()

static int rpmsg_eptdev_release(struct inode *inode, struct file *filp)
{
	struct rpmsg_eptdev *eptdev = cdev_to_eptdev(inode->i_cdev);
	struct device *dev = &eptdev->dev;

	/* Close the endpoint, if it's not already destroyed by the parent */
	mutex_lock(&eptdev->ept_lock);
	if (eptdev->ept) {
		rpmsg_destroy_ept(eptdev->ept);
		eptdev->ept = NULL;
	}
	mutex_unlock(&eptdev->ept_lock);

	/* Discard all SKBs */
	skb_queue_purge(&eptdev->queue);

	put_device(dev);

	return 0;
}

When the device needs to be completely removed (e.g., parent device remove or userspaceRPMSG_DESTROY_EPT_IOCTL), what is actually called isrpmsg_eptdev_destroy：

static int rpmsg_eptdev_destroy(struct device *dev, void *data)
{
	struct rpmsg_eptdev *eptdev = dev_to_eptdev(dev);

	mutex_lock(&eptdev->ept_lock);
	if (eptdev->ept) {
		rpmsg_destroy_ept(eptdev->ept);
		eptdev->ept = NULL;
	}
	mutex_unlock(&eptdev->ept_lock);

	/* wake up any blocked readers */
	wake_up_interruptible(&eptdev->readq);

	cdev_device_del(&eptdev->cdev, &eptdev->dev);
	put_device(&eptdev->dev);

	return 0;
}

rpmsg_eptdev_read_iter()

static ssize_t rpmsg_eptdev_read_iter(struct kiocb *iocb, struct iov_iter *to)
{
	struct file *filp = iocb->ki_filp;
	struct rpmsg_eptdev *eptdev = filp->private_data;
	unsigned long flags;
	struct sk_buff *skb;
	int use;

	if (!eptdev->ept)
		return -EPIPE;

	spin_lock_irqsave(&eptdev->queue_lock, flags);

	/* Wait for data in the queue */
	if (skb_queue_empty(&eptdev->queue)) {
		spin_unlock_irqrestore(&eptdev->queue_lock, flags);

		if (filp->f_flags & O_NONBLOCK)
			return -EAGAIN;

		/* Wait until we get data or the endpoint goes away */
		if (wait_event_interruptible(eptdev->readq,
					     !skb_queue_empty(&eptdev->queue) ||
					     !eptdev->ept))
			return -ERESTARTSYS;

		/* We lost the endpoint while waiting */
		if (!eptdev->ept)
			return -EPIPE;

		spin_lock_irqsave(&eptdev->queue_lock, flags);
	}

	skb = skb_dequeue(&eptdev->queue);
	spin_unlock_irqrestore(&eptdev->queue_lock, flags);
	if (!skb)
		return -EFAULT;

	use = min_t(size_t, iov_iter_count(to), skb->len);
	if (copy_to_iter(skb->data, use, to) != use)
		use = -EFAULT;

	kfree_skb(skb);

	return use;
}

• also check in the wait condition!ept: this wayWhen destroy sets ept to NULL and calls wake_up, the blocked read can wake up immediately and return -EPIPE, and will not be stuck forever.. This is the standard way to use “resource disappearance” as a wake-up condition.
• One read gets one message: rpmsg is a protocol with clear message boundaries,one skb = one rpmsg message。use = min_t(size_t, iov_iter_count(to), skb->len);This means that if the user buffer is smaller than the message, the excess is discarded (the skb is freed directly). This is a pitfall for beginners: the buffer must be large enough (the backend buffer is 512 bytes minus a 16-byte header = 496 bytes of payload).
• Usespin_lock_irqsaverather thanspin_lock: because the read may be in normal process context, but the lock may be held by the cb in interrupt context, so disabling interrupts is safer.

rpmsg_eptdev_write_iter()

static ssize_t rpmsg_eptdev_write_iter(struct kiocb *iocb,
				       struct iov_iter *from)
{
	struct file *filp = iocb->ki_filp;
	struct rpmsg_eptdev *eptdev = filp->private_data;
	size_t len = iov_iter_count(from);
	void *kbuf;
	int ret;

	kbuf = kzalloc(len, GFP_KERNEL);
	if (!kbuf)
		return -ENOMEM;

	if (!copy_from_iter_full(kbuf, len, from)) {
		ret = -EFAULT;
		goto free_kbuf;
	}

	if (mutex_lock_interruptible(&eptdev->ept_lock)) {
		ret = -ERESTARTSYS;
		goto free_kbuf;
	}

	if (!eptdev->ept) {
		ret = -EPIPE;
		goto unlock_eptdev;
	}

	if (filp->f_flags & O_NONBLOCK)
		ret = rpmsg_trysend(eptdev->ept, kbuf, len);
	else
		ret = rpmsg_send(eptdev->ept, kbuf, len);

unlock_eptdev:
	mutex_unlock(&eptdev->ept_lock);

free_kbuf:
	kfree(kbuf);
	return ret < 0 ? ret : len;
}

• firstcopy_from_iterthen take the lock again, becausethe userspace copy may page-fault and sleep, so it should not be done while holding ept_lock — otherwise it will block the destroy path for a long time.

• rpmsg_sendThe blocking version goes all the way to the backendrpmsg_send_offchannel_raw(..., wait=true): if there is no TX buffer,wait_event_interruptible_timeout(..., 15000)。rpmsg_trysendYeswait=false, if no buffer, immediately-ENOMEM。

• On success, return len (number of bytes written), not ret (the backend ret is 0). Note that a single message is limited by the backend buffer:len > buf_size - sizeof(hdr)that is, 496 bytes will-EMSGSIZE。

• Sending is also one write = one message; there is no streaming concatenation.

rpmsg_eptdev_poll()

static __poll_t rpmsg_eptdev_poll(struct file *filp, poll_table *wait)
{
	struct rpmsg_eptdev *eptdev = filp->private_data;
	__poll_t mask = 0;

	if (!eptdev->ept)
		return EPOLLERR;

	poll_wait(filp, &eptdev->readq, wait);

	if (!skb_queue_empty(&eptdev->queue))
		mask |= EPOLLIN | EPOLLRDNORM;

	mask |= rpmsg_poll(eptdev->ept, filp, wait);

	return mask;
}

rpmsg_eptdev_ioctl()

static long rpmsg_eptdev_ioctl(struct file *fp, unsigned int cmd,
			       unsigned long arg)
{
	struct rpmsg_eptdev *eptdev = fp->private_data;

	if (cmd != RPMSG_DESTROY_EPT_IOCTL)
		return -EINVAL;

	return rpmsg_eptdev_destroy(&eptdev->dev, NULL);
}

rpmsg_poll forwards to the backend’sops->poll— but note that the virtio backend’svirtio_endpoint_opsdoes not implement poll, sorpmsg_pollinif (!ept->ops->poll) return 0, i.e., under this backend, poll only reports readable, not writable. This is an optional backend implementation.

References

[1]

Asymmetric Multiprocessing and Embedded Linux

[2]

Using virtio to Talk With Remote Processors