[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_seen;
};

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

struct rbd_options {
	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 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;
}

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 (rbd_aio_is_complete(fri->completion)) {
		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_seen = 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_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");
			goto failed_comp;
		}

	} else if (io_u->ddir == DDIR_READ) {
		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");
			goto failed_comp;
		}
	} else if (io_u->ddir == DDIR_TRIM) {
		r = rbd_aio_discard(rbd_data->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_data->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_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)
{
#if defined(CONFIG_RBD_INVAL)
	struct rbd_data *rbd_data = td->io_ops->data;

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