lib/ftl: split IOs based on the number of iovecs

Split the IO in case io->iov_cnt exceeds the FTL_IO_MAX_IOVEC. It allows
the user to pass any number of iovecs. Each IO request is now split if
necessary and submitted via ftl_io_call_foreach_child submitting each
child individually. Any resubmissions need to be handled on a child
level, as parent request doesn't track its children submission status.

Change-Id: If54249d54225b34191216a366c227f79215abc90
Signed-off-by: Konrad Sztyber <konrad.sztyber@intel.com>
Reviewed-on: https://review.gerrithub.io/c/spdk/spdk/+/455527
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: Darek Stojaczyk <dariusz.stojaczyk@intel.com>
Reviewed-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>

lib/ftl/ftl_core.c

@@ -801,6 +801,8 @@ ftl_submit_read(struct ftl_io *io)
 	struct ftl_ppa ppa;
 	int rc = 0, lbk_cnt;
 
+	assert(LIST_EMPTY(&io->children));
+
 	while (io->pos < io->lbk_cnt) {
 		if (ftl_io_mode_ppa(io)) {
 			lbk_cnt = rc = ftl_ppa_read_next_ppa(io, &ppa);
@@ -1579,6 +1581,35 @@ _ftl_io_write(void *ctx)
 	ftl_io_write((struct ftl_io *)ctx);
 }
 
+static int
+ftl_rwb_fill_leaf(struct ftl_io *io)
+{
+	int rc;
+
+	rc = ftl_rwb_fill(io);
+	if (rc == -EAGAIN) {
+		spdk_thread_send_msg(spdk_io_channel_get_thread(io->ioch),
+				     _ftl_io_write, io);
+		return 0;
+	}
+
+	return rc;
+}
+
+static int
+ftl_submit_write_leaf(struct ftl_io *io)
+{
+	int rc;
+
+	rc = ftl_submit_write(ftl_wptr_from_band(io->band), io);
+	if (rc == -EAGAIN) {
+		/* EAGAIN means that the request was put on the pending queue */
+		return 0;
+	}
+
+	return rc;
+}
+
 void
 ftl_io_write(struct ftl_io *io)
 {
@@ -1586,24 +1617,17 @@ ftl_io_write(struct ftl_io *io)
 
 	/* For normal IOs we just need to copy the data onto the rwb */
 	if (!(io->flags & FTL_IO_MD)) {
-		/* Other errors should be handled by ftl_rwb_fill */
+		ftl_io_call_foreach_child(io, ftl_rwb_fill_leaf);
-		if (ftl_rwb_fill(io) == -EAGAIN) {
+	} else {
-			spdk_thread_send_msg(spdk_get_thread(), _ftl_io_write, io);
-		}
-
-		return;
-	}
-
 		/* Metadata has its own buffer, so it doesn't have to be copied, so just */
 		/* send it the the core thread and schedule the write immediately */
 		if (ftl_check_core_thread(dev)) {
-		/* We don't care about the errors, as the IO is either retried or completed
+			ftl_io_call_foreach_child(io, ftl_submit_write_leaf);
-		 * internally by ftl_submit_write */
-		ftl_submit_write(ftl_wptr_from_band(io->band), io);
 		} else {
 			spdk_thread_send_msg(ftl_get_core_thread(dev), _ftl_io_write, io);
 		}
 	}
+}
 
 int
 spdk_ftl_write(struct spdk_ftl_dev *dev, struct spdk_io_channel *ch, uint64_t lba, size_t lba_cnt,
@@ -1611,7 +1635,7 @@ spdk_ftl_write(struct spdk_ftl_dev *dev, struct spdk_io_channel *ch, uint64_t lb
 {
 	struct ftl_io *io;
 
-	if (iov_cnt == 0 || iov_cnt > FTL_IO_MAX_IOVEC) {
+	if (iov_cnt == 0) {
 		return -EINVAL;
 	}
 
@@ -1637,6 +1661,20 @@ spdk_ftl_write(struct spdk_ftl_dev *dev, struct spdk_io_channel *ch, uint64_t lb
 	return 0;
 }
 
+static int
+ftl_io_read_leaf(struct ftl_io *io)
+{
+	int rc;
+
+	rc = ftl_submit_read(io);
+	if (rc == -ENOMEM) {
+		/* ENOMEM means that the request was put on a pending queue */
+		return 0;
+	}
+
+	return rc;
+}
+
 static void
 _ftl_io_read(void *arg)
 {
@@ -1649,9 +1687,7 @@ ftl_io_read(struct ftl_io *io)
 	struct spdk_ftl_dev *dev = io->dev;
 
 	if (ftl_check_read_thread(dev)) {
-		/* We don't care about the errors, as the IO is either retried or completed
+		ftl_io_call_foreach_child(io, ftl_io_read_leaf);
-		 * internally by ftl_submit_read */
-		ftl_submit_read(io);
 	} else {
 		spdk_thread_send_msg(ftl_get_read_thread(dev), _ftl_io_read, io);
 	}
@@ -1663,7 +1699,7 @@ spdk_ftl_read(struct spdk_ftl_dev *dev, struct spdk_io_channel *ch, uint64_t lba
 {
 	struct ftl_io *io;
 
-	if (iov_cnt == 0 || iov_cnt > FTL_IO_MAX_IOVEC) {
+	if (iov_cnt == 0) {
 		return -EINVAL;
 	}
 

lib/ftl/ftl_io.c

@@ -34,6 +34,7 @@
 #include "spdk/stdinc.h"
 #include "spdk/ftl.h"
 #include "spdk/likely.h"
+#include "spdk/util.h"
 
 #include "ftl_io.h"
 #include "ftl_core.h"
@@ -156,24 +157,101 @@ ftl_io_iovec_len_left(struct ftl_io *io)
 }
 
 static void
-ftl_io_init_iovec(struct ftl_io *io, void *buf, size_t lbk_cnt)
+_ftl_io_init_iovec(struct ftl_io *io, const struct iovec *iov, size_t iov_cnt, size_t lbk_cnt)
 {
-	io->iov_pos = 0;
+	size_t iov_off;
-	io->lbk_cnt = lbk_cnt;
-	io->iov_cnt = 1;
 
-	io->iov[0].iov_base = buf;
+	io->iov_pos = 0;
-	io->iov[0].iov_len = lbk_cnt * PAGE_SIZE;
+	io->iov_cnt = iov_cnt;
+	io->lbk_cnt = lbk_cnt;
+
+	memcpy(io->iov, iov, iov_cnt * sizeof(*iov));
+
+	if (lbk_cnt == 0) {
+		for (iov_off = 0; iov_off < iov_cnt; ++iov_off) {
+			io->lbk_cnt += iov[iov_off].iov_len / PAGE_SIZE;
+		}
+	}
+}
+
+static void _ftl_io_free(struct ftl_io *io);
+
+static int
+ftl_io_add_child(struct ftl_io *io, const struct iovec *iov, size_t iov_cnt)
+{
+	struct ftl_io *child;
+
+	child = ftl_io_alloc_child(io);
+	if (spdk_unlikely(!child)) {
+		return -ENOMEM;
+	}
+
+	_ftl_io_init_iovec(child, iov, iov_cnt, 0);
+
+	if (io->flags & FTL_IO_VECTOR_LBA) {
+		child->lba.vector = io->lba.vector + io->lbk_cnt;
+	} else {
+		child->lba.single = io->lba.single + io->lbk_cnt;
+	}
+
+	io->lbk_cnt += child->lbk_cnt;
+	return 0;
+}
+
+static int
+ftl_io_init_iovec(struct ftl_io *io, const struct iovec *iov, size_t iov_cnt, size_t lbk_cnt)
+{
+	struct ftl_io *child;
+	size_t iov_off = 0, iov_left;
+	int rc;
+
+	if (spdk_likely(iov_cnt <= FTL_IO_MAX_IOVEC)) {
+		_ftl_io_init_iovec(io, iov, iov_cnt, lbk_cnt);
+		return 0;
+	}
+
+	while (iov_off < iov_cnt) {
+		iov_left = spdk_min(iov_cnt - iov_off, FTL_IO_MAX_IOVEC);
+
+		rc = ftl_io_add_child(io, &iov[iov_off], iov_left);
+		if (spdk_unlikely(rc != 0)) {
+			while ((child = LIST_FIRST(&io->children))) {
+				assert(LIST_EMPTY(&child->children));
+				LIST_REMOVE(child, child_entry);
+				_ftl_io_free(child);
+			}
+
+			return -ENOMEM;
+		}
+
+		iov_off += iov_left;
+	}
+
+	assert(io->lbk_cnt == lbk_cnt);
+	return 0;
 }
 
 void
 ftl_io_shrink_iovec(struct ftl_io *io, size_t lbk_cnt)
 {
-	assert(io->iov_cnt == 1);
+	size_t iov_off = 0, lbk_off = 0;
+
 	assert(io->lbk_cnt >= lbk_cnt);
 	assert(io->pos == 0 && io->iov_pos == 0 && io->iov_off == 0);
 
-	ftl_io_init_iovec(io, ftl_io_iovec_addr(io), lbk_cnt);
+	for (; iov_off < io->iov_cnt; ++iov_off) {
+		size_t num_iov = io->iov[iov_off].iov_len / PAGE_SIZE;
+		size_t num_left = lbk_cnt - lbk_off;
+
+		if (num_iov >= num_left) {
+			io->iov[iov_off].iov_len = num_left * PAGE_SIZE;
+			io->iov_cnt = iov_off + 1;
+			io->lbk_cnt = lbk_cnt;
+			break;
+		}
+
+		lbk_off += num_iov;
+	}
 }
 
 static void
@@ -195,6 +273,10 @@ ftl_io_init_internal(const struct ftl_io_init_opts *opts)
 {
 	struct ftl_io *io = opts->io;
 	struct spdk_ftl_dev *dev = opts->dev;
+	struct iovec iov = {
+		.iov_base = opts->data,
+		.iov_len  = opts->lbk_cnt * PAGE_SIZE
+	};
 
 	if (!io) {
 		if (opts->parent) {
@@ -215,7 +297,12 @@ ftl_io_init_internal(const struct ftl_io_init_opts *opts)
 	io->band = opts->band;
 	io->md = opts->md;
 
-	ftl_io_init_iovec(io, opts->data, opts->lbk_cnt);
+	if (ftl_io_init_iovec(io, &iov, 1, opts->lbk_cnt)) {
+		if (!opts->io) {
+			ftl_io_free(io);
+		}
+		return NULL;
+	}
 
 	return io;
 }
@@ -283,16 +370,14 @@ ftl_io_user_init(struct spdk_io_channel *_ioch, uint64_t lba, size_t lbk_cnt, st
 	}
 
 	ftl_io_init(io, dev, cb_fn, cb_arg, 0, type);
-
 	io->lba.single = lba;
-	io->lbk_cnt = lbk_cnt;
-	io->iov_cnt = iov_cnt;
 
-	assert(iov_cnt < FTL_IO_MAX_IOVEC);
+	if (ftl_io_init_iovec(io, iov, iov_cnt, lbk_cnt)) {
-	memcpy(io->iov, iov, iov_cnt * sizeof(*iov));
+		ftl_io_free(io);
+		return NULL;
+	}
 
 	ftl_trace_lba_io_init(io->dev, io);
-
 	return io;
 }
 
@@ -364,6 +449,7 @@ ftl_io_alloc_child(struct ftl_io *parent)
 		return NULL;
 	}
 
+	ftl_io_init(io, parent->dev, NULL, NULL, parent->flags, parent->type);
 	io->parent = parent;
 
 	pthread_spin_lock(&parent->lock);