[ceph.git] / ceph / src / spdk / lib / bdev / split / vbdev_split.c

/*-
 *   BSD LICENSE
 *
 *   Copyright (c) Intel Corporation.
 *   All rights reserved.
 *
 *   Redistribution and use in source and binary forms, with or without
 *   modification, are permitted provided that the following conditions
 *   are met:
 *
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in
 *       the documentation and/or other materials provided with the
 *       distribution.
 *     * Neither the name of Intel Corporation nor the names of its
 *       contributors may be used to endorse or promote products derived
 *       from this software without specific prior written permission.
 *
 *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

/*
 * This is a simple example of a virtual block device that takes a single
 * bdev and slices it into multiple smaller bdevs.
 */

#include "vbdev_split.h"

#include "spdk/rpc.h"
#include "spdk/conf.h"
#include "spdk/endian.h"
#include "spdk/string.h"
#include "spdk/thread.h"
#include "spdk/util.h"

#include "spdk/bdev_module.h"
#include "spdk_internal/log.h"

struct spdk_vbdev_split_config {
	char *base_bdev;
	unsigned split_count;
	uint64_t split_size_mb;

	SPDK_BDEV_PART_TAILQ splits;
	struct spdk_bdev_part_base *split_base;
	bool removed;

	TAILQ_ENTRY(spdk_vbdev_split_config) tailq;
};

static TAILQ_HEAD(, spdk_vbdev_split_config) g_split_config = TAILQ_HEAD_INITIALIZER(
			g_split_config);

struct vbdev_split_channel {
	struct spdk_bdev_part_channel	part_ch;
};

struct vbdev_split_bdev_io {
	struct spdk_io_channel *ch;
	struct spdk_bdev_io *bdev_io;

	/* for bdev_io_wait */
	struct spdk_bdev_io_wait_entry bdev_io_wait;
};

static void vbdev_split_del_config(struct spdk_vbdev_split_config *cfg);

static int vbdev_split_init(void);
static void vbdev_split_fini(void);
static void vbdev_split_examine(struct spdk_bdev *bdev);
static int vbdev_split_config_json(struct spdk_json_write_ctx *w);
static int vbdev_split_get_ctx_size(void);

static void
vbdev_split_submit_request(struct spdk_io_channel *_ch, struct spdk_bdev_io *bdev_io);

static struct spdk_bdev_module split_if = {
	.name = "split",
	.module_init = vbdev_split_init,
	.module_fini = vbdev_split_fini,
	.get_ctx_size = vbdev_split_get_ctx_size,
	.examine_config = vbdev_split_examine,
	.config_json = vbdev_split_config_json,
};

SPDK_BDEV_MODULE_REGISTER(split, &split_if)

static void
vbdev_split_base_free(void *ctx)
{
	struct spdk_vbdev_split_config *cfg = ctx;

	cfg->split_base = NULL;
	if (cfg->removed) {
		vbdev_split_del_config(cfg);
	}
}

static int
vbdev_split_destruct(void *ctx)
{
	struct spdk_bdev_part *part = ctx;

	return spdk_bdev_part_free(part);
}

static void
vbdev_split_base_bdev_hotremove_cb(void *_part_base)
{
	struct spdk_bdev_part_base *part_base = _part_base;
	struct spdk_vbdev_split_config *cfg = spdk_bdev_part_base_get_ctx(part_base);

	spdk_bdev_part_base_hotremove(part_base, &cfg->splits);
}

static void
vbdev_split_resubmit_io(void *arg)
{
	struct vbdev_split_bdev_io *split_io = (struct vbdev_split_bdev_io *)arg;

	vbdev_split_submit_request(split_io->ch, split_io->bdev_io);
}

static void
vbdev_split_queue_io(struct vbdev_split_bdev_io *split_io)
{
	int rc;

	split_io->bdev_io_wait.bdev = split_io->bdev_io->bdev;
	split_io->bdev_io_wait.cb_fn = vbdev_split_resubmit_io;
	split_io->bdev_io_wait.cb_arg = split_io;

	rc = spdk_bdev_queue_io_wait(split_io->bdev_io->bdev,
				     split_io->ch, &split_io->bdev_io_wait);
	if (rc != 0) {
		SPDK_ERRLOG("Queue io failed in vbdev_split_queue_io, rc=%d\n", rc);
		spdk_bdev_io_complete(split_io->bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
	}
}

static void
vbdev_split_submit_request(struct spdk_io_channel *_ch, struct spdk_bdev_io *bdev_io)
{
	struct vbdev_split_channel *ch = spdk_io_channel_get_ctx(_ch);
	struct vbdev_split_bdev_io *io_ctx = (struct vbdev_split_bdev_io *)bdev_io->driver_ctx;
	int rc;

	rc = spdk_bdev_part_submit_request(&ch->part_ch, bdev_io);
	if (rc) {
		if (rc == -ENOMEM) {
			SPDK_DEBUGLOG(SPDK_LOG_VBDEV_SPLIT, "split: no memory, queue io.\n");
			io_ctx->ch = _ch;
			io_ctx->bdev_io = bdev_io;
			vbdev_split_queue_io(io_ctx);
		} else {
			SPDK_ERRLOG("split: error on io submission, rc=%d.\n", rc);
			spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
		}
	}
}

static int
vbdev_split_dump_info_json(void *ctx, struct spdk_json_write_ctx *w)
{
	struct spdk_bdev_part *part = ctx;
	struct spdk_bdev *split_base_bdev = spdk_bdev_part_get_base_bdev(part);
	uint64_t offset_blocks = spdk_bdev_part_get_offset_blocks(part);

	spdk_json_write_named_object_begin(w, "split");

	spdk_json_write_named_string(w, "base_bdev", spdk_bdev_get_name(split_base_bdev));
	spdk_json_write_named_uint64(w, "offset_blocks", offset_blocks);

	spdk_json_write_object_end(w);

	return 0;
}

static void
vbdev_split_write_config_json(struct spdk_bdev *bdev, struct spdk_json_write_ctx *w)
{
	/* No config per bdev needed */
}

static struct spdk_bdev_fn_table vbdev_split_fn_table = {
	.destruct		= vbdev_split_destruct,
	.submit_request		= vbdev_split_submit_request,
	.dump_info_json		= vbdev_split_dump_info_json,
	.write_config_json	= vbdev_split_write_config_json
};

static int
vbdev_split_create(struct spdk_vbdev_split_config *cfg)
{
	uint64_t split_size_blocks, offset_blocks;
	uint64_t split_count, max_split_count;
	uint64_t mb = 1024 * 1024;
	uint64_t i;
	int rc;
	char *name;
	struct spdk_bdev *base_bdev;
	struct bdev_part_tailq *split_base_tailq;

	assert(cfg->split_count > 0);

	base_bdev = spdk_bdev_get_by_name(cfg->base_bdev);
	if (!base_bdev) {
		return -ENODEV;
	}

	if (cfg->split_size_mb) {
		if (((cfg->split_size_mb * mb) % base_bdev->blocklen) != 0) {
			SPDK_ERRLOG("Split size %" PRIu64 " MB is not possible with block size "
				    "%" PRIu32 "\n",
				    cfg->split_size_mb, base_bdev->blocklen);
			return -EINVAL;
		}
		split_size_blocks = (cfg->split_size_mb * mb) / base_bdev->blocklen;
		SPDK_DEBUGLOG(SPDK_LOG_VBDEV_SPLIT, "Split size %" PRIu64 " MB specified by user\n",
			      cfg->split_size_mb);
	} else {
		split_size_blocks = base_bdev->blockcnt / cfg->split_count;
		SPDK_DEBUGLOG(SPDK_LOG_VBDEV_SPLIT, "Split size not specified by user\n");
	}

	max_split_count = base_bdev->blockcnt / split_size_blocks;
	split_count = cfg->split_count;
	if (split_count > max_split_count) {
		SPDK_WARNLOG("Split count %" PRIu64 " is greater than maximum possible split count "
			     "%" PRIu64 " - clamping\n", split_count, max_split_count);
		split_count = max_split_count;
	}

	SPDK_DEBUGLOG(SPDK_LOG_VBDEV_SPLIT, "base_bdev: %s split_count: %" PRIu64
		      " split_size_blocks: %" PRIu64 "\n",
		      spdk_bdev_get_name(base_bdev), split_count, split_size_blocks);

	TAILQ_INIT(&cfg->splits);
	cfg->split_base = spdk_bdev_part_base_construct(base_bdev,
			  vbdev_split_base_bdev_hotremove_cb,
			  &split_if, &vbdev_split_fn_table,
			  &cfg->splits, vbdev_split_base_free, cfg,
			  sizeof(struct vbdev_split_channel), NULL, NULL);
	if (!cfg->split_base) {
		SPDK_ERRLOG("Cannot construct bdev part base\n");
		return -ENOMEM;
	}

	offset_blocks = 0;
	for (i = 0; i < split_count; i++) {
		struct spdk_bdev_part *d;

		d = calloc(1, sizeof(*d));
		if (d == NULL) {
			SPDK_ERRLOG("could not allocate bdev part\n");
			rc = -ENOMEM;
			goto err;
		}

		name = spdk_sprintf_alloc("%sp%" PRIu64, cfg->base_bdev, i);
		if (!name) {
			SPDK_ERRLOG("could not allocate name\n");
			free(d);
			rc = -ENOMEM;
			goto err;
		}

		rc = spdk_bdev_part_construct(d, cfg->split_base, name, offset_blocks, split_size_blocks,
					      "Split Disk");
		free(name);
		if (rc) {
			SPDK_ERRLOG("could not construct bdev part\n");
			/* spdk_bdev_part_construct will free name if it fails */
			free(d);
			rc = -ENOMEM;
			goto err;
		}

		offset_blocks += split_size_blocks;
	}

	return 0;
err:
	split_base_tailq = spdk_bdev_part_base_get_tailq(cfg->split_base);
	cfg->removed = true;
	spdk_bdev_part_base_hotremove(cfg->split_base, split_base_tailq);
	return rc;
}

static void
vbdev_split_del_config(struct spdk_vbdev_split_config *cfg)
{
	TAILQ_REMOVE(&g_split_config, cfg, tailq);
	free(cfg->base_bdev);
	free(cfg);
}

static void
vbdev_split_destruct_config(struct spdk_vbdev_split_config *cfg)
{
	struct bdev_part_tailq *split_base_tailq;

	cfg->removed = true;
	if (cfg->split_base != NULL) {
		split_base_tailq = spdk_bdev_part_base_get_tailq(cfg->split_base);
		spdk_bdev_part_base_hotremove(cfg->split_base, split_base_tailq);
	} else {
		vbdev_split_del_config(cfg);
	}
}

static void
vbdev_split_clear_config(void)
{
	struct spdk_vbdev_split_config *cfg, *tmp_cfg;

	TAILQ_FOREACH_SAFE(cfg, &g_split_config, tailq, tmp_cfg) {
		vbdev_split_destruct_config(cfg);
	}
}

static struct spdk_vbdev_split_config *
vbdev_split_config_find_by_base_name(const char *base_bdev_name)
{
	struct spdk_vbdev_split_config *cfg;

	TAILQ_FOREACH(cfg, &g_split_config, tailq) {
		if (strcmp(cfg->base_bdev, base_bdev_name) == 0) {
			return cfg;
		}
	}

	return NULL;
}

static int
vbdev_split_add_config(const char *base_bdev_name, unsigned split_count, uint64_t split_size,
		       struct spdk_vbdev_split_config **config)
{
	struct spdk_vbdev_split_config *cfg;
	assert(base_bdev_name);

	if (base_bdev_name == NULL) {
		SPDK_ERRLOG("Split bdev config: no base bdev provided.");
		return -EINVAL;
	}

	if (split_count == 0) {
		SPDK_ERRLOG("Split bdev config: split_count can't be 0.");
		return -EINVAL;
	}

	/* Check if we already have 'base_bdev_name' registered in config */
	cfg = vbdev_split_config_find_by_base_name(base_bdev_name);
	if (cfg) {
		SPDK_ERRLOG("Split bdev config for base bdev '%s' already exist.", base_bdev_name);
		return -EEXIST;
	}

	cfg = calloc(1, sizeof(*cfg));
	if (!cfg) {
		SPDK_ERRLOG("calloc(): Out of memory");
		return -ENOMEM;
	}

	cfg->base_bdev = strdup(base_bdev_name);
	if (!cfg->base_bdev) {
		SPDK_ERRLOG("strdup(): Out of memory");
		free(cfg);
		return -ENOMEM;
	}

	cfg->split_count = split_count;
	cfg->split_size_mb = split_size;
	TAILQ_INSERT_TAIL(&g_split_config, cfg, tailq);
	if (config) {
		*config = cfg;
	}

	return 0;
}

static int
vbdev_split_init(void)
{

	struct spdk_conf_section *sp;
	const char *base_bdev_name;
	const char *split_count_str;
	const char *split_size_str;
	int rc, i, split_count, split_size;

	sp = spdk_conf_find_section(NULL, "Split");
	if (sp == NULL) {
		return 0;
	}

	for (i = 0; ; i++) {
		if (!spdk_conf_section_get_nval(sp, "Split", i)) {
			break;
		}

		base_bdev_name = spdk_conf_section_get_nmval(sp, "Split", i, 0);
		if (!base_bdev_name) {
			SPDK_ERRLOG("Split configuration missing bdev name\n");
			rc = -EINVAL;
			goto err;
		}

		split_count_str = spdk_conf_section_get_nmval(sp, "Split", i, 1);
		if (!split_count_str) {
			SPDK_ERRLOG("Split configuration missing split count\n");
			rc = -EINVAL;
			goto err;
		}

		split_count = spdk_strtol(split_count_str, 10);
		if (split_count < 1) {
			SPDK_ERRLOG("Invalid Split count %d\n", split_count);
			rc = -EINVAL;
			goto err;
		}

		/* Optional split size in MB */
		split_size = 0;
		split_size_str = spdk_conf_section_get_nmval(sp, "Split", i, 2);
		if (split_size_str) {
			split_size = spdk_strtol(split_size_str, 10);
			if (split_size <= 0) {
				SPDK_ERRLOG("Invalid Split size %d\n", split_size);
				rc = -EINVAL;
				goto err;
			}
		}

		rc = vbdev_split_add_config(base_bdev_name, split_count, split_size, NULL);
		if (rc != 0) {
			goto err;
		}
	}

	return 0;
err:
	vbdev_split_clear_config();
	return rc;
}

static void
vbdev_split_fini(void)
{
	vbdev_split_clear_config();
}

static void
vbdev_split_examine(struct spdk_bdev *bdev)
{
	struct spdk_vbdev_split_config *cfg = vbdev_split_config_find_by_base_name(bdev->name);

	if (cfg != NULL && cfg->removed == false) {
		assert(cfg->split_base == NULL);

		if (vbdev_split_create(cfg)) {
			SPDK_ERRLOG("could not split bdev %s\n", bdev->name);
		}
	}
	spdk_bdev_module_examine_done(&split_if);
}

static int
vbdev_split_config_json(struct spdk_json_write_ctx *w)
{
	struct spdk_vbdev_split_config *cfg;

	TAILQ_FOREACH(cfg, &g_split_config, tailq) {
		spdk_json_write_object_begin(w);

		spdk_json_write_named_string(w, "method", "construct_split_vbdev");

		spdk_json_write_named_object_begin(w, "params");
		spdk_json_write_named_string(w, "base_bdev", cfg->base_bdev);
		spdk_json_write_named_uint32(w, "split_count", cfg->split_count);
		spdk_json_write_named_uint64(w, "split_size_mb", cfg->split_size_mb);
		spdk_json_write_object_end(w);

		spdk_json_write_object_end(w);
	}

	return 0;
}

int
create_vbdev_split(const char *base_bdev_name, unsigned split_count, uint64_t split_size_mb)
{
	int rc;
	struct spdk_vbdev_split_config *cfg;

	rc = vbdev_split_add_config(base_bdev_name, split_count, split_size_mb, &cfg);
	if (rc) {
		return rc;
	}

	rc = vbdev_split_create(cfg);
	if (rc == -ENODEV) {
		/* It is ok if base bdev does not exist yet. */
		rc = 0;
	}

	return rc;
}

int
spdk_vbdev_split_destruct(const char *base_bdev_name)
{
	struct spdk_vbdev_split_config *cfg = vbdev_split_config_find_by_base_name(base_bdev_name);

	if (!cfg) {
		SPDK_ERRLOG("Split configuration for '%s' not found\n", base_bdev_name);
		return -ENOENT;
	}

	vbdev_split_destruct_config(cfg);
	return 0;
}

struct spdk_bdev_part_base *
spdk_vbdev_split_get_part_base(struct spdk_bdev *bdev)
{
	struct spdk_vbdev_split_config *cfg;

	cfg = vbdev_split_config_find_by_base_name(spdk_bdev_get_name(bdev));

	if (cfg == NULL) {
		return NULL;
	}

	return cfg->split_base;
}

/*
 * During init we'll be asked how much memory we'd like passed to us
 * in bev_io structures as context. Here's where we specify how
 * much context we want per IO.
 */
static int
vbdev_split_get_ctx_size(void)
{
	return sizeof(struct vbdev_split_bdev_io);
}

SPDK_LOG_REGISTER_COMPONENT("vbdev_split", SPDK_LOG_VBDEV_SPLIT)
Commit	Line	Data
7c673cae FG	1	/*-
	2	* BSD LICENSE
	3	*
	4	* Copyright (c) Intel Corporation.
	5	* All rights reserved.
	6	*
	7	* Redistribution and use in source and binary forms, with or without
	8	* modification, are permitted provided that the following conditions
	9	* are met:
	10	*
	11	* * Redistributions of source code must retain the above copyright
	12	* notice, this list of conditions and the following disclaimer.
	13	* * Redistributions in binary form must reproduce the above copyright
	14	* notice, this list of conditions and the following disclaimer in
	15	* the documentation and/or other materials provided with the
	16	* distribution.
	17	* * Neither the name of Intel Corporation nor the names of its
	18	* contributors may be used to endorse or promote products derived
	19	* from this software without specific prior written permission.
	20	*
	21	* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
	22	* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
	23	* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
	24	* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
	25	* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
	26	* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
	27	* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
	28	* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
	29	* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
	30	* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
	31	* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
	32	*/
	33
	34	/*
	35	* This is a simple example of a virtual block device that takes a single
	36	* bdev and slices it into multiple smaller bdevs.
	37	*/
	38
11fdf7f2	39	#include "vbdev_split.h"
7c673cae FG	40
	41	#include "spdk/rpc.h"
	42	#include "spdk/conf.h"
	43	#include "spdk/endian.h"
11fdf7f2 TL	44	#include "spdk/string.h"
	45	#include "spdk/thread.h"
	46	#include "spdk/util.h"
7c673cae	47
11fdf7f2	48	#include "spdk/bdev_module.h"
7c673cae FG	49	#include "spdk_internal/log.h"
7c673cae FG	50
11fdf7f2 TL	51	struct spdk_vbdev_split_config {
	52	char *base_bdev;
	53	unsigned split_count;
	54	uint64_t split_size_mb;
	55
9f95a23c	56	SPDK_BDEV_PART_TAILQ splits;
11fdf7f2 TL	57	struct spdk_bdev_part_base *split_base;
	58	bool removed;
	59
	60	TAILQ_ENTRY(spdk_vbdev_split_config) tailq;
7c673cae FG	61	};
7c673cae FG	62
11fdf7f2 TL	63	static TAILQ_HEAD(, spdk_vbdev_split_config) g_split_config = TAILQ_HEAD_INITIALIZER(
11fdf7f2 TL	64	g_split_config);
11fdf7f2 TL	65
	66	struct vbdev_split_channel {
	67	struct spdk_bdev_part_channel part_ch;
7c673cae FG	68	};
7c673cae FG	69
11fdf7f2 TL	70	struct vbdev_split_bdev_io {
	71	struct spdk_io_channel *ch;
	72	struct spdk_bdev_io *bdev_io;
7c673cae	73
11fdf7f2 TL	74	/* for bdev_io_wait */
	75	struct spdk_bdev_io_wait_entry bdev_io_wait;
	76	};
7c673cae	77
11fdf7f2 TL	78	static void vbdev_split_del_config(struct spdk_vbdev_split_config *cfg);
	79
	80	static int vbdev_split_init(void);
	81	static void vbdev_split_fini(void);
	82	static void vbdev_split_examine(struct spdk_bdev *bdev);
	83	static int vbdev_split_config_json(struct spdk_json_write_ctx *w);
	84	static int vbdev_split_get_ctx_size(void);
7c673cae FG	85
7c673cae FG	86	static void
11fdf7f2 TL	87	vbdev_split_submit_request(struct spdk_io_channel _ch, struct spdk_bdev_io bdev_io);
	88
	89	static struct spdk_bdev_module split_if = {
	90	.name = "split",
	91	.module_init = vbdev_split_init,
	92	.module_fini = vbdev_split_fini,
	93	.get_ctx_size = vbdev_split_get_ctx_size,
	94	.examine_config = vbdev_split_examine,
	95	.config_json = vbdev_split_config_json,
	96	};
7c673cae	97
9f95a23c	98	SPDK_BDEV_MODULE_REGISTER(split, &split_if)
7c673cae FG	99
7c673cae FG	100	static void
11fdf7f2	101	vbdev_split_base_free(void *ctx)
7c673cae	102	{
11fdf7f2	103	struct spdk_vbdev_split_config *cfg = ctx;
7c673cae	104
11fdf7f2 TL	105	cfg->split_base = NULL;
	106	if (cfg->removed) {
	107	vbdev_split_del_config(cfg);
7c673cae FG	108	}
	109	}
	110
11fdf7f2 TL	111	static int
11fdf7f2 TL	112	vbdev_split_destruct(void *ctx)
7c673cae	113	{
11fdf7f2	114	struct spdk_bdev_part *part = ctx;
7c673cae	115
11fdf7f2	116	return spdk_bdev_part_free(part);
7c673cae FG	117	}
	118
	119	static void
9f95a23c	120	vbdev_split_base_bdev_hotremove_cb(void *_part_base)
7c673cae	121	{
9f95a23c TL	122	struct spdk_bdev_part_base *part_base = _part_base;
	123	struct spdk_vbdev_split_config *cfg = spdk_bdev_part_base_get_ctx(part_base);
	124
	125	spdk_bdev_part_base_hotremove(part_base, &cfg->splits);
7c673cae FG	126	}
	127
	128	static void
11fdf7f2	129	vbdev_split_resubmit_io(void *arg)
7c673cae	130	{
11fdf7f2 TL	131	struct vbdev_split_bdev_io split_io = (struct vbdev_split_bdev_io )arg;
	132
	133	vbdev_split_submit_request(split_io->ch, split_io->bdev_io);
7c673cae FG	134	}
	135
	136	static void
11fdf7f2	137	vbdev_split_queue_io(struct vbdev_split_bdev_io *split_io)
7c673cae	138	{
11fdf7f2	139	int rc;
7c673cae	140
11fdf7f2 TL	141	split_io->bdev_io_wait.bdev = split_io->bdev_io->bdev;
	142	split_io->bdev_io_wait.cb_fn = vbdev_split_resubmit_io;
	143	split_io->bdev_io_wait.cb_arg = split_io;
7c673cae	144
11fdf7f2 TL	145	rc = spdk_bdev_queue_io_wait(split_io->bdev_io->bdev,
	146	split_io->ch, &split_io->bdev_io_wait);
	147	if (rc != 0) {
	148	SPDK_ERRLOG("Queue io failed in vbdev_split_queue_io, rc=%d\n", rc);
	149	spdk_bdev_io_complete(split_io->bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
7c673cae FG	150	}
	151	}
	152
11fdf7f2 TL	153	static void
11fdf7f2 TL	154	vbdev_split_submit_request(struct spdk_io_channel _ch, struct spdk_bdev_io bdev_io)
7c673cae	155	{
11fdf7f2 TL	156	struct vbdev_split_channel *ch = spdk_io_channel_get_ctx(_ch);
	157	struct vbdev_split_bdev_io io_ctx = (struct vbdev_split_bdev_io )bdev_io->driver_ctx;
	158	int rc;
7c673cae	159
11fdf7f2 TL	160	rc = spdk_bdev_part_submit_request(&ch->part_ch, bdev_io);
	161	if (rc) {
	162	if (rc == -ENOMEM) {
	163	SPDK_DEBUGLOG(SPDK_LOG_VBDEV_SPLIT, "split: no memory, queue io.\n");
	164	io_ctx->ch = _ch;
	165	io_ctx->bdev_io = bdev_io;
	166	vbdev_split_queue_io(io_ctx);
	167	} else {
	168	SPDK_ERRLOG("split: error on io submission, rc=%d.\n", rc);
	169	spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
	170	}
	171	}
7c673cae FG	172	}
	173
	174	static int
11fdf7f2	175	vbdev_split_dump_info_json(void ctx, struct spdk_json_write_ctx w)
7c673cae	176	{
11fdf7f2 TL	177	struct spdk_bdev_part *part = ctx;
	178	struct spdk_bdev *split_base_bdev = spdk_bdev_part_get_base_bdev(part);
	179	uint64_t offset_blocks = spdk_bdev_part_get_offset_blocks(part);
7c673cae	180
9f95a23c	181	spdk_json_write_named_object_begin(w, "split");
7c673cae	182
9f95a23c TL	183	spdk_json_write_named_string(w, "base_bdev", spdk_bdev_get_name(split_base_bdev));
9f95a23c TL	184	spdk_json_write_named_uint64(w, "offset_blocks", offset_blocks);
7c673cae FG	185
	186	spdk_json_write_object_end(w);
	187
	188	return 0;
	189	}
	190
11fdf7f2 TL	191	static void
	192	vbdev_split_write_config_json(struct spdk_bdev bdev, struct spdk_json_write_ctx w)
	193	{
	194	/* No config per bdev needed */
	195	}
	196
7c673cae FG	197	static struct spdk_bdev_fn_table vbdev_split_fn_table = {
7c673cae FG	198	.destruct = vbdev_split_destruct,
7c673cae	199	.submit_request = vbdev_split_submit_request,
11fdf7f2 TL	200	.dump_info_json = vbdev_split_dump_info_json,
11fdf7f2 TL	201	.write_config_json = vbdev_split_write_config_json
7c673cae FG	202	};
	203
	204	static int
11fdf7f2	205	vbdev_split_create(struct spdk_vbdev_split_config *cfg)
7c673cae	206	{
11fdf7f2 TL	207	uint64_t split_size_blocks, offset_blocks;
11fdf7f2 TL	208	uint64_t split_count, max_split_count;
7c673cae FG	209	uint64_t mb = 1024 * 1024;
	210	uint64_t i;
	211	int rc;
11fdf7f2 TL	212	char *name;
11fdf7f2 TL	213	struct spdk_bdev *base_bdev;
11fdf7f2	214	struct bdev_part_tailq *split_base_tailq;
7c673cae	215
11fdf7f2 TL	216	assert(cfg->split_count > 0);
	217
	218	base_bdev = spdk_bdev_get_by_name(cfg->base_bdev);
	219	if (!base_bdev) {
	220	return -ENODEV;
7c673cae FG	221	}
7c673cae FG	222
11fdf7f2 TL	223	if (cfg->split_size_mb) {
11fdf7f2 TL	224	if (((cfg->split_size_mb * mb) % base_bdev->blocklen) != 0) {
7c673cae FG	225	SPDK_ERRLOG("Split size %" PRIu64 " MB is not possible with block size "
7c673cae FG	226	"%" PRIu32 "\n",
11fdf7f2 TL	227	cfg->split_size_mb, base_bdev->blocklen);
11fdf7f2 TL	228	return -EINVAL;
7c673cae	229	}
11fdf7f2 TL	230	split_size_blocks = (cfg->split_size_mb * mb) / base_bdev->blocklen;
	231	SPDK_DEBUGLOG(SPDK_LOG_VBDEV_SPLIT, "Split size %" PRIu64 " MB specified by user\n",
	232	cfg->split_size_mb);
7c673cae	233	} else {
11fdf7f2 TL	234	split_size_blocks = base_bdev->blockcnt / cfg->split_count;
11fdf7f2 TL	235	SPDK_DEBUGLOG(SPDK_LOG_VBDEV_SPLIT, "Split size not specified by user\n");
7c673cae FG	236	}
7c673cae FG	237
7c673cae	238	max_split_count = base_bdev->blockcnt / split_size_blocks;
11fdf7f2	239	split_count = cfg->split_count;
7c673cae FG	240	if (split_count > max_split_count) {
	241	SPDK_WARNLOG("Split count %" PRIu64 " is greater than maximum possible split count "
	242	"%" PRIu64 " - clamping\n", split_count, max_split_count);
	243	split_count = max_split_count;
	244	}
	245
11fdf7f2 TL	246	SPDK_DEBUGLOG(SPDK_LOG_VBDEV_SPLIT, "base_bdev: %s split_count: %" PRIu64
	247	" split_size_blocks: %" PRIu64 "\n",
	248	spdk_bdev_get_name(base_bdev), split_count, split_size_blocks);
	249
9f95a23c	250	TAILQ_INIT(&cfg->splits);
11fdf7f2 TL	251	cfg->split_base = spdk_bdev_part_base_construct(base_bdev,
	252	vbdev_split_base_bdev_hotremove_cb,
	253	&split_if, &vbdev_split_fn_table,
9f95a23c	254	&cfg->splits, vbdev_split_base_free, cfg,
11fdf7f2 TL	255	sizeof(struct vbdev_split_channel), NULL, NULL);
	256	if (!cfg->split_base) {
	257	SPDK_ERRLOG("Cannot construct bdev part base\n");
	258	return -ENOMEM;
7c673cae	259	}
7c673cae	260
7c673cae FG	261	offset_blocks = 0;
7c673cae FG	262	for (i = 0; i < split_count; i++) {
11fdf7f2	263	struct spdk_bdev_part *d;
7c673cae FG	264
7c673cae FG	265	d = calloc(1, sizeof(*d));
11fdf7f2 TL	266	if (d == NULL) {
	267	SPDK_ERRLOG("could not allocate bdev part\n");
	268	rc = -ENOMEM;
	269	goto err;
	270	}
	271
	272	name = spdk_sprintf_alloc("%sp%" PRIu64, cfg->base_bdev, i);
	273	if (!name) {
	274	SPDK_ERRLOG("could not allocate name\n");
	275	free(d);
	276	rc = -ENOMEM;
	277	goto err;
7c673cae FG	278	}
7c673cae FG	279
11fdf7f2 TL	280	rc = spdk_bdev_part_construct(d, cfg->split_base, name, offset_blocks, split_size_blocks,
	281	"Split Disk");
	282	free(name);
	283	if (rc) {
	284	SPDK_ERRLOG("could not construct bdev part\n");
	285	/* spdk_bdev_part_construct will free name if it fails */
	286	free(d);
	287	rc = -ENOMEM;
	288	goto err;
	289	}
7c673cae	290
11fdf7f2 TL	291	offset_blocks += split_size_blocks;
11fdf7f2 TL	292	}
7c673cae	293
11fdf7f2 TL	294	return 0;
11fdf7f2 TL	295	err:
11fdf7f2 TL	296	split_base_tailq = spdk_bdev_part_base_get_tailq(cfg->split_base);
11fdf7f2 TL	297	cfg->removed = true;
9f95a23c	298	spdk_bdev_part_base_hotremove(cfg->split_base, split_base_tailq);
11fdf7f2 TL	299	return rc;
11fdf7f2 TL	300	}
7c673cae	301
11fdf7f2 TL	302	static void
	303	vbdev_split_del_config(struct spdk_vbdev_split_config *cfg)
	304	{
	305	TAILQ_REMOVE(&g_split_config, cfg, tailq);
	306	free(cfg->base_bdev);
	307	free(cfg);
	308	}
7c673cae	309
11fdf7f2 TL	310	static void
	311	vbdev_split_destruct_config(struct spdk_vbdev_split_config *cfg)
	312	{
11fdf7f2 TL	313	struct bdev_part_tailq *split_base_tailq;
	314
	315	cfg->removed = true;
	316	if (cfg->split_base != NULL) {
11fdf7f2	317	split_base_tailq = spdk_bdev_part_base_get_tailq(cfg->split_base);
9f95a23c	318	spdk_bdev_part_base_hotremove(cfg->split_base, split_base_tailq);
11fdf7f2 TL	319	} else {
	320	vbdev_split_del_config(cfg);
	321	}
	322	}
7c673cae	323
11fdf7f2 TL	324	static void
	325	vbdev_split_clear_config(void)
	326	{
	327	struct spdk_vbdev_split_config cfg, tmp_cfg;
7c673cae	328
11fdf7f2 TL	329	TAILQ_FOREACH_SAFE(cfg, &g_split_config, tailq, tmp_cfg) {
	330	vbdev_split_destruct_config(cfg);
	331	}
	332	}
	333
	334	static struct spdk_vbdev_split_config *
	335	vbdev_split_config_find_by_base_name(const char *base_bdev_name)
	336	{
	337	struct spdk_vbdev_split_config *cfg;
	338
	339	TAILQ_FOREACH(cfg, &g_split_config, tailq) {
	340	if (strcmp(cfg->base_bdev, base_bdev_name) == 0) {
	341	return cfg;
	342	}
7c673cae FG	343	}
7c673cae FG	344
11fdf7f2 TL	345	return NULL;
11fdf7f2 TL	346	}
7c673cae	347
11fdf7f2 TL	348	static int
	349	vbdev_split_add_config(const char *base_bdev_name, unsigned split_count, uint64_t split_size,
	350	struct spdk_vbdev_split_config **config)
	351	{
	352	struct spdk_vbdev_split_config *cfg;
	353	assert(base_bdev_name);
	354
	355	if (base_bdev_name == NULL) {
	356	SPDK_ERRLOG("Split bdev config: no base bdev provided.");
	357	return -EINVAL;
7c673cae FG	358	}
7c673cae FG	359
11fdf7f2 TL	360	if (split_count == 0) {
	361	SPDK_ERRLOG("Split bdev config: split_count can't be 0.");
	362	return -EINVAL;
	363	}
	364
	365	/* Check if we already have 'base_bdev_name' registered in config */
	366	cfg = vbdev_split_config_find_by_base_name(base_bdev_name);
	367	if (cfg) {
	368	SPDK_ERRLOG("Split bdev config for base bdev '%s' already exist.", base_bdev_name);
	369	return -EEXIST;
	370	}
	371
	372	cfg = calloc(1, sizeof(*cfg));
	373	if (!cfg) {
	374	SPDK_ERRLOG("calloc(): Out of memory");
	375	return -ENOMEM;
	376	}
	377
	378	cfg->base_bdev = strdup(base_bdev_name);
	379	if (!cfg->base_bdev) {
	380	SPDK_ERRLOG("strdup(): Out of memory");
	381	free(cfg);
	382	return -ENOMEM;
	383	}
	384
	385	cfg->split_count = split_count;
	386	cfg->split_size_mb = split_size;
	387	TAILQ_INSERT_TAIL(&g_split_config, cfg, tailq);
	388	if (config) {
	389	*config = cfg;
	390	}
	391
	392	return 0;
7c673cae FG	393	}
	394
	395	static int
	396	vbdev_split_init(void)
	397	{
11fdf7f2	398
7c673cae FG	399	struct spdk_conf_section *sp;
	400	const char *base_bdev_name;
	401	const char *split_count_str;
	402	const char *split_size_str;
11fdf7f2	403	int rc, i, split_count, split_size;
7c673cae FG	404
	405	sp = spdk_conf_find_section(NULL, "Split");
	406	if (sp == NULL) {
	407	return 0;
	408	}
	409
	410	for (i = 0; ; i++) {
	411	if (!spdk_conf_section_get_nval(sp, "Split", i)) {
	412	break;
	413	}
	414
	415	base_bdev_name = spdk_conf_section_get_nmval(sp, "Split", i, 0);
	416	if (!base_bdev_name) {
11fdf7f2 TL	417	SPDK_ERRLOG("Split configuration missing bdev name\n");
	418	rc = -EINVAL;
	419	goto err;
7c673cae FG	420	}
	421
	422	split_count_str = spdk_conf_section_get_nmval(sp, "Split", i, 1);
	423	if (!split_count_str) {
	424	SPDK_ERRLOG("Split configuration missing split count\n");
11fdf7f2 TL	425	rc = -EINVAL;
11fdf7f2 TL	426	goto err;
7c673cae FG	427	}
7c673cae FG	428
9f95a23c	429	split_count = spdk_strtol(split_count_str, 10);
7c673cae FG	430	if (split_count < 1) {
7c673cae FG	431	SPDK_ERRLOG("Invalid Split count %d\n", split_count);
11fdf7f2 TL	432	rc = -EINVAL;
11fdf7f2 TL	433	goto err;
7c673cae FG	434	}
	435
	436	/* Optional split size in MB */
	437	split_size = 0;
	438	split_size_str = spdk_conf_section_get_nmval(sp, "Split", i, 2);
	439	if (split_size_str) {
9f95a23c	440	split_size = spdk_strtol(split_size_str, 10);
7c673cae FG	441	if (split_size <= 0) {
7c673cae FG	442	SPDK_ERRLOG("Invalid Split size %d\n", split_size);
11fdf7f2 TL	443	rc = -EINVAL;
11fdf7f2 TL	444	goto err;
7c673cae FG	445	}
	446	}
	447
11fdf7f2 TL	448	rc = vbdev_split_add_config(base_bdev_name, split_count, split_size, NULL);
	449	if (rc != 0) {
	450	goto err;
7c673cae FG	451	}
	452	}
	453
	454	return 0;
11fdf7f2 TL	455	err:
	456	vbdev_split_clear_config();
	457	return rc;
7c673cae FG	458	}
	459
	460	static void
	461	vbdev_split_fini(void)
	462	{
11fdf7f2 TL	463	vbdev_split_clear_config();
	464	}
	465
	466	static void
	467	vbdev_split_examine(struct spdk_bdev *bdev)
	468	{
	469	struct spdk_vbdev_split_config *cfg = vbdev_split_config_find_by_base_name(bdev->name);
	470
	471	if (cfg != NULL && cfg->removed == false) {
	472	assert(cfg->split_base == NULL);
	473
	474	if (vbdev_split_create(cfg)) {
	475	SPDK_ERRLOG("could not split bdev %s\n", bdev->name);
	476	}
	477	}
	478	spdk_bdev_module_examine_done(&split_if);
	479	}
	480
	481	static int
	482	vbdev_split_config_json(struct spdk_json_write_ctx *w)
	483	{
	484	struct spdk_vbdev_split_config *cfg;
	485
	486	TAILQ_FOREACH(cfg, &g_split_config, tailq) {
	487	spdk_json_write_object_begin(w);
	488
	489	spdk_json_write_named_string(w, "method", "construct_split_vbdev");
	490
	491	spdk_json_write_named_object_begin(w, "params");
	492	spdk_json_write_named_string(w, "base_bdev", cfg->base_bdev);
	493	spdk_json_write_named_uint32(w, "split_count", cfg->split_count);
	494	spdk_json_write_named_uint64(w, "split_size_mb", cfg->split_size_mb);
	495	spdk_json_write_object_end(w);
	496
	497	spdk_json_write_object_end(w);
	498	}
	499
	500	return 0;
	501	}
	502
	503	int
	504	create_vbdev_split(const char *base_bdev_name, unsigned split_count, uint64_t split_size_mb)
	505	{
	506	int rc;
	507	struct spdk_vbdev_split_config *cfg;
	508
	509	rc = vbdev_split_add_config(base_bdev_name, split_count, split_size_mb, &cfg);
	510	if (rc) {
	511	return rc;
	512	}
	513
	514	rc = vbdev_split_create(cfg);
	515	if (rc == -ENODEV) {
	516	/* It is ok if base bdev does not exist yet. */
	517	rc = 0;
	518	}
	519
	520	return rc;
	521	}
	522
	523	int
	524	spdk_vbdev_split_destruct(const char *base_bdev_name)
	525	{
	526	struct spdk_vbdev_split_config *cfg = vbdev_split_config_find_by_base_name(base_bdev_name);
527
528	if (!cfg) {
529	SPDK_ERRLOG("Split configuration for '%s' not found\n", base_bdev_name);
530	return -ENOENT;
531	}
532
533	vbdev_split_destruct_config(cfg);
534	return 0;
535	}
536
537	struct spdk_bdev_part_base *
538	spdk_vbdev_split_get_part_base(struct spdk_bdev *bdev)
539	{
540	struct spdk_vbdev_split_config *cfg;
541
542	cfg = vbdev_split_config_find_by_base_name(spdk_bdev_get_name(bdev));
7c673cae	543
11fdf7f2 TL	544	if (cfg == NULL) {
11fdf7f2 TL	545	return NULL;
7c673cae	546	}
11fdf7f2 TL	547
	548	return cfg->split_base;
	549	}
	550
	551	/*
	552	* During init we'll be asked how much memory we'd like passed to us
	553	* in bev_io structures as context. Here's where we specify how
	554	* much context we want per IO.
	555	*/
	556	static int
	557	vbdev_split_get_ctx_size(void)
	558	{
	559	return sizeof(struct vbdev_split_bdev_io);
7c673cae FG	560	}
7c673cae FG	561
11fdf7f2	562	SPDK_LOG_REGISTER_COMPONENT("vbdev_split", SPDK_LOG_VBDEV_SPLIT)