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 <[email protected]> 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....@redhat.com?part=1
