[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"
#include "../optgroup.h"

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

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

struct rbd_options {
	void *pad;
	char *rbd_name;
	char *pool_name;
	char *client_name;
	int busy_poll;
};

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		= "busy_poll",
		.lname		= "Busy poll",
		.type		= FIO_OPT_BOOL,
		.help		= "Busy poll for completions instead of sleeping",
		.off1		= offsetof(struct rbd_options, busy_poll),
		.def		= "0",
		.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;

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

	rbd = calloc(1, sizeof(struct rbd_data));
	if (!rbd)
		goto failed;

	rbd->aio_events = calloc(td->o.iodepth, sizeof(struct io_u *));
	if (!rbd->aio_events)
		goto failed;

	rbd->sort_events = calloc(td->o.iodepth, sizeof(struct io_u *));
	if (!rbd->sort_events)
		goto failed;

	*rbd_data_ptr = rbd;
	return 0;

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

}

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

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

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

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

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

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

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

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

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

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

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

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

	/*
	 * 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;

	fri->io_complete = 1;
}

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

	return rbd->aio_events[event];
}

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

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

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

	return 0;
}

static inline int rbd_io_u_seen(struct io_u *io_u)
{
	struct fio_rbd_iou *fri = io_u->engine_data;

	return fri->io_seen;
}

static void rbd_io_u_wait_complete(struct io_u *io_u)
{
	struct fio_rbd_iou *fri = io_u->engine_data;

	rbd_aio_wait_for_complete(fri->completion);
}

static int rbd_io_u_cmp(const void *p1, const void *p2)
{
	const struct io_u **a = (const struct io_u **) p1;
	const struct io_u **b = (const struct io_u **) p2;
	uint64_t at, bt;

	at = utime_since_now(&(*a)->start_time);
	bt = utime_since_now(&(*b)->start_time);

	if (at < bt)
		return -1;
	else if (at == bt)
		return 0;
	else
		return 1;
}

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

	sidx = 0;
	io_u_qiter(&td->io_u_all, io_u, i) {
		if (!(io_u->flags & IO_U_F_FLIGHT))
			continue;
		if (rbd_io_u_seen(io_u))
			continue;

		if (fri_check_complete(rbd, io_u, events))
			this_events++;
		else if (wait)
			rbd->sort_events[sidx++] = io_u;
	}

	if (!wait || !sidx)
		return this_events;

	/*
	 * Sort events, oldest issue first, then wait on as many as we
	 * need in order of age. If we have enough events, stop waiting,
	 * and just check if any of the older ones are done.
	 */
	if (sidx > 1)
		qsort(rbd->sort_events, sidx, sizeof(struct io_u *), rbd_io_u_cmp);

	for (i = 0; i < sidx; i++) {
		io_u = rbd->sort_events[i];

		if (fri_check_complete(rbd, io_u, events)) {
			this_events++;
			continue;
		}

		/*
		 * Stop waiting when we have enough, but continue checking
		 * all pending IOs if they are complete.
		 */
		if (*events >= min_evts)
			continue;

		rbd_io_u_wait_complete(io_u);

		if (fri_check_complete(rbd, io_u, events))
			this_events++;
	}

	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;
	struct rbd_options *o = td->eo;
	int wait = 0;

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

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

		if (!o->busy_poll)
			wait = 1;
		else
			nop;
	} while (1);

	return events;
}

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

	fio_ro_check(td, io_u);

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

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

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

	} else if (io_u->ddir == DDIR_READ) {
		r = rbd_aio_read(rbd->image, io_u->offset, io_u->xfer_buflen,
					io_u->xfer_buf, fri->completion);

		if (r < 0) {
			log_err("rbd_aio_read failed.\n");
			goto failed_comp;
		}
	} else if (io_u->ddir == DDIR_TRIM) {
		r = rbd_aio_discard(rbd->image, io_u->offset,
					io_u->xfer_buflen, fri->completion);
		if (r < 0) {
			log_err("rbd_aio_discard failed.\n");
			goto failed_comp;
		}
	} else if (io_u->ddir == DDIR_SYNC) {
		r = rbd_aio_flush(rbd->image, fri->completion);
		if (r < 0) {
			log_err("rbd_flush failed.\n");
			goto failed_comp;
		}
	} else {
		dprint(FD_IO, "%s: Warning: unhandled ddir: %d\n", __func__,
		       io_u->ddir);
		goto failed_comp;
	}

	return FIO_Q_QUEUED;
failed_comp:
	rbd_aio_release(fri->completion);
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 = td->io_ops->data;

	if (rbd) {
		_fio_rbd_disconnect(rbd);
		free(rbd->aio_events);
		free(rbd->sort_events);
		free(rbd);
	}
}

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

	/* 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);
	if (r) {
		log_err("fio_setup_rbd_data failed.\n");
		goto cleanup;
	}
	td->io_ops->data = rbd;

	/* 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->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);
	return 0;

disconnect:
	_fio_rbd_disconnect(rbd);
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)
{
#if defined(CONFIG_RBD_INVAL)
	struct rbd_data *rbd = td->io_ops->data;

	return rbd_invalidate_cache(rbd->image);
#else
	return 0;
#endif
}

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