lvol: add spdk_lvol_iter_immediate_clones()

Add an interator that calls a callback for each clone of a snapshot
volume. This follows the typical pattern of stopping iteration when the
callback returns non-zero.

Signed-off-by: Mike Gerdts <mgerdts@nvidia.com>
Change-Id: If88ad769b72a19ba0993303e89da107db8a6adfc
Reviewed-on: https://review.spdk.io/gerrit/c/spdk/spdk/+/17545
Reviewed-by: Ben Walker <benjamin.walker@intel.com>
Reviewed-by: Jim Harris <james.r.harris@intel.com>
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>

CHANGELOG.md

@@ -17,6 +17,10 @@ multiple readers.
 
 New function `spdk_env_get_main_core` was added.
 
+### lvol
+
+New API `spdk_lvol_iter_immediate_clones` was added to iterate the clones of an lvol.
+
 ### nvmf
 
 New `spdk_nvmf_request_copy_to/from_buf()` APIs have been added, which support

include/spdk/lvol.h

@@ -115,6 +115,15 @@ typedef void (*spdk_lvol_op_with_handle_complete)(void *cb_arg, struct spdk_lvol
  */
 typedef void (*spdk_lvol_op_complete)(void *cb_arg, int lvolerrno);
 
+/**
+ * Callback definition for spdk_lvol_iter_clones.
+ *
+ * \param lvol An iterated lvol.
+ * \param cb_arg Opaque context passed to spdk_lvol_iter_clone().
+ * \return 0 to continue iterating, any other value to stop iterating.
+ */
+typedef int (*spdk_lvol_iter_cb)(void *cb_arg, struct spdk_lvol *lvol);
+
 /**
  * Initialize lvolstore on given bs_bdev.
  *
@@ -260,6 +269,18 @@ void spdk_lvol_destroy(struct spdk_lvol *lvol, spdk_lvol_op_complete cb_fn, void
  */
 void spdk_lvol_close(struct spdk_lvol *lvol, spdk_lvol_op_complete cb_fn, void *cb_arg);
 
+/**
+ * Iterate clones of an lvol.
+ *
+ * Iteration stops if cb_fn(cb_arg, clone_lvol) returns non-zero.
+ *
+ * \param lvol Handle to lvol.
+ * \param cb_fn Function to call for each lvol that clones this lvol.
+ * \param cb_arg Context to pass wtih cb_fn.
+ * \return -ENOMEM if memory allocation failed, non-zero return from cb_fn(), or 0.
+ */
+int spdk_lvol_iter_immediate_clones(struct spdk_lvol *lvol, spdk_lvol_iter_cb cb_fn, void *cb_arg);
+
 /**
  * Get I/O channel of bdev associated with specified lvol.
  *

lib/lvol/lvol.c

@@ -2063,3 +2063,54 @@ spdk_lvs_notify_hotplug(const void *esnap_id, uint32_t id_len)
 
 	return ret;
 }
+
+int
+spdk_lvol_iter_immediate_clones(struct spdk_lvol *lvol, spdk_lvol_iter_cb cb_fn, void *cb_arg)
+{
+	struct spdk_lvol_store *lvs = lvol->lvol_store;
+	struct spdk_blob_store *bs = lvs->blobstore;
+	struct spdk_lvol *clone;
+	spdk_blob_id *ids;
+	size_t id_cnt = 0;
+	size_t i;
+	int rc;
+
+	rc = spdk_blob_get_clones(bs, lvol->blob_id, NULL, &id_cnt);
+	if (rc != -ENOMEM) {
+		/* -ENOMEM says id_cnt is valid, no other errors should be returned. */
+		assert(rc == 0);
+		return rc;
+	}
+
+	ids = calloc(id_cnt, sizeof(*ids));
+	if (ids == NULL) {
+		SPDK_ERRLOG("lvol %s: out of memory while iterating clones\n", lvol->unique_id);
+		return -ENOMEM;
+	}
+
+	rc = spdk_blob_get_clones(bs, lvol->blob_id, ids, &id_cnt);
+	if (rc != 0) {
+		SPDK_ERRLOG("lvol %s: unable to get clone blob IDs: %d\n", lvol->unique_id, rc);
+		free(ids);
+		return rc;
+	}
+
+	for (i = 0; i < id_cnt; i++) {
+		clone = lvs_get_lvol_by_blob_id(lvs, ids[i]);
+		if (clone == NULL) {
+			SPDK_NOTICELOG("lvol %s: unable to find clone lvol with blob id 0x%"
+				       PRIx64 "\n", lvol->unique_id, ids[i]);
+			continue;
+		}
+		rc = cb_fn(cb_arg, clone);
+		if (rc != 0) {
+			SPDK_DEBUGLOG(lvol, "lvol %s: iteration stopped when lvol %s (blob 0x%"
+				      PRIx64 ") returned %d\n", lvol->unique_id, clone->unique_id,
+				      ids[i], rc);
+			break;
+		}
+	}
+
+	free(ids);
+	return rc;
+}

lib/lvol/spdk_lvol.map

@@ -22,6 +22,7 @@
 	spdk_lvol_inflate;
 	spdk_lvol_decouple_parent;
 	spdk_lvol_create_esnap_clone;
+	spdk_lvol_iter_immediate_clones;
 
 	# internal functions
 	spdk_lvol_resize;

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

@@ -42,6 +42,8 @@ DEFINE_STUB(spdk_bdev_get_memory_domains, int, (struct spdk_bdev *bdev,
 DEFINE_STUB(spdk_blob_get_esnap_id, int,
 	    (struct spdk_blob *blob, const void **id, size_t *len), -ENOTSUP);
 DEFINE_STUB(spdk_blob_is_esnap_clone, bool, (const struct spdk_blob *blob), false);
+DEFINE_STUB(spdk_lvol_iter_immediate_clones, int,
+	    (struct spdk_lvol *lvol, spdk_lvol_iter_cb cb_fn, void *cb_arg), -ENOTSUP);
 
 const struct spdk_bdev_aliases_list *
 spdk_bdev_get_aliases(const struct spdk_bdev *bdev)

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

@@ -586,6 +586,24 @@ ut_cb_res_untouched(const struct ut_cb_res *res)
 	return !memcmp(&pristine, res, sizeof(pristine));
 }
 
+struct count_clones_ctx {
+	struct spdk_lvol *stop_on_lvol;
+	int stop_errno;
+	int count;
+};
+
+static int
+count_clones(void *_ctx, struct spdk_lvol *lvol)
+{
+	struct count_clones_ctx *ctx = _ctx;
+
+	if (ctx->stop_on_lvol == lvol) {
+		return ctx->stop_errno;
+	}
+	ctx->count++;
+	return 0;
+}
+
 static void
 lvs_init_unload_success(void)
 {
@@ -1695,6 +1713,109 @@ lvol_clone_fail(void)
 	free_dev(&dev);
 }
 
+static void
+lvol_iter_clones(void)
+{
+	struct lvol_ut_bs_dev dev;
+	struct spdk_lvol *lvol, *snap, *clone;
+	struct spdk_lvs_opts opts;
+	struct count_clones_ctx ctx = { 0 };
+	spdk_blob_id mock_clones[2];
+	int rc = 0;
+
+	init_dev(&dev);
+
+	spdk_lvs_opts_init(&opts);
+	snprintf(opts.name, sizeof(opts.name), "lvs");
+
+	g_spdk_blob_get_clones_ids = mock_clones;
+
+	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);
+
+	/* Create a volume */
+	spdk_lvol_create(g_lvol_store, "lvol", 10, true, LVOL_CLEAR_WITH_DEFAULT,
+			 lvol_op_with_handle_complete, NULL);
+	CU_ASSERT(g_lvserrno == 0);
+	SPDK_CU_ASSERT_FATAL(g_lvol != NULL);
+	lvol = g_lvol;
+
+	/* Create a snapshot of the volume */
+	spdk_lvol_create_snapshot(lvol, "snap", lvol_op_with_handle_complete, NULL);
+	CU_ASSERT(g_lvserrno == 0);
+	SPDK_CU_ASSERT_FATAL(g_lvol != NULL);
+	CU_ASSERT_STRING_EQUAL(g_lvol->name, "snap");
+	snap = g_lvol;
+
+	g_spdk_blob_get_clones_snap_id = snap->blob_id;
+	g_spdk_blob_get_clones_count = 1;
+	mock_clones[0] = lvol->blob_id;
+
+	/* The snapshot turned the lvol into a clone, so the snapshot now has one clone. */
+	memset(&ctx, 0, sizeof(ctx));
+	rc = spdk_lvol_iter_immediate_clones(snap, count_clones, &ctx);
+	CU_ASSERT(rc == 0);
+	CU_ASSERT(ctx.count == 1);
+
+	/* The snapshotted volume still has no clones. */
+	memset(&ctx, 0, sizeof(ctx));
+	rc = spdk_lvol_iter_immediate_clones(lvol, count_clones, &ctx);
+	CU_ASSERT(rc == 0);
+	CU_ASSERT(ctx.count == 0);
+
+	/* Iteration can be stopped and the return value is propagated. */
+	memset(&ctx, 0, sizeof(ctx));
+	ctx.stop_on_lvol = lvol;
+	ctx.stop_errno = 42;
+	rc = spdk_lvol_iter_immediate_clones(snap, count_clones, &ctx);
+	CU_ASSERT(rc == 42);
+	CU_ASSERT(ctx.count == 0);
+
+	/* Create a clone of the snapshot */
+	spdk_lvol_create_clone(snap, "clone", lvol_op_with_handle_complete, NULL);
+	CU_ASSERT(g_lvserrno == 0);
+	SPDK_CU_ASSERT_FATAL(g_lvol != NULL);
+	CU_ASSERT_STRING_EQUAL(g_lvol->name, "clone");
+	clone = g_lvol;
+
+	g_spdk_blob_get_clones_count = 2;
+	mock_clones[1] = clone->blob_id;
+
+	/* The snapshot now has two clones */
+	memset(&ctx, 0, sizeof(ctx));
+	rc = spdk_lvol_iter_immediate_clones(snap, count_clones, &ctx);
+	CU_ASSERT(rc == 0);
+	CU_ASSERT(ctx.count == 2);
+
+	/* Cleanup */
+	g_spdk_blob_get_clones_snap_id = 0xbad;
+	g_spdk_blob_get_clones_count = 0;
+	g_spdk_blob_get_clones_ids = NULL;
+
+	spdk_lvol_close(snap, op_complete, NULL);
+	CU_ASSERT(g_lvserrno == 0);
+
+	g_lvserrno = -1;
+	spdk_lvol_close(clone, op_complete, NULL);
+	CU_ASSERT(g_lvserrno == 0);
+
+	g_lvserrno = -1;
+	spdk_lvol_close(lvol, op_complete, NULL);
+	CU_ASSERT(g_lvserrno == 0);
+
+	g_lvserrno = -1;
+	rc = spdk_lvs_unload(g_lvol_store, op_complete, NULL);
+	CU_ASSERT(rc == 0);
+	CU_ASSERT(g_lvserrno == 0);
+	g_lvol_store = NULL;
+	g_lvol = NULL;
+
+	free_dev(&dev);
+}
+
 static void
 lvol_names(void)
 {
@@ -3076,6 +3197,7 @@ main(int argc, char **argv)
 	CU_ADD_TEST(suite, lvol_snapshot_fail);
 	CU_ADD_TEST(suite, lvol_clone);
 	CU_ADD_TEST(suite, lvol_clone_fail);
+	CU_ADD_TEST(suite, lvol_iter_clones);
 	CU_ADD_TEST(suite, lvol_refcnt);
 	CU_ADD_TEST(suite, lvol_names);
 	CU_ADD_TEST(suite, lvol_create_thin_provisioned);