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