bdev: add bdev timeout support

Add an API so that the user can enable/disable the bdev IO
timeout. Also, add the bdev io timeout handling callback.
So it means to let the upper user determine how to handle
the IO timeout scenario reset the device or abort the IO.

Change-Id: I9c7138ca46c74c045b687adab59a18d6bccc4996
Signed-off-by: Jin Yu <jin.yu@intel.com>
Reviewed-on: https://review.gerrithub.io/c/spdk/spdk/+/469228
Reviewed-by: Jim Harris <james.r.harris@intel.com>
Reviewed-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>
Reviewed-by: Changpeng Liu <changpeng.liu@intel.com>
Reviewed-by: Alexey Marchuk <alexeymar@mellanox.com>
Community-CI: Broadcom SPDK FC-NVMe CI <spdk-ci.pdl@broadcom.com>
Community-CI: SPDK CI Jenkins <sys_sgci@intel.com>
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>

include/spdk/bdev.h

@@ -199,6 +199,14 @@ typedef void (*spdk_bdev_fini_cb)(void *cb_arg);
 typedef void (*spdk_bdev_get_device_stat_cb)(struct spdk_bdev *bdev,
 		struct spdk_bdev_io_stat *stat, void *cb_arg, int rc);
 
+/**
+ * Block device channel IO timeout callback
+ *
+ * \param cb_arg Callback argument
+ * \param bdev_io The IO cause the timeout
+ */
+typedef void (*spdk_bdev_io_timeout_cb)(void *cb_arg, struct spdk_bdev_io *bdev_io);
+
 /**
  * Initialize block device modules.
  *
@@ -326,6 +334,27 @@ void spdk_bdev_close(struct spdk_bdev_desc *desc);
  */
 struct spdk_bdev *spdk_bdev_desc_get_bdev(struct spdk_bdev_desc *desc);
 
+/**
+ * Set a time limit for the timeout IO of the bdev and timeout callback.
+ * We can use this function to enable/disable the timeout handler. If
+ * the timeout_in_sec > 0 then it means to enable the timeout IO handling
+ * or change the time limit. If the timeout_in_sec == 0 it means to
+ * disable the timeout IO handling. If you want to enable or change the
+ * timeout IO handle you need to specify the spdk_bdev_io_timeout_cb it
+ * means the upper user determines what to do if you meet the timeout IO,
+ * for example, you can reset the device or abort the IO.
+ * Note: This function must run in the desc's thread.
+ *
+ * \param desc Block device descriptor.
+ * \param timeout_in_sec Timeout value
+ * \param cb_fn Bdev IO timeout callback
+ * \param cb_arg Callback argument
+ *
+ * \return 0 on success, negated errno on failure.
+ */
+int spdk_bdev_set_timeout(struct spdk_bdev_desc *desc, uint64_t timeout_in_sec,
+			  spdk_bdev_io_timeout_cb cb_fn, void *cb_arg);
+
 /**
  * Check whether the block device supports the I/O type.
  *

lib/bdev/bdev.c

@@ -81,6 +81,7 @@ int __itt_init_ittlib(const char *, __itt_group_id);
 #define SPDK_BDEV_QOS_MIN_IOS_PER_SEC		1000
 #define SPDK_BDEV_QOS_MIN_BYTES_PER_SEC		(1024 * 1024)
 #define SPDK_BDEV_QOS_LIMIT_NOT_DEFINED		UINT64_MAX
+#define SPDK_BDEV_IO_POLL_INTERVAL_IN_MSEC	1000
 
 #define SPDK_BDEV_POOL_ALIGNMENT 512
 
@@ -293,6 +294,11 @@ struct spdk_bdev_desc {
 	pthread_mutex_t			mutex;
 	uint32_t			refs;
 	TAILQ_ENTRY(spdk_bdev_desc)	link;
+
+	uint64_t		timeout_in_sec;
+	spdk_bdev_io_timeout_cb	cb_fn;
+	void			*cb_arg;
+	struct spdk_poller	*io_timeout_poller;
 };
 
 struct spdk_bdev_iostat_ctx {
@@ -2119,6 +2125,133 @@ bdev_enable_qos(struct spdk_bdev *bdev, struct spdk_bdev_channel *ch)
 	}
 }
 
+struct poll_timeout_ctx {
+	struct spdk_bdev_desc	*desc;
+	uint64_t		timeout_in_sec;
+	spdk_bdev_io_timeout_cb	cb_fn;
+	void			*cb_arg;
+};
+
+static void
+bdev_desc_free(struct spdk_bdev_desc *desc)
+{
+	pthread_mutex_destroy(&desc->mutex);
+	free(desc);
+}
+
+static void
+bdev_channel_poll_timeout_io_done(struct spdk_io_channel_iter *i, int status)
+{
+	struct poll_timeout_ctx *ctx  = spdk_io_channel_iter_get_ctx(i);
+	struct spdk_bdev_desc *desc = ctx->desc;
+
+	free(ctx);
+
+	pthread_mutex_lock(&desc->mutex);
+	desc->refs--;
+	if (desc->closed == true && desc->refs == 0) {
+		pthread_mutex_unlock(&desc->mutex);
+		bdev_desc_free(desc);
+		return;
+	}
+	pthread_mutex_unlock(&desc->mutex);
+}
+
+static void
+bdev_channel_poll_timeout_io(struct spdk_io_channel_iter *i)
+{
+	struct poll_timeout_ctx *ctx  = spdk_io_channel_iter_get_ctx(i);
+	struct spdk_io_channel *io_ch = spdk_io_channel_iter_get_channel(i);
+	struct spdk_bdev_channel *bdev_ch = spdk_io_channel_get_ctx(io_ch);
+	struct spdk_bdev_desc *desc = ctx->desc;
+	struct spdk_bdev_io *bdev_io;
+	uint64_t now;
+
+	pthread_mutex_lock(&desc->mutex);
+	if (desc->closed == true) {
+		pthread_mutex_unlock(&desc->mutex);
+		spdk_for_each_channel_continue(i, -1);
+		return;
+	}
+	pthread_mutex_unlock(&desc->mutex);
+
+	now = spdk_get_ticks();
+	TAILQ_FOREACH(bdev_io, &bdev_ch->io_submitted, internal.ch_link) {
+		/* I/O are added to this TAILQ as they are submitted.
+		 * So once we find an I/O that has not timed out, we can immediately exit the loop. */
+		if (now < (bdev_io->internal.submit_tsc +
+			   ctx->timeout_in_sec * spdk_get_ticks_hz())) {
+			goto end;
+		}
+
+		if (bdev_io->internal.desc == desc) {
+			ctx->cb_fn(ctx->cb_arg, bdev_io);
+		}
+	}
+
+end:
+	spdk_for_each_channel_continue(i, 0);
+}
+
+static int
+bdev_poll_timeout_io(void *arg)
+{
+	struct spdk_bdev_desc *desc = arg;
+	struct spdk_bdev *bdev = spdk_bdev_desc_get_bdev(desc);
+	struct poll_timeout_ctx *ctx;
+
+	ctx = calloc(1, sizeof(struct poll_timeout_ctx));
+	if (!ctx) {
+		SPDK_ERRLOG("failed to allocate memory\n");
+		return 1;
+	}
+	ctx->desc = desc;
+	ctx->cb_arg = desc->cb_arg;
+	ctx->cb_fn = desc->cb_fn;
+	ctx->timeout_in_sec = desc->timeout_in_sec;
+
+	/* Take a ref on the descriptor in case it gets closed while we are checking
+	 * all of the channels.
+	 */
+	pthread_mutex_lock(&desc->mutex);
+	desc->refs++;
+	pthread_mutex_unlock(&desc->mutex);
+
+	spdk_for_each_channel(__bdev_to_io_dev(bdev),
+			      bdev_channel_poll_timeout_io,
+			      ctx,
+			      bdev_channel_poll_timeout_io_done);
+
+	return 1;
+}
+
+int
+spdk_bdev_set_timeout(struct spdk_bdev_desc *desc, uint64_t timeout_in_sec,
+		      spdk_bdev_io_timeout_cb cb_fn, void *cb_arg)
+{
+	assert(desc->thread == spdk_get_thread());
+
+	spdk_poller_unregister(&desc->io_timeout_poller);
+
+	if (timeout_in_sec) {
+		assert(cb_fn != NULL);
+		desc->io_timeout_poller = spdk_poller_register(bdev_poll_timeout_io,
+					  desc,
+					  SPDK_BDEV_IO_POLL_INTERVAL_IN_MSEC * SPDK_SEC_TO_USEC /
+					  1000);
+		if (desc->io_timeout_poller == NULL) {
+			SPDK_ERRLOG("can not register the desc timeout IO poller\n");
+			return -1;
+		}
+	}
+
+	desc->cb_fn = cb_fn;
+	desc->cb_arg = cb_arg;
+	desc->timeout_in_sec = timeout_in_sec;
+
+	return 0;
+}
+
 static int
 bdev_channel_create(void *io_device, void *ctx_buf)
 {
@@ -2689,13 +2822,6 @@ spdk_bdev_set_qd_sampling_period(struct spdk_bdev *bdev, uint64_t period)
 	}
 }
 
-static void
-bdev_desc_free(struct spdk_bdev_desc *desc)
-{
-	pthread_mutex_destroy(&desc->mutex);
-	free(desc);
-}
-
 static void
 _resize_notify(void *arg)
 {
@@ -4559,6 +4685,8 @@ spdk_bdev_close(struct spdk_bdev_desc *desc)
 
 	assert(desc->thread == spdk_get_thread());
 
+	spdk_poller_unregister(&desc->io_timeout_poller);
+
 	pthread_mutex_lock(&bdev->internal.mutex);
 	pthread_mutex_lock(&desc->mutex);