engines/posixaio.c

   1 /*
   2  * posixaio engine
   3  *
   4  * IO engine that uses the posix defined 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
  13 #include "../fio.h"
  14
  15 #ifdef FIO_HAVE_POSIXAIO
  16
  17 struct posixaio_data {
  18         struct io_u **aio_events;
  19 };
  20
  21 static int fill_timespec(struct timespec *ts)
  22 {
  23 #ifdef _POSIX_TIMERS
  24         if (!clock_gettime(CLOCK_MONOTONIC, ts))
  25                 return 0;
  26
  27         perror("clock_gettime");
  28 #endif
  29         return 1;
  30 }
  31
  32 static unsigned long long ts_utime_since_now(struct timespec *t)
  33 {
  34         long long sec, nsec;
  35         struct timespec now;
  36
  37         if (fill_timespec(&now))
  38                 return 0;
  39
  40         sec = now.tv_sec - t->tv_sec;
  41         nsec = now.tv_nsec - t->tv_nsec;
  42         if (sec > 0 && nsec < 0) {
  43                 sec--;
  44                 nsec += 1000000000;
  45         }
  46
  47         sec *= 1000000;
  48         nsec /= 1000;
  49         return sec + nsec;
  50 }
  51
  52 static int fio_posixaio_cancel(struct thread_data fio_unused *td,
  53                                struct io_u *io_u)
  54 {
  55         struct fio_file *f = io_u->file;
  56         int r = aio_cancel(f->fd, &io_u->aiocb);
  57
  58         if (r == 1 || r == AIO_CANCELED)
  59                 return 0;
  60
  61         return 1;
  62 }
  63
  64 static int fio_posixaio_prep(struct thread_data fio_unused *td,
  65                              struct io_u *io_u)
  66 {
  67         struct aiocb *aiocb = &io_u->aiocb;
  68         struct fio_file *f = io_u->file;
  69
  70         aiocb->aio_fildes = f->fd;
  71         aiocb->aio_buf = io_u->xfer_buf;
  72         aiocb->aio_nbytes = io_u->xfer_buflen;
  73         aiocb->aio_offset = io_u->offset;
  74
  75         io_u->seen = 0;
  76         return 0;
  77 }
  78
  79 #define SUSPEND_ENTRIES 8
  80
  81 static int fio_posixaio_getevents(struct thread_data *td, unsigned int min,
  82                                   unsigned int max, struct timespec *t)
  83 {
  84         struct posixaio_data *pd = td->io_ops->data;
  85         struct aiocb *suspend_list[SUSPEND_ENTRIES];
  86         struct flist_head *entry;
  87         struct timespec start;
  88         int have_timeout = 0;
  89         int suspend_entries = 0;
  90         unsigned int r;
  91
  92         if (t && !fill_timespec(&start))
  93                 have_timeout = 1;
  94
  95         r = 0;
  96         memset(suspend_list, 0, sizeof(*suspend_list));
  97 restart:
  98         flist_for_each(entry, &td->io_u_busylist) {
  99                 struct io_u *io_u = flist_entry(entry, struct io_u, list);
 100                 int err;
 101
 102                 if (io_u->seen)
 103                         continue;
 104
 105                 err = aio_error(&io_u->aiocb);
 106                 if (err == EINPROGRESS) {
 107                         if (suspend_entries < SUSPEND_ENTRIES) {
 108                                 suspend_list[suspend_entries] = &io_u->aiocb;
 109                                 suspend_entries++;
 110                         }
 111                         continue;
 112                 }
 113
 114                 io_u->seen = 1;
 115                 pd->aio_events[r++] = io_u;
 116
 117                 if (err == ECANCELED)
 118                         io_u->resid = io_u->xfer_buflen;
 119                 else if (!err) {
 120                         ssize_t retval = aio_return(&io_u->aiocb);
 121
 122                         io_u->resid = io_u->xfer_buflen - retval;
 123                 } else
 124                         io_u->error = err;
 125
 126                 if (r >= max)
 127                         break;
 128         }
 129
 130         if (r >= min)
 131                 return r;
 132
 133         if (have_timeout) {
 134                 unsigned long long usec;
 135
 136                 usec = (t->tv_sec * 1000000) + (t->tv_nsec / 1000);
 137                 if (ts_utime_since_now(&start) > usec)
 138                         return r;
 139         }
 140
 141         /*
 142          * must have some in-flight, wait for at least one
 143          */
 144         aio_suspend((const struct aiocb * const *)suspend_list,
 145                                                         suspend_entries, t);
 146         goto restart;
 147 }
 148
 149 static struct io_u *fio_posixaio_event(struct thread_data *td, int event)
 150 {
 151         struct posixaio_data *pd = td->io_ops->data;
 152
 153         return pd->aio_events[event];
 154 }
 155
 156 static int fio_posixaio_queue(struct thread_data fio_unused *td,
 157                               struct io_u *io_u)
 158 {
 159         struct aiocb *aiocb = &io_u->aiocb;
 160         int ret;
 161
 162         fio_ro_check(td, io_u);
 163
 164         if (io_u->ddir == DDIR_READ)
 165                 ret = aio_read(aiocb);
 166         else if (io_u->ddir == DDIR_WRITE)
 167                 ret = aio_write(aiocb);
 168         else
 169                 ret = aio_fsync(O_SYNC, aiocb);
 170
 171         if (ret) {
 172                 io_u->error = errno;
 173                 td_verror(td, io_u->error, "xfer");
 174                 return FIO_Q_COMPLETED;
 175         }
 176
 177         return FIO_Q_QUEUED;
 178 }
 179
 180 static void fio_posixaio_cleanup(struct thread_data *td)
 181 {
 182         struct posixaio_data *pd = td->io_ops->data;
 183
 184         if (pd) {
 185                 free(pd->aio_events);
 186                 free(pd);
 187         }
 188 }
 189
 190 static int fio_posixaio_init(struct thread_data *td)
 191 {
 192         struct posixaio_data *pd = malloc(sizeof(*pd));
 193
 194         memset(pd, 0, sizeof(*pd));
 195         pd->aio_events = malloc(td->o.iodepth * sizeof(struct io_u *));
 196         memset(pd->aio_events, 0, td->o.iodepth * sizeof(struct io_u *));
 197
 198         td->io_ops->data = pd;
 199         return 0;
 200 }
 201
 202 static struct ioengine_ops ioengine = {
 203         .name           = "posixaio",
 204         .version        = FIO_IOOPS_VERSION,
 205         .init           = fio_posixaio_init,
 206         .prep           = fio_posixaio_prep,
 207         .queue          = fio_posixaio_queue,
 208         .cancel         = fio_posixaio_cancel,
 209         .getevents      = fio_posixaio_getevents,
 210         .event          = fio_posixaio_event,
 211         .cleanup        = fio_posixaio_cleanup,
 212         .open_file      = generic_open_file,
 213         .close_file     = generic_close_file,
 214 };
 215
 216 #else /* FIO_HAVE_POSIXAIO */
 217
 218 /*
 219  * When we have a proper configure system in place, we simply wont build
 220  * and install this io engine. For now install a crippled version that
 221  * just complains and fails to load.
 222  */
 223 static int fio_posixaio_init(struct thread_data fio_unused *td)
 224 {
 225         fprintf(stderr, "fio: posixaio not available\n");
 226         return 1;
 227 }
 228
 229 static struct ioengine_ops ioengine = {
 230         .name           = "posixaio",
 231         .version        = FIO_IOOPS_VERSION,
 232         .init           = fio_posixaio_init,
 233 };
 234
 235 #endif
 236
 237 static void fio_init fio_posixaio_register(void)
 238 {
 239         register_ioengine(&ioengine);
 240 }
 241
 242 static void fio_exit fio_posixaio_unregister(void)
 243 {
 244         unregister_ioengine(&ioengine);
 245 }