second async IO code drop
[fio.git] / engines / glusterfs_async.c
1 /*
2  * glusterfs engine
3  *
4  * IO engine using Glusterfs's gfapi async interface
5  *
6  */
7 #include "gfapi.h"
8 #define NOT_YET 1
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++;
42
43                 if (events >= max) 
44                     break;
45                         }
46
47                 }
48                 if (events < min)
49                         usleep(10);
50                 else
51                         break;
52
53         } while (1);
54
55         return events;
56 }
57
58 static void fio_gf_io_u_free(struct thread_data *td, struct io_u *io_u)
59 {
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 }
70
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;
74
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;
86     }
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
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
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)
114                 r = glfs_pwrite_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_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
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 }
171
172 static struct ioengine_ops ioengine = {
173         .name               = "gfapi_async",
174         .version            = FIO_IOOPS_VERSION,
175         .init           = fio_gf_async_setup,
176         .cleanup        = fio_gf_cleanup,
177     .prep           = fio_gf_async_prep,
178         .queue              = fio_gf_async_queue,
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,
184         .io_u_init      = fio_gf_io_u_init,
185         .io_u_free      = fio_gf_io_u_free,
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 }