Merge branch 'for-upstream' of git://repo.or.cz/qemu/agraf
[qemu.git] / 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 laiocb->nbytes = nb_sectors * 512;
170 laiocb->ctx = s;
171 laiocb->ret = -EINPROGRESS;
172 laiocb->is_read = (type == QEMU_AIO_READ);
173 laiocb->qiov = qiov;
174
175 iocbs = &laiocb->iocb;
176
177 switch (type) {
178 case QEMU_AIO_WRITE:
179 io_prep_pwritev(iocbs, fd, qiov->iov, qiov->niov, offset);
180 break;
181 case QEMU_AIO_READ:
182 io_prep_preadv(iocbs, fd, qiov->iov, qiov->niov, offset);
183 break;
184 /* Currently Linux kernel does not support other operations */
185 default:
186 fprintf(stderr, "%s: invalid AIO request type 0x%x.\n",
187 __func__, type);
188 goto out_free_aiocb;
189 }
190 io_set_eventfd(&laiocb->iocb, s->efd);
191 s->count++;
192
193 if (io_submit(s->ctx, 1, &iocbs) < 0)
194 goto out_dec_count;
195 return &laiocb->common;
196
197 out_dec_count:
198 s->count--;
199 out_free_aiocb:
200 qemu_aio_release(laiocb);
201 return NULL;
202 }
203
204 void *laio_init(void)
205 {
206 struct qemu_laio_state *s;
207
208 s = g_malloc0(sizeof(*s));
209 s->efd = eventfd(0, 0);
210 if (s->efd == -1)
211 goto out_free_state;
212 fcntl(s->efd, F_SETFL, O_NONBLOCK);
213
214 if (io_setup(MAX_EVENTS, &s->ctx) != 0)
215 goto out_close_efd;
216
217 qemu_aio_set_fd_handler(s->efd, qemu_laio_completion_cb, NULL,
218 qemu_laio_flush_cb, NULL, s);
219
220 return s;
221
222 out_close_efd:
223 close(s->efd);
224 out_free_state:
225 g_free(s);
226 return NULL;
227 }