nvme: asynchronous create io qpair

async_mode option is currently supported in PCIe transport layer
to create io qpair asynchronously. User polls the io_qpair for
completions, after create cq and sq completes in order, pqpair
is set to READY state. I/O submitted before the qpair is ready
is queued internally. Currently other transports only support
synchronous io qpair creation.

Signed-off-by: Monica Kenguva <monica.kenguva@intel.com>
Signed-off-by: Ben Walker <benjamin.walker@intel.com>
Change-Id: Ib2f9043872bd5602274e2508cf1fe9ff4211cabb
Reviewed-on: https://review.spdk.io/gerrit/c/spdk/spdk/+/8911
Community-CI: Mellanox Build Bot
Community-CI: Broadcom CI <spdk-ci.pdl@broadcom.com>
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: Konrad Sztyber <konrad.sztyber@intel.com>
Reviewed-by: Aleksey Marchuk <alexeymar@mellanox.com>

include/spdk/nvme.h

@@ -1434,6 +1434,14 @@ struct spdk_nvme_io_qpair_opts {
 	 * poll group and then connect it later.
 	 */
 	bool create_only;
+
+	/**
+	 * This flag if set to true enables the creation of submission and completion queue
+	 * asynchronously. This mode is currently supported at PCIe layer and tracks the
+	 * qpair creation with state machine and returns to the user.Default mode is set to
+	 * false to create io qpair synchronosuly.
+	 */
+	bool async_mode;
 };
 
 /**

lib/nvme/Makefile

@@ -34,7 +34,7 @@
 SPDK_ROOT_DIR := $(abspath $(CURDIR)/../..)
 include $(SPDK_ROOT_DIR)/mk/spdk.common.mk
 
-SO_VER := 6
+SO_VER := 7
 SO_MINOR := 0
 
 C_SRCS = nvme_ctrlr_cmd.c nvme_ctrlr.c nvme_fabric.c nvme_ns_cmd.c nvme_ns.c nvme_pcie_common.c nvme_pcie.c nvme_qpair.c nvme.c nvme_quirks.c nvme_transport.c \

lib/nvme/nvme_ctrlr.c

@@ -314,6 +314,10 @@ spdk_nvme_ctrlr_get_default_io_qpair_opts(struct spdk_nvme_ctrlr *ctrlr,
 		opts->create_only = false;
 	}
 
+	if (FIELD_OK(async_mode)) {
+		opts->async_mode = false;
+	}
+
 #undef FIELD_OK
 }
 
@@ -1449,6 +1453,7 @@ nvme_ctrlr_reinit_on_reset(struct spdk_nvme_ctrlr *ctrlr)
 {
 	struct spdk_nvme_qpair	*qpair;
 	int rc = 0, rc_tmp = 0;
+	bool async;
 
 	if (nvme_ctrlr_process_init(ctrlr) != 0) {
 		NVME_CTRLR_ERRLOG(ctrlr, "controller reinitialization failed\n");
@@ -1470,7 +1475,14 @@ nvme_ctrlr_reinit_on_reset(struct spdk_nvme_ctrlr *ctrlr)
 		TAILQ_FOREACH(qpair, &ctrlr->active_io_qpairs, tailq) {
 			assert(spdk_bit_array_get(ctrlr->free_io_qids, qpair->id));
 			spdk_bit_array_clear(ctrlr->free_io_qids, qpair->id);
+
+			/* Force a synchronous connect. We can't currently handle an asynchronous
+			 * operation here. */
+			async = qpair->async;
+			qpair->async = false;
 			rc_tmp = nvme_transport_ctrlr_connect_qpair(ctrlr, qpair);
+			qpair->async = async;
+
 			if (rc_tmp != 0) {
 				rc = rc_tmp;
 				qpair->transport_failure_reason = SPDK_NVME_QPAIR_FAILURE_LOCAL;
@@ -3292,6 +3304,7 @@ nvme_ctrlr_process_init(struct spdk_nvme_ctrlr *ctrlr)
 		break;
 
 	case NVME_CTRLR_STATE_CONNECT_ADMINQ: /* synonymous with NVME_CTRLR_STATE_INIT */
+		assert(ctrlr->adminq->async == false); /* not currently supported */
 		rc = nvme_transport_ctrlr_connect_qpair(ctrlr, ctrlr->adminq);
 		if (rc == 0) {
 			nvme_qpair_set_state(ctrlr->adminq, NVME_QPAIR_ENABLED);

lib/nvme/nvme_internal.h

@@ -408,6 +408,10 @@ struct spdk_nvme_qpair {
 
 	uint8_t					state : 3;
 
+	uint8_t					async: 1;
+
+	uint8_t					is_new_qpair: 1;
+
 	/*
 	 * Members for handling IO qpair deletion inside of a completion context.
 	 * These are specifically defined as single bits, so that they do not
@@ -1065,7 +1069,7 @@ void nvme_ctrlr_complete_queued_async_events(struct spdk_nvme_ctrlr *ctrlr);
 int nvme_qpair_init(struct spdk_nvme_qpair *qpair, uint16_t id,
 		    struct spdk_nvme_ctrlr *ctrlr,
 		    enum spdk_nvme_qprio qprio,
-		    uint32_t num_requests);
+		    uint32_t num_requests, bool async);
 void	nvme_qpair_deinit(struct spdk_nvme_qpair *qpair);
 void	nvme_qpair_complete_error_reqs(struct spdk_nvme_qpair *qpair);
 int	nvme_qpair_submit_request(struct spdk_nvme_qpair *qpair,
@@ -1073,7 +1077,6 @@ int	nvme_qpair_submit_request(struct spdk_nvme_qpair *qpair,
 void	nvme_qpair_abort_reqs(struct spdk_nvme_qpair *qpair, uint32_t dnr);
 uint32_t nvme_qpair_abort_queued_reqs(struct spdk_nvme_qpair *qpair, void *cmd_cb_arg);
 void	nvme_qpair_resubmit_requests(struct spdk_nvme_qpair *qpair, uint32_t num_requests);
-
 int	nvme_ctrlr_identify_active_ns(struct spdk_nvme_ctrlr *ctrlr);
 void	nvme_ns_set_identify_data(struct spdk_nvme_ns *ns);
 void	nvme_ns_set_id_desc_list_data(struct spdk_nvme_ns *ns);
@@ -1219,6 +1222,9 @@ static inline void
 nvme_qpair_set_state(struct spdk_nvme_qpair *qpair, enum nvme_qpair_state state)
 {
 	qpair->state = state;
+	if (state == NVME_QPAIR_ENABLED) {
+		qpair->is_new_qpair = false;
+	}
 }
 
 static inline enum nvme_qpair_state

lib/nvme/nvme_pcie_common.c

@@ -268,7 +268,8 @@ nvme_pcie_ctrlr_construct_admin_qpair(struct spdk_nvme_ctrlr *ctrlr, uint16_t nu
 			     0, /* qpair ID */
 			     ctrlr,
 			     SPDK_NVME_QPRIO_URGENT,
-			     num_entries);
+			     num_entries,
+			     false);
 	if (rc != 0) {
 		return rc;
 	}
@@ -1000,7 +1001,7 @@ nvme_pcie_ctrlr_create_io_qpair(struct spdk_nvme_ctrlr *ctrlr, uint16_t qid,
 
 	qpair = &pqpair->qpair;
 
-	rc = nvme_qpair_init(qpair, qid, ctrlr, opts->qprio, opts->io_queue_requests);
+	rc = nvme_qpair_init(qpair, qid, ctrlr, opts->qprio, opts->io_queue_requests, opts->async_mode);
 	if (rc != 0) {
 		nvme_pcie_qpair_destroy(qpair);
 		return NULL;

lib/nvme/nvme_qpair.c

@@ -620,9 +620,11 @@ nvme_qpair_check_enabled(struct spdk_nvme_qpair *qpair)
 		 * but we have historically not disconnected pcie qpairs during reset so we have to abort requests
 		 * here.
 		 */
-		if (qpair->ctrlr->trid.trtype == SPDK_NVME_TRANSPORT_PCIE) {
+		if (qpair->ctrlr->trid.trtype == SPDK_NVME_TRANSPORT_PCIE &&
+		    !qpair->is_new_qpair) {
 			nvme_qpair_abort_reqs(qpair, 0);
 		}
+
 		nvme_qpair_set_state(qpair, NVME_QPAIR_ENABLED);
 		while (!STAILQ_EMPTY(&qpair->queued_req)) {
 			req = STAILQ_FIRST(&qpair->queued_req);
@@ -753,7 +755,7 @@ int
 nvme_qpair_init(struct spdk_nvme_qpair *qpair, uint16_t id,
 		struct spdk_nvme_ctrlr *ctrlr,
 		enum spdk_nvme_qprio qprio,
-		uint32_t num_requests)
+		uint32_t num_requests, bool async)
 {
 	size_t req_size_padded;
 	uint32_t i;
@@ -767,6 +769,8 @@ nvme_qpair_init(struct spdk_nvme_qpair *qpair, uint16_t id,
 
 	qpair->ctrlr = ctrlr;
 	qpair->trtype = ctrlr->trid.trtype;
+	qpair->is_new_qpair = true;
+	qpair->async = async;
 
 	STAILQ_INIT(&qpair->free_req);
 	STAILQ_INIT(&qpair->queued_req);

lib/nvme/nvme_rdma.c

@@ -1612,7 +1612,7 @@ nvme_rdma_ctrlr_create_qpair(struct spdk_nvme_ctrlr *ctrlr,
 	rqpair->num_entries = qsize;
 	rqpair->delay_cmd_submit = delay_cmd_submit;
 	qpair = &rqpair->qpair;
-	rc = nvme_qpair_init(qpair, qid, ctrlr, qprio, num_requests);
+	rc = nvme_qpair_init(qpair, qid, ctrlr, qprio, num_requests, false);
 	if (rc != 0) {
 		nvme_rdma_free(rqpair);
 		return NULL;

lib/nvme/nvme_tcp.c

@@ -1964,7 +1964,7 @@ nvme_tcp_ctrlr_create_qpair(struct spdk_nvme_ctrlr *ctrlr,
 
 	tqpair->num_entries = qsize;
 	qpair = &tqpair->qpair;
-	rc = nvme_qpair_init(qpair, qid, ctrlr, qprio, num_requests);
+	rc = nvme_qpair_init(qpair, qid, ctrlr, qprio, num_requests, false);
 	if (rc != 0) {
 		free(tqpair);
 		return NULL;

lib/nvme/nvme_transport.c

@@ -369,6 +369,7 @@ nvme_transport_ctrlr_connect_qpair(struct spdk_nvme_ctrlr *ctrlr, struct spdk_nv
 		goto err;
 	}
 
+	if (!qpair->async) {
 		/* Busy wait until the qpair exits the connecting state */
 		while (nvme_qpair_get_state(qpair) == NVME_QPAIR_CONNECTING) {
 			rc = spdk_nvme_qpair_process_completions(qpair, 0);
@@ -376,6 +377,7 @@ nvme_transport_ctrlr_connect_qpair(struct spdk_nvme_ctrlr *ctrlr, struct spdk_nv
 				goto err;
 			}
 		}
+	}
 
 	if (qpair->poll_group) {
 		rc = nvme_poll_group_connect_qpair(qpair);

module/bdev/nvme/bdev_nvme.c

@@ -409,6 +409,7 @@ bdev_nvme_create_qpair(struct nvme_ctrlr_channel *ctrlr_ch)
 	spdk_nvme_ctrlr_get_default_io_qpair_opts(ctrlr, &opts, sizeof(opts));
 	opts.delay_cmd_submit = g_opts.delay_cmd_submit;
 	opts.create_only = true;
+	opts.async_mode = true;
 	opts.io_queue_requests = spdk_max(g_opts.io_queue_requests, opts.io_queue_requests);
 	g_opts.io_queue_requests = opts.io_queue_requests;
 

test/common/lib/nvme/common_stubs.h

@@ -113,11 +113,12 @@ int
 nvme_qpair_init(struct spdk_nvme_qpair *qpair, uint16_t id,
 		struct spdk_nvme_ctrlr *ctrlr,
 		enum spdk_nvme_qprio qprio,
-		uint32_t num_requests)
+		uint32_t num_requests, bool async)
 {
 	qpair->ctrlr = ctrlr;
 	qpair->id = id;
 	qpair->qprio = qprio;
+	qpair->async = async;
 	qpair->trtype = SPDK_NVME_TRANSPORT_TCP;
 	qpair->poll_group = (void *)0xDEADBEEF;
 

test/unit/lib/nvme/nvme_ctrlr.c/nvme_ctrlr_ut.c

@@ -240,11 +240,12 @@ nvme_driver_init(void)
 int nvme_qpair_init(struct spdk_nvme_qpair *qpair, uint16_t id,
 		    struct spdk_nvme_ctrlr *ctrlr,
 		    enum spdk_nvme_qprio qprio,
-		    uint32_t num_requests)
+		    uint32_t num_requests, bool async)
 {
 	qpair->id = id;
 	qpair->qprio = qprio;
 	qpair->ctrlr = ctrlr;
+	qpair->async = async;
 
 	return 0;
 }

test/unit/lib/nvme/nvme_pcie_common.c/nvme_pcie_common_ut.c

@@ -76,11 +76,12 @@ DEFINE_STUB(nvme_request_check_timeout, int, (struct nvme_request *req, uint16_t
 int nvme_qpair_init(struct spdk_nvme_qpair *qpair, uint16_t id,
 		    struct spdk_nvme_ctrlr *ctrlr,
 		    enum spdk_nvme_qprio qprio,
-		    uint32_t num_requests)
+		    uint32_t num_requests, bool async)
 {
 	qpair->id = id;
 	qpair->qprio = qprio;
 	qpair->ctrlr = ctrlr;
+	qpair->async = async;
 
 	return 0;
 }
@@ -318,7 +319,7 @@ test_nvme_pcie_ctrlr_connect_qpair(void)
 	struct spdk_nvme_transport_poll_group poll_group = {};
 	struct spdk_nvme_cpl cpl = {};
 	struct spdk_nvme_qpair adminq = {};
-	struct nvme_request req[2] = {};
+	struct nvme_request req[3] = {};
 	int rc;
 
 	pqpair.cpl = &cpl;
@@ -367,6 +368,8 @@ test_nvme_pcie_ctrlr_connect_qpair(void)
 
 	/* Complete the second request */
 	req[1].cb_fn(req[1].cb_arg, &cpl);
+	CU_ASSERT(pqpair.pcie_state == NVME_PCIE_QPAIR_READY);
+	CU_ASSERT(pqpair.qpair.state == NVME_QPAIR_CONNECTED);
 
 	/* doorbell stride and qid are 1 */
 	CU_ASSERT(pqpair.shadow_doorbell.sq_tdbl == pctrlr.ctrlr.shadow_doorbell + 2);
@@ -417,12 +420,90 @@ test_nvme_pcie_ctrlr_connect_qpair(void)
 
 	/* Complete the second request */
 	req[1].cb_fn(req[1].cb_arg, &cpl);
+	CU_ASSERT(pqpair.pcie_state == NVME_PCIE_QPAIR_READY);
+	CU_ASSERT(pqpair.qpair.state == NVME_QPAIR_CONNECTED);
 
 	CU_ASSERT(pqpair.shadow_doorbell.sq_tdbl == NULL);
 	CU_ASSERT(pqpair.shadow_doorbell.sq_eventidx == NULL);
 	CU_ASSERT(pqpair.flags.has_shadow_doorbell == 0);
 	CU_ASSERT(STAILQ_EMPTY(&pctrlr.ctrlr.adminq->free_req));
 
+	/* Completion error for CQ */
+	memset(req, 0, sizeof(struct nvme_request) * 2);
+	memset(&pqpair, 0, sizeof(pqpair));
+	pqpair.cpl = &cpl;
+	pqpair.qpair.ctrlr = &pctrlr.ctrlr;
+	pqpair.qpair.id = 1;
+	pqpair.num_entries = 1;
+	pqpair.cpl_bus_addr = 0xDEADBEEF;
+	pqpair.cmd_bus_addr = 0xDDADBEEF;
+	pqpair.qpair.qprio = SPDK_NVME_QPRIO_HIGH;
+	pqpair.stat = NULL;
+	pqpair.qpair.poll_group = &poll_group;
+	/* Modify cpl such that CQ fails */
+	pqpair.cpl->status.sc = SPDK_NVME_SC_INVALID_FIELD;
+	pqpair.cpl->status.sct = SPDK_NVME_SCT_GENERIC;
+	for (int i = 0; i < 2; i++) {
+		STAILQ_INSERT_TAIL(&pctrlr.ctrlr.adminq->free_req, &req[i], stailq);
+	}
+
+	rc = nvme_pcie_ctrlr_connect_qpair(&pctrlr.ctrlr, &pqpair.qpair);
+	CU_ASSERT(rc == 0);
+	CU_ASSERT(req[0].cmd.opc == SPDK_NVME_OPC_CREATE_IO_CQ);
+
+	/* Request to complete callback in async operation */
+	req[0].cb_fn(req[0].cb_arg, &cpl);
+	CU_ASSERT(pqpair.pcie_state == NVME_PCIE_QPAIR_FAILED);
+	CU_ASSERT(pqpair.qpair.state == NVME_QPAIR_DISCONNECTED);
+
+	/* Remove unused request */
+	STAILQ_REMOVE_HEAD(&pctrlr.ctrlr.adminq->free_req, stailq);
+	CU_ASSERT(STAILQ_EMPTY(&pctrlr.ctrlr.adminq->free_req));
+
+	/* Completion error for SQ */
+	memset(req, 0, sizeof(struct nvme_request) * 3);
+	memset(&pqpair, 0, sizeof(pqpair));
+	pqpair.cpl = &cpl;
+	pqpair.qpair.ctrlr = &pctrlr.ctrlr;
+	pqpair.qpair.id = 1;
+	pqpair.num_entries = 1;
+	pqpair.cpl_bus_addr = 0xDEADBEEF;
+	pqpair.cmd_bus_addr = 0xDDADBEEF;
+	pqpair.qpair.qprio = SPDK_NVME_QPRIO_HIGH;
+	pqpair.stat = NULL;
+	pqpair.cpl->status.sc = SPDK_NVME_SC_SUCCESS;
+	pqpair.cpl->status.sct = SPDK_NVME_SCT_GENERIC;
+	pqpair.qpair.poll_group = &poll_group;
+	for (int i = 0; i < 3; i++) {
+		STAILQ_INSERT_TAIL(&pctrlr.ctrlr.adminq->free_req, &req[i], stailq);
+	}
+
+	rc = nvme_pcie_ctrlr_connect_qpair(&pctrlr.ctrlr, &pqpair.qpair);
+	CU_ASSERT(rc == 0);
+	CU_ASSERT(req[0].cmd.opc == SPDK_NVME_OPC_CREATE_IO_CQ);
+	CU_ASSERT(pqpair.pcie_state == NVME_PCIE_QPAIR_WAIT_FOR_CQ);
+
+	/* Request to complete cq callback in async operation */
+	req[0].cb_fn(req[0].cb_arg, &cpl);
+	CU_ASSERT(req[1].cmd.opc == SPDK_NVME_OPC_CREATE_IO_SQ);
+	CU_ASSERT(pqpair.pcie_state == NVME_PCIE_QPAIR_WAIT_FOR_SQ);
+	/* Modify cpl such that SQ fails */
+	pqpair.cpl->status.sc = SPDK_NVME_SC_INVALID_FIELD;
+	pqpair.cpl->status.sct = SPDK_NVME_SCT_GENERIC;
+
+	/* Request to complete sq callback in async operation */
+	req[1].cb_fn(req[1].cb_arg, &cpl);
+	CU_ASSERT(req[2].cmd.opc == SPDK_NVME_OPC_DELETE_IO_CQ);
+	/* Modify cpl back to success */
+	pqpair.cpl->status.sc = SPDK_NVME_SC_SUCCESS;
+	pqpair.cpl->status.sct = SPDK_NVME_SCT_GENERIC;
+	req[2].cb_fn(req[2].cb_arg, &cpl);
+	CU_ASSERT(pqpair.pcie_state == NVME_PCIE_QPAIR_FAILED);
+	CU_ASSERT(pqpair.qpair.state == NVME_QPAIR_DISCONNECTED);
+	/* No need to remove unused requests here */
+	CU_ASSERT(STAILQ_EMPTY(&pctrlr.ctrlr.adminq->free_req));
+
+
 	/* No available request used */
 	memset(req, 0, sizeof(struct nvme_request) * 2);
 	memset(&pqpair, 0, sizeof(pqpair));

test/unit/lib/nvme/nvme_qpair.c/nvme_qpair_ut.c

@@ -87,7 +87,7 @@ prepare_submit_request_test(struct spdk_nvme_qpair *qpair,
 	TAILQ_INIT(&ctrlr->active_io_qpairs);
 	TAILQ_INIT(&ctrlr->active_procs);
 	MOCK_CLEAR(spdk_zmalloc);
-	nvme_qpair_init(qpair, 1, ctrlr, 0, 32);
+	nvme_qpair_init(qpair, 1, ctrlr, 0, 32, false);
 }
 
 static void
@@ -191,8 +191,8 @@ static void test_nvme_qpair_process_completions(void)
 
 	TAILQ_INIT(&ctrlr.active_io_qpairs);
 	TAILQ_INIT(&ctrlr.active_procs);
-	nvme_qpair_init(&qpair, 1, &ctrlr, 0, 32);
+	nvme_qpair_init(&qpair, 1, &ctrlr, 0, 32, false);
-	nvme_qpair_init(&admin_qp, 0, &ctrlr, 0, 32);
+	nvme_qpair_init(&admin_qp, 0, &ctrlr, 0, 32, false);
 
 	ctrlr.adminq = &admin_qp;
 
@@ -665,7 +665,7 @@ test_nvme_qpair_init_deinit(void)
 
 	ctrlr.trid.trtype = SPDK_NVME_TRANSPORT_PCIE;
 
-	rc = nvme_qpair_init(&qpair, 1, &ctrlr, SPDK_NVME_QPRIO_HIGH, 3);
+	rc = nvme_qpair_init(&qpair, 1, &ctrlr, SPDK_NVME_QPRIO_HIGH, 3, false);
 	CU_ASSERT(rc == 0);
 	CU_ASSERT(qpair.id == 1);
 	CU_ASSERT(qpair.qprio == SPDK_NVME_QPRIO_HIGH);