Rpmsg Char Driver

---

Timeline

Timeline

2026-06-29

init

This article introduces the implementation mechanism of the Rpmsg Char Driver character device driver based on Linux 5.10.x. It discusses in detail its classic architecture design that separates control devices from endpoint devices, and summarizes the specific processes and key code logic of driver registration, deferred endpoint creation, lock and wait queue synchronization mechanisms, and user-space message sending and receiving.

---

The current code analysis is based on linux 5.10.x

rpmsg.drawio

rpmsg_char.cimplementsrpmsg_driverand registered to the Rpmsg Bus, providing an interface for user-space interaction, and taking overvirtio_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 following hierarchy:

rpmsg_char.drawio

Note that it usespostcore_initcall, which is moremodule_initearlier, because rpmsg_core.c's bus also usespostcore_initcall, and both are registered after the driver model is ready.

rpmsg_char_initdoes three things:

1. alloc_chrdev_region(&rpmsg_major, 0, RPMSG_DEV_MAX, "rpmsg") —— Dynamically allocate the 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_addit will automatically/dev/generate device nodes under (relying 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	Created by	Function
Control device	/dev/rpmsg_ctrl0	rpmsg_ctrldev	rpmsg_chrdev_probe	One per channel; users “create endpoints” on it via ioctl
Endpoint device	/dev/rpmsg0	rpmsg_eptdev	rpmsg_ctrldev_ioctl(CREATE)	One per endpoint; users read/write/poll it to send and receive messages

This separation of “control device + data device” is a very classic pattern in Linux:The control node is used to instantiate the data node, and the data node carries 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, attached 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 during open (name/src/dst)

    struct mutex ept_lock;               // Protects modification of the ept pointer
    struct rpmsg_endpoint *ept;          // Actual endpoint created only on open, 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
};

Locks and wait queues:

• struct mutex ept_lock;: Protects the ept pointer. ept becomes NULL in three cases:release、destroy ioctl, parent deviceremove. The read/write path must acquire this lock first to confirm the ept is still alive.
• spinlock_t queue_lock;: Protects the queue (skb linked list). The callbackrpmsg_ept_cbenqueues in interrupt/atomic context, read_iter dequeues in process context, so a spinlock is used here and in the cbspin_lock, and in readspin_lock_irqsave。
• wait_queue_head_t readq;：rpmsg_ept_cbAfter enqueuing,wake_up_interruptible，read_iterWhen the queue is empty,wait_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_driverit is 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 device number, fromrpmsg_ctrl_idaget id (used forrpmsg_ctrl%d）。
4. cdev_device_add—— register cdev and device together,/dev/rpmsg_ctrlNappears.
5. Key point:dev->release = rpmsg_ctrldev_release_deviceis in thecdev_device_addis assigned afterwards. The comment says “We can now rely on the release function for cleanup”——meaningcdev_device_addIf it fails before success, it goes to the manual error label for release; after success, it is handed over to the device model’s release callback for cleanup (kfree when the last reference drops to zero). This is a very standard refcount + release paradigm in the Linux device model.
6. dev_set_drvdata(&rpdev->dev, ctrldev)—— so that during remove, ctrldev can be retrieved 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;, therefore the actualreleasefunction 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);
};

create ept through this function, callingrpmsg_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 and chinfo.
2. Initialize lock, skb queue, and wait queue head.
3. device_initialize + setclass/parent/groups/drvdata。groups = rpmsg_eptdev_groupsprovide name/src/dst read-only attributes for sysfs
4. cdev_init usesrpmsg_eptdev_fops。
5. ida twice_simple_get to fetch minor and ept id,dev_set_name(dev, "rpmsg%d", ret) →/dev/rpmsgN。
6. cdev_device_add, set dev->release upon success.
7. Note: At this point,eptdev->ept == NULL! The real endpoint hasn’t been created yet.The endpoint is created only upon open. This is lazy creation to avoid holding an address without anyone using it.

The goto order in the error handling path is strictly the reverse of the resource acquisition order, which is the standard unwind style. Note the free_eptdev’s put_device followed by kfree—there’s actually a detail here:device_initializeAfterwards, one should use put_device to let the release callback free it, but because release is set only after cdev_device_add succeeds, it hasn’t been set in this path, so put_device won’t actually kfree, a manual kfree is required. 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.cin thecompat_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 the 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 allocate starting from 1024), binding the callbackrpmsg_ept_cbandpriv=eptdevto that address.
• After the callback is registered, messages sent from the remote with a dst equal to this local address will triggerrpmsg_ept_cb (see backendrpmsg_recv_singlelook up bymsg->dstin the idr and callept->cb）。
• get_device/ open failureput_device: ensuring the device is not released while the file is open.

And the ept’s callback 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 mutexes cannot be used here, cannot sleep, use GFP_ATOMIC and spinlocks.
• A memory copy is performed here (buf → skb). The backend receives the vring buffer via zero-copy, but to hand the data over to userspace and prevent the buffer from being held by the user for a long time (the backend needs to return the buffer to the rvq as soon as possible), a copy is made into the skb here.rpmsg_recv_singleImmediately after calling the 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;
}

Actually callsdev->release

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

• The waiting condition also checks!ept: This way,when destroy sets ept to NULL and calls wake_up, the blocked read can wake up immediately and return -EPIPE, instead of getting stuck forever. This is the standard way to write it, using “resource disappearance” as the wake-up condition.
• One read retrieves 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);which means if the user buffer is smaller than the message, the excess part 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 16 bytes header = 496 bytes payload).
• Usespin_lock_irqsaverather thanspin_lock: Because the read might be in a normal process context, but the lock might be held by a cb in an interrupt context, 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_iterAcquire the lock again, becausethe userspace copy might page fault and sleep, and should not be done while holding ept_lock — otherwise it would block the destroy path for a long time.

• rpmsg_sendThe blocking version will go all the way to the backendrpmsg_send_offchannel_raw(..., wait=true): if there is no TX buffer, it willwait_event_interruptible_timeout(..., 15000)。rpmsg_trysendiswait=false, if there is no buffer, it immediately-ENOMEM。

• successfully returns 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)meaning 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, meaning that under this backend, poll only reports readable, not writable. This is an optional backend implementation.

Reference materials

[1]

Asymmetric Multiprocessing and Embedded Linux

[2]

Using virtio to Talk With Remote Processors