bdev/compress: incorporate reducelib read and write

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

lib/bdev/compress/vbdev_compress.c

@@ -62,6 +62,8 @@
 
 /* TODO: need to get this from RPC on create or reduce metadata on load */
 #define TEST_MD_PATH "/tmp"
+#define DEV_CHUNK_SZ (16 * 1024)
+#define DEV_LBA_SZ 512
 
 /* To add support for new device types, follow the examples of the following...
  * Note that the string names are defined by the DPDK PMD in question so be
@@ -358,6 +360,21 @@ comp_dev_poller(void *args)
 	return 0;
 }
 
+static void
+spdk_reduce_rw_blocks_cb(void *arg, int reduce_errno)
+{
+	struct spdk_bdev_io *bdev_io = arg;
+
+	/* TODO: need to decide which error codes are bdev_io success vs failure;
+	 * example examine calls reading metadata */
+	if (reduce_errno == 0) {
+		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_SUCCESS);
+	} else {
+		SPDK_ERRLOG("ERROR %d on operation from reduce API\n", reduce_errno);
+		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.
@@ -365,7 +382,12 @@ comp_dev_poller(void *args)
 static void
 comp_read_get_buf_cb(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io, bool success)
 {
-	/* TODO future patch */
+	struct vbdev_compress *comp_bdev = SPDK_CONTAINEROF(bdev_io->bdev, struct vbdev_compress,
+					   comp_bdev);
+
+	spdk_reduce_vol_readv(comp_bdev->vol, bdev_io->u.bdev.iovs, bdev_io->u.bdev.iovcnt,
+			      bdev_io->u.bdev.offset_blocks, bdev_io->u.bdev.num_blocks,
+			      spdk_reduce_rw_blocks_cb, bdev_io);
 }
 
 /* TODO: A future patch will add routines to complete IO up the stack, need
@@ -406,8 +428,12 @@ vbdev_compress_submit_request(struct spdk_io_channel *ch, struct spdk_bdev_io *b
 		spdk_bdev_io_get_buf(bdev_io, comp_read_get_buf_cb,
 				     bdev_io->u.bdev.num_blocks * bdev_io->bdev->blocklen);
 		break;
-	/* TODO in future patch in the series */
 	case SPDK_BDEV_IO_TYPE_WRITE:
+		spdk_reduce_vol_writev(comp_bdev->vol, bdev_io->u.bdev.iovs, bdev_io->u.bdev.iovcnt,
+				       bdev_io->u.bdev.offset_blocks, bdev_io->u.bdev.num_blocks,
+				       spdk_reduce_rw_blocks_cb, bdev_io);
+		break;
+	/* TODO in future patch in the series */
 	case SPDK_BDEV_IO_TYPE_WRITE_ZEROES:
 	case SPDK_BDEV_IO_TYPE_UNMAP:
 	case SPDK_BDEV_IO_TYPE_FLUSH:
@@ -437,8 +463,8 @@ vbdev_compress_io_type_supported(void *ctx, enum spdk_bdev_io_type io_type)
 
 	switch (io_type) {
 	case SPDK_BDEV_IO_TYPE_READ:
-		return spdk_bdev_io_type_supported(comp_bdev->base_bdev, io_type);
 	case SPDK_BDEV_IO_TYPE_WRITE:
+		return spdk_bdev_io_type_supported(comp_bdev->base_bdev, io_type);
 	case SPDK_BDEV_IO_TYPE_UNMAP:
 	case SPDK_BDEV_IO_TYPE_RESET:
 	case SPDK_BDEV_IO_TYPE_FLUSH:
@@ -487,18 +513,19 @@ _device_unregister_cb(void *io_device)
 
 	/* Done with this comp_bdev. */
 	pthread_mutex_destroy(&comp_bdev->reduce_lock);
-	free(comp_bdev->drv_name);
 	free(comp_bdev->comp_bdev.name);
 	free(comp_bdev);
 }
 
-/* Called after we've unregistered following a hot remove callback.
+/* Called by reduceLib after performing unload vol actions */
- * Our finish entry point will be called next.
+static void
- */
+spdk_reduce_vol_unload_cb(void *cb_arg, int reduce_errno)
-static int
-vbdev_compress_destruct(void *ctx)
 {
-	struct vbdev_compress *comp_bdev = (struct vbdev_compress *)ctx;
+	struct vbdev_compress *comp_bdev = (struct vbdev_compress *)cb_arg;
+
+	if (reduce_errno) {
+		SPDK_ERRLOG("error %d\n", reduce_errno);
+	}
 
 	/* Remove this device from the internal list */
 	TAILQ_REMOVE(&g_vbdev_comp, comp_bdev, link);
@@ -511,6 +538,18 @@ vbdev_compress_destruct(void *ctx)
 
 	/* Unregister the io_device. */
 	spdk_io_device_unregister(comp_bdev, _device_unregister_cb);
+}
+
+/* Called after we've unregistered following a hot remove callback.
+ * Our finish entry point will be called next.
+ */
+static int
+vbdev_compress_destruct(void *ctx)
+{
+	struct vbdev_compress *comp_bdev = (struct vbdev_compress *)ctx;
+
+	/* Tell reduceLiib that we're done with this volume. */
+	spdk_reduce_vol_unload(comp_bdev->vol, spdk_reduce_vol_unload_cb, comp_bdev);
 
 	return 0;
 }
@@ -576,11 +615,16 @@ vbdev_reduce_init_cb(void *cb_arg, struct spdk_reduce_vol *vol, int reduce_errno
 {
 	struct vbdev_compress *meta_ctx = cb_arg;
 
+	/* We're done with metadata operations */
+	spdk_put_io_channel(meta_ctx->base_ch);
+
 	if (reduce_errno == 0) {
+		SPDK_NOTICELOG("OK for vol %s, error %u\n",
+			       spdk_bdev_get_name(meta_ctx->base_bdev), reduce_errno);
 		meta_ctx->vol = vol;
 		vbdev_compress_claim(meta_ctx);
 	} else {
-		SPDK_ERRLOG("for vol %s, error %u\n",
+		SPDK_ERRLOG("ERR for vol %s, error %u\n",
 			    spdk_bdev_get_name(meta_ctx->base_bdev), reduce_errno);
 		spdk_put_io_channel(meta_ctx->base_ch);
 		spdk_bdev_close(meta_ctx->base_desc);
@@ -704,7 +748,7 @@ vbdev_compress_base_bdev_hotremove_cb(void *ctx)
  * params.vol_size
  * params.chunk_size
  * compression PMD, algorithm, window size, comp level, etc.
- * TEST_MD_PATH
+ * DEV_MD_PATH
  */
 
 /* Common function for init and load to allocate and populate the minimal
@@ -728,9 +772,10 @@ _prepare_for_load_init(struct spdk_bdev *bdev)
 	meta_ctx->backing_dev.blocklen = bdev->blocklen;
 	meta_ctx->backing_dev.blockcnt = bdev->blockcnt;
 
-	meta_ctx->params.chunk_size = CHUNK_SIZE;
+	/* TODO, configurable chunk size & logical block size */
+	meta_ctx->params.chunk_size = DEV_CHUNK_SZ;
+	meta_ctx->params.logical_block_size = DEV_LBA_SZ;
 	meta_ctx->params.backing_io_unit_size = BACKING_IO_UNIT_SZ;
-	meta_ctx->params.logical_block_size = meta_ctx->params.backing_io_unit_size;
 
 	return meta_ctx;
 }
@@ -947,10 +992,6 @@ vbdev_compress_claim(struct vbdev_compress *comp_bdev)
 	 * other parms to reduce via init and read them back in the load path.
 	 */
 	comp_bdev->drv_name = ISAL_PMD;
-	if (!comp_bdev->drv_name) {
-		SPDK_ERRLOG("could not allocate comb_bdev drv_name\n");
-		goto error_drv_name;
-	}
 
 	/* Note: some of the fields below will change in the future - for example,
 	 * blockcnt specifically will not match (the compressed volume size will
@@ -958,8 +999,14 @@ vbdev_compress_claim(struct vbdev_compress *comp_bdev)
 	 */
 	comp_bdev->comp_bdev.product_name = COMP_BDEV_NAME;
 	comp_bdev->comp_bdev.write_cache = comp_bdev->base_bdev->write_cache;
+
 	comp_bdev->comp_bdev.required_alignment = comp_bdev->base_bdev->required_alignment;
-	comp_bdev->comp_bdev.optimal_io_boundary = comp_bdev->base_bdev->optimal_io_boundary;
+
+	comp_bdev->comp_bdev.optimal_io_boundary =
+		comp_bdev->params.chunk_size / comp_bdev->params.logical_block_size;
+
+	comp_bdev->comp_bdev.split_on_optimal_io_boundary = true;
+
 	comp_bdev->comp_bdev.blocklen = comp_bdev->base_bdev->blocklen;
 	comp_bdev->comp_bdev.blockcnt = comp_bdev->base_bdev->blockcnt;
 
@@ -1008,9 +1055,7 @@ error_vbdev_register:
 error_claim:
 	TAILQ_REMOVE(&g_vbdev_comp, comp_bdev, link);
 	spdk_io_device_unregister(comp_bdev, NULL);
-	free(comp_bdev->drv_name);
 error_open:
-error_drv_name:
 	free(comp_bdev->comp_bdev.name);
 error_bdev_name:
 	spdk_put_io_channel(comp_bdev->base_ch);