X-Git-Url: https://repo.jachan.dev/qemu.git/blobdiff_plain/a97ff30e934c4d673122cf709e8e87f0effbe2f7..2d710006a0da4a9b7ddf5c02d072e178906d0ef6:/aio-posix.c diff --git a/aio-posix.c b/aio-posix.c index 88d09e1cfb..d4770336c5 100644 --- a/aio-posix.c +++ b/aio-posix.c @@ -23,7 +23,6 @@ struct AioHandler GPollFD pfd; IOHandler *io_read; IOHandler *io_write; - AioFlushHandler *io_flush; int deleted; void *opaque; QLIST_ENTRY(AioHandler) node; @@ -46,7 +45,6 @@ void aio_set_fd_handler(AioContext *ctx, int fd, IOHandler *io_read, IOHandler *io_write, - AioFlushHandler *io_flush, void *opaque) { AioHandler *node; @@ -74,7 +72,7 @@ void aio_set_fd_handler(AioContext *ctx, } else { if (node == NULL) { /* Alloc and insert if it's not already there */ - node = g_malloc0(sizeof(AioHandler)); + node = g_new0(AioHandler, 1); node->pfd.fd = fd; QLIST_INSERT_HEAD(&ctx->aio_handlers, node, node); @@ -83,11 +81,10 @@ void aio_set_fd_handler(AioContext *ctx, /* Update handler with latest information */ node->io_read = io_read; node->io_write = io_write; - node->io_flush = io_flush; node->opaque = opaque; - node->pfd.events = (io_read ? G_IO_IN | G_IO_HUP : 0); - node->pfd.events |= (io_write ? G_IO_OUT : 0); + node->pfd.events = (io_read ? G_IO_IN | G_IO_HUP | G_IO_ERR : 0); + node->pfd.events |= (io_write ? G_IO_OUT | G_IO_ERR : 0); } aio_notify(ctx); @@ -95,12 +92,15 @@ void aio_set_fd_handler(AioContext *ctx, void aio_set_event_notifier(AioContext *ctx, EventNotifier *notifier, - EventNotifierHandler *io_read, - AioFlushEventNotifierHandler *io_flush) + EventNotifierHandler *io_read) { aio_set_fd_handler(ctx, event_notifier_get_fd(notifier), - (IOHandler *)io_read, NULL, - (AioFlushHandler *)io_flush, notifier); + (IOHandler *)io_read, NULL, notifier); +} + +bool aio_prepare(AioContext *ctx) +{ + return false; } bool aio_pending(AioContext *ctx) @@ -110,13 +110,6 @@ bool aio_pending(AioContext *ctx) QLIST_FOREACH(node, &ctx->aio_handlers, node) { int revents; - /* - * FIXME: right now we cannot get G_IO_HUP and G_IO_ERR because - * main-loop.c is still select based (due to the slirp legacy). - * If main-loop.c ever switches to poll, G_IO_ERR should be - * tested too. Dispatching G_IO_ERR to both handlers should be - * okay, since handlers need to be ready for spurious wakeups. - */ revents = node->pfd.revents & node->pfd.events; if (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR) && node->io_read) { return true; @@ -129,31 +122,22 @@ bool aio_pending(AioContext *ctx) return false; } -bool aio_poll(AioContext *ctx, bool blocking) +bool aio_dispatch(AioContext *ctx) { - static struct timeval tv0; AioHandler *node; - fd_set rdfds, wrfds; - int max_fd = -1; - int ret; - bool busy, progress; - - progress = false; + bool progress = false; /* - * If there are callbacks left that have been queued, we need to call then. + * If there are callbacks left that have been queued, we need to call them. * Do not call select in this case, because it is possible that the caller - * does not need a complete flush (as is the case for qemu_aio_wait loops). + * does not need a complete flush (as is the case for aio_poll loops). */ if (aio_bh_poll(ctx)) { - blocking = false; progress = true; } /* - * Then dispatch any pending callbacks from the GSource. - * - * We have to walk very carefully in case qemu_aio_set_fd_handler is + * We have to walk very carefully in case aio_set_fd_handler is * called while we're walking. */ node = QLIST_FIRST(&ctx->aio_handlers); @@ -166,12 +150,19 @@ bool aio_poll(AioContext *ctx, bool blocking) revents = node->pfd.revents & node->pfd.events; node->pfd.revents = 0; - /* See comment in aio_pending. */ - if (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR) && node->io_read) { + if (!node->deleted && + (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR)) && + node->io_read) { node->io_read(node->opaque); - progress = true; + + /* aio_notify() does not count as progress */ + if (node->opaque != &ctx->notifier) { + progress = true; + } } - if (revents & (G_IO_OUT | G_IO_ERR) && node->io_write) { + if (!node->deleted && + (revents & (G_IO_OUT | G_IO_ERR)) && + node->io_write) { node->io_write(node->opaque); progress = true; } @@ -187,82 +178,122 @@ bool aio_poll(AioContext *ctx, bool blocking) } } - if (progress && !blocking) { - return true; + /* Run our timers */ + progress |= timerlistgroup_run_timers(&ctx->tlg); + + return progress; +} + +/* These thread-local variables are used only in a small part of aio_poll + * around the call to the poll() system call. In particular they are not + * used while aio_poll is performing callbacks, which makes it much easier + * to think about reentrancy! + * + * Stack-allocated arrays would be perfect but they have size limitations; + * heap allocation is expensive enough that we want to reuse arrays across + * calls to aio_poll(). And because poll() has to be called without holding + * any lock, the arrays cannot be stored in AioContext. Thread-local data + * has none of the disadvantages of these three options. + */ +static __thread GPollFD *pollfds; +static __thread AioHandler **nodes; +static __thread unsigned npfd, nalloc; +static __thread Notifier pollfds_cleanup_notifier; + +static void pollfds_cleanup(Notifier *n, void *unused) +{ + g_assert(npfd == 0); + g_free(pollfds); + g_free(nodes); + nalloc = 0; +} + +static void add_pollfd(AioHandler *node) +{ + if (npfd == nalloc) { + if (nalloc == 0) { + pollfds_cleanup_notifier.notify = pollfds_cleanup; + qemu_thread_atexit_add(&pollfds_cleanup_notifier); + nalloc = 8; + } else { + g_assert(nalloc <= INT_MAX); + nalloc *= 2; + } + pollfds = g_renew(GPollFD, pollfds, nalloc); + nodes = g_renew(AioHandler *, nodes, nalloc); + } + nodes[npfd] = node; + pollfds[npfd] = (GPollFD) { + .fd = node->pfd.fd, + .events = node->pfd.events, + }; + npfd++; +} + +bool aio_poll(AioContext *ctx, bool blocking) +{ + AioHandler *node; + int i, ret; + bool progress; + int64_t timeout; + + aio_context_acquire(ctx); + progress = false; + + /* aio_notify can avoid the expensive event_notifier_set if + * everything (file descriptors, bottom halves, timers) will + * be re-evaluated before the next blocking poll(). This is + * already true when aio_poll is called with blocking == false; + * if blocking == true, it is only true after poll() returns, + * so disable the optimization now. + */ + if (blocking) { + atomic_add(&ctx->notify_me, 2); } ctx->walking_handlers++; - FD_ZERO(&rdfds); - FD_ZERO(&wrfds); + assert(npfd == 0); - /* fill fd sets */ - busy = false; + /* fill pollfds */ QLIST_FOREACH(node, &ctx->aio_handlers, node) { - /* If there aren't pending AIO operations, don't invoke callbacks. - * Otherwise, if there are no AIO requests, qemu_aio_wait() would - * wait indefinitely. - */ - if (!node->deleted && node->io_flush) { - if (node->io_flush(node->opaque) == 0) { - continue; - } - busy = true; - } - if (!node->deleted && node->io_read) { - FD_SET(node->pfd.fd, &rdfds); - max_fd = MAX(max_fd, node->pfd.fd + 1); - } - if (!node->deleted && node->io_write) { - FD_SET(node->pfd.fd, &wrfds); - max_fd = MAX(max_fd, node->pfd.fd + 1); + if (!node->deleted && node->pfd.events) { + add_pollfd(node); } } - ctx->walking_handlers--; + timeout = blocking ? aio_compute_timeout(ctx) : 0; - /* No AIO operations? Get us out of here */ - if (!busy) { - return progress; + /* wait until next event */ + if (timeout) { + aio_context_release(ctx); + } + ret = qemu_poll_ns((GPollFD *)pollfds, npfd, timeout); + if (blocking) { + atomic_sub(&ctx->notify_me, 2); + } + if (timeout) { + aio_context_acquire(ctx); } - /* wait until next event */ - ret = select(max_fd, &rdfds, &wrfds, NULL, blocking ? NULL : &tv0); + aio_notify_accept(ctx); /* if we have any readable fds, dispatch event */ if (ret > 0) { - /* we have to walk very carefully in case - * qemu_aio_set_fd_handler is called while we're walking */ - node = QLIST_FIRST(&ctx->aio_handlers); - while (node) { - AioHandler *tmp; - - ctx->walking_handlers++; - - if (!node->deleted && - FD_ISSET(node->pfd.fd, &rdfds) && - node->io_read) { - node->io_read(node->opaque); - progress = true; - } - if (!node->deleted && - FD_ISSET(node->pfd.fd, &wrfds) && - node->io_write) { - node->io_write(node->opaque); - progress = true; - } - - tmp = node; - node = QLIST_NEXT(node, node); + for (i = 0; i < npfd; i++) { + nodes[i]->pfd.revents = pollfds[i].revents; + } + } - ctx->walking_handlers--; + npfd = 0; + ctx->walking_handlers--; - if (!ctx->walking_handlers && tmp->deleted) { - QLIST_REMOVE(tmp, node); - g_free(tmp); - } - } + /* Run dispatch even if there were no readable fds to run timers */ + if (aio_dispatch(ctx)) { + progress = true; } + aio_context_release(ctx); + return progress; }