[PATCH 0/5] qmp: cleanup QMP dispatcher coroutine

[PATCH 0/5] qmp: cleanup QMP dispatcher coroutine

[Paolo Bonzini]

The QMP dispatcher coroutine is a bit messy.  Clean it up by creating
small and understandable functions, and once the wakeup logic is clear
change it to remove the qatomic_mb_read()/qatomic_mb_set() pair.

With this change and the other HMP series posted, the last use of
qatomic_mb_read() is removed.

Paolo

Paolo Bonzini (5):
  monitor: cleanup detection of qmp_dispatcher_co shutting down
  monitor: cleanup fetching of QMP requests
  monitor: introduce qmp_dispatcher_co_wake
  monitor: extract request dequeuing to a new function
  monitor: do not use mb_read/mb_set

 monitor/monitor-internal.h |   2 +-
 monitor/monitor.c          |  33 +++--------
 monitor/qmp.c              | 110 ++++++++++++++++++++++++-------------
 3 files changed, 80 insertions(+), 65 deletions(-)

-- 
2.40.1

[PATCH 1/5] monitor: cleanup detection of qmp_dispatcher_co shutting down

[Paolo Bonzini]

Instead of overloading qmp_dispatcher_co_busy, make the coroutine
pointer NULL.  This will make things break spectacularly if somebody
tries to start a request after monitor_cleanup().

AIO_WAIT_WHILE_UNLOCKED() does not need qatomic_mb_read(), because
the macro contains all the necessary memory barriers.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 monitor/monitor.c | 2 +-
 monitor/qmp.c     | 2 ++
 2 files changed, 3 insertions(+), 1 deletion(-)

diff --git a/monitor/monitor.c b/monitor/monitor.c
index 15f97538ef2b..c4ed2547c25f 100644
--- a/monitor/monitor.c
+++ b/monitor/monitor.c
@@ -686,7 +686,7 @@ void monitor_cleanup(void)
 
     AIO_WAIT_WHILE_UNLOCKED(NULL,
                    (aio_poll(iohandler_get_aio_context(), false),
-                    qatomic_mb_read(&qmp_dispatcher_co_busy)));
+                    qatomic_read(&qmp_dispatcher_co)));
 
     /*
      * We need to explicitly stop the I/O thread (but not destroy it),
diff --git a/monitor/qmp.c b/monitor/qmp.c
index 092c527b6fc9..f0cc6dc886f8 100644
--- a/monitor/qmp.c
+++ b/monitor/qmp.c
@@ -226,6 +226,7 @@ void coroutine_fn monitor_qmp_dispatcher_co(void *data)
 
         /* On shutdown, don't take any more requests from the queue */
         if (qmp_dispatcher_co_shutdown) {
+            qatomic_set(&qmp_dispatcher_co, NULL);
             return;
         }
 
@@ -250,6 +251,7 @@ void coroutine_fn monitor_qmp_dispatcher_co(void *data)
              * yielded and were reentered from monitor_cleanup()
              */
             if (qmp_dispatcher_co_shutdown) {
+                qatomic_set(&qmp_dispatcher_co, NULL);
                 return;
             }
         }
-- 
2.40.1

[PATCH 2/5] monitor: cleanup fetching of QMP requests

[Paolo Bonzini]

Use a continue statement so that "after going to sleep" is treated the same
way as "after processing a request".  Pull the monitor_lock critical
section out of monitor_qmp_requests_pop_any_with_lock() and protect
qmp_dispatcher_co_shutdown with the monitor_lock.

The two changes are complex to separate because monitor_qmp_dispatcher_co()
previously had a complicated logic to check for shutdown both before
and after going to sleep.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 monitor/monitor.c |  9 +++++++--
 monitor/qmp.c     | 38 ++++++++++++++------------------------
 2 files changed, 21 insertions(+), 26 deletions(-)

diff --git a/monitor/monitor.c b/monitor/monitor.c
index c4ed2547c25f..042a1ab918f9 100644
--- a/monitor/monitor.c
+++ b/monitor/monitor.c
@@ -56,7 +56,10 @@ IOThread *mon_iothread;
 /* Coroutine to dispatch the requests received from I/O thread */
 Coroutine *qmp_dispatcher_co;
 
-/* Set to true when the dispatcher coroutine should terminate */
+/*
+ * Set to true when the dispatcher coroutine should terminate.  Protected
+ * by monitor_lock.
+ */
 bool qmp_dispatcher_co_shutdown;
 
 /*
@@ -679,7 +682,9 @@ void monitor_cleanup(void)
      * we'll just leave them in the queue without sending a response
      * and monitor_data_destroy() will free them.
      */
-    qmp_dispatcher_co_shutdown = true;
+    WITH_QEMU_LOCK_GUARD(&monitor_lock) {
+        qmp_dispatcher_co_shutdown = true;
+    }
     if (!qatomic_xchg(&qmp_dispatcher_co_busy, true)) {
         aio_co_wake(qmp_dispatcher_co);
     }
diff --git a/monitor/qmp.c b/monitor/qmp.c
index f0cc6dc886f8..7516b92a4d3e 100644
--- a/monitor/qmp.c
+++ b/monitor/qmp.c
@@ -178,7 +178,10 @@ static QMPRequest *monitor_qmp_requests_pop_any_with_lock(void)
     Monitor *mon;
     MonitorQMP *qmp_mon;
 
-    QEMU_LOCK_GUARD(&monitor_lock);
+    /* On shutdown, don't take any more requests from the queue */
+    if (qmp_dispatcher_co_shutdown) {
+        return NULL;
+    }
 
     QTAILQ_FOREACH(mon, &mon_list, entry) {
         if (!monitor_is_qmp(mon)) {
@@ -215,6 +218,10 @@ void coroutine_fn monitor_qmp_dispatcher_co(void *data)
     MonitorQMP *mon;
 
     while (true) {
+        /*
+         * busy must be set to true again by whoever
+         * rescheduled us to avoid double scheduling
+         */
         assert(qatomic_mb_read(&qmp_dispatcher_co_busy) == true);
 
         /*
@@ -224,36 +231,18 @@ void coroutine_fn monitor_qmp_dispatcher_co(void *data)
          */
         qatomic_mb_set(&qmp_dispatcher_co_busy, false);
 
-        /* On shutdown, don't take any more requests from the queue */
-        if (qmp_dispatcher_co_shutdown) {
-            qatomic_set(&qmp_dispatcher_co, NULL);
-            return;
+        WITH_QEMU_LOCK_GUARD(&monitor_lock) {
+            req_obj = monitor_qmp_requests_pop_any_with_lock();
         }
 
-        while (!(req_obj = monitor_qmp_requests_pop_any_with_lock())) {
+        if (!req_obj) {
             /*
              * No more requests to process.  Wait to be reentered from
              * handle_qmp_command() when it pushes more requests, or
              * from monitor_cleanup() when it requests shutdown.
              */
-            if (!qmp_dispatcher_co_shutdown) {
-                qemu_coroutine_yield();
-
-                /*
-                 * busy must be set to true again by whoever
-                 * rescheduled us to avoid double scheduling
-                 */
-                assert(qatomic_xchg(&qmp_dispatcher_co_busy, false) == true);
-            }
-
-            /*
-             * qmp_dispatcher_co_shutdown may have changed if we
-             * yielded and were reentered from monitor_cleanup()
-             */
-            if (qmp_dispatcher_co_shutdown) {
-                qatomic_set(&qmp_dispatcher_co, NULL);
-                return;
-            }
+            qemu_coroutine_yield();
+            continue;
         }
 
         trace_monitor_qmp_in_band_dequeue(req_obj,
@@ -342,6 +331,7 @@ void coroutine_fn monitor_qmp_dispatcher_co(void *data)
         aio_co_schedule(iohandler_get_aio_context(), qmp_dispatcher_co);
         qemu_coroutine_yield();
     }
+    qatomic_set(&qmp_dispatcher_co, NULL);
 }
 
 static void handle_qmp_command(void *opaque, QObject *req, Error *err)
-- 
2.40.1

[PATCH 3/5] monitor: introduce qmp_dispatcher_co_wake

[Paolo Bonzini]

This makes it possible to turn qmp_dispatcher_co_busy into a static
variable.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 monitor/monitor-internal.h |  2 +-
 monitor/monitor.c          | 26 +-------------------------
 monitor/qmp.c              | 32 +++++++++++++++++++++++++++++---
 3 files changed, 31 insertions(+), 29 deletions(-)

diff --git a/monitor/monitor-internal.h b/monitor/monitor-internal.h
index 61c9b6916db3..252de856812f 100644
--- a/monitor/monitor-internal.h
+++ b/monitor/monitor-internal.h
@@ -165,7 +165,6 @@ typedef QTAILQ_HEAD(MonitorList, Monitor) MonitorList;
 extern IOThread *mon_iothread;
 extern Coroutine *qmp_dispatcher_co;
 extern bool qmp_dispatcher_co_shutdown;
-extern bool qmp_dispatcher_co_busy;
 extern QmpCommandList qmp_commands, qmp_cap_negotiation_commands;
 extern QemuMutex monitor_lock;
 extern MonitorList mon_list;
@@ -183,6 +182,7 @@ void monitor_fdsets_cleanup(void);
 void qmp_send_response(MonitorQMP *mon, const QDict *rsp);
 void monitor_data_destroy_qmp(MonitorQMP *mon);
 void coroutine_fn monitor_qmp_dispatcher_co(void *data);
+void qmp_dispatcher_co_wake(void);
 
 int get_monitor_def(Monitor *mon, int64_t *pval, const char *name);
 void handle_hmp_command(MonitorHMP *mon, const char *cmdline);
diff --git a/monitor/monitor.c b/monitor/monitor.c
index 042a1ab918f9..dc352f9e9d95 100644
--- a/monitor/monitor.c
+++ b/monitor/monitor.c
@@ -62,27 +62,6 @@ Coroutine *qmp_dispatcher_co;
  */
 bool qmp_dispatcher_co_shutdown;
 
-/*
- * qmp_dispatcher_co_busy is used for synchronisation between the
- * monitor thread and the main thread to ensure that the dispatcher
- * coroutine never gets scheduled a second time when it's already
- * scheduled (scheduling the same coroutine twice is forbidden).
- *
- * It is true if the coroutine is active and processing requests.
- * Additional requests may then be pushed onto mon->qmp_requests,
- * and @qmp_dispatcher_co_shutdown may be set without further ado.
- * @qmp_dispatcher_co_busy must not be woken up in this case.
- *
- * If false, you also have to set @qmp_dispatcher_co_busy to true and
- * wake up @qmp_dispatcher_co after pushing the new requests.
- *
- * The coroutine will automatically change this variable back to false
- * before it yields.  Nobody else may set the variable to false.
- *
- * Access must be atomic for thread safety.
- */
-bool qmp_dispatcher_co_busy;
-
 /*
  * Protects mon_list, monitor_qapi_event_state, coroutine_mon,
  * monitor_destroyed.
@@ -685,9 +664,7 @@ void monitor_cleanup(void)
     WITH_QEMU_LOCK_GUARD(&monitor_lock) {
         qmp_dispatcher_co_shutdown = true;
     }
-    if (!qatomic_xchg(&qmp_dispatcher_co_busy, true)) {
-        aio_co_wake(qmp_dispatcher_co);
-    }
+    qmp_dispatcher_co_wake();
 
     AIO_WAIT_WHILE_UNLOCKED(NULL,
                    (aio_poll(iohandler_get_aio_context(), false),
@@ -742,7 +719,6 @@ void monitor_init_globals(void)
      * rid of those assumptions.
      */
     qmp_dispatcher_co = qemu_coroutine_create(monitor_qmp_dispatcher_co, NULL);
-    qatomic_mb_set(&qmp_dispatcher_co_busy, true);
     aio_co_schedule(iohandler_get_aio_context(), qmp_dispatcher_co);
 }
 
diff --git a/monitor/qmp.c b/monitor/qmp.c
index 7516b92a4d3e..baa1f6652e07 100644
--- a/monitor/qmp.c
+++ b/monitor/qmp.c
@@ -33,6 +33,27 @@
 #include "qapi/qmp/qlist.h"
 #include "trace.h"
 
+/*
+ * qmp_dispatcher_co_busy is used for synchronisation between the
+ * monitor thread and the main thread to ensure that the dispatcher
+ * coroutine never gets scheduled a second time when it's already
+ * scheduled (scheduling the same coroutine twice is forbidden).
+ *
+ * It is true if the coroutine is active and processing requests.
+ * Additional requests may then be pushed onto mon->qmp_requests,
+ * and @qmp_dispatcher_co_shutdown may be set without further ado.
+ * @qmp_dispatcher_co_busy must not be woken up in this case.
+ *
+ * If false, you also have to set @qmp_dispatcher_co_busy to true and
+ * wake up @qmp_dispatcher_co after pushing the new requests.
+ *
+ * The coroutine will automatically change this variable back to false
+ * before it yields.  Nobody else may set the variable to false.
+ *
+ * Access must be atomic for thread safety.
+ */
+static bool qmp_dispatcher_co_busy = true;
+
 struct QMPRequest {
     /* Owner of the request */
     MonitorQMP *mon;
@@ -334,6 +355,13 @@ void coroutine_fn monitor_qmp_dispatcher_co(void *data)
     qatomic_set(&qmp_dispatcher_co, NULL);
 }
 
+void qmp_dispatcher_co_wake(void)
+{
+    if (!qatomic_xchg(&qmp_dispatcher_co_busy, true)) {
+        aio_co_wake(qmp_dispatcher_co);
+    }
+}
+
 static void handle_qmp_command(void *opaque, QObject *req, Error *err)
 {
     MonitorQMP *mon = opaque;
@@ -395,9 +423,7 @@ static void handle_qmp_command(void *opaque, QObject *req, Error *err)
     }
 
     /* Kick the dispatcher routine */
-    if (!qatomic_xchg(&qmp_dispatcher_co_busy, true)) {
-        aio_co_wake(qmp_dispatcher_co);
-    }
+    qmp_dispatcher_co_wake();
 }
 
 static void monitor_qmp_read(void *opaque, const uint8_t *buf, int size)
-- 
2.40.1

[PATCH 4/5] monitor: extract request dequeuing to a new function

[Paolo Bonzini]

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 monitor/qmp.c | 49 ++++++++++++++++++++++++++++---------------------
 1 file changed, 28 insertions(+), 21 deletions(-)

diff --git a/monitor/qmp.c b/monitor/qmp.c
index baa1f6652e07..9ec28be2ee10 100644
--- a/monitor/qmp.c
+++ b/monitor/qmp.c
@@ -199,11 +199,6 @@ static QMPRequest *monitor_qmp_requests_pop_any_with_lock(void)
     Monitor *mon;
     MonitorQMP *qmp_mon;
 
-    /* On shutdown, don't take any more requests from the queue */
-    if (qmp_dispatcher_co_shutdown) {
-        return NULL;
-    }
-
     QTAILQ_FOREACH(mon, &mon_list, entry) {
         if (!monitor_is_qmp(mon)) {
             continue;
@@ -231,14 +226,9 @@ static QMPRequest *monitor_qmp_requests_pop_any_with_lock(void)
     return req_obj;
 }
 
-void coroutine_fn monitor_qmp_dispatcher_co(void *data)
+static QMPRequest *monitor_qmp_dispatcher_pop_any(void)
 {
-    QMPRequest *req_obj = NULL;
-    QDict *rsp;
-    bool oob_enabled;
-    MonitorQMP *mon;
-
-    while (true) {
+    for (;;) {
         /*
          * busy must be set to true again by whoever
          * rescheduled us to avoid double scheduling
@@ -253,19 +243,36 @@ void coroutine_fn monitor_qmp_dispatcher_co(void *data)
         qatomic_mb_set(&qmp_dispatcher_co_busy, false);
 
         WITH_QEMU_LOCK_GUARD(&monitor_lock) {
+            QMPRequest *req_obj;
+
+            /* On shutdown, don't take any more requests from the queue */
+            if (qmp_dispatcher_co_shutdown) {
+                return NULL;
+            }
+
             req_obj = monitor_qmp_requests_pop_any_with_lock();
+            if (req_obj) {
+                return req_obj;
+            }
         }
 
-        if (!req_obj) {
-            /*
-             * No more requests to process.  Wait to be reentered from
-             * handle_qmp_command() when it pushes more requests, or
-             * from monitor_cleanup() when it requests shutdown.
-             */
-            qemu_coroutine_yield();
-            continue;
-        }
+        /*
+         * No more requests to process.  Wait to be reentered from
+         * handle_qmp_command() when it pushes more requests, or
+         * from monitor_cleanup() when it requests shutdown.
+         */
+        qemu_coroutine_yield();
+    }
+}
 
+void coroutine_fn monitor_qmp_dispatcher_co(void *data)
+{
+    QMPRequest *req_obj;
+    QDict *rsp;
+    bool oob_enabled;
+    MonitorQMP *mon;
+
+    while ((req_obj = monitor_qmp_dispatcher_pop_any()) != NULL) {
         trace_monitor_qmp_in_band_dequeue(req_obj,
                                           req_obj->mon->qmp_requests->length);
 
-- 
2.40.1

[PATCH 5/5] monitor: do not use mb_read/mb_set

[Paolo Bonzini]

Instead of relying on magic memory barriers, document the pattern that
is being used.  It is the one based on Dekker's algorithm, and in this
case it is embodied as follows:

    enqueue request;              sleeping = true;
    smp_mb();                     smp_mb();
    if (sleeping) kick();         if (!have a request) yield();

qatomic_mb_set() can be kept---but it will be shortly renamed to
qatomic_set_mb() to clarify that the write occurs _before_ the
barrier, just in the right column of the pseudocode above.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 monitor/qmp.c | 27 ++++++++++++++++++---------
 1 file changed, 18 insertions(+), 9 deletions(-)

diff --git a/monitor/qmp.c b/monitor/qmp.c
index 9ec28be2ee10..aee8ce4cf006 100644
--- a/monitor/qmp.c
+++ b/monitor/qmp.c
@@ -39,13 +39,16 @@
  * coroutine never gets scheduled a second time when it's already
  * scheduled (scheduling the same coroutine twice is forbidden).
  *
- * It is true if the coroutine is active and processing requests.
- * Additional requests may then be pushed onto mon->qmp_requests,
- * and @qmp_dispatcher_co_shutdown may be set without further ado.
- * @qmp_dispatcher_co_busy must not be woken up in this case.
+ * It is true if the coroutine will process at least one more request
+ * before going to sleep.  Either it has been kicked already, or it
+ * is active and processing requests.  Additional requests may therefore
+ * be pushed onto mon->qmp_requests, and @qmp_dispatcher_co_shutdown may
+ * be set without further ado.  @qmp_dispatcher_co must not be woken up
+ * in this case.
  *
- * If false, you also have to set @qmp_dispatcher_co_busy to true and
- * wake up @qmp_dispatcher_co after pushing the new requests.
+ * If false, you have to wake up @qmp_dispatcher_co after pushing new
+ * requests. You also have to set @qmp_dispatcher_co_busy to true
+ * before waking up the coroutine.
  *
  * The coroutine will automatically change this variable back to false
  * before it yields.  Nobody else may set the variable to false.
@@ -230,15 +233,18 @@ static QMPRequest *monitor_qmp_dispatcher_pop_any(void)
 {
     for (;;) {
         /*
-         * busy must be set to true again by whoever
-         * rescheduled us to avoid double scheduling
+         * To avoid double scheduling, busy is true on entry to
+         * monitor_qmp_dispatcher_co(), and must be set again before
+         * aio_co_wake()-ing it.
          */
-        assert(qatomic_mb_read(&qmp_dispatcher_co_busy) == true);
+        assert(qatomic_read(&qmp_dispatcher_co_busy) == true);
 
         /*
          * Mark the dispatcher as not busy already here so that we
          * don't miss any new requests coming in the middle of our
          * processing.
+         *
+         * Clear qmp_dispatcher_co_busy before reading request.
          */
         qatomic_mb_set(&qmp_dispatcher_co_busy, false);
 
@@ -364,6 +370,9 @@ void coroutine_fn monitor_qmp_dispatcher_co(void *data)
 
 void qmp_dispatcher_co_wake(void)
 {
+    /* Write request before reading qmp_dispatcher_co_busy.  */
+    smp_mb__before_rmw();
+
     if (!qatomic_xchg(&qmp_dispatcher_co_busy, true)) {
         aio_co_wake(qmp_dispatcher_co);
     }
-- 
2.40.1