xref: /illumos-kvm-cmd/linux-aio.c (revision 68396ea9)
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     int async_context_id;
35     QLIST_ENTRY(qemu_laiocb) node;
36 };
37 
38 struct qemu_laio_state {
39     io_context_t ctx;
40     int efd;
41     int count;
42     QLIST_HEAD(, qemu_laiocb) completed_reqs;
43 };
44 
io_event_ret(struct io_event * ev)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  * Be sure to be in the right AsyncContext before calling this function.
53  */
qemu_laio_process_completion(struct qemu_laio_state * s,struct qemu_laiocb * laiocb)54 static void qemu_laio_process_completion(struct qemu_laio_state *s,
55     struct qemu_laiocb *laiocb)
56 {
57     int ret;
58 
59     s->count--;
60 
61     ret = laiocb->ret;
62     if (ret != -ECANCELED) {
63         if (ret == laiocb->nbytes)
64             ret = 0;
65         else if (ret >= 0)
66             ret = -EINVAL;
67 
68         laiocb->common.cb(laiocb->common.opaque, ret);
69     }
70 
71     qemu_aio_release(laiocb);
72 }
73 
74 /*
75  * Processes all queued AIO requests, i.e. requests that have return from OS
76  * but their callback was not called yet. Requests that cannot have their
77  * callback called in the current AsyncContext, remain in the queue.
78  *
79  * Returns 1 if at least one request could be completed, 0 otherwise.
80  */
qemu_laio_process_requests(void * opaque)81 static int qemu_laio_process_requests(void *opaque)
82 {
83     struct qemu_laio_state *s = opaque;
84     struct qemu_laiocb *laiocb, *next;
85     int res = 0;
86 
87     QLIST_FOREACH_SAFE (laiocb, &s->completed_reqs, node, next) {
88         if (laiocb->async_context_id == get_async_context_id()) {
89             qemu_laio_process_completion(s, laiocb);
90             QLIST_REMOVE(laiocb, node);
91             res = 1;
92         }
93     }
94 
95     return res;
96 }
97 
98 /*
99  * Puts a request in the completion queue so that its callback is called the
100  * next time when it's possible. If we already are in the right AsyncContext,
101  * the request is completed immediately instead.
102  */
qemu_laio_enqueue_completed(struct qemu_laio_state * s,struct qemu_laiocb * laiocb)103 static void qemu_laio_enqueue_completed(struct qemu_laio_state *s,
104     struct qemu_laiocb* laiocb)
105 {
106     if (laiocb->async_context_id == get_async_context_id()) {
107         qemu_laio_process_completion(s, laiocb);
108     } else {
109         QLIST_INSERT_HEAD(&s->completed_reqs, laiocb, node);
110     }
111 }
112 
qemu_laio_completion_cb(void * opaque)113 static void qemu_laio_completion_cb(void *opaque)
114 {
115     struct qemu_laio_state *s = opaque;
116 
117     while (1) {
118         struct io_event events[MAX_EVENTS];
119         uint64_t val;
120         ssize_t ret;
121         struct timespec ts = { 0 };
122         int nevents, i;
123 
124         do {
125             ret = read(s->efd, &val, sizeof(val));
126         } while (ret == -1 && errno == EINTR);
127 
128         if (ret == -1 && errno == EAGAIN)
129             break;
130 
131         if (ret != 8)
132             break;
133 
134         do {
135             nevents = io_getevents(s->ctx, val, MAX_EVENTS, events, &ts);
136         } while (nevents == -EINTR);
137 
138         for (i = 0; i < nevents; i++) {
139             struct iocb *iocb = events[i].obj;
140             struct qemu_laiocb *laiocb =
141                     container_of(iocb, struct qemu_laiocb, iocb);
142 
143             laiocb->ret = io_event_ret(&events[i]);
144             qemu_laio_enqueue_completed(s, laiocb);
145         }
146     }
147 }
148 
qemu_laio_flush_cb(void * opaque)149 static int qemu_laio_flush_cb(void *opaque)
150 {
151     struct qemu_laio_state *s = opaque;
152 
153     return (s->count > 0) ? 1 : 0;
154 }
155 
laio_cancel(BlockDriverAIOCB * blockacb)156 static void laio_cancel(BlockDriverAIOCB *blockacb)
157 {
158     struct qemu_laiocb *laiocb = (struct qemu_laiocb *)blockacb;
159     struct io_event event;
160     int ret;
161 
162     if (laiocb->ret != -EINPROGRESS)
163         return;
164 
165     /*
166      * Note that as of Linux 2.6.31 neither the block device code nor any
167      * filesystem implements cancellation of AIO request.
168      * Thus the polling loop below is the normal code path.
169      */
170     ret = io_cancel(laiocb->ctx->ctx, &laiocb->iocb, &event);
171     if (ret == 0) {
172         laiocb->ret = -ECANCELED;
173         return;
174     }
175 
176     /*
177      * We have to wait for the iocb to finish.
178      *
179      * The only way to get the iocb status update is by polling the io context.
180      * We might be able to do this slightly more optimal by removing the
181      * O_NONBLOCK flag.
182      */
183     while (laiocb->ret == -EINPROGRESS)
184         qemu_laio_completion_cb(laiocb->ctx);
185 }
186 
187 static AIOPool laio_pool = {
188     .aiocb_size         = sizeof(struct qemu_laiocb),
189     .cancel             = laio_cancel,
190 };
191 
laio_submit(BlockDriverState * bs,void * aio_ctx,int fd,int64_t sector_num,QEMUIOVector * qiov,int nb_sectors,BlockDriverCompletionFunc * cb,void * opaque,int type)192 BlockDriverAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
193         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
194         BlockDriverCompletionFunc *cb, void *opaque, int type)
195 {
196     struct qemu_laio_state *s = aio_ctx;
197     struct qemu_laiocb *laiocb;
198     struct iocb *iocbs;
199     off_t offset = sector_num * 512;
200 
201     laiocb = qemu_aio_get(&laio_pool, bs, cb, opaque);
202     if (!laiocb)
203         return NULL;
204     laiocb->nbytes = nb_sectors * 512;
205     laiocb->ctx = s;
206     laiocb->ret = -EINPROGRESS;
207     laiocb->async_context_id = get_async_context_id();
208 
209     iocbs = &laiocb->iocb;
210 
211     switch (type) {
212     case QEMU_AIO_WRITE:
213         io_prep_pwritev(iocbs, fd, qiov->iov, qiov->niov, offset);
214 	break;
215     case QEMU_AIO_READ:
216         io_prep_preadv(iocbs, fd, qiov->iov, qiov->niov, offset);
217 	break;
218     default:
219         fprintf(stderr, "%s: invalid AIO request type 0x%x.\n",
220                         __func__, type);
221         goto out_free_aiocb;
222     }
223     io_set_eventfd(&laiocb->iocb, s->efd);
224     s->count++;
225 
226     if (io_submit(s->ctx, 1, &iocbs) < 0)
227         goto out_dec_count;
228     return &laiocb->common;
229 
230 out_free_aiocb:
231     qemu_aio_release(laiocb);
232 out_dec_count:
233     s->count--;
234     return NULL;
235 }
236 
laio_init(void)237 void *laio_init(void)
238 {
239     struct qemu_laio_state *s;
240 
241     s = qemu_mallocz(sizeof(*s));
242     QLIST_INIT(&s->completed_reqs);
243     s->efd = eventfd(0, 0);
244     if (s->efd == -1)
245         goto out_free_state;
246     fcntl(s->efd, F_SETFL, O_NONBLOCK);
247 
248     if (io_setup(MAX_EVENTS, &s->ctx) != 0)
249         goto out_close_efd;
250 
251     qemu_aio_set_fd_handler(s->efd, qemu_laio_completion_cb, NULL,
252         qemu_laio_flush_cb, qemu_laio_process_requests, s);
253 
254     return s;
255 
256 out_close_efd:
257     close(s->efd);
258 out_free_state:
259     qemu_free(s);
260     return NULL;
261 }
262