*/
#include "qemu-common.h"
-#include "block.h"
-#include "qemu-queue.h"
-#include "qemu_socket.h"
+#include "block/block.h"
+#include "qemu/queue.h"
+#include "qemu/sockets.h"
struct AioHandler
{
GPollFD pfd;
IOHandler *io_read;
IOHandler *io_write;
- AioFlushHandler *io_flush;
int deleted;
void *opaque;
QLIST_ENTRY(AioHandler) node;
int fd,
IOHandler *io_read,
IOHandler *io_write,
- AioFlushHandler *io_flush,
void *opaque)
{
AioHandler *node;
/* Are we deleting the fd handler? */
if (!io_read && !io_write) {
if (node) {
+ g_source_remove_poll(&ctx->source, &node->pfd);
+
/* If the lock is held, just mark the node as deleted */
if (ctx->walking_handlers) {
node->deleted = 1;
} 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);
+
+ g_source_add_poll(&ctx->source, &node->pfd);
}
/* 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);
}
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)
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;
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);
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;
}
}
}
- 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;
+ bool was_dispatching;
+ int i, ret;
+ bool progress;
+ int64_t timeout;
+
+ aio_context_acquire(ctx);
+ was_dispatching = ctx->dispatching;
+ 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.
+ *
+ * If we're in a nested event loop, ctx->dispatching might be true.
+ * In that case we can restore it just before returning, but we
+ * have to clear it now.
+ */
+ aio_set_dispatching(ctx, !blocking);
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--;
-
- /* No AIO operations? Get us out of here */
- if (!busy) {
- return progress;
- }
+ timeout = blocking ? aio_compute_timeout(ctx) : 0;
/* wait until next event */
- ret = select(max_fd, &rdfds, &wrfds, NULL, blocking ? NULL : &tv0);
+ if (timeout) {
+ aio_context_release(ctx);
+ }
+ ret = qemu_poll_ns((GPollFD *)pollfds, npfd, timeout);
+ if (timeout) {
+ aio_context_acquire(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 */
+ aio_set_dispatching(ctx, true);
+ if (aio_dispatch(ctx)) {
+ progress = true;
}
+ aio_set_dispatching(ctx, was_dispatching);
+ aio_context_release(ctx);
+
return progress;
}
projects / qemu.git / blobdiff