blobstore: add decouple parent function

This patch adds an API to decouple blobs parent removing dependency
on it. Blob stays thin after this operation.

Also unit tests for blobstore inflate are improved and reused with
decouple parent functionality.

Change-Id: I96dfee467c78cf4f4d929ec7bc05263f7a23a8aa
Signed-off-by: Tomasz Kulasek <tomaszx.kulasek@intel.com>
Reviewed-on: https://review.gerrithub.io/410829
Tested-by: SPDK Automated Test System <sys_sgsw@intel.com>
Reviewed-by: Jim Harris <james.r.harris@intel.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>

include/spdk/blob.h

@@ -522,8 +522,10 @@ void spdk_bs_delete_blob(struct spdk_blob_store *bs, spdk_blob_id blobid,
 			 spdk_blob_op_complete cb_fn, void *cb_arg);
 
 /**
- * Allocate all unallocated clusters in this blob and copy data from backing blob.
+ * Allocate all clusters in this blob. Data for allocated clusters is copied
- * This call removes dependency on backing blob.
+ * from backing blob(s) if they exist.
+ *
+ * This call removes all dependencies on any backing blobs.
  *
  * \param bs blobstore.
  * \param channel IO channel used to inflate blob.
@@ -534,6 +536,24 @@ void spdk_bs_delete_blob(struct spdk_blob_store *bs, spdk_blob_id blobid,
 void spdk_bs_inflate_blob(struct spdk_blob_store *bs, struct spdk_io_channel *channel,
 			  spdk_blob_id blobid, spdk_blob_op_complete cb_fn, void *cb_arg);
 
+/**
+ * Remove dependency on parent blob.
+ *
+ * This call allocates and copies data for any clusters that are allocated in
+ * the parent blob, and decouples parent updating dependencies of blob to
+ * its ancestor.
+ *
+ * If blob have no parent -EINVAL error is reported.
+ *
+ * \param bs blobstore.
+ * \param channel IO channel used to inflate blob.
+ * \param blobid The id of the blob.
+ * \param cb_fn Called when the operation is complete.
+ * \param cb_arg Argument passed to function cb_fn.
+ */
+void spdk_bs_blob_decouple_parent(struct spdk_blob_store *bs, struct spdk_io_channel *channel,
+				  spdk_blob_id blobid, spdk_blob_op_complete cb_fn, void *cb_arg);
+
 /**
  * Open a blob from the given blobstore.
  *

include/spdk/lvol.h

@@ -260,6 +260,15 @@ void spdk_lvol_open(struct spdk_lvol *lvol, spdk_lvol_op_with_handle_complete cb
  */
 void spdk_lvol_inflate(struct spdk_lvol *lvol, spdk_lvol_op_complete cb_fn, void *cb_arg);
 
+/**
+ * Decouple parent of lvol
+ *
+ * \param lvol Handle to lvol
+ * \param cb_fn Completion callback
+ * \param cb_arg Completion callback custom arguments
+ */
+void spdk_lvol_decouple_parent(struct spdk_lvol *lvol, spdk_lvol_op_complete cb_fn, void *cb_arg);
+
 #ifdef __cplusplus
 }
 #endif

lib/bdev/lvol/vbdev_lvol_rpc.c

@@ -760,6 +760,51 @@ invalid:
 
 SPDK_RPC_REGISTER("inflate_lvol_bdev", spdk_rpc_inflate_lvol_bdev, SPDK_RPC_RUNTIME)
 
+static void
+spdk_rpc_decouple_parent_lvol_bdev(struct spdk_jsonrpc_request *request,
+				   const struct spdk_json_val *params)
+{
+	struct rpc_inflate_lvol_bdev req = {};
+	struct spdk_bdev *bdev;
+	struct spdk_lvol *lvol;
+	int rc = 0;
+
+	SPDK_INFOLOG(SPDK_LOG_LVOL_RPC, "Decoupling parent of lvol\n");
+
+	if (spdk_json_decode_object(params, rpc_inflate_lvol_bdev_decoders,
+				    SPDK_COUNTOF(rpc_inflate_lvol_bdev_decoders),
+				    &req)) {
+		SPDK_INFOLOG(SPDK_LOG_LVOL_RPC, "spdk_json_decode_object failed\n");
+		rc = -EINVAL;
+		goto invalid;
+	}
+
+	bdev = spdk_bdev_get_by_name(req.name);
+	if (bdev == NULL) {
+		SPDK_ERRLOG("bdev '%s' does not exist\n", req.name);
+		rc = -ENODEV;
+		goto invalid;
+	}
+
+	lvol = vbdev_lvol_get_from_bdev(bdev);
+	if (lvol == NULL) {
+		SPDK_ERRLOG("lvol does not exist\n");
+		rc = -ENODEV;
+		goto invalid;
+	}
+
+	spdk_lvol_decouple_parent(lvol, _spdk_rpc_inflate_lvol_bdev_cb, request);
+
+	free_rpc_inflate_lvol_bdev(&req);
+	return;
+
+invalid:
+	spdk_jsonrpc_send_error_response(request, SPDK_JSONRPC_ERROR_INVALID_PARAMS, spdk_strerror(-rc));
+	free_rpc_inflate_lvol_bdev(&req);
+}
+
+SPDK_RPC_REGISTER("decouple_parent_lvol_bdev", spdk_rpc_decouple_parent_lvol_bdev, SPDK_RPC_RUNTIME)
+
 struct rpc_resize_lvol_bdev {
 	char *name;
 	uint64_t size;

lib/blob/blobstore.c

@@ -4153,6 +4153,10 @@ struct spdk_clone_snapshot_ctx {
 	/* Current cluster for inflate operation */
 	uint64_t cluster;
 
+	/* For inflation force allocation of all unallocated clusters and remove
+	 * thin-provisioning. Otherwise only decouple parent and keep clone thin. */
+	bool allocate_all;
+
 	struct {
 		spdk_blob_id id;
 		struct spdk_blob *blob;
@@ -4581,7 +4585,7 @@ void spdk_bs_create_clone(struct spdk_blob_store *bs, spdk_blob_id blobid,
 /* START spdk_bs_inflate_blob */
 
 static void
-_spdk_bs_inflate_blob_sync(void *cb_arg, int bserrno)
+_spdk_bs_inflate_blob_set_parent_cpl(void *cb_arg, struct spdk_blob *_parent, int bserrno)
 {
 	struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg;
 	struct spdk_blob *_blob = ctx->original.blob;
@@ -4591,11 +4595,18 @@ _spdk_bs_inflate_blob_sync(void *cb_arg, int bserrno)
 		return;
 	}
 
-	/* Destroy back_bs_dev */
+	assert(_parent != NULL);
-	_blob->back_bs_dev->destroy(_blob->back_bs_dev);
-	_blob->back_bs_dev = NULL;
 
-	_spdk_bs_clone_snapshot_origblob_cleanup(ctx, 0);
+	_spdk_bs_blob_list_remove(_blob);
+	_blob->parent_id = _parent->id;
+	_spdk_blob_set_xattr(_blob, BLOB_SNAPSHOT, &_blob->parent_id,
+			     sizeof(spdk_blob_id), true);
+
+	_blob->back_bs_dev->destroy(_blob->back_bs_dev);
+	_blob->back_bs_dev = spdk_bs_create_blob_bs_dev(_parent);
+	_spdk_bs_blob_list_add(_blob);
+
+	spdk_blob_sync_md(_blob, _spdk_bs_clone_snapshot_origblob_cleanup, ctx);
 }
 
 static void
@@ -4603,21 +4614,61 @@ _spdk_bs_inflate_blob_done(void *cb_arg, int bserrno)
 {
 	struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg;
 	struct spdk_blob *_blob = ctx->original.blob;
+	struct spdk_blob *_parent;
 
 	if (bserrno != 0) {
 		_spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno);
 		return;
 	}
 
-	_spdk_bs_blob_list_remove(_blob);
+	if (ctx->allocate_all) {
+		/* remove thin provisioning */
+		_spdk_bs_blob_list_remove(_blob);
+		_spdk_blob_remove_xattr(_blob, BLOB_SNAPSHOT, true);
+		_blob->invalid_flags = _blob->invalid_flags & ~SPDK_BLOB_THIN_PROV;
+		_blob->back_bs_dev->destroy(_blob->back_bs_dev);
+		_blob->back_bs_dev = NULL;
+		_blob->parent_id = SPDK_BLOBID_INVALID;
+	} else {
+		_parent = ((struct spdk_blob_bs_dev *)(_blob->back_bs_dev))->blob;
+		if (_parent->parent_id != SPDK_BLOBID_INVALID) {
+			/* We must change the parent of the inflated blob */
+			spdk_bs_open_blob(_blob->bs, _parent->parent_id,
+					  _spdk_bs_inflate_blob_set_parent_cpl, ctx);
+			return;
+		}
 
-	_spdk_blob_remove_xattr(_blob, BLOB_SNAPSHOT, true);
+		_spdk_bs_blob_list_remove(_blob);
+		_spdk_blob_remove_xattr(_blob, BLOB_SNAPSHOT, true);
+		_blob->parent_id = SPDK_BLOBID_INVALID;
+		_blob->back_bs_dev->destroy(_blob->back_bs_dev);
+		_blob->back_bs_dev = spdk_bs_create_zeroes_dev();
+	}
 
-	/* Unset thin provision */
-	_blob->invalid_flags = _blob->invalid_flags & ~SPDK_BLOB_THIN_PROV;
 	_blob->state = SPDK_BLOB_STATE_DIRTY;
+	spdk_blob_sync_md(_blob, _spdk_bs_clone_snapshot_origblob_cleanup, ctx);
+}
 
-	spdk_blob_sync_md(_blob, _spdk_bs_inflate_blob_sync, ctx);
+/* Check if cluster needs allocation */
+static inline bool
+_spdk_bs_cluster_needs_allocation(struct spdk_blob *blob, uint64_t cluster, bool allocate_all)
+{
+	struct spdk_blob_bs_dev *b;
+
+	assert(blob != NULL);
+
+	if (blob->active.clusters[cluster] != 0) {
+		/* Cluster is already allocated */
+		return false;
+	}
+
+	if (blob->parent_id == SPDK_BLOBID_INVALID) {
+		/* Blob have no parent blob */
+		return allocate_all;
+	}
+
+	b = (struct spdk_blob_bs_dev *)blob->back_bs_dev;
+	return (allocate_all || b->blob->active.clusters[cluster] != 0);
 }
 
 static void
@@ -4633,7 +4684,7 @@ _spdk_bs_inflate_blob_touch_next(void *cb_arg, int bserrno)
 	}
 
 	for (; ctx->cluster < _blob->active.num_clusters; ctx->cluster++) {
-		if (_blob->active.clusters[ctx->cluster] == 0) {
+		if (_spdk_bs_cluster_needs_allocation(_blob, ctx->cluster, ctx->allocate_all)) {
 			break;
 		}
 	}
@@ -4665,6 +4716,13 @@ _spdk_bs_inflate_blob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrn
 	}
 	ctx->original.blob = _blob;
 
+	if (!ctx->allocate_all && _blob->parent_id == SPDK_BLOBID_INVALID) {
+		/* This blob have no parent, so we cannot decouple it. */
+		SPDK_ERRLOG("Cannot decouple parent of blob with no parent.\n");
+		_spdk_bs_clone_snapshot_origblob_cleanup(ctx, -EINVAL);
+		return;
+	}
+
 	if (spdk_blob_is_thin_provisioned(_blob) == false) {
 		/* This is not thin provisioned blob. No need to inflate. */
 		_spdk_bs_clone_snapshot_origblob_cleanup(ctx, 0);
@@ -4676,7 +4734,7 @@ _spdk_bs_inflate_blob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrn
 	 */
 	lfc = 0;
 	for (i = 0; i < _blob->active.num_clusters; i++) {
-		if (_blob->active.clusters[i] == 0) {
+		if (_spdk_bs_cluster_needs_allocation(_blob, i, ctx->allocate_all)) {
 			lfc = spdk_bit_array_find_first_clear(_blob->bs->used_clusters, lfc);
 			if (lfc >= _blob->bs->total_clusters) {
 				/* No more free clusters. Cannot satisfy the request */
@@ -4691,8 +4749,9 @@ _spdk_bs_inflate_blob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrn
 	_spdk_bs_inflate_blob_touch_next(ctx, 0);
 }
 
-void spdk_bs_inflate_blob(struct spdk_blob_store *bs, struct spdk_io_channel *channel,
+static void
-			  spdk_blob_id blobid, spdk_blob_op_complete cb_fn, void *cb_arg)
+_spdk_bs_inflate_blob(struct spdk_blob_store *bs, struct spdk_io_channel *channel,
+		      spdk_blob_id blobid, bool allocate_all, spdk_blob_op_complete cb_fn, void *cb_arg)
 {
 	struct spdk_clone_snapshot_ctx *ctx = calloc(1, sizeof(*ctx));
 
@@ -4706,10 +4765,24 @@ void spdk_bs_inflate_blob(struct spdk_blob_store *bs, struct spdk_io_channel *ch
 	ctx->bserrno = 0;
 	ctx->original.id = blobid;
 	ctx->channel = channel;
+	ctx->allocate_all = allocate_all;
 
 	spdk_bs_open_blob(bs, ctx->original.id, _spdk_bs_inflate_blob_open_cpl, ctx);
 }
 
+void
+spdk_bs_inflate_blob(struct spdk_blob_store *bs, struct spdk_io_channel *channel,
+		     spdk_blob_id blobid, spdk_blob_op_complete cb_fn, void *cb_arg)
+{
+	_spdk_bs_inflate_blob(bs, channel, blobid, true, cb_fn, cb_arg);
+}
+
+void
+spdk_bs_blob_decouple_parent(struct spdk_blob_store *bs, struct spdk_io_channel *channel,
+			     spdk_blob_id blobid, spdk_blob_op_complete cb_fn, void *cb_arg)
+{
+	_spdk_bs_inflate_blob(bs, channel, blobid, false, cb_fn, cb_arg);
+}
 /* END spdk_bs_inflate_blob */
 
 /* START spdk_blob_resize */

lib/lvol/lvol.c

@@ -1474,3 +1474,39 @@ spdk_lvol_inflate(struct spdk_lvol *lvol, spdk_lvol_op_complete cb_fn, void *cb_
 	spdk_bs_inflate_blob(lvol->lvol_store->blobstore, req->channel, blob_id, _spdk_lvol_inflate_cb,
 			     req);
 }
+
+void
+spdk_lvol_decouple_parent(struct spdk_lvol *lvol, spdk_lvol_op_complete cb_fn, void *cb_arg)
+{
+	struct spdk_lvol_req *req;
+	struct spdk_blob *blob = lvol->blob;
+	spdk_blob_id blob_id = spdk_blob_get_id(blob);
+
+	assert(cb_fn != NULL);
+
+	if (lvol == NULL) {
+		SPDK_ERRLOG("Lvol does not exist\n");
+		cb_fn(cb_arg, -ENODEV);
+		return;
+	}
+
+	req = calloc(1, sizeof(*req));
+	if (!req) {
+		SPDK_ERRLOG("Cannot alloc memory for lvol request pointer\n");
+		cb_fn(cb_arg, -ENOMEM);
+		return;
+	}
+
+	req->cb_fn = cb_fn;
+	req->cb_arg = cb_arg;
+	req->channel = spdk_bs_alloc_io_channel(lvol->lvol_store->blobstore);
+	if (req->channel == NULL) {
+		SPDK_ERRLOG("Cannot alloc io channel for lvol inflate request\n");
+		free(req);
+		cb_fn(cb_arg, -ENOMEM);
+		return;
+	}
+
+	spdk_bs_blob_decouple_parent(lvol->lvol_store->blobstore, req->channel, blob_id,
+				     _spdk_lvol_inflate_cb, req);
+}

scripts/rpc.py

@@ -713,6 +713,15 @@ if __name__ == "__main__":
     p.add_argument('name', help='lvol bdev name')
     p.set_defaults(func=inflate_lvol_bdev)
 
+    @call_cmd
+    def decouple_parent_lvol_bdev(args):
+        rpc.lvol.decouple_parent_lvol_bdev(args.client,
+                                           name=args.name)
+
+    p = subparsers.add_parser('decouple_parent_lvol_bdev', help='Decouple parent of lvol')
+    p.add_argument('name', help='lvol bdev name')
+    p.set_defaults(func=inflate_lvol_bdev)
+
     @call_cmd
     def resize_lvol_bdev(args):
         rpc.lvol.resize_lvol_bdev(args.client,

scripts/rpc/lvol.py

@@ -134,6 +134,18 @@ def inflate_lvol_bdev(client, name):
     return client.call('inflate_lvol_bdev', params)
 
 
+def decouple_parent_lvol_bdev(client, name):
+    """Decouple parent of a logical volume.
+
+    Args:
+        name: name of logical volume to decouple parent
+    """
+    params = {
+        'name': name,
+    }
+    return client.call('decouple_parent_lvol_bdev', params)
+
+
 def destroy_lvol_store(client, uuid=None, lvs_name=None):
     """Destroy a logical volume store.
 

test/unit/lib/blob/blob.c/blob_ut.c

@@ -867,7 +867,7 @@ blob_clone(void)
 }
 
 static void
-blob_inflate(void)
+_blob_inflate(bool decouple_parent)
 {
 	struct spdk_blob_store *bs;
 	struct spdk_bs_dev *dev;
@@ -891,6 +891,7 @@ blob_inflate(void)
 
 	spdk_blob_opts_init(&opts);
 	opts.num_clusters = 10;
+	opts.thin_provision = true;
 
 	spdk_bs_create_blob_ext(bs, &opts, blob_op_with_id_complete, NULL);
 	CU_ASSERT(g_bserrno == 0);
@@ -903,9 +904,19 @@ blob_inflate(void)
 	blob = g_blob;
 
 	CU_ASSERT(spdk_blob_get_num_clusters(blob) == 10)
-	CU_ASSERT(spdk_blob_is_thin_provisioned(blob) == false);
+	CU_ASSERT(spdk_blob_is_thin_provisioned(blob) == true);
 
-	/* Create snapshot */
+	/* 1) Blob with no parent */
+	if (decouple_parent) {
+		/* Decouple parent of blob with no parent (should fail) */
+		spdk_bs_blob_decouple_parent(bs, channel, blobid, blob_op_complete, NULL);
+		CU_ASSERT(g_bserrno != 0);
+	} else {
+		/* Inflate of thin blob with no parent should made it thick */
+		spdk_bs_inflate_blob(bs, channel, blobid, blob_op_complete, NULL);
+		CU_ASSERT(g_bserrno == 0);
+		CU_ASSERT(spdk_blob_is_thin_provisioned(blob) == false);
+	}
 
 	spdk_bs_create_snapshot(bs, blobid, NULL, blob_op_with_id_complete, NULL);
 	CU_ASSERT(g_bserrno == 0);
@@ -928,19 +939,27 @@ blob_inflate(void)
 
 	free_clusters = spdk_bs_free_cluster_count(bs);
 
-	/* Inflate blob */
+	/* 2) Blob with parent */
-	spdk_bs_inflate_blob(bs, channel, blobid, blob_op_complete, NULL);
+	if (!decouple_parent) {
-	CU_ASSERT(g_bserrno == 0);
+		/* Do full blob inflation */
-
+		spdk_bs_inflate_blob(bs, channel, blobid, blob_op_complete, NULL);
-	/* All 10 clusters should be allocated from blob store */
+		CU_ASSERT(g_bserrno == 0);
-	CU_ASSERT(spdk_bs_free_cluster_count(bs) == free_clusters - 10);
+		/* all 10 clusters should be allocated */
+		CU_ASSERT(spdk_bs_free_cluster_count(bs) == free_clusters - 10);
+	} else {
+		/* Decouple parent of blob */
+		spdk_bs_blob_decouple_parent(bs, channel, blobid, blob_op_complete, NULL);
+		CU_ASSERT(g_bserrno == 0);
+		/* when only parent is removed, none of the clusters should be allocated */
+		CU_ASSERT(spdk_bs_free_cluster_count(bs) == free_clusters);
+	}
 
 	/* Now, it should be possible to delete snapshot */
 	spdk_bs_delete_blob(bs, snapshotid, blob_op_complete, NULL);
 	CU_ASSERT(g_bserrno == 0);
 
 	CU_ASSERT(spdk_blob_get_num_clusters(blob) == 10)
-	CU_ASSERT(spdk_blob_is_thin_provisioned(blob) == false);
+	CU_ASSERT(spdk_blob_is_thin_provisioned(blob) == decouple_parent);
 
 	spdk_blob_close(blob, blob_op_complete, NULL);
 	CU_ASSERT(g_bserrno == 0);
@@ -952,6 +971,13 @@ blob_inflate(void)
 	spdk_bs_free_io_channel(channel);
 }
 
+static void
+blob_inflate(void)
+{
+	_blob_inflate(false);
+	_blob_inflate(true);
+}
+
 static void
 blob_delete(void)
 {
@@ -3913,16 +3939,49 @@ blob_snapshot_rw_iov(void)
 	g_blobid = 0;
 }
 
+/**
+ * Inflate / decouple parent rw unit tests.
+ *
+ * --------------
+ * original blob:         0         1         2         3         4
+ *                   ,---------+---------+---------+---------+---------.
+ *         snapshot  |xxxxxxxxx|xxxxxxxxx|xxxxxxxxx|xxxxxxxxx|    -    |
+ *                   +---------+---------+---------+---------+---------+
+ *         snapshot2 |    -    |yyyyyyyyy|    -    |yyyyyyyyy|    -    |
+ *                   +---------+---------+---------+---------+---------+
+ *         blob      |    -    |zzzzzzzzz|    -    |    -    |    -    |
+ *                   '---------+---------+---------+---------+---------'
+ *                   .         .         .         .         .         .
+ * --------          .         .         .         .         .         .
+ * inflate:          .         .         .         .         .         .
+ *                   ,---------+---------+---------+---------+---------.
+ *         blob      |xxxxxxxxx|zzzzzzzzz|xxxxxxxxx|yyyyyyyyy|000000000|
+ *                   '---------+---------+---------+---------+---------'
+ *
+ *         NOTE: needs to allocate 4 clusters, thin provisioning removed, dependency
+ *               on snapshot2 and snapshot removed .         .         .
+ *                   .         .         .         .         .         .
+ * ----------------  .         .         .         .         .         .
+ * decouple parent:  .         .         .         .         .         .
+ *                   ,---------+---------+---------+---------+---------.
+ *         snapshot  |xxxxxxxxx|xxxxxxxxx|xxxxxxxxx|xxxxxxxxx|    -    |
+ *                   +---------+---------+---------+---------+---------+
+ *         blob      |    -    |zzzzzzzzz|    -    |yyyyyyyyy|    -    |
+ *                   '---------+---------+---------+---------+---------'
+ *
+ *         NOTE: needs to allocate 1 cluster, 3 clusters unallocated, dependency
+ *               on snapshot2 removed and on snapshot still exists. Snapshot2
+ *               should remain a clone of snapshot.
+ */
 static void
-blob_inflate_rw(void)
+_blob_inflate_rw(bool decouple_parent)
 {
-	static uint8_t *zero;
 	struct spdk_blob_store *bs;
 	struct spdk_bs_dev *dev;
-	struct spdk_blob *blob, *snapshot;
+	struct spdk_blob *blob, *snapshot, *snapshot2;
 	struct spdk_io_channel *channel;
 	struct spdk_blob_opts opts;
-	spdk_blob_id blobid, snapshotid;
+	spdk_blob_id blobid, snapshotid, snapshot2id;
 	uint64_t free_clusters;
 	uint64_t cluster_size;
 
@@ -3935,6 +3994,8 @@ blob_inflate_rw(void)
 	uint64_t pages_per_payload;
 
 	int i;
+	spdk_blob_id ids[2];
+	size_t count;
 
 	dev = init_dev();
 
@@ -3959,9 +4020,6 @@ blob_inflate_rw(void)
 	payload_clone = malloc(payload_size);
 	SPDK_CU_ASSERT_FATAL(payload_clone != NULL);
 
-	zero = calloc(1, payload_size);
-	SPDK_CU_ASSERT_FATAL(zero != NULL);
-
 	channel = spdk_bs_alloc_io_channel(bs);
 	SPDK_CU_ASSERT_FATAL(channel != NULL);
 
@@ -3983,20 +4041,22 @@ blob_inflate_rw(void)
 
 	CU_ASSERT(spdk_blob_get_num_clusters(blob) == 5);
 
-	/* Initial read should return zeroed payload */
+	/* 1) Initial read should return zeroed payload */
 	memset(payload_read, 0xFF, payload_size);
-	spdk_blob_io_read(blob, channel, payload_read, 0, pages_per_payload, blob_op_complete, NULL);
+	spdk_blob_io_read(blob, channel, payload_read, 0, pages_per_payload,
+			  blob_op_complete, NULL);
 	CU_ASSERT(g_bserrno == 0);
-	CU_ASSERT(memcmp(zero, payload_read, payload_size) == 0);
+	CU_ASSERT(spdk_mem_all_zero(payload_read, payload_size));
 
-	/* Fill whole blob with a pattern */
+	/* Fill whole blob with a pattern, except last cluster (to be sure it
-	memset(payload_write, 0xE5, payload_size);
+	 * isn't allocated) */
-	spdk_blob_io_write(blob, channel, payload_write, 0, pages_per_payload,
+	memset(payload_write, 0xE5, payload_size - cluster_size);
-			   blob_op_complete, NULL);
+	spdk_blob_io_write(blob, channel, payload_write, 0, pages_per_payload -
+			   pages_per_cluster, blob_op_complete, NULL);
 	CU_ASSERT(g_bserrno == 0);
 	CU_ASSERT(free_clusters != spdk_bs_free_cluster_count(bs));
 
-	/* Create snapshot from blob */
+	/* 2) Create snapshot from blob (first level) */
 	spdk_bs_create_snapshot(bs, blobid, NULL, blob_op_with_id_complete, NULL);
 	CU_ASSERT(g_bserrno == 0);
 	CU_ASSERT(g_blobid != SPDK_BLOBID_INVALID);
@@ -4012,13 +4072,17 @@ blob_inflate_rw(void)
 	CU_ASSERT(spdk_blob_get_num_clusters(snapshot) == 5)
 
 	/* Write every second cluster with a pattern.
+	 *
+	 * Last cluster shouldn't be written, to be sure that snapshot nor clone
+	 * doesn't allocate it.
 	 *
 	 * payload_clone stores expected result on "blob" read at the time and
 	 * is used only to check data consistency on clone before and after
 	 * inflation. Initially we fill it with a backing snapshots pattern
 	 * used before.
 	 */
-	memset(payload_clone, 0xE5, payload_size);
+	memset(payload_clone, 0xE5, payload_size - cluster_size);
+	memset(payload_clone + payload_size - cluster_size, 0x00, cluster_size);
 	memset(payload_write, 0xAA, payload_size);
 	for (i = 1; i < 5; i += 2) {
 		spdk_blob_io_write(blob, channel, payload_write, i * pages_per_cluster,
@@ -4038,20 +4102,124 @@ blob_inflate_rw(void)
 	CU_ASSERT(g_bserrno == 0);
 	CU_ASSERT(memcmp(payload_clone, payload_read, payload_size) == 0);
 
+	/* 3) Create second levels snapshot from blob */
+	spdk_bs_create_snapshot(bs, blobid, NULL, blob_op_with_id_complete, NULL);
+	CU_ASSERT(g_bserrno == 0);
+	CU_ASSERT(g_blobid != SPDK_BLOBID_INVALID);
+	snapshot2id = g_blobid;
+
+	spdk_bs_open_blob(bs, snapshot2id, blob_op_with_handle_complete, NULL);
+	CU_ASSERT(g_bserrno == 0);
+	SPDK_CU_ASSERT_FATAL(g_blob != NULL);
+	snapshot2 = g_blob;
+	CU_ASSERT(snapshot2->data_ro == true)
+	CU_ASSERT(snapshot2->md_ro == true)
+
+	CU_ASSERT(spdk_blob_get_num_clusters(snapshot2) == 5)
+
+	CU_ASSERT(snapshot2->parent_id == snapshotid);
+
+	/* Write one cluster on the top level blob. This cluster (1) covers
+	 * already allocated cluster in the snapshot2, so shouldn't be inflated
+	 * at all */
+	spdk_blob_io_write(blob, channel, payload_write, pages_per_cluster,
+			   pages_per_cluster, blob_op_complete, NULL);
+	CU_ASSERT(g_bserrno == 0);
+
+	/* Update expected result */
+	memcpy(payload_clone + cluster_size, payload_write, cluster_size);
+
+	/* Check data consistency on clone */
+	memset(payload_read, 0xFF, payload_size);
+	spdk_blob_io_read(blob, channel, payload_read, 0, pages_per_payload,
+			  blob_op_complete, NULL);
+	CU_ASSERT(g_bserrno == 0);
+	CU_ASSERT(memcmp(payload_clone, payload_read, payload_size) == 0);
+
+
 	/* Close all blobs */
 	spdk_blob_close(blob, blob_op_complete, NULL);
 	CU_ASSERT(g_bserrno == 0);
 
+	spdk_blob_close(snapshot2, blob_op_complete, NULL);
+	CU_ASSERT(g_bserrno == 0);
+
 	spdk_blob_close(snapshot, blob_op_complete, NULL);
 	CU_ASSERT(g_bserrno == 0);
 
-	/* Inflate blob */
+	/* Check snapshot-clone relations */
-	spdk_bs_inflate_blob(bs, channel, blobid, blob_op_complete, NULL);
+	count = 2;
+	CU_ASSERT(spdk_blob_get_clones(bs, snapshotid, ids, &count) == 0);
+	CU_ASSERT(count == 1);
+	CU_ASSERT(ids[0] == snapshot2id);
+
+	count = 2;
+	CU_ASSERT(spdk_blob_get_clones(bs, snapshot2id, ids, &count) == 0);
+	CU_ASSERT(count == 1);
+	CU_ASSERT(ids[0] == blobid);
+
+	CU_ASSERT(spdk_blob_get_parent_snapshot(bs, blobid) == snapshot2id);
+
+	free_clusters = spdk_bs_free_cluster_count(bs);
+	if (!decouple_parent) {
+		/* Do full blob inflation */
+		spdk_bs_inflate_blob(bs, channel, blobid, blob_op_complete, NULL);
+		CU_ASSERT(g_bserrno == 0);
+
+		/* All clusters should be inflated (except one already allocated
+		 * in a top level blob) */
+		CU_ASSERT(spdk_bs_free_cluster_count(bs) == free_clusters - 4);
+
+		/* Check if relation tree updated correctly */
+		count = 2;
+		CU_ASSERT(spdk_blob_get_clones(bs, snapshotid, ids, &count) == 0);
+
+		/* snapshotid have one clone */
+		CU_ASSERT(count == 1);
+		CU_ASSERT(ids[0] == snapshot2id);
+
+		/* snapshot2id have no clones */
+		count = 2;
+		CU_ASSERT(spdk_blob_get_clones(bs, snapshot2id, ids, &count) == 0);
+		CU_ASSERT(count == 0);
+
+		CU_ASSERT(spdk_blob_get_parent_snapshot(bs, blobid) == SPDK_BLOBID_INVALID);
+	} else {
+		/* Decouple parent of blob */
+		spdk_bs_blob_decouple_parent(bs, channel, blobid, blob_op_complete, NULL);
+		CU_ASSERT(g_bserrno == 0);
+
+		/* Only one cluster from a parent should be inflated (second one
+		 * is covered by a cluster written on a top level blob, and
+		 * already allocated) */
+		CU_ASSERT(spdk_bs_free_cluster_count(bs) == free_clusters - 1);
+
+		/* Check if relation tree updated correctly */
+		count = 2;
+		CU_ASSERT(spdk_blob_get_clones(bs, snapshotid, ids, &count) == 0);
+
+		/* snapshotid have two clones now */
+		CU_ASSERT(count == 2);
+		CU_ASSERT(ids[0] == blobid || ids[1] == blobid);
+		CU_ASSERT(ids[0] == snapshot2id || ids[1] == snapshot2id);
+
+		/* snapshot2id have no clones */
+		count = 2;
+		CU_ASSERT(spdk_blob_get_clones(bs, snapshot2id, ids, &count) == 0);
+		CU_ASSERT(count == 0);
+
+		CU_ASSERT(spdk_blob_get_parent_snapshot(bs, blobid) == snapshotid);
+	}
+
+	/* Try to delete snapshot2 (should pass) */
+	spdk_bs_delete_blob(bs, snapshot2id, blob_op_complete, NULL);
 	CU_ASSERT(g_bserrno == 0);
 
-	/* Try to delete snapshot (should pass) */
+	/* Try to delete base snapshot (for decouple_parent should fail while
+	 * dependency still exists) */
 	spdk_bs_delete_blob(bs, snapshotid, blob_op_complete, NULL);
-	CU_ASSERT(g_bserrno == 0);
+	CU_ASSERT(decouple_parent || g_bserrno == 0);
+	CU_ASSERT(!decouple_parent || g_bserrno != 0);
 
 	/* Reopen blob after snapshot deletion */
 	spdk_bs_open_blob(bs, blobid, blob_op_with_handle_complete, NULL);
@@ -4063,7 +4231,8 @@ blob_inflate_rw(void)
 
 	/* Check data consistency on inflated blob */
 	memset(payload_read, 0xFF, payload_size);
-	spdk_blob_io_read(blob, channel, payload_read, 0, pages_per_payload, blob_op_complete, NULL);
+	spdk_blob_io_read(blob, channel, payload_read, 0, pages_per_payload,
+			  blob_op_complete, NULL);
 	CU_ASSERT(g_bserrno == 0);
 	CU_ASSERT(memcmp(payload_clone, payload_read, payload_size) == 0);
 
@@ -4082,7 +4251,13 @@ blob_inflate_rw(void)
 	free(payload_read);
 	free(payload_write);
 	free(payload_clone);
-	free(zero);
+}
+
+static void
+blob_inflate_rw(void)
+{
+	_blob_inflate_rw(false);
+	_blob_inflate_rw(true);
 }
 
 /**
@@ -4472,8 +4647,8 @@ int main(int argc, char **argv)
 		CU_add_test(suite, "bs_load_iter", bs_load_iter) == NULL ||
 		CU_add_test(suite, "blob_snapshot_rw", blob_snapshot_rw) == NULL ||
 		CU_add_test(suite, "blob_snapshot_rw_iov", blob_snapshot_rw_iov) == NULL ||
-		CU_add_test(suite, "blob_inflate_rw", blob_inflate_rw) == NULL ||
 		CU_add_test(suite, "blob_relations", blob_relations) == NULL ||
+		CU_add_test(suite, "blob_inflate_rw", blob_inflate_rw) == NULL ||
 		CU_add_test(suite, "blob_snapshot_freeze_io", blob_snapshot_freeze_io) == NULL
 	) {
 		CU_cleanup_registry();

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

@@ -100,6 +100,12 @@ void spdk_bs_inflate_blob(struct spdk_blob_store *bs, struct spdk_io_channel *ch
 	cb_fn(cb_arg, g_inflate_rc);
 }
 
+void spdk_bs_blob_decouple_parent(struct spdk_blob_store *bs, struct spdk_io_channel *channel,
+				  spdk_blob_id blobid, spdk_blob_op_complete cb_fn, void *cb_arg)
+{
+	cb_fn(cb_arg, g_inflate_rc);
+}
+
 void
 spdk_bs_iter_next(struct spdk_blob_store *bs, struct spdk_blob *b,
 		  spdk_blob_op_with_handle_complete cb_fn, void *cb_arg)
@@ -2000,6 +2006,59 @@ lvol_inflate(void)
 	spdk_free_thread();
 }
 
+static void
+lvol_decouple_parent(void)
+{
+	struct lvol_ut_bs_dev dev;
+	struct spdk_lvs_opts opts;
+	int rc = 0;
+
+	init_dev(&dev);
+
+	spdk_allocate_thread(_lvol_send_msg, NULL, NULL, NULL, NULL);
+
+	spdk_lvs_opts_init(&opts);
+	snprintf(opts.name, sizeof(opts.name), "lvs");
+
+	g_lvserrno = -1;
+	rc = spdk_lvs_init(&dev.bs_dev, &opts, lvol_store_op_with_handle_complete, NULL);
+	CU_ASSERT(rc == 0);
+	CU_ASSERT(g_lvserrno == 0);
+	SPDK_CU_ASSERT_FATAL(g_lvol_store != NULL);
+
+	spdk_lvol_create(g_lvol_store, "lvol", 10, false, lvol_op_with_handle_complete, NULL);
+	CU_ASSERT(g_lvserrno == 0);
+	SPDK_CU_ASSERT_FATAL(g_lvol != NULL);
+
+	g_inflate_rc = -1;
+	spdk_lvol_decouple_parent(g_lvol, lvol_op_complete, NULL);
+	CU_ASSERT(g_lvolerrno != 0);
+
+	g_inflate_rc = 0;
+	spdk_lvol_decouple_parent(g_lvol, lvol_op_complete, NULL);
+	CU_ASSERT(g_lvolerrno == 0);
+
+	spdk_lvol_close(g_lvol, close_cb, NULL);
+	CU_ASSERT(g_lvserrno == 0);
+	spdk_lvol_destroy(g_lvol, destroy_cb, NULL);
+	CU_ASSERT(g_lvserrno == 0);
+
+	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;
+
+	free_dev(&dev);
+
+	/* Make sure that all references to the io_channel was closed after
+	 * inflate call
+	 */
+	CU_ASSERT(g_io_channel == NULL);
+
+	spdk_free_thread();
+}
+
 int main(int argc, char **argv)
 {
 	CU_pSuite	suite = NULL;
@@ -2042,7 +2101,8 @@ int main(int argc, char **argv)
 		CU_add_test(suite, "lvol_create_thin_provisioned", lvol_create_thin_provisioned) == NULL ||
 		CU_add_test(suite, "lvol_rename", lvol_rename) == NULL ||
 		CU_add_test(suite, "lvs_rename", lvs_rename) == NULL ||
-		CU_add_test(suite, "lvol_inflate", lvol_inflate) == NULL
+		CU_add_test(suite, "lvol_inflate", lvol_inflate) == NULL ||
+		CU_add_test(suite, "lvol_decouple_parent", lvol_decouple_parent) == NULL
 	) {
 		CU_cleanup_registry();
 		return CU_get_error();