engines/libaio.c

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