The following changes since commit 9ac5df20f51fabcba0d902025df4bd7ea987c158:

Merge remote-tracking branch 'remotes/pmaydell/tags/pull-target-arm-20200221-1' into staging (2020-02-21 16:18:38 +0000)

are available in the Git repository at:

https://github.com/stefanha/qemu.git tags/block-pull-request

for you to fetch changes up to e5c59355ae9f724777c61c859292ec9db2c8c2ab:

fuzz: add documentation to docs/devel/ (2020-02-22 08:26:48 +0000)

Pull request

This pull request contains a virtio-blk/scsi performance optimization, event

loop scalability improvements, and a qtest-based device fuzzing framework. I

am including the fuzzing patches because I have reviewed them and Thomas Huth

is currently away on leave.

Alexander Bulekov (22):

softmmu: move vl.c to softmmu/

softmmu: split off vl.c:main() into main.c

module: check module wasn't already initialized

fuzz: add FUZZ_TARGET module type

qtest: add qtest_server_send abstraction

libqtest: add a layer of abstraction to send/recv

libqtest: make bufwrite rely on the TransportOps

qtest: add in-process incoming command handler

libqos: rename i2c_send and i2c_recv

libqos: split qos-test and libqos makefile vars

libqos: move useful qos-test funcs to qos_external

fuzz: add fuzzer skeleton

exec: keep ram block across fork when using qtest

main: keep rcu_atfork callback enabled for qtest

fuzz: support for fork-based fuzzing.

fuzz: add support for qos-assisted fuzz targets

fuzz: add target/fuzz makefile rules

fuzz: add configure flag --enable-fuzzing

fuzz: add i440fx fuzz targets

fuzz: add virtio-net fuzz target

fuzz: add virtio-scsi fuzz target

fuzz: add documentation to docs/devel/

Denis Plotnikov (1):

virtio: increase virtqueue size for virtio-scsi and virtio-blk

Paolo Bonzini (1):

rcu_queue: add QSLIST functions

Stefan Hajnoczi (7):

aio-posix: avoid reacquiring rcu_read_lock() when polling

util/async: make bh_aio_poll() O(1)

aio-posix: fix use after leaving scope in aio_poll()

aio-posix: don't pass ns timeout to epoll_wait()

qemu/queue.h: add QLIST_SAFE_REMOVE()

aio-posix: make AioHandler deletion O(1)

aio-posix: make AioHandler dispatch O(1) with epoll

MAINTAINERS | 11 +-

Makefile | 15 +-

Makefile.objs | 2 -

Makefile.target | 19 ++-

block.c | 5 +-

chardev/spice.c | 4 +-

configure | 39 +++++

docs/devel/fuzzing.txt | 116 ++++++++++++++

exec.c | 12 +-

hw/block/virtio-blk.c | 2 +-

hw/core/machine.c | 2 +

hw/scsi/virtio-scsi.c | 2 +-

include/block/aio.h | 26 ++-

include/qemu/module.h | 4 +-

include/qemu/queue.h | 32 +++-

include/qemu/rcu_queue.h | 47 ++++++

include/sysemu/qtest.h | 4 +

include/sysemu/sysemu.h | 4 +

qtest.c | 31 +++-

scripts/checkpatch.pl | 2 +-

scripts/get_maintainer.pl | 3 +-

softmmu/Makefile.objs | 3 +

softmmu/main.c | 53 +++++++

vl.c => softmmu/vl.c | 48 +++---

tests/Makefile.include | 2 +

tests/qtest/Makefile.include | 72 +++++----

tests/qtest/fuzz/Makefile.include | 18 +++

tests/qtest/fuzz/fork_fuzz.c | 55 +++++++

tests/qtest/fuzz/fork_fuzz.h | 23 +++

tests/qtest/fuzz/fork_fuzz.ld | 37 +++++

tests/qtest/fuzz/fuzz.c | 179 +++++++++++++++++++++

tests/qtest/fuzz/fuzz.h | 95 +++++++++++

tests/qtest/fuzz/i440fx_fuzz.c | 193 ++++++++++++++++++++++

tests/qtest/fuzz/qos_fuzz.c | 234 +++++++++++++++++++++++++++

tests/qtest/fuzz/qos_fuzz.h | 33 ++++

tests/qtest/fuzz/virtio_net_fuzz.c | 198 +++++++++++++++++++++++

tests/qtest/fuzz/virtio_scsi_fuzz.c | 213 +++++++++++++++++++++++++

tests/qtest/libqos/i2c.c | 10 +-

tests/qtest/libqos/i2c.h | 4 +-

tests/qtest/libqos/qos_external.c | 168 ++++++++++++++++++++

tests/qtest/libqos/qos_external.h | 28 ++++

tests/qtest/libqtest.c | 119 ++++++++++++--

tests/qtest/libqtest.h | 4 +

tests/qtest/pca9552-test.c | 10 +-

tests/qtest/qos-test.c | 132 +---------------

tests/test-aio.c | 3 +-

tests/test-rcu-list.c | 16 ++

tests/test-rcu-slist.c | 2 +

util/aio-posix.c | 187 +++++++++++++++-------

util/async.c | 237 ++++++++++++++++------------

util/module.c | 7 +

51 files changed, 2365 insertions(+), 400 deletions(-)

create mode 100644 docs/devel/fuzzing.txt

create mode 100644 softmmu/Makefile.objs

create mode 100644 softmmu/main.c

rename vl.c => softmmu/vl.c (99%)

create mode 100644 tests/qtest/fuzz/Makefile.include

create mode 100644 tests/qtest/fuzz/fork_fuzz.c

create mode 100644 tests/qtest/fuzz/fork_fuzz.h

create mode 100644 tests/qtest/fuzz/fork_fuzz.ld

create mode 100644 tests/qtest/fuzz/fuzz.c

create mode 100644 tests/qtest/fuzz/fuzz.h

create mode 100644 tests/qtest/fuzz/i440fx_fuzz.c

create mode 100644 tests/qtest/fuzz/qos_fuzz.c

create mode 100644 tests/qtest/fuzz/qos_fuzz.h

create mode 100644 tests/qtest/fuzz/virtio_net_fuzz.c

create mode 100644 tests/qtest/fuzz/virtio_scsi_fuzz.c

create mode 100644 tests/qtest/libqos/qos_external.c

create mode 100644 tests/qtest/libqos/qos_external.h

create mode 100644 tests/test-rcu-slist.c

2.24.1

The ctx->first_bh list contains all created BHs, including those that

are not scheduled. The list is iterated by the event loop and therefore

has O(n) time complexity with respected to the number of created BHs.

Rewrite BHs so that only scheduled or deleted BHs are enqueued.

Only BHs that actually require action will be iterated.

One semantic change is required: qemu_bh_delete() enqueues the BH and

therefore invokes aio_notify(). The

tests/test-aio.c:test_source_bh_delete_from_cb() test case assumed that

g_main_context_iteration(NULL, false) returns false after

qemu_bh_delete() but it now returns true for one iteration. Fix up the

test case.

This patch makes aio_compute_timeout() and aio_bh_poll() drop from a CPU

profile reported by perf-top(1). Previously they combined to 9% CPU

utilization when AioContext polling is commented out and the guest has 2

virtio-blk,num-queues=1 and 99 virtio-blk,num-queues=32 devices.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>

Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>

Message-id: 20200221093951.1414693-1-stefanha@redhat.com

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>

---

include/block/aio.h | 20 +++-

tests/test-aio.c | 3 +-

util/async.c | 237 ++++++++++++++++++++++++++------------------

3 files changed, 158 insertions(+), 102 deletions(-)

diff --git a/include/block/aio.h b/include/block/aio.h

index XXXXXXX..XXXXXXX 100644

--- a/include/block/aio.h

+++ b/include/block/aio.h

@@ -XXX,XX +XXX,XX @@ struct ThreadPool;

struct LinuxAioState;

struct LuringState;

+/*

+ * Each aio_bh_poll() call carves off a slice of the BH list, so that newly

+ * scheduled BHs are not processed until the next aio_bh_poll() call. All

+ * active aio_bh_poll() calls chain their slices together in a list, so that

+ * nested aio_bh_poll() calls process all scheduled bottom halves.

+ */

+typedef QSLIST_HEAD(, QEMUBH) BHList;

+typedef struct BHListSlice BHListSlice;

+struct BHListSlice {

+ BHList bh_list;

+ QSIMPLEQ_ENTRY(BHListSlice) next;

+};

+

struct AioContext {

GSource source;

@@ -XXX,XX +XXX,XX @@ struct AioContext {

*/

QemuLockCnt list_lock;

- /* Anchor of the list of Bottom Halves belonging to the context */

- struct QEMUBH *first_bh;

+ /* Bottom Halves pending aio_bh_poll() processing */

+ BHList bh_list;

+

+ /* Chained BH list slices for each nested aio_bh_poll() call */

+ QSIMPLEQ_HEAD(, BHListSlice) bh_slice_list;

/* Used by aio_notify.

*

diff --git a/tests/test-aio.c b/tests/test-aio.c

index XXXXXXX..XXXXXXX 100644

--- a/tests/test-aio.c

+++ b/tests/test-aio.c

@@ -XXX,XX +XXX,XX @@ static void test_source_bh_delete_from_cb(void)

g_assert_cmpint(data1.n, ==, data1.max);

g_assert(data1.bh == NULL);

- g_assert(!g_main_context_iteration(NULL, false));

+ assert(g_main_context_iteration(NULL, false));

+ assert(!g_main_context_iteration(NULL, false));

}

static void test_source_bh_delete_from_cb_many(void)

diff --git a/util/async.c b/util/async.c

index XXXXXXX..XXXXXXX 100644

--- a/util/async.c

+++ b/util/async.c

@@ -XXX,XX +XXX,XX @@

#include "block/thread-pool.h"

#include "qemu/main-loop.h"

#include "qemu/atomic.h"

+#include "qemu/rcu_queue.h"

#include "block/raw-aio.h"

#include "qemu/coroutine_int.h"

#include "trace.h"

@@ -XXX,XX +XXX,XX @@

/***********************************************************/

/* bottom halves (can be seen as timers which expire ASAP) */

+/* QEMUBH::flags values */

+enum {

+ /* Already enqueued and waiting for aio_bh_poll() */

+ BH_PENDING = (1 << 0),

+

+ /* Invoke the callback */

+ BH_SCHEDULED = (1 << 1),

+

+ /* Delete without invoking callback */

+ BH_DELETED = (1 << 2),

+

+ /* Delete after invoking callback */

+ BH_ONESHOT = (1 << 3),

+

+ /* Schedule periodically when the event loop is idle */

+ BH_IDLE = (1 << 4),

+};

+

struct QEMUBH {

AioContext *ctx;

QEMUBHFunc *cb;

void *opaque;

- QEMUBH *next;

- bool scheduled;

- bool idle;

- bool deleted;

+ QSLIST_ENTRY(QEMUBH) next;

+ unsigned flags;

};

+/* Called concurrently from any thread */

+static void aio_bh_enqueue(QEMUBH *bh, unsigned new_flags)

+{

+ AioContext *ctx = bh->ctx;

+ unsigned old_flags;

+

+ /*

+ * The memory barrier implicit in atomic_fetch_or makes sure that:

+ * 1. idle & any writes needed by the callback are done before the

+ * locations are read in the aio_bh_poll.

+ * 2. ctx is loaded before the callback has a chance to execute and bh

+ * could be freed.

+ */

+ old_flags = atomic_fetch_or(&bh->flags, BH_PENDING | new_flags);

+ if (!(old_flags & BH_PENDING)) {

+ QSLIST_INSERT_HEAD_ATOMIC(&ctx->bh_list, bh, next);

+ }

+

+ aio_notify(ctx);

+}

+

+/* Only called from aio_bh_poll() and aio_ctx_finalize() */

+static QEMUBH *aio_bh_dequeue(BHList *head, unsigned *flags)

+{

+ QEMUBH *bh = QSLIST_FIRST_RCU(head);

+

+ if (!bh) {

+ return NULL;

+ }

+

+ QSLIST_REMOVE_HEAD(head, next);

+

+ /*

+ * The atomic_and is paired with aio_bh_enqueue(). The implicit memory

+ * barrier ensures that the callback sees all writes done by the scheduling

+ * thread. It also ensures that the scheduling thread sees the cleared

+ * flag before bh->cb has run, and thus will call aio_notify again if

+ * necessary.

+ */

+ *flags = atomic_fetch_and(&bh->flags,

+ ~(BH_PENDING | BH_SCHEDULED | BH_IDLE));

+ return bh;

+}

+

void aio_bh_schedule_oneshot(AioContext *ctx, QEMUBHFunc *cb, void *opaque)

{

QEMUBH *bh;

@@ -XXX,XX +XXX,XX @@ void aio_bh_schedule_oneshot(AioContext *ctx, QEMUBHFunc *cb, void *opaque)

.cb = cb,

.opaque = opaque,

};

- qemu_lockcnt_lock(&ctx->list_lock);

- bh->next = ctx->first_bh;

- bh->scheduled = 1;

- bh->deleted = 1;

- /* Make sure that the members are ready before putting bh into list */

- smp_wmb();

- ctx->first_bh = bh;

- qemu_lockcnt_unlock(&ctx->list_lock);

- aio_notify(ctx);

+ aio_bh_enqueue(bh, BH_SCHEDULED | BH_ONESHOT);

}

QEMUBH *aio_bh_new(AioContext *ctx, QEMUBHFunc *cb, void *opaque)

@@ -XXX,XX +XXX,XX @@ QEMUBH *aio_bh_new(AioContext *ctx, QEMUBHFunc *cb, void *opaque)

.cb = cb,

.opaque = opaque,

};

- qemu_lockcnt_lock(&ctx->list_lock);

- bh->next = ctx->first_bh;

- /* Make sure that the members are ready before putting bh into list */

- smp_wmb();

- ctx->first_bh = bh;

- qemu_lockcnt_unlock(&ctx->list_lock);

return bh;

}

@@ -XXX,XX +XXX,XX @@ void aio_bh_call(QEMUBH *bh)

bh->cb(bh->opaque);

}

-/* Multiple occurrences of aio_bh_poll cannot be called concurrently.

- * The count in ctx->list_lock is incremented before the call, and is

- * not affected by the call.

- */

+/* Multiple occurrences of aio_bh_poll cannot be called concurrently. */

int aio_bh_poll(AioContext *ctx)

{

- QEMUBH *bh, **bhp, *next;

- int ret;

- bool deleted = false;

-

- ret = 0;

- for (bh = atomic_rcu_read(&ctx->first_bh); bh; bh = next) {

- next = atomic_rcu_read(&bh->next);

- /* The atomic_xchg is paired with the one in qemu_bh_schedule. The

- * implicit memory barrier ensures that the callback sees all writes

- * done by the scheduling thread. It also ensures that the scheduling

- * thread sees the zero before bh->cb has run, and thus will call

- * aio_notify again if necessary.

- */

- if (atomic_xchg(&bh->scheduled, 0)) {

+ BHListSlice slice;

+ BHListSlice *s;

+ int ret = 0;

+

+ QSLIST_MOVE_ATOMIC(&slice.bh_list, &ctx->bh_list);

+ QSIMPLEQ_INSERT_TAIL(&ctx->bh_slice_list, &slice, next);

+

+ while ((s = QSIMPLEQ_FIRST(&ctx->bh_slice_list))) {

+ QEMUBH *bh;

+ unsigned flags;

+

+ bh = aio_bh_dequeue(&s->bh_list, &flags);

+ if (!bh) {

+ QSIMPLEQ_REMOVE_HEAD(&ctx->bh_slice_list, next);

+ continue;

+ }

+

+ if ((flags & (BH_SCHEDULED | BH_DELETED)) == BH_SCHEDULED) {

/* Idle BHs don't count as progress */

- if (!bh->idle) {

+ if (!(flags & BH_IDLE)) {

ret = 1;

}

- bh->idle = 0;

aio_bh_call(bh);

}

- if (bh->deleted) {

- deleted = true;

+ if (flags & (BH_DELETED | BH_ONESHOT)) {

+ g_free(bh);

}

}

- /* remove deleted bhs */

- if (!deleted) {

- return ret;

- }

-

- if (qemu_lockcnt_dec_if_lock(&ctx->list_lock)) {

- bhp = &ctx->first_bh;

- while (*bhp) {

- bh = *bhp;

- if (bh->deleted && !bh->scheduled) {

- *bhp = bh->next;

- g_free(bh);

- } else {

- bhp = &bh->next;

- }

- }

- qemu_lockcnt_inc_and_unlock(&ctx->list_lock);

- }

return ret;

}

void qemu_bh_schedule_idle(QEMUBH *bh)

{

- bh->idle = 1;

- /* Make sure that idle & any writes needed by the callback are done

- * before the locations are read in the aio_bh_poll.

- */

- atomic_mb_set(&bh->scheduled, 1);

+ aio_bh_enqueue(bh, BH_SCHEDULED | BH_IDLE);

}

void qemu_bh_schedule(QEMUBH *bh)

{

- AioContext *ctx;

-

- ctx = bh->ctx;

- bh->idle = 0;

- /* The memory barrier implicit in atomic_xchg makes sure that:

- * 1. idle & any writes needed by the callback are done before the

- * locations are read in the aio_bh_poll.

- * 2. ctx is loaded before scheduled is set and the callback has a chance

- * to execute.

- */

- if (atomic_xchg(&bh->scheduled, 1) == 0) {

- aio_notify(ctx);

- }

+ aio_bh_enqueue(bh, BH_SCHEDULED);

}

-

/* This func is async.

*/

void qemu_bh_cancel(QEMUBH *bh)

{

- atomic_mb_set(&bh->scheduled, 0);

+ atomic_and(&bh->flags, ~BH_SCHEDULED);

}

/* This func is async.The bottom half will do the delete action at the finial

@@ -XXX,XX +XXX,XX @@ void qemu_bh_cancel(QEMUBH *bh)

*/

void qemu_bh_delete(QEMUBH *bh)

{

- bh->scheduled = 0;

- bh->deleted = 1;

+ aio_bh_enqueue(bh, BH_DELETED);

}

-int64_t

-aio_compute_timeout(AioContext *ctx)

+static int64_t aio_compute_bh_timeout(BHList *head, int timeout)

{

- int64_t deadline;

- int timeout = -1;

QEMUBH *bh;

- for (bh = atomic_rcu_read(&ctx->first_bh); bh;

- bh = atomic_rcu_read(&bh->next)) {

- if (bh->scheduled) {

- if (bh->idle) {

+ QSLIST_FOREACH_RCU(bh, head, next) {

+ if ((bh->flags & (BH_SCHEDULED | BH_DELETED)) == BH_SCHEDULED) {

+ if (bh->flags & BH_IDLE) {

/* idle bottom halves will be polled at least

* every 10ms */

timeout = 10000000;

@@ -XXX,XX +XXX,XX @@ aio_compute_timeout(AioContext *ctx)

}

}

+ return timeout;

+}

+

+int64_t

+aio_compute_timeout(AioContext *ctx)

+{

+ BHListSlice *s;

+ int64_t deadline;

+ int timeout = -1;

+

+ timeout = aio_compute_bh_timeout(&ctx->bh_list, timeout);

+ if (timeout == 0) {

+ return 0;

+ }

+

+ QSIMPLEQ_FOREACH(s, &ctx->bh_slice_list, next) {

+ timeout = aio_compute_bh_timeout(&s->bh_list, timeout);

+ if (timeout == 0) {

+ return 0;

+ }

+ }

+

deadline = timerlistgroup_deadline_ns(&ctx->tlg);

if (deadline == 0) {

return 0;

@@ -XXX,XX +XXX,XX @@ aio_ctx_check(GSource *source)

{

AioContext *ctx = (AioContext *) source;

QEMUBH *bh;

+ BHListSlice *s;

atomic_and(&ctx->notify_me, ~1);

aio_notify_accept(ctx);

- for (bh = ctx->first_bh; bh; bh = bh->next) {

- if (bh->scheduled) {

+ QSLIST_FOREACH_RCU(bh, &ctx->bh_list, next) {

+ if ((bh->flags & (BH_SCHEDULED | BH_DELETED)) == BH_SCHEDULED) {

return true;

}

}

+

+ QSIMPLEQ_FOREACH(s, &ctx->bh_slice_list, next) {

+ QSLIST_FOREACH_RCU(bh, &s->bh_list, next) {

+ if ((bh->flags & (BH_SCHEDULED | BH_DELETED)) == BH_SCHEDULED) {

+ return true;

+ }

+ }

+ }

return aio_pending(ctx) || (timerlistgroup_deadline_ns(&ctx->tlg) == 0);

}

@@ -XXX,XX +XXX,XX @@ static void

aio_ctx_finalize(GSource *source)

{

AioContext *ctx = (AioContext *) source;

+ QEMUBH *bh;

+ unsigned flags;

thread_pool_free(ctx->thread_pool);

@@ -XXX,XX +XXX,XX @@ aio_ctx_finalize(GSource *source)

assert(QSLIST_EMPTY(&ctx->scheduled_coroutines));

qemu_bh_delete(ctx->co_schedule_bh);

- qemu_lockcnt_lock(&ctx->list_lock);

- assert(!qemu_lockcnt_count(&ctx->list_lock));

- while (ctx->first_bh) {

- QEMUBH *next = ctx->first_bh->next;

+ /* There must be no aio_bh_poll() calls going on */

+ assert(QSIMPLEQ_EMPTY(&ctx->bh_slice_list));

+ while ((bh = aio_bh_dequeue(&ctx->bh_list, &flags))) {

/* qemu_bh_delete() must have been called on BHs in this AioContext */

- assert(ctx->first_bh->deleted);

+ assert(flags & BH_DELETED);

- g_free(ctx->first_bh);

- ctx->first_bh = next;

+ g_free(bh);

}

- qemu_lockcnt_unlock(&ctx->list_lock);

aio_set_event_notifier(ctx, &ctx->notifier, false, NULL, NULL);

event_notifier_cleanup(&ctx->notifier);

@@ -XXX,XX +XXX,XX @@ AioContext *aio_context_new(Error **errp)

AioContext *ctx;

ctx = (AioContext *) g_source_new(&aio_source_funcs, sizeof(AioContext));

+ QSLIST_INIT(&ctx->bh_list);

+ QSIMPLEQ_INIT(&ctx->bh_slice_list);

aio_context_setup(ctx);

ret = event_notifier_init(&ctx->notifier, false);

--

2.24.1

It is not necessary to scan all AioHandlers for deletion. Keep a list

of deleted handlers instead of scanning the full list of all handlers.

The AioHandler->deleted field can be dropped. Let's check if the

handler has been inserted into the deleted list instead. Add a new

QLIST_IS_INSERTED() API for this check.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>

Reviewed-by: Sergio Lopez <slp@redhat.com>

Message-id: 20200214171712.541358-5-stefanha@redhat.com

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>

---

include/block/aio.h | 6 ++++-

include/qemu/queue.h | 3 +++

util/aio-posix.c | 53 +++++++++++++++++++++++++++++---------------

3 files changed, 43 insertions(+), 19 deletions(-)

diff --git a/include/block/aio.h b/include/block/aio.h

index XXXXXXX..XXXXXXX 100644

--- a/include/block/aio.h

+++ b/include/block/aio.h

@@ -XXX,XX +XXX,XX @@ void qemu_aio_unref(void *p);

void qemu_aio_ref(void *p);

typedef struct AioHandler AioHandler;

+typedef QLIST_HEAD(, AioHandler) AioHandlerList;

typedef void QEMUBHFunc(void *opaque);

typedef bool AioPollFn(void *opaque);

typedef void IOHandler(void *opaque);

@@ -XXX,XX +XXX,XX @@ struct AioContext {

QemuRecMutex lock;

/* The list of registered AIO handlers. Protected by ctx->list_lock. */

- QLIST_HEAD(, AioHandler) aio_handlers;

+ AioHandlerList aio_handlers;

+

+ /* The list of AIO handlers to be deleted. Protected by ctx->list_lock. */

+ AioHandlerList deleted_aio_handlers;

/* Used to avoid unnecessary event_notifier_set calls in aio_notify;

* accessed with atomic primitives. If this field is 0, everything

diff --git a/include/qemu/queue.h b/include/qemu/queue.h

index XXXXXXX..XXXXXXX 100644

--- a/include/qemu/queue.h

+++ b/include/qemu/queue.h

@@ -XXX,XX +XXX,XX @@ struct { \

} \

} while (/*CONSTCOND*/0)

+/* Is elm in a list? */

+#define QLIST_IS_INSERTED(elm, field) ((elm)->field.le_prev != NULL)

+

#define QLIST_FOREACH(var, head, field) \

for ((var) = ((head)->lh_first); \

(var); \

diff --git a/util/aio-posix.c b/util/aio-posix.c

index XXXXXXX..XXXXXXX 100644

--- a/util/aio-posix.c

+++ b/util/aio-posix.c

@@ -XXX,XX +XXX,XX @@ struct AioHandler

AioPollFn *io_poll;

IOHandler *io_poll_begin;

IOHandler *io_poll_end;

- int deleted;

void *opaque;

bool is_external;

QLIST_ENTRY(AioHandler) node;

+ QLIST_ENTRY(AioHandler) node_deleted;

};

#ifdef CONFIG_EPOLL_CREATE1

@@ -XXX,XX +XXX,XX @@ static bool aio_epoll_try_enable(AioContext *ctx)

QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {

int r;

- if (node->deleted || !node->pfd.events) {

+ if (QLIST_IS_INSERTED(node, node_deleted) || !node->pfd.events) {

continue;

}

event.events = epoll_events_from_pfd(node->pfd.events);

@@ -XXX,XX +XXX,XX @@ static AioHandler *find_aio_handler(AioContext *ctx, int fd)

AioHandler *node;

QLIST_FOREACH(node, &ctx->aio_handlers, node) {

- if (node->pfd.fd == fd)

- if (!node->deleted)

+ if (node->pfd.fd == fd) {

+ if (!QLIST_IS_INSERTED(node, node_deleted)) {

return node;

+ }

+ }

}

return NULL;

@@ -XXX,XX +XXX,XX @@ static bool aio_remove_fd_handler(AioContext *ctx, AioHandler *node)

/* If a read is in progress, just mark the node as deleted */

if (qemu_lockcnt_count(&ctx->list_lock)) {

- node->deleted = 1;

+ QLIST_INSERT_HEAD_RCU(&ctx->deleted_aio_handlers, node, node_deleted);

node->pfd.revents = 0;

return false;

}

@@ -XXX,XX +XXX,XX @@ static void poll_set_started(AioContext *ctx, bool started)

QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {

IOHandler *fn;

- if (node->deleted) {

+ if (QLIST_IS_INSERTED(node, node_deleted)) {

continue;

}

@@ -XXX,XX +XXX,XX @@ bool aio_pending(AioContext *ctx)

return result;

}

+static void aio_free_deleted_handlers(AioContext *ctx)

+{

+ AioHandler *node;

+

+ if (QLIST_EMPTY_RCU(&ctx->deleted_aio_handlers)) {

+ return;

+ }

+ if (!qemu_lockcnt_dec_if_lock(&ctx->list_lock)) {

+ return; /* we are nested, let the parent do the freeing */

+ }

+

+ while ((node = QLIST_FIRST_RCU(&ctx->deleted_aio_handlers))) {

+ QLIST_REMOVE(node, node);

+ QLIST_REMOVE(node, node_deleted);

+ g_free(node);

+ }

+

+ qemu_lockcnt_inc_and_unlock(&ctx->list_lock);

+}

+

static bool aio_dispatch_handlers(AioContext *ctx)

{

AioHandler *node, *tmp;

@@ -XXX,XX +XXX,XX @@ static bool aio_dispatch_handlers(AioContext *ctx)

revents = node->pfd.revents & node->pfd.events;

node->pfd.revents = 0;

- if (!node->deleted &&

+ if (!QLIST_IS_INSERTED(node, node_deleted) &&

(revents & (G_IO_IN | G_IO_HUP | G_IO_ERR)) &&

aio_node_check(ctx, node->is_external) &&

node->io_read) {

@@ -XXX,XX +XXX,XX @@ static bool aio_dispatch_handlers(AioContext *ctx)

progress = true;

}

}

- if (!node->deleted &&

+ if (!QLIST_IS_INSERTED(node, node_deleted) &&

(revents & (G_IO_OUT | G_IO_ERR)) &&

aio_node_check(ctx, node->is_external) &&

node->io_write) {

node->io_write(node->opaque);

progress = true;

}

-

- if (node->deleted) {

- if (qemu_lockcnt_dec_if_lock(&ctx->list_lock)) {

- QLIST_REMOVE(node, node);

- g_free(node);

- qemu_lockcnt_inc_and_unlock(&ctx->list_lock);

- }

- }

}

return progress;

@@ -XXX,XX +XXX,XX @@ void aio_dispatch(AioContext *ctx)

qemu_lockcnt_inc(&ctx->list_lock);

aio_bh_poll(ctx);

aio_dispatch_handlers(ctx);

+ aio_free_deleted_handlers(ctx);

qemu_lockcnt_dec(&ctx->list_lock);

timerlistgroup_run_timers(&ctx->tlg);

@@ -XXX,XX +XXX,XX @@ static bool run_poll_handlers_once(AioContext *ctx, int64_t *timeout)

RCU_READ_LOCK_GUARD();

QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {

- if (!node->deleted && node->io_poll &&

+ if (!QLIST_IS_INSERTED(node, node_deleted) && node->io_poll &&

aio_node_check(ctx, node->is_external) &&

node->io_poll(node->opaque)) {

/*

@@ -XXX,XX +XXX,XX @@ bool aio_poll(AioContext *ctx, bool blocking)

if (!aio_epoll_enabled(ctx)) {

QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {

- if (!node->deleted && node->pfd.events

+ if (!QLIST_IS_INSERTED(node, node_deleted) && node->pfd.events

&& aio_node_check(ctx, node->is_external)) {

add_pollfd(node);

}

@@ -XXX,XX +XXX,XX @@ bool aio_poll(AioContext *ctx, bool blocking)

progress |= aio_dispatch_handlers(ctx);

}

+ aio_free_deleted_handlers(ctx);

+

qemu_lockcnt_dec(&ctx->list_lock);

progress |= timerlistgroup_run_timers(&ctx->tlg);

--

2.24.1

File descriptor monitoring is O(1) with epoll(7), but

aio_dispatch_handlers() still scans all AioHandlers instead of

dispatching just those that are ready. This makes aio_poll() O(n) with

respect to the total number of registered handlers.

Add a local ready_list to aio_poll() so that each nested aio_poll()

builds a list of handlers ready to be dispatched. Since file descriptor

polling is level-triggered, nested aio_poll() calls also see fds that

were ready in the parent but not yet dispatched. This guarantees that

nested aio_poll() invocations will dispatch all fds, even those that

became ready before the nested invocation.

Since only handlers ready to be dispatched are placed onto the

ready_list, the new aio_dispatch_ready_handlers() function provides O(1)

dispatch.

Note that AioContext polling is still O(n) and currently cannot be fully

disabled. This still needs to be fixed before aio_poll() is fully O(1).

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>

Reviewed-by: Sergio Lopez <slp@redhat.com>

Message-id: 20200214171712.541358-6-stefanha@redhat.com

[Fix compilation error on macOS where there is no epoll(87). The

aio_epoll() prototype was out of date and aio_add_ready_list() needed to

be moved outside the ifdef.

--Stefan]

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>

---

util/aio-posix.c | 110 +++++++++++++++++++++++++++++++++--------------

1 file changed, 78 insertions(+), 32 deletions(-)

diff --git a/util/aio-posix.c b/util/aio-posix.c

index XXXXXXX..XXXXXXX 100644

--- a/util/aio-posix.c

+++ b/util/aio-posix.c

@@ -XXX,XX +XXX,XX @@ struct AioHandler