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

[sashiko-bot@kernel.org]

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