bdev: add common partition helper functions

split, gpt and error all have a lot of similar code, and
are based on the presumption of one or more "partition" bdevs
on a "base" bdev.  This results in quite a bit of duplicated
code between these three bdev modules.  The error bdev
module does follow this same model - it just always has only
a one-to-one mapping between the error bdev and its base bdev.

As all of the modules move to allocating their own bdev_io
rather than allowing for adjusting an existing bdev_io and
resubmitting it, there will be even more duplicated code
between these modules.

So this patch adds a set of helper functions in the common
bdev library to eliminate all of this duplicated code.

This patch also moves the split module to use it - future
patches will also convert gpt and error.

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

Reviewed-on: https://review.gerrithub.io/376423
Tested-by: SPDK Automated Test System <sys_sgsw@intel.com>
Reviewed-by: Daniel Verkamp <daniel.verkamp@intel.com>

include/spdk_internal/bdev.h

@@ -44,6 +44,7 @@
 #include "spdk/bdev.h"
 #include "spdk/queue.h"
 #include "spdk/scsi_spec.h"
+#include "spdk/io_channel.h"
 
 /** \page block_backend_modules Block Device Backend Modules
  *
@@ -437,6 +438,50 @@ spdk_bdev_io_from_ctx(void *ctx)
 	       ((uintptr_t)ctx - offsetof(struct spdk_bdev_io, driver_ctx));
 }
 
+struct spdk_bdev_part_base {
+	struct spdk_bdev		*bdev;
+	struct spdk_bdev_desc		*desc;
+	uint32_t			ref;
+	uint32_t			channel_size;
+	bool				claimed;
+	struct spdk_bdev_module_if	*module;
+	struct spdk_bdev_fn_table	*fn_table;
+	struct bdev_part_tailq		*tailq;
+	spdk_io_channel_create_cb	ch_create_cb;
+	spdk_io_channel_destroy_cb	ch_destroy_cb;
+};
+
+struct spdk_bdev_part {
+	struct spdk_bdev		bdev;
+	struct spdk_bdev_part_base	*base;
+	uint64_t			offset_blocks;
+	TAILQ_ENTRY(spdk_bdev_part)	tailq;
+};
+
+struct spdk_bdev_part_channel {
+	struct spdk_bdev_part		*part;
+	struct spdk_io_channel		*base_ch;
+};
+
+typedef TAILQ_HEAD(bdev_part_tailq, spdk_bdev_part)	SPDK_BDEV_PART_TAILQ;
+
+void spdk_bdev_part_base_free(struct spdk_bdev_part_base *base);
+void spdk_bdev_part_free(struct spdk_bdev_part *part);
+void spdk_bdev_part_tailq_fini(struct bdev_part_tailq *tailq);
+void spdk_bdev_part_base_hotremove(struct spdk_bdev *base_bdev, struct bdev_part_tailq *tailq);
+int spdk_bdev_part_base_construct(struct spdk_bdev_part_base *base, struct spdk_bdev *bdev,
+				  spdk_bdev_remove_cb_t remove_cb,
+				  struct spdk_bdev_module_if *module,
+				  struct spdk_bdev_fn_table *fn_table,
+				  struct bdev_part_tailq *tailq,
+				  uint32_t channel_size,
+				  spdk_io_channel_create_cb ch_create_cb,
+				  spdk_io_channel_destroy_cb ch_destroy_cb);
+int spdk_bdev_part_construct(struct spdk_bdev_part *part, struct spdk_bdev_part_base *base,
+			     char *name, uint64_t offset_blocks, uint64_t num_blocks,
+			     char *product_name);
+void spdk_bdev_part_submit_request(struct spdk_bdev_part_channel *ch, struct spdk_bdev_io *bdev_io);
+
 #define SPDK_BDEV_MODULE_REGISTER(_name, init_fn, fini_fn, config_fn, ctx_size_fn, examine_fn)	\
 	static struct spdk_bdev_module_if _name ## _if = {					\
 	.name		= #_name,								\

lib/bdev/bdev.c

@@ -42,6 +42,7 @@
 #include "spdk/queue.h"
 #include "spdk/nvme_spec.h"
 #include "spdk/scsi_spec.h"
+#include "spdk/util.h"
 
 #include "spdk_internal/bdev.h"
 #include "spdk_internal/log.h"
@@ -1797,4 +1798,236 @@ spdk_bdev_module_list_add(struct spdk_bdev_module_if *bdev_module)
 	}
 }
 
+void
+spdk_bdev_part_base_free(struct spdk_bdev_part_base *base)
+{
+	assert(base->bdev);
+	assert(base->desc);
+	spdk_bdev_module_release_bdev(base->bdev);
+	spdk_bdev_close(base->desc);
+}
+
+void
+spdk_bdev_part_free(struct spdk_bdev_part *part)
+{
+	struct spdk_bdev_part_base *base;
+
+	assert(part);
+	assert(part->base);
+
+	base = part->base;
+	spdk_io_device_unregister(&part->base, NULL);
+	TAILQ_REMOVE(base->tailq, part, tailq);
+	free(part->bdev.name);
+
+	if (__sync_sub_and_fetch(&base->ref, 1) == 0) {
+		spdk_bdev_part_base_free(base);
+	}
+}
+
+void
+spdk_bdev_part_tailq_fini(struct bdev_part_tailq *tailq)
+{
+	struct spdk_bdev_part *part, *tmp;
+
+	TAILQ_FOREACH_SAFE(part, tailq, tailq, tmp) {
+		spdk_bdev_part_free(part);
+	}
+}
+
+void
+spdk_bdev_part_base_hotremove(struct spdk_bdev *base_bdev, struct bdev_part_tailq *tailq)
+{
+	struct spdk_bdev_part *part, *tmp;
+
+	TAILQ_FOREACH_SAFE(part, tailq, tailq, tmp) {
+		if (part->base->bdev == base_bdev) {
+			spdk_bdev_unregister(&part->bdev);
+		}
+	}
+}
+
+static bool
+spdk_bdev_part_io_type_supported(void *_part, enum spdk_bdev_io_type io_type)
+{
+	struct spdk_bdev_part *part = _part;
+
+	return part->base->bdev->fn_table->io_type_supported(part->base->bdev, io_type);
+}
+
+static struct spdk_io_channel *
+spdk_bdev_part_get_io_channel(void *_part)
+{
+	struct spdk_bdev_part *part = _part;
+
+	return spdk_get_io_channel(&part->base);
+}
+
+static void
+spdk_bdev_part_complete_io(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg)
+{
+	struct spdk_bdev_io *part_io = cb_arg;
+	int status = success ? SPDK_BDEV_IO_STATUS_SUCCESS : SPDK_BDEV_IO_STATUS_FAILED;
+
+	spdk_bdev_io_complete(part_io, status);
+	spdk_bdev_free_io(bdev_io);
+}
+
+void
+spdk_bdev_part_submit_request(struct spdk_bdev_part_channel *ch, struct spdk_bdev_io *bdev_io)
+{
+	struct spdk_bdev_part *part = ch->part;
+	struct spdk_io_channel *base_ch = ch->base_ch;
+	struct spdk_bdev_desc *base_desc = part->base->desc;
+	uint64_t offset;
+	int rc = 0;
+
+	/* Modify the I/O to adjust for the offset within the base bdev. */
+	switch (bdev_io->type) {
+	case SPDK_BDEV_IO_TYPE_READ:
+		offset = bdev_io->u.read.offset_blocks + part->offset_blocks;
+		rc = spdk_bdev_readv_blocks(base_desc, base_ch, bdev_io->u.read.iovs,
+					    bdev_io->u.read.iovcnt, offset,
+					    bdev_io->u.read.num_blocks, spdk_bdev_part_complete_io,
+					    bdev_io);
+		break;
+	case SPDK_BDEV_IO_TYPE_WRITE:
+		offset = bdev_io->u.write.offset_blocks + part->offset_blocks;
+		rc = spdk_bdev_writev_blocks(base_desc, base_ch, bdev_io->u.write.iovs,
+					     bdev_io->u.write.iovcnt, offset,
+					     bdev_io->u.write.num_blocks, spdk_bdev_part_complete_io,
+					     bdev_io);
+		break;
+	case SPDK_BDEV_IO_TYPE_WRITE_ZEROES:
+		offset = bdev_io->u.write.offset_blocks + part->offset_blocks;
+		rc = spdk_bdev_write_zeroes_blocks(base_desc, base_ch, offset, bdev_io->u.write.num_blocks,
+						   spdk_bdev_part_complete_io, bdev_io);
+		break;
+	case SPDK_BDEV_IO_TYPE_UNMAP:
+		offset = bdev_io->u.unmap.offset_blocks + part->offset_blocks;
+		rc = spdk_bdev_unmap_blocks(base_desc, base_ch, offset, bdev_io->u.unmap.num_blocks,
+					    spdk_bdev_part_complete_io, bdev_io);
+		break;
+	case SPDK_BDEV_IO_TYPE_FLUSH:
+		offset = bdev_io->u.flush.offset_blocks + part->offset_blocks;
+		rc = spdk_bdev_flush_blocks(base_desc, base_ch, offset, bdev_io->u.flush.num_blocks,
+					    spdk_bdev_part_complete_io, bdev_io);
+		break;
+	case SPDK_BDEV_IO_TYPE_RESET:
+		rc = spdk_bdev_reset(base_desc, base_ch,
+				     spdk_bdev_part_complete_io, bdev_io);
+		break;
+	default:
+		SPDK_ERRLOG("split: unknown I/O type %d\n", bdev_io->type);
+		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
+		return;
+	}
+
+	if (rc != 0) {
+		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
+	}
+}
+static int
+spdk_bdev_part_channel_create_cb(void *io_device, void *ctx_buf)
+{
+	struct spdk_bdev_part *part = SPDK_CONTAINEROF(io_device, struct spdk_bdev_part, base);
+	struct spdk_bdev_part_channel *ch = ctx_buf;
+
+	ch->part = part;
+	ch->base_ch = spdk_bdev_get_io_channel(part->base->desc);
+	if (ch->base_ch == NULL) {
+		return -1;
+	}
+
+	if (part->base->ch_create_cb) {
+		return part->base->ch_create_cb(io_device, ctx_buf);
+	} else {
+		return 0;
+	}
+}
+
+static void
+spdk_bdev_part_channel_destroy_cb(void *io_device, void *ctx_buf)
+{
+	struct spdk_bdev_part *part = SPDK_CONTAINEROF(io_device, struct spdk_bdev_part, base);
+	struct spdk_bdev_part_channel *ch = ctx_buf;
+
+	if (part->base->ch_destroy_cb) {
+		part->base->ch_destroy_cb(io_device, ctx_buf);
+	}
+	spdk_put_io_channel(ch->base_ch);
+}
+
+int
+spdk_bdev_part_base_construct(struct spdk_bdev_part_base *base, struct spdk_bdev *bdev,
+			      spdk_bdev_remove_cb_t remove_cb, struct spdk_bdev_module_if *module,
+			      struct spdk_bdev_fn_table *fn_table, struct bdev_part_tailq *tailq,
+			      uint32_t channel_size, spdk_io_channel_create_cb ch_create_cb,
+			      spdk_io_channel_destroy_cb ch_destroy_cb)
+{
+	int rc;
+
+	fn_table->get_io_channel = spdk_bdev_part_get_io_channel;
+	fn_table->io_type_supported = spdk_bdev_part_io_type_supported;
+
+	base->bdev = bdev;
+	base->ref = 0;
+	base->module = module;
+	base->fn_table = fn_table;
+	base->tailq = tailq;
+	base->claimed = false;
+	base->channel_size = channel_size;
+	base->ch_create_cb = ch_create_cb;
+	base->ch_destroy_cb = ch_destroy_cb;
+
+	rc = spdk_bdev_open(bdev, false, remove_cb, bdev, &base->desc);
+	if (rc) {
+		SPDK_ERRLOG("could not open bdev %s\n", spdk_bdev_get_name(bdev));
+		return -1;
+	}
+
+	return 0;
+}
+
+int
+spdk_bdev_part_construct(struct spdk_bdev_part *part, struct spdk_bdev_part_base *base,
+			 char *name, uint64_t offset_blocks, uint64_t num_blocks,
+			 char *product_name)
+{
+	part->bdev.name = name;
+	part->bdev.blocklen = base->bdev->blocklen;
+	part->bdev.blockcnt = num_blocks;
+	part->offset_blocks = offset_blocks;
+
+	part->bdev.write_cache = base->bdev->write_cache;
+	part->bdev.need_aligned_buffer = base->bdev->need_aligned_buffer;
+	part->bdev.product_name = product_name;
+	part->bdev.ctxt = part;
+	part->bdev.module = base->module;
+	part->bdev.fn_table = base->fn_table;
+
+	__sync_fetch_and_add(&base->ref, 1);
+	part->base = base;
+
+	if (!base->claimed) {
+		int rc;
+
+		rc = spdk_bdev_module_claim_bdev(base->bdev, base->desc, base->module);
+		if (rc) {
+			SPDK_ERRLOG("could not claim bdev %s\n", spdk_bdev_get_name(base->bdev));
+			free(part->bdev.name);
+			return -1;
+		}
+		base->claimed = true;
+	}
+
+	spdk_io_device_register(&part->base, spdk_bdev_part_channel_create_cb,
+				spdk_bdev_part_channel_destroy_cb,
+				base->channel_size);
+	spdk_vbdev_register(&part->bdev, &base->bdev, 1);
+	TAILQ_INSERT_TAIL(base->tailq, part, tailq);
+
+	return 0;
+}
+
 SPDK_LOG_REGISTER_TRACE_FLAG("bdev", SPDK_TRACE_BDEV)

lib/bdev/split/vbdev_split.c

@@ -50,172 +50,47 @@
 
 SPDK_DECLARE_BDEV_MODULE(split);
 
-/* Base block device split context */
+static SPDK_BDEV_PART_TAILQ g_split_disks = TAILQ_HEAD_INITIALIZER(g_split_disks);
-struct split_base {
+
-	struct spdk_bdev	*bdev;
+struct vbdev_split_channel {
-	struct spdk_bdev_desc	*desc;
+	struct spdk_bdev_part_channel	part_ch;
-	uint32_t		ref;
 };
 
-/* Context for each split virtual bdev */
-struct split_disk {
-	struct spdk_bdev	disk;
-	struct split_base	*base;
-	uint64_t		offset_blocks;
-	TAILQ_ENTRY(split_disk)	tailq;
-};
-
-static TAILQ_HEAD(, split_disk) g_split_disks = TAILQ_HEAD_INITIALIZER(g_split_disks);
-
-struct split_channel {
-	struct split_disk	*disk;
-	struct spdk_io_channel	*base_ch;
-};
-
-static void
-vbdev_split_complete_io(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg)
-{
-	struct spdk_bdev_io *split_io = cb_arg;
-	int status = success ? SPDK_BDEV_IO_STATUS_SUCCESS : SPDK_BDEV_IO_STATUS_FAILED;
-
-	spdk_bdev_io_complete(split_io, status);
-	spdk_bdev_free_io(bdev_io);
-}
-
-static void
-vbdev_split_submit_request(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io)
-{
-	struct split_channel *split_ch = spdk_io_channel_get_ctx(ch);
-	struct split_disk *split_disk = split_ch->disk;
-	struct spdk_io_channel *base_ch = split_ch->base_ch;
-	struct spdk_bdev_desc *base_desc = split_disk->base->desc;
-	uint32_t block_size = split_disk->disk.blocklen;
-	uint64_t offset_blocks;
-	int rc = 0;
-
-	/* Modify the I/O to adjust for the offset within the base bdev. */
-	switch (bdev_io->type) {
-	case SPDK_BDEV_IO_TYPE_READ:
-		offset_blocks = bdev_io->u.read.offset_blocks + split_disk->offset_blocks;
-		rc = spdk_bdev_readv(base_desc, base_ch, bdev_io->u.read.iovs,
-				     bdev_io->u.read.iovcnt, offset_blocks * block_size,
-				     bdev_io->u.read.num_blocks * block_size, vbdev_split_complete_io,
-				     bdev_io);
-		break;
-	case SPDK_BDEV_IO_TYPE_WRITE:
-		offset_blocks = bdev_io->u.write.offset_blocks + split_disk->offset_blocks;
-		rc = spdk_bdev_writev(base_desc, base_ch, bdev_io->u.write.iovs,
-				      bdev_io->u.write.iovcnt, offset_blocks * block_size,
-				      bdev_io->u.write.num_blocks * block_size, vbdev_split_complete_io,
-				      bdev_io);
-		break;
-	case SPDK_BDEV_IO_TYPE_UNMAP:
-		offset_blocks = bdev_io->u.unmap.offset_blocks + split_disk->offset_blocks;
-		rc = spdk_bdev_unmap(base_desc, base_ch, offset_blocks * block_size,
-				     bdev_io->u.unmap.num_blocks * block_size,
-				     vbdev_split_complete_io, bdev_io);
-		break;
-	case SPDK_BDEV_IO_TYPE_FLUSH:
-		offset_blocks = bdev_io->u.flush.offset_blocks + split_disk->offset_blocks;
-		rc = spdk_bdev_flush(base_desc, base_ch, offset_blocks * block_size,
-				     bdev_io->u.flush.num_blocks * block_size,
-				     vbdev_split_complete_io, bdev_io);
-		break;
-	case SPDK_BDEV_IO_TYPE_RESET:
-		rc = spdk_bdev_reset(split_disk->base->desc, base_ch,
-				     vbdev_split_complete_io, bdev_io);
-		break;
-	default:
-		SPDK_ERRLOG("split: unknown I/O type %d\n", bdev_io->type);
-		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
-		return;
-	}
-
-	if (rc != 0) {
-		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
-	}
-}
-
-static void
-vbdev_split_base_free(struct split_base *split_base)
-{
-	assert(split_base->bdev);
-	assert(split_base->desc);
-	spdk_bdev_module_release_bdev(split_base->bdev);
-	spdk_bdev_close(split_base->desc);
-	free(split_base);
-}
-
-static void
-vbdev_split_free(struct split_disk *split_disk)
-{
-	struct split_base *split_base;
-
-	assert(split_disk);
-	assert(split_disk->base);
-
-	split_base = split_disk->base;
-
-	spdk_io_device_unregister(&split_disk->base, NULL);
-	TAILQ_REMOVE(&g_split_disks, split_disk, tailq);
-	free(split_disk->disk.name);
-	free(split_disk);
-
-	if (__sync_sub_and_fetch(&split_base->ref, 1) == 0) {
-		vbdev_split_base_free(split_base);
-	}
-}
-
 static int
 vbdev_split_destruct(void *ctx)
 {
-	struct split_disk *split_disk = ctx;
+	struct spdk_bdev_part *part = ctx;
 
-	vbdev_split_free(split_disk);
+	spdk_bdev_part_free(part);
 	return 0;
 }
 
 static void
-vbdev_split_base_bdev_hotremove_cb(void *remove_ctx)
+vbdev_split_base_bdev_hotremove_cb(void *_base_bdev)
 {
-	struct spdk_bdev *base_bdev = remove_ctx;
+	spdk_bdev_part_base_hotremove(_base_bdev, &g_split_disks);
-	struct split_disk *split_disk, *tmp;
-
-	TAILQ_FOREACH_SAFE(split_disk, &g_split_disks, tailq, tmp) {
-		if (split_disk->base->bdev == base_bdev) {
-			spdk_bdev_unregister(&split_disk->disk);
-		}
-	}
 }
 
-static bool
+static void
-vbdev_split_io_type_supported(void *ctx, enum spdk_bdev_io_type io_type)
+vbdev_split_submit_request(struct spdk_io_channel *_ch, struct spdk_bdev_io *bdev_io)
 {
-	struct split_disk *split_disk = ctx;
+	struct vbdev_split_channel *ch = spdk_io_channel_get_ctx(_ch);
 
-	return split_disk->base->bdev->fn_table->io_type_supported(split_disk->base->bdev, io_type);
+	spdk_bdev_part_submit_request(&ch->part_ch, bdev_io);
-}
-
-static struct spdk_io_channel *
-vbdev_split_get_io_channel(void *ctx)
-{
-	struct split_disk *split_disk = ctx;
-
-	return spdk_get_io_channel(&split_disk->base);
 }
 
 static int
 vbdev_split_dump_config_json(void *ctx, struct spdk_json_write_ctx *w)
 {
-	struct split_disk *split_disk = ctx;
+	struct spdk_bdev_part *part = ctx;
 
 	spdk_json_write_name(w, "split");
 	spdk_json_write_object_begin(w);
 
 	spdk_json_write_name(w, "base_bdev");
-	spdk_json_write_string(w, spdk_bdev_get_name(split_disk->base->bdev));
+	spdk_json_write_string(w, spdk_bdev_get_name(part->base->bdev));
 	spdk_json_write_name(w, "offset_blocks");
-	spdk_json_write_uint64(w, split_disk->offset_blocks);
+	spdk_json_write_uint64(w, part->offset_blocks);
 
 	spdk_json_write_object_end(w);
 
@@ -224,35 +99,10 @@ vbdev_split_dump_config_json(void *ctx, struct spdk_json_write_ctx *w)
 
 static struct spdk_bdev_fn_table vbdev_split_fn_table = {
 	.destruct		= vbdev_split_destruct,
-	.io_type_supported	= vbdev_split_io_type_supported,
 	.submit_request		= vbdev_split_submit_request,
-	.get_io_channel		= vbdev_split_get_io_channel,
 	.dump_config_json	= vbdev_split_dump_config_json,
 };
 
-static int
-split_channel_create_cb(void *io_device, void *ctx_buf)
-{
-	struct split_disk *disk = SPDK_CONTAINEROF(io_device, struct split_disk, base);
-	struct split_channel *ch = ctx_buf;
-
-	ch->disk = disk;
-	ch->base_ch = spdk_bdev_get_io_channel(disk->base->desc);
-	if (ch->base_ch == NULL) {
-		return -1;
-	}
-
-	return 0;
-}
-
-static void
-split_channel_destroy_cb(void *io_device, void *ctx_buf)
-{
-	struct split_channel *ch = ctx_buf;
-
-	spdk_put_io_channel(ch->base_ch);
-}
-
 static int
 vbdev_split_create(struct spdk_bdev *base_bdev, uint64_t split_count, uint64_t split_size_mb)
 {
@@ -261,7 +111,10 @@ vbdev_split_create(struct spdk_bdev *base_bdev, uint64_t split_count, uint64_t s
 	uint64_t mb = 1024 * 1024;
 	uint64_t i;
 	int rc;
-	struct split_base *split_base;
+	char *name;
+	struct spdk_bdev_part_base *split_base;
+
+	assert(split_count > 0);
 
 	if (split_size_mb) {
 		if (((split_size_mb * mb) % base_bdev->blocklen) != 0) {
@@ -291,84 +144,51 @@ vbdev_split_create(struct spdk_bdev *base_bdev, uint64_t split_count, uint64_t s
 
 	split_base = calloc(1, sizeof(*split_base));
 	if (!split_base) {
-		SPDK_ERRLOG("Cannot alloc memory for split base pointer\n");
+		SPDK_ERRLOG("Cannot allocate bdev part base\n");
-		return -1;
-	}
-	split_base->bdev = base_bdev;
-	split_base->ref = 0;
-
-	rc = spdk_bdev_open(base_bdev, false, vbdev_split_base_bdev_hotremove_cb, base_bdev,
-			    &split_base->desc);
-	if (rc) {
-		SPDK_ERRLOG("could not open bdev %s\n", spdk_bdev_get_name(base_bdev));
-		free(split_base);
 		return -1;
 	}
 
-	rc = spdk_bdev_module_claim_bdev(base_bdev, split_base->desc, SPDK_GET_BDEV_MODULE(split));
+	rc = spdk_bdev_part_base_construct(split_base, base_bdev,
+					   vbdev_split_base_bdev_hotremove_cb,
+					   SPDK_GET_BDEV_MODULE(split), &vbdev_split_fn_table,
+					   &g_split_disks, sizeof(struct vbdev_split_channel),
+					   NULL, NULL);
 	if (rc) {
-		SPDK_ERRLOG("could not claim bdev %s\n", spdk_bdev_get_name(base_bdev));
+		SPDK_ERRLOG("Cannot construct bdev part base\n");
-		spdk_bdev_close(split_base->desc);
 		free(split_base);
 		return -1;
 	}
 
 	offset_blocks = 0;
 	for (i = 0; i < split_count; i++) {
-		struct split_disk *d;
+		struct spdk_bdev_part *d;
 
 		d = calloc(1, sizeof(*d));
-		if (!d) {
+		if (d == NULL) {
-			SPDK_ERRLOG("Memory allocation failure\n");
+			SPDK_ERRLOG("could not allocate bdev part\n");
-			rc = -1;
+			return -1;
-			goto cleanup;
 		}
 
-		/* Copy properties of the base bdev */
+		name = spdk_sprintf_alloc("%sp%" PRIu64, spdk_bdev_get_name(base_bdev), i);
-		d->disk.blocklen = base_bdev->blocklen;
+		if (!name) {
-		d->disk.write_cache = base_bdev->write_cache;
+			SPDK_ERRLOG("could not allocate name\n");
-		d->disk.need_aligned_buffer = base_bdev->need_aligned_buffer;
-
-		/* Append partition number to the base bdev's name, e.g. Malloc0 -> Malloc0p0 */
-		d->disk.name = spdk_sprintf_alloc("%sp%" PRIu64, spdk_bdev_get_name(base_bdev), i);
-		if (!d->disk.name) {
 			free(d);
-			rc = -ENOMEM;
+			return -1;
-			goto cleanup;
 		}
-		d->disk.product_name = "Split Disk";
-		d->offset_blocks = offset_blocks;
-		d->disk.blockcnt = split_size_blocks;
-		d->disk.ctxt = d;
-		d->disk.fn_table = &vbdev_split_fn_table;
-		d->disk.module = SPDK_GET_BDEV_MODULE(split);
 
-		SPDK_DEBUGLOG(SPDK_TRACE_VBDEV_SPLIT, "Split vbdev %s: base bdev: %s"
+		rc = spdk_bdev_part_construct(d, split_base, name, offset_blocks, split_size_blocks,
-			      " offset_blocks: %" PRIu64 "\n",
+					      "Split Disk");
-			      d->disk.name, spdk_bdev_get_name(base_bdev), d->offset_blocks);
+		if (rc) {
-
+			SPDK_ERRLOG("could not construct bdev part\n");
-		__sync_fetch_and_add(&split_base->ref, 1);
+			/* spdk_bdev_part_construct will free name if it fails */
-		d->base = split_base;
+			free(d);
-
+			return -1;
-		spdk_io_device_register(&d->base, split_channel_create_cb, split_channel_destroy_cb,
+		}
-					sizeof(struct split_channel));
-
-		spdk_vbdev_register(&d->disk, &base_bdev, 1);
-
-		TAILQ_INSERT_TAIL(&g_split_disks, d, tailq);
 
 		offset_blocks += split_size_blocks;
 	}
 
-	rc = 0;
+	return 0;
-
-cleanup:
-	if (split_base->ref == 0) {
-		/* If no split_disk instances were created, free the resources */
-		vbdev_split_base_free(split_base);
-	}
-
-	return rc;
 }
 
 static int
@@ -442,11 +262,7 @@ vbdev_split_examine(struct spdk_bdev *bdev)
 static void
 vbdev_split_fini(void)
 {
-	struct split_disk *split_disk, *tmp;
+	spdk_bdev_part_tailq_fini(&g_split_disks);
-
-	TAILQ_FOREACH_SAFE(split_disk, &g_split_disks, tailq, tmp) {
-		vbdev_split_free(split_disk);
-	}
 }
 
 static int

test/unit/lib/bdev/bdev.c/Makefile

@@ -34,7 +34,7 @@ SPDK_ROOT_DIR := $(abspath $(CURDIR)/../../../../..)
 include $(SPDK_ROOT_DIR)/mk/spdk.common.mk
 include $(SPDK_ROOT_DIR)/mk/spdk.app.mk
 
-SPDK_LIB_LIST = log cunit
+SPDK_LIB_LIST = log cunit util
 
 CFLAGS += -I$(SPDK_ROOT_DIR)/test
 CFLAGS += -I$(SPDK_ROOT_DIR)/lib/bdev