[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@xxxxxxxxxx>
---
 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