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 <fcntl.h>
  12
  13 #include "../fio.h"
  14
  15 struct posixaio_data {
  16         struct io_u **aio_events;
  17         unsigned int queued;
  18 };
  19
  20 static int fill_timespec(struct timespec *ts)
  21 {
  22 #ifdef _POSIX_TIMERS
  23         if (!clock_gettime(CLOCK_MONOTONIC, ts))
  24                 return 0;
  25
  26         perror("clock_gettime");
  27 #endif
  28         return 1;
  29 }
  30
  31 static unsigned long long ts_utime_since_now(struct timespec *t)
  32 {
  33         long long sec, nsec;
  34         struct timespec now;
  35
  36         if (fill_timespec(&now))
  37                 return 0;
  38
  39         sec = now.tv_sec - t->tv_sec;
  40         nsec = now.tv_nsec - t->tv_nsec;
  41         if (sec > 0 && nsec < 0) {
  42                 sec--;
  43                 nsec += 1000000000;
  44         }
  45
  46         sec *= 1000000;
  47         nsec /= 1000;
  48         return sec + nsec;
  49 }
  50
  51 static int fio_posixaio_cancel(struct thread_data fio_unused *td,
  52                                struct io_u *io_u)
  53 {
  54         struct fio_file *f = io_u->file;
  55         int r = aio_cancel(f->fd, &io_u->aiocb);
  56
  57         if (r == AIO_ALLDONE || r == AIO_CANCELED)
  58                 return 0;
  59
  60         return 1;
  61 }
  62
  63 static int fio_posixaio_prep(struct thread_data fio_unused *td,
  64                              struct io_u *io_u)
  65 {
  66         os_aiocb_t *aiocb = &io_u->aiocb;
  67         struct fio_file *f = io_u->file;
  68
  69         aiocb->aio_fildes = f->fd;
  70         aiocb->aio_buf = io_u->xfer_buf;
  71         aiocb->aio_nbytes = io_u->xfer_buflen;
  72         aiocb->aio_offset = io_u->offset;
  73         aiocb->aio_sigevent.sigev_notify = SIGEV_NONE;
  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         os_aiocb_t *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->queued--;
 116                 pd->aio_events[r++] = io_u;
 117
 118                 if (err == ECANCELED)
 119                         io_u->resid = io_u->xfer_buflen;
 120                 else if (!err) {
 121                         ssize_t retval = aio_return(&io_u->aiocb);
 122
 123                         io_u->resid = io_u->xfer_buflen - retval;
 124                 } else
 125                         io_u->error = err;
 126         }
 127
 128         if (r >= min)
 129                 return r;
 130
 131         if (have_timeout) {
 132                 unsigned long long usec;
 133
 134                 usec = (t->tv_sec * 1000000) + (t->tv_nsec / 1000);
 135                 if (ts_utime_since_now(&start) > usec)
 136                         return r;
 137         }
 138
 139         /*
 140          * must have some in-flight, wait for at least one
 141          */
 142         aio_suspend((const os_aiocb_t * const *)suspend_list,
 143                                                         suspend_entries, t);
 144         goto restart;
 145 }
 146
 147 static struct io_u *fio_posixaio_event(struct thread_data *td, int event)
 148 {
 149         struct posixaio_data *pd = td->io_ops->data;
 150
 151         return pd->aio_events[event];
 152 }
 153
 154 static int fio_posixaio_queue(struct thread_data *td,
 155                               struct io_u *io_u)
 156 {
 157         struct posixaio_data *pd = td->io_ops->data;
 158         os_aiocb_t *aiocb = &io_u->aiocb;
 159         int ret;
 160
 161         fio_ro_check(td, io_u);
 162
 163         if (io_u->ddir == DDIR_READ)
 164                 ret = aio_read(aiocb);
 165         else if (io_u->ddir == DDIR_WRITE)
 166                 ret = aio_write(aiocb);
 167         else if (io_u->ddir == DDIR_TRIM) {
 168                 if (pd->queued)
 169                         return FIO_Q_BUSY;
 170
 171                 do_io_u_trim(td, io_u);
 172                 return FIO_Q_COMPLETED;
 173         } else {
 174 #ifdef CONFIG_POSIXAIO_FSYNC
 175                 ret = aio_fsync(O_SYNC, aiocb);
 176 #else
 177                 if (pd->queued)
 178                         return FIO_Q_BUSY;
 179
 180                 do_io_u_sync(td, io_u);
 181                 return FIO_Q_COMPLETED;
 182 #endif
 183         }
 184
 185         if (ret) {
 186                 /*
 187                  * At least OSX has a very low limit on the number of pending
 188                  * IOs, so if it returns EAGAIN, we are out of resources
 189                  * to queue more. Just return FIO_Q_BUSY to naturally
 190                  * drop off at this depth.
 191                  */
 192                 if (errno == EAGAIN)
 193                         return FIO_Q_BUSY;
 194
 195                 io_u->error = errno;
 196                 td_verror(td, io_u->error, "xfer");
 197                 return FIO_Q_COMPLETED;
 198         }
 199
 200         pd->queued++;
 201         return FIO_Q_QUEUED;
 202 }
 203
 204 static void fio_posixaio_cleanup(struct thread_data *td)
 205 {
 206         struct posixaio_data *pd = td->io_ops->data;
 207
 208         if (pd) {
 209                 free(pd->aio_events);
 210                 free(pd);
 211         }
 212 }
 213
 214 static int fio_posixaio_init(struct thread_data *td)
 215 {
 216         struct posixaio_data *pd = malloc(sizeof(*pd));
 217
 218         memset(pd, 0, sizeof(*pd));
 219         pd->aio_events = malloc(td->o.iodepth * sizeof(struct io_u *));
 220         memset(pd->aio_events, 0, td->o.iodepth * sizeof(struct io_u *));
 221
 222         td->io_ops->data = pd;
 223         return 0;
 224 }
 225
 226 static struct ioengine_ops ioengine = {
 227         .name           = "posixaio",
 228         .version        = FIO_IOOPS_VERSION,
 229         .init           = fio_posixaio_init,
 230         .prep           = fio_posixaio_prep,
 231         .queue          = fio_posixaio_queue,
 232         .cancel         = fio_posixaio_cancel,
 233         .getevents      = fio_posixaio_getevents,
 234         .event          = fio_posixaio_event,
 235         .cleanup        = fio_posixaio_cleanup,
 236         .open_file      = generic_open_file,
 237         .close_file     = generic_close_file,
 238         .get_file_size  = generic_get_file_size,
 239 };
 240
 241 static void fio_init fio_posixaio_register(void)
 242 {
 243         register_ioengine(&ioengine);
 244 }
 245
 246 static void fio_exit fio_posixaio_unregister(void)
 247 {
 248         unregister_ioengine(&ioengine);
 249 }