virtio: expose virtio descriptors as soon as possible

With upcoming event index patch we will need
to keep track of how many descriptor chains we
have really put into the avail vring. This patch
is a step towards that.

Our virtio layer abstracts away descriptor chains
as "requests". We can start requests, add descriptors
to them, and finally flush them. So far we used to put
any descriptors directly into the virtqueue desc ring,
and made them visible to the device only through
virtqueue_req_flush().

All of our virtio bdev modules currently flush the
virtqueue after adding each single request, but the
docs for the virtio API say it's possible to start
multiple subsequent requests and flush them all at
once. This was conceptually broken so far and only
the last request would be exposed to the device.

It's now fixed and subsequent requests are put
into the avail vring as soon as they're complete
(either the next request is started, or the
virtqueue is flushed).

Change-Id: I76b7db88ab094c38430edd8ff0e65681775dcb31
Signed-off-by: Dariusz Stojaczyk <dariuszx.stojaczyk@intel.com>
Reviewed-on: https://review.gerrithub.io/415590
Tested-by: SPDK Automated Test System <sys_sgsw@intel.com>
Reviewed-by: Daniel Verkamp <daniel.verkamp@intel.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>
Reviewed-by: Changpeng Liu <changpeng.liu@intel.com>

include/spdk_internal/virtio.h

@@ -193,16 +193,18 @@ typedef int (*virtio_pci_create_cb)(struct virtio_pci_ctx *pci_ctx, void *ctx);
 uint16_t virtio_recv_pkts(struct virtqueue *vq, void **io, uint32_t *len, uint16_t io_cnt);
 
 /**
- * Start a new request on the current vring head position. The request will
+ * Start a new request on the current vring head position and associate it
- * be bound to given opaque cookie object. All previous requests will be
+ * with an opaque cookie object. The previous request in given vq will be
- * still kept in a ring until they are flushed or the request is aborted.
+ * made visible to the device in hopes it can be processed early, but there's
- * If a previous request is empty (no descriptors have been added) this call
+ * no guarantee it will be until the device is notified with \c
- * will overwrite it. The device owning given virtqueue must be started.
+ * virtqueue_req_flush. This behavior is simply an optimization and virtqueues
+ * must always be flushed. Empty requests (with no descriptors added) will be
+ * ignored. The device owning given virtqueue must be started.
  *
  * \param vq virtio queue
- * \param cookie opaque object to bind with this request. Once the request
+ * \param cookie opaque object to associate with this request. Once the request
  * is sent, processed and a response is received, the same object will be
- * returned to the user calling the virtio poll API.
+ * returned to the user after calling the virtio poll API.
  * \param iovcnt number of required iovectors for the request. This can be
  * higher than than the actual number of iovectors to be added.
  * \return 0 on success or negative errno otherwise. If the `iovcnt` is
@@ -212,9 +214,8 @@ uint16_t virtio_recv_pkts(struct virtqueue *vq, void **io, uint32_t *len, uint16
 int virtqueue_req_start(struct virtqueue *vq, void *cookie, int iovcnt);
 
 /**
- * Flush a virtqueue. This will make the host device see and process all
+ * Flush a virtqueue. This will notify the device if it's required.
- * previously queued requests. An interrupt might be automatically sent if
+ * The device owning given virtqueue must be started.
- * the host device expects it. The device owning given virtqueue must be started.
  *
  * \param vq virtio queue
  */

lib/virtio/virtio.c

@@ -415,19 +415,41 @@ virtqueue_dequeue_burst_rx(struct virtqueue *vq, void **rx_pkts,
 	return i;
 }
 
+static void
+finish_req(struct virtqueue *vq)
+{
+	struct vring_desc *desc;
+	uint16_t avail_idx;
+
+	desc = &vq->vq_ring.desc[vq->req_end];
+	desc->flags &= ~VRING_DESC_F_NEXT;
+
+	/*
+	 * Place the head of the descriptor chain into the next slot and make
+	 * it usable to the host. The chain is made available now rather than
+	 * deferring to virtqueue_req_flush() in the hopes that if the host is
+	 * currently running on another CPU, we can keep it processing the new
+	 * descriptor.
+	 */
+	avail_idx = (uint16_t)(vq->vq_avail_idx & (vq->vq_nentries - 1));
+	vq->vq_ring.avail->ring[avail_idx] = vq->req_start;
+	vq->vq_avail_idx++;
+	vq->req_end = VQ_RING_DESC_CHAIN_END;
+	virtio_wmb();
+	vq->vq_ring.avail->idx = vq->vq_avail_idx;
+}
+
 int
 virtqueue_req_start(struct virtqueue *vq, void *cookie, int iovcnt)
 {
-	struct vring_desc *desc;
 	struct vq_desc_extra *dxp;
 
 	if (iovcnt > vq->vq_free_cnt) {
 		return iovcnt > vq->vq_nentries ? -EINVAL : -ENOMEM;
 	}
 
-	if (vq->req_start != VQ_RING_DESC_CHAIN_END) {
+	if (vq->req_end != VQ_RING_DESC_CHAIN_END) {
-		desc = &vq->vq_ring.desc[vq->req_end];
+		finish_req(vq);
-		desc->flags &= ~VRING_DESC_F_NEXT;
 	}
 
 	vq->req_start = vq->vq_desc_head_idx;
@@ -441,34 +463,14 @@ virtqueue_req_start(struct virtqueue *vq, void *cookie, int iovcnt)
 void
 virtqueue_req_flush(struct virtqueue *vq)
 {
-	struct vring_desc *desc;
+	if (vq->req_end == VQ_RING_DESC_CHAIN_END) {
-	uint16_t avail_idx;
+		/* no non-empty requests have been started */
-
-	if (vq->req_start == VQ_RING_DESC_CHAIN_END) {
-		/* no requests have been started */
 		return;
 	}
 
-	desc = &vq->vq_ring.desc[vq->req_end];
+	finish_req(vq);
-	desc->flags &= ~VRING_DESC_F_NEXT;
-
-	/*
-	 * Place the head of the descriptor chain into the next slot and make
-	 * it usable to the host. The chain is made available now rather than
-	 * deferring to virtqueue_notify() in the hopes that if the host is
-	 * currently running on another CPU, we can keep it processing the new
-	 * descriptor.
-	 */
-	avail_idx = (uint16_t)(vq->vq_avail_idx & (vq->vq_nentries - 1));
-	vq->vq_ring.avail->ring[avail_idx] = vq->req_start;
-
-	vq->vq_avail_idx++;
-	vq->req_start = VQ_RING_DESC_CHAIN_END;
-
-	virtio_wmb();
-	vq->vq_ring.avail->idx = vq->vq_avail_idx;
-
 	virtio_mb();
+
 	if (spdk_unlikely(!(vq->vq_ring.used->flags & VRING_USED_F_NO_NOTIFY))) {
 		virtio_dev_backend_ops(vq->vdev)->notify_queue(vq->vdev, vq);
 		SPDK_DEBUGLOG(SPDK_LOG_VIRTIO_DEV, "Notified backend after xmit\n");