The following changes since commit 56f9e46b841c7be478ca038d8d4085d776ab4b0d:

Merge remote-tracking branch 'remotes/armbru/tags/pull-qapi-2017-02-20' into staging (2017-02-20 17:42:47 +0000)

are available in the git repository at:

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

for you to fetch changes up to a7b91d35bab97a2d3e779d0c64c9b837b52a6cf7:

coroutine-lock: make CoRwlock thread-safe and fair (2017-02-21 11:39:40 +0000)

Pull request

v2:

* Rebased to resolve scsi conflicts

Paolo Bonzini (24):

block: move AioContext, QEMUTimer, main-loop to libqemuutil

aio: introduce aio_co_schedule and aio_co_wake

block-backend: allow blk_prw from coroutine context

test-thread-pool: use generic AioContext infrastructure

io: add methods to set I/O handlers on AioContext

io: make qio_channel_yield aware of AioContexts

nbd: convert to use qio_channel_yield

coroutine-lock: reschedule coroutine on the AioContext it was running

on

blkdebug: reschedule coroutine on the AioContext it is running on

qed: introduce qed_aio_start_io and qed_aio_next_io_cb

aio: push aio_context_acquire/release down to dispatching

block: explicitly acquire aiocontext in timers that need it

block: explicitly acquire aiocontext in callbacks that need it

block: explicitly acquire aiocontext in bottom halves that need it

block: explicitly acquire aiocontext in aio callbacks that need it

aio-posix: partially inline aio_dispatch into aio_poll

async: remove unnecessary inc/dec pairs

block: document fields protected by AioContext lock

coroutine-lock: make CoMutex thread-safe

coroutine-lock: add limited spinning to CoMutex

test-aio-multithread: add performance comparison with thread-based

mutexes

coroutine-lock: place CoMutex before CoQueue in header

coroutine-lock: add mutex argument to CoQueue APIs

coroutine-lock: make CoRwlock thread-safe and fair

Makefile.objs | 4 -

stubs/Makefile.objs | 1 +

tests/Makefile.include | 19 +-

util/Makefile.objs | 6 +-

block/nbd-client.h | 2 +-

block/qed.h | 3 +

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

include/block/block_int.h | 64 +++--

include/io/channel.h | 72 +++++-

include/qemu/coroutine.h | 84 ++++---

include/qemu/coroutine_int.h | 11 +-

include/sysemu/block-backend.h | 14 +-

tests/iothread.h | 25 ++

block/backup.c | 2 +-

block/blkdebug.c | 9 +-

block/blkreplay.c | 2 +-

block/block-backend.c | 13 +-

block/curl.c | 44 +++-

block/gluster.c | 9 +-

block/io.c | 42 +---

block/iscsi.c | 15 +-

block/linux-aio.c | 10 +-

block/mirror.c | 12 +-

block/nbd-client.c | 119 +++++----

block/nfs.c | 9 +-

block/qcow2-cluster.c | 4 +-

block/qed-cluster.c | 2 +

block/qed-table.c | 12 +-

block/qed.c | 58 +++--

block/sheepdog.c | 31 +--

block/ssh.c | 29 +--

block/throttle-groups.c | 4 +-

block/win32-aio.c | 9 +-

dma-helpers.c | 2 +

hw/9pfs/9p.c | 2 +-

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

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

hw/scsi/scsi-disk.c | 15 ++

hw/scsi/scsi-generic.c | 20 +-

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

io/channel-command.c | 13 +

io/channel-file.c | 11 +

io/channel-socket.c | 16 +-

io/channel-tls.c | 12 +

io/channel-watch.c | 6 +

io/channel.c | 97 ++++++--

nbd/client.c | 2 +-

nbd/common.c | 9 +-

nbd/server.c | 94 +++-----

stubs/linux-aio.c | 32 +++

stubs/set-fd-handler.c | 11 -

tests/iothread.c | 91 +++++++

tests/test-aio-multithread.c | 463 ++++++++++++++++++++++++++++++++++++

tests/test-thread-pool.c | 12 +-

aio-posix.c => util/aio-posix.c | 62 ++---

aio-win32.c => util/aio-win32.c | 30 +--

util/aiocb.c | 55 +++++

async.c => util/async.c | 84 ++++++-

iohandler.c => util/iohandler.c | 0

main-loop.c => util/main-loop.c | 0

util/qemu-coroutine-lock.c | 254 ++++++++++++++++++--

util/qemu-coroutine-sleep.c | 2 +-

util/qemu-coroutine.c | 8 +

qemu-timer.c => util/qemu-timer.c | 0

thread-pool.c => util/thread-pool.c | 8 +-

trace-events | 11 -

util/trace-events | 17 +-

67 files changed, 1712 insertions(+), 533 deletions(-)

create mode 100644 tests/iothread.h

create mode 100644 stubs/linux-aio.c

create mode 100644 tests/iothread.c

create mode 100644 tests/test-aio-multithread.c

rename aio-posix.c => util/aio-posix.c (94%)

rename aio-win32.c => util/aio-win32.c (95%)

create mode 100644 util/aiocb.c

rename async.c => util/async.c (82%)

rename iohandler.c => util/iohandler.c (100%)

rename main-loop.c => util/main-loop.c (100%)

rename qemu-timer.c => util/qemu-timer.c (100%)

rename thread-pool.c => util/thread-pool.c (97%)

2.9.3

From: Paolo Bonzini <pbonzini@redhat.com>

AioContext is fairly self contained, the only dependency is QEMUTimer but

that in turn doesn't need anything else. So move them out of block-obj-y

to avoid introducing a dependency from io/ to block-obj-y.

main-loop and its dependency iohandler also need to be moved, because

later in this series io/ will call iohandler_get_aio_context.

[Changed copyright "the QEMU team" to "other QEMU contributors" as

suggested by Daniel Berrange and agreed by Paolo.

--Stefan]

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

Reviewed-by: Fam Zheng <famz@redhat.com>

Message-id: 20170213135235.12274-2-pbonzini@redhat.com

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

---

Makefile.objs | 4 ---

stubs/Makefile.objs | 1 +

tests/Makefile.include | 11 ++++----

util/Makefile.objs | 6 +++-

block/io.c | 29 -------------------

stubs/linux-aio.c | 32 +++++++++++++++++++++

stubs/set-fd-handler.c | 11 --------

aio-posix.c => util/aio-posix.c | 2 +-

aio-win32.c => util/aio-win32.c | 0

util/aiocb.c | 55 +++++++++++++++++++++++++++++++++++++

async.c => util/async.c | 3 +-

iohandler.c => util/iohandler.c | 0

main-loop.c => util/main-loop.c | 0

qemu-timer.c => util/qemu-timer.c | 0

thread-pool.c => util/thread-pool.c | 2 +-

trace-events | 11 --------

util/trace-events | 11 ++++++++

17 files changed, 114 insertions(+), 64 deletions(-)

create mode 100644 stubs/linux-aio.c

rename aio-posix.c => util/aio-posix.c (99%)

rename aio-win32.c => util/aio-win32.c (100%)

create mode 100644 util/aiocb.c

rename async.c => util/async.c (99%)

rename iohandler.c => util/iohandler.c (100%)

rename main-loop.c => util/main-loop.c (100%)

rename qemu-timer.c => util/qemu-timer.c (100%)

rename thread-pool.c => util/thread-pool.c (99%)

diff --git a/Makefile.objs b/Makefile.objs

index XXXXXXX..XXXXXXX 100644

--- a/Makefile.objs

+++ b/Makefile.objs

@@ -XXX,XX +XXX,XX @@ chardev-obj-y = chardev/

#######################################################################

# block-obj-y is code used by both qemu system emulation and qemu-img

-block-obj-y = async.o thread-pool.o

block-obj-y += nbd/

block-obj-y += block.o blockjob.o

-block-obj-y += main-loop.o iohandler.o qemu-timer.o

-block-obj-$(CONFIG_POSIX) += aio-posix.o

-block-obj-$(CONFIG_WIN32) += aio-win32.o

block-obj-y += block/

block-obj-y += qemu-io-cmds.o

block-obj-$(CONFIG_REPLICATION) += replication.o

diff --git a/stubs/Makefile.objs b/stubs/Makefile.objs

index XXXXXXX..XXXXXXX 100644

--- a/stubs/Makefile.objs

+++ b/stubs/Makefile.objs

@@ -XXX,XX +XXX,XX @@ stub-obj-y += get-vm-name.o

stub-obj-y += iothread.o

stub-obj-y += iothread-lock.o

stub-obj-y += is-daemonized.o

+stub-obj-$(CONFIG_LINUX_AIO) += linux-aio.o

stub-obj-y += machine-init-done.o

stub-obj-y += migr-blocker.o

stub-obj-y += monitor.o

diff --git a/tests/Makefile.include b/tests/Makefile.include

index XXXXXXX..XXXXXXX 100644

--- a/tests/Makefile.include

+++ b/tests/Makefile.include

@@ -XXX,XX +XXX,XX @@ check-unit-y += tests/test-visitor-serialization$(EXESUF)

check-unit-y += tests/test-iov$(EXESUF)

gcov-files-test-iov-y = util/iov.c

check-unit-y += tests/test-aio$(EXESUF)

+gcov-files-test-aio-y = util/async.c util/qemu-timer.o

+gcov-files-test-aio-$(CONFIG_WIN32) += util/aio-win32.c

+gcov-files-test-aio-$(CONFIG_POSIX) += util/aio-posix.c

check-unit-y += tests/test-throttle$(EXESUF)

gcov-files-test-aio-$(CONFIG_WIN32) = aio-win32.c

gcov-files-test-aio-$(CONFIG_POSIX) = aio-posix.c

@@ -XXX,XX +XXX,XX @@ tests/check-qjson$(EXESUF): tests/check-qjson.o $(test-util-obj-y)

tests/check-qom-interface$(EXESUF): tests/check-qom-interface.o $(test-qom-obj-y)

tests/check-qom-proplist$(EXESUF): tests/check-qom-proplist.o $(test-qom-obj-y)

-tests/test-char$(EXESUF): tests/test-char.o qemu-timer.o \

-    $(test-util-obj-y) $(qtest-obj-y) $(test-block-obj-y) $(chardev-obj-y)

+tests/test-char$(EXESUF): tests/test-char.o $(test-util-obj-y) $(qtest-obj-y) $(test-io-obj-y) $(chardev-obj-y)

tests/test-coroutine$(EXESUF): tests/test-coroutine.o $(test-block-obj-y)

tests/test-aio$(EXESUF): tests/test-aio.o $(test-block-obj-y)

tests/test-throttle$(EXESUF): tests/test-throttle.o $(test-block-obj-y)

@@ -XXX,XX +XXX,XX @@ tests/test-vmstate$(EXESUF): tests/test-vmstate.o \

    migration/vmstate.o migration/qemu-file.o \

migration/qemu-file-channel.o migration/qjson.o \

    $(test-io-obj-y)

-tests/test-timed-average$(EXESUF): tests/test-timed-average.o qemu-timer.o \

-    $(test-util-obj-y)

+tests/test-timed-average$(EXESUF): tests/test-timed-average.o $(test-util-obj-y)

tests/test-base64$(EXESUF): tests/test-base64.o \

    libqemuutil.a libqemustub.a

tests/ptimer-test$(EXESUF): tests/ptimer-test.o tests/ptimer-test-stubs.o hw/core/ptimer.o libqemustub.a

@@ -XXX,XX +XXX,XX @@ tests/usb-hcd-ehci-test$(EXESUF): tests/usb-hcd-ehci-test.o $(libqos-usb-obj-y)

tests/usb-hcd-xhci-test$(EXESUF): tests/usb-hcd-xhci-test.o $(libqos-usb-obj-y)

tests/pc-cpu-test$(EXESUF): tests/pc-cpu-test.o

tests/postcopy-test$(EXESUF): tests/postcopy-test.o

-tests/vhost-user-test$(EXESUF): tests/vhost-user-test.o qemu-timer.o \

+tests/vhost-user-test$(EXESUF): tests/vhost-user-test.o $(test-util-obj-y) \

    $(qtest-obj-y) $(test-io-obj-y) $(libqos-virtio-obj-y) $(libqos-pc-obj-y) \

    $(chardev-obj-y)

tests/qemu-iotests/socket_scm_helper$(EXESUF): tests/qemu-iotests/socket_scm_helper.o

diff --git a/util/Makefile.objs b/util/Makefile.objs

index XXXXXXX..XXXXXXX 100644

--- a/util/Makefile.objs

+++ b/util/Makefile.objs

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

util-obj-y = osdep.o cutils.o unicode.o qemu-timer-common.o

util-obj-y += bufferiszero.o

util-obj-y += lockcnt.o

+util-obj-y += aiocb.o async.o thread-pool.o qemu-timer.o

+util-obj-y += main-loop.o iohandler.o

+util-obj-$(CONFIG_POSIX) += aio-posix.o

util-obj-$(CONFIG_POSIX) += compatfd.o

util-obj-$(CONFIG_POSIX) += event_notifier-posix.o

util-obj-$(CONFIG_POSIX) += mmap-alloc.o

util-obj-$(CONFIG_POSIX) += oslib-posix.o

util-obj-$(CONFIG_POSIX) += qemu-openpty.o

util-obj-$(CONFIG_POSIX) += qemu-thread-posix.o

-util-obj-$(CONFIG_WIN32) += event_notifier-win32.o

util-obj-$(CONFIG_POSIX) += memfd.o

+util-obj-$(CONFIG_WIN32) += aio-win32.o

+util-obj-$(CONFIG_WIN32) += event_notifier-win32.o

util-obj-$(CONFIG_WIN32) += oslib-win32.o

util-obj-$(CONFIG_WIN32) += qemu-thread-win32.o

util-obj-y += envlist.o path.o module.o

diff --git a/block/io.c b/block/io.c

index XXXXXXX..XXXXXXX 100644

--- a/block/io.c

+++ b/block/io.c

@@ -XXX,XX +XXX,XX @@ BlockAIOCB *bdrv_aio_flush(BlockDriverState *bs,

return &acb->common;

}

-void *qemu_aio_get(const AIOCBInfo *aiocb_info, BlockDriverState *bs,

- BlockCompletionFunc *cb, void *opaque)

-{

- BlockAIOCB *acb;

-

- acb = g_malloc(aiocb_info->aiocb_size);

- acb->aiocb_info = aiocb_info;

- acb->bs = bs;

- acb->cb = cb;

- acb->opaque = opaque;

- acb->refcnt = 1;

- return acb;

-}

-

-void qemu_aio_ref(void *p)

-{

- BlockAIOCB *acb = p;

- acb->refcnt++;

-}

-

-void qemu_aio_unref(void *p)

-{

- BlockAIOCB *acb = p;

- assert(acb->refcnt > 0);

- if (--acb->refcnt == 0) {

- g_free(acb);

- }

-}

-

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

/* Coroutine block device emulation */

diff --git a/stubs/linux-aio.c b/stubs/linux-aio.c

new file mode 100644

index XXXXXXX..XXXXXXX

--- /dev/null

+++ b/stubs/linux-aio.c

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

+/*

+ * Linux native AIO support.

+ *

+ * Copyright (C) 2009 IBM, Corp.

+ * Copyright (C) 2009 Red Hat, Inc.

+ *

+ * This work is licensed under the terms of the GNU GPL, version 2 or later.

+ * See the COPYING file in the top-level directory.

+ */

+#include "qemu/osdep.h"

+#include "block/aio.h"

+#include "block/raw-aio.h"

+

+void laio_detach_aio_context(LinuxAioState *s, AioContext *old_context)

+{

+ abort();

+}

+

+void laio_attach_aio_context(LinuxAioState *s, AioContext *new_context)

+{

+ abort();

+}

+

+LinuxAioState *laio_init(void)

+{

+ abort();

+}

+

+void laio_cleanup(LinuxAioState *s)

+{

+ abort();

+}

diff --git a/stubs/set-fd-handler.c b/stubs/set-fd-handler.c

index XXXXXXX..XXXXXXX 100644

--- a/stubs/set-fd-handler.c

+++ b/stubs/set-fd-handler.c

@@ -XXX,XX +XXX,XX @@ void qemu_set_fd_handler(int fd,

{

abort();

}

-

-void aio_set_fd_handler(AioContext *ctx,

- int fd,

- bool is_external,

- IOHandler *io_read,

- IOHandler *io_write,

- AioPollFn *io_poll,

- void *opaque)

-{

- abort();

-}

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

similarity index 99%

rename from aio-posix.c

rename to util/aio-posix.c

index XXXXXXX..XXXXXXX 100644

--- a/aio-posix.c

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

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

#include "qemu/rcu_queue.h"

#include "qemu/sockets.h"

#include "qemu/cutils.h"

-#include "trace-root.h"

+#include "trace.h"

#ifdef CONFIG_EPOLL_CREATE1

#include <sys/epoll.h>

#endif

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

similarity index 100%

rename from aio-win32.c

rename to util/aio-win32.c

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

new file mode 100644

index XXXXXXX..XXXXXXX

--- /dev/null

+++ b/util/aiocb.c

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

+/*

+ * BlockAIOCB allocation

+ *

+ * Copyright (c) 2003-2017 Fabrice Bellard and other QEMU contributors

+ *

+ * Permission is hereby granted, free of charge, to any person obtaining a copy

+ * of this software and associated documentation files (the "Software"), to deal

+ * in the Software without restriction, including without limitation the rights

+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell

+ * copies of the Software, and to permit persons to whom the Software is

+ * furnished to do so, subject to the following conditions:

+ *

+ * The above copyright notice and this permission notice shall be included in

+ * all copies or substantial portions of the Software.

+ *

+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR

+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,

+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL

+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER

+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,

+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN

+ * THE SOFTWARE.

+ */

+

+#include "qemu/osdep.h"

+#include "block/aio.h"

+

+void *qemu_aio_get(const AIOCBInfo *aiocb_info, BlockDriverState *bs,

+ BlockCompletionFunc *cb, void *opaque)

+{

+ BlockAIOCB *acb;

+

+ acb = g_malloc(aiocb_info->aiocb_size);

+ acb->aiocb_info = aiocb_info;

+ acb->bs = bs;

+ acb->cb = cb;

+ acb->opaque = opaque;

+ acb->refcnt = 1;

+ return acb;

+}

+

+void qemu_aio_ref(void *p)

+{

+ BlockAIOCB *acb = p;

+ acb->refcnt++;

+}

+

+void qemu_aio_unref(void *p)

+{

+ BlockAIOCB *acb = p;

+ assert(acb->refcnt > 0);

+ if (--acb->refcnt == 0) {

+ g_free(acb);

+ }

+}

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

similarity index 99%

rename from async.c

rename to util/async.c

index XXXXXXX..XXXXXXX 100644

--- a/async.c

+++ b/util/async.c

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

/*

- * QEMU System Emulator

+ * Data plane event loop

*

* Copyright (c) 2003-2008 Fabrice Bellard

+ * Copyright (c) 2009-2017 QEMU contributors

*

* Permission is hereby granted, free of charge, to any person obtaining a copy

* of this software and associated documentation files (the "Software"), to deal

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

similarity index 100%

rename from iohandler.c

rename to util/iohandler.c

diff --git a/main-loop.c b/util/main-loop.c

similarity index 100%

rename from main-loop.c

rename to util/main-loop.c

diff --git a/qemu-timer.c b/util/qemu-timer.c

similarity index 100%

rename from qemu-timer.c

rename to util/qemu-timer.c

diff --git a/thread-pool.c b/util/thread-pool.c

similarity index 99%

rename from thread-pool.c

rename to util/thread-pool.c

index XXXXXXX..XXXXXXX 100644

--- a/thread-pool.c

+++ b/util/thread-pool.c

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

#include "qemu/queue.h"

#include "qemu/thread.h"

#include "qemu/coroutine.h"

-#include "trace-root.h"

+#include "trace.h"

#include "block/thread-pool.h"

#include "qemu/main-loop.h"

diff --git a/trace-events b/trace-events

index XXXXXXX..XXXXXXX 100644

--- a/trace-events

+++ b/trace-events

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

#

# The <format-string> should be a sprintf()-compatible format string.

-# aio-posix.c

-run_poll_handlers_begin(void *ctx, int64_t max_ns) "ctx %p max_ns %"PRId64

-run_poll_handlers_end(void *ctx, bool progress) "ctx %p progress %d"

-poll_shrink(void *ctx, int64_t old, int64_t new) "ctx %p old %"PRId64" new %"PRId64

-poll_grow(void *ctx, int64_t old, int64_t new) "ctx %p old %"PRId64" new %"PRId64

-

-# thread-pool.c

-thread_pool_submit(void *pool, void *req, void *opaque) "pool %p req %p opaque %p"

-thread_pool_complete(void *pool, void *req, void *opaque, int ret) "pool %p req %p opaque %p ret %d"

-thread_pool_cancel(void *req, void *opaque) "req %p opaque %p"

-

# ioport.c

cpu_in(unsigned int addr, char size, unsigned int val) "addr %#x(%c) value %u"

cpu_out(unsigned int addr, char size, unsigned int val) "addr %#x(%c) value %u"

diff --git a/util/trace-events b/util/trace-events

index XXXXXXX..XXXXXXX 100644

--- a/util/trace-events

+++ b/util/trace-events

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

# See docs/tracing.txt for syntax documentation.

+# util/aio-posix.c

+run_poll_handlers_begin(void *ctx, int64_t max_ns) "ctx %p max_ns %"PRId64

+run_poll_handlers_end(void *ctx, bool progress) "ctx %p progress %d"

+poll_shrink(void *ctx, int64_t old, int64_t new) "ctx %p old %"PRId64" new %"PRId64

+poll_grow(void *ctx, int64_t old, int64_t new) "ctx %p old %"PRId64" new %"PRId64

+

+# util/thread-pool.c

+thread_pool_submit(void *pool, void *req, void *opaque) "pool %p req %p opaque %p"

+thread_pool_complete(void *pool, void *req, void *opaque, int ret) "pool %p req %p opaque %p ret %d"

+thread_pool_cancel(void *req, void *opaque) "req %p opaque %p"

+

# util/buffer.c

buffer_resize(const char *buf, size_t olen, size_t len) "%s: old %zd, new %zd"

buffer_move_empty(const char *buf, size_t len, const char *from) "%s: %zd bytes from %s"

--

2.9.3

From: Paolo Bonzini <pbonzini@redhat.com>

aio_co_wake provides the infrastructure to start a coroutine on a "home"

AioContext. It will be used by CoMutex and CoQueue, so that coroutines

don't jump from one context to another when they go to sleep on a

mutex or waitqueue. However, it can also be used as a more efficient

alternative to one-shot bottom halves, and saves the effort of tracking

which AioContext a coroutine is running on.

aio_co_schedule is the part of aio_co_wake that starts a coroutine

on a remove AioContext, but it is also useful to implement e.g.

bdrv_set_aio_context callbacks.

The implementation of aio_co_schedule is based on a lock-free

multiple-producer, single-consumer queue. The multiple producers use

cmpxchg to add to a LIFO stack. The consumer (a per-AioContext bottom

half) grabs all items added so far, inverts the list to make it FIFO,

and goes through it one item at a time until it's empty. The data

structure was inspired by OSv, which uses it in the very code we'll

"port" to QEMU for the thread-safe CoMutex.

Most of the new code is really tests.

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

Reviewed-by: Fam Zheng <famz@redhat.com>

Message-id: 20170213135235.12274-3-pbonzini@redhat.com

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

---

tests/Makefile.include | 8 +-

include/block/aio.h | 32 +++++++

include/qemu/coroutine_int.h | 11 ++-

tests/iothread.h | 25 +++++

tests/iothread.c | 91 ++++++++++++++++++

tests/test-aio-multithread.c | 213 +++++++++++++++++++++++++++++++++++++++++++

util/async.c | 65 +++++++++++++

util/qemu-coroutine.c | 8 ++

util/trace-events | 4 +

9 files changed, 453 insertions(+), 4 deletions(-)

create mode 100644 tests/iothread.h

create mode 100644 tests/iothread.c

create mode 100644 tests/test-aio-multithread.c

diff --git a/tests/Makefile.include b/tests/Makefile.include

index XXXXXXX..XXXXXXX 100644

--- a/tests/Makefile.include

+++ b/tests/Makefile.include

@@ -XXX,XX +XXX,XX @@ check-unit-y += tests/test-aio$(EXESUF)

gcov-files-test-aio-y = util/async.c util/qemu-timer.o

gcov-files-test-aio-$(CONFIG_WIN32) += util/aio-win32.c

gcov-files-test-aio-$(CONFIG_POSIX) += util/aio-posix.c

+check-unit-y += tests/test-aio-multithread$(EXESUF)

+gcov-files-test-aio-multithread-y = $(gcov-files-test-aio-y)

+gcov-files-test-aio-multithread-y += util/qemu-coroutine.c tests/iothread.c

check-unit-y += tests/test-throttle$(EXESUF)

-gcov-files-test-aio-$(CONFIG_WIN32) = aio-win32.c

-gcov-files-test-aio-$(CONFIG_POSIX) = aio-posix.c

check-unit-y += tests/test-thread-pool$(EXESUF)

gcov-files-test-thread-pool-y = thread-pool.c

gcov-files-test-hbitmap-y = util/hbitmap.c

@@ -XXX,XX +XXX,XX @@ test-qapi-obj-y = tests/test-qapi-visit.o tests/test-qapi-types.o \

    $(test-qom-obj-y)

test-crypto-obj-y = $(crypto-obj-y) $(test-qom-obj-y)

test-io-obj-y = $(io-obj-y) $(test-crypto-obj-y)

-test-block-obj-y = $(block-obj-y) $(test-io-obj-y)

+test-block-obj-y = $(block-obj-y) $(test-io-obj-y) tests/iothread.o

tests/check-qint$(EXESUF): tests/check-qint.o $(test-util-obj-y)

tests/check-qstring$(EXESUF): tests/check-qstring.o $(test-util-obj-y)

@@ -XXX,XX +XXX,XX @@ tests/check-qom-proplist$(EXESUF): tests/check-qom-proplist.o $(test-qom-obj-y)

tests/test-char$(EXESUF): tests/test-char.o $(test-util-obj-y) $(qtest-obj-y) $(test-io-obj-y) $(chardev-obj-y)

tests/test-coroutine$(EXESUF): tests/test-coroutine.o $(test-block-obj-y)

tests/test-aio$(EXESUF): tests/test-aio.o $(test-block-obj-y)

+tests/test-aio-multithread$(EXESUF): tests/test-aio-multithread.o $(test-block-obj-y)

tests/test-throttle$(EXESUF): tests/test-throttle.o $(test-block-obj-y)

tests/test-blockjob$(EXESUF): tests/test-blockjob.o $(test-block-obj-y) $(test-util-obj-y)

tests/test-blockjob-txn$(EXESUF): tests/test-blockjob-txn.o $(test-block-obj-y) $(test-util-obj-y)

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 @@ typedef void QEMUBHFunc(void *opaque);

typedef bool AioPollFn(void *opaque);

typedef void IOHandler(void *opaque);

+struct Coroutine;

struct ThreadPool;

struct LinuxAioState;

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

bool notified;

EventNotifier notifier;

+ QSLIST_HEAD(, Coroutine) scheduled_coroutines;

+ QEMUBH *co_schedule_bh;

+

/* Thread pool for performing work and receiving completion callbacks.

* Has its own locking.

*/

@@ -XXX,XX +XXX,XX @@ static inline bool aio_node_check(AioContext *ctx, bool is_external)

}

/**

+ * aio_co_schedule:

+ * @ctx: the aio context

+ * @co: the coroutine

+ *

+ * Start a coroutine on a remote AioContext.

+ *

+ * The coroutine must not be entered by anyone else while aio_co_schedule()

+ * is active. In addition the coroutine must have yielded unless ctx

+ * is the context in which the coroutine is running (i.e. the value of

+ * qemu_get_current_aio_context() from the coroutine itself).

+ */

+void aio_co_schedule(AioContext *ctx, struct Coroutine *co);

+

+/**

+ * aio_co_wake:

+ * @co: the coroutine

+ *

+ * Restart a coroutine on the AioContext where it was running last, thus

+ * preventing coroutines from jumping from one context to another when they

+ * go to sleep.

+ *

+ * aio_co_wake may be executed either in coroutine or non-coroutine

+ * context. The coroutine must not be entered by anyone else while

+ * aio_co_wake() is active.

+ */

+void aio_co_wake(struct Coroutine *co);

+

+/**

* Return the AioContext whose event loop runs in the current thread.

*

* If called from an IOThread this will be the IOThread's AioContext. If

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

index XXXXXXX..XXXXXXX 100644

--- a/include/qemu/coroutine_int.h

+++ b/include/qemu/coroutine_int.h

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

CoroutineEntry *entry;

void *entry_arg;

Coroutine *caller;

+

+ /* Only used when the coroutine has terminated. */

QSLIST_ENTRY(Coroutine) pool_next;

+

size_t locks_held;

- /* Coroutines that should be woken up when we yield or terminate */

+ /* Coroutines that should be woken up when we yield or terminate.

+ * Only used when the coroutine is running.

+ */

QSIMPLEQ_HEAD(, Coroutine) co_queue_wakeup;

+

+ /* Only used when the coroutine has yielded. */

+ AioContext *ctx;

QSIMPLEQ_ENTRY(Coroutine) co_queue_next;

+ QSLIST_ENTRY(Coroutine) co_scheduled_next;

};

Coroutine *qemu_coroutine_new(void);

diff --git a/tests/iothread.h b/tests/iothread.h

new file mode 100644

index XXXXXXX..XXXXXXX

--- /dev/null

+++ b/tests/iothread.h

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

+/*

+ * Event loop thread implementation for unit tests

+ *

+ * Copyright Red Hat Inc., 2013, 2016

+ *

+ * Authors:

+ * Stefan Hajnoczi <stefanha@redhat.com>

+ * Paolo Bonzini <pbonzini@redhat.com>

+ *

+ * This work is licensed under the terms of the GNU GPL, version 2 or later.

+ * See the COPYING file in the top-level directory.

+ */

+#ifndef TEST_IOTHREAD_H

+#define TEST_IOTHREAD_H

+

+#include "block/aio.h"

+#include "qemu/thread.h"

+

+typedef struct IOThread IOThread;

+

+IOThread *iothread_new(void);

+void iothread_join(IOThread *iothread);

+AioContext *iothread_get_aio_context(IOThread *iothread);

+

+#endif

diff --git a/tests/iothread.c b/tests/iothread.c

new file mode 100644

index XXXXXXX..XXXXXXX

--- /dev/null

+++ b/tests/iothread.c

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

+/*

+ * Event loop thread implementation for unit tests

+ *

+ * Copyright Red Hat Inc., 2013, 2016

+ *

+ * Authors:

+ * Stefan Hajnoczi <stefanha@redhat.com>

+ * Paolo Bonzini <pbonzini@redhat.com>

+ *

+ * This work is licensed under the terms of the GNU GPL, version 2 or later.

+ * See the COPYING file in the top-level directory.

+ *

+ */

+

+#include "qemu/osdep.h"

+#include "qapi/error.h"

+#include "block/aio.h"

+#include "qemu/main-loop.h"

+#include "qemu/rcu.h"

+#include "iothread.h"

+

+struct IOThread {

+ AioContext *ctx;

+

+ QemuThread thread;

+ QemuMutex init_done_lock;

+ QemuCond init_done_cond; /* is thread initialization done? */

+ bool stopping;

+};

+

+static __thread IOThread *my_iothread;

+

+AioContext *qemu_get_current_aio_context(void)

+{

+ return my_iothread ? my_iothread->ctx : qemu_get_aio_context();

+}

+

+static void *iothread_run(void *opaque)

+{

+ IOThread *iothread = opaque;

+

+ rcu_register_thread();

+

+ my_iothread = iothread;

+ qemu_mutex_lock(&iothread->init_done_lock);

+ iothread->ctx = aio_context_new(&error_abort);

+ qemu_cond_signal(&iothread->init_done_cond);

+ qemu_mutex_unlock(&iothread->init_done_lock);

+

+ while (!atomic_read(&iothread->stopping)) {

+ aio_poll(iothread->ctx, true);

+ }

+

+ rcu_unregister_thread();

+ return NULL;

+}

+

+void iothread_join(IOThread *iothread)

+{

+ iothread->stopping = true;

+ aio_notify(iothread->ctx);

+ qemu_thread_join(&iothread->thread);

+ qemu_cond_destroy(&iothread->init_done_cond);

+ qemu_mutex_destroy(&iothread->init_done_lock);

+ aio_context_unref(iothread->ctx);

+ g_free(iothread);

+}

+

+IOThread *iothread_new(void)

+{

+ IOThread *iothread = g_new0(IOThread, 1);

+

+ qemu_mutex_init(&iothread->init_done_lock);

+ qemu_cond_init(&iothread->init_done_cond);

+ qemu_thread_create(&iothread->thread, NULL, iothread_run,

+ iothread, QEMU_THREAD_JOINABLE);

+

+ /* Wait for initialization to complete */

+ qemu_mutex_lock(&iothread->init_done_lock);

+ while (iothread->ctx == NULL) {

+ qemu_cond_wait(&iothread->init_done_cond,

+ &iothread->init_done_lock);

+ }

+ qemu_mutex_unlock(&iothread->init_done_lock);

+ return iothread;

+}

+

+AioContext *iothread_get_aio_context(IOThread *iothread)

+{

+ return iothread->ctx;

+}

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

new file mode 100644

index XXXXXXX..XXXXXXX

--- /dev/null

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

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

+/*

+ * AioContext multithreading tests

+ *

+ * Copyright Red Hat, Inc. 2016

+ *

+ * Authors:

+ * Paolo Bonzini <pbonzini@redhat.com>

+ *

+ * This work is licensed under the terms of the GNU LGPL, version 2 or later.

+ * See the COPYING.LIB file in the top-level directory.

+ */

+

+#include "qemu/osdep.h"

+#include <glib.h>

+#include "block/aio.h"

+#include "qapi/error.h"

+#include "qemu/coroutine.h"

+#include "qemu/thread.h"

+#include "qemu/error-report.h"

+#include "iothread.h"

+

+/* AioContext management */

+

+#define NUM_CONTEXTS 5

+

+static IOThread *threads[NUM_CONTEXTS];

+static AioContext *ctx[NUM_CONTEXTS];

+static __thread int id = -1;

+

+static QemuEvent done_event;

+

+/* Run a function synchronously on a remote iothread. */

+

+typedef struct CtxRunData {

+ QEMUBHFunc *cb;

+ void *arg;

+} CtxRunData;

+

+static void ctx_run_bh_cb(void *opaque)

+{

+ CtxRunData *data = opaque;

+

+ data->cb(data->arg);

+ qemu_event_set(&done_event);

+}

+

+static void ctx_run(int i, QEMUBHFunc *cb, void *opaque)

+{

+ CtxRunData data = {

+ .cb = cb,

+ .arg = opaque

+ };

+

+ qemu_event_reset(&done_event);

+ aio_bh_schedule_oneshot(ctx[i], ctx_run_bh_cb, &data);

+ qemu_event_wait(&done_event);

+}

+

+/* Starting the iothreads. */

+

+static void set_id_cb(void *opaque)

+{

+ int *i = opaque;

+

+ id = *i;

+}

+

+static void create_aio_contexts(void)

+{

+ int i;

+

+ for (i = 0; i < NUM_CONTEXTS; i++) {

+ threads[i] = iothread_new();

+ ctx[i] = iothread_get_aio_context(threads[i]);

+ }

+

+ qemu_event_init(&done_event, false);

+ for (i = 0; i < NUM_CONTEXTS; i++) {

+ ctx_run(i, set_id_cb, &i);

+ }

+}

+

+/* Stopping the iothreads. */

+

+static void join_aio_contexts(void)

+{

+ int i;

+

+ for (i = 0; i < NUM_CONTEXTS; i++) {

+ aio_context_ref(ctx[i]);

+ }

+ for (i = 0; i < NUM_CONTEXTS; i++) {

+ iothread_join(threads[i]);

+ }

+ for (i = 0; i < NUM_CONTEXTS; i++) {

+ aio_context_unref(ctx[i]);

+ }

+ qemu_event_destroy(&done_event);

+}

+

+/* Basic test for the stuff above. */

+

+static void test_lifecycle(void)

+{

+ create_aio_contexts();

+ join_aio_contexts();

+}

+

+/* aio_co_schedule test. */

+

+static Coroutine *to_schedule[NUM_CONTEXTS];

+

+static bool now_stopping;

+

+static int count_retry;

+static int count_here;

+static int count_other;

+

+static bool schedule_next(int n)

+{

+ Coroutine *co;

+

+ co = atomic_xchg(&to_schedule[n], NULL);

+ if (!co) {

+ atomic_inc(&count_retry);

+ return false;

+ }

+

+ if (n == id) {

+ atomic_inc(&count_here);

+ } else {

+ atomic_inc(&count_other);

+ }

+

+ aio_co_schedule(ctx[n], co);

+ return true;

+}

+

+static void finish_cb(void *opaque)

+{

+ schedule_next(id);

+}

+

+static coroutine_fn void test_multi_co_schedule_entry(void *opaque)

+{

+ g_assert(to_schedule[id] == NULL);

+ atomic_mb_set(&to_schedule[id], qemu_coroutine_self());

+

+ while (!atomic_mb_read(&now_stopping)) {

+ int n;

+

+ n = g_test_rand_int_range(0, NUM_CONTEXTS);

+ schedule_next(n);

+ qemu_coroutine_yield();

+

+ g_assert(to_schedule[id] == NULL);

+ atomic_mb_set(&to_schedule[id], qemu_coroutine_self());

+ }

+}

+

+

+static void test_multi_co_schedule(int seconds)

+{

+ int i;

+

+ count_here = count_other = count_retry = 0;

+ now_stopping = false;

+

+ create_aio_contexts();

+ for (i = 0; i < NUM_CONTEXTS; i++) {

+ Coroutine *co1 = qemu_coroutine_create(test_multi_co_schedule_entry, NULL);

+ aio_co_schedule(ctx[i], co1);

+ }

+

+ g_usleep(seconds * 1000000);

+

+ atomic_mb_set(&now_stopping, true);

+ for (i = 0; i < NUM_CONTEXTS; i++) {

+ ctx_run(i, finish_cb, NULL);

+ to_schedule[i] = NULL;

+ }

+

+ join_aio_contexts();

+ g_test_message("scheduled %d, queued %d, retry %d, total %d\n",

+ count_other, count_here, count_retry,

+ count_here + count_other + count_retry);

+}

+

+static void test_multi_co_schedule_1(void)

+{

+ test_multi_co_schedule(1);

+}

+

+static void test_multi_co_schedule_10(void)

+{

+ test_multi_co_schedule(10);

+}

+

+/* End of tests. */

+

+int main(int argc, char **argv)

+{

+ init_clocks();

+

+ g_test_init(&argc, &argv, NULL);

+ g_test_add_func("/aio/multi/lifecycle", test_lifecycle);

+ if (g_test_quick()) {

+ g_test_add_func("/aio/multi/schedule", test_multi_co_schedule_1);

+ } else {

+ g_test_add_func("/aio/multi/schedule", test_multi_co_schedule_10);

+ }

+ return g_test_run();

+}

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 "qemu/main-loop.h"

#include "qemu/atomic.h"

#include "block/raw-aio.h"

+#include "qemu/coroutine_int.h"

+#include "trace.h"

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

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

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

}

#endif

+ 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) {

@@ -XXX,XX +XXX,XX @@ static bool event_notifier_poll(void *opaque)

return atomic_read(&ctx->notified);

}

+static void co_schedule_bh_cb(void *opaque)

+{

+ AioContext *ctx = opaque;

+ QSLIST_HEAD(, Coroutine) straight, reversed;

+

+ QSLIST_MOVE_ATOMIC(&reversed, &ctx->scheduled_coroutines);

+ QSLIST_INIT(&straight);

+

+ while (!QSLIST_EMPTY(&reversed)) {

+ Coroutine *co = QSLIST_FIRST(&reversed);

+ QSLIST_REMOVE_HEAD(&reversed, co_scheduled_next);

+ QSLIST_INSERT_HEAD(&straight, co, co_scheduled_next);

+ }

+

+ while (!QSLIST_EMPTY(&straight)) {

+ Coroutine *co = QSLIST_FIRST(&straight);

+ QSLIST_REMOVE_HEAD(&straight, co_scheduled_next);

+ trace_aio_co_schedule_bh_cb(ctx, co);

+ qemu_coroutine_enter(co);

+ }

+}

+

AioContext *aio_context_new(Error **errp)

{

int ret;

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

}

g_source_set_can_recurse(&ctx->source, true);

qemu_lockcnt_init(&ctx->list_lock);

+

+ ctx->co_schedule_bh = aio_bh_new(ctx, co_schedule_bh_cb, ctx);

+ QSLIST_INIT(&ctx->scheduled_coroutines);

+

aio_set_event_notifier(ctx, &ctx->notifier,

false,

(EventNotifierHandler *)

@@ -XXX,XX +XXX,XX @@ fail:

return NULL;

}

+void aio_co_schedule(AioContext *ctx, Coroutine *co)

+{

+ trace_aio_co_schedule(ctx, co);

+ QSLIST_INSERT_HEAD_ATOMIC(&ctx->scheduled_coroutines,

+ co, co_scheduled_next);

+ qemu_bh_schedule(ctx->co_schedule_bh);

+}

+

+void aio_co_wake(struct Coroutine *co)

+{

+ AioContext *ctx;

+

+ /* Read coroutine before co->ctx. Matches smp_wmb in

+ * qemu_coroutine_enter.

+ */

+ smp_read_barrier_depends();

+ ctx = atomic_read(&co->ctx);

+

+ if (ctx != qemu_get_current_aio_context()) {

+ aio_co_schedule(ctx, co);

+ return;

+ }

+

+ if (qemu_in_coroutine()) {

+ Coroutine *self = qemu_coroutine_self();

+ assert(self != co);

+ QSIMPLEQ_INSERT_TAIL(&self->co_queue_wakeup, co, co_queue_next);

+ } else {

+ aio_context_acquire(ctx);

+ qemu_coroutine_enter(co);

+ aio_context_release(ctx);

+ }

+}

+

void aio_context_ref(AioContext *ctx)

{

g_source_ref(&ctx->source);

diff --git a/util/qemu-coroutine.c b/util/qemu-coroutine.c

index XXXXXXX..XXXXXXX 100644

--- a/util/qemu-coroutine.c

+++ b/util/qemu-coroutine.c

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

#include "qemu/atomic.h"

#include "qemu/coroutine.h"

#include "qemu/coroutine_int.h"

+#include "block/aio.h"

enum {

POOL_BATCH_SIZE = 64,

@@ -XXX,XX +XXX,XX @@ void qemu_coroutine_enter(Coroutine *co)

}

co->caller = self;

+ co->ctx = qemu_get_current_aio_context();

+

+ /* Store co->ctx before anything that stores co. Matches

+ * barrier in aio_co_wake.

+ */

+ smp_wmb();

+

ret = qemu_coroutine_switch(self, co, COROUTINE_ENTER);

qemu_co_queue_run_restart(co);

diff --git a/util/trace-events b/util/trace-events

index XXXXXXX..XXXXXXX 100644

--- a/util/trace-events

+++ b/util/trace-events

@@ -XXX,XX +XXX,XX @@ run_poll_handlers_end(void *ctx, bool progress) "ctx %p progress %d"

poll_shrink(void *ctx, int64_t old, int64_t new) "ctx %p old %"PRId64" new %"PRId64

poll_grow(void *ctx, int64_t old, int64_t new) "ctx %p old %"PRId64" new %"PRId64

+# util/async.c

+aio_co_schedule(void *ctx, void *co) "ctx %p co %p"

+aio_co_schedule_bh_cb(void *ctx, void *co) "ctx %p co %p"

+

# util/thread-pool.c

thread_pool_submit(void *pool, void *req, void *opaque) "pool %p req %p opaque %p"

thread_pool_complete(void *pool, void *req, void *opaque, int ret) "pool %p req %p opaque %p ret %d"

--

2.9.3

From: Paolo Bonzini <pbonzini@redhat.com>

qcow2_create2 calls this. Do not run a nested event loop, as that

breaks when aio_co_wake tries to queue the coroutine on the co_queue_wakeup

list of the currently running one.

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

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

Reviewed-by: Fam Zheng <famz@redhat.com>

Message-id: 20170213135235.12274-4-pbonzini@redhat.com

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

block/block-backend.c | 12 ++++++++----

1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/block/block-backend.c b/block/block-backend.c

index XXXXXXX..XXXXXXX 100644

--- a/block/block-backend.c

+++ b/block/block-backend.c

@@ -XXX,XX +XXX,XX @@ static int blk_prw(BlockBackend *blk, int64_t offset, uint8_t *buf,

{

QEMUIOVector qiov;

struct iovec iov;

- Coroutine *co;

BlkRwCo rwco;

iov = (struct iovec) {

@@ -XXX,XX +XXX,XX @@ static int blk_prw(BlockBackend *blk, int64_t offset, uint8_t *buf,

.ret = NOT_DONE,

};

- co = qemu_coroutine_create(co_entry, &rwco);

- qemu_coroutine_enter(co);

- BDRV_POLL_WHILE(blk_bs(blk), rwco.ret == NOT_DONE);

+ if (qemu_in_coroutine()) {

+ /* Fast-path if already in coroutine context */

+ co_entry(&rwco);

+ } else {

+ Coroutine *co = qemu_coroutine_create(co_entry, &rwco);

+ qemu_coroutine_enter(co);

+ BDRV_POLL_WHILE(blk_bs(blk), rwco.ret == NOT_DONE);

+ }

return rwco.ret;

}

2.9.3