reduce: add spdk_reduce_vol_load()

Signed-off-by: Jim Harris <james.r.harris@intel.com>
Change-Id: I7923d461af8ec8835fc258fc537230eff52f787e

Reviewed-on: https://review.gerrithub.io/433088
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Chandler-Test-Pool: SPDK Automated Test System <sys_sgsw@intel.com>
Reviewed-by: Paul Luse <paul.e.luse@intel.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>
Reviewed-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>

include/spdk/reduce.h

@@ -141,6 +141,17 @@ void spdk_reduce_vol_init(struct spdk_reduce_vol_params *params,
 			  spdk_reduce_vol_op_with_handle_complete cb_fn,
 			  void *cb_arg);
 
+/**
+ * Load an existing libreduce compressed volume.
+ *
+ * \param backing_dev Structure describing the backing device containing the compressed volume.
+ * \param cb_fn Callback function to signal completion of the loading process.
+ * \param cb_arg Argument to pass to the callback function.
+ */
+void spdk_reduce_vol_load(struct spdk_reduce_backing_dev *backing_dev,
+			  spdk_reduce_vol_op_with_handle_complete cb_fn,
+			  void *cb_arg);
+
 /**
  * Unload a previously initialized or loaded libreduce compressed volume.
  *

lib/reduce/reduce.c

@@ -373,6 +373,127 @@ spdk_reduce_vol_init(struct spdk_reduce_vol_params *params,
 				 &init_ctx->backing_cb_args);
 }
 
+static void
+_load_read_super_and_path_cpl(void *cb_arg, int ziperrno)
+{
+	struct reduce_init_load_ctx *load_ctx = cb_arg;
+	struct spdk_reduce_vol *vol = load_ctx->vol;
+	struct spdk_reduce_vol_params *params = &vol->backing_super->params;
+	int64_t size, size_needed;
+	size_t mapped_len;
+	int rc;
+
+	if (memcmp(vol->backing_super->signature,
+		   SPDK_REDUCE_SIGNATURE,
+		   sizeof(vol->backing_super->signature)) != 0) {
+		/* This backing device isn't a libreduce backing device. */
+		rc = -EILSEQ;
+		goto error;
+	}
+
+	size_needed = spdk_reduce_get_backing_device_size(params);
+	size = vol->backing_dev->blockcnt * vol->backing_dev->blocklen;
+	if (size_needed > size) {
+		SPDK_ERRLOG("backing device size %" PRIi64 " but %" PRIi64 " expected\n",
+			    size, size_needed);
+		rc = -EILSEQ;
+		goto error;
+	}
+
+	memcpy(vol->pm_file.path, load_ctx->path, sizeof(vol->pm_file.path));
+	vol->pm_file.size = spdk_reduce_get_pm_file_size(params);
+	vol->pm_file.pm_buf = pmem_map_file(vol->pm_file.path, vol->pm_file.size,
+					    0, 0, &mapped_len, &vol->pm_file.pm_is_pmem);
+	if (vol->pm_file.pm_buf == NULL) {
+		SPDK_ERRLOG("could not pmem_map_file(%s): %s\n", vol->pm_file.path, strerror(errno));
+		rc = -errno;
+		goto error;
+	}
+
+	if (vol->pm_file.size != mapped_len) {
+		SPDK_ERRLOG("could not map entire pmem file (size=%" PRIu64 " mapped=%" PRIu64 ")\n",
+			    vol->pm_file.size, mapped_len);
+		rc = -ENOMEM;
+		goto error;
+	}
+
+	memcpy(&vol->uuid, &params->uuid, sizeof(params->uuid));
+
+	load_ctx->cb_fn(load_ctx->cb_arg, vol, 0);
+	spdk_dma_free(load_ctx->path);
+	free(load_ctx);
+	return;
+
+error:
+	load_ctx->cb_fn(load_ctx->cb_arg, NULL, rc);
+	spdk_dma_free(load_ctx->path);
+	free(load_ctx);
+	spdk_dma_free(vol->backing_super);
+	free(vol);
+}
+
+void
+spdk_reduce_vol_load(struct spdk_reduce_backing_dev *backing_dev,
+		     spdk_reduce_vol_op_with_handle_complete cb_fn, void *cb_arg)
+{
+	struct spdk_reduce_vol *vol;
+	struct reduce_init_load_ctx *load_ctx;
+
+	if (backing_dev->close == NULL || backing_dev->readv == NULL ||
+	    backing_dev->writev == NULL || backing_dev->unmap == NULL) {
+		SPDK_ERRLOG("backing_dev function pointer not specified\n");
+		cb_fn(cb_arg, NULL, -EINVAL);
+		return;
+	}
+
+	vol = calloc(1, sizeof(*vol));
+	if (vol == NULL) {
+		cb_fn(cb_arg, NULL, -ENOMEM);
+		return;
+	}
+
+	vol->backing_super = spdk_dma_zmalloc(sizeof(*vol->backing_super), 64, NULL);
+	if (vol->backing_super == NULL) {
+		free(vol);
+		cb_fn(cb_arg, NULL, -ENOMEM);
+		return;
+	}
+
+	vol->backing_dev = backing_dev;
+
+	load_ctx = calloc(1, sizeof(*load_ctx));
+	if (load_ctx == NULL) {
+		spdk_dma_free(vol->backing_super);
+		free(vol);
+		cb_fn(cb_arg, NULL, -ENOMEM);
+		return;
+	}
+
+	load_ctx->path = spdk_dma_zmalloc(REDUCE_PATH_MAX, 64, NULL);
+	if (load_ctx->path == NULL) {
+		free(load_ctx);
+		spdk_dma_free(vol->backing_super);
+		free(vol);
+		cb_fn(cb_arg, NULL, -ENOMEM);
+		return;
+	}
+
+	load_ctx->vol = vol;
+	load_ctx->cb_fn = cb_fn;
+	load_ctx->cb_arg = cb_arg;
+
+	load_ctx->iov[0].iov_base = vol->backing_super;
+	load_ctx->iov[0].iov_len = sizeof(*vol->backing_super);
+	load_ctx->iov[1].iov_base = load_ctx->path;
+	load_ctx->iov[1].iov_len = REDUCE_PATH_MAX;
+	load_ctx->backing_cb_args.cb_fn = _load_read_super_and_path_cpl;
+	load_ctx->backing_cb_args.cb_arg = load_ctx;
+	vol->backing_dev->readv(vol->backing_dev, load_ctx->iov, 2, 0,
+				(sizeof(*vol->backing_super) + REDUCE_PATH_MAX) /
+				vol->backing_dev->blocklen,
+				&load_ctx->backing_cb_args);
+}
+
 void
 spdk_reduce_vol_unload(struct spdk_reduce_vol *vol,
 		       spdk_reduce_vol_op_complete cb_fn, void *cb_arg)

test/unit/lib/reduce/reduce.c/reduce_ut.c

@@ -199,6 +199,13 @@ init_cb(void *cb_arg, struct spdk_reduce_vol *vol, int ziperrno)
 	g_ziperrno = ziperrno;
 }
 
+static void
+load_cb(void *cb_arg, struct spdk_reduce_vol *vol, int ziperrno)
+{
+	g_vol = vol;
+	g_ziperrno = ziperrno;
+}
+
 static void
 unload_cb(void *cb_arg, int ziperrno)
 {
@@ -397,6 +404,49 @@ init_backing_dev(void)
 	backing_dev_destroy(&backing_dev);
 }
 
+static void
+load(void)
+{
+	struct spdk_reduce_vol_params params = {};
+	struct spdk_reduce_backing_dev backing_dev = {};
+	char pmem_file_path[REDUCE_PATH_MAX];
+
+	params.vol_size = 1024 * 1024; /* 1MB */
+	params.chunk_size = 16 * 1024;
+	params.backing_io_unit_size = 512;
+	spdk_uuid_generate(&params.uuid);
+
+	backing_dev_init(&backing_dev, &params);
+
+	g_vol = NULL;
+	g_ziperrno = -1;
+	spdk_reduce_vol_init(&params, &backing_dev, TEST_MD_PATH, init_cb, NULL);
+	CU_ASSERT(g_ziperrno == 0);
+	SPDK_CU_ASSERT_FATAL(g_vol != NULL);
+	SPDK_CU_ASSERT_FATAL(g_path != NULL);
+	memcpy(pmem_file_path, g_path, sizeof(pmem_file_path));
+
+	g_ziperrno = -1;
+	spdk_reduce_vol_unload(g_vol, unload_cb, NULL);
+	CU_ASSERT(g_ziperrno == 0);
+
+	g_vol = NULL;
+	g_path = NULL;
+	g_ziperrno = -1;
+	spdk_reduce_vol_load(&backing_dev, load_cb, NULL);
+	CU_ASSERT(g_ziperrno == 0);
+	SPDK_CU_ASSERT_FATAL(g_vol != NULL);
+	SPDK_CU_ASSERT_FATAL(g_path != NULL);
+	CU_ASSERT(strncmp(g_path, pmem_file_path, sizeof(pmem_file_path)) == 0);
+
+	g_ziperrno = -1;
+	spdk_reduce_vol_unload(g_vol, unload_cb, NULL);
+	CU_ASSERT(g_ziperrno == 0);
+
+	persistent_pm_buf_destroy();
+	backing_dev_destroy(&backing_dev);
+}
+
 int
 main(int argc, char **argv)
 {
@@ -418,7 +468,8 @@ main(int argc, char **argv)
 		CU_add_test(suite, "get_backing_device_size", get_backing_device_size) == NULL ||
 		CU_add_test(suite, "init_failure", init_failure) == NULL ||
 		CU_add_test(suite, "init_md", init_md) == NULL ||
-		CU_add_test(suite, "init_backing_dev", init_backing_dev) == NULL
+		CU_add_test(suite, "init_backing_dev", init_backing_dev) == NULL ||
+		CU_add_test(suite, "load", load) == NULL
 	) {
 		CU_cleanup_registry();
 		return CU_get_error();