virtqueue_split_pop() reads the next split avail ring entry after
checking that the ring is configured. Call virtqueue_num_heads() before
consuming that entry, so an avail index distance larger than the queue
size is rejected.

virtqueue_num_heads() also reports an empty queue and keeps the read
barrier needed before the avail ring entry is read. Drop the now unused
virtio_queue_empty_rcu() helper.

This prevents an invalid split queue state from being expanded into
repeated device command processing.

Link: https://gitlab.com/qemu-project/qemu/-/issues/3930
Signed-off-by: Jia Jia <physicalmtea@gmail.com>
---
v2:
- Drop virtio_queue_empty_rcu() from the split pop path.
- Keep only the existing !vq->vring.avail early exit before
  virtqueue_num_heads().
- Remove the now unused virtio_queue_empty_rcu() helper.

Tested with qemu-system-x86_64 11.0.50, built from origin/master
f893c46c3931 plus this patch, configured with:
  --target-list=x86_64-softmmu --enable-kvm --disable-tcg

The original virtio-iommu live-vring qtest reproducer used for the
report completed successfully on the patched build without host OOM,
confirming that this reproducer is fixed by the patch.

 hw/virtio/virtio.c | 30 ++++++------------------------
 1 file changed, 6 insertions(+), 24 deletions(-)

diff --git a/hw/virtio/virtio.c b/hw/virtio/virtio.c
index f4d86a3655..68ec3f0751 100644
--- a/hw/virtio/virtio.c
+++ b/hw/virtio/virtio.c
@@ -716,26 +716,6 @@ static inline bool is_desc_avail(uint16_t flags, bool wrap_counter)
     return (avail != used) && (avail == wrap_counter);
 }
 
-/* Fetch avail_idx from VQ memory only when we really need to know if
- * guest has added some buffers.
- * Called within rcu_read_lock().  */
-static int virtio_queue_empty_rcu(VirtQueue *vq)
-{
-    if (virtio_device_disabled(vq->vdev)) {
-        return 1;
-    }
-
-    if (unlikely(!vq->vring.avail)) {
-        return 1;
-    }
-
-    if (vq->shadow_avail_idx != vq->last_avail_idx) {
-        return 0;
-    }
-
-    return vring_avail_idx(vq) == vq->last_avail_idx;
-}
-
 static int virtio_queue_split_empty(VirtQueue *vq)
 {
     bool empty;
@@ -1748,12 +1728,14 @@ static void *virtqueue_split_pop(VirtQueue *vq, size_t sz)
     address_space_cache_init_empty(&indirect_desc_cache);
 
     RCU_READ_LOCK_GUARD();
-    if (virtio_queue_empty_rcu(vq)) {
+    if (unlikely(!vq->vring.avail)) {
+        goto done;
+    }
+
+    rc = virtqueue_num_heads(vq, vq->last_avail_idx);
+    if (rc <= 0) {
         goto done;
     }
-    /* Needed after virtio_queue_empty(), see comment in
-     * virtqueue_num_heads(). */
-    smp_rmb();
 
     /* When we start there are none of either input nor output. */
     out_num = in_num = elem_entries = 0;
-- 
2.34.1