Re: [PATCH] rpmsg: virtio: EPOLLOUT support

[Mathieu Poirier <mathieu.poirier@linaro.org>]

On Thu, Oct 23, 2025 at 08:59:22AM +1100, Patrick Oppenlander wrote:
> On Thu, 23 Oct 2025 at 00:35, Zhongqiu Han
> <zhongqiu.han@oss.qualcomm.com> wrote:
> >
> > On 10/22/2025 11:28 AM, patrick.oppenlander@gmail.com wrote:
> > > From: Patrick Oppenlander <patrick.oppenlander@gmail.com>
> > >
> > > 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 <patrick.oppenlander@gmail.com>
> > > ---
> > >   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.

Please add "Calling virtqueue ... in the queue." to the in-line comment you
already have.

I have pondered over this patch long and hard.  As Zhongqiu Han pointed out,
there is potential for unwanted side effect but as you pointed out, EPOLLOUT is
currently not supported.  We currently sit at rc3.  I will highlight your patch
in the next OpenAMP community meeting, asking people to test it in their
environment.  From there it could be added to my queue for 6.19-rc1, under the
reserve that if someone complains, it will be backed out and you will need to
find a different approach.

Thanks,
Mathieu   

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