lvol: allow to configure cluster size of lvol store

New optional parameter -c or --cluster_sz in construct_lvol_store() RPC,
as well as in vbdev_lvol and lvol lib API.

This parameter allows to configure cluster size of blobstore that lvol store
is build upon.

When this parameter is not specified, default of 1GiB is used.

spdk_lvs_opts struct was created to facilitate any future options when
creating lvol store.

Signed-off-by: Tomasz Zawadzki <tomasz.zawadzki@intel.com>
Change-Id: Ibfe8765ede3e78ff19c36f46043e4cec2e5c9f97
Reviewed-on: https://review.gerrithub.io/379356
Tested-by: SPDK Automated Test System <sys_sgsw@intel.com>
Reviewed-by: Jim Harris <james.r.harris@intel.com>
Reviewed-by: Piotr Pelpliński <piotr.pelplinski@intel.com>

include/spdk/lvol.h

@@ -45,6 +45,10 @@
 struct spdk_lvol_store;
 struct spdk_lvol;
 
+struct spdk_lvs_opts {
+	uint32_t cluster_sz;
+};
+
 typedef void (*spdk_lvs_op_with_handle_complete)(void *cb_arg, struct spdk_lvol_store *lvol_store,
 		int lvserrno);
 typedef void (*spdk_lvs_op_complete)(void *cb_arg, int lvserrno);
@@ -52,7 +56,8 @@ typedef void (*spdk_lvol_op_complete)(void *cb_arg, int lvolerrno);
 typedef void (*spdk_lvol_op_with_handle_complete)(void *cb_arg, struct spdk_lvol *lvol,
 		int lvolerrno);
 
-int spdk_lvs_init(struct spdk_bs_dev *bs_dev, spdk_lvs_op_with_handle_complete cb_fn, void *cb_arg);
+int spdk_lvs_init(struct spdk_bs_dev *bs_dev, struct spdk_lvs_opts *o,
+		  spdk_lvs_op_with_handle_complete cb_fn, void *cb_arg);
 int spdk_lvs_unload(struct spdk_lvol_store *lvol_store, spdk_lvs_op_complete cb_fn, void *cb_arg);
 int spdk_lvol_create(struct spdk_lvol_store *lvs, uint64_t sz,
 		     spdk_lvol_op_with_handle_complete cb_fn, void *cb_arg);

include/spdk_internal/lvolstore.h

@@ -37,6 +37,9 @@
 #include "spdk/lvol.h"
 #include "spdk_internal/bdev.h"
 
+/* Default size of blobstore cluster */
+#define SPDK_LVS_OPTS_CLUSTER_SZ (1024 * 1024 * 1024)
+
 /* Length of string returned from uuid_unparse() */
 #define UUID_STRING_LEN 37
 

lib/bdev/lvol/vbdev_lvol.c

@@ -92,14 +92,25 @@ end:
 }
 
 int
-vbdev_lvs_create(struct spdk_bdev *base_bdev,
+vbdev_lvs_create(struct spdk_bdev *base_bdev, uint32_t cluster_sz,
-		 spdk_lvs_op_with_handle_complete cb_fn,
+		 spdk_lvs_op_with_handle_complete cb_fn, void *cb_arg)
-		 void *cb_arg)
 {
 	struct spdk_bs_dev *bs_dev;
 	struct spdk_lvs_with_handle_req *lvs_req;
+	struct spdk_lvs_opts *opts = NULL;
+	struct spdk_lvs_opts temp;
 	int rc;
 
+	if (base_bdev == NULL) {
+		SPDK_ERRLOG("Bdev does not exist\n");
+		return -ENODEV;
+	}
+
+	if (cluster_sz != 0) {
+		temp.cluster_sz = cluster_sz;
+		opts = &temp;
+	}
+
 	lvs_req = calloc(1, sizeof(*lvs_req));
 	if (!lvs_req) {
 		SPDK_ERRLOG("Cannot alloc memory for vbdev lvol store request pointer\n");
@@ -118,7 +129,7 @@ vbdev_lvs_create(struct spdk_bdev *base_bdev,
 	lvs_req->cb_fn = cb_fn;
 	lvs_req->cb_arg = cb_arg;
 
-	rc = spdk_lvs_init(bs_dev, _vbdev_lvs_create_cb, lvs_req);
+	rc = spdk_lvs_init(bs_dev, opts, _vbdev_lvs_create_cb, lvs_req);
 	if (rc < 0) {
 		free(lvs_req);
 		bs_dev->destroy(bs_dev);

lib/bdev/lvol/vbdev_lvol.h

@@ -45,8 +45,8 @@ struct lvol_store_bdev {
 
 	TAILQ_ENTRY(lvol_store_bdev)	lvol_stores;
 };
-int vbdev_lvs_create(struct spdk_bdev *base_bdev, spdk_lvs_op_with_handle_complete cb_fn,
+int vbdev_lvs_create(struct spdk_bdev *base_bdev, uint32_t cluster_sz,
-		     void *cb_arg);
+		     spdk_lvs_op_with_handle_complete cb_fn, void *cb_arg);
 void vbdev_lvs_destruct(struct spdk_lvol_store *lvs, spdk_lvs_op_complete cb_fn, void *cb_arg);
 
 int vbdev_lvol_create(uuid_t uuid, size_t sz, spdk_lvol_op_with_handle_complete cb_fn,

lib/bdev/lvol/vbdev_lvol_rpc.c

@@ -42,6 +42,7 @@ SPDK_LOG_REGISTER_TRACE_FLAG("lvolrpc", SPDK_TRACE_LVOL_RPC)
 
 struct rpc_construct_lvol_store {
 	char *base_name;
+	uint32_t cluster_sz;
 };
 
 static void
@@ -52,6 +53,7 @@ free_rpc_construct_lvol_store(struct rpc_construct_lvol_store *req)
 
 static const struct spdk_json_object_decoder rpc_construct_lvol_store_decoders[] = {
 	{"base_name", offsetof(struct rpc_construct_lvol_store, base_name), spdk_json_decode_string},
+	{"cluster_sz", offsetof(struct rpc_construct_lvol_store, cluster_sz), spdk_json_decode_uint32, true},
 };
 
 static void
@@ -114,7 +116,7 @@ spdk_rpc_construct_lvol_store(struct spdk_jsonrpc_request *request,
 		goto invalid;
 	}
 
-	rc = vbdev_lvs_create(bdev, _spdk_rpc_lvol_store_construct_cb, request);
+	rc = vbdev_lvs_create(bdev, req.cluster_sz, _spdk_rpc_lvol_store_construct_cb, request);
 	if (rc < 0) {
 		goto invalid;
 	}

lib/lvol/lvol.c

@@ -70,18 +70,32 @@ _lvs_init_cb(void *cb_arg, struct spdk_blob_store *bs, int lvserrno)
 	free(lvs_req);
 }
 
+static void
+spdk_setup_lvs_opts(struct spdk_bs_opts *bs_opts, struct spdk_lvs_opts *o)
+{
+	spdk_bs_opts_init(bs_opts);
+	if (o) {
+		bs_opts->cluster_sz = o->cluster_sz;
+	} else {
+		bs_opts->cluster_sz = SPDK_LVS_OPTS_CLUSTER_SZ;
+	}
+}
+
 int
-spdk_lvs_init(struct spdk_bs_dev *bs_dev, spdk_lvs_op_with_handle_complete cb_fn,
+spdk_lvs_init(struct spdk_bs_dev *bs_dev, struct spdk_lvs_opts *o,
-	      void *cb_arg)
+	      spdk_lvs_op_with_handle_complete cb_fn, void *cb_arg)
 {
 	struct spdk_lvol_store *lvs;
 	struct spdk_lvs_with_handle_req *lvs_req;
+	struct spdk_bs_opts opts;
 
 	if (bs_dev == NULL) {
 		SPDK_ERRLOG("Blobstore device does not exist\n");
 		return -ENODEV;
 	}
 
+	spdk_setup_lvs_opts(&opts, o);
+
 	lvs = calloc(1, sizeof(*lvs));
 	if (!lvs) {
 		SPDK_ERRLOG("Cannot alloc memory for lvol store base pointer\n");
@@ -103,7 +117,7 @@ spdk_lvs_init(struct spdk_bs_dev *bs_dev, spdk_lvs_op_with_handle_complete cb_fn
 	lvs->bs_dev = bs_dev;
 
 	SPDK_INFOLOG(SPDK_TRACE_LVOL, "Initializing lvol store\n");
-	spdk_bs_init(bs_dev, NULL, _lvs_init_cb, lvs_req);
+	spdk_bs_init(bs_dev, &opts, _lvs_init_cb, lvs_req);
 
 	return 0;
 }

scripts/rpc.py

@@ -258,9 +258,14 @@ p.set_defaults(func=construct_error_bdev)
 
 def construct_lvol_store(args):
     params = {'base_name': args.base_name}
+
+    if args.cluster_sz:
+        params['cluster_sz'] = args.cluster_sz
+
     print_array(jsonrpc_call('construct_lvol_store', params))
 p = subparsers.add_parser('construct_lvol_store', help='Add logical volume store on base bdev')
 p.add_argument('base_name', help='base bdev name')
+p.add_argument('-c', '--cluster-sz', help='size of cluster (in bytes)', type=int, required=False)
 p.set_defaults(func=construct_lvol_store)
 
 

test/iscsi_tgt/lvol/iscsi_lvol.sh

@@ -40,7 +40,7 @@ for i in `seq 0 9`; do
     INITIATOR_TAG=$((i+2))
     $rpc_py add_initiator_group $INITIATOR_TAG $INITIATOR_NAME $NETMASK
     bdev=$($rpc_py construct_malloc_bdev $MALLOC_BDEV_SIZE $MALLOC_BLOCK_SIZE)
-    ls_guid=$($rpc_py construct_lvol_store $bdev)
+    ls_guid=$($rpc_py construct_lvol_store $bdev -c 1048576)
     LUNs=""
     for j in `seq 0 0`; do
         lb_guid=$($rpc_py construct_lvol_bdev $ls_guid 10)

test/nvmf/lvol/nvmf_lvol.sh

@@ -47,7 +47,7 @@ lvol_bdevs=()
 # Create malloc backends and creat lvol store on each
 for i in `seq 1 $SUBSYS_NR`; do
 	bdev="$($rpc_py construct_malloc_bdev $MALLOC_BDEV_SIZE $MALLOC_BLOCK_SIZE)"
-	ls_guid="$($rpc_py construct_lvol_store $bdev)"
+	ls_guid="$($rpc_py construct_lvol_store $bdev -c 1048576)"
 	lvol_stores+=("$ls_guid")
 
 	# 1 NVMe-OF subsystem per malloc bdev / lvol store / 10 lvol bdevs

test/unit/lib/bdev/vbdev_lvol.c/vbdev_lvol_ut.c

@@ -96,7 +96,8 @@ spdk_bdev_create_bs_dev(struct spdk_bdev *bdev, spdk_bdev_remove_cb_t remove_cb,
 }
 
 int
-spdk_lvs_init(struct spdk_bs_dev *bs_dev, spdk_lvs_op_with_handle_complete cb_fn, void *cb_arg)
+spdk_lvs_init(struct spdk_bs_dev *bs_dev, struct spdk_lvs_opts *o,
+	      spdk_lvs_op_with_handle_complete cb_fn, void *cb_arg)
 {
 	struct spdk_lvol_store *lvs;
 	int error = 0;
@@ -373,7 +374,7 @@ ut_lvol_hotremove(void)
 	g_bs_dev = NULL;
 
 	/* Lvol store is succesfully created */
-	rc = vbdev_lvs_create(&g_bdev, lvol_store_op_with_handle_complete, NULL);
+	rc = vbdev_lvs_create(&g_bdev, 0, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc == 0);
 	CU_ASSERT(g_lvserrno == 0);
 	CU_ASSERT(g_lvol_store != NULL);
@@ -460,7 +461,7 @@ ut_lvs_init(void)
 	/* spdk_lvs_init() fails */
 	lvol_store_initialize_fail = true;
 
-	rc = vbdev_lvs_create(&g_bdev, lvol_store_op_with_handle_complete, NULL);
+	rc = vbdev_lvs_create(&g_bdev, 0, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc != 0);
 	CU_ASSERT(g_lvserrno == 0);
 	CU_ASSERT(g_lvol_store == NULL);
@@ -471,7 +472,7 @@ ut_lvs_init(void)
 	/* spdk_lvs_init_cb() fails */
 	lvol_store_initialize_cb_fail = true;
 
-	rc = vbdev_lvs_create(&g_bdev, lvol_store_op_with_handle_complete, NULL);
+	rc = vbdev_lvs_create(&g_bdev, 0, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc == 0);
 	CU_ASSERT(g_lvserrno != 0);
 	CU_ASSERT(g_lvol_store == NULL);
@@ -480,7 +481,7 @@ ut_lvs_init(void)
 	lvol_store_initialize_cb_fail = false;
 
 	/* Lvol store is succesfully created */
-	rc = vbdev_lvs_create(&g_bdev, lvol_store_op_with_handle_complete, NULL);
+	rc = vbdev_lvs_create(&g_bdev, 0, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc == 0);
 	CU_ASSERT(g_lvserrno == 0);
 	CU_ASSERT(g_lvol_store != NULL);
@@ -492,7 +493,7 @@ ut_lvs_init(void)
 	g_bs_dev = NULL;
 
 	/* Bdev with lvol store already claimed */
-	rc = vbdev_lvs_create(&g_bdev, lvol_store_op_with_handle_complete, NULL);
+	rc = vbdev_lvs_create(&g_bdev, 0, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc != 0);
 	CU_ASSERT(g_lvserrno == 0);
 	CU_ASSERT(g_lvol_store == NULL);

test/unit/lib/lvol/lvol.c/lvol_ut.c

@@ -45,10 +45,16 @@
 #define BS_CLUSTER_SIZE (1024 * 1024)
 #define BS_FREE_CLUSTERS (DEV_BUFFER_SIZE / BS_CLUSTER_SIZE)
 
+#define SPDK_BLOB_OPTS_CLUSTER_SZ (1024 * 1024)
+#define SPDK_BLOB_OPTS_NUM_MD_PAGES UINT32_MAX
+#define SPDK_BLOB_OPTS_MAX_MD_OPS 32
+#define SPDK_BLOB_OPTS_MAX_CHANNEL_OPS 512
+
 int g_lvserrno;
 int g_resize_rc;
 struct spdk_lvol_store *g_lvol_store;
 struct spdk_lvol *g_lvol;
+struct spdk_bs_opts g_bs_opts;
 struct spdk_blob_store {};
 
 struct spdk_io_channel *spdk_bs_alloc_io_channel(struct spdk_blob_store *bs)
@@ -85,6 +91,8 @@ spdk_bs_init(struct spdk_bs_dev *dev, struct spdk_bs_opts *o,
 
 	bs = calloc(1, sizeof(*bs));
 
+	memcpy(&g_bs_opts, o, sizeof(struct spdk_bs_opts));
+
 	cb_fn(cb_arg, bs, 0);
 }
 
@@ -110,6 +118,15 @@ spdk_blob_get_id(struct spdk_blob *blob)
 	return id;
 }
 
+void
+spdk_bs_opts_init(struct spdk_bs_opts *opts)
+{
+	opts->cluster_sz = SPDK_BLOB_OPTS_CLUSTER_SZ;
+	opts->num_md_pages = SPDK_BLOB_OPTS_NUM_MD_PAGES;
+	opts->max_md_ops = SPDK_BLOB_OPTS_MAX_MD_OPS;
+	opts->max_channel_ops = SPDK_BLOB_OPTS_MAX_CHANNEL_OPS;
+}
+
 uint64_t
 spdk_bs_get_cluster_size(struct spdk_blob_store *bs)
 {
@@ -191,7 +208,8 @@ lvs_init_unload_success(void)
 	spdk_allocate_thread(_lvol_send_msg, NULL, NULL);
 
 	g_lvserrno = -1;
-	rc = spdk_lvs_init(&bs_dev, lvol_store_op_with_handle_complete, NULL);
+
+	rc = spdk_lvs_init(&bs_dev, NULL, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc == 0);
 	CU_ASSERT(g_lvserrno == 0);
 	SPDK_CU_ASSERT_FATAL(g_lvol_store != NULL);
@@ -205,6 +223,35 @@ lvs_init_unload_success(void)
 	spdk_free_thread();
 }
 
+static void
+lvs_init_opts_success(void)
+{
+	struct spdk_bs_dev bs_dev;
+	struct spdk_lvs_opts opts;
+	int rc = 0;
+
+	init_dev(&bs_dev);
+
+	spdk_allocate_thread(_lvol_send_msg, NULL, NULL);
+
+	g_lvserrno = -1;
+
+	opts.cluster_sz = 8192;
+	rc = spdk_lvs_init(&bs_dev, &opts, lvol_store_op_with_handle_complete, NULL);
+	CU_ASSERT(rc == 0);
+	CU_ASSERT(g_lvserrno == 0);
+	CU_ASSERT(g_bs_opts.cluster_sz == opts.cluster_sz);
+	SPDK_CU_ASSERT_FATAL(g_lvol_store != NULL);
+
+	g_lvserrno = -1;
+	rc = spdk_lvs_unload(g_lvol_store, lvol_store_op_complete, NULL);
+	CU_ASSERT(rc == 0);
+	CU_ASSERT(g_lvserrno == 0);
+	g_lvol_store = NULL;
+
+	spdk_free_thread();
+}
+
 static void
 lvs_unload_lvs_is_null_fail(void)
 {
@@ -231,7 +278,7 @@ lvol_create_destroy_success(void)
 	spdk_allocate_thread(_lvol_send_msg, NULL, NULL);
 
 	g_lvserrno = -1;
-	rc = spdk_lvs_init(&bs_dev, lvol_store_op_with_handle_complete, NULL);
+	rc = spdk_lvs_init(&bs_dev, NULL, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc == 0);
 	CU_ASSERT(g_lvserrno == 0);
 	SPDK_CU_ASSERT_FATAL(g_lvol_store != NULL);
@@ -263,11 +310,11 @@ lvol_create_fail(void)
 
 	g_lvol_store = NULL;
 	g_lvserrno = 0;
-	rc = spdk_lvs_init(NULL, lvol_store_op_with_handle_complete, NULL);
+	rc = spdk_lvs_init(NULL, NULL, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc != 0);
 	CU_ASSERT(g_lvol_store == NULL);
 
-	rc = spdk_lvs_init(&bs_dev, lvol_store_op_with_handle_complete, NULL);
+	rc = spdk_lvs_init(&bs_dev, NULL, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc == 0);
 	SPDK_CU_ASSERT_FATAL(g_lvol_store != NULL);
 
@@ -299,7 +346,7 @@ lvol_destroy_fail(void)
 
 	spdk_allocate_thread(_lvol_send_msg, NULL, NULL);
 
-	rc = spdk_lvs_init(&bs_dev, lvol_store_op_with_handle_complete, NULL);
+	rc = spdk_lvs_init(&bs_dev, NULL, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc == 0);
 	CU_ASSERT(g_lvserrno == 0);
 	SPDK_CU_ASSERT_FATAL(g_lvol_store != NULL);
@@ -330,7 +377,7 @@ lvol_close_fail(void)
 
 	spdk_allocate_thread(_lvol_send_msg, NULL, NULL);
 
-	rc = spdk_lvs_init(&bs_dev, lvol_store_op_with_handle_complete, NULL);
+	rc = spdk_lvs_init(&bs_dev, NULL, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc == 0);
 	CU_ASSERT(g_lvserrno == 0);
 	SPDK_CU_ASSERT_FATAL(g_lvol_store != NULL);
@@ -362,7 +409,7 @@ lvol_close_success(void)
 	spdk_allocate_thread(_lvol_send_msg, NULL, NULL);
 
 	g_lvserrno = -1;
-	rc = spdk_lvs_init(&bs_dev, lvol_store_op_with_handle_complete, NULL);
+	rc = spdk_lvs_init(&bs_dev, NULL, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc == 0);
 	CU_ASSERT(g_lvserrno == 0);
 	SPDK_CU_ASSERT_FATAL(g_lvol_store != NULL);
@@ -394,7 +441,7 @@ lvol_resize(void)
 
 	g_resize_rc = 0;
 	g_lvserrno = -1;
-	rc = spdk_lvs_init(&bs_dev, lvol_store_op_with_handle_complete, NULL);
+	rc = spdk_lvs_init(&bs_dev, NULL, lvol_store_op_with_handle_complete, NULL);
 	CU_ASSERT(rc == 0);
 	CU_ASSERT(g_lvserrno == 0);
 	SPDK_CU_ASSERT_FATAL(g_lvol_store != NULL);
@@ -462,6 +509,7 @@ int main(int argc, char **argv)
 
 	if (
 		CU_add_test(suite, "lvs_init_unload_success", lvs_init_unload_success) == NULL ||
+		CU_add_test(suite, "lvs_init_opts_success", lvs_init_opts_success) == NULL ||
 		CU_add_test(suite, "lvs_unload_lvs_is_null_fail", lvs_unload_lvs_is_null_fail) == NULL ||
 		CU_add_test(suite, "lvol_create_destroy_success", lvol_create_destroy_success) == NULL ||
 		CU_add_test(suite, "lvol_create_fail", lvol_create_fail) == NULL ||