reduce: write metadata to backing_dev

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

Reviewed-on: https://review.gerrithub.io/432504
Chandler-Test-Pool: SPDK Automated Test System <sys_sgsw@intel.com>
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: Changpeng Liu <changpeng.liu@intel.com>
Reviewed-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>

lib/reduce/reduce.c

@@ -34,6 +34,7 @@
 #include "spdk/stdinc.h"
 
 #include "spdk/reduce.h"
+#include "spdk/env.h"
 #include "spdk/string.h"
 #include "spdk_internal/log.h"
 
@@ -56,6 +57,7 @@ struct spdk_reduce_vol {
 	struct spdk_uuid			uuid;
 	struct spdk_reduce_pm_file		pm_file;
 	struct spdk_reduce_backing_dev		*backing_dev;
+	struct spdk_reduce_vol_superblock	*backing_super;
 };
 
 /*
@@ -163,6 +165,23 @@ spdk_reduce_get_backing_device_size(struct spdk_reduce_vol_params *params)
 	return total_backing_size;
 }
 
+struct reduce_init_load_ctx {
+	struct spdk_reduce_vol			*vol;
+	struct spdk_reduce_vol_cb_args		backing_cb_args;
+	spdk_reduce_vol_op_with_handle_complete	cb_fn;
+	void					*cb_arg;
+	struct iovec				iov;
+};
+
+static void
+_init_write_super_cpl(void *cb_arg, int ziperrno)
+{
+	struct reduce_init_load_ctx *init_ctx = cb_arg;
+
+	init_ctx->cb_fn(init_ctx->cb_arg, init_ctx->vol, ziperrno);
+	free(init_ctx);
+}
+
 void
 spdk_reduce_vol_init(struct spdk_reduce_vol_params *params,
 		     struct spdk_reduce_backing_dev *backing_dev,
@@ -171,6 +190,7 @@ spdk_reduce_vol_init(struct spdk_reduce_vol_params *params,
 {
 	struct spdk_reduce_vol *vol;
 	struct spdk_reduce_vol_superblock *pm_super;
+	struct reduce_init_load_ctx *init_ctx;
 	int64_t size, size_needed;
 	int rc;
 
@@ -220,19 +240,47 @@ spdk_reduce_vol_init(struct spdk_reduce_vol_params *params,
 
 	memcpy(&vol->pm_file, pm_file, sizeof(*pm_file));
 
+	vol->backing_super = spdk_dma_zmalloc(sizeof(*vol->backing_super), 0, NULL);
+	if (vol->backing_super == NULL) {
+		free(vol);
+		cb_fn(cb_arg, NULL, -ENOMEM);
+		return;
+	}
+
+	init_ctx = calloc(1, sizeof(*init_ctx));
+	if (init_ctx == NULL) {
+		spdk_dma_free(vol->backing_super);
+		free(vol);
+		cb_fn(cb_arg, NULL, -ENOMEM);
+		return;
+	}
+
 	memcpy(&vol->uuid, &params->uuid, sizeof(params->uuid));
 	vol->backing_dev = backing_dev;
-	pm_super = (struct spdk_reduce_vol_superblock *)vol->pm_file.pm_buf;
 
-	memcpy(pm_super->signature, SPDK_REDUCE_SIGNATURE, 8);
+	memcpy(vol->backing_super->signature, SPDK_REDUCE_SIGNATURE,
-	memcpy(&pm_super->params, params, sizeof(*params));
+	       sizeof(vol->backing_super->signature));
+	memcpy(&vol->backing_super->params, params, sizeof(*params));
+
+	pm_super = (struct spdk_reduce_vol_superblock *)vol->pm_file.pm_buf;
+	memcpy(pm_super, vol->backing_super, sizeof(*vol->backing_super));
+
 	if (vol->pm_file.pm_is_pmem) {
 		pmem_persist(pm_super, sizeof(*pm_super));
 	} else {
 		pmem_msync(pm_super, sizeof(*pm_super));
 	}
 
-	cb_fn(cb_arg, vol, 0);
+	init_ctx->vol = vol;
+	init_ctx->cb_fn = cb_fn;
+	init_ctx->cb_arg = cb_arg;
+	init_ctx->iov.iov_base = vol->backing_super;
+	init_ctx->iov.iov_len = sizeof(*vol->backing_super);
+	init_ctx->backing_cb_args.cb_fn = _init_write_super_cpl;
+	init_ctx->backing_cb_args.cb_arg = init_ctx;
+	vol->backing_dev->writev(vol->backing_dev, &init_ctx->iov, 1,
+				 0, sizeof(*vol->backing_super) / vol->backing_dev->blocklen,
+				 &init_ctx->backing_cb_args);
 }
 
 void
@@ -252,6 +300,7 @@ spdk_reduce_vol_unload(struct spdk_reduce_vol *vol,
 
 	vol->backing_dev->close(vol->backing_dev);
 
+	spdk_dma_free(vol->backing_super);
 	free(vol);
 	cb_fn(cb_arg, 0);
 }

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

@@ -353,6 +353,43 @@ init_md(void)
 	backing_dev_destroy(&backing_dev);
 }
 
+static void
+init_backing_dev(void)
+{
+	struct spdk_reduce_vol_params params = {};
+	struct spdk_reduce_vol_params *persistent_params;
+	struct spdk_reduce_backing_dev backing_dev = {};
+	struct spdk_reduce_pm_file pm_file = {};
+
+	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);
+	pm_file_init(&pm_file, &params);
+
+	g_vol = NULL;
+	g_ziperrno = -1;
+	spdk_reduce_vol_init(&params, &backing_dev, &pm_file, init_cb, NULL);
+	CU_ASSERT(g_ziperrno == 0);
+	SPDK_CU_ASSERT_FATAL(g_vol != NULL);
+	/* Confirm that libreduce persisted the params to the backing device. */
+	CU_ASSERT(memcmp(g_backing_dev_buf, SPDK_REDUCE_SIGNATURE, 8) == 0);
+	persistent_params = (struct spdk_reduce_vol_params *)(g_backing_dev_buf + 8);
+	CU_ASSERT(memcmp(persistent_params, &params, sizeof(params)) == 0);
+	CU_ASSERT(backing_dev.close != NULL);
+
+	g_ziperrno = -1;
+	g_backing_dev_closed = false;
+	spdk_reduce_vol_unload(g_vol, unload_cb, NULL);
+	CU_ASSERT(g_ziperrno == 0);
+	CU_ASSERT(g_backing_dev_closed == true);
+
+	pm_file_destroy();
+	backing_dev_destroy(&backing_dev);
+}
+
 int
 main(int argc, char **argv)
 {
@@ -373,7 +410,8 @@ main(int argc, char **argv)
 		CU_add_test(suite, "get_pm_file_size", get_pm_file_size) == NULL ||
 		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_md", init_md) == NULL ||
+		CU_add_test(suite, "init_backing_dev", init_backing_dev) == NULL
 	) {
 		CU_cleanup_registry();
 		return CU_get_error();