quorum: Inline quorum_aio_cb()
[qemu.git] / aio-posix.c
1 /*
2 * QEMU aio implementation
3 *
4 * Copyright IBM, Corp. 2008
5 *
6 * Authors:
7 * Anthony Liguori <aliguori@us.ibm.com>
8 *
9 * This work is licensed under the terms of the GNU GPL, version 2. See
10 * the COPYING file in the top-level directory.
11 *
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.
14 */
15
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 #include "qemu/cutils.h"
22 #include "trace.h"
23 #ifdef CONFIG_EPOLL_CREATE1
24 #include <sys/epoll.h>
25 #endif
26
27 struct AioHandler
28 {
29 GPollFD pfd;
30 IOHandler *io_read;
31 IOHandler *io_write;
32 AioPollFn *io_poll;
33 IOHandler *io_poll_begin;
34 IOHandler *io_poll_end;
35 int deleted;
36 void *opaque;
37 bool is_external;
38 QLIST_ENTRY(AioHandler) node;
39 };
40
41 #ifdef CONFIG_EPOLL_CREATE1
42
43 /* The fd number threashold to switch to epoll */
44 #define EPOLL_ENABLE_THRESHOLD 64
45
46 static void aio_epoll_disable(AioContext *ctx)
47 {
48 ctx->epoll_available = false;
49 if (!ctx->epoll_enabled) {
50 return;
51 }
52 ctx->epoll_enabled = false;
53 close(ctx->epollfd);
54 }
55
56 static inline int epoll_events_from_pfd(int pfd_events)
57 {
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);
62 }
63
64 static bool aio_epoll_try_enable(AioContext *ctx)
65 {
66 AioHandler *node;
67 struct epoll_event event;
68
69 QLIST_FOREACH(node, &ctx->aio_handlers, node) {
70 int r;
71 if (node->deleted || !node->pfd.events) {
72 continue;
73 }
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);
77 if (r) {
78 return false;
79 }
80 }
81 ctx->epoll_enabled = true;
82 return true;
83 }
84
85 static void aio_epoll_update(AioContext *ctx, AioHandler *node, bool is_new)
86 {
87 struct epoll_event event;
88 int r;
89 int ctl;
90
91 if (!ctx->epoll_enabled) {
92 return;
93 }
94 if (!node->pfd.events) {
95 ctl = EPOLL_CTL_DEL;
96 } else {
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;
100 }
101
102 r = epoll_ctl(ctx->epollfd, ctl, node->pfd.fd, &event);
103 if (r) {
104 aio_epoll_disable(ctx);
105 }
106 }
107
108 static int aio_epoll(AioContext *ctx, GPollFD *pfds,
109 unsigned npfd, int64_t timeout)
110 {
111 AioHandler *node;
112 int i, ret = 0;
113 struct epoll_event events[128];
114
115 assert(npfd == 1);
116 assert(pfds[0].fd == ctx->epollfd);
117 if (timeout > 0) {
118 ret = qemu_poll_ns(pfds, npfd, timeout);
119 }
120 if (timeout <= 0 || ret > 0) {
121 ret = epoll_wait(ctx->epollfd, events,
122 sizeof(events) / sizeof(events[0]),
123 timeout);
124 if (ret <= 0) {
125 goto out;
126 }
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);
134 }
135 }
136 out:
137 return ret;
138 }
139
140 static bool aio_epoll_enabled(AioContext *ctx)
141 {
142 /* Fall back to ppoll when external clients are disabled. */
143 return !aio_external_disabled(ctx) && ctx->epoll_enabled;
144 }
145
146 static bool aio_epoll_check_poll(AioContext *ctx, GPollFD *pfds,
147 unsigned npfd, int64_t timeout)
148 {
149 if (!ctx->epoll_available) {
150 return false;
151 }
152 if (aio_epoll_enabled(ctx)) {
153 return true;
154 }
155 if (npfd >= EPOLL_ENABLE_THRESHOLD) {
156 if (aio_epoll_try_enable(ctx)) {
157 return true;
158 } else {
159 aio_epoll_disable(ctx);
160 }
161 }
162 return false;
163 }
164
165 #else
166
167 static void aio_epoll_update(AioContext *ctx, AioHandler *node, bool is_new)
168 {
169 }
170
171 static int aio_epoll(AioContext *ctx, GPollFD *pfds,
172 unsigned npfd, int64_t timeout)
173 {
174 assert(false);
175 }
176
177 static bool aio_epoll_enabled(AioContext *ctx)
178 {
179 return false;
180 }
181
182 static bool aio_epoll_check_poll(AioContext *ctx, GPollFD *pfds,
183 unsigned npfd, int64_t timeout)
184 {
185 return false;
186 }
187
188 #endif
189
190 static AioHandler *find_aio_handler(AioContext *ctx, int fd)
191 {
192 AioHandler *node;
193
194 QLIST_FOREACH(node, &ctx->aio_handlers, node) {
195 if (node->pfd.fd == fd)
196 if (!node->deleted)
197 return node;
198 }
199
200 return NULL;
201 }
202
203 void aio_set_fd_handler(AioContext *ctx,
204 int fd,
205 bool is_external,
206 IOHandler *io_read,
207 IOHandler *io_write,
208 AioPollFn *io_poll,
209 void *opaque)
210 {
211 AioHandler *node;
212 bool is_new = false;
213 bool deleted = false;
214
215 node = find_aio_handler(ctx, fd);
216
217 /* Are we deleting the fd handler? */
218 if (!io_read && !io_write && !io_poll) {
219 if (node == NULL) {
220 return;
221 }
222
223 g_source_remove_poll(&ctx->source, &node->pfd);
224
225 /* If the lock is held, just mark the node as deleted */
226 if (ctx->walking_handlers) {
227 node->deleted = 1;
228 node->pfd.revents = 0;
229 } else {
230 /* Otherwise, delete it for real. We can't just mark it as
231 * deleted because deleted nodes are only cleaned up after
232 * releasing the walking_handlers lock.
233 */
234 QLIST_REMOVE(node, node);
235 deleted = true;
236 }
237
238 if (!node->io_poll) {
239 ctx->poll_disable_cnt--;
240 }
241 } else {
242 if (node == NULL) {
243 /* Alloc and insert if it's not already there */
244 node = g_new0(AioHandler, 1);
245 node->pfd.fd = fd;
246 QLIST_INSERT_HEAD(&ctx->aio_handlers, node, node);
247
248 g_source_add_poll(&ctx->source, &node->pfd);
249 is_new = true;
250
251 ctx->poll_disable_cnt += !io_poll;
252 } else {
253 ctx->poll_disable_cnt += !io_poll - !node->io_poll;
254 }
255
256 /* Update handler with latest information */
257 node->io_read = io_read;
258 node->io_write = io_write;
259 node->io_poll = io_poll;
260 node->opaque = opaque;
261 node->is_external = is_external;
262
263 node->pfd.events = (io_read ? G_IO_IN | G_IO_HUP | G_IO_ERR : 0);
264 node->pfd.events |= (io_write ? G_IO_OUT | G_IO_ERR : 0);
265 }
266
267 aio_epoll_update(ctx, node, is_new);
268 aio_notify(ctx);
269
270 if (deleted) {
271 g_free(node);
272 }
273 }
274
275 void aio_set_fd_poll(AioContext *ctx, int fd,
276 IOHandler *io_poll_begin,
277 IOHandler *io_poll_end)
278 {
279 AioHandler *node = find_aio_handler(ctx, fd);
280
281 if (!node) {
282 return;
283 }
284
285 node->io_poll_begin = io_poll_begin;
286 node->io_poll_end = io_poll_end;
287 }
288
289 void aio_set_event_notifier(AioContext *ctx,
290 EventNotifier *notifier,
291 bool is_external,
292 EventNotifierHandler *io_read,
293 AioPollFn *io_poll)
294 {
295 aio_set_fd_handler(ctx, event_notifier_get_fd(notifier), is_external,
296 (IOHandler *)io_read, NULL, io_poll, notifier);
297 }
298
299 void aio_set_event_notifier_poll(AioContext *ctx,
300 EventNotifier *notifier,
301 EventNotifierHandler *io_poll_begin,
302 EventNotifierHandler *io_poll_end)
303 {
304 aio_set_fd_poll(ctx, event_notifier_get_fd(notifier),
305 (IOHandler *)io_poll_begin,
306 (IOHandler *)io_poll_end);
307 }
308
309 static void poll_set_started(AioContext *ctx, bool started)
310 {
311 AioHandler *node;
312
313 if (started == ctx->poll_started) {
314 return;
315 }
316
317 ctx->poll_started = started;
318
319 ctx->walking_handlers++;
320 QLIST_FOREACH(node, &ctx->aio_handlers, node) {
321 IOHandler *fn;
322
323 if (node->deleted) {
324 continue;
325 }
326
327 if (started) {
328 fn = node->io_poll_begin;
329 } else {
330 fn = node->io_poll_end;
331 }
332
333 if (fn) {
334 fn(node->opaque);
335 }
336 }
337 ctx->walking_handlers--;
338 }
339
340
341 bool aio_prepare(AioContext *ctx)
342 {
343 /* Poll mode cannot be used with glib's event loop, disable it. */
344 poll_set_started(ctx, false);
345
346 return false;
347 }
348
349 bool aio_pending(AioContext *ctx)
350 {
351 AioHandler *node;
352
353 QLIST_FOREACH(node, &ctx->aio_handlers, node) {
354 int revents;
355
356 revents = node->pfd.revents & node->pfd.events;
357 if (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR) && node->io_read &&
358 aio_node_check(ctx, node->is_external)) {
359 return true;
360 }
361 if (revents & (G_IO_OUT | G_IO_ERR) && node->io_write &&
362 aio_node_check(ctx, node->is_external)) {
363 return true;
364 }
365 }
366
367 return false;
368 }
369
370 /*
371 * Note that dispatch_fds == false has the side-effect of post-poning the
372 * freeing of deleted handlers.
373 */
374 bool aio_dispatch(AioContext *ctx, bool dispatch_fds)
375 {
376 AioHandler *node = NULL;
377 bool progress = false;
378
379 /*
380 * If there are callbacks left that have been queued, we need to call them.
381 * Do not call select in this case, because it is possible that the caller
382 * does not need a complete flush (as is the case for aio_poll loops).
383 */
384 if (aio_bh_poll(ctx)) {
385 progress = true;
386 }
387
388 /*
389 * We have to walk very carefully in case aio_set_fd_handler is
390 * called while we're walking.
391 */
392 if (dispatch_fds) {
393 node = QLIST_FIRST(&ctx->aio_handlers);
394 }
395 while (node) {
396 AioHandler *tmp;
397 int revents;
398
399 ctx->walking_handlers++;
400
401 revents = node->pfd.revents & node->pfd.events;
402 node->pfd.revents = 0;
403
404 if (!node->deleted &&
405 (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR)) &&
406 aio_node_check(ctx, node->is_external) &&
407 node->io_read) {
408 node->io_read(node->opaque);
409
410 /* aio_notify() does not count as progress */
411 if (node->opaque != &ctx->notifier) {
412 progress = true;
413 }
414 }
415 if (!node->deleted &&
416 (revents & (G_IO_OUT | G_IO_ERR)) &&
417 aio_node_check(ctx, node->is_external) &&
418 node->io_write) {
419 node->io_write(node->opaque);
420 progress = true;
421 }
422
423 tmp = node;
424 node = QLIST_NEXT(node, node);
425
426 ctx->walking_handlers--;
427
428 if (!ctx->walking_handlers && tmp->deleted) {
429 QLIST_REMOVE(tmp, node);
430 g_free(tmp);
431 }
432 }
433
434 /* Run our timers */
435 progress |= timerlistgroup_run_timers(&ctx->tlg);
436
437 return progress;
438 }
439
440 /* These thread-local variables are used only in a small part of aio_poll
441 * around the call to the poll() system call. In particular they are not
442 * used while aio_poll is performing callbacks, which makes it much easier
443 * to think about reentrancy!
444 *
445 * Stack-allocated arrays would be perfect but they have size limitations;
446 * heap allocation is expensive enough that we want to reuse arrays across
447 * calls to aio_poll(). And because poll() has to be called without holding
448 * any lock, the arrays cannot be stored in AioContext. Thread-local data
449 * has none of the disadvantages of these three options.
450 */
451 static __thread GPollFD *pollfds;
452 static __thread AioHandler **nodes;
453 static __thread unsigned npfd, nalloc;
454 static __thread Notifier pollfds_cleanup_notifier;
455
456 static void pollfds_cleanup(Notifier *n, void *unused)
457 {
458 g_assert(npfd == 0);
459 g_free(pollfds);
460 g_free(nodes);
461 nalloc = 0;
462 }
463
464 static void add_pollfd(AioHandler *node)
465 {
466 if (npfd == nalloc) {
467 if (nalloc == 0) {
468 pollfds_cleanup_notifier.notify = pollfds_cleanup;
469 qemu_thread_atexit_add(&pollfds_cleanup_notifier);
470 nalloc = 8;
471 } else {
472 g_assert(nalloc <= INT_MAX);
473 nalloc *= 2;
474 }
475 pollfds = g_renew(GPollFD, pollfds, nalloc);
476 nodes = g_renew(AioHandler *, nodes, nalloc);
477 }
478 nodes[npfd] = node;
479 pollfds[npfd] = (GPollFD) {
480 .fd = node->pfd.fd,
481 .events = node->pfd.events,
482 };
483 npfd++;
484 }
485
486 static bool run_poll_handlers_once(AioContext *ctx)
487 {
488 bool progress = false;
489 AioHandler *node;
490
491 QLIST_FOREACH(node, &ctx->aio_handlers, node) {
492 if (!node->deleted && node->io_poll &&
493 node->io_poll(node->opaque)) {
494 progress = true;
495 }
496
497 /* Caller handles freeing deleted nodes. Don't do it here. */
498 }
499
500 return progress;
501 }
502
503 /* run_poll_handlers:
504 * @ctx: the AioContext
505 * @max_ns: maximum time to poll for, in nanoseconds
506 *
507 * Polls for a given time.
508 *
509 * Note that ctx->notify_me must be non-zero so this function can detect
510 * aio_notify().
511 *
512 * Note that the caller must have incremented ctx->walking_handlers.
513 *
514 * Returns: true if progress was made, false otherwise
515 */
516 static bool run_poll_handlers(AioContext *ctx, int64_t max_ns)
517 {
518 bool progress;
519 int64_t end_time;
520
521 assert(ctx->notify_me);
522 assert(ctx->walking_handlers > 0);
523 assert(ctx->poll_disable_cnt == 0);
524
525 trace_run_poll_handlers_begin(ctx, max_ns);
526
527 end_time = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + max_ns;
528
529 do {
530 progress = run_poll_handlers_once(ctx);
531 } while (!progress && qemu_clock_get_ns(QEMU_CLOCK_REALTIME) < end_time);
532
533 trace_run_poll_handlers_end(ctx, progress);
534
535 return progress;
536 }
537
538 /* try_poll_mode:
539 * @ctx: the AioContext
540 * @blocking: busy polling is only attempted when blocking is true
541 *
542 * ctx->notify_me must be non-zero so this function can detect aio_notify().
543 *
544 * Note that the caller must have incremented ctx->walking_handlers.
545 *
546 * Returns: true if progress was made, false otherwise
547 */
548 static bool try_poll_mode(AioContext *ctx, bool blocking)
549 {
550 if (blocking && ctx->poll_max_ns && ctx->poll_disable_cnt == 0) {
551 /* See qemu_soonest_timeout() uint64_t hack */
552 int64_t max_ns = MIN((uint64_t)aio_compute_timeout(ctx),
553 (uint64_t)ctx->poll_ns);
554
555 if (max_ns) {
556 poll_set_started(ctx, true);
557
558 if (run_poll_handlers(ctx, max_ns)) {
559 return true;
560 }
561 }
562 }
563
564 poll_set_started(ctx, false);
565
566 /* Even if we don't run busy polling, try polling once in case it can make
567 * progress and the caller will be able to avoid ppoll(2)/epoll_wait(2).
568 */
569 return run_poll_handlers_once(ctx);
570 }
571
572 bool aio_poll(AioContext *ctx, bool blocking)
573 {
574 AioHandler *node;
575 int i;
576 int ret = 0;
577 bool progress;
578 int64_t timeout;
579 int64_t start = 0;
580
581 aio_context_acquire(ctx);
582 progress = false;
583
584 /* aio_notify can avoid the expensive event_notifier_set if
585 * everything (file descriptors, bottom halves, timers) will
586 * be re-evaluated before the next blocking poll(). This is
587 * already true when aio_poll is called with blocking == false;
588 * if blocking == true, it is only true after poll() returns,
589 * so disable the optimization now.
590 */
591 if (blocking) {
592 atomic_add(&ctx->notify_me, 2);
593 }
594
595 ctx->walking_handlers++;
596
597 if (ctx->poll_max_ns) {
598 start = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
599 }
600
601 if (try_poll_mode(ctx, blocking)) {
602 progress = true;
603 } else {
604 assert(npfd == 0);
605
606 /* fill pollfds */
607
608 if (!aio_epoll_enabled(ctx)) {
609 QLIST_FOREACH(node, &ctx->aio_handlers, node) {
610 if (!node->deleted && node->pfd.events
611 && aio_node_check(ctx, node->is_external)) {
612 add_pollfd(node);
613 }
614 }
615 }
616
617 timeout = blocking ? aio_compute_timeout(ctx) : 0;
618
619 /* wait until next event */
620 if (timeout) {
621 aio_context_release(ctx);
622 }
623 if (aio_epoll_check_poll(ctx, pollfds, npfd, timeout)) {
624 AioHandler epoll_handler;
625
626 epoll_handler.pfd.fd = ctx->epollfd;
627 epoll_handler.pfd.events = G_IO_IN | G_IO_OUT | G_IO_HUP | G_IO_ERR;
628 npfd = 0;
629 add_pollfd(&epoll_handler);
630 ret = aio_epoll(ctx, pollfds, npfd, timeout);
631 } else {
632 ret = qemu_poll_ns(pollfds, npfd, timeout);
633 }
634 if (timeout) {
635 aio_context_acquire(ctx);
636 }
637 }
638
639 if (blocking) {
640 atomic_sub(&ctx->notify_me, 2);
641 }
642
643 /* Adjust polling time */
644 if (ctx->poll_max_ns) {
645 int64_t block_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start;
646
647 if (block_ns <= ctx->poll_ns) {
648 /* This is the sweet spot, no adjustment needed */
649 } else if (block_ns > ctx->poll_max_ns) {
650 /* We'd have to poll for too long, poll less */
651 int64_t old = ctx->poll_ns;
652
653 if (ctx->poll_shrink) {
654 ctx->poll_ns /= ctx->poll_shrink;
655 } else {
656 ctx->poll_ns = 0;
657 }
658
659 trace_poll_shrink(ctx, old, ctx->poll_ns);
660 } else if (ctx->poll_ns < ctx->poll_max_ns &&
661 block_ns < ctx->poll_max_ns) {
662 /* There is room to grow, poll longer */
663 int64_t old = ctx->poll_ns;
664 int64_t grow = ctx->poll_grow;
665
666 if (grow == 0) {
667 grow = 2;
668 }
669
670 if (ctx->poll_ns) {
671 ctx->poll_ns *= grow;
672 } else {
673 ctx->poll_ns = 4000; /* start polling at 4 microseconds */
674 }
675
676 if (ctx->poll_ns > ctx->poll_max_ns) {
677 ctx->poll_ns = ctx->poll_max_ns;
678 }
679
680 trace_poll_grow(ctx, old, ctx->poll_ns);
681 }
682 }
683
684 aio_notify_accept(ctx);
685
686 /* if we have any readable fds, dispatch event */
687 if (ret > 0) {
688 for (i = 0; i < npfd; i++) {
689 nodes[i]->pfd.revents = pollfds[i].revents;
690 }
691 }
692
693 npfd = 0;
694 ctx->walking_handlers--;
695
696 /* Run dispatch even if there were no readable fds to run timers */
697 if (aio_dispatch(ctx, ret > 0)) {
698 progress = true;
699 }
700
701 aio_context_release(ctx);
702
703 return progress;
704 }
705
706 void aio_context_setup(AioContext *ctx)
707 {
708 /* TODO remove this in final patch submission */
709 if (getenv("QEMU_AIO_POLL_MAX_NS")) {
710 fprintf(stderr, "The QEMU_AIO_POLL_MAX_NS environment variable has "
711 "been replaced with -object iothread,poll-max-ns=NUM\n");
712 exit(1);
713 }
714
715 #ifdef CONFIG_EPOLL_CREATE1
716 assert(!ctx->epollfd);
717 ctx->epollfd = epoll_create1(EPOLL_CLOEXEC);
718 if (ctx->epollfd == -1) {
719 fprintf(stderr, "Failed to create epoll instance: %s", strerror(errno));
720 ctx->epoll_available = false;
721 } else {
722 ctx->epoll_available = true;
723 }
724 #endif
725 }
726
727 void aio_context_set_poll_params(AioContext *ctx, int64_t max_ns,
728 int64_t grow, int64_t shrink, Error **errp)
729 {
730 /* No thread synchronization here, it doesn't matter if an incorrect value
731 * is used once.
732 */
733 ctx->poll_max_ns = max_ns;
734 ctx->poll_ns = 0;
735 ctx->poll_grow = grow;
736 ctx->poll_shrink = shrink;
737
738 aio_notify(ctx);
739 }