nvmf: simplify the qpair_mask handling.

We should not use mutex, but use the spdk_send_msg policy,
then we can let only one thread to handle that and
eliminates the segement fault issue.

Now in the code, the qpair_mask is handled by the same
thread, e.g., the thread which owns the admin qpair of
the ctrlr.

Change-Id: I609fd4d49f5ecc85bc47bf9c23afbb507900be7c
Signed-off-by: Ziye Yang <optimistyzy@gmail.com>
Reviewed-on: https://review.gerrithub.io/420827
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Chandler-Test-Pool: SPDK Automated Test System <sys_sgsw@intel.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>
Reviewed-by: Changpeng Liu <changpeng.liu@intel.com>
Reviewed-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>

lib/nvmf/ctrlr.c

@@ -77,14 +77,14 @@ ctrlr_add_qpair_and_update_rsp(struct spdk_nvmf_qpair *qpair,
 			       struct spdk_nvmf_ctrlr *ctrlr,
 			       struct spdk_nvmf_fabric_connect_rsp *rsp)
 {
-	pthread_mutex_lock(&ctrlr->mtx);
+	assert(ctrlr->admin_qpair->group->thread == spdk_get_thread());
+
 	/* check if we would exceed ctrlr connection limit */
 	if (qpair->qid >= spdk_bit_array_capacity(ctrlr->qpair_mask)) {
 		SPDK_ERRLOG("Requested QID %u but Max QID is %u\n",
 			    qpair->qid, spdk_bit_array_capacity(ctrlr->qpair_mask) - 1);
 		rsp->status.sct = SPDK_NVME_SCT_COMMAND_SPECIFIC;
 		rsp->status.sc = SPDK_NVME_SC_INVALID_QUEUE_IDENTIFIER;
-		pthread_mutex_unlock(&ctrlr->mtx);
 		return;
 	}
 
@@ -92,7 +92,6 @@ ctrlr_add_qpair_and_update_rsp(struct spdk_nvmf_qpair *qpair,
 		SPDK_ERRLOG("Got I/O connect with duplicate QID %u\n", qpair->qid);
 		rsp->status.sct = SPDK_NVME_SCT_COMMAND_SPECIFIC;
 		rsp->status.sc = SPDK_NVME_SC_INVALID_QUEUE_IDENTIFIER;
-		pthread_mutex_unlock(&ctrlr->mtx);
 		return;
 	}
 
@@ -103,8 +102,6 @@ ctrlr_add_qpair_and_update_rsp(struct spdk_nvmf_qpair *qpair,
 	rsp->status_code_specific.success.cntlid = ctrlr->cntlid;
 	SPDK_DEBUGLOG(SPDK_LOG_NVMF, "connect capsule response: cntlid = 0x%04x\n",
 		      rsp->status_code_specific.success.cntlid);
-
-	pthread_mutex_unlock(&ctrlr->mtx);
 }
 
 static void
@@ -168,11 +165,6 @@ spdk_nvmf_ctrlr_create(struct spdk_nvmf_subsystem *subsystem,
 	req->qpair->ctrlr = ctrlr;
 	ctrlr->subsys = subsystem;
 
-	if (pthread_mutex_init(&ctrlr->mtx, NULL) != 0) {
-		SPDK_ERRLOG("Failed to initialize controller mutex\n");
-		free(ctrlr);
-		return NULL;
-	}
 	ctrlr->qpair_mask = spdk_bit_array_create(tgt->opts.max_qpairs_per_ctrlr);
 	if (!ctrlr->qpair_mask) {
 		SPDK_ERRLOG("Failed to allocate controller qpair mask\n");
@@ -224,10 +216,6 @@ spdk_nvmf_ctrlr_create(struct spdk_nvmf_subsystem *subsystem,
 void
 spdk_nvmf_ctrlr_destruct(struct spdk_nvmf_ctrlr *ctrlr)
 {
-	/* TODO: Verify that qpair mask has been cleared. */
-
-	spdk_bit_array_free(&ctrlr->qpair_mask);
-
 	spdk_nvmf_subsystem_remove_ctrlr(ctrlr->subsys, ctrlr);
 
 	free(ctrlr);
@@ -876,10 +864,7 @@ spdk_nvmf_ctrlr_set_features_number_of_queues(struct spdk_nvmf_request *req)
 	SPDK_DEBUGLOG(SPDK_LOG_NVMF, "Set Features - Number of Queues, cdw11 0x%x\n",
 		      req->cmd->nvme_cmd.cdw11);
 
-	pthread_mutex_lock(&ctrlr->mtx);
 	count = spdk_bit_array_count_set(ctrlr->qpair_mask);
-	pthread_mutex_unlock(&ctrlr->mtx);
-
 	/* verify that the contoller is ready to process commands */
 	if (count > 1) {
 		SPDK_DEBUGLOG(SPDK_LOG_NVMF, "Queue pairs already active!\n");

lib/nvmf/nvmf.c

@@ -65,6 +65,7 @@ struct nvmf_qpair_disconnect_ctx {
 	nvmf_qpair_disconnect_cb cb_fn;
 	struct spdk_thread *thread;
 	void *ctx;
+	uint16_t qid;
 };
 
 /*
@@ -655,13 +656,29 @@ spdk_nvmf_poll_group_remove(struct spdk_nvmf_poll_group *group,
 	return rc;
 }
 
+static
+void _nvmf_ctrlr_destruct(void *ctx)
+{
+	struct spdk_nvmf_ctrlr *ctrlr = ctx;
+
+	spdk_nvmf_ctrlr_destruct(ctrlr);
+}
+
 static void
 _spdk_nvmf_ctrlr_free_from_qpair(void *ctx)
 {
 	struct nvmf_qpair_disconnect_ctx *qpair_ctx = ctx;
 	struct spdk_nvmf_ctrlr *ctrlr = qpair_ctx->ctrlr;
+	uint32_t count;
 
-	spdk_nvmf_ctrlr_destruct(ctrlr);
+	spdk_bit_array_clear(ctrlr->qpair_mask, qpair_ctx->qid);
+	count = spdk_bit_array_count_set(ctrlr->qpair_mask);
+	if (count == 0) {
+		/* TODO: Verify that qpair mask has been cleared. */
+		spdk_bit_array_free(&ctrlr->qpair_mask);
+
+		spdk_thread_send_msg(ctrlr->subsys->thread, _nvmf_ctrlr_destruct, ctrlr);
+	}
 
 	if (qpair_ctx->cb_fn) {
 		spdk_thread_send_msg(qpair_ctx->thread, qpair_ctx->cb_fn, qpair_ctx->ctx);
@@ -675,17 +692,21 @@ _spdk_nvmf_qpair_destroy(void *ctx, int status)
 	struct nvmf_qpair_disconnect_ctx *qpair_ctx = ctx;
 	struct spdk_nvmf_qpair *qpair = qpair_ctx->qpair;
 	struct spdk_nvmf_ctrlr *ctrlr = qpair->ctrlr;
-	uint16_t qid = qpair->qid;
+	struct spdk_thread *thread = NULL;
-	uint32_t count;
 
+	if (ctrlr && ctrlr->admin_qpair && ctrlr->admin_qpair->group) {
+		/* store the thread of admin_qpair and use it later */
+		thread = ctrlr->admin_qpair->group->thread;
+	}
 	spdk_nvmf_poll_group_remove(qpair->group, qpair);
 
 	assert(qpair->state == SPDK_NVMF_QPAIR_DEACTIVATING);
 	qpair->state = SPDK_NVMF_QPAIR_INACTIVE;
+	qpair_ctx->qid = qpair->qid;
 
 	spdk_nvmf_transport_qpair_fini(qpair);
 
-	if (!ctrlr) {
+	if (!thread) {
 		if (qpair_ctx->cb_fn) {
 			spdk_thread_send_msg(qpair_ctx->thread, qpair_ctx->cb_fn, qpair_ctx->ctx);
 		}
@@ -693,22 +714,9 @@ _spdk_nvmf_qpair_destroy(void *ctx, int status)
 		return;
 	}
 
-	pthread_mutex_lock(&ctrlr->mtx);
+	qpair_ctx->ctrlr = ctrlr;
-	spdk_bit_array_clear(ctrlr->qpair_mask, qid);
+	spdk_thread_send_msg(thread, _spdk_nvmf_ctrlr_free_from_qpair, qpair_ctx);
-	count = spdk_bit_array_count_set(ctrlr->qpair_mask);
-	pthread_mutex_unlock(&ctrlr->mtx);
 
-	if (count == 0) {
-		/* If this was the last queue pair on the controller, also send a message
-		 * to the subsystem to remove the controller. */
-		qpair_ctx->ctrlr = ctrlr;
-		spdk_thread_send_msg(ctrlr->subsys->thread, _spdk_nvmf_ctrlr_free_from_qpair, qpair_ctx);
-	} else {
-		if (qpair_ctx->cb_fn) {
-			spdk_thread_send_msg(qpair_ctx->thread, qpair_ctx->cb_fn, qpair_ctx->ctx);
-		}
-		free(qpair_ctx);
-	}
 }
 
 static void

lib/nvmf/nvmf_internal.h

@@ -211,8 +211,6 @@ struct spdk_nvmf_ctrlr {
 
 	struct spdk_nvmf_qpair *admin_qpair;
 
-	/* Mutex to protect the qpair mask */
-	pthread_mutex_t		mtx;
 	struct spdk_bit_array	*qpair_mask;
 
 	struct spdk_nvmf_request *aer_req;