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>
92 lines
2.2 KiB
C
92 lines
2.2 KiB
C
/*
|
|
* 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;
|
|
}
|