0c330a734b
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>
169 lines
4.3 KiB
C
169 lines
4.3 KiB
C
/*
|
|
* QEMU coroutines
|
|
*
|
|
* Copyright IBM, Corp. 2011
|
|
*
|
|
* Authors:
|
|
* Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
|
|
* Kevin Wolf <kwolf@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 "trace.h"
|
|
#include "qemu-common.h"
|
|
#include "qemu/thread.h"
|
|
#include "qemu/atomic.h"
|
|
#include "qemu/coroutine.h"
|
|
#include "qemu/coroutine_int.h"
|
|
#include "block/aio.h"
|
|
|
|
enum {
|
|
POOL_BATCH_SIZE = 64,
|
|
};
|
|
|
|
/** Free list to speed up creation */
|
|
static QSLIST_HEAD(, Coroutine) release_pool = QSLIST_HEAD_INITIALIZER(pool);
|
|
static unsigned int release_pool_size;
|
|
static __thread QSLIST_HEAD(, Coroutine) alloc_pool = QSLIST_HEAD_INITIALIZER(pool);
|
|
static __thread unsigned int alloc_pool_size;
|
|
static __thread Notifier coroutine_pool_cleanup_notifier;
|
|
|
|
static void coroutine_pool_cleanup(Notifier *n, void *value)
|
|
{
|
|
Coroutine *co;
|
|
Coroutine *tmp;
|
|
|
|
QSLIST_FOREACH_SAFE(co, &alloc_pool, pool_next, tmp) {
|
|
QSLIST_REMOVE_HEAD(&alloc_pool, pool_next);
|
|
qemu_coroutine_delete(co);
|
|
}
|
|
}
|
|
|
|
Coroutine *qemu_coroutine_create(CoroutineEntry *entry, void *opaque)
|
|
{
|
|
Coroutine *co = NULL;
|
|
|
|
if (CONFIG_COROUTINE_POOL) {
|
|
co = QSLIST_FIRST(&alloc_pool);
|
|
if (!co) {
|
|
if (release_pool_size > POOL_BATCH_SIZE) {
|
|
/* Slow path; a good place to register the destructor, too. */
|
|
if (!coroutine_pool_cleanup_notifier.notify) {
|
|
coroutine_pool_cleanup_notifier.notify = coroutine_pool_cleanup;
|
|
qemu_thread_atexit_add(&coroutine_pool_cleanup_notifier);
|
|
}
|
|
|
|
/* This is not exact; there could be a little skew between
|
|
* release_pool_size and the actual size of release_pool. But
|
|
* it is just a heuristic, it does not need to be perfect.
|
|
*/
|
|
alloc_pool_size = atomic_xchg(&release_pool_size, 0);
|
|
QSLIST_MOVE_ATOMIC(&alloc_pool, &release_pool);
|
|
co = QSLIST_FIRST(&alloc_pool);
|
|
}
|
|
}
|
|
if (co) {
|
|
QSLIST_REMOVE_HEAD(&alloc_pool, pool_next);
|
|
alloc_pool_size--;
|
|
}
|
|
}
|
|
|
|
if (!co) {
|
|
co = qemu_coroutine_new();
|
|
}
|
|
|
|
co->entry = entry;
|
|
co->entry_arg = opaque;
|
|
QSIMPLEQ_INIT(&co->co_queue_wakeup);
|
|
return co;
|
|
}
|
|
|
|
static void coroutine_delete(Coroutine *co)
|
|
{
|
|
co->caller = NULL;
|
|
|
|
if (CONFIG_COROUTINE_POOL) {
|
|
if (release_pool_size < POOL_BATCH_SIZE * 2) {
|
|
QSLIST_INSERT_HEAD_ATOMIC(&release_pool, co, pool_next);
|
|
atomic_inc(&release_pool_size);
|
|
return;
|
|
}
|
|
if (alloc_pool_size < POOL_BATCH_SIZE) {
|
|
QSLIST_INSERT_HEAD(&alloc_pool, co, pool_next);
|
|
alloc_pool_size++;
|
|
return;
|
|
}
|
|
}
|
|
|
|
qemu_coroutine_delete(co);
|
|
}
|
|
|
|
void qemu_coroutine_enter(Coroutine *co)
|
|
{
|
|
Coroutine *self = qemu_coroutine_self();
|
|
CoroutineAction ret;
|
|
|
|
trace_qemu_coroutine_enter(self, co, co->entry_arg);
|
|
|
|
if (co->caller) {
|
|
fprintf(stderr, "Co-routine re-entered recursively\n");
|
|
abort();
|
|
}
|
|
|
|
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);
|
|
|
|
switch (ret) {
|
|
case COROUTINE_YIELD:
|
|
return;
|
|
case COROUTINE_TERMINATE:
|
|
assert(!co->locks_held);
|
|
trace_qemu_coroutine_terminate(co);
|
|
coroutine_delete(co);
|
|
return;
|
|
default:
|
|
abort();
|
|
}
|
|
}
|
|
|
|
void qemu_coroutine_enter_if_inactive(Coroutine *co)
|
|
{
|
|
if (!qemu_coroutine_entered(co)) {
|
|
qemu_coroutine_enter(co);
|
|
}
|
|
}
|
|
|
|
void coroutine_fn qemu_coroutine_yield(void)
|
|
{
|
|
Coroutine *self = qemu_coroutine_self();
|
|
Coroutine *to = self->caller;
|
|
|
|
trace_qemu_coroutine_yield(self, to);
|
|
|
|
if (!to) {
|
|
fprintf(stderr, "Co-routine is yielding to no one\n");
|
|
abort();
|
|
}
|
|
|
|
self->caller = NULL;
|
|
qemu_coroutine_switch(self, to, COROUTINE_YIELD);
|
|
}
|
|
|
|
bool qemu_coroutine_entered(Coroutine *co)
|
|
{
|
|
return co->caller;
|
|
}
|