nvmf: Make spdk_nvmf_subsystem_add_listener asynchronous

In the next patch a new transport call will be added to notify
transports of subsystem->listener associations. The operations the
transports may need to perform there are likely asynchronous, so make
this top level call asynchronous here in preparation.

Change-Id: I7674f56dc3ec0d127ce1026f980d436b4269cb56
Signed-off-by: Ben Walker <benjamin.walker@intel.com>
Reviewed-on: https://review.spdk.io/gerrit/c/spdk/spdk/+/628
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: Changpeng Liu <changpeng.liu@intel.com>
Reviewed-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>
Reviewed-by: Aleksey Marchuk <alexeymar@mellanox.com>

include/spdk/nvmf.h

@@ -122,6 +122,14 @@ struct spdk_nvmf_transport_poll_group_stat {
  */
 typedef void (*new_qpair_fn)(struct spdk_nvmf_qpair *qpair, void *cb_arg);
 
+/**
+ * Function to be called once the listener is associated with a subsystem.
+ *
+ * \param ctx Context argument passed to this function.
+ * \param status 0 if it completed successfully, or negative errno if it failed.
+ */
+typedef void (*spdk_nvmf_tgt_subsystem_listen_done_fn)(void *ctx, int status);
+
 /**
  * Construct an NVMe-oF target.
  *
@@ -556,11 +564,13 @@ const char *spdk_nvmf_host_get_nqn(const struct spdk_nvmf_host *host);
  *
  * \param subsystem Subsystem to add listener to.
  * \param trid The address to accept connections from.
- *
+ * \param cb_fn A callback that will be called once the association is complete.
- * \return 0 on success, or negated errno value on failure.
+ * \param cb_arg Argument passed to cb_fn.
  */
-int spdk_nvmf_subsystem_add_listener(struct spdk_nvmf_subsystem *subsystem,
+void spdk_nvmf_subsystem_add_listener(struct spdk_nvmf_subsystem *subsystem,
-				     struct spdk_nvme_transport_id *trid);
+				      struct spdk_nvme_transport_id *trid,
+				      spdk_nvmf_tgt_subsystem_listen_done_fn cb_fn,
+				      void *cb_arg);
 
 /**
  * Remove the listener from subsystem.

lib/nvmf/fc.c

@@ -2956,6 +2956,7 @@ out:
 }
 
 struct nvmf_fc_add_rem_listener_ctx {
+	struct spdk_nvmf_subsystem *subsystem;
 	bool add_listener;
 	struct spdk_nvme_transport_id trid;
 };
@@ -2968,6 +2969,18 @@ nvmf_fc_adm_subsystem_resume_cb(struct spdk_nvmf_subsystem *subsystem, void *cb_
 	free(ctx);
 }
 
+static void
+nvmf_fc_adm_listen_done(void *cb_arg, int status)
+{
+	ASSERT_SPDK_FC_MASTER_THREAD();
+	struct nvmf_fc_add_rem_listener_ctx *ctx = cb_arg;
+
+	if (spdk_nvmf_subsystem_resume(ctx->subsystem, nvmf_fc_adm_subsystem_resume_cb, ctx)) {
+		SPDK_ERRLOG("Failed to resume subsystem: %s\n", subsystem->subnqn);
+		free(ctx);
+	}
+}
+
 static void
 nvmf_fc_adm_subsystem_paused_cb(struct spdk_nvmf_subsystem *subsystem, void *cb_arg, int status)
 {
@@ -2975,13 +2988,10 @@ nvmf_fc_adm_subsystem_paused_cb(struct spdk_nvmf_subsystem *subsystem, void *cb_
 	struct nvmf_fc_add_rem_listener_ctx *ctx = (struct nvmf_fc_add_rem_listener_ctx *)cb_arg;
 
 	if (ctx->add_listener) {
-		spdk_nvmf_subsystem_add_listener(subsystem, &ctx->trid);
+		spdk_nvmf_subsystem_add_listener(subsystem, &ctx->trid, nvmf_fc_adm_listen_done, ctx);
 	} else {
 		spdk_nvmf_subsystem_remove_listener(subsystem, &ctx->trid);
-	}
+		nvmf_fc_adm_listen_done(ctx, 0);
-	if (spdk_nvmf_subsystem_resume(subsystem, nvmf_fc_adm_subsystem_resume_cb, ctx)) {
-		SPDK_ERRLOG("Failed to resume subsystem: %s\n", subsystem->subnqn);
-		free(ctx);
 	}
 }
 
@@ -3004,6 +3014,7 @@ nvmf_fc_adm_add_rem_nport_listener(struct spdk_nvmf_fc_nport *nport, bool add)
 			ctx = calloc(1, sizeof(struct nvmf_fc_add_rem_listener_ctx));
 			if (ctx) {
 				ctx->add_listener = add;
+				ctx->subsystem = subsystem;
 				spdk_nvmf_fc_create_trid(&ctx->trid,
 							 nport->fc_nodename.u.wwn,
 							 nport->fc_portname.u.wwn);

lib/nvmf/nvmf_rpc.c

@@ -633,6 +633,30 @@ nvmf_rpc_listen_resumed(struct spdk_nvmf_subsystem *subsystem,
 	spdk_jsonrpc_end_result(request, w);
 }
 
+static void
+nvmf_rpc_subsystem_listen(void *cb_arg, int status)
+{
+	struct nvmf_rpc_listener_ctx *ctx = cb_arg;
+
+	if (status) {
+		/* Destroy the listener that we just created. Ignore the error code because
+		 * the RPC is failing already anyway. */
+		spdk_nvmf_tgt_stop_listen(ctx->tgt, &ctx->trid);
+
+		spdk_jsonrpc_send_error_response(ctx->request, SPDK_JSONRPC_ERROR_INVALID_PARAMS,
+						 "Invalid parameters");
+		ctx->response_sent = true;
+	}
+
+	if (spdk_nvmf_subsystem_resume(ctx->subsystem, nvmf_rpc_listen_resumed, ctx)) {
+		if (!ctx->response_sent) {
+			spdk_jsonrpc_send_error_response(ctx->request, SPDK_JSONRPC_ERROR_INTERNAL_ERROR, "Internal error");
+		}
+		nvmf_rpc_listener_ctx_free(ctx);
+		/* Can't really do anything to recover here - subsystem will remain paused. */
+	}
+}
+
 static void
 nvmf_rpc_listen_paused(struct spdk_nvmf_subsystem *subsystem,
 		       void *cb_arg, int status)
@@ -644,16 +668,13 @@ nvmf_rpc_listen_paused(struct spdk_nvmf_subsystem *subsystem,
 		if (!spdk_nvmf_subsystem_find_listener(subsystem, &ctx->trid)) {
 			rc = spdk_nvmf_tgt_listen(ctx->tgt, &ctx->trid);
 			if (rc == 0) {
-				if (spdk_nvmf_subsystem_add_listener(ctx->subsystem, &ctx->trid)) {
+				spdk_nvmf_subsystem_add_listener(ctx->subsystem, &ctx->trid, nvmf_rpc_subsystem_listen, ctx);
-					spdk_jsonrpc_send_error_response(ctx->request, SPDK_JSONRPC_ERROR_INVALID_PARAMS,
+				return;
-									 "Invalid parameters");
-					ctx->response_sent = true;
-				}
-			} else {
-				spdk_jsonrpc_send_error_response(ctx->request, SPDK_JSONRPC_ERROR_INVALID_PARAMS,
-								 "Invalid parameters");
-				ctx->response_sent = true;
 			}
+
+			spdk_jsonrpc_send_error_response(ctx->request, SPDK_JSONRPC_ERROR_INVALID_PARAMS,
+							 "Invalid parameters");
+			ctx->response_sent = true;
 		}
 	} else if (ctx->op == NVMF_RPC_LISTEN_REMOVE) {
 		if (spdk_nvmf_subsystem_remove_listener(subsystem, &ctx->trid)) {

lib/nvmf/subsystem.c

@@ -747,39 +747,48 @@ spdk_nvmf_subsystem_find_listener(struct spdk_nvmf_subsystem *subsystem,
 	return NULL;
 }
 
-int
+void
 spdk_nvmf_subsystem_add_listener(struct spdk_nvmf_subsystem *subsystem,
-				 struct spdk_nvme_transport_id *trid)
+				 struct spdk_nvme_transport_id *trid,
+				 spdk_nvmf_tgt_subsystem_listen_done_fn cb_fn,
+				 void *cb_arg)
 {
 	struct spdk_nvmf_transport *transport;
 	struct spdk_nvmf_subsystem_listener *listener;
 	struct spdk_nvmf_listener *tr_listener;
 
+	assert(cb_fn != NULL);
+
 	if (!(subsystem->state == SPDK_NVMF_SUBSYSTEM_INACTIVE ||
 	      subsystem->state == SPDK_NVMF_SUBSYSTEM_PAUSED)) {
-		return -EAGAIN;
+		cb_fn(cb_arg, -EAGAIN);
+		return;
 	}
 
 	if (spdk_nvmf_subsystem_find_listener(subsystem, trid)) {
 		/* Listener already exists in this subsystem */
-		return 0;
+		cb_fn(cb_arg, 0);
+		return;
 	}
 
 	transport = spdk_nvmf_tgt_get_transport(subsystem->tgt, trid->trstring);
 	if (transport == NULL) {
 		SPDK_ERRLOG("Unknown transport type %d\n", trid->trtype);
-		return -EINVAL;
+		cb_fn(cb_arg, -EINVAL);
+		return;
 	}
 
 	tr_listener = spdk_nvmf_transport_find_listener(transport, trid);
 	if (!tr_listener) {
 		SPDK_ERRLOG("Cannot find transport listener for %s\n", trid->traddr);
-		return -EINVAL;
+		cb_fn(cb_arg, -EINVAL);
+		return;
 	}
 
 	listener = calloc(1, sizeof(*listener));
 	if (!listener) {
-		return -ENOMEM;
+		cb_fn(cb_arg, -ENOMEM);
+		return;
 	}
 
 	listener->trid = &tr_listener->trid;
@@ -788,7 +797,7 @@ spdk_nvmf_subsystem_add_listener(struct spdk_nvmf_subsystem *subsystem,
 	TAILQ_INSERT_HEAD(&subsystem->listeners, listener, link);
 	subsystem->tgt->discovery_genctr++;
 
-	return 0;
+	cb_fn(cb_arg, 0);
 }
 
 int

module/event/subsystems/nvmf/conf.c

@@ -281,6 +281,16 @@ spdk_nvmf_tgt_parse_listen_fc_addr(const char *address,
 	return 0;
 }
 
+static void
+spdk_nvmf_tgt_listen_done(void *cb_arg, int status)
+{
+	/* TODO: Config parsing should wait for this operation to finish. */
+
+	if (status) {
+		SPDK_ERRLOG("Failed to listen on transport address\n");
+	}
+}
+
 static int
 spdk_nvmf_parse_subsystem(struct spdk_conf_section *sp)
 {
@@ -476,7 +486,7 @@ spdk_nvmf_parse_subsystem(struct spdk_conf_section *sp)
 			SPDK_ERRLOG("Failed to listen on transport address\n");
 		}
 
-		spdk_nvmf_subsystem_add_listener(subsystem, &trid);
+		spdk_nvmf_subsystem_add_listener(subsystem, &trid, spdk_nvmf_tgt_listen_done, NULL);
 		allow_any_listener = false;
 	}
 

test/unit/lib/nvmf/ctrlr_discovery.c/ctrlr_discovery_ut.c

@@ -196,6 +196,12 @@ spdk_nvmf_poll_group_resume_subsystem(struct spdk_nvmf_poll_group *group,
 {
 }
 
+static void
+_subsystem_add_listen_done(void *cb_arg, int status)
+{
+	SPDK_CU_ASSERT_FATAL(status == 0);
+}
+
 static void
 test_discovery_log(void)
 {
@@ -224,7 +230,7 @@ test_discovery_log(void)
 	trid.adrfam = SPDK_NVMF_ADRFAM_IPV4;
 	snprintf(trid.traddr, sizeof(trid.traddr), "1234");
 	snprintf(trid.trsvcid, sizeof(trid.trsvcid), "5678");
-	SPDK_CU_ASSERT_FATAL(spdk_nvmf_subsystem_add_listener(subsystem, &trid) == 0);
+	spdk_nvmf_subsystem_add_listener(subsystem, &trid, _subsystem_add_listen_done, NULL);
 	subsystem->state = SPDK_NVMF_SUBSYSTEM_ACTIVE;
 
 	/* Get only genctr (first field in the header) */