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_init(void) 181 { 182 struct qemu_laio_state *s; 183 184 s = g_malloc0(sizeof(*s)); 185 if (event_notifier_init(&s->e, false) < 0) { 186 goto out_free_state; 187 } 188 189 if (io_setup(MAX_EVENTS, &s->ctx) != 0) { 190 goto out_close_efd; 191 } 192 193 qemu_aio_set_event_notifier(&s->e, qemu_laio_completion_cb); 194 195 return s; 196 197 out_close_efd: 198 event_notifier_cleanup(&s->e); 199 out_free_state: 200 g_free(s); 201 return NULL; 202 } 203