bdev_virtio: implement controlq and RESETs

A single thread that initialized
bdev_virtio subsystem will be used
to poll control queues for all
virtio devices.

Every io_channel will put it's e.g.
RESET requests to the controlq MP-SC
spdk_ring. The controlq poller will
dequeue these requests and put them
into the virtio queue. The same
poller will then poll for completions
and send them to proper threads
that submitted given requests.

Change-Id: I90ae7c5d76dc89cc52ff69cf2a60e27a9314557c
Signed-off-by: Dariusz Stojaczyk <dariuszx.stojaczyk@intel.com>
Reviewed-on: https://review.gerrithub.io/382201
Tested-by: SPDK Automated Test System <sys_sgsw@intel.com>
Reviewed-by: Jim Harris <james.r.harris@intel.com>
Reviewed-by: Daniel Verkamp <daniel.verkamp@intel.com>

lib/bdev/virtio/bdev_virtio.c

@@ -52,6 +52,8 @@
 
 #define BDEV_VIRTIO_MAX_TARGET 64
 #define BDEV_VIRTIO_SCAN_PAYLOAD_SIZE 256
+#define CTRLQ_POLL_PERIOD_US (1000 * 5)
+#define CTRLQ_RING_SIZE 16
 
 #define VIRTIO_SCSI_CONTROLQ	0
 #define VIRTIO_SCSI_EVENTQ	1
@@ -62,8 +64,14 @@ static void bdev_virtio_finish(void);
 
 struct virtio_scsi_io_ctx {
 	struct virtio_req 		vreq;
+	union {
 		struct virtio_scsi_cmd_req req;
+		struct virtio_scsi_ctrl_tmf_req tmf_req;
+	};
+	union {
 		struct virtio_scsi_cmd_resp resp;
+		struct virtio_scsi_ctrl_tmf_resp tmf_resp;
+	};
 };
 
 struct virtio_scsi_scan_base {
@@ -140,6 +148,34 @@ bdev_virtio_init_io_vreq(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_i
 	return vreq;
 }
 
+static struct virtio_req *
+bdev_virtio_init_tmf_vreq(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io)
+{
+	struct virtio_req *vreq;
+	struct virtio_scsi_ctrl_tmf_req *tmf_req;
+	struct virtio_scsi_ctrl_tmf_resp *tmf_resp;
+	struct virtio_scsi_disk *disk = SPDK_CONTAINEROF(bdev_io->bdev, struct virtio_scsi_disk, bdev);
+	struct virtio_scsi_io_ctx *io_ctx = (struct virtio_scsi_io_ctx *)bdev_io->driver_ctx;
+
+	vreq = &io_ctx->vreq;
+	tmf_req = &io_ctx->tmf_req;
+	tmf_resp = &io_ctx->tmf_resp;
+
+	vreq->iov = NULL;
+	vreq->iov_req.iov_base = tmf_req;
+	vreq->iov_req.iov_len = sizeof(*tmf_req);
+	vreq->iov_resp.iov_base = tmf_resp;
+	vreq->iov_resp.iov_len = sizeof(*tmf_resp);
+	vreq->iovcnt = 0;
+	vreq->is_write = false;
+
+	memset(tmf_req, 0, sizeof(*tmf_req));
+	tmf_req->lun[0] = 1;
+	tmf_req->lun[1] = disk->target;
+
+	return vreq;
+}
+
 static void
 bdev_virtio_rw(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io)
 {
@@ -164,6 +200,20 @@ bdev_virtio_rw(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io)
 	virtio_xmit_pkts(virtio_channel->vq, vreq);
 }
 
+static void
+bdev_virtio_reset(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io)
+{
+	struct bdev_virtio_io_channel *virtio_ch = spdk_io_channel_get_ctx(ch);
+	struct virtio_req *vreq = bdev_virtio_init_tmf_vreq(ch, bdev_io);
+	struct virtio_scsi_ctrl_tmf_req *tmf_req = vreq->iov_req.iov_base;
+	struct spdk_ring *ctrlq_send_ring = virtio_ch->vdev->vqs[VIRTIO_SCSI_CONTROLQ]->poller_ctx;
+
+	tmf_req->type = VIRTIO_SCSI_T_TMF;
+	tmf_req->subtype = VIRTIO_SCSI_T_TMF_LOGICAL_UNIT_RESET;
+
+	spdk_ring_enqueue(ctrlq_send_ring, (void **)&vreq, 1);
+}
+
 static void
 bdev_virtio_unmap(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io)
 {
@@ -223,7 +273,7 @@ static int _bdev_virtio_submit_request(struct spdk_io_channel *ch, struct spdk_b
 		bdev_virtio_rw(ch, bdev_io);
 		return 0;
 	case SPDK_BDEV_IO_TYPE_RESET:
-		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_SUCCESS);
+		bdev_virtio_reset(ch, bdev_io);
 		return 0;
 	case SPDK_BDEV_IO_TYPE_UNMAP: {
 		uint64_t buf_len = 8 /* header size */ +
@@ -340,6 +390,49 @@ bdev_virtio_poll(void *arg)
 	}
 }
 
+static void
+bdev_virtio_tmf_cpl_cb(void *ctx)
+{
+	struct virtio_req *req = ctx;
+	struct virtio_scsi_io_ctx *io_ctx = SPDK_CONTAINEROF(req, struct virtio_scsi_io_ctx, vreq);
+	struct spdk_bdev_io *bdev_io = spdk_bdev_io_from_ctx(io_ctx);
+
+	if (io_ctx->tmf_resp.response == VIRTIO_SCSI_S_OK) {
+		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_SUCCESS);
+	} else {
+		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
+	}
+}
+
+static void
+bdev_virtio_tmf_cpl(struct virtio_req *req)
+{
+	struct virtio_scsi_io_ctx *io_ctx = SPDK_CONTAINEROF(req, struct virtio_scsi_io_ctx, vreq);
+	struct spdk_bdev_io *bdev_io = spdk_bdev_io_from_ctx(io_ctx);
+
+	spdk_thread_send_msg(spdk_bdev_io_get_thread(bdev_io), bdev_virtio_tmf_cpl_cb, req);
+}
+
+static void
+bdev_virtio_ctrlq_poll(void *arg)
+{
+	struct virtio_dev *vdev = arg;
+	struct virtqueue *ctrlq = vdev->vqs[VIRTIO_SCSI_CONTROLQ];
+	struct spdk_ring *send_ring = ctrlq->poller_ctx;
+	struct virtio_req *req[16];
+	uint16_t i, cnt;
+
+	cnt = spdk_ring_dequeue(send_ring, (void **)req, SPDK_COUNTOF(req));
+	for (i = 0; i < cnt; ++i) {
+		virtio_xmit_pkts(ctrlq, req[i]);
+	}
+
+	cnt = virtio_recv_pkts(ctrlq, req, SPDK_COUNTOF(req));
+	for (i = 0; i < cnt; ++i) {
+		bdev_virtio_tmf_cpl(req[i]);
+	}
+}
+
 static int
 bdev_virtio_create_cb(void *io_device, void *ctx_buf)
 {
@@ -377,10 +470,32 @@ bdev_virtio_destroy_cb(void *io_device, void *ctx_buf)
 	virtio_dev_release_queue(vdev, vq->vq_queue_index);
 }
 
+static void
+scan_target_abort(struct virtio_scsi_scan_base *base)
+{
+	struct virtio_scsi_disk *disk;
+
+	while ((disk = TAILQ_FIRST(&base->found_disks))) {
+		TAILQ_REMOVE(&base->found_disks, disk, link);
+		free(disk);
+	}
+
+	TAILQ_REMOVE(&g_virtio_driver.init_ctrlrs, base->vdev, tailq);
+	virtio_dev_free(base->vdev);
+	spdk_dma_free(base);
+
+	if (TAILQ_EMPTY(&g_virtio_driver.init_ctrlrs)) {
+		spdk_bdev_module_init_done(SPDK_GET_BDEV_MODULE(virtio_scsi));
+	}
+}
+
 static void
 scan_target_finish(struct virtio_scsi_scan_base *base)
 {
 	struct virtio_scsi_disk *disk;
+	struct virtqueue *ctrlq;
+	struct spdk_ring *ctrlq_ring;
+	int rc;
 
 	base->target++;
 	if (base->target < BDEV_VIRTIO_MAX_TARGET) {
@@ -389,8 +504,31 @@ scan_target_finish(struct virtio_scsi_scan_base *base)
 	}
 
 	spdk_bdev_poller_stop(&base->vq->poller);
+	base->vq->poller_ctx = NULL;
 	virtio_dev_release_queue(base->vdev, base->vq->vq_queue_index);
 
+	ctrlq_ring = spdk_ring_create(SPDK_RING_TYPE_MP_SC, CTRLQ_RING_SIZE,
+				      SPDK_ENV_SOCKET_ID_ANY);
+	if (ctrlq_ring == NULL) {
+		SPDK_ERRLOG("Failed to allocate send ring for the controlq.\n");
+		scan_target_abort(base);
+		return;
+	}
+
+	rc = virtio_dev_acquire_queue(base->vdev, VIRTIO_SCSI_CONTROLQ);
+	if (rc != 0) {
+		SPDK_ERRLOG("Failed to acquire the controlq.\n");
+		assert(false);
+		spdk_ring_free(ctrlq_ring);
+		scan_target_abort(base);
+		return;
+	}
+
+	ctrlq = base->vdev->vqs[VIRTIO_SCSI_CONTROLQ];
+	ctrlq->poller_ctx = ctrlq_ring;
+	spdk_bdev_poller_start(&ctrlq->poller, bdev_virtio_ctrlq_poll, base->vdev,
+			       ctrlq->owner_lcore, CTRLQ_POLL_PERIOD_US);
+
 	while ((disk = TAILQ_FIRST(&base->found_disks))) {
 		TAILQ_REMOVE(&base->found_disks, disk, link);
 		spdk_io_device_register(&disk->vdev, bdev_virtio_create_cb, bdev_virtio_destroy_cb,
@@ -770,8 +908,21 @@ out:
 static void bdev_virtio_finish(void)
 {
 	struct virtio_dev *vdev, *next;
+	struct virtqueue *vq;
+	struct spdk_ring *send_ring;
 
 	TAILQ_FOREACH_SAFE(vdev, &g_virtio_driver.attached_ctrlrs, tailq, next) {
+		TAILQ_REMOVE(&g_virtio_driver.attached_ctrlrs, vdev, tailq);
+		if (virtio_dev_queue_is_acquired(vdev, VIRTIO_SCSI_CONTROLQ)) {
+			vq = vdev->vqs[VIRTIO_SCSI_CONTROLQ];
+			spdk_bdev_poller_stop(&vq->poller);
+			send_ring = vq->poller_ctx;
+			/* bdevs built on top of this vdev mustn't be destroyed with outstanding I/O. */
+			assert(spdk_ring_count(send_ring) == 0);
+			spdk_ring_free(send_ring);
+			vq->poller_ctx = NULL;
+			virtio_dev_release_queue(vdev, VIRTIO_SCSI_CONTROLQ);
+		}
 		virtio_dev_free(vdev);
 	}
 }