2 * QEMU aio implementation
4 * Copyright IBM, Corp. 2008
7 * Anthony Liguori <aliguori@us.ibm.com>
9 * This work is licensed under the terms of the GNU GPL, version 2. See
10 * the COPYING file in the top-level directory.
12 * Contributions after 2012-01-13 are licensed under the terms of the
13 * GNU GPL, version 2 or (at your option) any later version.
16 #include "qemu/osdep.h"
17 #include "qemu-common.h"
18 #include "block/block.h"
19 #include "qemu/queue.h"
20 #include "qemu/sockets.h"
21 #ifdef CONFIG_EPOLL_CREATE1
22 #include <sys/epoll.h>
33 QLIST_ENTRY(AioHandler
) node
;
36 #ifdef CONFIG_EPOLL_CREATE1
38 /* The fd number threashold to switch to epoll */
39 #define EPOLL_ENABLE_THRESHOLD 64
41 static void aio_epoll_disable(AioContext
*ctx
)
43 ctx
->epoll_available
= false;
44 if (!ctx
->epoll_enabled
) {
47 ctx
->epoll_enabled
= false;
51 static inline int epoll_events_from_pfd(int pfd_events
)
53 return (pfd_events
& G_IO_IN
? EPOLLIN
: 0) |
54 (pfd_events
& G_IO_OUT
? EPOLLOUT
: 0) |
55 (pfd_events
& G_IO_HUP
? EPOLLHUP
: 0) |
56 (pfd_events
& G_IO_ERR
? EPOLLERR
: 0);
59 static bool aio_epoll_try_enable(AioContext
*ctx
)
62 struct epoll_event event
;
64 QLIST_FOREACH(node
, &ctx
->aio_handlers
, node
) {
66 if (node
->deleted
|| !node
->pfd
.events
) {
69 event
.events
= epoll_events_from_pfd(node
->pfd
.events
);
70 event
.data
.ptr
= node
;
71 r
= epoll_ctl(ctx
->epollfd
, EPOLL_CTL_ADD
, node
->pfd
.fd
, &event
);
76 ctx
->epoll_enabled
= true;
80 static void aio_epoll_update(AioContext
*ctx
, AioHandler
*node
, bool is_new
)
82 struct epoll_event event
;
85 if (!ctx
->epoll_enabled
) {
88 if (!node
->pfd
.events
) {
89 r
= epoll_ctl(ctx
->epollfd
, EPOLL_CTL_DEL
, node
->pfd
.fd
, &event
);
91 aio_epoll_disable(ctx
);
94 event
.data
.ptr
= node
;
95 event
.events
= epoll_events_from_pfd(node
->pfd
.events
);
97 r
= epoll_ctl(ctx
->epollfd
, EPOLL_CTL_ADD
, node
->pfd
.fd
, &event
);
99 aio_epoll_disable(ctx
);
102 r
= epoll_ctl(ctx
->epollfd
, EPOLL_CTL_MOD
, node
->pfd
.fd
, &event
);
104 aio_epoll_disable(ctx
);
110 static int aio_epoll(AioContext
*ctx
, GPollFD
*pfds
,
111 unsigned npfd
, int64_t timeout
)
115 struct epoll_event events
[128];
118 assert(pfds
[0].fd
== ctx
->epollfd
);
120 ret
= qemu_poll_ns(pfds
, npfd
, timeout
);
122 if (timeout
<= 0 || ret
> 0) {
123 ret
= epoll_wait(ctx
->epollfd
, events
,
124 sizeof(events
) / sizeof(events
[0]),
129 for (i
= 0; i
< ret
; i
++) {
130 int ev
= events
[i
].events
;
131 node
= events
[i
].data
.ptr
;
132 node
->pfd
.revents
= (ev
& EPOLLIN
? G_IO_IN
: 0) |
133 (ev
& EPOLLOUT
? G_IO_OUT
: 0) |
134 (ev
& EPOLLHUP
? G_IO_HUP
: 0) |
135 (ev
& EPOLLERR
? G_IO_ERR
: 0);
142 static bool aio_epoll_enabled(AioContext
*ctx
)
144 /* Fall back to ppoll when external clients are disabled. */
145 return !aio_external_disabled(ctx
) && ctx
->epoll_enabled
;
148 static bool aio_epoll_check_poll(AioContext
*ctx
, GPollFD
*pfds
,
149 unsigned npfd
, int64_t timeout
)
151 if (!ctx
->epoll_available
) {
154 if (aio_epoll_enabled(ctx
)) {
157 if (npfd
>= EPOLL_ENABLE_THRESHOLD
) {
158 if (aio_epoll_try_enable(ctx
)) {
161 aio_epoll_disable(ctx
);
169 static void aio_epoll_update(AioContext
*ctx
, AioHandler
*node
, bool is_new
)
173 static int aio_epoll(AioContext
*ctx
, GPollFD
*pfds
,
174 unsigned npfd
, int64_t timeout
)
179 static bool aio_epoll_enabled(AioContext
*ctx
)
184 static bool aio_epoll_check_poll(AioContext
*ctx
, GPollFD
*pfds
,
185 unsigned npfd
, int64_t timeout
)
192 static AioHandler
*find_aio_handler(AioContext
*ctx
, int fd
)
196 QLIST_FOREACH(node
, &ctx
->aio_handlers
, node
) {
197 if (node
->pfd
.fd
== fd
)
205 void aio_set_fd_handler(AioContext
*ctx
,
214 bool deleted
= false;
216 node
= find_aio_handler(ctx
, fd
);
218 /* Are we deleting the fd handler? */
219 if (!io_read
&& !io_write
) {
224 g_source_remove_poll(&ctx
->source
, &node
->pfd
);
226 /* If the lock is held, just mark the node as deleted */
227 if (ctx
->walking_handlers
) {
229 node
->pfd
.revents
= 0;
231 /* Otherwise, delete it for real. We can't just mark it as
232 * deleted because deleted nodes are only cleaned up after
233 * releasing the walking_handlers lock.
235 QLIST_REMOVE(node
, node
);
240 /* Alloc and insert if it's not already there */
241 node
= g_new0(AioHandler
, 1);
243 QLIST_INSERT_HEAD(&ctx
->aio_handlers
, node
, node
);
245 g_source_add_poll(&ctx
->source
, &node
->pfd
);
248 /* Update handler with latest information */
249 node
->io_read
= io_read
;
250 node
->io_write
= io_write
;
251 node
->opaque
= opaque
;
252 node
->is_external
= is_external
;
254 node
->pfd
.events
= (io_read
? G_IO_IN
| G_IO_HUP
| G_IO_ERR
: 0);
255 node
->pfd
.events
|= (io_write
? G_IO_OUT
| G_IO_ERR
: 0);
258 aio_epoll_update(ctx
, node
, is_new
);
265 void aio_set_event_notifier(AioContext
*ctx
,
266 EventNotifier
*notifier
,
268 EventNotifierHandler
*io_read
)
270 aio_set_fd_handler(ctx
, event_notifier_get_fd(notifier
),
271 is_external
, (IOHandler
*)io_read
, NULL
, notifier
);
274 bool aio_prepare(AioContext
*ctx
)
279 bool aio_pending(AioContext
*ctx
)
283 QLIST_FOREACH(node
, &ctx
->aio_handlers
, node
) {
286 revents
= node
->pfd
.revents
& node
->pfd
.events
;
287 if (revents
& (G_IO_IN
| G_IO_HUP
| G_IO_ERR
) && node
->io_read
&&
288 aio_node_check(ctx
, node
->is_external
)) {
291 if (revents
& (G_IO_OUT
| G_IO_ERR
) && node
->io_write
&&
292 aio_node_check(ctx
, node
->is_external
)) {
300 bool aio_dispatch(AioContext
*ctx
)
303 bool progress
= false;
306 * If there are callbacks left that have been queued, we need to call them.
307 * Do not call select in this case, because it is possible that the caller
308 * does not need a complete flush (as is the case for aio_poll loops).
310 if (aio_bh_poll(ctx
)) {
315 * We have to walk very carefully in case aio_set_fd_handler is
316 * called while we're walking.
318 node
= QLIST_FIRST(&ctx
->aio_handlers
);
323 ctx
->walking_handlers
++;
325 revents
= node
->pfd
.revents
& node
->pfd
.events
;
326 node
->pfd
.revents
= 0;
328 if (!node
->deleted
&&
329 (revents
& (G_IO_IN
| G_IO_HUP
| G_IO_ERR
)) &&
330 aio_node_check(ctx
, node
->is_external
) &&
332 node
->io_read(node
->opaque
);
334 /* aio_notify() does not count as progress */
335 if (node
->opaque
!= &ctx
->notifier
) {
339 if (!node
->deleted
&&
340 (revents
& (G_IO_OUT
| G_IO_ERR
)) &&
341 aio_node_check(ctx
, node
->is_external
) &&
343 node
->io_write(node
->opaque
);
348 node
= QLIST_NEXT(node
, node
);
350 ctx
->walking_handlers
--;
352 if (!ctx
->walking_handlers
&& tmp
->deleted
) {
353 QLIST_REMOVE(tmp
, node
);
359 progress
|= timerlistgroup_run_timers(&ctx
->tlg
);
364 /* These thread-local variables are used only in a small part of aio_poll
365 * around the call to the poll() system call. In particular they are not
366 * used while aio_poll is performing callbacks, which makes it much easier
367 * to think about reentrancy!
369 * Stack-allocated arrays would be perfect but they have size limitations;
370 * heap allocation is expensive enough that we want to reuse arrays across
371 * calls to aio_poll(). And because poll() has to be called without holding
372 * any lock, the arrays cannot be stored in AioContext. Thread-local data
373 * has none of the disadvantages of these three options.
375 static __thread GPollFD
*pollfds
;
376 static __thread AioHandler
**nodes
;
377 static __thread
unsigned npfd
, nalloc
;
378 static __thread Notifier pollfds_cleanup_notifier
;
380 static void pollfds_cleanup(Notifier
*n
, void *unused
)
388 static void add_pollfd(AioHandler
*node
)
390 if (npfd
== nalloc
) {
392 pollfds_cleanup_notifier
.notify
= pollfds_cleanup
;
393 qemu_thread_atexit_add(&pollfds_cleanup_notifier
);
396 g_assert(nalloc
<= INT_MAX
);
399 pollfds
= g_renew(GPollFD
, pollfds
, nalloc
);
400 nodes
= g_renew(AioHandler
*, nodes
, nalloc
);
403 pollfds
[npfd
] = (GPollFD
) {
405 .events
= node
->pfd
.events
,
410 bool aio_poll(AioContext
*ctx
, bool blocking
)
417 aio_context_acquire(ctx
);
420 /* aio_notify can avoid the expensive event_notifier_set if
421 * everything (file descriptors, bottom halves, timers) will
422 * be re-evaluated before the next blocking poll(). This is
423 * already true when aio_poll is called with blocking == false;
424 * if blocking == true, it is only true after poll() returns,
425 * so disable the optimization now.
428 atomic_add(&ctx
->notify_me
, 2);
431 ctx
->walking_handlers
++;
437 if (!aio_epoll_enabled(ctx
)) {
438 QLIST_FOREACH(node
, &ctx
->aio_handlers
, node
) {
439 if (!node
->deleted
&& node
->pfd
.events
440 && aio_node_check(ctx
, node
->is_external
)) {
446 timeout
= blocking
? aio_compute_timeout(ctx
) : 0;
448 /* wait until next event */
450 aio_context_release(ctx
);
452 if (aio_epoll_check_poll(ctx
, pollfds
, npfd
, timeout
)) {
453 AioHandler epoll_handler
;
455 epoll_handler
.pfd
.fd
= ctx
->epollfd
;
456 epoll_handler
.pfd
.events
= G_IO_IN
| G_IO_OUT
| G_IO_HUP
| G_IO_ERR
;
458 add_pollfd(&epoll_handler
);
459 ret
= aio_epoll(ctx
, pollfds
, npfd
, timeout
);
461 ret
= qemu_poll_ns(pollfds
, npfd
, timeout
);
464 atomic_sub(&ctx
->notify_me
, 2);
467 aio_context_acquire(ctx
);
470 aio_notify_accept(ctx
);
472 /* if we have any readable fds, dispatch event */
474 for (i
= 0; i
< npfd
; i
++) {
475 nodes
[i
]->pfd
.revents
= pollfds
[i
].revents
;
480 ctx
->walking_handlers
--;
482 /* Run dispatch even if there were no readable fds to run timers */
483 if (aio_dispatch(ctx
)) {
487 aio_context_release(ctx
);
492 void aio_context_setup(AioContext
*ctx
)
494 #ifdef CONFIG_EPOLL_CREATE1
495 assert(!ctx
->epollfd
);
496 ctx
->epollfd
= epoll_create1(EPOLL_CLOEXEC
);
497 if (ctx
->epollfd
== -1) {
498 fprintf(stderr
, "Failed to create epoll instance: %s", strerror(errno
));
499 ctx
->epoll_available
= false;
501 ctx
->epoll_available
= true;