[fio.git] / engines / rbd.c

/*
 * rbd engine
 *
 * IO engine using Ceph's librbd to test RADOS Block Devices.
 *
 */

#include <rbd/librbd.h>

#include "../fio.h"

struct fio_rbd_iou {
	struct io_u *io_u;
	rbd_completion_t completion;
	int io_complete;
	int io_seen;
};

struct rbd_data {
	rados_t cluster;
	rados_ioctx_t io_ctx;
	rbd_image_t image;
	struct io_u **aio_events;
};

struct rbd_options {
	struct thread_data *td;
	char *rbd_name;
	char *pool_name;
	char *client_name;
};

static struct fio_option options[] = {
	{
		.name		= "rbdname",
		.lname		= "rbd engine rbdname",
		.type		= FIO_OPT_STR_STORE,
		.help		= "RBD name for RBD engine",
		.off1		= offsetof(struct rbd_options, rbd_name),
		.category	= FIO_OPT_C_ENGINE,
		.group		= FIO_OPT_G_RBD,
	},
	{
		.name     = "pool",
		.lname    = "rbd engine pool",
		.type     = FIO_OPT_STR_STORE,
		.help     = "Name of the pool hosting the RBD for the RBD engine",
		.off1     = offsetof(struct rbd_options, pool_name),
		.category = FIO_OPT_C_ENGINE,
		.group    = FIO_OPT_G_RBD,
	},
	{
		.name     = "clientname",
		.lname    = "rbd engine clientname",
		.type     = FIO_OPT_STR_STORE,
		.help     = "Name of the ceph client to access the RBD for the RBD engine",
		.off1     = offsetof(struct rbd_options, client_name),
		.category = FIO_OPT_C_ENGINE,
		.group    = FIO_OPT_G_RBD,
	},
	{
		.name = NULL,
	},
};

static int _fio_setup_rbd_data(struct thread_data *td,
			       struct rbd_data **rbd_data_ptr)
{
	struct rbd_data *rbd_data;

	if (td->io_ops->data)
		return 0;

	rbd_data = malloc(sizeof(struct rbd_data));
	if (!rbd_data)
		goto failed;

	memset(rbd_data, 0, sizeof(struct rbd_data));

	rbd_data->aio_events = malloc(td->o.iodepth * sizeof(struct io_u *));
	if (!rbd_data->aio_events)
		goto failed;

	memset(rbd_data->aio_events, 0, td->o.iodepth * sizeof(struct io_u *));

	*rbd_data_ptr = rbd_data;

	return 0;

failed:
	if (rbd_data)
		free(rbd_data);
	return 1;

}

static int _fio_rbd_connect(struct thread_data *td)
{
	struct rbd_data *rbd_data = td->io_ops->data;
	struct rbd_options *o = td->eo;
	int r;

	r = rados_create(&(rbd_data->cluster), o->client_name);
	if (r < 0) {
		log_err("rados_create failed.\n");
		goto failed_early;
	}

	r = rados_conf_read_file(rbd_data->cluster, NULL);
	if (r < 0) {
		log_err("rados_conf_read_file failed.\n");
		goto failed_early;
	}

	r = rados_connect(rbd_data->cluster);
	if (r < 0) {
		log_err("rados_connect failed.\n");
		goto failed_shutdown;
	}

	r = rados_ioctx_create(rbd_data->cluster, o->pool_name,
			       &(rbd_data->io_ctx));
	if (r < 0) {
		log_err("rados_ioctx_create failed.\n");
		goto failed_shutdown;
	}

	r = rbd_open(rbd_data->io_ctx, o->rbd_name, &(rbd_data->image),
		     NULL /*snap */ );
	if (r < 0) {
		log_err("rbd_open failed.\n");
		goto failed_open;
	}
	return 0;

failed_open:
	rados_ioctx_destroy(rbd_data->io_ctx);
	rbd_data->io_ctx = NULL;
failed_shutdown:
	rados_shutdown(rbd_data->cluster);
	rbd_data->cluster = NULL;
failed_early:
	return 1;
}

static void _fio_rbd_disconnect(struct rbd_data *rbd_data)
{
	if (!rbd_data)
		return;

	/* shutdown everything */
	if (rbd_data->image) {
		rbd_close(rbd_data->image);
		rbd_data->image = NULL;
	}

	if (rbd_data->io_ctx) {
		rados_ioctx_destroy(rbd_data->io_ctx);
		rbd_data->io_ctx = NULL;
	}

	if (rbd_data->cluster) {
		rados_shutdown(rbd_data->cluster);
		rbd_data->cluster = NULL;
	}
}

static void _fio_rbd_finish_aiocb(rbd_completion_t comp, void *data)
{
	struct io_u *io_u = data;
	struct fio_rbd_iou *fri = io_u->engine_data;
	ssize_t ret;

	fri->io_complete = 1;

	/*
	 * Looks like return value is 0 for success, or < 0 for
	 * a specific error. So we have to assume that it can't do
	 * partial completions.
	 */
	ret = rbd_aio_get_return_value(fri->completion);
	if (ret < 0) {
		io_u->error = ret;
		io_u->resid = io_u->xfer_buflen;
	} else
		io_u->error = 0;
}

static struct io_u *fio_rbd_event(struct thread_data *td, int event)
{
	struct rbd_data *rbd_data = td->io_ops->data;

	return rbd_data->aio_events[event];
}

static inline int fri_check_complete(struct rbd_data *rbd_data,
				     struct io_u *io_u,
				     unsigned int *events)
{
	struct fio_rbd_iou *fri = io_u->engine_data;

	if (fri->io_complete) {
		fri->io_complete = 0;
		fri->io_seen = 1;
		rbd_data->aio_events[*events] = io_u;
		(*events)++;

		rbd_aio_release(fri->completion);
		return 1;
	}

	return 0;
}

static int rbd_iter_events(struct thread_data *td, unsigned int *events,
			   unsigned int min_evts, int wait)
{
	struct rbd_data *rbd_data = td->io_ops->data;
	unsigned int this_events = 0;
	struct io_u *io_u;
	int i;

	io_u_qiter(&td->io_u_all, io_u, i) {
		struct fio_rbd_iou *fri = io_u->engine_data;

		if (!(io_u->flags & IO_U_F_FLIGHT))
			continue;
		if (fri->io_seen)
			continue;

		if (fri_check_complete(rbd_data, io_u, events))
			this_events++;
		else if (wait) {
			rbd_aio_wait_for_complete(fri->completion);

			if (fri_check_complete(rbd_data, io_u, events))
				this_events++;
		}
		if (*events >= min_evts)
			break;
	}

	return this_events;
}

static int fio_rbd_getevents(struct thread_data *td, unsigned int min,
			     unsigned int max, const struct timespec *t)
{
	unsigned int this_events, events = 0;
	int wait = 0;

	do {
		this_events = rbd_iter_events(td, &events, min, wait);

		if (events >= min)
			break;
		if (this_events)
			continue;

		wait = 1;
	} while (1);

	return events;
}

static int fio_rbd_queue(struct thread_data *td, struct io_u *io_u)
{
	struct rbd_data *rbd_data = td->io_ops->data;
	struct fio_rbd_iou *fri = io_u->engine_data;
	int r = -1;

	fio_ro_check(td, io_u);

	fri->io_complete = 0;
	fri->io_seen = 0;

	if (io_u->ddir == DDIR_WRITE) {
		r = rbd_aio_create_completion(io_u, _fio_rbd_finish_aiocb,
						&fri->completion);
		if (r < 0) {
			log_err
			    ("rbd_aio_create_completion for DDIR_WRITE failed.\n");
			goto failed;
		}

		r = rbd_aio_write(rbd_data->image, io_u->offset,
				  io_u->xfer_buflen, io_u->xfer_buf,
				  fri->completion);
		if (r < 0) {
			log_err("rbd_aio_write failed.\n");
			rbd_aio_release(fri->completion);
			goto failed;
		}

	} else if (io_u->ddir == DDIR_READ) {
		r = rbd_aio_create_completion(io_u, _fio_rbd_finish_aiocb,
						&fri->completion);
		if (r < 0) {
			log_err
			    ("rbd_aio_create_completion for DDIR_READ failed.\n");
			goto failed;
		}

		r = rbd_aio_read(rbd_data->image, io_u->offset,
				 io_u->xfer_buflen, io_u->xfer_buf,
				 fri->completion);

		if (r < 0) {
			log_err("rbd_aio_read failed.\n");
			rbd_aio_release(fri->completion);
			goto failed;
		}

	} else if (io_u->ddir == DDIR_SYNC) {
		r = rbd_aio_create_completion(io_u, _fio_rbd_finish_aiocb,
						&fri->completion);
		if (r < 0) {
			log_err
			    ("rbd_aio_create_completion for DDIR_SYNC failed.\n");
			goto failed;
		}

		r = rbd_aio_flush(rbd_data->image, fri->completion);
		if (r < 0) {
			log_err("rbd_flush failed.\n");
			rbd_aio_release(fri->completion);
			goto failed;
		}

	} else {
		dprint(FD_IO, "%s: Warning: unhandled ddir: %d\n", __func__,
		       io_u->ddir);
		return FIO_Q_COMPLETED;
	}

	return FIO_Q_QUEUED;

failed:
	io_u->error = r;
	td_verror(td, io_u->error, "xfer");
	return FIO_Q_COMPLETED;
}

static int fio_rbd_init(struct thread_data *td)
{
	int r;

	r = _fio_rbd_connect(td);
	if (r) {
		log_err("fio_rbd_connect failed, return code: %d .\n", r);
		goto failed;
	}

	return 0;

failed:
	return 1;
}

static void fio_rbd_cleanup(struct thread_data *td)
{
	struct rbd_data *rbd_data = td->io_ops->data;

	if (rbd_data) {
		_fio_rbd_disconnect(rbd_data);
		free(rbd_data->aio_events);
		free(rbd_data);
	}

}

static int fio_rbd_setup(struct thread_data *td)
{
	int r = 0;
	rbd_image_info_t info;
	struct fio_file *f;
	struct rbd_data *rbd_data = NULL;
	int major, minor, extra;

	/* log version of librbd. No cluster connection required. */
	rbd_version(&major, &minor, &extra);
	log_info("rbd engine: RBD version: %d.%d.%d\n", major, minor, extra);

	/* allocate engine specific structure to deal with librbd. */
	r = _fio_setup_rbd_data(td, &rbd_data);
	if (r) {
		log_err("fio_setup_rbd_data failed.\n");
		goto cleanup;
	}
	td->io_ops->data = rbd_data;

	/* librbd does not allow us to run first in the main thread and later
	 * in a fork child. It needs to be the same process context all the
	 * time. 
	 */
	td->o.use_thread = 1;

	/* connect in the main thread to determine to determine
	 * the size of the given RADOS block device. And disconnect
	 * later on.
	 */
	r = _fio_rbd_connect(td);
	if (r) {
		log_err("fio_rbd_connect failed.\n");
		goto cleanup;
	}

	/* get size of the RADOS block device */
	r = rbd_stat(rbd_data->image, &info, sizeof(info));
	if (r < 0) {
		log_err("rbd_status failed.\n");
		goto disconnect;
	}
	dprint(FD_IO, "rbd-engine: image size: %lu\n", info.size);

	/* taken from "net" engine. Pretend we deal with files,
	 * even if we do not have any ideas about files.
	 * The size of the RBD is set instead of a artificial file.
	 */
	if (!td->files_index) {
		add_file(td, td->o.filename ? : "rbd", 0, 0);
		td->o.nr_files = td->o.nr_files ? : 1;
		td->o.open_files++;
	}
	f = td->files[0];
	f->real_file_size = info.size;

	/* disconnect, then we were only connected to determine
	 * the size of the RBD.
	 */
	_fio_rbd_disconnect(rbd_data);
	return 0;

disconnect:
	_fio_rbd_disconnect(rbd_data);
cleanup:
	fio_rbd_cleanup(td);
	return r;
}

static int fio_rbd_open(struct thread_data *td, struct fio_file *f)
{
	return 0;
}

static int fio_rbd_invalidate(struct thread_data *td, struct fio_file *f)
{
	return 0;
}

static void fio_rbd_io_u_free(struct thread_data *td, struct io_u *io_u)
{
	struct fio_rbd_iou *fri = io_u->engine_data;

	if (fri) {
		io_u->engine_data = NULL;
		free(fri);
	}
}

static int fio_rbd_io_u_init(struct thread_data *td, struct io_u *io_u)
{
	struct fio_rbd_iou *fri;

	fri = calloc(1, sizeof(*fri));
	fri->io_u = io_u;
	io_u->engine_data = fri;
	return 0;
}

static struct ioengine_ops ioengine = {
	.name			= "rbd",
	.version		= FIO_IOOPS_VERSION,
	.setup			= fio_rbd_setup,
	.init			= fio_rbd_init,
	.queue			= fio_rbd_queue,
	.getevents		= fio_rbd_getevents,
	.event			= fio_rbd_event,
	.cleanup		= fio_rbd_cleanup,
	.open_file		= fio_rbd_open,
	.invalidate		= fio_rbd_invalidate,
	.options		= options,
	.io_u_init		= fio_rbd_io_u_init,
	.io_u_free		= fio_rbd_io_u_free,
	.option_struct_size	= sizeof(struct rbd_options),
};

static void fio_init fio_rbd_register(void)
{
	register_ioengine(&ioengine);
}

static void fio_exit fio_rbd_unregister(void)
{
	unregister_ioengine(&ioengine);
}
Commit	Line	Data
fc5c0345 DG	1	/*
	2	* rbd engine
	3	*
	4	* IO engine using Ceph's librbd to test RADOS Block Devices.
	5	*
	6	*/
	7
	8	#include <rbd/librbd.h>
	9
	10	#include "../fio.h"
	11
	12	struct fio_rbd_iou {
	13	struct io_u *io_u;
d8b64af2	14	rbd_completion_t completion;
fc5c0345	15	int io_complete;
d8b64af2	16	int io_seen;
fc5c0345 DG	17	};
	18
	19	struct rbd_data {
	20	rados_t cluster;
	21	rados_ioctx_t io_ctx;
	22	rbd_image_t image;
	23	struct io_u **aio_events;
	24	};
	25
	26	struct rbd_options {
	27	struct thread_data *td;
	28	char *rbd_name;
	29	char *pool_name;
	30	char *client_name;
	31	};
	32
	33	static struct fio_option options[] = {
	34	{
d8b64af2 JA	35	.name = "rbdname",
	36	.lname = "rbd engine rbdname",
	37	.type = FIO_OPT_STR_STORE,
	38	.help = "RBD name for RBD engine",
	39	.off1 = offsetof(struct rbd_options, rbd_name),
	40	.category = FIO_OPT_C_ENGINE,
	41	.group = FIO_OPT_G_RBD,
	42	},
fc5c0345	43	{
d8b64af2 JA	44	.name = "pool",
	45	.lname = "rbd engine pool",
	46	.type = FIO_OPT_STR_STORE,
	47	.help = "Name of the pool hosting the RBD for the RBD engine",
	48	.off1 = offsetof(struct rbd_options, pool_name),
	49	.category = FIO_OPT_C_ENGINE,
	50	.group = FIO_OPT_G_RBD,
	51	},
fc5c0345	52	{
d8b64af2 JA	53	.name = "clientname",
	54	.lname = "rbd engine clientname",
	55	.type = FIO_OPT_STR_STORE,
	56	.help = "Name of the ceph client to access the RBD for the RBD engine",
	57	.off1 = offsetof(struct rbd_options, client_name),
	58	.category = FIO_OPT_C_ENGINE,
	59	.group = FIO_OPT_G_RBD,
	60	},
fc5c0345	61	{
d8b64af2 JA	62	.name = NULL,
d8b64af2 JA	63	},
fc5c0345 DG	64	};
	65
	66	static int _fio_setup_rbd_data(struct thread_data *td,
	67	struct rbd_data **rbd_data_ptr)
	68	{
	69	struct rbd_data *rbd_data;
	70
	71	if (td->io_ops->data)
	72	return 0;
	73
	74	rbd_data = malloc(sizeof(struct rbd_data));
	75	if (!rbd_data)
	76	goto failed;
	77
	78	memset(rbd_data, 0, sizeof(struct rbd_data));
	79
	80	rbd_data->aio_events = malloc(td->o.iodepth * sizeof(struct io_u *));
	81	if (!rbd_data->aio_events)
	82	goto failed;
	83
	84	memset(rbd_data->aio_events, 0, td->o.iodepth * sizeof(struct io_u *));
	85
	86	*rbd_data_ptr = rbd_data;
	87
	88	return 0;
	89
	90	failed:
d9cb70a8 JA	91	if (rbd_data)
d9cb70a8 JA	92	free(rbd_data);
fc5c0345 DG	93	return 1;
	94
	95	}
	96
	97	static int _fio_rbd_connect(struct thread_data *td)
	98	{
	99	struct rbd_data *rbd_data = td->io_ops->data;
	100	struct rbd_options *o = td->eo;
	101	int r;
	102
	103	r = rados_create(&(rbd_data->cluster), o->client_name);
	104	if (r < 0) {
	105	log_err("rados_create failed.\n");
	106	goto failed_early;
	107	}
	108
	109	r = rados_conf_read_file(rbd_data->cluster, NULL);
	110	if (r < 0) {
	111	log_err("rados_conf_read_file failed.\n");
	112	goto failed_early;
	113	}
	114
	115	r = rados_connect(rbd_data->cluster);
	116	if (r < 0) {
	117	log_err("rados_connect failed.\n");
	118	goto failed_shutdown;
	119	}
	120
	121	r = rados_ioctx_create(rbd_data->cluster, o->pool_name,
	122	&(rbd_data->io_ctx));
	123	if (r < 0) {
	124	log_err("rados_ioctx_create failed.\n");
	125	goto failed_shutdown;
	126	}
	127
	128	r = rbd_open(rbd_data->io_ctx, o->rbd_name, &(rbd_data->image),
	129	NULL /snap / );
	130	if (r < 0) {
	131	log_err("rbd_open failed.\n");
	132	goto failed_open;
	133	}
	134	return 0;
	135
	136	failed_open:
	137	rados_ioctx_destroy(rbd_data->io_ctx);
086f9935	138	rbd_data->io_ctx = NULL;
fc5c0345 DG	139	failed_shutdown:
fc5c0345 DG	140	rados_shutdown(rbd_data->cluster);
086f9935	141	rbd_data->cluster = NULL;
fc5c0345 DG	142	failed_early:
	143	return 1;
	144	}
	145
	146	static void _fio_rbd_disconnect(struct rbd_data *rbd_data)
	147	{
	148	if (!rbd_data)
	149	return;
	150
	151	/* shutdown everything */
	152	if (rbd_data->image) {
	153	rbd_close(rbd_data->image);
	154	rbd_data->image = NULL;
	155	}
	156
	157	if (rbd_data->io_ctx) {
	158	rados_ioctx_destroy(rbd_data->io_ctx);
	159	rbd_data->io_ctx = NULL;
	160	}
	161
	162	if (rbd_data->cluster) {
	163	rados_shutdown(rbd_data->cluster);
	164	rbd_data->cluster = NULL;
	165	}
	166	}
	167
d8b64af2	168	static void _fio_rbd_finish_aiocb(rbd_completion_t comp, void *data)
fc5c0345	169	{
d8b64af2 JA	170	struct io_u *io_u = data;
	171	struct fio_rbd_iou *fri = io_u->engine_data;
	172	ssize_t ret;
fc5c0345	173
d8b64af2	174	fri->io_complete = 1;
fc5c0345	175
d8b64af2 JA	176	/*
	177	* Looks like return value is 0 for success, or < 0 for
	178	* a specific error. So we have to assume that it can't do
	179	* partial completions.
	180	*/
	181	ret = rbd_aio_get_return_value(fri->completion);
	182	if (ret < 0) {
	183	io_u->error = ret;
	184	io_u->resid = io_u->xfer_buflen;
	185	} else
	186	io_u->error = 0;
	187	}
fc5c0345	188
d8b64af2 JA	189	static struct io_u fio_rbd_event(struct thread_data td, int event)
	190	{
	191	struct rbd_data *rbd_data = td->io_ops->data;
fc5c0345	192
d8b64af2	193	return rbd_data->aio_events[event];
fc5c0345 DG	194	}
fc5c0345 DG	195
d8b64af2 JA	196	static inline int fri_check_complete(struct rbd_data *rbd_data,
	197	struct io_u *io_u,
	198	unsigned int *events)
fc5c0345	199	{
d8b64af2	200	struct fio_rbd_iou *fri = io_u->engine_data;
fc5c0345	201
d8b64af2 JA	202	if (fri->io_complete) {
	203	fri->io_complete = 0;
	204	fri->io_seen = 1;
	205	rbd_data->aio_events[*events] = io_u;
	206	(*events)++;
fc5c0345	207
d8b64af2 JA	208	rbd_aio_release(fri->completion);
	209	return 1;
	210	}
fc5c0345	211
d8b64af2	212	return 0;
fc5c0345 DG	213	}
fc5c0345 DG	214
d8b64af2 JA	215	static int rbd_iter_events(struct thread_data td, unsigned int events,
d8b64af2 JA	216	unsigned int min_evts, int wait)
82340a9f	217	{
d8b64af2 JA	218	struct rbd_data *rbd_data = td->io_ops->data;
	219	unsigned int this_events = 0;
	220	struct io_u *io_u;
	221	int i;
82340a9f	222
d8b64af2 JA	223	io_u_qiter(&td->io_u_all, io_u, i) {
d8b64af2 JA	224	struct fio_rbd_iou *fri = io_u->engine_data;
82340a9f	225
d8b64af2 JA	226	if (!(io_u->flags & IO_U_F_FLIGHT))
	227	continue;
	228	if (fri->io_seen)
	229	continue;
82340a9f	230
d8b64af2 JA	231	if (fri_check_complete(rbd_data, io_u, events))
	232	this_events++;
	233	else if (wait) {
	234	rbd_aio_wait_for_complete(fri->completion);
82340a9f	235
d8b64af2 JA	236	if (fri_check_complete(rbd_data, io_u, events))
	237	this_events++;
	238	}
	239	if (*events >= min_evts)
	240	break;
	241	}
fc5c0345	242
d8b64af2	243	return this_events;
fc5c0345 DG	244	}
	245
	246	static int fio_rbd_getevents(struct thread_data *td, unsigned int min,
1f440ece	247	unsigned int max, const struct timespec *t)
fc5c0345	248	{
d8b64af2 JA	249	unsigned int this_events, events = 0;
d8b64af2 JA	250	int wait = 0;
fc5c0345 DG	251
fc5c0345 DG	252	do {
d8b64af2	253	this_events = rbd_iter_events(td, &events, min, wait);
fc5c0345	254
d8b64af2	255	if (events >= min)
fc5c0345	256	break;
d8b64af2 JA	257	if (this_events)
d8b64af2 JA	258	continue;
fc5c0345	259
d8b64af2	260	wait = 1;
fc5c0345 DG	261	} while (1);
	262
	263	return events;
	264	}
	265
	266	static int fio_rbd_queue(struct thread_data td, struct io_u io_u)
	267	{
fc5c0345	268	struct rbd_data *rbd_data = td->io_ops->data;
d8b64af2 JA	269	struct fio_rbd_iou *fri = io_u->engine_data;
d8b64af2 JA	270	int r = -1;
fc5c0345 DG	271
	272	fio_ro_check(td, io_u);
	273
d8b64af2 JA	274	fri->io_complete = 0;
	275	fri->io_seen = 0;
	276
fc5c0345	277	if (io_u->ddir == DDIR_WRITE) {
d8b64af2 JA	278	r = rbd_aio_create_completion(io_u, _fio_rbd_finish_aiocb,
d8b64af2 JA	279	&fri->completion);
fc5c0345 DG	280	if (r < 0) {
	281	log_err
	282	("rbd_aio_create_completion for DDIR_WRITE failed.\n");
	283	goto failed;
	284	}
	285
	286	r = rbd_aio_write(rbd_data->image, io_u->offset,
d8b64af2 JA	287	io_u->xfer_buflen, io_u->xfer_buf,
d8b64af2 JA	288	fri->completion);
fc5c0345 DG	289	if (r < 0) {
fc5c0345 DG	290	log_err("rbd_aio_write failed.\n");
d8b64af2	291	rbd_aio_release(fri->completion);
fc5c0345 DG	292	goto failed;
	293	}
	294
	295	} else if (io_u->ddir == DDIR_READ) {
d8b64af2 JA	296	r = rbd_aio_create_completion(io_u, _fio_rbd_finish_aiocb,
d8b64af2 JA	297	&fri->completion);
fc5c0345 DG	298	if (r < 0) {
	299	log_err
	300	("rbd_aio_create_completion for DDIR_READ failed.\n");
	301	goto failed;
	302	}
	303
	304	r = rbd_aio_read(rbd_data->image, io_u->offset,
d8b64af2 JA	305	io_u->xfer_buflen, io_u->xfer_buf,
d8b64af2 JA	306	fri->completion);
fc5c0345 DG	307
	308	if (r < 0) {
	309	log_err("rbd_aio_read failed.\n");
d8b64af2	310	rbd_aio_release(fri->completion);
fc5c0345 DG	311	goto failed;
	312	}
	313
	314	} else if (io_u->ddir == DDIR_SYNC) {
d8b64af2 JA	315	r = rbd_aio_create_completion(io_u, _fio_rbd_finish_aiocb,
d8b64af2 JA	316	&fri->completion);
82340a9f HW	317	if (r < 0) {
	318	log_err
	319	("rbd_aio_create_completion for DDIR_SYNC failed.\n");
	320	goto failed;
	321	}
	322
d8b64af2	323	r = rbd_aio_flush(rbd_data->image, fri->completion);
fc5c0345 DG	324	if (r < 0) {
fc5c0345 DG	325	log_err("rbd_flush failed.\n");
d8b64af2	326	rbd_aio_release(fri->completion);
fc5c0345 DG	327	goto failed;
	328	}
	329
fc5c0345 DG	330	} else {
	331	dprint(FD_IO, "%s: Warning: unhandled ddir: %d\n", __func__,
	332	io_u->ddir);
	333	return FIO_Q_COMPLETED;
	334	}
	335
	336	return FIO_Q_QUEUED;
	337
	338	failed:
	339	io_u->error = r;
	340	td_verror(td, io_u->error, "xfer");
	341	return FIO_Q_COMPLETED;
	342	}
	343
	344	static int fio_rbd_init(struct thread_data *td)
	345	{
	346	int r;
	347
	348	r = _fio_rbd_connect(td);
	349	if (r) {
	350	log_err("fio_rbd_connect failed, return code: %d .\n", r);
	351	goto failed;
	352	}
	353
	354	return 0;
	355
	356	failed:
	357	return 1;
fc5c0345 DG	358	}
	359
	360	static void fio_rbd_cleanup(struct thread_data *td)
	361	{
	362	struct rbd_data *rbd_data = td->io_ops->data;
	363
	364	if (rbd_data) {
	365	_fio_rbd_disconnect(rbd_data);
	366	free(rbd_data->aio_events);
	367	free(rbd_data);
	368	}
	369
	370	}
	371
	372	static int fio_rbd_setup(struct thread_data *td)
	373	{
	374	int r = 0;
	375	rbd_image_info_t info;
	376	struct fio_file *f;
	377	struct rbd_data *rbd_data = NULL;
	378	int major, minor, extra;
	379
	380	/* log version of librbd. No cluster connection required. */
	381	rbd_version(&major, &minor, &extra);
	382	log_info("rbd engine: RBD version: %d.%d.%d\n", major, minor, extra);
	383
	384	/* allocate engine specific structure to deal with librbd. */
	385	r = _fio_setup_rbd_data(td, &rbd_data);
	386	if (r) {
	387	log_err("fio_setup_rbd_data failed.\n");
	388	goto cleanup;
	389	}
	390	td->io_ops->data = rbd_data;
	391
d8b64af2 JA	392	/* librbd does not allow us to run first in the main thread and later
	393	* in a fork child. It needs to be the same process context all the
	394	* time.
fc5c0345 DG	395	*/
	396	td->o.use_thread = 1;
	397
	398	/* connect in the main thread to determine to determine
	399	* the size of the given RADOS block device. And disconnect
	400	* later on.
	401	*/
	402	r = _fio_rbd_connect(td);
	403	if (r) {
	404	log_err("fio_rbd_connect failed.\n");
	405	goto cleanup;
	406	}
	407
	408	/* get size of the RADOS block device */
	409	r = rbd_stat(rbd_data->image, &info, sizeof(info));
	410	if (r < 0) {
	411	log_err("rbd_status failed.\n");
	412	goto disconnect;
	413	}
	414	dprint(FD_IO, "rbd-engine: image size: %lu\n", info.size);
	415
	416	/* taken from "net" engine. Pretend we deal with files,
	417	* even if we do not have any ideas about files.
	418	* The size of the RBD is set instead of a artificial file.
	419	*/
	420	if (!td->files_index) {
5903e7b7	421	add_file(td, td->o.filename ? : "rbd", 0, 0);
fc5c0345	422	td->o.nr_files = td->o.nr_files ? : 1;
b53f2c54	423	td->o.open_files++;
fc5c0345 DG	424	}
	425	f = td->files[0];
	426	f->real_file_size = info.size;
	427
	428	/* disconnect, then we were only connected to determine
	429	* the size of the RBD.
	430	*/
	431	_fio_rbd_disconnect(rbd_data);
	432	return 0;
	433
	434	disconnect:
	435	_fio_rbd_disconnect(rbd_data);
	436	cleanup:
	437	fio_rbd_cleanup(td);
	438	return r;
	439	}
	440
	441	static int fio_rbd_open(struct thread_data td, struct fio_file f)
	442	{
	443	return 0;
	444	}
	445
d9b100fc JA	446	static int fio_rbd_invalidate(struct thread_data td, struct fio_file f)
	447	{
	448	return 0;
	449	}
	450
fc5c0345 DG	451	static void fio_rbd_io_u_free(struct thread_data td, struct io_u io_u)
fc5c0345 DG	452	{
d8b64af2	453	struct fio_rbd_iou *fri = io_u->engine_data;
fc5c0345	454
d8b64af2	455	if (fri) {
fc5c0345	456	io_u->engine_data = NULL;
d8b64af2	457	free(fri);
fc5c0345 DG	458	}
	459	}
	460
	461	static int fio_rbd_io_u_init(struct thread_data td, struct io_u io_u)
	462	{
d8b64af2	463	struct fio_rbd_iou *fri;
fc5c0345	464
d8b64af2 JA	465	fri = calloc(1, sizeof(*fri));
	466	fri->io_u = io_u;
	467	io_u->engine_data = fri;
fc5c0345 DG	468	return 0;
	469	}
	470
10aa136b	471	static struct ioengine_ops ioengine = {
d9b100fc JA	472	.name = "rbd",
	473	.version = FIO_IOOPS_VERSION,
	474	.setup = fio_rbd_setup,
	475	.init = fio_rbd_init,
	476	.queue = fio_rbd_queue,
	477	.getevents = fio_rbd_getevents,
	478	.event = fio_rbd_event,
	479	.cleanup = fio_rbd_cleanup,
	480	.open_file = fio_rbd_open,
	481	.invalidate = fio_rbd_invalidate,
	482	.options = options,
	483	.io_u_init = fio_rbd_io_u_init,
	484	.io_u_free = fio_rbd_io_u_free,
	485	.option_struct_size = sizeof(struct rbd_options),
fc5c0345 DG	486	};
	487
	488	static void fio_init fio_rbd_register(void)
	489	{
	490	register_ioengine(&ioengine);
	491	}
	492
	493	static void fio_exit fio_rbd_unregister(void)
	494	{
	495	unregister_ioengine(&ioengine);
	496	}