lib/nvme: Make internal of spdk_nvme_detach() asynchronous

Add two new helper functions, nvme_ctrlr_detach_async() and
nvme_ctrlr_detach_poll_async() to make the internal of
spdk_nvme_detach() asynchronous.

Use callback function to remove controller from the attached list after
completing shutdown and before freeing to avoid conflict between
attach and detach.

Update MOCKs in the corresponding unit test cases.

The next patch will add two public APIs spdk_nvme_detach_async()
and spdk_nvme_detach_poll_async() based on this patch.

Signed-off-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>
Change-Id: Ifbdfec2a1facde9354007c6248f280e245a36eed
Reviewed-on: https://review.spdk.io/gerrit/c/spdk/spdk/+/4416
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: Aleksey Marchuk <alexeymar@mellanox.com>
Reviewed-by: Jacek Kalwas <jacek.kalwas@intel.com>
Reviewed-by: Jim Harris <james.r.harris@intel.com>

lib/nvme/nvme.c

@@ -63,24 +63,98 @@ nvme_ctrlr_connected(struct spdk_nvme_probe_ctx *probe_ctx,
 	TAILQ_INSERT_TAIL(&probe_ctx->init_ctrlrs, ctrlr, tailq);
 }
 
-int
-spdk_nvme_detach(struct spdk_nvme_ctrlr *ctrlr)
+static void
+nvme_ctrlr_detach_async_finish(struct spdk_nvme_ctrlr *ctrlr)
 {
 	nvme_robust_mutex_lock(&g_spdk_nvme_driver->lock);
+	if (nvme_ctrlr_shared(ctrlr)) {
+		TAILQ_REMOVE(&g_spdk_nvme_driver->shared_attached_ctrlrs, ctrlr, tailq);
+	} else {
+		TAILQ_REMOVE(&g_nvme_attached_ctrlrs, ctrlr, tailq);
+	}
+	nvme_robust_mutex_unlock(&g_spdk_nvme_driver->lock);
+}
 
-	nvme_ctrlr_proc_put_ref(ctrlr);
+static int
+nvme_ctrlr_detach_async(struct spdk_nvme_ctrlr *ctrlr,
+			struct nvme_ctrlr_detach_ctx **_ctx)
+{
+	struct nvme_ctrlr_detach_ctx *ctx;
+	int ref_count;
 
-	if (nvme_ctrlr_get_ref_count(ctrlr) == 0) {
-		nvme_io_msg_ctrlr_detach(ctrlr);
-		if (nvme_ctrlr_shared(ctrlr)) {
-			TAILQ_REMOVE(&g_spdk_nvme_driver->shared_attached_ctrlrs, ctrlr, tailq);
-		} else {
-			TAILQ_REMOVE(&g_nvme_attached_ctrlrs, ctrlr, tailq);
+	nvme_robust_mutex_lock(&g_spdk_nvme_driver->lock);
+
+	ref_count = nvme_ctrlr_get_ref_count(ctrlr);
+	assert(ref_count > 0);
+
+	if (ref_count == 1) {
+		/* This is the last reference to the controller, so we need to
+		 * allocate a context to destruct it.
+		 */
+		ctx = calloc(1, sizeof(*ctx));
+		if (ctx == NULL) {
+			nvme_robust_mutex_unlock(&g_spdk_nvme_driver->lock);
+
+			return -ENOMEM;
 		}
-		nvme_ctrlr_destruct(ctrlr);
+		ctx->cb_fn = nvme_ctrlr_detach_async_finish;
+
+		nvme_ctrlr_proc_put_ref(ctrlr);
+
+		nvme_io_msg_ctrlr_detach(ctrlr);
+
+		nvme_ctrlr_destruct_async(ctrlr, ctx);
+
+		*_ctx = ctx;
+	} else {
+		nvme_ctrlr_proc_put_ref(ctrlr);
 	}
 
 	nvme_robust_mutex_unlock(&g_spdk_nvme_driver->lock);
+
+	return 0;
+}
+
+static int
+nvme_ctrlr_detach_poll_async(struct spdk_nvme_ctrlr *ctrlr,
+			     struct nvme_ctrlr_detach_ctx *ctx)
+{
+	int rc;
+
+	rc = nvme_ctrlr_destruct_poll_async(ctrlr, ctx);
+	if (rc == -EAGAIN) {
+		return -EAGAIN;
+	}
+
+	free(ctx);
+
+	return rc;
+}
+
+int
+spdk_nvme_detach(struct spdk_nvme_ctrlr *ctrlr)
+{
+	struct nvme_ctrlr_detach_ctx *ctx = NULL;
+	int rc;
+
+	rc = nvme_ctrlr_detach_async(ctrlr, &ctx);
+	if (rc != 0) {
+		return rc;
+	} else if (ctx == NULL) {
+		/* ctrlr was detached from the caller process but any other process
+		 * still attaches it.
+		 */
+		return 0;
+	}
+
+	while (1) {
+		rc = nvme_ctrlr_detach_poll_async(ctrlr, ctx);
+		if (rc != -EAGAIN) {
+			break;
+		}
+		nvme_delay(1000);
+	}
+
 	return 0;
 }
 

lib/nvme/nvme_ctrlr.c

@@ -3317,6 +3317,10 @@ nvme_ctrlr_destruct_poll_async(struct spdk_nvme_ctrlr *ctrlr,
 		/* Destruct ctrlr forcefully for any other error. */
 	}
 
+	if (ctx->cb_fn) {
+		ctx->cb_fn(ctrlr);
+	}
+
 	nvme_ctrlr_destruct_namespaces(ctrlr);
 
 	spdk_bit_array_free(&ctrlr->free_io_qids);

lib/nvme/nvme_internal.h

@@ -843,10 +843,13 @@ struct spdk_nvme_probe_ctx {
 	TAILQ_HEAD(, spdk_nvme_ctrlr)		init_ctrlrs;
 };
 
+typedef void (*nvme_ctrlr_detach_cb)(struct spdk_nvme_ctrlr *ctrlr);
+
 struct nvme_ctrlr_detach_ctx {
-	uint64_t	shutdown_start_tsc;
-	uint32_t	shutdown_timeout_ms;
-	bool		shutdown_complete;
+	nvme_ctrlr_detach_cb	cb_fn;
+	uint64_t		shutdown_start_tsc;
+	uint32_t		shutdown_timeout_ms;
+	bool			shutdown_complete;
 };
 
 struct nvme_driver {

test/unit/lib/nvme/nvme.c/nvme_ut.c

@@ -74,6 +74,23 @@ nvme_ctrlr_destruct(struct spdk_nvme_ctrlr *ctrlr)
 	ut_destruct_called = true;
 }
 
+void
+nvme_ctrlr_destruct_async(struct spdk_nvme_ctrlr *ctrlr, struct nvme_ctrlr_detach_ctx *ctx)
+{
+	ut_destruct_called = true;
+}
+
+int
+nvme_ctrlr_destruct_poll_async(struct spdk_nvme_ctrlr *ctrlr,
+			       struct nvme_ctrlr_detach_ctx *ctx)
+{
+	if (ctx->cb_fn) {
+		ctx->cb_fn(ctrlr);
+	}
+
+	return 0;
+}
+
 void
 spdk_nvme_ctrlr_get_default_ctrlr_opts(struct spdk_nvme_ctrlr_opts *opts, size_t opts_size)
 {
@@ -284,6 +301,7 @@ test_spdk_nvme_connect(void)
 	CU_ASSERT_EQUAL(ret_ctrlr->opts.num_io_queues, 1);
 	CU_ASSERT_EQUAL(ret_ctrlr->opts.opts_size, 4);
 	/* remove the attached ctrlr on the attached_list */
+	MOCK_SET(nvme_ctrlr_get_ref_count, 1);
 	CU_ASSERT(spdk_nvme_detach(&ctrlr) == 0);
 	CU_ASSERT(TAILQ_EMPTY(&g_spdk_nvme_driver->shared_attached_ctrlrs));
 
@@ -506,7 +524,7 @@ test_spdk_nvme_detach(void)
 	 * called (we aren't testing what the real destruct function does
 	 * here.)
 	 */
-	MOCK_SET(nvme_ctrlr_get_ref_count, 0);
+	MOCK_SET(nvme_ctrlr_get_ref_count, 1);
 	rc = spdk_nvme_detach(&ctrlr);
 	ret_ctrlr = TAILQ_FIRST(&test_driver.shared_attached_ctrlrs);
 	CU_ASSERT(ret_ctrlr == NULL);
@@ -518,7 +536,7 @@ test_spdk_nvme_detach(void)
 	 * function is not called and that attached ctrl list is
 	 * not empty.
 	 */
-	MOCK_SET(nvme_ctrlr_get_ref_count, 1);
+	MOCK_SET(nvme_ctrlr_get_ref_count, 2);
 	TAILQ_INSERT_TAIL(&test_driver.shared_attached_ctrlrs, &ctrlr, tailq);
 	ut_destruct_called = false;
 	rc = spdk_nvme_detach(&ctrlr);
@@ -536,7 +554,7 @@ test_spdk_nvme_detach(void)
 	ctrlr.trid.trtype = SPDK_NVME_TRANSPORT_RDMA;
 	TAILQ_INIT(&g_nvme_attached_ctrlrs);
 	TAILQ_INSERT_TAIL(&g_nvme_attached_ctrlrs, &ctrlr, tailq);
-	MOCK_SET(nvme_ctrlr_get_ref_count, 0);
+	MOCK_SET(nvme_ctrlr_get_ref_count, 1);
 	rc = spdk_nvme_detach(&ctrlr);
 	CU_ASSERT(TAILQ_EMPTY(&g_nvme_attached_ctrlrs));
 	CU_ASSERT(ut_destruct_called == true);