Commit 19ddb108 authored by Dariusz Stojaczyk's avatar Dariusz Stojaczyk Committed by Jim Harris
Browse files

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: default avatarDariusz Stojaczyk <dariuszx.stojaczyk@intel.com>
Reviewed-on: https://review.gerrithub.io/382201


Tested-by: default avatarSPDK Automated Test System <sys_sgsw@intel.com>
Reviewed-by: default avatarJim Harris <james.r.harris@intel.com>
Reviewed-by: default avatarDaniel Verkamp <daniel.verkamp@intel.com>
parent 922aaac6

lib/bdev/virtio/bdev_virtio.c

+154 −3
Original line number Diff line number Diff line
@@ -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);

	}

}