block/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 "block/aio.h"
  12 #include "qemu/queue.h"
  13 #include "block/raw-aio.h"
  14 #include "qemu/event_notifier.h"
  15
  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     QEMUIOVector *qiov;
  35     bool is_read;
  36     QLIST_ENTRY(qemu_laiocb) node;
  37 };
  38
  39 struct qemu_laio_state {
  40     io_context_t ctx;
  41     EventNotifier e;
  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     ret = laiocb->ret;
  58     if (ret != -ECANCELED) {
  59         if (ret == laiocb->nbytes) {
  60             ret = 0;
  61         } else if (ret >= 0) {
  62             /* Short reads mean EOF, pad with zeros. */
  63             if (laiocb->is_read) {
  64                 qemu_iovec_memset(laiocb->qiov, ret, 0,
  65                     laiocb->qiov->size - ret);
  66             } else {
  67                 ret = -EINVAL;
  68             }
  69         }
  70
  71         laiocb->common.cb(laiocb->common.opaque, ret);
  72     }
  73
  74     qemu_aio_release(laiocb);
  75 }
  76
  77 static void qemu_laio_completion_cb(EventNotifier *e)
  78 {
  79     struct qemu_laio_state *s = container_of(e, struct qemu_laio_state, e);
  80
  81     while (event_notifier_test_and_clear(&s->e)) {
  82         struct io_event events[MAX_EVENTS];
  83         struct timespec ts = { 0 };
  84         int nevents, i;
  85
  86         do {
  87             nevents = io_getevents(s->ctx, MAX_EVENTS, MAX_EVENTS, events, &ts);
  88         } while (nevents == -EINTR);
  89
  90         for (i = 0; i < nevents; i++) {
  91             struct iocb *iocb = events[i].obj;
  92             struct qemu_laiocb *laiocb =
  93                     container_of(iocb, struct qemu_laiocb, iocb);
  94
  95             laiocb->ret = io_event_ret(&events[i]);
  96             qemu_laio_process_completion(s, laiocb);
  97         }
  98     }
  99 }
 100
 101 static void laio_cancel(BlockDriverAIOCB *blockacb)
 102 {
 103     struct qemu_laiocb *laiocb = (struct qemu_laiocb *)blockacb;
 104     struct io_event event;
 105     int ret;
 106
 107     if (laiocb->ret != -EINPROGRESS)
 108         return;
 109
 110     /*
 111      * Note that as of Linux 2.6.31 neither the block device code nor any
 112      * filesystem implements cancellation of AIO request.
 113      * Thus the polling loop below is the normal code path.
 114      */
 115     ret = io_cancel(laiocb->ctx->ctx, &laiocb->iocb, &event);
 116     if (ret == 0) {
 117         laiocb->ret = -ECANCELED;
 118         return;
 119     }
 120
 121     /*
 122      * We have to wait for the iocb to finish.
 123      *
 124      * The only way to get the iocb status update is by polling the io context.
 125      * We might be able to do this slightly more optimal by removing the
 126      * O_NONBLOCK flag.
 127      */
 128     while (laiocb->ret == -EINPROGRESS) {
 129         qemu_laio_completion_cb(&laiocb->ctx->e);
 130     }
 131 }
 132
 133 static const AIOCBInfo laio_aiocb_info = {
 134     .aiocb_size         = sizeof(struct qemu_laiocb),
 135     .cancel             = laio_cancel,
 136 };
 137
 138 BlockDriverAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
 139         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
 140         BlockDriverCompletionFunc *cb, void *opaque, int type)
 141 {
 142     struct qemu_laio_state *s = aio_ctx;
 143     struct qemu_laiocb *laiocb;
 144     struct iocb *iocbs;
 145     off_t offset = sector_num * 512;
 146
 147     laiocb = qemu_aio_get(&laio_aiocb_info, bs, cb, opaque);
 148     laiocb->nbytes = nb_sectors * 512;
 149     laiocb->ctx = s;
 150     laiocb->ret = -EINPROGRESS;
 151     laiocb->is_read = (type == QEMU_AIO_READ);
 152     laiocb->qiov = qiov;
 153
 154     iocbs = &laiocb->iocb;
 155
 156     switch (type) {
 157     case QEMU_AIO_WRITE:
 158         io_prep_pwritev(iocbs, fd, qiov->iov, qiov->niov, offset);
 159         break;
 160     case QEMU_AIO_READ:
 161         io_prep_preadv(iocbs, fd, qiov->iov, qiov->niov, offset);
 162         break;
 163     /* Currently Linux kernel does not support other operations */
 164     default:
 165         fprintf(stderr, "%s: invalid AIO request type 0x%x.\n",
 166                         __func__, type);
 167         goto out_free_aiocb;
 168     }
 169     io_set_eventfd(&laiocb->iocb, event_notifier_get_fd(&s->e));
 170
 171     if (io_submit(s->ctx, 1, &iocbs) < 0)
 172         goto out_free_aiocb;
 173     return &laiocb->common;
 174
 175 out_free_aiocb:
 176     qemu_aio_release(laiocb);
 177     return NULL;
 178 }
 179
 180 void laio_detach_aio_context(void *s_, AioContext *old_context)
 181 {
 182     struct qemu_laio_state *s = s_;
 183
 184     aio_set_event_notifier(old_context, &s->e, NULL);
 185 }
 186
 187 void laio_attach_aio_context(void *s_, AioContext *new_context)
 188 {
 189     struct qemu_laio_state *s = s_;
 190
 191     aio_set_event_notifier(new_context, &s->e, qemu_laio_completion_cb);
 192 }
 193
 194 void *laio_init(void)
 195 {
 196     struct qemu_laio_state *s;
 197
 198     s = g_malloc0(sizeof(*s));
 199     if (event_notifier_init(&s->e, false) < 0) {
 200         goto out_free_state;
 201     }
 202
 203     if (io_setup(MAX_EVENTS, &s->ctx) != 0) {
 204         goto out_close_efd;
 205     }
 206
 207     return s;
 208
 209 out_close_efd:
 210     event_notifier_cleanup(&s->e);
 211 out_free_state:
 212     g_free(s);
 213     return NULL;
 214 }
 215
 216 void laio_cleanup(void *s_)
 217 {
 218     struct qemu_laio_state *s = s_;
 219
 220     event_notifier_cleanup(&s->e);
 221     g_free(s);
 222 }