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_int.h"
  13 #include "block/raw-posix-aio.h"
  14
  15 #include <sys/eventfd.h>
  16 #include <libaio.h>
  17
  18 /*
  19  * Queue size (per-device).
  20  *
  21  * XXX: eventually we need to communicate this to the guest and/or make it
  22  *      tunable by the guest.  If we get more outstanding requests at a time
  23  *      than this we will get EAGAIN from io_submit which is communicated to
  24  *      the guest as an I/O error.
  25  */
  26 #define MAX_EVENTS 128
  27
  28 struct qemu_laiocb {
  29     BlockDriverAIOCB common;
  30     struct qemu_laio_state *ctx;
  31     struct iocb iocb;
  32     ssize_t ret;
  33     size_t nbytes;
  34     int async_context_id;
  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     QLIST_HEAD(, qemu_laiocb) completed_reqs;
  43 };
  44
  45 static inline ssize_t io_event_ret(struct io_event *ev)
  46 {
  47     return (ssize_t)(((uint64_t)ev->res2 << 32) | ev->res);
  48 }
  49
  50 /*
  51  * Completes an AIO request (calls the callback and frees the ACB).
  52  * Be sure to be in the right AsyncContext before calling this function.
  53  */
  54 static void qemu_laio_process_completion(struct qemu_laio_state *s,
  55     struct qemu_laiocb *laiocb)
  56 {
  57     int ret;
  58
  59     s->count--;
  60
  61     ret = laiocb->ret;
  62     if (ret != -ECANCELED) {
  63         if (ret == laiocb->nbytes)
  64             ret = 0;
  65         else if (ret >= 0)
  66             ret = -EINVAL;
  67
  68         laiocb->common.cb(laiocb->common.opaque, ret);
  69     }
  70
  71     qemu_aio_release(laiocb);
  72 }
  73
  74 /*
  75  * Processes all queued AIO requests, i.e. requests that have return from OS
  76  * but their callback was not called yet. Requests that cannot have their
  77  * callback called in the current AsyncContext, remain in the queue.
  78  *
  79  * Returns 1 if at least one request could be completed, 0 otherwise.
  80  */
  81 static int qemu_laio_process_requests(void *opaque)
  82 {
  83     struct qemu_laio_state *s = opaque;
  84     struct qemu_laiocb *laiocb, *next;
  85     int res = 0;
  86
  87     QLIST_FOREACH_SAFE (laiocb, &s->completed_reqs, node, next) {
  88         if (laiocb->async_context_id == get_async_context_id()) {
  89             qemu_laio_process_completion(s, laiocb);
  90             QLIST_REMOVE(laiocb, node);
  91             res = 1;
  92         }
  93     }
  94
  95     return res;
  96 }
  97
  98 /*
  99  * Puts a request in the completion queue so that its callback is called the
 100  * next time when it's possible. If we already are in the right AsyncContext,
 101  * the request is completed immediately instead.
 102  */
 103 static void qemu_laio_enqueue_completed(struct qemu_laio_state *s,
 104     struct qemu_laiocb* laiocb)
 105 {
 106     if (laiocb->async_context_id == get_async_context_id()) {
 107         qemu_laio_process_completion(s, laiocb);
 108     } else {
 109         QLIST_INSERT_HEAD(&s->completed_reqs, laiocb, node);
 110     }
 111 }
 112
 113 static void qemu_laio_completion_cb(void *opaque)
 114 {
 115     struct qemu_laio_state *s = opaque;
 116
 117     while (1) {
 118         struct io_event events[MAX_EVENTS];
 119         uint64_t val;
 120         ssize_t ret;
 121         struct timespec ts = { 0 };
 122         int nevents, i;
 123
 124         do {
 125             ret = read(s->efd, &val, sizeof(val));
 126         } while (ret == -1 && errno == EINTR);
 127
 128         if (ret == -1 && errno == EAGAIN)
 129             break;
 130
 131         if (ret != 8)
 132             break;
 133
 134         do {
 135             nevents = io_getevents(s->ctx, val, MAX_EVENTS, events, &ts);
 136         } while (nevents == -EINTR);
 137
 138         for (i = 0; i < nevents; i++) {
 139             struct iocb *iocb = events[i].obj;
 140             struct qemu_laiocb *laiocb =
 141                     container_of(iocb, struct qemu_laiocb, iocb);
 142
 143             laiocb->ret = io_event_ret(&events[i]);
 144             qemu_laio_enqueue_completed(s, laiocb);
 145         }
 146     }
 147 }
 148
 149 static int qemu_laio_flush_cb(void *opaque)
 150 {
 151     struct qemu_laio_state *s = opaque;
 152
 153     return (s->count > 0) ? 1 : 0;
 154 }
 155
 156 static void laio_cancel(BlockDriverAIOCB *blockacb)
 157 {
 158     struct qemu_laiocb *laiocb = (struct qemu_laiocb *)blockacb;
 159     struct io_event event;
 160     int ret;
 161
 162     if (laiocb->ret != -EINPROGRESS)
 163         return;
 164
 165     /*
 166      * Note that as of Linux 2.6.31 neither the block device code nor any
 167      * filesystem implements cancellation of AIO request.
 168      * Thus the polling loop below is the normal code path.
 169      */
 170     ret = io_cancel(laiocb->ctx->ctx, &laiocb->iocb, &event);
 171     if (ret == 0) {
 172         laiocb->ret = -ECANCELED;
 173         return;
 174     }
 175
 176     /*
 177      * We have to wait for the iocb to finish.
 178      *
 179      * The only way to get the iocb status update is by polling the io context.
 180      * We might be able to do this slightly more optimal by removing the
 181      * O_NONBLOCK flag.
 182      */
 183     while (laiocb->ret == -EINPROGRESS)
 184         qemu_laio_completion_cb(laiocb->ctx);
 185 }
 186
 187 static AIOPool laio_pool = {
 188     .aiocb_size         = sizeof(struct qemu_laiocb),
 189     .cancel             = laio_cancel,
 190 };
 191
 192 BlockDriverAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
 193         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
 194         BlockDriverCompletionFunc *cb, void *opaque, int type)
 195 {
 196     struct qemu_laio_state *s = aio_ctx;
 197     struct qemu_laiocb *laiocb;
 198     struct iocb *iocbs;
 199     off_t offset = sector_num * 512;
 200
 201     laiocb = qemu_aio_get(&laio_pool, bs, cb, opaque);
 202     if (!laiocb)
 203         return NULL;
 204     laiocb->nbytes = nb_sectors * 512;
 205     laiocb->ctx = s;
 206     laiocb->ret = -EINPROGRESS;
 207     laiocb->async_context_id = get_async_context_id();
 208
 209     iocbs = &laiocb->iocb;
 210
 211     switch (type) {
 212     case QEMU_AIO_WRITE:
 213         io_prep_pwritev(iocbs, fd, qiov->iov, qiov->niov, offset);
 214         break;
 215     case QEMU_AIO_READ:
 216         io_prep_preadv(iocbs, fd, qiov->iov, qiov->niov, offset);
 217         break;
 218     default:
 219         fprintf(stderr, "%s: invalid AIO request type 0x%x.\n",
 220                         __func__, type);
 221         goto out_free_aiocb;
 222     }
 223     io_set_eventfd(&laiocb->iocb, s->efd);
 224     s->count++;
 225
 226     if (io_submit(s->ctx, 1, &iocbs) < 0)
 227         goto out_dec_count;
 228     return &laiocb->common;
 229
 230 out_free_aiocb:
 231     qemu_aio_release(laiocb);
 232 out_dec_count:
 233     s->count--;
 234     return NULL;
 235 }
 236
 237 void *laio_init(void)
 238 {
 239     struct qemu_laio_state *s;
 240
 241     s = qemu_mallocz(sizeof(*s));
 242     QLIST_INIT(&s->completed_reqs);
 243     s->efd = eventfd(0, 0);
 244     if (s->efd == -1)
 245         goto out_free_state;
 246     fcntl(s->efd, F_SETFL, O_NONBLOCK);
 247
 248     if (io_setup(MAX_EVENTS, &s->ctx) != 0)
 249         goto out_close_efd;
 250
 251     qemu_aio_set_fd_handler(s->efd, qemu_laio_completion_cb, NULL,
 252         qemu_laio_flush_cb, qemu_laio_process_requests, s);
 253
 254     return s;
 255
 256 out_close_efd:
 257     close(s->efd);
 258 out_free_state:
 259     qemu_free(s);
 260     return NULL;
 261 }