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/rcu_queue.h"
20 #include "qemu/sockets.h"
21 #include "qemu/cutils.h"
23 #ifdef CONFIG_EPOLL_CREATE1
24 #include <sys/epoll.h>
33 IOHandler
*io_poll_begin
;
34 IOHandler
*io_poll_end
;
38 QLIST_ENTRY(AioHandler
) node
;
41 #ifdef CONFIG_EPOLL_CREATE1
43 /* The fd number threashold to switch to epoll */
44 #define EPOLL_ENABLE_THRESHOLD 64
46 static void aio_epoll_disable(AioContext
*ctx
)
48 ctx
->epoll_enabled
= false;
49 if (!ctx
->epoll_available
) {
52 ctx
->epoll_available
= false;
56 static inline int epoll_events_from_pfd(int pfd_events
)
58 return (pfd_events
& G_IO_IN
? EPOLLIN
: 0) |
59 (pfd_events
& G_IO_OUT
? EPOLLOUT
: 0) |
60 (pfd_events
& G_IO_HUP
? EPOLLHUP
: 0) |
61 (pfd_events
& G_IO_ERR
? EPOLLERR
: 0);
64 static bool aio_epoll_try_enable(AioContext
*ctx
)
67 struct epoll_event event
;
69 QLIST_FOREACH_RCU(node
, &ctx
->aio_handlers
, node
) {
71 if (node
->deleted
|| !node
->pfd
.events
) {
74 event
.events
= epoll_events_from_pfd(node
->pfd
.events
);
75 event
.data
.ptr
= node
;
76 r
= epoll_ctl(ctx
->epollfd
, EPOLL_CTL_ADD
, node
->pfd
.fd
, &event
);
81 ctx
->epoll_enabled
= true;
85 static void aio_epoll_update(AioContext
*ctx
, AioHandler
*node
, bool is_new
)
87 struct epoll_event event
;
91 if (!ctx
->epoll_enabled
) {
94 if (!node
->pfd
.events
) {
97 event
.data
.ptr
= node
;
98 event
.events
= epoll_events_from_pfd(node
->pfd
.events
);
99 ctl
= is_new
? EPOLL_CTL_ADD
: EPOLL_CTL_MOD
;
102 r
= epoll_ctl(ctx
->epollfd
, ctl
, node
->pfd
.fd
, &event
);
104 aio_epoll_disable(ctx
);
108 static int aio_epoll(AioContext
*ctx
, GPollFD
*pfds
,
109 unsigned npfd
, int64_t timeout
)
113 struct epoll_event events
[128];
116 assert(pfds
[0].fd
== ctx
->epollfd
);
118 ret
= qemu_poll_ns(pfds
, npfd
, timeout
);
120 if (timeout
<= 0 || ret
> 0) {
121 ret
= epoll_wait(ctx
->epollfd
, events
,
127 for (i
= 0; i
< ret
; i
++) {
128 int ev
= events
[i
].events
;
129 node
= events
[i
].data
.ptr
;
130 node
->pfd
.revents
= (ev
& EPOLLIN
? G_IO_IN
: 0) |
131 (ev
& EPOLLOUT
? G_IO_OUT
: 0) |
132 (ev
& EPOLLHUP
? G_IO_HUP
: 0) |
133 (ev
& EPOLLERR
? G_IO_ERR
: 0);
140 static bool aio_epoll_enabled(AioContext
*ctx
)
142 /* Fall back to ppoll when external clients are disabled. */
143 return !aio_external_disabled(ctx
) && ctx
->epoll_enabled
;
146 static bool aio_epoll_check_poll(AioContext
*ctx
, GPollFD
*pfds
,
147 unsigned npfd
, int64_t timeout
)
149 if (!ctx
->epoll_available
) {
152 if (aio_epoll_enabled(ctx
)) {
155 if (npfd
>= EPOLL_ENABLE_THRESHOLD
) {
156 if (aio_epoll_try_enable(ctx
)) {
159 aio_epoll_disable(ctx
);
167 static void aio_epoll_update(AioContext
*ctx
, AioHandler
*node
, bool is_new
)
171 static int aio_epoll(AioContext
*ctx
, GPollFD
*pfds
,
172 unsigned npfd
, int64_t timeout
)
177 static bool aio_epoll_enabled(AioContext
*ctx
)
182 static bool aio_epoll_check_poll(AioContext
*ctx
, GPollFD
*pfds
,
183 unsigned npfd
, int64_t timeout
)
190 static AioHandler
*find_aio_handler(AioContext
*ctx
, int fd
)
194 QLIST_FOREACH(node
, &ctx
->aio_handlers
, node
) {
195 if (node
->pfd
.fd
== fd
)
203 void aio_set_fd_handler(AioContext
*ctx
,
213 bool deleted
= false;
214 int poll_disable_change
;
216 qemu_lockcnt_lock(&ctx
->list_lock
);
218 node
= find_aio_handler(ctx
, fd
);
220 /* Are we deleting the fd handler? */
221 if (!io_read
&& !io_write
&& !io_poll
) {
223 qemu_lockcnt_unlock(&ctx
->list_lock
);
227 /* If the GSource is in the process of being destroyed then
228 * g_source_remove_poll() causes an assertion failure. Skip
229 * removal in that case, because glib cleans up its state during
230 * destruction anyway.
232 if (!g_source_is_destroyed(&ctx
->source
)) {
233 g_source_remove_poll(&ctx
->source
, &node
->pfd
);
236 /* If a read is in progress, just mark the node as deleted */
237 if (qemu_lockcnt_count(&ctx
->list_lock
)) {
239 node
->pfd
.revents
= 0;
241 /* Otherwise, delete it for real. We can't just mark it as
242 * deleted because deleted nodes are only cleaned up while
243 * no one is walking the handlers list.
245 QLIST_REMOVE(node
, node
);
248 poll_disable_change
= -!node
->io_poll
;
250 poll_disable_change
= !io_poll
- (node
&& !node
->io_poll
);
252 /* Alloc and insert if it's not already there */
253 node
= g_new0(AioHandler
, 1);
255 QLIST_INSERT_HEAD_RCU(&ctx
->aio_handlers
, node
, node
);
257 g_source_add_poll(&ctx
->source
, &node
->pfd
);
261 /* Update handler with latest information */
262 node
->io_read
= io_read
;
263 node
->io_write
= io_write
;
264 node
->io_poll
= io_poll
;
265 node
->opaque
= opaque
;
266 node
->is_external
= is_external
;
268 node
->pfd
.events
= (io_read
? G_IO_IN
| G_IO_HUP
| G_IO_ERR
: 0);
269 node
->pfd
.events
|= (io_write
? G_IO_OUT
| G_IO_ERR
: 0);
272 /* No need to order poll_disable_cnt writes against other updates;
273 * the counter is only used to avoid wasting time and latency on
274 * iterated polling when the system call will be ultimately necessary.
275 * Changing handlers is a rare event, and a little wasted polling until
276 * the aio_notify below is not an issue.
278 atomic_set(&ctx
->poll_disable_cnt
,
279 atomic_read(&ctx
->poll_disable_cnt
) + poll_disable_change
);
281 aio_epoll_update(ctx
, node
, is_new
);
282 qemu_lockcnt_unlock(&ctx
->list_lock
);
290 void aio_set_fd_poll(AioContext
*ctx
, int fd
,
291 IOHandler
*io_poll_begin
,
292 IOHandler
*io_poll_end
)
294 AioHandler
*node
= find_aio_handler(ctx
, fd
);
300 node
->io_poll_begin
= io_poll_begin
;
301 node
->io_poll_end
= io_poll_end
;
304 void aio_set_event_notifier(AioContext
*ctx
,
305 EventNotifier
*notifier
,
307 EventNotifierHandler
*io_read
,
310 aio_set_fd_handler(ctx
, event_notifier_get_fd(notifier
), is_external
,
311 (IOHandler
*)io_read
, NULL
, io_poll
, notifier
);
314 void aio_set_event_notifier_poll(AioContext
*ctx
,
315 EventNotifier
*notifier
,
316 EventNotifierHandler
*io_poll_begin
,
317 EventNotifierHandler
*io_poll_end
)
319 aio_set_fd_poll(ctx
, event_notifier_get_fd(notifier
),
320 (IOHandler
*)io_poll_begin
,
321 (IOHandler
*)io_poll_end
);
324 static void poll_set_started(AioContext
*ctx
, bool started
)
328 if (started
== ctx
->poll_started
) {
332 ctx
->poll_started
= started
;
334 qemu_lockcnt_inc(&ctx
->list_lock
);
335 QLIST_FOREACH_RCU(node
, &ctx
->aio_handlers
, node
) {
343 fn
= node
->io_poll_begin
;
345 fn
= node
->io_poll_end
;
352 qemu_lockcnt_dec(&ctx
->list_lock
);
356 bool aio_prepare(AioContext
*ctx
)
358 /* Poll mode cannot be used with glib's event loop, disable it. */
359 poll_set_started(ctx
, false);
364 bool aio_pending(AioContext
*ctx
)
370 * We have to walk very carefully in case aio_set_fd_handler is
371 * called while we're walking.
373 qemu_lockcnt_inc(&ctx
->list_lock
);
375 QLIST_FOREACH_RCU(node
, &ctx
->aio_handlers
, node
) {
378 revents
= node
->pfd
.revents
& node
->pfd
.events
;
379 if (revents
& (G_IO_IN
| G_IO_HUP
| G_IO_ERR
) && node
->io_read
&&
380 aio_node_check(ctx
, node
->is_external
)) {
384 if (revents
& (G_IO_OUT
| G_IO_ERR
) && node
->io_write
&&
385 aio_node_check(ctx
, node
->is_external
)) {
390 qemu_lockcnt_dec(&ctx
->list_lock
);
395 static bool aio_dispatch_handlers(AioContext
*ctx
)
397 AioHandler
*node
, *tmp
;
398 bool progress
= false;
400 QLIST_FOREACH_SAFE_RCU(node
, &ctx
->aio_handlers
, node
, tmp
) {
403 revents
= node
->pfd
.revents
& node
->pfd
.events
;
404 node
->pfd
.revents
= 0;
406 if (!node
->deleted
&&
407 (revents
& (G_IO_IN
| G_IO_HUP
| G_IO_ERR
)) &&
408 aio_node_check(ctx
, node
->is_external
) &&
410 node
->io_read(node
->opaque
);
412 /* aio_notify() does not count as progress */
413 if (node
->opaque
!= &ctx
->notifier
) {
417 if (!node
->deleted
&&
418 (revents
& (G_IO_OUT
| G_IO_ERR
)) &&
419 aio_node_check(ctx
, node
->is_external
) &&
421 node
->io_write(node
->opaque
);
426 if (qemu_lockcnt_dec_if_lock(&ctx
->list_lock
)) {
427 QLIST_REMOVE(node
, node
);
429 qemu_lockcnt_inc_and_unlock(&ctx
->list_lock
);
437 void aio_dispatch(AioContext
*ctx
)
439 qemu_lockcnt_inc(&ctx
->list_lock
);
441 aio_dispatch_handlers(ctx
);
442 qemu_lockcnt_dec(&ctx
->list_lock
);
444 timerlistgroup_run_timers(&ctx
->tlg
);
447 /* These thread-local variables are used only in a small part of aio_poll
448 * around the call to the poll() system call. In particular they are not
449 * used while aio_poll is performing callbacks, which makes it much easier
450 * to think about reentrancy!
452 * Stack-allocated arrays would be perfect but they have size limitations;
453 * heap allocation is expensive enough that we want to reuse arrays across
454 * calls to aio_poll(). And because poll() has to be called without holding
455 * any lock, the arrays cannot be stored in AioContext. Thread-local data
456 * has none of the disadvantages of these three options.
458 static __thread GPollFD
*pollfds
;
459 static __thread AioHandler
**nodes
;
460 static __thread
unsigned npfd
, nalloc
;
461 static __thread Notifier pollfds_cleanup_notifier
;
463 static void pollfds_cleanup(Notifier
*n
, void *unused
)
471 static void add_pollfd(AioHandler
*node
)
473 if (npfd
== nalloc
) {
475 pollfds_cleanup_notifier
.notify
= pollfds_cleanup
;
476 qemu_thread_atexit_add(&pollfds_cleanup_notifier
);
479 g_assert(nalloc
<= INT_MAX
);
482 pollfds
= g_renew(GPollFD
, pollfds
, nalloc
);
483 nodes
= g_renew(AioHandler
*, nodes
, nalloc
);
486 pollfds
[npfd
] = (GPollFD
) {
488 .events
= node
->pfd
.events
,
493 static bool run_poll_handlers_once(AioContext
*ctx
, int64_t *timeout
)
495 bool progress
= false;
498 QLIST_FOREACH_RCU(node
, &ctx
->aio_handlers
, node
) {
499 if (!node
->deleted
&& node
->io_poll
&&
500 aio_node_check(ctx
, node
->is_external
) &&
501 node
->io_poll(node
->opaque
) &&
502 node
->opaque
!= &ctx
->notifier
) {
507 /* Caller handles freeing deleted nodes. Don't do it here. */
513 /* run_poll_handlers:
514 * @ctx: the AioContext
515 * @max_ns: maximum time to poll for, in nanoseconds
517 * Polls for a given time.
519 * Note that ctx->notify_me must be non-zero so this function can detect
522 * Note that the caller must have incremented ctx->list_lock.
524 * Returns: true if progress was made, false otherwise
526 static bool run_poll_handlers(AioContext
*ctx
, int64_t max_ns
, int64_t *timeout
)
529 int64_t start_time
, elapsed_time
;
531 assert(ctx
->notify_me
);
532 assert(qemu_lockcnt_count(&ctx
->list_lock
) > 0);
534 trace_run_poll_handlers_begin(ctx
, max_ns
, *timeout
);
536 start_time
= qemu_clock_get_ns(QEMU_CLOCK_REALTIME
);
538 progress
= run_poll_handlers_once(ctx
, timeout
);
539 elapsed_time
= qemu_clock_get_ns(QEMU_CLOCK_REALTIME
) - start_time
;
540 } while (!progress
&& elapsed_time
< max_ns
541 && !atomic_read(&ctx
->poll_disable_cnt
));
543 /* If time has passed with no successful polling, adjust *timeout to
544 * keep the same ending time.
546 if (*timeout
!= -1) {
547 *timeout
-= MIN(*timeout
, elapsed_time
);
550 trace_run_poll_handlers_end(ctx
, progress
, *timeout
);
555 * @ctx: the AioContext
556 * @timeout: timeout for blocking wait, computed by the caller and updated if
559 * ctx->notify_me must be non-zero so this function can detect aio_notify().
561 * Note that the caller must have incremented ctx->list_lock.
563 * Returns: true if progress was made, false otherwise
565 static bool try_poll_mode(AioContext
*ctx
, int64_t *timeout
)
567 /* See qemu_soonest_timeout() uint64_t hack */
568 int64_t max_ns
= MIN((uint64_t)*timeout
, (uint64_t)ctx
->poll_ns
);
570 if (max_ns
&& !atomic_read(&ctx
->poll_disable_cnt
)) {
571 poll_set_started(ctx
, true);
573 if (run_poll_handlers(ctx
, max_ns
, timeout
)) {
578 poll_set_started(ctx
, false);
580 /* Even if we don't run busy polling, try polling once in case it can make
581 * progress and the caller will be able to avoid ppoll(2)/epoll_wait(2).
583 return run_poll_handlers_once(ctx
, timeout
);
586 bool aio_poll(AioContext
*ctx
, bool blocking
)
595 /* aio_notify can avoid the expensive event_notifier_set if
596 * everything (file descriptors, bottom halves, timers) will
597 * be re-evaluated before the next blocking poll(). This is
598 * already true when aio_poll is called with blocking == false;
599 * if blocking == true, it is only true after poll() returns,
600 * so disable the optimization now.
603 assert(in_aio_context_home_thread(ctx
));
604 atomic_add(&ctx
->notify_me
, 2);
607 qemu_lockcnt_inc(&ctx
->list_lock
);
609 if (ctx
->poll_max_ns
) {
610 start
= qemu_clock_get_ns(QEMU_CLOCK_REALTIME
);
613 timeout
= blocking
? aio_compute_timeout(ctx
) : 0;
614 progress
= try_poll_mode(ctx
, &timeout
);
615 assert(!(timeout
&& progress
));
617 /* If polling is allowed, non-blocking aio_poll does not need the
618 * system call---a single round of run_poll_handlers_once suffices.
620 if (timeout
|| atomic_read(&ctx
->poll_disable_cnt
)) {
625 if (!aio_epoll_enabled(ctx
)) {
626 QLIST_FOREACH_RCU(node
, &ctx
->aio_handlers
, node
) {
627 if (!node
->deleted
&& node
->pfd
.events
628 && aio_node_check(ctx
, node
->is_external
)) {
634 /* wait until next event */
635 if (aio_epoll_check_poll(ctx
, pollfds
, npfd
, timeout
)) {
636 AioHandler epoll_handler
;
638 epoll_handler
.pfd
.fd
= ctx
->epollfd
;
639 epoll_handler
.pfd
.events
= G_IO_IN
| G_IO_OUT
| G_IO_HUP
| G_IO_ERR
;
641 add_pollfd(&epoll_handler
);
642 ret
= aio_epoll(ctx
, pollfds
, npfd
, timeout
);
644 ret
= qemu_poll_ns(pollfds
, npfd
, timeout
);
649 atomic_sub(&ctx
->notify_me
, 2);
650 aio_notify_accept(ctx
);
653 /* Adjust polling time */
654 if (ctx
->poll_max_ns
) {
655 int64_t block_ns
= qemu_clock_get_ns(QEMU_CLOCK_REALTIME
) - start
;
657 if (block_ns
<= ctx
->poll_ns
) {
658 /* This is the sweet spot, no adjustment needed */
659 } else if (block_ns
> ctx
->poll_max_ns
) {
660 /* We'd have to poll for too long, poll less */
661 int64_t old
= ctx
->poll_ns
;
663 if (ctx
->poll_shrink
) {
664 ctx
->poll_ns
/= ctx
->poll_shrink
;
669 trace_poll_shrink(ctx
, old
, ctx
->poll_ns
);
670 } else if (ctx
->poll_ns
< ctx
->poll_max_ns
&&
671 block_ns
< ctx
->poll_max_ns
) {
672 /* There is room to grow, poll longer */
673 int64_t old
= ctx
->poll_ns
;
674 int64_t grow
= ctx
->poll_grow
;
681 ctx
->poll_ns
*= grow
;
683 ctx
->poll_ns
= 4000; /* start polling at 4 microseconds */
686 if (ctx
->poll_ns
> ctx
->poll_max_ns
) {
687 ctx
->poll_ns
= ctx
->poll_max_ns
;
690 trace_poll_grow(ctx
, old
, ctx
->poll_ns
);
694 /* if we have any readable fds, dispatch event */
696 for (i
= 0; i
< npfd
; i
++) {
697 nodes
[i
]->pfd
.revents
= pollfds
[i
].revents
;
703 progress
|= aio_bh_poll(ctx
);
706 progress
|= aio_dispatch_handlers(ctx
);
709 qemu_lockcnt_dec(&ctx
->list_lock
);
711 progress
|= timerlistgroup_run_timers(&ctx
->tlg
);
716 void aio_context_setup(AioContext
*ctx
)
718 #ifdef CONFIG_EPOLL_CREATE1
719 assert(!ctx
->epollfd
);
720 ctx
->epollfd
= epoll_create1(EPOLL_CLOEXEC
);
721 if (ctx
->epollfd
== -1) {
722 fprintf(stderr
, "Failed to create epoll instance: %s", strerror(errno
));
723 ctx
->epoll_available
= false;
725 ctx
->epoll_available
= true;
730 void aio_context_destroy(AioContext
*ctx
)
732 #ifdef CONFIG_EPOLL_CREATE1
733 aio_epoll_disable(ctx
);
737 void aio_context_set_poll_params(AioContext
*ctx
, int64_t max_ns
,
738 int64_t grow
, int64_t shrink
, Error
**errp
)
740 /* No thread synchronization here, it doesn't matter if an incorrect value
743 ctx
->poll_max_ns
= max_ns
;
745 ctx
->poll_grow
= grow
;
746 ctx
->poll_shrink
= shrink
;