[PATCH 1/2] virtio_ring: add virtqueue_enable_cb_delayed_max

[PATCH 1/2] virtio_ring: add virtqueue_enable_cb_delayed_max

[Michael S. Tsirkin]

virtqueue_enable_cb_delayed places the used event at 3/4 of the
currently queued buffers, so with VIRTIO_RING_F_EVENT_IDX the device
interrupts early and often - this is a good fit for drivers that poll
and process buffers as they trickle in, like virtio net.

Some drivers queue a batch of buffers and then sleep waiting for the
whole batch to complete.  For these, an interrupt before the last
buffer is used is pure waste.

To fix, add virtqueue_enable_cb_delayed_max: same API, but the used
event is placed at the last currently queued buffer, so the device
only notifies us once it has used everything.

Signed-off-by: Michael S. Tsirkin <mst@redhat.com>
---
 drivers/virtio/virtio_ring.c | 125 +++++++++++++++++++++++++++++++++++
 include/linux/virtio.h       |   2 +
 2 files changed, 127 insertions(+)

diff --git a/drivers/virtio/virtio_ring.c b/drivers/virtio/virtio_ring.c
index fbca7ce1c6bf..0428c0ae9a8a 100644
--- a/drivers/virtio/virtio_ring.c
+++ b/drivers/virtio/virtio_ring.c
@@ -180,6 +180,7 @@ struct virtqueue_ops {
 	bool (*kick_prepare)(struct vring_virtqueue *vq);
 	void (*disable_cb)(struct vring_virtqueue *vq);
 	bool (*enable_cb_delayed)(struct vring_virtqueue *vq);
+	bool (*enable_cb_delayed_max)(struct vring_virtqueue *vq);
 	unsigned int (*enable_cb_prepare)(struct vring_virtqueue *vq);
 	bool (*poll)(const struct vring_virtqueue *vq,
 		     unsigned int last_used_idx);
@@ -1122,6 +1123,42 @@ static bool virtqueue_enable_cb_delayed_split(struct vring_virtqueue *vq)
 	return true;
 }
 
+static bool virtqueue_enable_cb_delayed_max_split(struct vring_virtqueue *vq)
+{
+	u16 bufs;
+
+	START_USE(vq);
+
+	/* We optimistically turn back on interrupts, then check if there was
+	 * more to do. */
+	/* Depending on the VIRTIO_RING_F_USED_EVENT_IDX feature, we need to
+	 * either clear the flags bit or point the event index at the next
+	 * entry. Always update the event index to keep code simple. */
+	if (vq->split.avail_flags_shadow & VRING_AVAIL_F_NO_INTERRUPT) {
+		vq->split.avail_flags_shadow &= ~VRING_AVAIL_F_NO_INTERRUPT;
+		if (!vq->event)
+			vq->split.vring.avail->flags =
+				cpu_to_virtio16(vq->vq.vdev,
+						vq->split.avail_flags_shadow);
+	}
+	/* Point the used event at the last queued buffer: the device
+	 * will only notify us once it has used everything. */
+	bufs = (u16)(vq->split.avail_idx_shadow - vq->last_used_idx);
+
+	virtio_store_mb(vq->weak_barriers,
+			&vring_used_event(&vq->split.vring),
+			cpu_to_virtio16(vq->vq.vdev, vq->last_used_idx + bufs));
+
+	if (unlikely((u16)(virtio16_to_cpu(vq->vq.vdev, vq->split.vring.used->idx)
+						- vq->last_used_idx) > bufs)) {
+		END_USE(vq);
+		return false;
+	}
+
+	END_USE(vq);
+	return true;
+}
+
 static void *virtqueue_detach_unused_buf_split(struct vring_virtqueue *vq)
 {
 	unsigned int i;
@@ -2318,6 +2355,67 @@ static bool virtqueue_enable_cb_delayed_packed(struct vring_virtqueue *vq)
 	return true;
 }
 
+static bool virtqueue_enable_cb_delayed_max_packed(struct vring_virtqueue *vq)
+{
+	u16 used_idx, wrap_counter, last_used_idx;
+	u16 bufs;
+
+	START_USE(vq);
+
+	/*
+	 * We optimistically turn back on interrupts, then check if there was
+	 * more to do.
+	 */
+
+	if (vq->event) {
+		/* Point the event at the last queued buffer: the device
+		 * will only notify us once it has used everything. */
+		bufs = vq->packed.vring.num - vq->vq.num_free;
+		last_used_idx = READ_ONCE(vq->last_used_idx);
+		wrap_counter = packed_used_wrap_counter(last_used_idx);
+
+		used_idx = packed_last_used(last_used_idx) + bufs;
+		if (used_idx >= vq->packed.vring.num) {
+			used_idx -= vq->packed.vring.num;
+			wrap_counter ^= 1;
+		}
+
+		vq->packed.vring.driver->off_wrap = cpu_to_le16(used_idx |
+			(wrap_counter << VRING_PACKED_EVENT_F_WRAP_CTR));
+
+		/*
+		 * We need to update event offset and event wrap
+		 * counter first before updating event flags.
+		 */
+		virtio_wmb(vq->weak_barriers);
+	}
+
+	if (vq->packed.event_flags_shadow == VRING_PACKED_EVENT_FLAG_DISABLE) {
+		vq->packed.event_flags_shadow = vq->event ?
+				VRING_PACKED_EVENT_FLAG_DESC :
+				VRING_PACKED_EVENT_FLAG_ENABLE;
+		vq->packed.vring.driver->flags =
+				cpu_to_le16(vq->packed.event_flags_shadow);
+	}
+
+	/*
+	 * We need to update event suppression structure first
+	 * before re-checking for more used buffers.
+	 */
+	virtio_mb(vq->weak_barriers);
+
+	last_used_idx = READ_ONCE(vq->last_used_idx);
+	wrap_counter = packed_used_wrap_counter(last_used_idx);
+	used_idx = packed_last_used(last_used_idx);
+	if (is_used_desc_packed(vq, used_idx, wrap_counter)) {
+		END_USE(vq);
+		return false;
+	}
+
+	END_USE(vq);
+	return true;
+}
+
 static void *virtqueue_detach_unused_buf_packed(struct vring_virtqueue *vq)
 {
 	unsigned int i;
@@ -2640,6 +2738,7 @@ static const struct virtqueue_ops split_ops = {
 	.kick_prepare = virtqueue_kick_prepare_split,
 	.disable_cb = virtqueue_disable_cb_split,
 	.enable_cb_delayed = virtqueue_enable_cb_delayed_split,
+	.enable_cb_delayed_max = virtqueue_enable_cb_delayed_max_split,
 	.enable_cb_prepare = virtqueue_enable_cb_prepare_split,
 	.poll = virtqueue_poll_split,
 	.detach_unused_buf = virtqueue_detach_unused_buf_split,
@@ -2654,6 +2753,7 @@ static const struct virtqueue_ops packed_ops = {
 	.kick_prepare = virtqueue_kick_prepare_packed,
 	.disable_cb = virtqueue_disable_cb_packed,
 	.enable_cb_delayed = virtqueue_enable_cb_delayed_packed,
+	.enable_cb_delayed_max = virtqueue_enable_cb_delayed_max_packed,
 	.enable_cb_prepare = virtqueue_enable_cb_prepare_packed,
 	.poll = virtqueue_poll_packed,
 	.detach_unused_buf = virtqueue_detach_unused_buf_packed,
@@ -2668,6 +2768,7 @@ static const struct virtqueue_ops split_in_order_ops = {
 	.kick_prepare = virtqueue_kick_prepare_split,
 	.disable_cb = virtqueue_disable_cb_split,
 	.enable_cb_delayed = virtqueue_enable_cb_delayed_split,
+	.enable_cb_delayed_max = virtqueue_enable_cb_delayed_max_split,
 	.enable_cb_prepare = virtqueue_enable_cb_prepare_split,
 	.poll = virtqueue_poll_split,
 	.detach_unused_buf = virtqueue_detach_unused_buf_split,
@@ -2682,6 +2783,7 @@ static const struct virtqueue_ops packed_in_order_ops = {
 	.kick_prepare = virtqueue_kick_prepare_packed,
 	.disable_cb = virtqueue_disable_cb_packed,
 	.enable_cb_delayed = virtqueue_enable_cb_delayed_packed,
+	.enable_cb_delayed_max = virtqueue_enable_cb_delayed_max_packed,
 	.enable_cb_prepare = virtqueue_enable_cb_prepare_packed,
 	.poll = virtqueue_poll_packed,
 	.detach_unused_buf = virtqueue_detach_unused_buf_packed,
@@ -3197,6 +3299,29 @@ bool virtqueue_enable_cb_delayed(struct virtqueue *_vq)
 }
 EXPORT_SYMBOL_GPL(virtqueue_enable_cb_delayed);
 
+/**
+ * virtqueue_enable_cb_delayed_max - callback for all queued buffers
+ * @_vq: the struct virtqueue we're talking about.
+ *
+ * Like virtqueue_enable_cb_delayed, but with VIRTIO_RING_F_EVENT_IDX
+ * the used event is placed at the last currently queued buffer, so the
+ * device only notifies us once it has used everything we queued.
+ *
+ * Returns "true" if the callback is armed and we can sleep waiting
+ * for the notification, or "false" if some buffers were already used
+ * and we should poll instead.
+ */
+bool virtqueue_enable_cb_delayed_max(struct virtqueue *_vq)
+{
+	struct vring_virtqueue *vq = to_vvq(_vq);
+
+	if (vq->event_triggered)
+		data_race(vq->event_triggered = false);
+
+	return VIRTQUEUE_CALL(vq, enable_cb_delayed_max);
+}
+EXPORT_SYMBOL_GPL(virtqueue_enable_cb_delayed_max);
+
 /**
  * virtqueue_detach_unused_buf - detach first unused buffer
  * @_vq: the struct virtqueue we're talking about.
diff --git a/include/linux/virtio.h b/include/linux/virtio.h
index 3bbc4cb6a672..214fa36b8eed 100644
--- a/include/linux/virtio.h
+++ b/include/linux/virtio.h
@@ -114,6 +114,8 @@ bool virtqueue_poll(struct virtqueue *vq, unsigned);
 
 bool virtqueue_enable_cb_delayed(struct virtqueue *vq);
 
+bool virtqueue_enable_cb_delayed_max(struct virtqueue *vq);
+
 void *virtqueue_detach_unused_buf(struct virtqueue *vq);
 
 unsigned int virtqueue_get_vring_size(const struct virtqueue *vq);
-- 
MST

[PATCH 2/2] virtio_balloon: submit reported pages as individual buffers

[Michael S. Tsirkin]

The virtio_balloon driver reports a scatter-gather list of free pages
to the VMM so it may discard unused pages.

For an arm64 kernel with 64 KiB pages, the default reporting and
allocation orders allow entries ranging from 2 MiB to 512 MiB. If the
report contains a sufficient number of large entries, we will end up
with a descriptor chain that represents over 4 GiB of memory. This
violates virtio spec, which says:
	Drivers MUST NOT add a descriptor chain longer than $2^{32}$ bytes in total;

Specifically, submitting $2^{32}$ or more breaks Rust-VMM.

Submit each reported page as a separate virtqueue buffer instead of one
buffer with an sg list of all pages.  This also avoids indirect
descriptor allocation (kmalloc in the reporting path).

To avoid multiple buffers causing interrupt storms, keep callbacks
disabled while we wait for an event.

Note: if the virtqueue is broken, wait_event on
virtqueue_get_buf hangs.  This is a pre-existing issue:
the old single-buffer code had the same hang.

Reported-by: Ruben Hakobyan <hruben@meta.com>
Signed-off-by: Michael S. Tsirkin <mst@redhat.com>
---

Lightly tested. Ruben, could you pls help test?

 drivers/virtio/virtio_balloon.c | 48 ++++++++++++++++++++++++---------
 1 file changed, 35 insertions(+), 13 deletions(-)

diff --git a/drivers/virtio/virtio_balloon.c b/drivers/virtio/virtio_balloon.c
index f6c2dff33f8a..363519fa890d 100644
--- a/drivers/virtio/virtio_balloon.c
+++ b/drivers/virtio/virtio_balloon.c
@@ -202,25 +202,47 @@ static int virtballoon_free_page_report(struct page_reporting_dev_info *pr_dev_i
 	struct virtio_balloon *vb =
 		container_of(pr_dev_info, struct virtio_balloon, pr_dev_info);
 	struct virtqueue *vq = vb->reporting_vq;
-	unsigned int unused, err;
+	unsigned int i, err = 0;
 
 	/* We should always be able to add these buffers to an empty queue. */
-	err = virtqueue_add_inbuf(vq, sg, nents, vb, GFP_NOWAIT);
+	for (i = 0; i < nents; i++) {
+		struct scatterlist one;
 
-	/*
-	 * In the extremely unlikely case that something has occurred and we
-	 * are able to trigger an error we will simply display a warning
-	 * and exit without actually processing the pages.
-	 */
-	if (WARN_ON_ONCE(err))
-		return err;
+		sg_init_table(&one, 1);
+		sg_set_page(&one, sg_page(&sg[i]), sg[i].length,
+			    sg[i].offset);
+		err = virtqueue_add_inbuf(vq, &one, 1, &sg[i], GFP_NOWAIT);
 
-	virtqueue_kick(vq);
+		/*
+		 * In the extremely unlikely case that something has occurred
+		 * and we are able to trigger an error we will simply display
+		 * a warning and exit without actually processing the pages.
+		 */
+		if (WARN_ON_ONCE(err)) {
+			nents = i;
+			break;
+		}
+	}
 
-	/* When host has read buffer, this completes via balloon_ack */
-	wait_event(vb->acked, virtqueue_get_buf(vq, &unused));
+	if (nents) {
+		virtqueue_kick(vq);
 
-	return 0;
+		/* When host has read buffer, this completes via balloon_ack.
+		 * We report pages as individual buffers, so arm the callback
+		 * to only interrupt once all the queued buffers have been
+		 * used: otherwise we would get one interrupt per page.
+		 */
+		for (i = 0; i < nents; i++) {
+			unsigned int unused;
+
+			virtqueue_enable_cb_delayed_max(vq);
+			wait_event(vb->acked,
+				   virtqueue_get_buf(vq, &unused));
+			virtqueue_disable_cb(vq);
+		}
+	}
+
+	return err;
 }
 
 static void set_page_pfns(struct virtio_balloon *vb,
-- 
MST

Re: [PATCH 2/2] virtio_balloon: submit reported pages as individual buffers

[sashiko-bot]

Thank you for your contribution! Sashiko AI review found 2 potential issue(s) to consider:
- [High] virtio: off-by-one in virtqueue_enable_cb_delayed_max() hangs page reporting
- [Medium] virtio_balloon: broken callback suppression in virtballoon_free_page_report()
--

commit e2548c4b5befc523db0e3106c2a724d4255859e6
Author: Michael S. Tsirkin <mst@redhat.com>

virtio_balloon: submit reported pages as individual buffers

This commit submits each reported page in virtballoon_free_page_report() as a
separate virtqueue buffer instead of a single scatter-gather descriptor chain
to avoid exceeding the virtio specification's 2^32 byte limit. It uses
virtqueue_enable_cb_delayed_max() and virtqueue_disable_cb() around
wait_event() when draining the buffers.

> diff --git a/drivers/virtio/virtio_balloon.c b/drivers/virtio/virtio_balloon.c
> index ab3e3d887e00f..64999f05f9f1d 100644
> --- a/drivers/virtio/virtio_balloon.c
> +++ b/drivers/virtio/virtio_balloon.c
> @@ -205,25 +205,47 @@ static int virtballoon_free_page_report(struct page_reporting_dev_info *pr_dev_i

[ ... ]

> -	/* When host has read buffer, this completes via balloon_ack */
> -	wait_event(vb->acked, virtqueue_get_buf(vq, &unused));
> +	if (nents) {
> +		virtqueue_kick(vq);
>  
> -	return 0;
> +		/* When host has read buffer, this completes via balloon_ack.
> +		 * We report pages as individual buffers, so arm the callback
> +		 * to only interrupt once all the queued buffers have been
> +		 * used: otherwise we would get one interrupt per page.
> +		 */
> +		for (i = 0; i < nents; i++) {
> +			unsigned int unused;
> +
> +			virtqueue_enable_cb_delayed_max(vq);

[Severity: High]
Can calling virtqueue_enable_cb_delayed_max(vq) here cause wait_event() to
hang indefinitely when VIRTIO_RING_F_EVENT_IDX is negotiated?

In drivers/virtio/virtio_ring.c, virtqueue_enable_cb_delayed_max_split()
computes the event index as:

    bufs = (u16)(vq->split.avail_idx_shadow - vq->last_used_idx);

    virtio_store_mb(vq->weak_barriers,
                    &vring_used_event(&vq->split.vring),
                    cpu_to_virtio16(vq->vq.vdev, vq->last_used_idx + bufs));

    if (unlikely((u16)(virtio16_to_cpu(vq->vq.vdev, vq->split.vring.used->idx)
                                    - vq->last_used_idx) > bufs)) {
            END_USE(vq);
            return false;
    }

In include/uapi/linux/virtio_ring.h, vring_need_event() checks:

    static inline int vring_need_event(__u16 event_idx, __u16 new_idx, __u16 old)
    {
            return (__u16)(new_idx - event_idx - 1) < (__u16)(new_idx - old);
    }

Because event_idx is 0-based, an interrupt is triggered when used->idx
(new_idx) advances to event_idx + 1. When bufs buffers are queued at ring
slots last_used_idx .. last_used_idx + bufs - 1, consuming all queued buffers
advances used->idx to last_used_idx + bufs.

By setting vring_used_event to last_used_idx + bufs instead of
last_used_idx + bufs - 1, when the device finishes consuming all queued
buffers (new_idx == event_idx), (__u16)(new_idx - event_idx - 1) underflows
to 0xffff and vring_need_event() returns false. In addition, the race check
tests (used->idx - last_used_idx) > bufs, which can never be true since the
device cannot consume more than the bufs available buffers.

Similarly, in virtqueue_enable_cb_delayed_max_packed():

    bufs = vq->packed.vring.num - vq->vq.num_free;
    last_used_idx = READ_ONCE(vq->last_used_idx);
    wrap_counter = packed_used_wrap_counter(last_used_idx);

    used_idx = packed_last_used(last_used_idx) + bufs;
    if (used_idx >= vq->packed.vring.num) {
            used_idx -= vq->packed.vring.num;
            wrap_counter ^= 1;
    }

    vq->packed.vring.driver->off_wrap = cpu_to_le16(used_idx |
            (wrap_counter << VRING_PACKED_EVENT_F_WRAP_CTR));

This sets driver->off_wrap to packed_last_used(last_used_idx) + bufs
(next_avail_idx, one descriptor past the last queued descriptor) instead of
the last queued descriptor. Since that descriptor is not available to the
device, the device never marks it used and never interrupts, hanging the
page_reporting_process() -> page_reporting_cycle() ->
virtballoon_free_page_report() worker in wait_event().

> +			wait_event(vb->acked,
> +				   virtqueue_get_buf(vq, &unused));
> +			virtqueue_disable_cb(vq);
> +		}
> +	}

[Severity: Medium]
Does ordering virtqueue_enable_cb_delayed_max(), virtqueue_get_buf(), and
virtqueue_disable_cb() this way defeat the delayed callback suppression?

There appear to be three issues with the callback state machine in
virtballoon_free_page_report():

1. Reporting virtqueue callbacks are enabled by default in init_vqs(), and
virtqueue_kick(vq) is called before callbacks are disabled or delayed, so the
device is notified while interrupts are still armed for the first buffer.

2. Inside the loop, virtqueue_enable_cb_delayed_max(vq) is called before
wait_event(vb->acked, virtqueue_get_buf(vq, &unused)). Because callbacks are
enabled when virtqueue_get_buf() runs, virtqueue_get_buf_ctx_split() (and
update_last_used_idx_packed() for packed rings) overwrites the delayed event
threshold on every dequeued buffer with the very next single buffer:

    vq->last_used_idx++;
    /* If we expect an interrupt for the next entry, tell host
     * by writing event index and flush out the write before
     * the read in the next get_buf call. */
    if (!(vq->split.avail_flags_shadow & VRING_AVAIL_F_NO_INTERRUPT))
            virtio_store_mb(vq->weak_barriers,
                            &vring_used_event(&vq->split.vring),
                            cpu_to_virtio16(vq->vq.vdev, vq->last_used_idx));

3. When an interrupt fires, vring_interrupt() sets vq->event_triggered = true.
After wait_event() wakes and virtqueue_get_buf() writes
vring_used_event = vq->last_used_idx, the subsequent virtqueue_disable_cb(vq)
call hits the early return in virtqueue_disable_cb_split() (and similarly in
virtqueue_disable_cb_packed()):

    if (!(vq->split.avail_flags_shadow & VRING_AVAIL_F_NO_INTERRUPT)) {
            vq->split.avail_flags_shadow |= VRING_AVAIL_F_NO_INTERRUPT;

            /*
             * If device triggered an event already it won't trigger one again:
             * no need to disable.
             */
            if (vq->event_triggered)
                    return;

            if (vq->event)
                    /* TODO: this is a hack. Figure out a cleaner value to write. */
                    vring_used_event(&vq->split.vring) = 0x0;

This leaves vring_used_event armed at vq->last_used_idx in the shared ring at
the end of the batch, so the next call to virtballoon_free_page_report()
kicks the virtqueue with interrupts armed for the first buffer.

Should callbacks be kept disabled while draining available buffers with
virtqueue_get_buf(), only arming virtqueue_enable_cb_delayed_max() when
virtqueue_get_buf() returns NULL?

-- 
Sashiko AI review · https://sashiko.dev/#/patchset/ad6a530f49c4a0e182112a73789e31b5e91f3bb7.1793026707.git.mst@redhat.com?part=2

Re: [PATCH 1/2] virtio_ring: add virtqueue_enable_cb_delayed_max

[sashiko-bot]

Thank you for your contribution! Sashiko AI review found 4 potential issue(s) to consider:
- [High] virtio_ring: off-by-one in virtqueue_enable_cb_delayed_max_split()
- [High] virtio_ring: off-by-one in virtqueue_enable_cb_delayed_max_packed()
- [High] virtio_ring: missing vq->broken check in virtqueue_enable_cb_delayed_max()
- [Medium] virtio_ring: missing READ_ONCE() in virtqueue_enable_cb_delayed_max_split()
--

commit f77f0cd19bbfba432c2d11a8e7e8a7c7597a610a
Author: Michael S. Tsirkin <mst@redhat.com>

virtio_ring: add virtqueue_enable_cb_delayed_max

This commit introduces virtqueue_enable_cb_delayed_max(), which arms the
virtqueue callback with the used event placed at the last currently queued
buffer so that the device only notifies the driver once all queued buffers
have been used.

> diff --git a/drivers/virtio/virtio_ring.c b/drivers/virtio/virtio_ring.c
> index db678f5a80e03..98b630fc6f8f6 100644
> --- a/drivers/virtio/virtio_ring.c
> +++ b/drivers/virtio/virtio_ring.c
[ ... ]
> @@ -1165,6 +1166,42 @@ static bool virtqueue_enable_cb_delayed_split(struct vring_virtqueue *vq)
>  	return true;
>  }
>  
> +static bool virtqueue_enable_cb_delayed_max_split(struct vring_virtqueue *vq)
> +{
> +	u16 bufs;
> +
> +	START_USE(vq);
> +
> +	/* We optimistically turn back on interrupts, then check if there was
> +	 * more to do. */
> +	/* Depending on the VIRTIO_RING_F_USED_EVENT_IDX feature, we need to
> +	 * either clear the flags bit or point the event index at the next
> +	 * entry. Always update the event index to keep code simple. */
> +	if (vq->split.avail_flags_shadow & VRING_AVAIL_F_NO_INTERRUPT) {
> +		vq->split.avail_flags_shadow &= ~VRING_AVAIL_F_NO_INTERRUPT;
> +		if (!vq->event)
> +			vq->split.vring.avail->flags =
> +				cpu_to_virtio16(vq->vq.vdev,
> +						vq->split.avail_flags_shadow);
> +	}
> +	/* Point the used event at the last queued buffer: the device
> +	 * will only notify us once it has used everything. */
> +	bufs = (u16)(vq->split.avail_idx_shadow - vq->last_used_idx);
> +
> +	virtio_store_mb(vq->weak_barriers,
> +			&vring_used_event(&vq->split.vring),
> +			cpu_to_virtio16(vq->vq.vdev, vq->last_used_idx + bufs));
> +
> +	if (unlikely((u16)(virtio16_to_cpu(vq->vq.vdev, vq->split.vring.used->idx)
> +						- vq->last_used_idx) > bufs)) {
> +		END_USE(vq);
> +		return false;
> +	}

[Severity: High]
Is there an off-by-one in how bufs and vring_used_event are computed in
virtqueue_enable_cb_delayed_max_split()?

Here bufs is set to (u16)(vq->split.avail_idx_shadow - vq->last_used_idx),
so vq->last_used_idx + bufs equals vq->split.avail_idx_shadow. Per the
virtio specification and vring_need_event():

include/uapi/linux/virtio_ring.h:vring_need_event() {
	return (__u16)(new_idx - event_idx - 1) < (__u16)(new_idx - old);
}

event_idx is 0-indexed and triggers a notification when used->idx (new_idx)
transitions from event_idx to event_idx + 1 (for example,
virtqueue_enable_cb_prepare_split() writes vq->last_used_idx + 0 to be
notified when the first buffer is used and used->idx reaches
vq->last_used_idx + 1). When the device consumes all queued buffers,
used->idx advances to vq->split.avail_idx_shadow. With event_idx also set
to vq->split.avail_idx_shadow, (u16)(new_idx - event_idx - 1) underflows to
0xffff, so vring_need_event() evaluates to false and the device never sends
an interrupt when the batch finishes.

In addition, in the post-barrier race check:

	if (unlikely((u16)(virtio16_to_cpu(vq->vq.vdev, vq->split.vring.used->idx)
						- vq->last_used_idx) > bufs)) {

because only bufs (avail_idx_shadow - last_used_idx) buffers have been made
available to the device in total, (u16)(used->idx - vq->last_used_idx) can
at most equal bufs when every queued buffer has been used and can never be
strictly greater than bufs. Even if the device already consumed all queued
buffers before vring_used_event or avail->flags was updated, this check
always evaluates to false and virtqueue_enable_cb_delayed_max_split()
returns true, causing callers to sleep forever waiting for a notification
that will never arrive.

[Severity: Medium]
Also in virtqueue_enable_cb_delayed_max_split(), should the read of
vq->split.vring.used->idx use vring_read_split_used_idx(vq) instead of
reading vq->split.vring.used->idx directly?

Commit 32fe1de5c124 ("virtio_ring: Add READ_ONCE annotations for
device-writable fields") introduced vring_read_split_used_idx(vq) to wrap
the lockless read of device-writable vq->split.vring.used->idx in
READ_ONCE(), which is what virtqueue_enable_cb_delayed_split() uses:

drivers/virtio/virtio_ring.c:virtqueue_enable_cb_delayed_split() {
    ...
	if (unlikely((u16)(vring_read_split_used_idx(vq)
					- vq->last_used_idx) > bufs)) {
    ...
}

> +
> +	END_USE(vq);
> +	return true;
> +}
[ ... ]
> @@ -2391,6 +2428,67 @@ static bool virtqueue_enable_cb_delayed_packed(struct vring_virtqueue *vq)
>  	return true;
>  }
>  
> +static bool virtqueue_enable_cb_delayed_max_packed(struct vring_virtqueue *vq)
> +{
> +	u16 used_idx, wrap_counter, last_used_idx;
> +	u16 bufs;
> +
> +	START_USE(vq);
> +
> +	/*
> +	 * We optimistically turn back on interrupts, then check if there was
> +	 * more to do.
> +	 */
> +
> +	if (vq->event) {
> +		/* Point the event at the last queued buffer: the device
> +		 * will only notify us once it has used everything. */
> +		bufs = vq->packed.vring.num - vq->vq.num_free;
> +		last_used_idx = READ_ONCE(vq->last_used_idx);
> +		wrap_counter = packed_used_wrap_counter(last_used_idx);
> +
> +		used_idx = packed_last_used(last_used_idx) + bufs;
> +		if (used_idx >= vq->packed.vring.num) {
> +			used_idx -= vq->packed.vring.num;
> +			wrap_counter ^= 1;
> +		}
> +
> +		vq->packed.vring.driver->off_wrap = cpu_to_le16(used_idx |
> +			(wrap_counter << VRING_PACKED_EVENT_F_WRAP_CTR));
> +
> +		/*
> +		 * We need to update event offset and event wrap
> +		 * counter first before updating event flags.
> +		 */
> +		virtio_wmb(vq->weak_barriers);
> +	}
> +
> +	if (vq->packed.event_flags_shadow == VRING_PACKED_EVENT_FLAG_DISABLE) {
> +		vq->packed.event_flags_shadow = vq->event ?
> +				VRING_PACKED_EVENT_FLAG_DESC :
> +				VRING_PACKED_EVENT_FLAG_ENABLE;
> +		vq->packed.vring.driver->flags =
> +				cpu_to_le16(vq->packed.event_flags_shadow);
> +	}
> +
> +	/*
> +	 * We need to update event suppression structure first
> +	 * before re-checking for more used buffers.
> +	 */
> +	virtio_mb(vq->weak_barriers);
> +
> +	last_used_idx = READ_ONCE(vq->last_used_idx);
> +	wrap_counter = packed_used_wrap_counter(last_used_idx);
> +	used_idx = packed_last_used(last_used_idx);
> +	if (is_used_desc_packed(vq, used_idx, wrap_counter)) {
> +		END_USE(vq);
> +		return false;
> +	}

[Severity: High]
Won't virtqueue_enable_cb_delayed_max_packed() also place driver->off_wrap
one past the last queued descriptor and miss the completion interrupt?

When bufs = vq->packed.vring.num - vq->vq.num_free descriptors are queued,
they occupy offsets 0 through bufs - 1 starting from
packed_last_used(last_used_idx). Adding bufs to
packed_last_used(last_used_idx) sets used_idx to vq->packed.next_avail_idx
(with vq->packed.avail_wrap_counter), which is the next free descriptor
slot that has not been made available to the device.

With VRING_PACKED_EVENT_FLAG_DESC, the device notifies the driver when it
marks the descriptor at driver->off_wrap as used (for comparison,
virtqueue_enable_cb_prepare_packed() sets driver->off_wrap to
vq->last_used_idx, i.e. offset 0, to trigger when the first queued
descriptor is used). Because no descriptor is queued at next_avail_idx (and
if the last queued buffer is a multi-descriptor direct chain, the device
writes the used completion to the head descriptor of that chain), the
device stops at next_avail_idx without ever using the descriptor at
driver->off_wrap.

When the post-barrier check then tests is_used_desc_packed() at
packed_last_used(last_used_idx) and sees the first queued buffer is not yet
used, virtqueue_enable_cb_delayed_max_packed() returns true and the caller
sleeps forever waiting for an interrupt at next_avail_idx that never
arrives.

> +
> +	END_USE(vq);
> +	return true;
> +}
[ ... ]
> @@ -3278,6 +3380,29 @@ bool virtqueue_enable_cb_delayed(struct virtqueue *_vq)
>  }
>  EXPORT_SYMBOL_GPL(virtqueue_enable_cb_delayed);
>  
> +/**
> + * virtqueue_enable_cb_delayed_max - callback for all queued buffers
> + * @_vq: the struct virtqueue we're talking about.
> + *
> + * Like virtqueue_enable_cb_delayed, but with VIRTIO_RING_F_EVENT_IDX
> + * the used event is placed at the last currently queued buffer, so the
> + * device only notifies us once it has used everything we queued.
> + *
> + * Returns "true" if the callback is armed and we can sleep waiting
> + * for the notification, or "false" if some buffers were already used
> + * and we should poll instead.
> + */
> +bool virtqueue_enable_cb_delayed_max(struct virtqueue *_vq)
> +{
> +	struct vring_virtqueue *vq = to_vvq(_vq);
> +
> +	if (vq->event_triggered)
> +		data_race(vq->event_triggered = false);
> +
> +	return VIRTQUEUE_CALL(vq, enable_cb_delayed_max);
> +}

[Severity: High]
Should virtqueue_enable_cb_delayed_max() check if (unlikely(vq->broken))
and return true, matching virtqueue_enable_cb_delayed()?

Commit 0d0eff39ceb3 ("virtio_ring: fix infinite loop in
virtnet_poll_cleantx when device is broken") added:

drivers/virtio/virtio_ring.c:virtqueue_enable_cb_delayed() {
	struct vring_virtqueue *vq = to_vvq(_vq);

	if (unlikely(vq->broken))
		return true;
    ...
}

because virtqueue_get_buf_ctx_split() and virtqueue_get_buf_ctx_packed()
immediately return NULL when vq->broken is true without detaching buffers
or advancing vq->last_used_idx. Without this check in
virtqueue_enable_cb_delayed_max(), if vq->broken becomes true while used
descriptors remain in the ring, a caller polling until
virtqueue_enable_cb_delayed_max() returns true will spin forever because
virtqueue_get_buf() returns NULL without advancing vq->last_used_idx while
virtqueue_enable_cb_delayed_max() repeatedly returns false.

-- 
Sashiko AI review · https://sashiko.dev/#/patchset/ad6a530f49c4a0e182112a73789e31b5e91f3bb7.1793026707.git.mst@redhat.com?part=1

Re: [PATCH 1/2] virtio_ring: add virtqueue_enable_cb_delayed_max

[Michael S. Tsirkin]

On Mon, Oct 26, 2026 at 03:24:23PM +0000, sashiko-bot@kernel.org wrote:
> Thank you for your contribution! Sashiko AI review found 4 potential issue(s) to consider:
> - [High] virtio_ring: off-by-one in virtqueue_enable_cb_delayed_max_split()
> - [High] virtio_ring: off-by-one in virtqueue_enable_cb_delayed_max_packed()
> - [High] virtio_ring: missing vq->broken check in virtqueue_enable_cb_delayed_max()
> - [Medium] virtio_ring: missing READ_ONCE() in virtqueue_enable_cb_delayed_max_split()
> --
> 
> commit f77f0cd19bbfba432c2d11a8e7e8a7c7597a610a
> Author: Michael S. Tsirkin <mst@redhat.com>
> 
> virtio_ring: add virtqueue_enable_cb_delayed_max
> 
> This commit introduces virtqueue_enable_cb_delayed_max(), which arms the
> virtqueue callback with the used event placed at the last currently queued
> buffer so that the device only notifies the driver once all queued buffers
> have been used.
> 
> > diff --git a/drivers/virtio/virtio_ring.c b/drivers/virtio/virtio_ring.c
> > index db678f5a80e03..98b630fc6f8f6 100644
> > --- a/drivers/virtio/virtio_ring.c
> > +++ b/drivers/virtio/virtio_ring.c
> [ ... ]
> > @@ -1165,6 +1166,42 @@ static bool virtqueue_enable_cb_delayed_split(struct vring_virtqueue *vq)
> >  	return true;
> >  }
> >  
> > +static bool virtqueue_enable_cb_delayed_max_split(struct vring_virtqueue *vq)
> > +{
> > +	u16 bufs;
> > +
> > +	START_USE(vq);
> > +
> > +	/* We optimistically turn back on interrupts, then check if there was
> > +	 * more to do. */
> > +	/* Depending on the VIRTIO_RING_F_USED_EVENT_IDX feature, we need to
> > +	 * either clear the flags bit or point the event index at the next
> > +	 * entry. Always update the event index to keep code simple. */
> > +	if (vq->split.avail_flags_shadow & VRING_AVAIL_F_NO_INTERRUPT) {
> > +		vq->split.avail_flags_shadow &= ~VRING_AVAIL_F_NO_INTERRUPT;
> > +		if (!vq->event)
> > +			vq->split.vring.avail->flags =
> > +				cpu_to_virtio16(vq->vq.vdev,
> > +						vq->split.avail_flags_shadow);
> > +	}
> > +	/* Point the used event at the last queued buffer: the device
> > +	 * will only notify us once it has used everything. */
> > +	bufs = (u16)(vq->split.avail_idx_shadow - vq->last_used_idx);
> > +
> > +	virtio_store_mb(vq->weak_barriers,
> > +			&vring_used_event(&vq->split.vring),
> > +			cpu_to_virtio16(vq->vq.vdev, vq->last_used_idx + bufs));
> > +
> > +	if (unlikely((u16)(virtio16_to_cpu(vq->vq.vdev, vq->split.vring.used->idx)
> > +						- vq->last_used_idx) > bufs)) {
> > +		END_USE(vq);
> > +		return false;
> > +	}
> 
> [Severity: High]
> Is there an off-by-one in how bufs and vring_used_event are computed in
> virtqueue_enable_cb_delayed_max_split()?

Ugh. I actually fixed it, but forgot to commit (
Will fix in v2.


> 
> Here bufs is set to (u16)(vq->split.avail_idx_shadow - vq->last_used_idx),
> so vq->last_used_idx + bufs equals vq->split.avail_idx_shadow. Per the
> virtio specification and vring_need_event():
> 
> include/uapi/linux/virtio_ring.h:vring_need_event() {
> 	return (__u16)(new_idx - event_idx - 1) < (__u16)(new_idx - old);
> }
> 
> event_idx is 0-indexed and triggers a notification when used->idx (new_idx)
> transitions from event_idx to event_idx + 1 (for example,
> virtqueue_enable_cb_prepare_split() writes vq->last_used_idx + 0 to be
> notified when the first buffer is used and used->idx reaches
> vq->last_used_idx + 1). When the device consumes all queued buffers,
> used->idx advances to vq->split.avail_idx_shadow. With event_idx also set
> to vq->split.avail_idx_shadow, (u16)(new_idx - event_idx - 1) underflows to
> 0xffff, so vring_need_event() evaluates to false and the device never sends
> an interrupt when the batch finishes.
> 
> In addition, in the post-barrier race check:
> 
> 	if (unlikely((u16)(virtio16_to_cpu(vq->vq.vdev, vq->split.vring.used->idx)
> 						- vq->last_used_idx) > bufs)) {
> 
> because only bufs (avail_idx_shadow - last_used_idx) buffers have been made
> available to the device in total, (u16)(used->idx - vq->last_used_idx) can
> at most equal bufs when every queued buffer has been used and can never be
> strictly greater than bufs. Even if the device already consumed all queued
> buffers before vring_used_event or avail->flags was updated, this check
> always evaluates to false and virtqueue_enable_cb_delayed_max_split()
> returns true, causing callers to sleep forever waiting for a notification
> that will never arrive.
> 
> [Severity: Medium]
> Also in virtqueue_enable_cb_delayed_max_split(), should the read of
> vq->split.vring.used->idx use vring_read_split_used_idx(vq) instead of
> reading vq->split.vring.used->idx directly?
> 
> Commit 32fe1de5c124 ("virtio_ring: Add READ_ONCE annotations for
> device-writable fields") introduced vring_read_split_used_idx(vq) to wrap
> the lockless read of device-writable vq->split.vring.used->idx in
> READ_ONCE(), which is what virtqueue_enable_cb_delayed_split() uses:
> 
> drivers/virtio/virtio_ring.c:virtqueue_enable_cb_delayed_split() {
>     ...
> 	if (unlikely((u16)(vring_read_split_used_idx(vq)
> 					- vq->last_used_idx) > bufs)) {
>     ...
> }
> 
> > +
> > +	END_USE(vq);
> > +	return true;
> > +}
> [ ... ]
> > @@ -2391,6 +2428,67 @@ static bool virtqueue_enable_cb_delayed_packed(struct vring_virtqueue *vq)
> >  	return true;
> >  }
> >  
> > +static bool virtqueue_enable_cb_delayed_max_packed(struct vring_virtqueue *vq)
> > +{
> > +	u16 used_idx, wrap_counter, last_used_idx;
> > +	u16 bufs;
> > +
> > +	START_USE(vq);
> > +
> > +	/*
> > +	 * We optimistically turn back on interrupts, then check if there was
> > +	 * more to do.
> > +	 */
> > +
> > +	if (vq->event) {
> > +		/* Point the event at the last queued buffer: the device
> > +		 * will only notify us once it has used everything. */
> > +		bufs = vq->packed.vring.num - vq->vq.num_free;
> > +		last_used_idx = READ_ONCE(vq->last_used_idx);
> > +		wrap_counter = packed_used_wrap_counter(last_used_idx);
> > +
> > +		used_idx = packed_last_used(last_used_idx) + bufs;
> > +		if (used_idx >= vq->packed.vring.num) {
> > +			used_idx -= vq->packed.vring.num;
> > +			wrap_counter ^= 1;
> > +		}
> > +
> > +		vq->packed.vring.driver->off_wrap = cpu_to_le16(used_idx |
> > +			(wrap_counter << VRING_PACKED_EVENT_F_WRAP_CTR));
> > +
> > +		/*
> > +		 * We need to update event offset and event wrap
> > +		 * counter first before updating event flags.
> > +		 */
> > +		virtio_wmb(vq->weak_barriers);
> > +	}
> > +
> > +	if (vq->packed.event_flags_shadow == VRING_PACKED_EVENT_FLAG_DISABLE) {
> > +		vq->packed.event_flags_shadow = vq->event ?
> > +				VRING_PACKED_EVENT_FLAG_DESC :
> > +				VRING_PACKED_EVENT_FLAG_ENABLE;
> > +		vq->packed.vring.driver->flags =
> > +				cpu_to_le16(vq->packed.event_flags_shadow);
> > +	}
> > +
> > +	/*
> > +	 * We need to update event suppression structure first
> > +	 * before re-checking for more used buffers.
> > +	 */
> > +	virtio_mb(vq->weak_barriers);
> > +
> > +	last_used_idx = READ_ONCE(vq->last_used_idx);
> > +	wrap_counter = packed_used_wrap_counter(last_used_idx);
> > +	used_idx = packed_last_used(last_used_idx);
> > +	if (is_used_desc_packed(vq, used_idx, wrap_counter)) {
> > +		END_USE(vq);
> > +		return false;
> > +	}
> 
> [Severity: High]
> Won't virtqueue_enable_cb_delayed_max_packed() also place driver->off_wrap
> one past the last queued descriptor and miss the completion interrupt?
> 
> When bufs = vq->packed.vring.num - vq->vq.num_free descriptors are queued,
> they occupy offsets 0 through bufs - 1 starting from
> packed_last_used(last_used_idx). Adding bufs to
> packed_last_used(last_used_idx) sets used_idx to vq->packed.next_avail_idx
> (with vq->packed.avail_wrap_counter), which is the next free descriptor
> slot that has not been made available to the device.
> 
> With VRING_PACKED_EVENT_FLAG_DESC, the device notifies the driver when it
> marks the descriptor at driver->off_wrap as used (for comparison,
> virtqueue_enable_cb_prepare_packed() sets driver->off_wrap to
> vq->last_used_idx, i.e. offset 0, to trigger when the first queued
> descriptor is used). Because no descriptor is queued at next_avail_idx (and
> if the last queued buffer is a multi-descriptor direct chain, the device
> writes the used completion to the head descriptor of that chain), the
> device stops at next_avail_idx without ever using the descriptor at
> driver->off_wrap.
> 
> When the post-barrier check then tests is_used_desc_packed() at
> packed_last_used(last_used_idx) and sees the first queued buffer is not yet
> used, virtqueue_enable_cb_delayed_max_packed() returns true and the caller
> sleeps forever waiting for an interrupt at next_avail_idx that never
> arrives.
> 
> > +
> > +	END_USE(vq);
> > +	return true;
> > +}
> [ ... ]
> > @@ -3278,6 +3380,29 @@ bool virtqueue_enable_cb_delayed(struct virtqueue *_vq)
> >  }
> >  EXPORT_SYMBOL_GPL(virtqueue_enable_cb_delayed);
> >  
> > +/**
> > + * virtqueue_enable_cb_delayed_max - callback for all queued buffers
> > + * @_vq: the struct virtqueue we're talking about.
> > + *
> > + * Like virtqueue_enable_cb_delayed, but with VIRTIO_RING_F_EVENT_IDX
> > + * the used event is placed at the last currently queued buffer, so the
> > + * device only notifies us once it has used everything we queued.
> > + *
> > + * Returns "true" if the callback is armed and we can sleep waiting
> > + * for the notification, or "false" if some buffers were already used
> > + * and we should poll instead.
> > + */
> > +bool virtqueue_enable_cb_delayed_max(struct virtqueue *_vq)
> > +{
> > +	struct vring_virtqueue *vq = to_vvq(_vq);
> > +
> > +	if (vq->event_triggered)
> > +		data_race(vq->event_triggered = false);
> > +
> > +	return VIRTQUEUE_CALL(vq, enable_cb_delayed_max);
> > +}
> 
> [Severity: High]
> Should virtqueue_enable_cb_delayed_max() check if (unlikely(vq->broken))
> and return true, matching virtqueue_enable_cb_delayed()?
> 
> Commit 0d0eff39ceb3 ("virtio_ring: fix infinite loop in
> virtnet_poll_cleantx when device is broken") added:
> 
> drivers/virtio/virtio_ring.c:virtqueue_enable_cb_delayed() {
> 	struct vring_virtqueue *vq = to_vvq(_vq);
> 
> 	if (unlikely(vq->broken))
> 		return true;
>     ...
> }
> 
> because virtqueue_get_buf_ctx_split() and virtqueue_get_buf_ctx_packed()
> immediately return NULL when vq->broken is true without detaching buffers
> or advancing vq->last_used_idx. Without this check in
> virtqueue_enable_cb_delayed_max(), if vq->broken becomes true while used
> descriptors remain in the ring, a caller polling until
> virtqueue_enable_cb_delayed_max() returns true will spin forever because
> virtqueue_get_buf() returns NULL without advancing vq->last_used_idx while
> virtqueue_enable_cb_delayed_max() repeatedly returns false.
> 
> -- 
> Sashiko AI review · https://sashiko.dev/#/patchset/ad6a530f49c4a0e182112a73789e31b5e91f3bb7.1793026707.git.mst@redhat.com?part=1

Re: [PATCH 2/2] virtio_balloon: submit reported pages as individual buffers

[Michael S. Tsirkin]

On Mon, Oct 26, 2026 at 11:09:59AM -0400, Michael S. Tsirkin wrote:
> The virtio_balloon driver reports a scatter-gather list of free pages
> to the VMM so it may discard unused pages.
> 
> For an arm64 kernel with 64 KiB pages, the default reporting and
> allocation orders allow entries ranging from 2 MiB to 512 MiB. If the
> report contains a sufficient number of large entries, we will end up
> with a descriptor chain that represents over 4 GiB of memory. This
> violates virtio spec, which says:
> 	Drivers MUST NOT add a descriptor chain longer than $2^{32}$ bytes in total;
> 
> Specifically, submitting $2^{32}$ or more breaks Rust-VMM.
> 
> Submit each reported page as a separate virtqueue buffer instead of one
> buffer with an sg list of all pages.  This also avoids indirect
> descriptor allocation (kmalloc in the reporting path).
> 
> To avoid multiple buffers causing interrupt storms, keep callbacks
> disabled while we wait for an event.
> 
> Note: if the virtqueue is broken, wait_event on
> virtqueue_get_buf hangs.  This is a pre-existing issue:
> the old single-buffer code had the same hang.
> 
> Reported-by: Ruben Hakobyan <hruben@meta.com>
> Signed-off-by: Michael S. Tsirkin <mst@redhat.com>

brown paperbag post,forgot to commit.

> ---
> 
> Lightly tested. Ruben, could you pls help test?
> 
>  drivers/virtio/virtio_balloon.c | 48 ++++++++++++++++++++++++---------
>  1 file changed, 35 insertions(+), 13 deletions(-)
> 
> diff --git a/drivers/virtio/virtio_balloon.c b/drivers/virtio/virtio_balloon.c
> index f6c2dff33f8a..363519fa890d 100644
> --- a/drivers/virtio/virtio_balloon.c
> +++ b/drivers/virtio/virtio_balloon.c
> @@ -202,25 +202,47 @@ static int virtballoon_free_page_report(struct page_reporting_dev_info *pr_dev_i
>  	struct virtio_balloon *vb =
>  		container_of(pr_dev_info, struct virtio_balloon, pr_dev_info);
>  	struct virtqueue *vq = vb->reporting_vq;
> -	unsigned int unused, err;
> +	unsigned int i, err = 0;
>  
>  	/* We should always be able to add these buffers to an empty queue. */
> -	err = virtqueue_add_inbuf(vq, sg, nents, vb, GFP_NOWAIT);
> +	for (i = 0; i < nents; i++) {
> +		struct scatterlist one;
>  
> -	/*
> -	 * In the extremely unlikely case that something has occurred and we
> -	 * are able to trigger an error we will simply display a warning
> -	 * and exit without actually processing the pages.
> -	 */
> -	if (WARN_ON_ONCE(err))
> -		return err;
> +		sg_init_table(&one, 1);
> +		sg_set_page(&one, sg_page(&sg[i]), sg[i].length,
> +			    sg[i].offset);
> +		err = virtqueue_add_inbuf(vq, &one, 1, &sg[i], GFP_NOWAIT);
>  
> -	virtqueue_kick(vq);
> +		/*
> +		 * In the extremely unlikely case that something has occurred
> +		 * and we are able to trigger an error we will simply display
> +		 * a warning and exit without actually processing the pages.
> +		 */
> +		if (WARN_ON_ONCE(err)) {
> +			nents = i;
> +			break;
> +		}
> +	}
>  
> -	/* When host has read buffer, this completes via balloon_ack */
> -	wait_event(vb->acked, virtqueue_get_buf(vq, &unused));
> +	if (nents) {
> +		virtqueue_kick(vq);
>  
> -	return 0;
> +		/* When host has read buffer, this completes via balloon_ack.
> +		 * We report pages as individual buffers, so arm the callback
> +		 * to only interrupt once all the queued buffers have been
> +		 * used: otherwise we would get one interrupt per page.
> +		 */
> +		for (i = 0; i < nents; i++) {
> +			unsigned int unused;
> +
> +			virtqueue_enable_cb_delayed_max(vq);
> +			wait_event(vb->acked,
> +				   virtqueue_get_buf(vq, &unused));
> +			virtqueue_disable_cb(vq);
> +		}
> +	}
> +
> +	return err;
>  }
>  
>  static void set_page_pfns(struct virtio_balloon *vb,
> -- 
> MST
>