libaio: switch to newer libaio polled IO API
[fio.git] / engines / libaio.c
1 /*
2  * libaio engine
3  *
4  * IO engine using the Linux native aio interface.
5  *
6  */
7 #include <stdlib.h>
8 #include <unistd.h>
9 #include <errno.h>
10 #include <libaio.h>
11
12 #include "../fio.h"
13 #include "../lib/pow2.h"
14 #include "../optgroup.h"
15
16 #ifndef IOCB_FLAG_HIPRI
17 #define IOCB_FLAG_HIPRI (1 << 2)
18 #endif
19
20 static int fio_libaio_commit(struct thread_data *td);
21
22 struct libaio_data {
23         io_context_t aio_ctx;
24         struct io_event *aio_events;
25         struct iocb **iocbs;
26         struct io_u **io_us;
27
28         /*
29          * Basic ring buffer. 'head' is incremented in _queue(), and
30          * 'tail' is incremented in _commit(). We keep 'queued' so
31          * that we know if the ring is full or empty, when
32          * 'head' == 'tail'. 'entries' is the ring size, and
33          * 'is_pow2' is just an optimization to use AND instead of
34          * modulus to get the remainder on ring increment.
35          */
36         int is_pow2;
37         unsigned int entries;
38         unsigned int queued;
39         unsigned int head;
40         unsigned int tail;
41 };
42
43 struct libaio_options {
44         void *pad;
45         unsigned int userspace_reap;
46         unsigned int hipri;
47 };
48
49 static struct fio_option options[] = {
50         {
51                 .name   = "userspace_reap",
52                 .lname  = "Libaio userspace reaping",
53                 .type   = FIO_OPT_STR_SET,
54                 .off1   = offsetof(struct libaio_options, userspace_reap),
55                 .help   = "Use alternative user-space reap implementation",
56                 .category = FIO_OPT_C_ENGINE,
57                 .group  = FIO_OPT_G_LIBAIO,
58         },
59         {
60                 .name   = "hipri",
61                 .lname  = "High Priority",
62                 .type   = FIO_OPT_STR_SET,
63                 .off1   = offsetof(struct libaio_options, hipri),
64                 .help   = "Use polled IO completions",
65                 .category = FIO_OPT_C_ENGINE,
66                 .group  = FIO_OPT_G_LIBAIO,
67         },
68         {
69                 .name   = NULL,
70         },
71 };
72
73 static inline void ring_inc(struct libaio_data *ld, unsigned int *val,
74                             unsigned int add)
75 {
76         if (ld->is_pow2)
77                 *val = (*val + add) & (ld->entries - 1);
78         else
79                 *val = (*val + add) % ld->entries;
80 }
81
82 static int fio_libaio_prep(struct thread_data fio_unused *td, struct io_u *io_u)
83 {
84         struct fio_file *f = io_u->file;
85         struct libaio_options *o = td->eo;
86
87         if (io_u->ddir == DDIR_READ) {
88                 io_prep_pread(&io_u->iocb, f->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
89                 if (o->hipri)
90                         io_u->iocb.u.c.flags |= IOCB_FLAG_HIPRI;
91         } else if (io_u->ddir == DDIR_WRITE) {
92                 io_prep_pwrite(&io_u->iocb, f->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
93                 if (o->hipri)
94                         io_u->iocb.u.c.flags |= IOCB_FLAG_HIPRI;
95         } else if (ddir_sync(io_u->ddir))
96                 io_prep_fsync(&io_u->iocb, f->fd);
97
98         return 0;
99 }
100
101 static struct io_u *fio_libaio_event(struct thread_data *td, int event)
102 {
103         struct libaio_data *ld = td->io_ops_data;
104         struct io_event *ev;
105         struct io_u *io_u;
106
107         ev = ld->aio_events + event;
108         io_u = container_of(ev->obj, struct io_u, iocb);
109
110         if (ev->res != io_u->xfer_buflen) {
111                 if (ev->res > io_u->xfer_buflen)
112                         io_u->error = -ev->res;
113                 else
114                         io_u->resid = io_u->xfer_buflen - ev->res;
115         } else
116                 io_u->error = 0;
117
118         return io_u;
119 }
120
121 struct aio_ring {
122         unsigned id;             /** kernel internal index number */
123         unsigned nr;             /** number of io_events */
124         unsigned head;
125         unsigned tail;
126
127         unsigned magic;
128         unsigned compat_features;
129         unsigned incompat_features;
130         unsigned header_length; /** size of aio_ring */
131
132         struct io_event events[0];
133 };
134
135 #define AIO_RING_MAGIC  0xa10a10a1
136
137 static int user_io_getevents(io_context_t aio_ctx, unsigned int max,
138                              struct io_event *events)
139 {
140         long i = 0;
141         unsigned head;
142         struct aio_ring *ring = (struct aio_ring*) aio_ctx;
143
144         while (i < max) {
145                 head = ring->head;
146
147                 if (head == ring->tail) {
148                         /* There are no more completions */
149                         break;
150                 } else {
151                         /* There is another completion to reap */
152                         events[i] = ring->events[head];
153                         read_barrier();
154                         ring->head = (head + 1) % ring->nr;
155                         i++;
156                 }
157         }
158
159         return i;
160 }
161
162 static int fio_libaio_getevents(struct thread_data *td, unsigned int min,
163                                 unsigned int max, const struct timespec *t)
164 {
165         struct libaio_data *ld = td->io_ops_data;
166         struct libaio_options *o = td->eo;
167         unsigned actual_min = td->o.iodepth_batch_complete_min == 0 ? 0 : min;
168         struct timespec __lt, *lt = NULL;
169         int r, events = 0;
170
171         if (t) {
172                 __lt = *t;
173                 lt = &__lt;
174         }
175
176         do {
177                 if (o->userspace_reap == 1
178                     && actual_min == 0
179                     && ((struct aio_ring *)(ld->aio_ctx))->magic
180                                 == AIO_RING_MAGIC) {
181                         r = user_io_getevents(ld->aio_ctx, max,
182                                 ld->aio_events + events);
183                 } else {
184                         r = io_getevents(ld->aio_ctx, actual_min,
185                                 max, ld->aio_events + events, lt);
186                 }
187                 if (r > 0)
188                         events += r;
189                 else if ((min && r == 0) || r == -EAGAIN) {
190                         fio_libaio_commit(td);
191                         if (actual_min)
192                                 usleep(10);
193                 } else if (r != -EINTR)
194                         break;
195         } while (events < min);
196
197         return r < 0 ? r : events;
198 }
199
200 static enum fio_q_status fio_libaio_queue(struct thread_data *td,
201                                           struct io_u *io_u)
202 {
203         struct libaio_data *ld = td->io_ops_data;
204
205         fio_ro_check(td, io_u);
206
207         if (ld->queued == td->o.iodepth)
208                 return FIO_Q_BUSY;
209
210         /*
211          * fsync is tricky, since it can fail and we need to do it
212          * serialized with other io. the reason is that linux doesn't
213          * support aio fsync yet. So return busy for the case where we
214          * have pending io, to let fio complete those first.
215          */
216         if (ddir_sync(io_u->ddir)) {
217                 if (ld->queued)
218                         return FIO_Q_BUSY;
219
220                 do_io_u_sync(td, io_u);
221                 return FIO_Q_COMPLETED;
222         }
223
224         if (io_u->ddir == DDIR_TRIM) {
225                 if (ld->queued)
226                         return FIO_Q_BUSY;
227
228                 do_io_u_trim(td, io_u);
229                 io_u_mark_submit(td, 1);
230                 io_u_mark_complete(td, 1);
231                 return FIO_Q_COMPLETED;
232         }
233
234         ld->iocbs[ld->head] = &io_u->iocb;
235         ld->io_us[ld->head] = io_u;
236         ring_inc(ld, &ld->head, 1);
237         ld->queued++;
238         return FIO_Q_QUEUED;
239 }
240
241 static void fio_libaio_queued(struct thread_data *td, struct io_u **io_us,
242                               unsigned int nr)
243 {
244         struct timespec now;
245         unsigned int i;
246
247         if (!fio_fill_issue_time(td))
248                 return;
249
250         fio_gettime(&now, NULL);
251
252         for (i = 0; i < nr; i++) {
253                 struct io_u *io_u = io_us[i];
254
255                 memcpy(&io_u->issue_time, &now, sizeof(now));
256                 io_u_queued(td, io_u);
257         }
258 }
259
260 static int fio_libaio_commit(struct thread_data *td)
261 {
262         struct libaio_data *ld = td->io_ops_data;
263         struct iocb **iocbs;
264         struct io_u **io_us;
265         struct timespec ts;
266         int ret, wait_start = 0;
267
268         if (!ld->queued)
269                 return 0;
270
271         do {
272                 long nr = ld->queued;
273
274                 nr = min((unsigned int) nr, ld->entries - ld->tail);
275                 io_us = ld->io_us + ld->tail;
276                 iocbs = ld->iocbs + ld->tail;
277
278                 ret = io_submit(ld->aio_ctx, nr, iocbs);
279                 if (ret > 0) {
280                         fio_libaio_queued(td, io_us, ret);
281                         io_u_mark_submit(td, ret);
282
283                         ld->queued -= ret;
284                         ring_inc(ld, &ld->tail, ret);
285                         ret = 0;
286                         wait_start = 0;
287                 } else if (ret == -EINTR || !ret) {
288                         if (!ret)
289                                 io_u_mark_submit(td, ret);
290                         wait_start = 0;
291                         continue;
292                 } else if (ret == -EAGAIN) {
293                         /*
294                          * If we get EAGAIN, we should break out without
295                          * error and let the upper layer reap some
296                          * events for us. If we have no queued IO, we
297                          * must loop here. If we loop for more than 30s,
298                          * just error out, something must be buggy in the
299                          * IO path.
300                          */
301                         if (ld->queued) {
302                                 ret = 0;
303                                 break;
304                         }
305                         if (!wait_start) {
306                                 fio_gettime(&ts, NULL);
307                                 wait_start = 1;
308                         } else if (mtime_since_now(&ts) > 30000) {
309                                 log_err("fio: aio appears to be stalled, giving up\n");
310                                 break;
311                         }
312                         usleep(1);
313                         continue;
314                 } else if (ret == -ENOMEM) {
315                         /*
316                          * If we get -ENOMEM, reap events if we can. If
317                          * we cannot, treat it as a fatal event since there's
318                          * nothing we can do about it.
319                          */
320                         if (ld->queued)
321                                 ret = 0;
322                         break;
323                 } else
324                         break;
325         } while (ld->queued);
326
327         return ret;
328 }
329
330 static int fio_libaio_cancel(struct thread_data *td, struct io_u *io_u)
331 {
332         struct libaio_data *ld = td->io_ops_data;
333
334         return io_cancel(ld->aio_ctx, &io_u->iocb, ld->aio_events);
335 }
336
337 static void fio_libaio_cleanup(struct thread_data *td)
338 {
339         struct libaio_data *ld = td->io_ops_data;
340
341         if (ld) {
342                 /*
343                  * Work-around to avoid huge RCU stalls at exit time. If we
344                  * don't do this here, then it'll be torn down by exit_aio().
345                  * But for that case we can parallellize the freeing, thus
346                  * speeding it up a lot.
347                  */
348                 if (!(td->flags & TD_F_CHILD))
349                         io_destroy(ld->aio_ctx);
350                 free(ld->aio_events);
351                 free(ld->iocbs);
352                 free(ld->io_us);
353                 free(ld);
354         }
355 }
356
357 static int fio_libaio_init(struct thread_data *td)
358 {
359         struct libaio_options *o = td->eo;
360         struct libaio_data *ld;
361         int err = 0;
362
363         ld = calloc(1, sizeof(*ld));
364
365         /*
366          * First try passing in 0 for queue depth, since we don't
367          * care about the user ring. If that fails, the kernel is too old
368          * and we need the right depth.
369          */
370         if (!o->userspace_reap)
371                 err = io_queue_init(INT_MAX, &ld->aio_ctx);
372         if (o->userspace_reap || err == -EINVAL)
373                 err = io_queue_init(td->o.iodepth, &ld->aio_ctx);
374         if (err) {
375                 td_verror(td, -err, "io_queue_init");
376                 log_err("fio: check /proc/sys/fs/aio-max-nr\n");
377                 free(ld);
378                 return 1;
379         }
380
381         ld->entries = td->o.iodepth;
382         ld->is_pow2 = is_power_of_2(ld->entries);
383         ld->aio_events = calloc(ld->entries, sizeof(struct io_event));
384         ld->iocbs = calloc(ld->entries, sizeof(struct iocb *));
385         ld->io_us = calloc(ld->entries, sizeof(struct io_u *));
386
387         td->io_ops_data = ld;
388         return 0;
389 }
390
391 static struct ioengine_ops ioengine = {
392         .name                   = "libaio",
393         .version                = FIO_IOOPS_VERSION,
394         .init                   = fio_libaio_init,
395         .prep                   = fio_libaio_prep,
396         .queue                  = fio_libaio_queue,
397         .commit                 = fio_libaio_commit,
398         .cancel                 = fio_libaio_cancel,
399         .getevents              = fio_libaio_getevents,
400         .event                  = fio_libaio_event,
401         .cleanup                = fio_libaio_cleanup,
402         .open_file              = generic_open_file,
403         .close_file             = generic_close_file,
404         .get_file_size          = generic_get_file_size,
405         .options                = options,
406         .option_struct_size     = sizeof(struct libaio_options),
407 };
408
409 static void fio_init fio_libaio_register(void)
410 {
411         register_ioengine(&ioengine);
412 }
413
414 static void fio_exit fio_libaio_unregister(void)
415 {
416         unregister_ioengine(&ioengine);
417 }