#include "qemu/osdep.h"
#include "block/block.h"
+#include "block/thread-pool.h"
+#include "qemu/main-loop.h"
#include "qemu/rcu.h"
#include "qemu/rcu_queue.h"
#include "qemu/sockets.h"
bool aio_poll_disabled(AioContext *ctx)
{
- return atomic_read(&ctx->poll_disable_cnt);
+ return qatomic_read(&ctx->poll_disable_cnt);
}
void aio_add_ready_handler(AioHandlerList *ready_list,
QLIST_INSERT_HEAD(ready_list, node, node_ready);
}
+static void aio_add_poll_ready_handler(AioHandlerList *ready_list,
+ AioHandler *node)
+{
+ QLIST_SAFE_REMOVE(node, node_ready); /* remove from nested parent's list */
+ node->poll_ready = true;
+ QLIST_INSERT_HEAD(ready_list, node, node_ready);
+}
+
static AioHandler *find_aio_handler(AioContext *ctx, int fd)
{
AioHandler *node;
}
node->pfd.revents = 0;
+ node->poll_ready = false;
/* If the fd monitor has already marked it deleted, leave it alone */
if (QLIST_IS_INSERTED(node, node_deleted)) {
IOHandler *io_read,
IOHandler *io_write,
AioPollFn *io_poll,
+ IOHandler *io_poll_ready,
void *opaque)
{
AioHandler *node;
bool deleted = false;
int poll_disable_change;
+ if (io_poll && !io_poll_ready) {
+ io_poll = NULL; /* polling only makes sense if there is a handler */
+ }
+
qemu_lockcnt_lock(&ctx->list_lock);
node = find_aio_handler(ctx, fd);
new_node->io_read = io_read;
new_node->io_write = io_write;
new_node->io_poll = io_poll;
+ new_node->io_poll_ready = io_poll_ready;
new_node->opaque = opaque;
new_node->is_external = is_external;
* Changing handlers is a rare event, and a little wasted polling until
* the aio_notify below is not an issue.
*/
- atomic_set(&ctx->poll_disable_cnt,
- atomic_read(&ctx->poll_disable_cnt) + poll_disable_change);
+ qatomic_set(&ctx->poll_disable_cnt,
+ qatomic_read(&ctx->poll_disable_cnt) + poll_disable_change);
ctx->fdmon_ops->update(ctx, node, new_node);
if (node) {
EventNotifier *notifier,
bool is_external,
EventNotifierHandler *io_read,
- AioPollFn *io_poll)
+ AioPollFn *io_poll,
+ EventNotifierHandler *io_poll_ready)
{
aio_set_fd_handler(ctx, event_notifier_get_fd(notifier), is_external,
- (IOHandler *)io_read, NULL, io_poll, notifier);
+ (IOHandler *)io_read, NULL, io_poll,
+ (IOHandler *)io_poll_ready, notifier);
}
void aio_set_event_notifier_poll(AioContext *ctx,
(IOHandler *)io_poll_end);
}
-static bool poll_set_started(AioContext *ctx, bool started)
+static bool poll_set_started(AioContext *ctx, AioHandlerList *ready_list,
+ bool started)
{
AioHandler *node;
bool progress = false;
}
/* Poll one last time in case ->io_poll_end() raced with the event */
- if (!started) {
- progress = node->io_poll(node->opaque) || progress;
+ if (!started && node->io_poll(node->opaque)) {
+ aio_add_poll_ready_handler(ready_list, node);
+ progress = true;
}
}
qemu_lockcnt_dec(&ctx->list_lock);
bool aio_prepare(AioContext *ctx)
{
+ AioHandlerList ready_list = QLIST_HEAD_INITIALIZER(ready_list);
+
/* Poll mode cannot be used with glib's event loop, disable it. */
- poll_set_started(ctx, false);
+ poll_set_started(ctx, &ready_list, false);
+ /* TODO what to do with this list? */
return false;
}
QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {
int revents;
+ /* TODO should this check poll ready? */
revents = node->pfd.revents & node->pfd.events;
if (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR) && node->io_read &&
aio_node_check(ctx, node->is_external)) {
static bool aio_dispatch_handler(AioContext *ctx, AioHandler *node)
{
bool progress = false;
+ bool poll_ready;
int revents;
revents = node->pfd.revents & node->pfd.events;
node->pfd.revents = 0;
+ poll_ready = node->poll_ready;
+ node->poll_ready = false;
+
/*
* Start polling AioHandlers when they become ready because activity is
* likely to continue. Note that starvation is theoretically possible when
}
QLIST_INSERT_HEAD(&ctx->poll_aio_handlers, node, node_poll);
}
+ if (!QLIST_IS_INSERTED(node, node_deleted) &&
+ poll_ready && revents == 0 &&
+ aio_node_check(ctx, node->is_external) &&
+ node->io_poll_ready) {
+ node->io_poll_ready(node->opaque);
+
+ /*
+ * Return early since revents was zero. aio_notify() does not count as
+ * progress.
+ */
+ return node->opaque != &ctx->notifier;
+ }
if (!QLIST_IS_INSERTED(node, node_deleted) &&
(revents & (G_IO_IN | G_IO_HUP | G_IO_ERR)) &&
}
static bool run_poll_handlers_once(AioContext *ctx,
+ AioHandlerList *ready_list,
int64_t now,
int64_t *timeout)
{
QLIST_FOREACH_SAFE(node, &ctx->poll_aio_handlers, node_poll, tmp) {
if (aio_node_check(ctx, node->is_external) &&
node->io_poll(node->opaque)) {
+ aio_add_poll_ready_handler(ready_list, node);
+
node->poll_idle_timeout = now + POLL_IDLE_INTERVAL_NS;
/*
return ctx->fdmon_ops->need_wait != aio_poll_disabled;
}
-static bool remove_idle_poll_handlers(AioContext *ctx, int64_t now)
+static bool remove_idle_poll_handlers(AioContext *ctx,
+ AioHandlerList *ready_list,
+ int64_t now)
{
AioHandler *node;
AioHandler *tmp;
* Nevermind about re-adding the handler in the rare case where
* this causes progress.
*/
- progress = node->io_poll(node->opaque) || progress;
+ if (node->io_poll(node->opaque)) {
+ aio_add_poll_ready_handler(ready_list, node);
+ progress = true;
+ }
}
}
}
/* run_poll_handlers:
* @ctx: the AioContext
+ * @ready_list: the list to place ready handlers on
* @max_ns: maximum time to poll for, in nanoseconds
*
* Polls for a given time.
*
- * Note that ctx->notify_me must be non-zero so this function can detect
- * aio_notify().
- *
* Note that the caller must have incremented ctx->list_lock.
*
* Returns: true if progress was made, false otherwise
*/
-static bool run_poll_handlers(AioContext *ctx, int64_t max_ns, int64_t *timeout)
+static bool run_poll_handlers(AioContext *ctx, AioHandlerList *ready_list,
+ int64_t max_ns, int64_t *timeout)
{
bool progress;
int64_t start_time, elapsed_time;
- assert(ctx->notify_me);
assert(qemu_lockcnt_count(&ctx->list_lock) > 0);
trace_run_poll_handlers_begin(ctx, max_ns, *timeout);
start_time = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
do {
- progress = run_poll_handlers_once(ctx, start_time, timeout);
+ progress = run_poll_handlers_once(ctx, ready_list,
+ start_time, timeout);
elapsed_time = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start_time;
max_ns = qemu_soonest_timeout(*timeout, max_ns);
assert(!(max_ns && progress));
} while (elapsed_time < max_ns && !ctx->fdmon_ops->need_wait(ctx));
- if (remove_idle_poll_handlers(ctx, start_time + elapsed_time)) {
+ if (remove_idle_poll_handlers(ctx, ready_list,
+ start_time + elapsed_time)) {
*timeout = 0;
progress = true;
}
/* try_poll_mode:
* @ctx: the AioContext
+ * @ready_list: list to add handlers that need to be run
* @timeout: timeout for blocking wait, computed by the caller and updated if
* polling succeeds.
*
- * ctx->notify_me must be non-zero so this function can detect aio_notify().
- *
* Note that the caller must have incremented ctx->list_lock.
*
* Returns: true if progress was made, false otherwise
*/
-static bool try_poll_mode(AioContext *ctx, int64_t *timeout)
+static bool try_poll_mode(AioContext *ctx, AioHandlerList *ready_list,
+ int64_t *timeout)
{
int64_t max_ns;
max_ns = qemu_soonest_timeout(*timeout, ctx->poll_ns);
if (max_ns && !ctx->fdmon_ops->need_wait(ctx)) {
- poll_set_started(ctx, true);
+ poll_set_started(ctx, ready_list, true);
- if (run_poll_handlers(ctx, max_ns, timeout)) {
+ if (run_poll_handlers(ctx, ready_list, max_ns, timeout)) {
return true;
}
}
- if (poll_set_started(ctx, false)) {
+ if (poll_set_started(ctx, ready_list, false)) {
*timeout = 0;
return true;
}
bool aio_poll(AioContext *ctx, bool blocking)
{
AioHandlerList ready_list = QLIST_HEAD_INITIALIZER(ready_list);
- int ret = 0;
bool progress;
+ bool use_notify_me;
int64_t timeout;
int64_t start = 0;
- assert(in_aio_context_home_thread(ctx));
-
- /* 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.
+ /*
+ * There cannot be two concurrent aio_poll calls for the same AioContext (or
+ * an aio_poll concurrent with a GSource prepare/check/dispatch callback).
+ * We rely on this below to avoid slow locked accesses to ctx->notify_me.
+ *
+ * aio_poll() may only be called in the AioContext's thread. iohandler_ctx
+ * is special in that it runs in the main thread, but that thread's context
+ * is qemu_aio_context.
*/
- if (blocking) {
- atomic_add(&ctx->notify_me, 2);
- }
+ assert(in_aio_context_home_thread(ctx == iohandler_get_aio_context() ?
+ qemu_get_aio_context() : ctx));
qemu_lockcnt_inc(&ctx->list_lock);
}
timeout = blocking ? aio_compute_timeout(ctx) : 0;
- progress = try_poll_mode(ctx, &timeout);
+ progress = try_poll_mode(ctx, &ready_list, &timeout);
assert(!(timeout && progress));
+ /*
+ * 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.
+ */
+ use_notify_me = timeout != 0;
+ if (use_notify_me) {
+ qatomic_set(&ctx->notify_me, qatomic_read(&ctx->notify_me) + 2);
+ /*
+ * Write ctx->notify_me before reading ctx->notified. Pairs with
+ * smp_mb in aio_notify().
+ */
+ smp_mb();
+
+ /* Don't block if aio_notify() was called */
+ if (qatomic_read(&ctx->notified)) {
+ timeout = 0;
+ }
+ }
+
/* If polling is allowed, non-blocking aio_poll does not need the
* system call---a single round of run_poll_handlers_once suffices.
*/
if (timeout || ctx->fdmon_ops->need_wait(ctx)) {
- ret = ctx->fdmon_ops->wait(ctx, &ready_list, timeout);
+ ctx->fdmon_ops->wait(ctx, &ready_list, timeout);
}
- if (blocking) {
- atomic_sub(&ctx->notify_me, 2);
- aio_notify_accept(ctx);
+ if (use_notify_me) {
+ /* Finish the poll before clearing the flag. */
+ qatomic_store_release(&ctx->notify_me,
+ qatomic_read(&ctx->notify_me) - 2);
}
+ aio_notify_accept(ctx);
+
/* Adjust polling time */
if (ctx->poll_max_ns) {
int64_t block_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start;
}
progress |= aio_bh_poll(ctx);
-
- if (ret > 0) {
- progress |= aio_dispatch_ready_handlers(ctx, &ready_list);
- }
+ progress |= aio_dispatch_ready_handlers(ctx, &ready_list);
aio_free_deleted_handlers(ctx);
{
fdmon_io_uring_destroy(ctx);
fdmon_epoll_disable(ctx);
+ aio_free_deleted_handlers(ctx);
+}
+
+void aio_context_use_g_source(AioContext *ctx)
+{
+ /*
+ * Disable io_uring when the glib main loop is used because it doesn't
+ * support mixed glib/aio_poll() usage. It relies on aio_poll() being
+ * called regularly so that changes to the monitored file descriptors are
+ * submitted, otherwise a list of pending fd handlers builds up.
+ */
+ fdmon_io_uring_destroy(ctx);
+ aio_free_deleted_handlers(ctx);
}
void aio_context_set_poll_params(AioContext *ctx, int64_t max_ns,
aio_notify(ctx);
}
+
+void aio_context_set_aio_params(AioContext *ctx, int64_t max_batch,
+ Error **errp)
+{
+ /*
+ * No thread synchronization here, it doesn't matter if an incorrect value
+ * is used once.
+ */
+ ctx->aio_max_batch = max_batch;
+
+ aio_notify(ctx);
+}
projects / mirror_qemu.git / blobdiff