linux-aio.c

   1 /*
   2  * Linux native AIO support.
   3  *
   4  * Copyright (C) 2009 IBM, Corp.
   5  * Copyright (C) 2009 Red Hat, Inc.
   6  *
   7  * This work is licensed under the terms of the GNU GPL, version 2 or later.
   8  * See the COPYING file in the top-level directory.
   9  */
  10 #include "qemu-common.h"
  11 #include "qemu-aio.h"
  12 #include "block/raw-posix-aio.h"
  13
  14 #include <sys/eventfd.h>
  15 #include <libaio.h>
  16
  17 /*
  18  * Queue size (per-device).
  19  *
  20  * XXX: eventually we need to communicate this to the guest and/or make it
  21  *      tunable by the guest.  If we get more outstanding requests at a time
  22  *      than this we will get EAGAIN from io_submit which is communicated to
  23  *      the guest as an I/O error.
  24  */
  25 #define MAX_EVENTS 128
  26
  27 struct qemu_laiocb {
  28     BlockDriverAIOCB common;
  29     struct qemu_laio_state *ctx;
  30     struct iocb iocb;
  31     ssize_t ret;
  32     size_t nbytes;
  33     QEMUIOVector *qiov;
  34     bool is_read;
  35     QLIST_ENTRY(qemu_laiocb) node;
  36 };
  37
  38 struct qemu_laio_state {
  39     io_context_t ctx;
  40     int efd;
  41     int count;
  42 };
  43
  44 static inline ssize_t io_event_ret(struct io_event *ev)
  45 {
  46     return (ssize_t)(((uint64_t)ev->res2 << 32) | ev->res);
  47 }
  48
  49 /*
  50  * Completes an AIO request (calls the callback and frees the ACB).
  51  */
  52 static void qemu_laio_process_completion(struct qemu_laio_state *s,
  53     struct qemu_laiocb *laiocb)
  54 {
  55     int ret;
  56
  57     s->count--;
  58
  59     ret = laiocb->ret;
  60     if (ret != -ECANCELED) {
  61         if (ret == laiocb->nbytes) {
  62             ret = 0;
  63         } else if (ret >= 0) {
  64             /* Short reads mean EOF, pad with zeros. */
  65             if (laiocb->is_read) {
  66                 qemu_iovec_memset(laiocb->qiov, ret, 0,
  67                     laiocb->qiov->size - ret);
  68             } else {
  69                 ret = -EINVAL;
  70             }
  71         }
  72
  73         laiocb->common.cb(laiocb->common.opaque, ret);
  74     }
  75
  76     qemu_aio_release(laiocb);
  77 }
  78
  79 static void qemu_laio_completion_cb(void *opaque)
  80 {
  81     struct qemu_laio_state *s = opaque;
  82
  83     while (1) {
  84         struct io_event events[MAX_EVENTS];
  85         uint64_t val;
  86         ssize_t ret;
  87         struct timespec ts = { 0 };
  88         int nevents, i;
  89
  90         do {
  91             ret = read(s->efd, &val, sizeof(val));
  92         } while (ret == -1 && errno == EINTR);
  93
  94         if (ret == -1 && errno == EAGAIN)
  95             break;
  96
  97         if (ret != 8)
  98             break;
  99
 100         do {
 101             nevents = io_getevents(s->ctx, val, MAX_EVENTS, events, &ts);
 102         } while (nevents == -EINTR);
 103
 104         for (i = 0; i < nevents; i++) {
 105             struct iocb *iocb = events[i].obj;
 106             struct qemu_laiocb *laiocb =
 107                     container_of(iocb, struct qemu_laiocb, iocb);
 108
 109             laiocb->ret = io_event_ret(&events[i]);
 110             qemu_laio_process_completion(s, laiocb);
 111         }
 112     }
 113 }
 114
 115 static int qemu_laio_flush_cb(void *opaque)
 116 {
 117     struct qemu_laio_state *s = opaque;
 118
 119     return (s->count > 0) ? 1 : 0;
 120 }
 121
 122 static void laio_cancel(BlockDriverAIOCB *blockacb)
 123 {
 124     struct qemu_laiocb *laiocb = (struct qemu_laiocb *)blockacb;
 125     struct io_event event;
 126     int ret;
 127
 128     if (laiocb->ret != -EINPROGRESS)
 129         return;
 130
 131     /*
 132      * Note that as of Linux 2.6.31 neither the block device code nor any
 133      * filesystem implements cancellation of AIO request.
 134      * Thus the polling loop below is the normal code path.
 135      */
 136     ret = io_cancel(laiocb->ctx->ctx, &laiocb->iocb, &event);
 137     if (ret == 0) {
 138         laiocb->ret = -ECANCELED;
 139         return;
 140     }
 141
 142     /*
 143      * We have to wait for the iocb to finish.
 144      *
 145      * The only way to get the iocb status update is by polling the io context.
 146      * We might be able to do this slightly more optimal by removing the
 147      * O_NONBLOCK flag.
 148      */
 149     while (laiocb->ret == -EINPROGRESS)
 150         qemu_laio_completion_cb(laiocb->ctx);
 151 }
 152
 153 static AIOPool laio_pool = {
 154     .aiocb_size         = sizeof(struct qemu_laiocb),
 155     .cancel             = laio_cancel,
 156 };
 157
 158 BlockDriverAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
 159         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
 160         BlockDriverCompletionFunc *cb, void *opaque, int type)
 161 {
 162     struct qemu_laio_state *s = aio_ctx;
 163     struct qemu_laiocb *laiocb;
 164     struct iocb *iocbs;
 165     off_t offset = sector_num * 512;
 166
 167     laiocb = qemu_aio_get(&laio_pool, bs, cb, opaque);
 168     laiocb->nbytes = nb_sectors * 512;
 169     laiocb->ctx = s;
 170     laiocb->ret = -EINPROGRESS;
 171     laiocb->is_read = (type == QEMU_AIO_READ);
 172     laiocb->qiov = qiov;
 173
 174     iocbs = &laiocb->iocb;
 175
 176     switch (type) {
 177     case QEMU_AIO_WRITE:
 178         io_prep_pwritev(iocbs, fd, qiov->iov, qiov->niov, offset);
 179         break;
 180     case QEMU_AIO_READ:
 181         io_prep_preadv(iocbs, fd, qiov->iov, qiov->niov, offset);
 182         break;
 183     /* Currently Linux kernel does not support other operations */
 184     default:
 185         fprintf(stderr, "%s: invalid AIO request type 0x%x.\n",
 186                         __func__, type);
 187         goto out_free_aiocb;
 188     }
 189     io_set_eventfd(&laiocb->iocb, s->efd);
 190     s->count++;
 191
 192     if (io_submit(s->ctx, 1, &iocbs) < 0)
 193         goto out_dec_count;
 194     return &laiocb->common;
 195
 196 out_dec_count:
 197     s->count--;
 198 out_free_aiocb:
 199     qemu_aio_release(laiocb);
 200     return NULL;
 201 }
 202
 203 void *laio_init(void)
 204 {
 205     struct qemu_laio_state *s;
 206
 207     s = g_malloc0(sizeof(*s));
 208     s->efd = eventfd(0, 0);
 209     if (s->efd == -1)
 210         goto out_free_state;
 211     fcntl(s->efd, F_SETFL, O_NONBLOCK);
 212
 213     if (io_setup(MAX_EVENTS, &s->ctx) != 0)
 214         goto out_close_efd;
 215
 216     qemu_aio_set_fd_handler(s->efd, qemu_laio_completion_cb, NULL,
 217         qemu_laio_flush_cb, s);
 218
 219     return s;
 220
 221 out_close_efd:
 222     close(s->efd);
 223 out_free_state:
 224     g_free(s);
 225     return NULL;
 226 }