lib/bdev: Add support for new per bdev_io auxiliary buffer.

A new API was added `spdk_bdev_io_get_aux_buf` allowing the caller to request
an auxiliary buffer for its own private use. The API is used in the same manner that
`spdk_bdev_io_get_buf` is used and the length of the buffer is always the same as the
bdev_io primary buffer.  'spdk_bdev_io_put_aux_buf' is called to free the
auxiliary buffer.

The initial use case is crypto, in the next patch in series it is used. No UT were
added as the logic isn't that complicated and it is fully tested with each run
of crypto.

Fixed a comment typo also (not mine for once).

Signed-off-by: paul luse <paul.e.luse@intel.com>
Signed-off-by: Jim Harris <james.r.harris@intel.com>
Change-Id: Ib1939fcbc8e5db36fd909ef26771a725a551e8e6
Reviewed-on: https://review.gerrithub.io/c/spdk/spdk/+/478383
Community-CI: SPDK CI Jenkins <sys_sgci@intel.com>
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>

CHANGELOG.md

@@ -36,6 +36,14 @@ specified during NVMF subsystem creation RPC will be overwritten).
 This handler can be enabled via the `nvmf_set_config` RPC.
 Note: In a future version of SPDK, this handler will be enabled by default.
 
+### bdev
+
+A new API was added `spdk_bdev_io_get_aux_buf` allowing the caller to request
+an auxiliary buffer for its own private use. The API is used in the same manner that
+`spdk_bdev_io_get_buf` is used and the length of the buffer is always the same as the
+bdev_io primary buffer. 'spdk_bdev_io_put_aux_buf' frees the allocated auxiliary
+buffer.
+
 ### sock
 
 Added spdk_sock_writev_async for performing asynchronous writes to sockets. This call will

include/spdk/bdev_module.h

@@ -456,6 +456,17 @@ struct spdk_bdev {
 typedef void (*spdk_bdev_io_get_buf_cb)(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io,
 					bool success);
 
+/**
+ * Callback when an auxiliary buffer is allocated for the bdev I/O.
+ *
+ * \param ch The I/O channel the bdev I/O was handled on.
+ * \param bdev_io The bdev I/O
+ * \param aux_buf Pointer to the allocated buffer.  NULL if there was a failuer such as
+ * the size of the buffer to allocate is greater than the permitted maximum.
+ */
+typedef void (*spdk_bdev_io_get_aux_buf_cb)(struct spdk_io_channel *ch,
+		struct spdk_bdev_io *bdev_io, void *aux_buf);
+
 #define BDEV_IO_NUM_CHILD_IOV 32
 
 struct spdk_bdev_io {
@@ -629,6 +640,9 @@ struct spdk_bdev_io {
 		int           orig_iovcnt;
 		void	     *orig_md_buf;
 
+		/** Callback for when the aux buf is allocated */
+		spdk_bdev_io_get_aux_buf_cb get_aux_buf_cb;
+
 		/** Callback for when buf is allocated */
 		spdk_bdev_io_get_buf_cb get_buf_cb;
 
@@ -818,6 +832,26 @@ const struct spdk_bdev_aliases_list *spdk_bdev_get_aliases(const struct spdk_bde
  */
 void spdk_bdev_io_get_buf(struct spdk_bdev_io *bdev_io, spdk_bdev_io_get_buf_cb cb, uint64_t len);
 
+/**
+ * Allocate an auxillary buffer for given bdev_io. The length of the
+ * buffer will be the same size as the bdev_io primary buffer. The buffer
+ * must be freed using \c spdk_bdev_io_put_aux_buf() before completing
+ * the associated bdev_io.  This call will never fail. In case of lack of
+ * memory given callback \c cb will be deferred until enough memory is freed.
+ *
+ * \param bdev_io I/O to allocate buffer for.
+ * \param cb callback to be called when the buffer is allocated
+ */
+void spdk_bdev_io_get_aux_buf(struct spdk_bdev_io *bdev_io, spdk_bdev_io_get_aux_buf_cb cb);
+
+/**
+ * Free an auxiliary buffer previously allocated by \c spdk_bdev_io_get_aux_buf().
+ *
+ * \param bdev_io bdev_io specified when the aux_buf was allocated.
+ * \param aux_buf auxiliary buffer to free
+ */
+void spdk_bdev_io_put_aux_buf(struct spdk_bdev_io *bdev_io, void *aux_buf);
+
 /**
  * Set the given buffer as the data buffer described by this bdev_io.
  *

lib/bdev/bdev.c

@@ -595,12 +595,19 @@ _bdev_io_set_bounce_md_buf(struct spdk_bdev_io *bdev_io, void *md_buf, size_t le
 }
 
 static void
-bdev_io_get_buf_complete(struct spdk_bdev_io *bdev_io, bool status)
+bdev_io_get_buf_complete(struct spdk_bdev_io *bdev_io, void *buf, bool status)
 {
 	struct spdk_io_channel *ch = spdk_bdev_io_get_io_channel(bdev_io);
 
-	bdev_io->internal.get_buf_cb(ch, bdev_io, status);
+	if (spdk_unlikely(bdev_io->internal.get_aux_buf_cb != NULL)) {
-	bdev_io->internal.get_buf_cb = NULL;
+		bdev_io->internal.get_aux_buf_cb(ch, bdev_io, buf);
+		bdev_io->internal.get_aux_buf_cb = NULL;
+	} else {
+		assert(bdev_io->internal.get_buf_cb != NULL);
+		bdev_io->internal.buf = buf;
+		bdev_io->internal.get_buf_cb(ch, bdev_io, status);
+		bdev_io->internal.get_buf_cb = NULL;
+	}
 }
 
 static void
@@ -611,6 +618,11 @@ _bdev_io_set_buf(struct spdk_bdev_io *bdev_io, void *buf, uint64_t len)
 	uint64_t md_len, alignment;
 	void *aligned_buf;
 
+	if (spdk_unlikely(bdev_io->internal.get_aux_buf_cb != NULL)) {
+		bdev_io_get_buf_complete(bdev_io, buf, true);
+		return;
+	}
+
 	alignment = spdk_bdev_get_buf_align(bdev);
 	buf_allocated = _is_buf_allocated(bdev_io->u.bdev.iovs);
 	aligned_buf = (void *)(((uintptr_t)buf + (alignment - 1)) & ~(alignment - 1));
@@ -633,30 +645,23 @@ _bdev_io_set_buf(struct spdk_bdev_io *bdev_io, void *buf, uint64_t len)
 			spdk_bdev_io_set_md_buf(bdev_io, aligned_buf, md_len);
 		}
 	}
-
+	bdev_io_get_buf_complete(bdev_io, buf, true);
-	bdev_io->internal.buf = buf;
-	bdev_io_get_buf_complete(bdev_io, true);
 }
 
 static void
-bdev_io_put_buf(struct spdk_bdev_io *bdev_io)
+_bdev_io_put_buf(struct spdk_bdev_io *bdev_io, void *buf, uint64_t buf_len)
 {
 	struct spdk_bdev *bdev = bdev_io->bdev;
 	struct spdk_mempool *pool;
 	struct spdk_bdev_io *tmp;
 	bdev_io_stailq_t *stailq;
 	struct spdk_bdev_mgmt_channel *ch;
-	uint64_t buf_len, md_len, alignment;
+	uint64_t md_len, alignment;
-	void *buf;
 
-	buf = bdev_io->internal.buf;
-	buf_len = bdev_io->internal.buf_len;
 	md_len = spdk_bdev_is_md_separate(bdev) ? bdev_io->u.bdev.num_blocks * bdev->md_len : 0;
 	alignment = spdk_bdev_get_buf_align(bdev);
 	ch = bdev_io->internal.ch->shared_resource->mgmt_ch;
 
-	bdev_io->internal.buf = NULL;
-
 	if (buf_len + alignment + md_len <= SPDK_BDEV_BUF_SIZE_WITH_MD(SPDK_BDEV_SMALL_BUF_MAX_SIZE) +
 	    SPDK_BDEV_POOL_ALIGNMENT) {
 		pool = g_bdev_mgr.buf_small_pool;
@@ -675,6 +680,23 @@ bdev_io_put_buf(struct spdk_bdev_io *bdev_io)
 	}
 }
 
+static void
+bdev_io_put_buf(struct spdk_bdev_io *bdev_io)
+{
+	assert(bdev_io->internal.buf != NULL);
+	_bdev_io_put_buf(bdev_io, bdev_io->internal.buf, bdev_io->internal.buf_len);
+	bdev_io->internal.buf = NULL;
+}
+
+void
+spdk_bdev_io_put_aux_buf(struct spdk_bdev_io *bdev_io, void *buf)
+{
+	uint64_t len = bdev_io->u.bdev.num_blocks * bdev_io->bdev->blocklen;
+
+	assert(buf != NULL);
+	_bdev_io_put_buf(bdev_io, buf, len);
+}
+
 static void
 _bdev_io_unset_bounce_buf(struct spdk_bdev_io *bdev_io)
 {
@@ -735,7 +757,7 @@ bdev_io_get_buf(struct spdk_bdev_io *bdev_io, uint64_t len)
 	    SPDK_BDEV_POOL_ALIGNMENT) {
 		SPDK_ERRLOG("Length + alignment %" PRIu64 " is larger than allowed\n",
 			    len + alignment);
-		bdev_io_get_buf_complete(bdev_io, false);
+		bdev_io_get_buf_complete(bdev_io, NULL, false);
 		return;
 	}
 
@@ -781,6 +803,17 @@ spdk_bdev_io_get_buf(struct spdk_bdev_io *bdev_io, spdk_bdev_io_get_buf_cb cb, u
 	bdev_io_get_buf(bdev_io, len);
 }
 
+void
+spdk_bdev_io_get_aux_buf(struct spdk_bdev_io *bdev_io, spdk_bdev_io_get_aux_buf_cb cb)
+{
+	uint64_t len = bdev_io->u.bdev.num_blocks * bdev_io->bdev->blocklen;
+
+	assert(cb != NULL);
+	assert(bdev_io->internal.get_aux_buf_cb == NULL);
+	bdev_io->internal.get_aux_buf_cb = cb;
+	bdev_io_get_buf(bdev_io, len);
+}
+
 static int
 bdev_module_get_max_ctx_size(void)
 {
@@ -2073,6 +2106,8 @@ bdev_io_init(struct spdk_bdev_io *bdev_io,
 	bdev_io->internal.orig_md_buf = NULL;
 	bdev_io->internal.error.nvme.cdw0 = 0;
 	bdev_io->num_retries = 0;
+	bdev_io->internal.get_buf_cb = NULL;
+	bdev_io->internal.get_aux_buf_cb = NULL;
 }
 
 static bool