[fio.git] / engines / glusterfs_async.c

/*
 * glusterfs engine
 *
 * IO engine using Glusterfs's gfapi async interface
 *
 */
#include "gfapi.h"
#define NOT_YET 1
struct fio_gf_iou {
	struct io_u *io_u;
	int io_complete;
};

static struct io_u *fio_gf_event(struct thread_data *td, int event)
{
	struct gf_data *gf_data = td->io_ops->data;
	dprint(FD_IO, "%s\n", __FUNCTION__);
	return gf_data->aio_events[event];
}

static int fio_gf_getevents(struct thread_data *td, unsigned int min,
			     unsigned int max, struct timespec *t)
{
	struct gf_data *g = td->io_ops->data;
	unsigned int events = 0;
	struct io_u *io_u;
	int i = 0;
	struct fio_gf_iou *io = NULL;

	dprint(FD_IO, "%s\n", __FUNCTION__);
	do {
		io_u_qiter(&td->io_u_all, io_u, i) {
			if (!(io_u->flags & IO_U_F_FLIGHT))
				continue;

			io = (struct fio_gf_iou *)io_u->engine_data;

			if (io && io->io_complete) {
				io->io_complete = 0;
				g->aio_events[events] = io_u;
				events++;

                if (events >= max) 
                    break;
			}

		}
		if (events < min)
			usleep(10);
		else
			break;

	} while (1);

	return events;
}

static void fio_gf_io_u_free(struct thread_data *td, struct io_u *io_u)
{
	struct fio_gf_iou *io = io_u->engine_data;

	if (io) {
        if (io->io_complete){
            log_err("incomplete IO found.\n");
        }
		io_u->engine_data = NULL;
		free(io);
	}
}

static int fio_gf_io_u_init(struct thread_data *td, struct io_u *io_u)
{
	struct fio_gf_iou *io = NULL;

	dprint(FD_FILE, "%s\n", __FUNCTION__);

    if (!io_u->engine_data){
        io = malloc(sizeof(struct fio_gf_iou));
        if (!io){
            td_verror(td, errno, "malloc");
            return 1;
        }
        io->io_complete = 0;
        io->io_u = io_u;
        io_u->engine_data = io;
    }
	return 0;
}

static void gf_async_cb(glfs_fd_t *fd, ssize_t ret, void *data) 
{
	struct io_u *io_u = (struct io_u *)data;
	struct fio_gf_iou *iou =
	    (struct fio_gf_iou *)io_u->engine_data;

    dprint(FD_IO, "%s ret %lu\n", __FUNCTION__, ret);    
    iou->io_complete = 1;
}

static int fio_gf_async_queue(struct thread_data fio_unused *td, struct io_u *io_u)
{
	struct gf_data *g = td->io_ops->data;
    int r = 0;

    dprint(FD_IO, "%s op %s\n", __FUNCTION__,  
           io_u->ddir == DDIR_READ? "read": io_u->ddir == DDIR_WRITE? "write":io_u->ddir == DDIR_SYNC? "sync":"unknown");    

	fio_ro_check(td, io_u);

	if (io_u->ddir == DDIR_READ)
		r = glfs_pread_async(g->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset,
                         0, gf_async_cb, (void *)io_u);
	else if (io_u->ddir == DDIR_WRITE)
		r = glfs_pwrite_async(g->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset,
                         0, gf_async_cb, (void *)io_u);
    else if (io_u->ddir == DDIR_SYNC) {
        r = glfs_fsync_async(g->fd, gf_async_cb, (void *)io_u);
    }else { 	    
        log_err("unsupported operation.\n");
        io_u->error = -EINVAL;
        goto failed;
    }
    if (r){
        log_err("glfs failed.\n");
        io_u->error = r;
        goto failed;
    }

	return FIO_Q_QUEUED;

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

int fio_gf_async_setup(struct thread_data *td)
{
	int r = 0;
	struct gf_data *g = NULL;
#if defined(NOT_YET)
    fprintf(stderr, "the async interface is still very experimental...\n");
#endif
    r = fio_gf_setup(td);
    if (r){
        return r;
    }
    g = td->io_ops->data;
    g->aio_events = malloc(td->o.iodepth * sizeof(struct io_u *));
	if (!g->aio_events){
        r = -ENOMEM;
        fio_gf_cleanup(td);
        return r;
    }

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

    return r;

}

static int fio_gf_async_prep(struct thread_data *td, struct io_u *io_u)
{
	dprint(FD_FILE, "%s\n", __FUNCTION__);

	if (!ddir_rw(io_u->ddir))
		return 0;

	return 0;
}

static struct ioengine_ops ioengine = {
	.name		    = "gfapi_async",
	.version	    = FIO_IOOPS_VERSION,
	.init           = fio_gf_async_setup,
	.cleanup        = fio_gf_cleanup,
    .prep           = fio_gf_async_prep,
  	.queue		    = fio_gf_async_queue,
	.open_file	    = fio_gf_open_file,
	.close_file	    = fio_gf_close_file,
	.get_file_size	= fio_gf_get_file_size,
	.getevents      = fio_gf_getevents,
	.event          = fio_gf_event,
	.io_u_init      = fio_gf_io_u_init,
	.io_u_free      = fio_gf_io_u_free,
	.options        = gfapi_options,
	.option_struct_size = sizeof(struct gf_options),
	.flags		    = FIO_DISKLESSIO,
};

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

static void fio_exit fio_gf_unregister(void)
{
    unregister_ioengine(&ioengine);
}
Commit	Line	Data
cc47f094	1	/*
	2	* glusterfs engine
	3	*
	4	* IO engine using Glusterfs's gfapi async interface
	5	*
	6	*/
	7	#include "gfapi.h"
88db2717	8	#define NOT_YET 1
cc47f094	9	struct fio_gf_iou {
	10	struct io_u *io_u;
	11	int io_complete;
	12	};
	13
	14	static struct io_u fio_gf_event(struct thread_data td, int event)
	15	{
	16	struct gf_data *gf_data = td->io_ops->data;
	17	dprint(FD_IO, "%s\n", __FUNCTION__);
	18	return gf_data->aio_events[event];
	19	}
	20
	21	static int fio_gf_getevents(struct thread_data *td, unsigned int min,
	22	unsigned int max, struct timespec *t)
	23	{
	24	struct gf_data *g = td->io_ops->data;
	25	unsigned int events = 0;
	26	struct io_u *io_u;
	27	int i = 0;
	28	struct fio_gf_iou *io = NULL;
	29
	30	dprint(FD_IO, "%s\n", __FUNCTION__);
	31	do {
	32	io_u_qiter(&td->io_u_all, io_u, i) {
	33	if (!(io_u->flags & IO_U_F_FLIGHT))
	34	continue;
	35
	36	io = (struct fio_gf_iou *)io_u->engine_data;
	37
	38	if (io && io->io_complete) {
	39	io->io_complete = 0;
	40	g->aio_events[events] = io_u;
	41	events++;
88db2717	42
	43	if (events >= max)
	44	break;
cc47f094	45	}
	46
	47	}
	48	if (events < min)
88db2717	49	usleep(10);
cc47f094	50	else
	51	break;
	52
	53	} while (1);
	54
	55	return events;
	56	}
	57
88db2717	58	static void fio_gf_io_u_free(struct thread_data td, struct io_u io_u)
cc47f094	59	{
88db2717	60	struct fio_gf_iou *io = io_u->engine_data;
	61
	62	if (io) {
	63	if (io->io_complete){
	64	log_err("incomplete IO found.\n");
	65	}
	66	io_u->engine_data = NULL;
	67	free(io);
	68	}
	69	}
cc47f094	70
88db2717	71	static int fio_gf_io_u_init(struct thread_data td, struct io_u io_u)
	72	{
	73	struct fio_gf_iou *io = NULL;
cc47f094	74
88db2717	75	dprint(FD_FILE, "%s\n", __FUNCTION__);
	76
	77	if (!io_u->engine_data){
	78	io = malloc(sizeof(struct fio_gf_iou));
	79	if (!io){
	80	td_verror(td, errno, "malloc");
	81	return 1;
	82	}
	83	io->io_complete = 0;
	84	io->io_u = io_u;
	85	io_u->engine_data = io;
cc47f094	86	}
cc47f094	87	return 0;
	88	}
	89
	90	static void gf_async_cb(glfs_fd_t fd, ssize_t ret, void data)
	91	{
	92	struct io_u io_u = (struct io_u )data;
	93	struct fio_gf_iou *iou =
	94	(struct fio_gf_iou *)io_u->engine_data;
	95
	96	dprint(FD_IO, "%s ret %lu\n", __FUNCTION__, ret);
	97	iou->io_complete = 1;
	98	}
	99
	100	static int fio_gf_async_queue(struct thread_data fio_unused td, struct io_u io_u)
	101	{
	102	struct gf_data *g = td->io_ops->data;
	103	int r = 0;
	104
88db2717	105	dprint(FD_IO, "%s op %s\n", __FUNCTION__,
	106	io_u->ddir == DDIR_READ? "read": io_u->ddir == DDIR_WRITE? "write":io_u->ddir == DDIR_SYNC? "sync":"unknown");
	107
cc47f094	108	fio_ro_check(td, io_u);
	109
	110	if (io_u->ddir == DDIR_READ)
	111	r = glfs_pread_async(g->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset,
	112	0, gf_async_cb, (void *)io_u);
	113	else if (io_u->ddir == DDIR_WRITE)
88db2717	114	r = glfs_pwrite_async(g->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset,
cc47f094	115	0, gf_async_cb, (void *)io_u);
	116	else if (io_u->ddir == DDIR_SYNC) {
	117	r = glfs_fsync_async(g->fd, gf_async_cb, (void *)io_u);
	118	}else {
	119	log_err("unsupported operation.\n");
	120	io_u->error = -EINVAL;
	121	goto failed;
	122	}
	123	if (r){
	124	log_err("glfs failed.\n");
	125	io_u->error = r;
	126	goto failed;
	127	}
	128
	129	return FIO_Q_QUEUED;
	130
	131	failed:
	132	io_u->error = r;
	133	td_verror(td, io_u->error, "xfer");
	134	return FIO_Q_COMPLETED;
	135	}
	136
88db2717	137	int fio_gf_async_setup(struct thread_data *td)
	138	{
	139	int r = 0;
	140	struct gf_data *g = NULL;
	141	#if defined(NOT_YET)
	142	fprintf(stderr, "the async interface is still very experimental...\n");
	143	#endif
	144	r = fio_gf_setup(td);
	145	if (r){
	146	return r;
	147	}
	148	g = td->io_ops->data;
	149	g->aio_events = malloc(td->o.iodepth * sizeof(struct io_u *));
	150	if (!g->aio_events){
	151	r = -ENOMEM;
	152	fio_gf_cleanup(td);
	153	return r;
	154	}
	155
	156	memset(g->aio_events, 0, td->o.iodepth * sizeof(struct io_u *));
	157
	158	return r;
	159
	160	}
	161
	162	static int fio_gf_async_prep(struct thread_data td, struct io_u io_u)
	163	{
	164	dprint(FD_FILE, "%s\n", __FUNCTION__);
	165
	166	if (!ddir_rw(io_u->ddir))
	167	return 0;
	168
	169	return 0;
	170	}
cc47f094	171
	172	static struct ioengine_ops ioengine = {
	173	.name = "gfapi_async",
	174	.version = FIO_IOOPS_VERSION,
88db2717	175	.init = fio_gf_async_setup,
cc47f094	176	.cleanup = fio_gf_cleanup,
88db2717	177	.prep = fio_gf_async_prep,
88db2717	178	.queue = fio_gf_async_queue,
cc47f094	179	.open_file = fio_gf_open_file,
	180	.close_file = fio_gf_close_file,
	181	.get_file_size = fio_gf_get_file_size,
	182	.getevents = fio_gf_getevents,
	183	.event = fio_gf_event,
88db2717	184	.io_u_init = fio_gf_io_u_init,
88db2717	185	.io_u_free = fio_gf_io_u_free,
cc47f094	186	.options = gfapi_options,
	187	.option_struct_size = sizeof(struct gf_options),
	188	.flags = FIO_DISKLESSIO,
	189	};
	190
	191	static void fio_init fio_gf_register(void)
	192	{
	193	register_ioengine(&ioengine);
	194	}
	195
	196	static void fio_exit fio_gf_unregister(void)
	197	{
	198	unregister_ioengine(&ioengine);
	199	}