[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 ulong cb_count = 0, issued = 0;

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(100);
		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);
	}
    fprintf(stderr, "issued %lu finished %lu\n", issued, cb_count);
}

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;
    cb_count ++;
}

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