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 #define MAX_QUEUED_IO  128
  29
  30 struct qemu_laiocb {
  31     BlockDriverAIOCB common;
  32     struct qemu_laio_state *ctx;
  33     struct iocb iocb;
  34     ssize_t ret;
  35     size_t nbytes;
  36     QEMUIOVector *qiov;
  37     bool is_read;
  38     QLIST_ENTRY(qemu_laiocb) node;
  39 };
  40
  41 typedef struct {
  42     struct iocb *iocbs[MAX_QUEUED_IO];
  43     int plugged;
  44     unsigned int size;
  45     unsigned int idx;
  46 } LaioQueue;
  47
  48 struct qemu_laio_state {
  49     io_context_t ctx;
  50     EventNotifier e;
  51
  52     /* io queue for submit at batch */
  53     LaioQueue io_q;
  54
  55     /* I/O completion processing */
  56     QEMUBH *completion_bh;
  57     struct io_event events[MAX_EVENTS];
  58     int event_idx;
  59     int event_max;
  60 };
  61
  62 static inline ssize_t io_event_ret(struct io_event *ev)
  63 {
  64     return (ssize_t)(((uint64_t)ev->res2 << 32) | ev->res);
  65 }
  66
  67 /*
  68  * Completes an AIO request (calls the callback and frees the ACB).
  69  */
  70 static void qemu_laio_process_completion(struct qemu_laio_state *s,
  71     struct qemu_laiocb *laiocb)
  72 {
  73     int ret;
  74
  75     ret = laiocb->ret;
  76     if (ret != -ECANCELED) {
  77         if (ret == laiocb->nbytes) {
  78             ret = 0;
  79         } else if (ret >= 0) {
  80             /* Short reads mean EOF, pad with zeros. */
  81             if (laiocb->is_read) {
  82                 qemu_iovec_memset(laiocb->qiov, ret, 0,
  83                     laiocb->qiov->size - ret);
  84             } else {
  85                 ret = -EINVAL;
  86             }
  87         }
  88     }
  89     laiocb->common.cb(laiocb->common.opaque, ret);
  90
  91     qemu_aio_unref(laiocb);
  92 }
  93
  94 /* The completion BH fetches completed I/O requests and invokes their
  95  * callbacks.
  96  *
  97  * The function is somewhat tricky because it supports nested event loops, for
  98  * example when a request callback invokes aio_poll().  In order to do this,
  99  * the completion events array and index are kept in qemu_laio_state.  The BH
 100  * reschedules itself as long as there are completions pending so it will
 101  * either be called again in a nested event loop or will be called after all
 102  * events have been completed.  When there are no events left to complete, the
 103  * BH returns without rescheduling.
 104  */
 105 static void qemu_laio_completion_bh(void *opaque)
 106 {
 107     struct qemu_laio_state *s = opaque;
 108
 109     /* Fetch more completion events when empty */
 110     if (s->event_idx == s->event_max) {
 111         do {
 112             struct timespec ts = { 0 };
 113             s->event_max = io_getevents(s->ctx, MAX_EVENTS, MAX_EVENTS,
 114                                         s->events, &ts);
 115         } while (s->event_max == -EINTR);
 116
 117         s->event_idx = 0;
 118         if (s->event_max <= 0) {
 119             s->event_max = 0;
 120             return; /* no more events */
 121         }
 122     }
 123
 124     /* Reschedule so nested event loops see currently pending completions */
 125     qemu_bh_schedule(s->completion_bh);
 126
 127     /* Process completion events */
 128     while (s->event_idx < s->event_max) {
 129         struct iocb *iocb = s->events[s->event_idx].obj;
 130         struct qemu_laiocb *laiocb =
 131                 container_of(iocb, struct qemu_laiocb, iocb);
 132
 133         laiocb->ret = io_event_ret(&s->events[s->event_idx]);
 134         s->event_idx++;
 135
 136         qemu_laio_process_completion(s, laiocb);
 137     }
 138 }
 139
 140 static void qemu_laio_completion_cb(EventNotifier *e)
 141 {
 142     struct qemu_laio_state *s = container_of(e, struct qemu_laio_state, e);
 143
 144     if (event_notifier_test_and_clear(&s->e)) {
 145         qemu_bh_schedule(s->completion_bh);
 146     }
 147 }
 148
 149 static void laio_cancel(BlockDriverAIOCB *blockacb)
 150 {
 151     struct qemu_laiocb *laiocb = (struct qemu_laiocb *)blockacb;
 152     struct io_event event;
 153     int ret;
 154
 155     if (laiocb->ret != -EINPROGRESS) {
 156         return;
 157     }
 158     ret = io_cancel(laiocb->ctx->ctx, &laiocb->iocb, &event);
 159     laiocb->ret = -ECANCELED;
 160     if (ret != 0) {
 161         /* iocb is not cancelled, cb will be called by the event loop later */
 162         return;
 163     }
 164
 165     laiocb->common.cb(laiocb->common.opaque, laiocb->ret);
 166 }
 167
 168 static const AIOCBInfo laio_aiocb_info = {
 169     .aiocb_size         = sizeof(struct qemu_laiocb),
 170     .cancel_async       = laio_cancel,
 171 };
 172
 173 static void ioq_init(LaioQueue *io_q)
 174 {
 175     io_q->size = MAX_QUEUED_IO;
 176     io_q->idx = 0;
 177     io_q->plugged = 0;
 178 }
 179
 180 static int ioq_submit(struct qemu_laio_state *s)
 181 {
 182     int ret, i = 0;
 183     int len = s->io_q.idx;
 184
 185     do {
 186         ret = io_submit(s->ctx, len, s->io_q.iocbs);
 187     } while (i++ < 3 && ret == -EAGAIN);
 188
 189     /* empty io queue */
 190     s->io_q.idx = 0;
 191
 192     if (ret < 0) {
 193         i = 0;
 194     } else {
 195         i = ret;
 196     }
 197
 198     for (; i < len; i++) {
 199         struct qemu_laiocb *laiocb =
 200             container_of(s->io_q.iocbs[i], struct qemu_laiocb, iocb);
 201
 202         laiocb->ret = (ret < 0) ? ret : -EIO;
 203         qemu_laio_process_completion(s, laiocb);
 204     }
 205     return ret;
 206 }
 207
 208 static void ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb)
 209 {
 210     unsigned int idx = s->io_q.idx;
 211
 212     s->io_q.iocbs[idx++] = iocb;
 213     s->io_q.idx = idx;
 214
 215     /* submit immediately if queue is full */
 216     if (idx == s->io_q.size) {
 217         ioq_submit(s);
 218     }
 219 }
 220
 221 void laio_io_plug(BlockDriverState *bs, void *aio_ctx)
 222 {
 223     struct qemu_laio_state *s = aio_ctx;
 224
 225     s->io_q.plugged++;
 226 }
 227
 228 int laio_io_unplug(BlockDriverState *bs, void *aio_ctx, bool unplug)
 229 {
 230     struct qemu_laio_state *s = aio_ctx;
 231     int ret = 0;
 232
 233     assert(s->io_q.plugged > 0 || !unplug);
 234
 235     if (unplug && --s->io_q.plugged > 0) {
 236         return 0;
 237     }
 238
 239     if (s->io_q.idx > 0) {
 240         ret = ioq_submit(s);
 241     }
 242
 243     return ret;
 244 }
 245
 246 BlockDriverAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
 247         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
 248         BlockDriverCompletionFunc *cb, void *opaque, int type)
 249 {
 250     struct qemu_laio_state *s = aio_ctx;
 251     struct qemu_laiocb *laiocb;
 252     struct iocb *iocbs;
 253     off_t offset = sector_num * 512;
 254
 255     laiocb = qemu_aio_get(&laio_aiocb_info, bs, cb, opaque);
 256     laiocb->nbytes = nb_sectors * 512;
 257     laiocb->ctx = s;
 258     laiocb->ret = -EINPROGRESS;
 259     laiocb->is_read = (type == QEMU_AIO_READ);
 260     laiocb->qiov = qiov;
 261
 262     iocbs = &laiocb->iocb;
 263
 264     switch (type) {
 265     case QEMU_AIO_WRITE:
 266         io_prep_pwritev(iocbs, fd, qiov->iov, qiov->niov, offset);
 267         break;
 268     case QEMU_AIO_READ:
 269         io_prep_preadv(iocbs, fd, qiov->iov, qiov->niov, offset);
 270         break;
 271     /* Currently Linux kernel does not support other operations */
 272     default:
 273         fprintf(stderr, "%s: invalid AIO request type 0x%x.\n",
 274                         __func__, type);
 275         goto out_free_aiocb;
 276     }
 277     io_set_eventfd(&laiocb->iocb, event_notifier_get_fd(&s->e));
 278
 279     if (!s->io_q.plugged) {
 280         if (io_submit(s->ctx, 1, &iocbs) < 0) {
 281             goto out_free_aiocb;
 282         }
 283     } else {
 284         ioq_enqueue(s, iocbs);
 285     }
 286     return &laiocb->common;
 287
 288 out_free_aiocb:
 289     qemu_aio_unref(laiocb);
 290     return NULL;
 291 }
 292
 293 void laio_detach_aio_context(void *s_, AioContext *old_context)
 294 {
 295     struct qemu_laio_state *s = s_;
 296
 297     aio_set_event_notifier(old_context, &s->e, NULL);
 298     qemu_bh_delete(s->completion_bh);
 299 }
 300
 301 void laio_attach_aio_context(void *s_, AioContext *new_context)
 302 {
 303     struct qemu_laio_state *s = s_;
 304
 305     s->completion_bh = aio_bh_new(new_context, qemu_laio_completion_bh, s);
 306     aio_set_event_notifier(new_context, &s->e, qemu_laio_completion_cb);
 307 }
 308
 309 void *laio_init(void)
 310 {
 311     struct qemu_laio_state *s;
 312
 313     s = g_malloc0(sizeof(*s));
 314     if (event_notifier_init(&s->e, false) < 0) {
 315         goto out_free_state;
 316     }
 317
 318     if (io_setup(MAX_EVENTS, &s->ctx) != 0) {
 319         goto out_close_efd;
 320     }
 321
 322     ioq_init(&s->io_q);
 323
 324     return s;
 325
 326 out_close_efd:
 327     event_notifier_cleanup(&s->e);
 328 out_free_state:
 329     g_free(s);
 330     return NULL;
 331 }
 332
 333 void laio_cleanup(void *s_)
 334 {
 335     struct qemu_laio_state *s = s_;
 336
 337     event_notifier_cleanup(&s->e);
 338
 339     if (io_destroy(s->ctx) != 0) {
 340         fprintf(stderr, "%s: destroy AIO context %p failed\n",
 341                         __func__, &s->ctx);
 342     }
 343     g_free(s);
 344 }