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>

This covers both file descriptor callbacks and polling callbacks,

since they execute related code.

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

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

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

Reviewed-by: Daniel P. Berrange <berrange@redhat.com>

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

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

block/curl.c | 16 +++++++++++++---

block/iscsi.c | 4 ++++

block/linux-aio.c | 4 ++++

block/nfs.c | 6 ++++++

block/sheepdog.c | 29 +++++++++++++++--------------

block/ssh.c | 29 +++++++++--------------------

block/win32-aio.c | 10 ++++++----

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

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

util/aio-posix.c | 7 -------

util/aio-win32.c | 6 ------

11 files changed, 68 insertions(+), 55 deletions(-)

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

--- a/block/curl.c

+++ b/block/curl.c

@@ -XXX,XX +XXX,XX @@ static void curl_multi_check_completion(BDRVCURLState *s)

}

-static void curl_multi_do(void *arg)

+static void curl_multi_do_locked(CURLState *s)

{

- CURLState *s = (CURLState *)arg;

CURLSocket *socket, *next_socket;

int running;

int r;

@@ -XXX,XX +XXX,XX @@ static void curl_multi_do(void *arg)

}

}

+static void curl_multi_do(void *arg)

+ CURLState *s = (CURLState *)arg;

+ aio_context_acquire(s->s->aio_context);

+ curl_multi_do_locked(s);

+ aio_context_release(s->s->aio_context);

static void curl_multi_read(void *arg)

CURLState *s = (CURLState *)arg;

- curl_multi_do(arg);

+ aio_context_acquire(s->s->aio_context);

+ curl_multi_do_locked(s);

curl_multi_check_completion(s->s);

+ aio_context_release(s->s->aio_context);

}

static void curl_multi_timeout_do(void *arg)

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

index XXXXXXX..XXXXXXX 100644

--- a/block/iscsi.c

+++ b/block/iscsi.c

@@ -XXX,XX +XXX,XX @@ iscsi_process_read(void *arg)

IscsiLun *iscsilun = arg;

struct iscsi_context *iscsi = iscsilun->iscsi;

+ aio_context_acquire(iscsilun->aio_context);

iscsi_service(iscsi, POLLIN);

iscsi_set_events(iscsilun);

+ aio_context_release(iscsilun->aio_context);

}

static void

@@ -XXX,XX +XXX,XX @@ iscsi_process_write(void *arg)

IscsiLun *iscsilun = arg;

struct iscsi_context *iscsi = iscsilun->iscsi;

+ aio_context_acquire(iscsilun->aio_context);

iscsi_service(iscsi, POLLOUT);

iscsi_set_events(iscsilun);

+ aio_context_release(iscsilun->aio_context);

}

static int64_t sector_lun2qemu(int64_t sector, IscsiLun *iscsilun)

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

index XXXXXXX..XXXXXXX 100644

--- a/block/linux-aio.c

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

@@ -XXX,XX +XXX,XX @@ static void qemu_laio_completion_cb(EventNotifier *e)

LinuxAioState *s = container_of(e, LinuxAioState, e);

if (event_notifier_test_and_clear(&s->e)) {

+ aio_context_acquire(s->aio_context);

qemu_laio_process_completions_and_submit(s);

+ aio_context_release(s->aio_context);

}

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

return false;

+ aio_context_acquire(s->aio_context);

qemu_laio_process_completions_and_submit(s);

+ aio_context_release(s->aio_context);

return true;

}

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

index XXXXXXX..XXXXXXX 100644

--- a/block/nfs.c

+++ b/block/nfs.c

@@ -XXX,XX +XXX,XX @@ static void nfs_set_events(NFSClient *client)

static void nfs_process_read(void *arg)

{

NFSClient *client = arg;

+

+ aio_context_acquire(client->aio_context);

nfs_service(client->context, POLLIN);

nfs_set_events(client);

+ aio_context_release(client->aio_context);

}

static void nfs_process_write(void *arg)

{

NFSClient *client = arg;

+

+ aio_context_acquire(client->aio_context);

nfs_service(client->context, POLLOUT);

nfs_set_events(client);

+ aio_context_release(client->aio_context);

}

static void nfs_co_init_task(BlockDriverState *bs, NFSRPC *task)

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

index XXXXXXX..XXXXXXX 100644

--- a/block/sheepdog.c

+++ b/block/sheepdog.c

@@ -XXX,XX +XXX,XX @@ static coroutine_fn int send_co_req(int sockfd, SheepdogReq *hdr, void *data,

return ret;

}

-static void restart_co_req(void *opaque)

-{

- Coroutine *co = opaque;

-

- qemu_coroutine_enter(co);

-}

-

typedef struct SheepdogReqCo {

int sockfd;

BlockDriverState *bs;

@@ -XXX,XX +XXX,XX @@ typedef struct SheepdogReqCo {

unsigned int *rlen;

int ret;

bool finished;

+ Coroutine *co;

} SheepdogReqCo;

+static void restart_co_req(void *opaque)

+{

+ SheepdogReqCo *srco = opaque;

+

+ aio_co_wake(srco->co);

+}

+

static coroutine_fn void do_co_req(void *opaque)

{

int ret;

- Coroutine *co;

SheepdogReqCo *srco = opaque;

int sockfd = srco->sockfd;

SheepdogReq *hdr = srco->hdr;

@@ -XXX,XX +XXX,XX @@ static coroutine_fn void do_co_req(void *opaque)

unsigned int *wlen = srco->wlen;

unsigned int *rlen = srco->rlen;

- co = qemu_coroutine_self();

+ srco->co = qemu_coroutine_self();

aio_set_fd_handler(srco->aio_context, sockfd, false,

- NULL, restart_co_req, NULL, co);

+ NULL, restart_co_req, NULL, srco);

ret = send_co_req(sockfd, hdr, data, wlen);

if (ret < 0) {

@@ -XXX,XX +XXX,XX @@ static coroutine_fn void do_co_req(void *opaque)

aio_set_fd_handler(srco->aio_context, sockfd, false,

- restart_co_req, NULL, NULL, co);

+ restart_co_req, NULL, NULL, srco);

ret = qemu_co_recv(sockfd, hdr, sizeof(*hdr));

if (ret != sizeof(*hdr)) {

@@ -XXX,XX +XXX,XX @@ out:

aio_set_fd_handler(srco->aio_context, sockfd, false,

NULL, NULL, NULL, NULL);

+ srco->co = NULL;

srco->ret = ret;

srco->finished = true;

if (srco->bs) {

@@ -XXX,XX +XXX,XX @@ static void coroutine_fn aio_read_response(void *opaque)

* We've finished all requests which belong to the AIOCB, so

* we can switch back to sd_co_readv/writev now.

*/

- qemu_coroutine_enter(acb->coroutine);

+ aio_co_wake(acb->coroutine);

return;

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

s->co_recv = qemu_coroutine_create(aio_read_response, opaque);

}

- qemu_coroutine_enter(s->co_recv);

+ aio_co_wake(s->co_recv);

}

static void co_write_request(void *opaque)

{

BDRVSheepdogState *s = opaque;

- qemu_coroutine_enter(s->co_send);

+ aio_co_wake(s->co_send);

}

/*

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

index XXXXXXX..XXXXXXX 100644

--- a/block/ssh.c

+++ b/block/ssh.c

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

DPRINTF("co=%p", co);

- qemu_coroutine_enter(co);

+ aio_co_wake(co);

}

-static coroutine_fn void set_fd_handler(BDRVSSHState *s, BlockDriverState *bs)

+/* A non-blocking call returned EAGAIN, so yield, ensuring the

+ * handlers are set up so that we'll be rescheduled when there is an

+ * interesting event on the socket.

+ */

+static coroutine_fn void co_yield(BDRVSSHState *s, BlockDriverState *bs)

{

int r;

IOHandler *rd_handler = NULL, *wr_handler = NULL;

@@ -XXX,XX +XXX,XX @@ static coroutine_fn void set_fd_handler(BDRVSSHState *s, BlockDriverState *bs)

aio_set_fd_handler(bdrv_get_aio_context(bs), s->sock,

false, rd_handler, wr_handler, NULL, co);

-}

-

-static coroutine_fn void clear_fd_handler(BDRVSSHState *s,

- BlockDriverState *bs)

-{

- DPRINTF("s->sock=%d", s->sock);

- aio_set_fd_handler(bdrv_get_aio_context(bs), s->sock,

- false, NULL, NULL, NULL, NULL);

-}

-

-/* A non-blocking call returned EAGAIN, so yield, ensuring the

- * handlers are set up so that we'll be rescheduled when there is an

- * interesting event on the socket.

- */

-static coroutine_fn void co_yield(BDRVSSHState *s, BlockDriverState *bs)

-{

- set_fd_handler(s, bs);

qemu_coroutine_yield();

- clear_fd_handler(s, bs);

+ DPRINTF("s->sock=%d - back", s->sock);

+ aio_set_fd_handler(bdrv_get_aio_context(bs), s->sock, false,

+ NULL, NULL, NULL, NULL);

}

/* SFTP has a function `libssh2_sftp_seek64' which seeks to a position

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

index XXXXXXX..XXXXXXX 100644

--- a/block/win32-aio.c

+++ b/block/win32-aio.c

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

HANDLE hIOCP;

EventNotifier e;

int count;

- bool is_aio_context_attached;

+ AioContext *aio_ctx;

};

typedef struct QEMUWin32AIOCB {

@@ -XXX,XX +XXX,XX @@ static void win32_aio_process_completion(QEMUWin32AIOState *s,

}

+ aio_context_acquire(s->aio_ctx);

waiocb->common.cb(waiocb->common.opaque, ret);

+ aio_context_release(s->aio_ctx);

qemu_aio_unref(waiocb);

}

@@ -XXX,XX +XXX,XX @@ void win32_aio_detach_aio_context(QEMUWin32AIOState *aio,

AioContext *old_context)

{

aio_set_event_notifier(old_context, &aio->e, false, NULL, NULL);

- aio->is_aio_context_attached = false;

+ aio->aio_ctx = NULL;

}

void win32_aio_attach_aio_context(QEMUWin32AIOState *aio,

AioContext *new_context)

{

- aio->is_aio_context_attached = true;

+ aio->aio_ctx = new_context;

aio_set_event_notifier(new_context, &aio->e, false,

win32_aio_completion_cb, NULL);

}

@@ -XXX,XX +XXX,XX @@ out_free_state:

void win32_aio_cleanup(QEMUWin32AIOState *aio)

{

- assert(!aio->is_aio_context_attached);

+ assert(!aio->aio_ctx);

CloseHandle(aio->hIOCP);

event_notifier_cleanup(&aio->e);

g_free(aio);

diff --git a/hw/block/virtio-blk.c b/hw/block/virtio-blk.c

index XXXXXXX..XXXXXXX 100644

--- a/hw/block/virtio-blk.c

+++ b/hw/block/virtio-blk.c

@@ -XXX,XX +XXX,XX @@ static void virtio_blk_ioctl_complete(void *opaque, int status)

{

VirtIOBlockIoctlReq *ioctl_req = opaque;

VirtIOBlockReq *req = ioctl_req->req;

- VirtIODevice *vdev = VIRTIO_DEVICE(req->dev);

+ VirtIOBlock *s = req->dev;

+ VirtIODevice *vdev = VIRTIO_DEVICE(s);

struct virtio_scsi_inhdr *scsi;

struct sg_io_hdr *hdr;

@@ -XXX,XX +XXX,XX @@ bool virtio_blk_handle_vq(VirtIOBlock *s, VirtQueue *vq)

MultiReqBuffer mrb = {};

bool progress = false;

+ aio_context_acquire(blk_get_aio_context(s->blk));

blk_io_plug(s->blk);

do {

@@ -XXX,XX +XXX,XX @@ bool virtio_blk_handle_vq(VirtIOBlock *s, VirtQueue *vq)

}

blk_io_unplug(s->blk);

+ aio_context_release(blk_get_aio_context(s->blk));

return progress;

}

diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c

index XXXXXXX..XXXXXXX 100644

--- a/hw/scsi/virtio-scsi.c

+++ b/hw/scsi/virtio-scsi.c

@@ -XXX,XX +XXX,XX @@ bool virtio_scsi_handle_ctrl_vq(VirtIOSCSI *s, VirtQueue *vq)

VirtIOSCSIReq *req;

bool progress = false;

+ virtio_scsi_acquire(s);

while ((req = virtio_scsi_pop_req(s, vq))) {

progress = true;

virtio_scsi_handle_ctrl_req(s, req);

}

+ virtio_scsi_release(s);

return progress;

}

@@ -XXX,XX +XXX,XX @@ bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq)

QTAILQ_HEAD(, VirtIOSCSIReq) reqs = QTAILQ_HEAD_INITIALIZER(reqs);

+ virtio_scsi_acquire(s);

do {

virtio_queue_set_notification(vq, 0);

@@ -XXX,XX +XXX,XX @@ bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq)

QTAILQ_FOREACH_SAFE(req, &reqs, next, next) {

virtio_scsi_handle_cmd_req_submit(s, req);

}

+ virtio_scsi_release(s);

return progress;

}

@@ -XXX,XX +XXX,XX @@ out:

bool virtio_scsi_handle_event_vq(VirtIOSCSI *s, VirtQueue *vq)

{

+ virtio_scsi_acquire(s);

if (s->events_dropped) {

virtio_scsi_push_event(s, NULL, VIRTIO_SCSI_T_NO_EVENT, 0);

+ virtio_scsi_release(s);

return true;

}

+ virtio_scsi_release(s);

return false;

}

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 @@ static bool aio_dispatch_handlers(AioContext *ctx)

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

aio_node_check(ctx, node->is_external) &&

node->io_read) {

- aio_context_acquire(ctx);

node->io_read(node->opaque);

- aio_context_release(ctx);

/* aio_notify() does not count as progress */

if (node->opaque != &ctx->notifier) {

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

(revents & (G_IO_OUT | G_IO_ERR)) &&

aio_node_check(ctx, node->is_external) &&

node->io_write) {

- aio_context_acquire(ctx);

node->io_write(node->opaque);

- aio_context_release(ctx);

progress = true;

}

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

start = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);

}

- aio_context_acquire(ctx);

progress = try_poll_mode(ctx, blocking);

- aio_context_release(ctx);

-

if (!progress) {

assert(npfd == 0);

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

index XXXXXXX..XXXXXXX 100644

--- a/util/aio-win32.c

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

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

(revents || event_notifier_get_handle(node->e) == event) &&

node->io_notify) {

node->pfd.revents = 0;

- aio_context_acquire(ctx);

node->io_notify(node->e);

- aio_context_release(ctx);

/* aio_notify() does not count as progress */

if (node->e != &ctx->notifier) {

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

(node->io_read || node->io_write)) {

node->pfd.revents = 0;

if ((revents & G_IO_IN) && node->io_read) {

- aio_context_acquire(ctx);

node->io_read(node->opaque);

- aio_context_release(ctx);

progress = true;

}

if ((revents & G_IO_OUT) && node->io_write) {

- aio_context_acquire(ctx);

node->io_write(node->opaque);

- aio_context_release(ctx);

progress = true;

}

2.9.3

From: Paolo Bonzini <pbonzini@redhat.com>

Running a very small critical section on pthread_mutex_t and CoMutex

shows that pthread_mutex_t is much faster because it doesn't actually

go to sleep. What happens is that the critical section is shorter

than the latency of entering the kernel and thus FUTEX_WAIT always

fails. With CoMutex there is no such latency but you still want to

avoid wait and wakeup. So introduce it artificially.

This only works with one waiters; because CoMutex is fair, it will

always have more waits and wakeups than a pthread_mutex_t.

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

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

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

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

include/qemu/coroutine.h | 5 +++++

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

util/qemu-coroutine.c | 2 +-

3 files changed, 51 insertions(+), 7 deletions(-)

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

--- a/include/qemu/coroutine.h

+++ b/include/qemu/coroutine.h

@@ -XXX,XX +XXX,XX @@ typedef struct CoMutex {

*/

unsigned locked;

+ /* Context that is holding the lock. Useful to avoid spinning

+ * when two coroutines on the same AioContext try to get the lock. :)

+ */

+ AioContext *ctx;

+

/* A queue of waiters. Elements are added atomically in front of

* from_push. to_pop is only populated, and popped from, by whoever

* is in charge of the next wakeup. This can be an unlocker or,

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

index XXXXXXX..XXXXXXX 100644

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

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

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

#include "qemu-common.h"

#include "qemu/coroutine.h"

#include "qemu/coroutine_int.h"

+#include "qemu/processor.h"

#include "qemu/queue.h"

#include "block/aio.h"

#include "trace.h"

@@ -XXX,XX +XXX,XX @@ void qemu_co_mutex_init(CoMutex *mutex)

memset(mutex, 0, sizeof(*mutex));

}

-static void coroutine_fn qemu_co_mutex_lock_slowpath(CoMutex *mutex)

+static void coroutine_fn qemu_co_mutex_wake(CoMutex *mutex, Coroutine *co)

+{

+ /* Read co before co->ctx; pairs with smp_wmb() in

+ * qemu_coroutine_enter().

+ */

+ smp_read_barrier_depends();

+ mutex->ctx = co->ctx;

+ aio_co_wake(co);

+}

+

+static void coroutine_fn qemu_co_mutex_lock_slowpath(AioContext *ctx,

+ CoMutex *mutex)

Coroutine *self = qemu_coroutine_self();

CoWaitRecord w;

@@ -XXX,XX +XXX,XX @@ static void coroutine_fn qemu_co_mutex_lock_slowpath(CoMutex *mutex)

if (co == self) {

/* We got the lock ourselves! */

assert(to_wake == &w);

+ mutex->ctx = ctx;

return;

}

- aio_co_wake(co);

+ qemu_co_mutex_wake(mutex, co);

qemu_coroutine_yield();

@@ -XXX,XX +XXX,XX @@ static void coroutine_fn qemu_co_mutex_lock_slowpath(CoMutex *mutex)

void coroutine_fn qemu_co_mutex_lock(CoMutex *mutex)

{

+ AioContext *ctx = qemu_get_current_aio_context();

Coroutine *self = qemu_coroutine_self();

+ int waiters, i;

- if (atomic_fetch_inc(&mutex->locked) == 0) {

+ /* Running a very small critical section on pthread_mutex_t and CoMutex

+ * shows that pthread_mutex_t is much faster because it doesn't actually

+ * go to sleep. What happens is that the critical section is shorter

+ * than the latency of entering the kernel and thus FUTEX_WAIT always

+ * fails. With CoMutex there is no such latency but you still want to

+ * avoid wait and wakeup. So introduce it artificially.

+ */

+ i = 0;

+retry_fast_path:

+ waiters = atomic_cmpxchg(&mutex->locked, 0, 1);

+ if (waiters != 0) {

+ while (waiters == 1 && ++i < 1000) {

+ if (atomic_read(&mutex->ctx) == ctx) {

+ break;

+ }

+ if (atomic_read(&mutex->locked) == 0) {

+ goto retry_fast_path;

+ }

+ cpu_relax();

+ }

+ waiters = atomic_fetch_inc(&mutex->locked);

+ }

+

+ if (waiters == 0) {

/* Uncontended. */

trace_qemu_co_mutex_lock_uncontended(mutex, self);

+ mutex->ctx = ctx;

} else {

- qemu_co_mutex_lock_slowpath(mutex);

+ qemu_co_mutex_lock_slowpath(ctx, mutex);

}

mutex->holder = self;

self->locks_held++;

@@ -XXX,XX +XXX,XX @@ void coroutine_fn qemu_co_mutex_unlock(CoMutex *mutex)

assert(mutex->holder == self);

assert(qemu_in_coroutine());

+ mutex->ctx = NULL;

mutex->holder = NULL;

self->locks_held--;

if (atomic_fetch_dec(&mutex->locked) == 1) {

@@ -XXX,XX +XXX,XX @@ void coroutine_fn qemu_co_mutex_unlock(CoMutex *mutex)

unsigned our_handoff;

if (to_wake) {

- Coroutine *co = to_wake->co;

- aio_co_wake(co);

+ qemu_co_mutex_wake(mutex, to_wake->co);

break;

}

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 @@ void qemu_coroutine_enter(Coroutine *co)

co->ctx = qemu_get_current_aio_context();

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

- * barrier in aio_co_wake.

+ * barrier in aio_co_wake and qemu_co_mutex_wake.

*/

smp_wmb();

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.