[Qemu-devel] [PATCH 05/11] coroutine-lock: reschedule coroutine on the AioContext it was running on

[Paolo Bonzini <pbonzini@redhat.com>]

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 include/block/aio.h        |  5 +++++
 include/qemu/coroutine.h   | 18 ++++++++++++++++++
 iothread.c                 | 16 ++++++++++++++++
 stubs/iothread.c           | 11 +++++++++++
 tests/iothread.c           | 16 ++++++++++++++++
 trace-events               |  3 ++-
 util/qemu-coroutine-lock.c | 13 ++++++++++---
 7 files changed, 78 insertions(+), 4 deletions(-)

diff --git a/include/block/aio.h b/include/block/aio.h
index 0a344c3..7fa909d 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -450,6 +450,11 @@ static inline bool aio_node_check(AioContext *ctx, bool is_external)
 void aio_co_schedule(AioContext *ctx, struct Coroutine *co);
 
 /**
+ * Return the AioContext whose event loop runs in the current I/O thread.
+ */
+AioContext *qemu_get_current_aio_context(void);
+
+/**
  * @ctx: the aio context
  *
  * Return whether we are running in the I/O thread that manages @ctx.
diff --git a/include/qemu/coroutine.h b/include/qemu/coroutine.h
index bb23be0..179ab0b 100644
--- a/include/qemu/coroutine.h
+++ b/include/qemu/coroutine.h
@@ -73,6 +73,24 @@ Coroutine *qemu_coroutine_create(CoroutineEntry *entry);
 void qemu_coroutine_enter(Coroutine *coroutine, void *opaque);
 
 /**
+ * Wake up a suspended coroutine in the current thread.
+ *
+ * This function is used by the implementation of qemu_coroutine_wake.
+ * It arranges for @co to restart at the next yield point of the
+ * current coroutine.
+ */
+void qemu_coroutine_queue_next(Coroutine *co);
+
+/**
+ * Wake up a suspended coroutine
+ *
+ * This function is used by the synchronization primitives to wake up
+ * a sleeping coroutine.  For single-threaded scenarios it can simply
+ * call qemu_coroutine_queue_next.
+ */
+void qemu_coroutine_wake(AioContext *ctx, Coroutine *co);
+
+/**
  * Transfer control back to a coroutine's caller
  *
  * This function does not return until the coroutine is re-entered using
diff --git a/iothread.c b/iothread.c
index f66ec95..986d125 100644
--- a/iothread.c
+++ b/iothread.c
@@ -21,6 +21,7 @@
 #include "qemu/error-report.h"
 #include "qemu/rcu.h"
 #include "qemu/main-loop.h"
+#include "qemu/coroutine.h"
 
 typedef ObjectClass IOThreadClass;
 
@@ -31,11 +32,26 @@ typedef ObjectClass IOThreadClass;
 
 static __thread IOThread *my_iothread;
 
+void qemu_get_current_aio_context(void)
+{
+    return my_iothread ? my_iothread->ctx : qemu_get_aio_context();
+}
+
 bool aio_context_in_iothread(AioContext *ctx)
 {
     return ctx == (my_iothread ? my_iothread->ctx : qemu_get_aio_context());
 }
 
+void qemu_coroutine_wake(AioContext *ctx, Coroutine *co)
+{
+    if (ctx != qemu_get_current_aio_context()) {
+        aio_co_schedule(ctx, co);
+        return;
+    }
+
+    qemu_coroutine_queue_next(co);
+}
+
 static void *iothread_run(void *opaque)
 {
     IOThread *iothread = opaque;
diff --git a/stubs/iothread.c b/stubs/iothread.c
index 6c02323..fb3e5e3 100644
--- a/stubs/iothread.c
+++ b/stubs/iothread.c
@@ -1,8 +1,19 @@
 #include "qemu/osdep.h"
 #include "block/aio.h"
 #include "qemu/main-loop.h"
+#include "qemu/coroutine.h"
+
+AioContext *qemu_get_current_aio_context(void)
+{
+    return qemu_get_aio_context();
+}
 
 bool aio_context_in_iothread(AioContext *ctx)
 {
     return ctx == qemu_get_aio_context();
 }
+
+void qemu_coroutine_wake(AioContext *ctx, Coroutine *co)
+{
+    qemu_coroutine_queue_next(co);
+}
diff --git a/tests/iothread.c b/tests/iothread.c
index 00ab316..e0d61b5 100644
--- a/tests/iothread.c
+++ b/tests/iothread.c
@@ -16,6 +16,7 @@
 #include "qapi/error.h"
 #include "block/aio.h"
 #include "qemu/main-loop.h"
+#include "qemu/coroutine.h"
 #include "qemu/rcu.h"
 #include "iothread.h"
 
@@ -30,11 +31,26 @@ struct IOThread {
 
 static __thread IOThread *my_iothread;
 
+AioContext *qemu_get_current_aio_context(void)
+{
+    return my_iothread ? my_iothread->ctx : qemu_get_aio_context();
+}
+
 bool aio_context_in_iothread(AioContext *ctx)
 {
     return ctx == (my_iothread ? my_iothread->ctx : qemu_get_aio_context());
 }
 
+void qemu_coroutine_wake(AioContext *ctx, Coroutine *co)
+{
+    if (ctx != qemu_get_current_aio_context()) {
+        aio_co_schedule(ctx, co);
+        return;
+    }
+
+    qemu_coroutine_queue_next(co);
+}
+
 static void *iothread_run(void *opaque)
 {
     IOThread *iothread = opaque;
diff --git a/trace-events b/trace-events
index 78b042c..be17ba8 100644
--- a/trace-events
+++ b/trace-events
@@ -982,8 +982,9 @@ qemu_coroutine_yield(void *from, void *to) "from %p to %p"
 qemu_coroutine_terminate(void *co) "self %p"
 
 # qemu-coroutine-lock.c
+qemu_coroutine_queue_next(void *from, void *nxt) "%p->%p"
 qemu_co_queue_run_restart(void *co) "co %p"
-qemu_co_queue_next(void *nxt) "next %p"
+qemu_co_queue_next(void *ctx, void *nxt) "context %p next %p"
 qemu_co_mutex_lock_entry(void *mutex, void *self) "mutex %p self %p"
 qemu_co_mutex_lock_return(void *mutex, void *self) "mutex %p self %p"
 qemu_co_mutex_unlock_entry(void *mutex, void *self) "mutex %p self %p"
diff --git a/util/qemu-coroutine-lock.c b/util/qemu-coroutine-lock.c
index 1f46970..0af14ab 100644
--- a/util/qemu-coroutine-lock.c
+++ b/util/qemu-coroutine-lock.c
@@ -27,8 +27,16 @@
 #include "qemu/coroutine.h"
 #include "qemu/coroutine_int.h"
 #include "qemu/queue.h"
+#include "block/aio.h"
 #include "trace.h"
 
+void qemu_coroutine_queue_next(Coroutine *co)
+{
+    Coroutine *self = qemu_coroutine_self();
+    trace_qemu_coroutine_queue_next(self, co);
+    QSIMPLEQ_INSERT_TAIL(&self->co_queue_wakeup, co, co_queue_next);
+}
+
 void qemu_co_queue_init(CoQueue *queue)
 {
     QSIMPLEQ_INIT(&queue->entries);
@@ -37,6 +45,7 @@ void qemu_co_queue_init(CoQueue *queue)
 void coroutine_fn qemu_co_queue_wait(CoQueue *queue)
 {
     Coroutine *self = qemu_coroutine_self();
+    self->ctx = qemu_get_current_aio_context();
     QSIMPLEQ_INSERT_TAIL(&queue->entries, self, co_queue_next);
     qemu_coroutine_yield();
     assert(qemu_in_coroutine());
@@ -63,7 +72,6 @@ void qemu_co_queue_run_restart(Coroutine *co)
 
 static bool qemu_co_queue_do_restart(CoQueue *queue, bool single)
 {
-    Coroutine *self = qemu_coroutine_self();
     Coroutine *next;
 
     if (QSIMPLEQ_EMPTY(&queue->entries)) {
@@ -72,8 +80,7 @@ static bool qemu_co_queue_do_restart(CoQueue *queue, bool single)
 
     while ((next = QSIMPLEQ_FIRST(&queue->entries)) != NULL) {
         QSIMPLEQ_REMOVE_HEAD(&queue->entries, co_queue_next);
-        QSIMPLEQ_INSERT_TAIL(&self->co_queue_wakeup, next, co_queue_next);
-        trace_qemu_co_queue_next(next);
+        qemu_coroutine_wake(next->ctx, next);
         if (single) {
             break;
         }
-- 
2.5.5