On Thu, 23 Oct 2025 at 00:35, Zhongqiu Han
<[email protected]> wrote:
>
> On 10/22/2025 11:28 AM, [email protected] wrote:
> > From: Patrick Oppenlander <[email protected]>
> >
> > Previously, polling an rpmsg endpoint (e.g. /dev/ttyRPMSGx) would
> > generate EPOLLIN events but no EPOLLOUT events.
> >
> > Unfortunately, poll support means that we can no longer disable
> > tx-complete interrupts as there is no way to know whether a poller is
> > waiting in sendq, so we always need notifications.
> >
> > Signed-off-by: Patrick Oppenlander <[email protected]>
> > ---
> > drivers/rpmsg/virtio_rpmsg_bus.c | 101 ++++++++++---------------------
> > 1 file changed, 32 insertions(+), 69 deletions(-)
> >
> > diff --git a/drivers/rpmsg/virtio_rpmsg_bus.c
> > b/drivers/rpmsg/virtio_rpmsg_bus.c
> > index 484890b4a6a74..79d983055b4d6 100644
> > --- a/drivers/rpmsg/virtio_rpmsg_bus.c
> > +++ b/drivers/rpmsg/virtio_rpmsg_bus.c
> > @@ -41,13 +41,12 @@
> > * @buf_size: size of one rx or tx buffer
> > * @last_sbuf: index of last tx buffer used
> > * @bufs_dma: dma base addr of the buffers
> > - * @tx_lock: protects svq, sbufs and sleepers, to allow concurrent senders.
> > + * @tx_lock: protects svq and sbufs, to allow concurrent senders.
> > * sending a message might require waking up a dozing remote
> > * processor, which involves sleeping, hence the mutex.
> > * @endpoints: idr of local endpoints, allows fast retrieval
> > * @endpoints_lock: lock of the endpoints set
> > * @sendq: wait queue of sending contexts waiting for a tx buffers
> > - * @sleepers: number of senders that are waiting for a tx buffer
> > *
> > * This structure stores the rpmsg state of a given virtio remote
> > processor
> > * device (there might be several virtio proc devices for each physical
> > @@ -65,7 +64,6 @@ struct virtproc_info {
> > struct idr endpoints;
> > struct mutex endpoints_lock;
> > wait_queue_head_t sendq;
> > - atomic_t sleepers;
> > };
> >
> > /* The feature bitmap for virtio rpmsg */
> > @@ -144,6 +142,8 @@ static int virtio_rpmsg_sendto(struct rpmsg_endpoint
> > *ept, void *data, int len,
> > static int virtio_rpmsg_trysend(struct rpmsg_endpoint *ept, void *data,
> > int len);
> > static int virtio_rpmsg_trysendto(struct rpmsg_endpoint *ept, void *data,
> > int len, u32 dst);
> > +static __poll_t virtio_rpmsg_poll(struct rpmsg_endpoint *ept, struct file
> > *filp,
> > + poll_table *wait);
> > static ssize_t virtio_rpmsg_get_mtu(struct rpmsg_endpoint *ept);
> > static struct rpmsg_device *__rpmsg_create_channel(struct virtproc_info
> > *vrp,
> > struct rpmsg_channel_info
> > *chinfo);
> > @@ -154,6 +154,7 @@ static const struct rpmsg_endpoint_ops
> > virtio_endpoint_ops = {
> > .sendto = virtio_rpmsg_sendto,
> > .trysend = virtio_rpmsg_trysend,
> > .trysendto = virtio_rpmsg_trysendto,
> > + .poll = virtio_rpmsg_poll,
> > .get_mtu = virtio_rpmsg_get_mtu,
> > };
> >
> > @@ -436,7 +437,6 @@ static void *get_a_tx_buf(struct virtproc_info *vrp)
> > unsigned int len;
> > void *ret;
> >
> > - /* support multiple concurrent senders */
> > mutex_lock(&vrp->tx_lock);
> >
> > /*
> > @@ -454,62 +454,6 @@ static void *get_a_tx_buf(struct virtproc_info *vrp)
> > return ret;
> > }
> >
> > -/**
> > - * rpmsg_upref_sleepers() - enable "tx-complete" interrupts, if needed
> > - * @vrp: virtual remote processor state
> > - *
> > - * This function is called before a sender is blocked, waiting for
> > - * a tx buffer to become available.
> > - *
> > - * If we already have blocking senders, this function merely increases
> > - * the "sleepers" reference count, and exits.
> > - *
> > - * Otherwise, if this is the first sender to block, we also enable
> > - * virtio's tx callbacks, so we'd be immediately notified when a tx
> > - * buffer is consumed (we rely on virtio's tx callback in order
> > - * to wake up sleeping senders as soon as a tx buffer is used by the
> > - * remote processor).
> > - */
> > -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);
> > -}
> > -
> > -/**
> > - * rpmsg_downref_sleepers() - disable "tx-complete" interrupts, if needed
> > - * @vrp: virtual remote processor state
> > - *
> > - * This function is called after a sender, that waited for a tx buffer
> > - * to become available, is unblocked.
> > - *
> > - * If we still have blocking senders, this function merely decreases
> > - * the "sleepers" reference count, and exits.
> > - *
> > - * Otherwise, if there are no more blocking senders, we also disable
> > - * virtio's tx callbacks, to avoid the overhead incurred with handling
> > - * those (now redundant) interrupts.
> > - */
> > -static void rpmsg_downref_sleepers(struct virtproc_info *vrp)
> > -{
> > - /* support multiple concurrent senders */
> > - mutex_lock(&vrp->tx_lock);
> > -
> > - /* are we the last sleeping context waiting for tx buffers ? */
> > - if (atomic_dec_and_test(&vrp->sleepers))
> > - /* disable "tx-complete" interrupts */
> > - virtqueue_disable_cb(vrp->svq);
> > -
> > - mutex_unlock(&vrp->tx_lock);
> > -}
> > -
>
> Hi Patrick,
>
> I'd like to go over a few aspects of this patch, please feel free to
> correct me if there is any misunderstanding.
Hi, thanks for the review, I'll address your comments below.
> > /**
> > * rpmsg_send_offchannel_raw() - send a message across to the remote
> > processor
> > * @rpdev: the rpmsg channel
> > @@ -582,9 +526,6 @@ static int rpmsg_send_offchannel_raw(struct
> > rpmsg_device *rpdev,
> >
> > /* 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
> > @@ -595,9 +536,6 @@ static int rpmsg_send_offchannel_raw(struct
> > rpmsg_device *rpdev,
> > (msg = get_a_tx_buf(vrp)),
> > msecs_to_jiffies(15000));
> >
> > - /* disable "tx-complete" interrupts if we're the last sleeper
> > */
> > - rpmsg_downref_sleepers(vrp);
>
> 1.The original code dynamically disabled tx-complete interrupts during
> normal operation and only enabled them when waiting for buffers. This
> patch removes all virtqueue_disable_cb() calls, meaning interrupts are
> always enabled. For high-frequency messaging, could this significantly
> increase interrupt overhead, even irq storm? May I know do you have
> performance data showing the interrupt frequency increase?
I mentioned this in the commit message.
The problem is that the poller can be removed at any time by
poll_freewait, and there is no notification of this happening. The
simplest solution to this is what I have proposed in this patch, i.e.
always leave callbacks enabled.
There is an alternate solution, but it comes with a possibly
significant caveat. We could add a check along the lines of:
if wq_has_sleeper(vrp->sendq) {
virtqueue_disable_cb(vrp->svq);
}
But this is not guaranteed to work as expected as poll_freewait can
remove the poller at any time (i.e. after we've performed this check),
meaning that the callbacks will sometimes remain enabled regardless.
In terms of performance, I have not measured a difference in my workload.
> > -
> > /* timeout ? */
> > if (!err) {
> > dev_err(dev, "timeout waiting for a tx buffer\n");
> > @@ -676,6 +614,34 @@ static int virtio_rpmsg_trysendto(struct
> > rpmsg_endpoint *ept, void *data,
> > return rpmsg_send_offchannel_raw(rpdev, src, dst, data, len, false);
> > }
> >
> > +static __poll_t virtio_rpmsg_poll(struct rpmsg_endpoint *ept, struct file
> > *filp,
> > + poll_table *wait)
> > +{
> > + struct rpmsg_device *rpdev = ept->rpdev;
> > + struct virtio_rpmsg_channel *vch = to_virtio_rpmsg_channel(rpdev);
> > + struct virtproc_info *vrp = vch->vrp;
> > + __poll_t mask = 0;
> > +
> > + poll_wait(filp, &vrp->sendq, wait);
> > +
> > + /* support multiple concurrent senders */
> > + mutex_lock(&vrp->tx_lock);
> > +
> > + /*
> > + * check for a free buffer, either:
> > + * - we haven't used all of the available transmit buffers (half of
> > the
> > + * allocated buffers are used for transmit, hence num_bufs / 2), or,
> > + * - we ask the virtqueue if there's a buffer available
> > + */
> > + if (vrp->last_sbuf < vrp->num_bufs / 2 ||
> > + !virtqueue_enable_cb(vrp->svq))
> > + mask |= EPOLLOUT;
> > +
>
> 2.The virtio_rpmsg_poll() calls virtqueue_enable_cb() but seems never
> disables it. Once any process polls the device, interrupts remain
> permanently enabled even after poll completes?
virtio_rpmsg_poll returns immediately after registering the poller and
checking whether there are any free buffers. poll_wait does not block,
so there's no way to know that poll has completed.
With this change callbacks are always enabled (they're enabled by the
virtqueue initialisation if a callback is registered). Calling
virtqueue_enable_cb here doesn't actually enable callbacks. We call it
because its return value tells us whether there's any pending buffers
in the queue. I'm not aware of another virtqueue API which can perform
this check.
We could propose a new virtqueue_buf_pending() or something along
those lines if necessary, but, looking at the implementation of
virtqueue_enable_cb(), we'd effectively be causing a bunch of code
churn to avoid a couple of if tests. Given that we'd be the only
consumer of this new API it didn't seem worth it to me?
Kind regards,
Patrick
> > + mutex_unlock(&vrp->tx_lock);
> > +
> > + return mask;
> > +}
> > +
> > static ssize_t virtio_rpmsg_get_mtu(struct rpmsg_endpoint *ept)
> > {
> > struct rpmsg_device *rpdev = ept->rpdev;
> > @@ -922,9 +888,6 @@ static int rpmsg_probe(struct virtio_device *vdev)
> > WARN_ON(err); /* sanity check; this can't really happen */
> > }
> >
> > - /* suppress "tx-complete" interrupts */
> > - virtqueue_disable_cb(vrp->svq);
> > -
> > vdev->priv = vrp;
> >
> > rpdev_ctrl = rpmsg_virtio_add_ctrl_dev(vdev);
>
>
> --
> Thx and BRs,
> Zhongqiu Han
