bdev/crypto: add IO queueing for out of mem condition via bdev layer

Also made on the prints a DEBUG message instead and noticed the really
name that was being registered by this component so updated it to
make it look like the rest of SPDK.

Signed-off-by: paul luse <paul.e.luse@intel.com>
Change-Id: I747a846cb365e7db49be50db941e83fb1b265ea0
Reviewed-on: https://review.gerrithub.io/c/spdk/spdk/+/460244
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>

lib/bdev/crypto/vbdev_crypto.c

@@ -38,7 +38,7 @@
 #include "spdk/endian.h"
 #include "spdk/io_channel.h"
 #include "spdk/bdev_module.h"
-
+#include "spdk_internal/log.h"
 
 #include <rte_config.h>
 #include <rte_version.h>
@@ -127,6 +127,7 @@ static void _complete_internal_read(struct spdk_bdev_io *bdev_io, bool success,
 static void _complete_internal_write(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg);
 static void vbdev_crypto_examine(struct spdk_bdev *bdev);
 static int vbdev_crypto_claim(struct spdk_bdev *bdev);
+static void vbdev_crypto_submit_request(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io);
 
 /* list of crypto_bdev names and their base bdevs via configuration file.
  * Used so we can parse the conf once at init and use this list in examine().
@@ -192,6 +193,10 @@ struct crypto_bdev_io {
 	uint64_t cry_num_blocks;			/* num of blocks for the contiguous buffer */
 	uint64_t cry_offset_blocks;			/* block offset on media */
 	struct iovec cry_iov;				/* iov representing contig write buffer */
+
+	/* for bdev_io_wait */
+	struct spdk_bdev_io_wait_entry bdev_io_wait;
+	struct spdk_io_channel *ch;
 };
 
 /* Called by vbdev_crypto_init_crypto_drivers() to init each discovered crypto device */
@@ -902,6 +907,32 @@ _complete_internal_read(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg
 	spdk_bdev_free_io(bdev_io);
 }
 
+static void
+vbdev_crypto_resubmit_io(void *arg)
+{
+	struct spdk_bdev_io *bdev_io = (struct spdk_bdev_io *)arg;
+	struct crypto_bdev_io *io_ctx = (struct crypto_bdev_io *)bdev_io->driver_ctx;
+
+	vbdev_crypto_submit_request(io_ctx->ch, bdev_io);
+}
+
+static void
+vbdev_crypto_queue_io(struct spdk_bdev_io *bdev_io)
+{
+	struct crypto_bdev_io *io_ctx = (struct crypto_bdev_io *)bdev_io->driver_ctx;
+	int rc;
+
+	io_ctx->bdev_io_wait.bdev = bdev_io->bdev;
+	io_ctx->bdev_io_wait.cb_fn = vbdev_crypto_resubmit_io;
+	io_ctx->bdev_io_wait.cb_arg = bdev_io;
+
+	rc = spdk_bdev_queue_io_wait(bdev_io->bdev, io_ctx->ch, &io_ctx->bdev_io_wait);
+	if (rc != 0) {
+		SPDK_ERRLOG("Queue io failed in vbdev_crypto_queue_io, rc=%d.\n", rc);
+		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
+	}
+}
+
 /* Callback for getting a buf from the bdev pool in the event that the caller passed
  * in NULL, we need to own the buffer so it doesn't get freed by another vbdev module
  * beneath us before we're done with it.
@@ -913,6 +944,7 @@ crypto_read_get_buf_cb(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io,
 	struct vbdev_crypto *crypto_bdev = SPDK_CONTAINEROF(bdev_io->bdev, struct vbdev_crypto,
 					   crypto_bdev);
 	struct crypto_io_channel *crypto_ch = spdk_io_channel_get_ctx(ch);
+	struct crypto_bdev_io *io_ctx = (struct crypto_bdev_io *)bdev_io->driver_ctx;
 	int rc;
 
 	if (!success) {
@@ -925,8 +957,14 @@ crypto_read_get_buf_cb(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io,
 				    bdev_io->u.bdev.num_blocks, _complete_internal_read,
 				    bdev_io);
 	if (rc != 0) {
-		SPDK_ERRLOG("ERROR on bdev_io submission!\n");
+		if (rc == -ENOMEM) {
-		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
+			SPDK_DEBUGLOG(SPDK_LOG_CRYPTO, "No memory, queue the IO.\n");
+			io_ctx->ch = ch;
+			vbdev_crypto_queue_io(bdev_io);
+		} else {
+			SPDK_ERRLOG("ERROR on bdev_io submission!\n");
+			spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
+		}
 	}
 }
 
@@ -983,8 +1021,14 @@ vbdev_crypto_submit_request(struct spdk_io_channel *ch, struct spdk_bdev_io *bde
 	}
 
 	if (rc != 0) {
-		SPDK_ERRLOG("ERROR on bdev_io submission!\n");
+		if (rc == -ENOMEM) {
-		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
+			SPDK_DEBUGLOG(SPDK_LOG_CRYPTO, "No memory, queue the IO.\n");
+			io_ctx->ch = ch;
+			vbdev_crypto_queue_io(bdev_io);
+		} else {
+			SPDK_ERRLOG("ERROR on bdev_io submission!\n");
+			spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
+		}
 	}
 }
 
@@ -1475,7 +1519,7 @@ vbdev_crypto_claim(struct spdk_bdev *bdev)
 		if (strcmp(name->bdev_name, bdev->name) != 0) {
 			continue;
 		}
-		SPDK_DEBUGLOG(SPDK_LOG_VBDEV_crypto, "Match on %s\n", bdev->name);
+		SPDK_DEBUGLOG(SPDK_LOG_CRYPTO, "Match on %s\n", bdev->name);
 
 		vbdev = calloc(1, sizeof(struct vbdev_crypto));
 		if (!vbdev) {
@@ -1616,7 +1660,7 @@ vbdev_crypto_claim(struct spdk_bdev *bdev)
 			rc = -EINVAL;
 			goto error_bdev_register;
 		}
-		SPDK_DEBUGLOG(SPDK_LOG_VBDEV_crypto, "registered io_device and virtual bdev for: %s\n",
+		SPDK_DEBUGLOG(SPDK_LOG_CRYPTO, "registered io_device and virtual bdev for: %s\n",
 			      name->vbdev_name);
 		break;
 	}
@@ -1693,4 +1737,4 @@ vbdev_crypto_examine(struct spdk_bdev *bdev)
 	spdk_bdev_module_examine_done(&crypto_if);
 }
 
-SPDK_LOG_REGISTER_COMPONENT("vbdev_crypto", SPDK_LOG_VBDEV_crypto)
+SPDK_LOG_REGISTER_COMPONENT("vbdev_crypto", SPDK_LOG_CRYPTO)

lib/bdev/crypto/vbdev_crypto_rpc.c

@@ -73,7 +73,7 @@ spdk_rpc_construct_crypto_bdev(struct spdk_jsonrpc_request *request,
 	if (spdk_json_decode_object(params, rpc_construct_crypto_decoders,
 				    SPDK_COUNTOF(rpc_construct_crypto_decoders),
 				    &req)) {
-		SPDK_DEBUGLOG(SPDK_LOG_VBDEV_crypto, "spdk_json_decode_object failed\n");
+		SPDK_DEBUGLOG(SPDK_LOG_CRYPTO, "spdk_json_decode_object failed\n");
 		goto invalid;
 	}
 

test/unit/lib/bdev/crypto.c/crypto_ut.c

@@ -139,6 +139,8 @@ mock_rte_crypto_op_attach_sym_session(struct rte_crypto_op *op,
 #include "bdev/crypto/vbdev_crypto.c"
 
 /* SPDK stubs */
+DEFINE_STUB(spdk_bdev_queue_io_wait, int, (struct spdk_bdev *bdev, struct spdk_io_channel *ch,
+		struct spdk_bdev_io_wait_entry *entry), 0);
 DEFINE_STUB(spdk_conf_find_section, struct spdk_conf_section *,
 	    (struct spdk_conf *cp, const char *name), NULL);
 DEFINE_STUB(spdk_conf_section_get_nval, char *,
@@ -374,11 +376,13 @@ test_error_paths(void)
 	g_bdev_io->type = SPDK_BDEV_IO_TYPE_WRITE;
 	g_enqueue_mock = g_dequeue_mock = ut_rte_crypto_op_bulk_alloc = 1;
 
-	/* test failure of spdk_mempool_get_bulk() */
+	/* test failure of spdk_mempool_get_bulk(), will result in success becuase it
+	 * will get queued.
+	 */
 	g_bdev_io->internal.status = SPDK_BDEV_IO_STATUS_SUCCESS;
 	MOCK_SET(spdk_mempool_get, NULL);
 	vbdev_crypto_submit_request(g_io_ch, g_bdev_io);
-	CU_ASSERT(g_bdev_io->internal.status == SPDK_BDEV_IO_STATUS_FAILED);
+	CU_ASSERT(g_bdev_io->internal.status == SPDK_BDEV_IO_STATUS_SUCCESS);
 
 	/* same thing but switch to reads to test error path in _crypto_complete_io() */
 	g_bdev_io->type = SPDK_BDEV_IO_TYPE_READ;