io_channel: change return type of spdk_channel_msg

Change the return type of spdk_channel_msg from
void to int. If this msg executed in a failure,
we do not need continue sending the message to other
threads, we can just tell the original thread, and
let the orgiginal thread call the spdk_channel_for_each_cpl
call back.

Thus we can track the qpair creation/destroy case for bdev
reset in nvme bdev module;

Change-Id: Ide9dffd1f84a29fcf61d8339a9ece2a0245d968d
Signed-off-by: Ziye Yang <optimistyzy@gmail.com>
Reviewed-on: https://review.gerrithub.io/387284
Tested-by: SPDK Automated Test System <sys_sgsw@intel.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>
Reviewed-by: Jim Harris <james.r.harris@intel.com>

include/spdk/io_channel.h

@@ -54,9 +54,9 @@ typedef void (*spdk_io_channel_destroy_cb)(void *io_device, void *ctx_buf);
 
 typedef void (*spdk_io_device_unregister_cb)(void *io_device);
 
-typedef void (*spdk_channel_msg)(void *io_device, struct spdk_io_channel *ch,
-				 void *ctx);
-typedef void (*spdk_channel_for_each_cpl)(void *io_device, void *ctx);
+typedef int (*spdk_channel_msg)(void *io_device, struct spdk_io_channel *ch,
+				void *ctx);
+typedef void (*spdk_channel_for_each_cpl)(void *io_device, void *ctx, int status);
 
 /**
  * \brief Initializes the calling thread for I/O channel allocation.

lib/bdev/bdev.c

@@ -1336,7 +1336,7 @@ spdk_bdev_flush_blocks(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch,
 }
 
 static void
-_spdk_bdev_reset_dev(void *io_device, void *ctx)
+_spdk_bdev_reset_dev(void *io_device, void *ctx, int status)
 {
 	struct spdk_bdev_channel *ch = ctx;
 	struct spdk_bdev_io *bdev_io;
@@ -1346,7 +1346,7 @@ _spdk_bdev_reset_dev(void *io_device, void *ctx)
 	spdk_bdev_io_submit_reset(bdev_io);
 }
 
-static void
+static int
 _spdk_bdev_reset_abort_channel(void *io_device, struct spdk_io_channel *ch,
 			       void *ctx)
 {
@@ -1361,6 +1361,8 @@ _spdk_bdev_reset_abort_channel(void *io_device, struct spdk_io_channel *ch,
 	_spdk_bdev_abort_queued_io(&channel->nomem_io, channel);
 	_spdk_bdev_abort_buf_io(&mgmt_channel->need_buf_small, channel);
 	_spdk_bdev_abort_buf_io(&mgmt_channel->need_buf_large, channel);
+
+	return 0;
 }
 
 static void
@@ -1395,7 +1397,7 @@ _spdk_bdev_channel_start_reset(struct spdk_bdev_channel *ch)
 	pthread_mutex_unlock(&bdev->mutex);
 }
 
-static void
+static int
 _spdk_bdev_complete_reset_channel(void *io_device, struct spdk_io_channel *_ch, void *ctx)
 {
 	struct spdk_bdev_channel *ch = spdk_io_channel_get_ctx(_ch);
@@ -1404,6 +1406,8 @@ _spdk_bdev_complete_reset_channel(void *io_device, struct spdk_io_channel *_ch,
 	if (!TAILQ_EMPTY(&ch->queued_resets)) {
 		_spdk_bdev_channel_start_reset(ch);
 	}
+
+	return 0;
 }
 
 int

lib/bdev/nvme/bdev_nvme.c

@@ -267,12 +267,16 @@ bdev_nvme_flush(struct nvme_bdev *nbdev, struct nvme_bdev_io *bio,
 }
 
 static void
-_bdev_nvme_reset_done(void *io_device, void *ctx)
+_bdev_nvme_reset_done(void *io_device, void *ctx, int status)
 {
-	spdk_bdev_io_complete(spdk_bdev_io_from_ctx(ctx), SPDK_BDEV_IO_STATUS_SUCCESS);
+	int rc = SPDK_BDEV_IO_STATUS_SUCCESS;
+	if (status) {
+		rc = SPDK_BDEV_IO_STATUS_FAILED;
+	}
+	spdk_bdev_io_complete(spdk_bdev_io_from_ctx(ctx), rc);
 }
 
-static void
+static int
 _bdev_nvme_reset_create_qpair(void *io_device, struct spdk_io_channel *ch,
 			      void *ctx)
 {
@@ -280,16 +284,25 @@ _bdev_nvme_reset_create_qpair(void *io_device, struct spdk_io_channel *ch,
 	struct nvme_io_channel *nvme_ch = spdk_io_channel_get_ctx(ch);
 
 	nvme_ch->qpair = spdk_nvme_ctrlr_alloc_io_qpair(ctrlr, NULL, 0);
-	assert(nvme_ch->qpair != NULL); /* Currently, no good way to handle this error */
+	if (!nvme_ch->qpair) {
+		return -1;
+	}
+
+	return 0;
 }
 
 static void
-_bdev_nvme_reset(void *io_device, void *ctx)
+_bdev_nvme_reset(void *io_device, void *ctx, int status)
 {
 	struct spdk_nvme_ctrlr *ctrlr = io_device;
 	struct nvme_bdev_io *bio = ctx;
 	int rc;
 
+	if (status) {
+		spdk_bdev_io_complete(spdk_bdev_io_from_ctx(bio), SPDK_BDEV_IO_STATUS_FAILED);
+		return;
+	}
+
 	rc = spdk_nvme_ctrlr_reset(ctrlr);
 	if (rc != 0) {
 		spdk_bdev_io_complete(spdk_bdev_io_from_ctx(bio), SPDK_BDEV_IO_STATUS_FAILED);
@@ -305,14 +318,19 @@ _bdev_nvme_reset(void *io_device, void *ctx)
 
 }
 
-static void
+static int
 _bdev_nvme_reset_destroy_qpair(void *io_device, struct spdk_io_channel *ch,
 			       void *ctx)
 {
 	struct nvme_io_channel *nvme_ch = spdk_io_channel_get_ctx(ch);
+	int rc;
 
-	spdk_nvme_ctrlr_free_io_qpair(nvme_ch->qpair);
-	nvme_ch->qpair = NULL;
+	rc = spdk_nvme_ctrlr_free_io_qpair(nvme_ch->qpair);
+	if (!rc) {
+		nvme_ch->qpair = NULL;
+	}
+
+	return rc;
 }
 
 static int

lib/util/io_channel.c

@@ -416,6 +416,7 @@ struct call_channel {
 	void *io_device;
 	struct io_device *dev;
 	spdk_channel_msg fn;
+	int status;
 	void *ctx;
 
 	struct spdk_thread *cur_thread;
@@ -430,7 +431,7 @@ _call_completion(void *ctx)
 	struct call_channel *ch_ctx = ctx;
 
 	if (ch_ctx->cpl != NULL) {
-		ch_ctx->cpl(ch_ctx->io_device, ch_ctx->ctx);
+		ch_ctx->cpl(ch_ctx->io_device, ch_ctx->ctx, ch_ctx->status);
 	}
 	free(ch_ctx);
 }
@@ -457,10 +458,13 @@ _call_channel(void *ctx)
 	 *  the fn() on this thread.
 	 */
 	if (ch != NULL) {
-		ch_ctx->fn(ch_ctx->io_device, ch, ch_ctx->ctx);
+		ch_ctx->status = ch_ctx->fn(ch_ctx->io_device, ch, ch_ctx->ctx);
 	}
 
 	pthread_mutex_lock(&g_devlist_mutex);
+	if (ch_ctx->status) {
+		goto end;
+	}
 	thread = TAILQ_NEXT(thread, tailq);
 	while (thread) {
 		TAILQ_FOREACH(ch, &thread->io_channels, tailq) {
@@ -474,6 +478,7 @@ _call_channel(void *ctx)
 		thread = TAILQ_NEXT(thread, tailq);
 	}
 
+end:
 	ch_ctx->dev->for_each_count--;
 	pthread_mutex_unlock(&g_devlist_mutex);
 
@@ -519,5 +524,5 @@ spdk_for_each_channel(void *io_device, spdk_channel_msg fn, void *ctx,
 
 	pthread_mutex_unlock(&g_devlist_mutex);
 
-	cpl(io_device, ctx);
+	cpl(io_device, ctx, 0);
 }

test/unit/lib/util/io_channel.c/io_channel_ut.c

@@ -108,16 +108,18 @@ channel_destroy(void *io_device, void *ctx_buf)
 {
 }
 
-static void
+static int
 channel_msg(void *io_device, struct spdk_io_channel *ch, void *ctx)
 {
 	int *count = spdk_io_channel_get_ctx(ch);
 
 	(*count)++;
+
+	return 0;
 }
 
 static void
-channel_cpl(void *io_device, void *ctx)
+channel_cpl(void *io_device, void *ctx, int status)
 {
 }
 
@@ -175,16 +177,18 @@ struct unreg_ctx {
 	bool	foreach_done;
 };
 
-static void
+static int
 unreg_ch_done(void *io_device, struct spdk_io_channel *_ch, void *_ctx)
 {
 	struct unreg_ctx *ctx = _ctx;
 
 	ctx->ch_done = true;
+
+	return 0;
 }
 
 static void
-unreg_foreach_done(void *io_device, void *_ctx)
+unreg_foreach_done(void *io_device, void *_ctx, int status)
 {
 	struct unreg_ctx *ctx = _ctx;