[PATCH v6 11/25] QmpSession: return orderly

[Marc-André Lureau]

QEMU will gain support for asynchronous commands, and may thus finish
commands in various order. However, the clients expect replies in
order. Let's enforce ordering of replies in QmpReturn: starting from
the older command, process each pending QmpReturn, and return until
reaching one that is unfinished.

Or if the command is OOB, it should return immediately.

Signed-off-by: Marc-André Lureau <address@hidden>
---
 include/qapi/qmp/dispatch.h |  2 ++
 qapi/qmp-dispatch.c         | 61 ++++++++++++++++++++++++++++++-------
 tests/test-qmp-cmds.c       | 33 ++++++++++++++++++++
 3 files changed, 85 insertions(+), 11 deletions(-)

diff --git a/include/qapi/qmp/dispatch.h b/include/qapi/qmp/dispatch.h
index 7c9de9780d..92d6fd1afb 100644
--- a/include/qapi/qmp/dispatch.h
+++ b/include/qapi/qmp/dispatch.h
@@ -55,6 +55,8 @@ struct QmpSession {
 struct QmpReturn {
     QmpSession *session;
     QDict *rsp;
+    bool oob;
+    bool finished;
     QTAILQ_ENTRY(QmpReturn) entry;
 };
 
diff --git a/qapi/qmp-dispatch.c b/qapi/qmp-dispatch.c
index aed5c91057..3eb7e4b610 100644
--- a/qapi/qmp-dispatch.c
+++ b/qapi/qmp-dispatch.c
@@ -25,6 +25,7 @@ QmpReturn *qmp_return_new(QmpSession *session, const QObject 
*request)
     const QDict *req = qobject_to(QDict, request);
     QObject *id = req ? qdict_get(req, "id") : NULL;
 
+    qret->oob = req ? qmp_is_oob(req) : false;
     qret->session = session;
     qret->rsp = qdict_new();
     if (id) {
@@ -39,6 +40,15 @@ QmpReturn *qmp_return_new(QmpSession *session, const QObject 
*request)
     return qret;
 }
 
+static void qmp_return_free_with_lock(QmpReturn *qret)
+{
+    if (qret->session) {
+        QTAILQ_REMOVE(&qret->session->pending, qret, entry);
+    }
+    qobject_unref(qret->rsp);
+    g_free(qret);
+}
+
 void qmp_return_free(QmpReturn *qret)
 {
     QmpSession *session = qret->session;
@@ -46,21 +56,53 @@ void qmp_return_free(QmpReturn *qret)
     if (session) {
         qemu_mutex_lock(&session->pending_lock);
     }
-    QTAILQ_REMOVE(&session->pending, qret, entry);
+
+    qmp_return_free_with_lock(qret);
+
     if (session) {
         qemu_mutex_unlock(&session->pending_lock);
     }
-    qobject_unref(qret->rsp);
-    g_free(qret);
+}
+
+static void qmp_return_orderly(QmpReturn *qret)
+{
+    QmpSession *session = qret->session;
+    QmpReturn *ret, *next;
+
+    if (!session) {
+        /* the session was destroyed before return, discard */
+        qmp_return_free(qret);
+        return;
+    }
+    if (qret->oob) {
+        session->return_cb(session, qret->rsp);
+        qmp_return_free(qret);
+        return;
+    }
+
+    qret->finished = true;
+
+    qemu_mutex_lock(&session->pending_lock);
+    /*
+     * Process the list of pending and call return_cb until reaching
+     * an unfinished.
+     */
+    QTAILQ_FOREACH_SAFE(ret, &session->pending, entry, next) {
+        if (!ret->finished) {
+            break;
+        }
+        session->return_cb(session, ret->rsp);
+        ret->session = session;
+        qmp_return_free_with_lock(ret);
+    }
+
+    qemu_mutex_unlock(&session->pending_lock);
 }
 
 void qmp_return(QmpReturn *qret, QObject *rsp)
 {
     qdict_put_obj(qret->rsp, "return", rsp ?: QOBJECT(qdict_new()));
-    if (qret->session) {
-        qret->session->return_cb(qret->session, qret->rsp);
-    }
-    qmp_return_free(qret);
+    qmp_return_orderly(qret);
 }
 
 void qmp_return_error(QmpReturn *qret, Error *err)
@@ -70,10 +112,7 @@ void qmp_return_error(QmpReturn *qret, Error *err)
     qdict_put_str(qdict, "desc", error_get_pretty(err));
     qdict_put_obj(qret->rsp, "error", QOBJECT(qdict));
     error_free(err);
-    if (qret->session) {
-        qret->session->return_cb(qret->session, qret->rsp);
-    }
-    qmp_return_free(qret);
+    qmp_return_orderly(qret);
 }
 
 static QDict *qmp_dispatch_check_obj(const QObject *request, bool allow_oob,
diff --git a/tests/test-qmp-cmds.c b/tests/test-qmp-cmds.c
index d9623d10b6..213df7e53a 100644
--- a/tests/test-qmp-cmds.c
+++ b/tests/test-qmp-cmds.c
@@ -361,6 +361,38 @@ static void test_dealloc_partial(void)
     qapi_free_UserDefTwo(ud2);
 }
 
+typedef struct QmpReturnOrderly {
+    QmpSession session;
+    int returns;
+} QmpReturnOrderly;
+
+static void dispatch_return_orderly(QmpSession *session, QDict *resp)
+{
+    QmpReturnOrderly *o = container_of(session, QmpReturnOrderly, session);
+
+    o->returns++;
+}
+
+static void test_qmp_return_orderly(void)
+{
+    QDict *dict = qdict_new();
+    QmpReturnOrderly o = { { 0 }, };
+    QmpReturn *r1, *r2, *r3;
+
+    qmp_session_init(&o.session, &qmp_commands, NULL, dispatch_return_orderly);
+    r1 = qmp_return_new(&o.session, NULL);
+    qdict_put_str(dict, "exec-oob", "test");
+    r2 = qmp_return_new(&o.session, QOBJECT(dict));
+    r3 = qmp_return_new(&o.session, NULL);
+    qmp_return(r3, NULL);
+    g_assert_cmpint(o.returns, ==, 0);
+    qmp_return(r2, NULL);
+    g_assert_cmpint(o.returns, ==, 1);
+    qmp_return(r1, NULL);
+    g_assert_cmpint(o.returns, ==, 3);
+    qmp_session_destroy(&o.session);
+    qobject_unref(dict);
+}
 
 int main(int argc, char **argv)
 {
@@ -374,6 +406,7 @@ int main(int argc, char **argv)
                     test_dispatch_cmd_success_response);
     g_test_add_func("/qmp/dealloc_types", test_dealloc_types);
     g_test_add_func("/qmp/dealloc_partial", test_dealloc_partial);
+    g_test_add_func("/qmp/return_orderly", test_qmp_return_orderly);
 
     test_qmp_init_marshal(&qmp_commands);
     g_test_run();
-- 
2.24.0