[fio.git] / engines / rados.c

/*
 *  Ceph Rados engine
 *
 * IO engine using Ceph's RADOS interface to test low-level performance of
 * Ceph OSDs.
 *
 */

#include <rados/librados.h>
#include <pthread.h>
#include "fio.h"
#include "../optgroup.h"

struct fio_rados_iou {
	struct thread_data *td;
	struct io_u *io_u;
	rados_completion_t completion;
	rados_write_op_t write_op;
};

struct rados_data {
	rados_t cluster;
	rados_ioctx_t io_ctx;
	char **objects;
	size_t object_count;
	struct io_u **aio_events;
	bool connected;
};

/* fio configuration options read from the job file */
struct rados_options {
	void *pad;
	char *cluster_name;
	char *pool_name;
	char *client_name;
	int busy_poll;
};

static struct fio_option options[] = {
	{
		.name     = "clustername",
		.lname    = "ceph cluster name",
		.type     = FIO_OPT_STR_STORE,
		.help     = "Cluster name for ceph",
		.off1     = offsetof(struct rados_options, cluster_name),
		.category = FIO_OPT_C_ENGINE,
		.group    = FIO_OPT_G_RBD,
	},
	{
		.name     = "pool",
		.lname    = "pool name to use",
		.type     = FIO_OPT_STR_STORE,
		.help     = "Ceph pool name to benchmark against",
		.off1     = offsetof(struct rados_options, pool_name),
		.category = FIO_OPT_C_ENGINE,
		.group    = FIO_OPT_G_RBD,
	},
	{
		.name     = "clientname",
		.lname    = "rados engine clientname",
		.type     = FIO_OPT_STR_STORE,
		.help     = "Name of the ceph client to access RADOS engine",
		.off1     = offsetof(struct rados_options, client_name),
		.category = FIO_OPT_C_ENGINE,
		.group    = FIO_OPT_G_RBD,
	},
	{
		.name     = "busy_poll",
		.lname    = "busy poll mode",
		.type     = FIO_OPT_BOOL,
		.help     = "Busy poll for completions instead of sleeping",
		.off1     = offsetof(struct rados_options, busy_poll),
		.def	  = "0",
		.category = FIO_OPT_C_ENGINE,
		.group    = FIO_OPT_G_RBD,
	},
	{
		.name     = NULL,
	},
};

static int _fio_setup_rados_data(struct thread_data *td,
				struct rados_data **rados_data_ptr)
{
	struct rados_data *rados;

	if (td->io_ops_data)
		return 0;

	rados = calloc(1, sizeof(struct rados_data));
	if (!rados)
		goto failed;

	rados->connected = false;

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

	rados->object_count = td->o.nr_files;
	rados->objects = calloc(rados->object_count, sizeof(char*));
	if (!rados->objects)
		goto failed;

	*rados_data_ptr = rados;
	return 0;

failed:
	if (rados) {
		rados->object_count = 0;
		if (rados->aio_events)
			free(rados->aio_events);
		free(rados);
	}
	return 1;
}

static void _fio_rados_rm_objects(struct rados_data *rados)
{
	size_t i;
	for (i = 0; i < rados->object_count; ++i) {
		if (rados->objects[i]) {
			rados_remove(rados->io_ctx, rados->objects[i]);
			free(rados->objects[i]);
			rados->objects[i] = NULL;
		}
	}
}

static int _fio_rados_connect(struct thread_data *td)
{
	struct rados_data *rados = td->io_ops_data;
	struct rados_options *o = td->eo;
	int r;
	const uint64_t file_size =
		td->o.size / (td->o.nr_files ? td->o.nr_files : 1u);
	struct fio_file *f;
	uint32_t i;
	size_t oname_len = 0;

	if (o->cluster_name) {
		char *client_name = NULL;

		/*
		* If we specify cluser name, the rados_create2
		* will not assume 'client.'. name is considered
		* as a full type.id namestr
		*/
		if (o->client_name) {
			if (!index(o->client_name, '.')) {
				client_name = calloc(1, strlen("client.") +
					strlen(o->client_name) + 1);
				strcat(client_name, "client.");
				strcat(client_name, o->client_name);
			} else {
				client_name = o->client_name;
			}
		}

		r = rados_create2(&rados->cluster, o->cluster_name,
			client_name, 0);

		if (client_name && !index(o->client_name, '.'))
			free(client_name);
	} else
		r = rados_create(&rados->cluster, o->client_name);

	if (r < 0) {
		log_err("rados_create failed.\n");
		goto failed_early;
	}

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

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

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

	for (i = 0; i < rados->object_count; i++) {
		f = td->files[i];
		f->real_file_size = file_size;
		f->engine_pos = i;

		oname_len = strlen(f->file_name) + 32;
		rados->objects[i] = malloc(oname_len);
		/* vary objects for different jobs */
		snprintf(rados->objects[i], oname_len - 1,
			"fio_rados_bench.%s.%x",
			f->file_name, td->thread_number);
		r = rados_write(rados->io_ctx, rados->objects[i], "", 0, 0);
		if (r < 0) {
			free(rados->objects[i]);
			rados->objects[i] = NULL;
			log_err("error creating object.\n");
			goto failed_obj_create;
		}
	}

  return 0;

failed_obj_create:
	_fio_rados_rm_objects(rados);
	rados_ioctx_destroy(rados->io_ctx);
	rados->io_ctx = NULL;
failed_shutdown:
	rados_shutdown(rados->cluster);
	rados->cluster = NULL;
failed_early:
	return 1;
}

static void _fio_rados_disconnect(struct rados_data *rados)
{
	if (!rados)
		return;

	_fio_rados_rm_objects(rados);

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

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

static void fio_rados_cleanup(struct thread_data *td)
{
	struct rados_data *rados = td->io_ops_data;

	if (rados) {
		_fio_rados_disconnect(rados);
		free(rados->objects);
		free(rados->aio_events);
		free(rados);
	}
}

static enum fio_q_status
fio_rados_queue(struct thread_data *td, struct io_u *io_u)
{
	struct rados_data *rados = td->io_ops_data;
	struct fio_rados_iou *fri = io_u->engine_data;
	char *object = rados->objects[io_u->file->engine_pos];
	int r = -1;

	fio_ro_check(td, io_u);

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

		r = rados_aio_write(rados->io_ctx, object, fri->completion,
			io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
		if (r < 0) {
			log_err("rados_write failed.\n");
			goto failed_comp;
		}
		return FIO_Q_QUEUED;
	} else if (io_u->ddir == DDIR_READ) {
		r = rados_aio_create_completion(fri, NULL,
			NULL, &fri->completion);
		if (r < 0) {
			log_err("rados_aio_create_completion failed.\n");
			goto failed;
		}
		r = rados_aio_read(rados->io_ctx, object, fri->completion,
			io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
		if (r < 0) {
			log_err("rados_aio_read failed.\n");
			goto failed_comp;
		}
		return FIO_Q_QUEUED;
	} else if (io_u->ddir == DDIR_TRIM) {
		r = rados_aio_create_completion(fri, NULL,
			NULL , &fri->completion);
		if (r < 0) {
			log_err("rados_aio_create_completion failed.\n");
			goto failed;
		}
		fri->write_op = rados_create_write_op();
		if (fri->write_op == NULL) {
			log_err("rados_create_write_op failed.\n");
			goto failed_comp;
		}
		rados_write_op_zero(fri->write_op, io_u->offset,
			io_u->xfer_buflen);
		r = rados_aio_write_op_operate(fri->write_op, rados->io_ctx,
			fri->completion, object, NULL, 0);
		if (r < 0) {
			log_err("rados_aio_write_op_operate failed.\n");
			goto failed_write_op;
		}
		return FIO_Q_QUEUED;
	 }

	log_err("WARNING: Only DDIR_READ, DDIR_WRITE and DDIR_TRIM are supported!");

failed_write_op:
	rados_release_write_op(fri->write_op);
failed_comp:
	rados_aio_release(fri->completion);
failed:
	io_u->error = -r;
	td_verror(td, io_u->error, "xfer");
	return FIO_Q_COMPLETED;
}

static struct io_u *fio_rados_event(struct thread_data *td, int event)
{
	struct rados_data *rados = td->io_ops_data;
	return rados->aio_events[event];
}

int fio_rados_getevents(struct thread_data *td, unsigned int min,
	unsigned int max, const struct timespec *t)
{
	struct rados_data *rados = td->io_ops_data;
	struct rados_options *o = td->eo;
	int busy_poll = o->busy_poll;
	unsigned int events = 0;
	struct io_u *u;
	struct fio_rados_iou *fri;
	unsigned int i;
	rados_completion_t first_unfinished;
	int observed_new = 0;

	/* loop through inflight ios until we find 'min' completions */
	do {
		first_unfinished = NULL;
		io_u_qiter(&td->io_u_all, u, i) {
			if (!(u->flags & IO_U_F_FLIGHT))
				continue;

			fri = u->engine_data;
			if (fri->completion) {
				if (rados_aio_is_complete(fri->completion)) {
					if (fri->write_op != NULL) {
						rados_release_write_op(fri->write_op);
						fri->write_op = NULL;
					}
					rados_aio_release(fri->completion);
					fri->completion = NULL;
					rados->aio_events[events] = u;
					events++;
					observed_new = 1;
				} else if (first_unfinished == NULL) {
					first_unfinished = fri->completion;
				}
			}
			if (events >= max)
				break;
		}
		if (events >= min)
			return events;
		if (first_unfinished == NULL || busy_poll)
			continue;

		if (!observed_new)
			rados_aio_wait_for_complete(first_unfinished);
	} while (1);
  return events;
}

static int fio_rados_setup(struct thread_data *td)
{
	struct rados_data *rados = NULL;
	int r;
	/* allocate engine specific structure to deal with librados. */
	r = _fio_setup_rados_data(td, &rados);
	if (r) {
		log_err("fio_setup_rados_data failed.\n");
		goto cleanup;
	}
	td->io_ops_data = rados;

	/* Force single process mode.
	*/
	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_rados_connect(td);
	if (r) {
		log_err("fio_rados_connect failed.\n");
		goto cleanup;
	}
	rados->connected = true;

	return 0;
cleanup:
	fio_rados_cleanup(td);
	return r;
}

/* open/invalidate are noops. we set the FIO_DISKLESSIO flag in ioengine_ops to
   prevent fio from creating the files
*/
static int fio_rados_open(struct thread_data *td, struct fio_file *f)
{
	return 0;
}
static int fio_rados_invalidate(struct thread_data *td, struct fio_file *f)
{
	return 0;
}

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

	if (fri) {
		io_u->engine_data = NULL;
		fri->td = NULL;
		if (fri->completion)
			rados_aio_release(fri->completion);
		if (fri->write_op)
			rados_release_write_op(fri->write_op);
		free(fri);
	}
}

static int fio_rados_io_u_init(struct thread_data *td, struct io_u *io_u)
{
	struct fio_rados_iou *fri;
	fri = calloc(1, sizeof(*fri));
	fri->io_u = io_u;
	fri->td = td;
	io_u->engine_data = fri;
	return 0;
}

/* ioengine_ops for get_ioengine() */
static struct ioengine_ops ioengine = {
	.name = "rados",
	.version		= FIO_IOOPS_VERSION,
	.flags			= FIO_DISKLESSIO,
	.setup			= fio_rados_setup,
	.queue			= fio_rados_queue,
	.getevents		= fio_rados_getevents,
	.event			= fio_rados_event,
	.cleanup		= fio_rados_cleanup,
	.open_file		= fio_rados_open,
	.invalidate		= fio_rados_invalidate,
	.options		= options,
	.io_u_init		= fio_rados_io_u_init,
	.io_u_free		= fio_rados_io_u_free,
	.option_struct_size	= sizeof(struct rados_options),
};

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

static void fio_exit fio_rados_unregister(void)
{
	unregister_ioengine(&ioengine);
}
Commit	Line	Data
d5f9b0ea IF	1	/*
	2	* Ceph Rados engine
	3	*
	4	* IO engine using Ceph's RADOS interface to test low-level performance of
	5	* Ceph OSDs.
	6	*
	7	*/
	8
	9	#include <rados/librados.h>
	10	#include <pthread.h>
	11	#include "fio.h"
	12	#include "../optgroup.h"
	13
	14	struct fio_rados_iou {
	15	struct thread_data *td;
	16	struct io_u *io_u;
	17	rados_completion_t completion;
	18	rados_write_op_t write_op;
	19	};
	20
	21	struct rados_data {
	22	rados_t cluster;
	23	rados_ioctx_t io_ctx;
	24	char **objects;
	25	size_t object_count;
	26	struct io_u **aio_events;
	27	bool connected;
	28	};
	29
	30	/* fio configuration options read from the job file */
	31	struct rados_options {
	32	void *pad;
	33	char *cluster_name;
	34	char *pool_name;
	35	char *client_name;
	36	int busy_poll;
	37	};
	38
	39	static struct fio_option options[] = {
	40	{
	41	.name = "clustername",
	42	.lname = "ceph cluster name",
	43	.type = FIO_OPT_STR_STORE,
	44	.help = "Cluster name for ceph",
	45	.off1 = offsetof(struct rados_options, cluster_name),
	46	.category = FIO_OPT_C_ENGINE,
	47	.group = FIO_OPT_G_RBD,
	48	},
	49	{
	50	.name = "pool",
	51	.lname = "pool name to use",
	52	.type = FIO_OPT_STR_STORE,
	53	.help = "Ceph pool name to benchmark against",
	54	.off1 = offsetof(struct rados_options, pool_name),
	55	.category = FIO_OPT_C_ENGINE,
	56	.group = FIO_OPT_G_RBD,
	57	},
	58	{
	59	.name = "clientname",
	60	.lname = "rados engine clientname",
	61	.type = FIO_OPT_STR_STORE,
	62	.help = "Name of the ceph client to access RADOS engine",
	63	.off1 = offsetof(struct rados_options, client_name),
	64	.category = FIO_OPT_C_ENGINE,
65	.group = FIO_OPT_G_RBD,
66	},
67	{
68	.name = "busy_poll",
69	.lname = "busy poll mode",
70	.type = FIO_OPT_BOOL,
71	.help = "Busy poll for completions instead of sleeping",
72	.off1 = offsetof(struct rados_options, busy_poll),
73	.def = "0",
74	.category = FIO_OPT_C_ENGINE,
75	.group = FIO_OPT_G_RBD,
76	},
77	{
78	.name = NULL,
79	},
80	};
81
82	static int _fio_setup_rados_data(struct thread_data *td,
83	struct rados_data **rados_data_ptr)
84	{
85	struct rados_data *rados;
86
87	if (td->io_ops_data)
88	return 0;
89
90	rados = calloc(1, sizeof(struct rados_data));
91	if (!rados)
92	goto failed;
93
94	rados->connected = false;
95
96	rados->aio_events = calloc(td->o.iodepth, sizeof(struct io_u *));
97	if (!rados->aio_events)
98	goto failed;
99
100	rados->object_count = td->o.nr_files;
101	rados->objects = calloc(rados->object_count, sizeof(char*));
102	if (!rados->objects)
103	goto failed;
104
105	*rados_data_ptr = rados;
106	return 0;
107
108	failed:
109	if (rados) {
110	rados->object_count = 0;
111	if (rados->aio_events)
112	free(rados->aio_events);
113	free(rados);
114	}
115	return 1;
116	}
117
118	static void _fio_rados_rm_objects(struct rados_data *rados)
119	{
120	size_t i;
121	for (i = 0; i < rados->object_count; ++i) {
122	if (rados->objects[i]) {
123	rados_remove(rados->io_ctx, rados->objects[i]);
124	free(rados->objects[i]);
125	rados->objects[i] = NULL;
126	}
127	}
128	}
129
130	static int _fio_rados_connect(struct thread_data *td)
131	{
132	struct rados_data *rados = td->io_ops_data;
133	struct rados_options *o = td->eo;
134	int r;
135	const uint64_t file_size =
136	td->o.size / (td->o.nr_files ? td->o.nr_files : 1u);
137	struct fio_file *f;
138	uint32_t i;
139	size_t oname_len = 0;
140
141	if (o->cluster_name) {
142	char *client_name = NULL;
143
144	/*
145	* If we specify cluser name, the rados_create2
146	* will not assume 'client.'. name is considered
147	* as a full type.id namestr
148	*/
149	if (o->client_name) {
150	if (!index(o->client_name, '.')) {
151	client_name = calloc(1, strlen("client.") +
152	strlen(o->client_name) + 1);
153	strcat(client_name, "client.");
154	strcat(client_name, o->client_name);
155	} else {
156	client_name = o->client_name;
157	}
158	}
159
160	r = rados_create2(&rados->cluster, o->cluster_name,
161	client_name, 0);
162
163	if (client_name && !index(o->client_name, '.'))
164	free(client_name);
165	} else
166	r = rados_create(&rados->cluster, o->client_name);
167
168	if (r < 0) {
169	log_err("rados_create failed.\n");
170	goto failed_early;
171	}
172
173	r = rados_conf_read_file(rados->cluster, NULL);
174	if (r < 0) {
175	log_err("rados_conf_read_file failed.\n");
176	goto failed_early;
177	}
178
179	r = rados_connect(rados->cluster);
180	if (r < 0) {
181	log_err("rados_connect failed.\n");
182	goto failed_early;
183	}
184
185	r = rados_ioctx_create(rados->cluster, o->pool_name, &rados->io_ctx);
186	if (r < 0) {
187	log_err("rados_ioctx_create failed.\n");
188	goto failed_shutdown;
189	}
190
191	for (i = 0; i < rados->object_count; i++) {
192	f = td->files[i];
193	f->real_file_size = file_size;
194	f->engine_pos = i;
195
196	oname_len = strlen(f->file_name) + 32;
197	rados->objects[i] = malloc(oname_len);
198	/* vary objects for different jobs */
199	snprintf(rados->objects[i], oname_len - 1,
200	"fio_rados_bench.%s.%x",
201	f->file_name, td->thread_number);
202	r = rados_write(rados->io_ctx, rados->objects[i], "", 0, 0);
203	if (r < 0) {
204	free(rados->objects[i]);
205	rados->objects[i] = NULL;
206	log_err("error creating object.\n");
207	goto failed_obj_create;
208	}
209	}
210
211	return 0;
212
213	failed_obj_create:
214	_fio_rados_rm_objects(rados);
215	rados_ioctx_destroy(rados->io_ctx);
216	rados->io_ctx = NULL;
217	failed_shutdown:
218	rados_shutdown(rados->cluster);
219	rados->cluster = NULL;
220	failed_early:
221	return 1;
222	}
223
224	static void _fio_rados_disconnect(struct rados_data *rados)
225	{
226	if (!rados)
227	return;
228
229	_fio_rados_rm_objects(rados);
230
231	if (rados->io_ctx) {
232	rados_ioctx_destroy(rados->io_ctx);
233	rados->io_ctx = NULL;
234	}
235
236	if (rados->cluster) {
237	rados_shutdown(rados->cluster);
238	rados->cluster = NULL;
239	}
240	}
241
242	static void fio_rados_cleanup(struct thread_data *td)
243	{
244	struct rados_data *rados = td->io_ops_data;
245
246	if (rados) {
247	_fio_rados_disconnect(rados);
248	free(rados->objects);
249	free(rados->aio_events);
250	free(rados);
251	}
252	}
253
d3b07186 BVA	254	static enum fio_q_status
d3b07186 BVA	255	fio_rados_queue(struct thread_data td, struct io_u io_u)
d5f9b0ea IF	256	{
	257	struct rados_data *rados = td->io_ops_data;
	258	struct fio_rados_iou *fri = io_u->engine_data;
	259	char *object = rados->objects[io_u->file->engine_pos];
	260	int r = -1;
	261
	262	fio_ro_check(td, io_u);
	263
	264	if (io_u->ddir == DDIR_WRITE) {
	265	r = rados_aio_create_completion(fri, NULL,
	266	NULL, &fri->completion);
	267	if (r < 0) {
	268	log_err("rados_aio_create_completion failed.\n");
	269	goto failed;
	270	}
	271
	272	r = rados_aio_write(rados->io_ctx, object, fri->completion,
	273	io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
	274	if (r < 0) {
	275	log_err("rados_write failed.\n");
	276	goto failed_comp;
	277	}
	278	return FIO_Q_QUEUED;
	279	} else if (io_u->ddir == DDIR_READ) {
	280	r = rados_aio_create_completion(fri, NULL,
	281	NULL, &fri->completion);
	282	if (r < 0) {
	283	log_err("rados_aio_create_completion failed.\n");
	284	goto failed;
	285	}
	286	r = rados_aio_read(rados->io_ctx, object, fri->completion,
	287	io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
	288	if (r < 0) {
	289	log_err("rados_aio_read failed.\n");
	290	goto failed_comp;
	291	}
	292	return FIO_Q_QUEUED;
	293	} else if (io_u->ddir == DDIR_TRIM) {
	294	r = rados_aio_create_completion(fri, NULL,
	295	NULL , &fri->completion);
	296	if (r < 0) {
	297	log_err("rados_aio_create_completion failed.\n");
	298	goto failed;
	299	}
	300	fri->write_op = rados_create_write_op();
	301	if (fri->write_op == NULL) {
	302	log_err("rados_create_write_op failed.\n");
	303	goto failed_comp;
	304	}
	305	rados_write_op_zero(fri->write_op, io_u->offset,
	306	io_u->xfer_buflen);
	307	r = rados_aio_write_op_operate(fri->write_op, rados->io_ctx,
	308	fri->completion, object, NULL, 0);
	309	if (r < 0) {
	310	log_err("rados_aio_write_op_operate failed.\n");
	311	goto failed_write_op;
	312	}
	313	return FIO_Q_QUEUED;
	314	}
	315
	316	log_err("WARNING: Only DDIR_READ, DDIR_WRITE and DDIR_TRIM are supported!");
	317
	318	failed_write_op:
	319	rados_release_write_op(fri->write_op);
320	failed_comp:
321	rados_aio_release(fri->completion);
322	failed:
323	io_u->error = -r;
324	td_verror(td, io_u->error, "xfer");
325	return FIO_Q_COMPLETED;
326	}
327
328	static struct io_u fio_rados_event(struct thread_data td, int event)
329	{
330	struct rados_data *rados = td->io_ops_data;
331	return rados->aio_events[event];
332	}
333
334	int fio_rados_getevents(struct thread_data *td, unsigned int min,
335	unsigned int max, const struct timespec *t)
336	{
337	struct rados_data *rados = td->io_ops_data;
338	struct rados_options *o = td->eo;
339	int busy_poll = o->busy_poll;
340	unsigned int events = 0;
341	struct io_u *u;
342	struct fio_rados_iou *fri;
343	unsigned int i;
344	rados_completion_t first_unfinished;
345	int observed_new = 0;
346
347	/* loop through inflight ios until we find 'min' completions */
348	do {
349	first_unfinished = NULL;
350	io_u_qiter(&td->io_u_all, u, i) {
351	if (!(u->flags & IO_U_F_FLIGHT))
352	continue;
353
354	fri = u->engine_data;
355	if (fri->completion) {
356	if (rados_aio_is_complete(fri->completion)) {
357	if (fri->write_op != NULL) {
358	rados_release_write_op(fri->write_op);
359	fri->write_op = NULL;
360	}
361	rados_aio_release(fri->completion);
362	fri->completion = NULL;
363	rados->aio_events[events] = u;
364	events++;
365	observed_new = 1;
366	} else if (first_unfinished == NULL) {
367	first_unfinished = fri->completion;
368	}
369	}
370	if (events >= max)
371	break;
372	}
373	if (events >= min)
374	return events;
375	if (first_unfinished == NULL \|\| busy_poll)
376	continue;
377
378	if (!observed_new)
379	rados_aio_wait_for_complete(first_unfinished);
380	} while (1);
381	return events;
382	}
383
384	static int fio_rados_setup(struct thread_data *td)
385	{
386	struct rados_data *rados = NULL;
387	int r;
388	/* allocate engine specific structure to deal with librados. */
389	r = _fio_setup_rados_data(td, &rados);
390	if (r) {
391	log_err("fio_setup_rados_data failed.\n");
392	goto cleanup;
393	}
394	td->io_ops_data = rados;
395
396	/* Force single process mode.
397	*/
398	td->o.use_thread = 1;
399
400	/* connect in the main thread to determine to determine
401	* the size of the given RADOS block device. And disconnect
402	* later on.
403	*/
404	r = _fio_rados_connect(td);
405	if (r) {
406	log_err("fio_rados_connect failed.\n");
407	goto cleanup;
408	}
409	rados->connected = true;
410
411	return 0;
412	cleanup:
413	fio_rados_cleanup(td);
414	return r;
415	}
416
417	/* open/invalidate are noops. we set the FIO_DISKLESSIO flag in ioengine_ops to
418	prevent fio from creating the files
419	*/
420	static int fio_rados_open(struct thread_data td, struct fio_file f)
421	{
422	return 0;
423	}
424	static int fio_rados_invalidate(struct thread_data td, struct fio_file f)
425	{
426	return 0;
427	}
428
429	static void fio_rados_io_u_free(struct thread_data td, struct io_u io_u)
430	{
431	struct fio_rados_iou *fri = io_u->engine_data;
432
433	if (fri) {
434	io_u->engine_data = NULL;
435	fri->td = NULL;
436	if (fri->completion)
437	rados_aio_release(fri->completion);
438	if (fri->write_op)
439	rados_release_write_op(fri->write_op);
440	free(fri);
441	}
442	}
443
444	static int fio_rados_io_u_init(struct thread_data td, struct io_u io_u)
445	{
446	struct fio_rados_iou *fri;
447	fri = calloc(1, sizeof(*fri));
448	fri->io_u = io_u;
449	fri->td = td;
450	io_u->engine_data = fri;
451	return 0;
452	}
453
454	/* ioengine_ops for get_ioengine() */
455	static struct ioengine_ops ioengine = {
456	.name = "rados",
457	.version = FIO_IOOPS_VERSION,
458	.flags = FIO_DISKLESSIO,
459	.setup = fio_rados_setup,
460	.queue = fio_rados_queue,
461	.getevents = fio_rados_getevents,
462	.event = fio_rados_event,
463	.cleanup = fio_rados_cleanup,
464	.open_file = fio_rados_open,
465	.invalidate = fio_rados_invalidate,
466	.options = options,
467	.io_u_init = fio_rados_io_u_init,
468	.io_u_free = fio_rados_io_u_free,
469	.option_struct_size = sizeof(struct rados_options),
470	};
471
472	static void fio_init fio_rados_register(void)
473	{
474	register_ioengine(&ioengine);
475	}
476
477	static void fio_exit fio_rados_unregister(void)
478	{
479	unregister_ioengine(&ioengine);
480	}