All that CoQueue needs in order to become thread-safe is help
from an external mutex. Add this to the API.
Signed-off-by: Paolo Bonzini <pbonz...@redhat.com>
---
block/backup.c | 2 +-
block/io.c | 2 +-
block/qcow2-cluster.c | 4 +---
block/sheepdog.c | 2 +-
block/throttle-groups.c | 2 +-
hw/9pfs/9p.c | 2 +-
include/qemu/coroutine.h | 8 +++++---
util/qemu-coroutine-lock.c | 24 +++++++++++++++++++++---
8 files changed, 32 insertions(+), 14 deletions(-)
diff --git a/block/backup.c b/block/backup.c
index 11949f1..db0231f 100644
--- a/block/backup.c
+++ b/block/backup.c
@@ -68,7 +68,7 @@ static void coroutine_fn
wait_for_overlapping_requests(BackupBlockJob *job,
retry = false;
QLIST_FOREACH(req, &job->inflight_reqs, list) {
if (end > req->start && start < req->end) {
- qemu_co_queue_wait(&req->wait_queue);
+ qemu_co_queue_wait(&req->wait_queue, NULL);
retry = true;
break;
}
diff --git a/block/io.c b/block/io.c
index c6ea980..279d9dc 100644
--- a/block/io.c
+++ b/block/io.c
@@ -541,7 +541,7 @@ static bool coroutine_fn
wait_serialising_requests(BdrvTrackedRequest *self)
* (instead of producing a deadlock in the former case). */
if (!req->waiting_for) {
self->waiting_for = req;
- qemu_co_queue_wait(&req->wait_queue);
+ qemu_co_queue_wait(&req->wait_queue, NULL);
self->waiting_for = NULL;
retry = true;
waited = true;
diff --git a/block/qcow2-cluster.c b/block/qcow2-cluster.c
index 31ecc10..42bab15 100644
--- a/block/qcow2-cluster.c
+++ b/block/qcow2-cluster.c
@@ -927,9 +927,7 @@ static int handle_dependencies(BlockDriverState *bs,
uint64_t guest_offset,
if (bytes == 0) {
/* Wait for the dependency to complete. We need to recheck
* the free/allocated clusters when we continue. */
- qemu_co_mutex_unlock(&s->lock);
- qemu_co_queue_wait(&old_alloc->dependent_requests);
- qemu_co_mutex_lock(&s->lock);
+ qemu_co_queue_wait(&old_alloc->dependent_requests, &s->lock);
return -EAGAIN;
}
}
diff --git a/block/sheepdog.c b/block/sheepdog.c
index 790541f..753ae59 100644
--- a/block/sheepdog.c
+++ b/block/sheepdog.c
@@ -2165,7 +2165,7 @@ static void wait_for_overlapping_aiocb(BDRVSheepdogState
*s, SheepdogAIOCB *aioc
retry:
QLIST_FOREACH(cb, &s->inflight_aiocb_head, aiocb_siblings) {
if (AIOCBOverlapping(aiocb, cb)) {
- qemu_co_queue_wait(&s->overlapping_queue);
+ qemu_co_queue_wait(&s->overlapping_queue, NULL);
goto retry;
}
}
diff --git a/block/throttle-groups.c b/block/throttle-groups.c
index 53e910e..5630606 100644
--- a/block/throttle-groups.c
+++ b/block/throttle-groups.c
@@ -302,7 +302,7 @@ void coroutine_fn
throttle_group_co_io_limits_intercept(BlockDriverState *bs,
if (must_wait || bs->pending_reqs[is_write]) {
bs->pending_reqs[is_write]++;
qemu_mutex_unlock(&tg->lock);
- qemu_co_queue_wait(&bs->throttled_reqs[is_write]);
+ qemu_co_queue_wait(&bs->throttled_reqs[is_write], NULL);
qemu_mutex_lock(&tg->lock);
bs->pending_reqs[is_write]--;
}
diff --git a/hw/9pfs/9p.c b/hw/9pfs/9p.c
index f5e3012..5f077e8 100644
--- a/hw/9pfs/9p.c
+++ b/hw/9pfs/9p.c
@@ -2282,7 +2282,7 @@ static void v9fs_flush(void *opaque)
/*
* Wait for pdu to complete.
*/
- qemu_co_queue_wait(&cancel_pdu->complete);
+ qemu_co_queue_wait(&cancel_pdu->complete, NULL);
cancel_pdu->cancelled = 0;
pdu_free(cancel_pdu);
}
diff --git a/include/qemu/coroutine.h b/include/qemu/coroutine.h
index e8e3431..25e31a1 100644
--- a/include/qemu/coroutine.h
+++ b/include/qemu/coroutine.h
@@ -159,7 +159,8 @@ void coroutine_fn qemu_co_mutex_unlock(CoMutex *mutex);
/**
* CoQueues are a mechanism to queue coroutines in order to continue executing
- * them later.
+ * them later. They are similar to condition variables, but they need help
+ * from an external mutex in order to maintain thread-safety.
*/
typedef struct CoQueue {
QSIMPLEQ_HEAD(, Coroutine) entries;
@@ -173,9 +174,10 @@ void qemu_co_queue_init(CoQueue *queue);
/**
* Adds the current coroutine to the CoQueue and transfers control to the
- * caller of the coroutine.
+ * caller of the coroutine. The mutex is unlocked during the wait and
+ * locked again afterwards.
*/
-void coroutine_fn qemu_co_queue_wait(CoQueue *queue);
+void coroutine_fn qemu_co_queue_wait(CoQueue *queue, CoMutex *mutex);
/**
* Restarts the next coroutine in the CoQueue and removes it from the queue.
diff --git a/util/qemu-coroutine-lock.c b/util/qemu-coroutine-lock.c
index aa59e82..828d79a 100644
--- a/util/qemu-coroutine-lock.c
+++ b/util/qemu-coroutine-lock.c
@@ -46,13 +46,31 @@ void qemu_co_queue_init(CoQueue *queue)
QSIMPLEQ_INIT(&queue->entries);
}
-void coroutine_fn qemu_co_queue_wait(CoQueue *queue)
+void coroutine_fn qemu_co_queue_wait(CoQueue *queue, CoMutex *mutex)
{
Coroutine *self = qemu_coroutine_self();
self->ctx = qemu_get_current_aio_context();
QSIMPLEQ_INSERT_TAIL(&queue->entries, self, co_queue_next);
+
+ if (mutex) {
+ qemu_co_mutex_unlock(mutex);
+ }
+
+ /* There is no race condition here. Other threads will call
+ * aio_co_schedule on our AioContext, which can reenter this
+ * coroutine but only after this yield and after the main loop
+ * has gone through the next iteration.
+ */
qemu_coroutine_yield();
assert(qemu_in_coroutine());
+
+ /* TODO: OSv implements wait morphing here, where the wakeup
+ * primitive automatically places the woken coroutine on the
+ * mutex's queue. This avoids the thundering herd effect.
+ */
+ if (mutex) {
+ qemu_co_mutex_lock(mutex);
+ }
}
/**
@@ -309,7 +327,7 @@ void qemu_co_rwlock_init(CoRwlock *lock)
void qemu_co_rwlock_rdlock(CoRwlock *lock)
{
while (lock->writer) {
- qemu_co_queue_wait(&lock->queue);
+ qemu_co_queue_wait(&lock->queue, NULL);
}
lock->reader++;
}
@@ -333,7 +351,7 @@ void qemu_co_rwlock_unlock(CoRwlock *lock)
void qemu_co_rwlock_wrlock(CoRwlock *lock)
{
while (lock->writer || lock->reader) {
- qemu_co_queue_wait(&lock->queue);
+ qemu_co_queue_wait(&lock->queue, NULL);
}
lock->writer = true;
}
--
2.5.5
