Merge remote-tracking branch 'remotes/armbru/tags/pull-monitor-2018-07-03-v2' into staging

Monitor patches for 2018-07-03

# gpg: Signature made Tue 03 Jul 2018 22:20:13 BST
# gpg:                using RSA key 3870B400EB918653
# gpg: Good signature from "Markus Armbruster <armbru@redhat.com>"
# gpg:                 aka "Markus Armbruster <armbru@pond.sub.org>"
# Primary key fingerprint: 354B C8B3 D7EB 2A6B 6867  4E5F 3870 B400 EB91 8653

* remotes/armbru/tags/pull-monitor-2018-07-03-v2: (32 commits)
  qapi: Polish command flags documentation in qapi-code-gen.txt
  monitor: Improve some comments
  qmp: Clean up capability negotiation after commit 02130314d8c
  qobject: Let qobject_from_jsonf() fail instead of abort
  qmp: Switch timestamp_put() to qdict_from_jsonf_nofail()
  qmp: Add some comments around null responses
  qmp: Simplify monitor_qmp_respond()
  qmp: Replace get_qmp_greeting() by qmp_greeting()
  qmp: Replace monitor_json_emitter{,raw}() by qmp_{queue,send}_response()
  qmp: Use QDict * instead of QObject * for response objects
  qmp: De-duplicate error response building
  qobject: New qdict_from_jsonf_nofail()
  monitor: Peel off @mon_global wrapper
  monitor: Rename use_io_thr to use_io_thread
  qmp: Don't let JSON errors jump the queue
  qmp: Don't let malformed in-band commands jump the queue
  tests/qmp-test: Demonstrate QMP errors jumping the queue
  qmp: Simplify code around monitor_qmp_dispatch_one()
  qmp: Always free QMPRequest with qmp_request_free()
  qmp: Revert change to handle_qmp_command tracepoint
  ...

Signed-off-by: Peter Maydell <peter.maydell@linaro.org>

1 files changed, 212 insertions, 295 deletions

diff --git a/monitor.c b/monitor.c
index 8e1aea7826..3c9c97b73f 100644
--- a/monitor.c
+++ b/monitor.c
@@ -169,14 +169,16 @@ typedef struct {
     JSONMessageParser parser;
     /*
      * When a client connects, we're in capabilities negotiation mode.
-     * When command qmp_capabilities succeeds, we go into command
-     * mode.
+     * @commands is &qmp_cap_negotiation_commands then.  When command
+     * qmp_capabilities succeeds, we go into command mode, and
+     * @command becomes &qmp_commands.
      */
     QmpCommandList *commands;
-    bool qmp_caps[QMP_CAPABILITY__MAX];
+    bool capab_offered[QMP_CAPABILITY__MAX]; /* capabilities offered */
+    bool capab[QMP_CAPABILITY__MAX];         /* offered and accepted */
     /*
-     * Protects qmp request/response queue.  Please take monitor_lock
-     * first when used together.
+     * Protects qmp request/response queue.
+     * Take monitor_lock first when you need both.
      */
     QemuMutex qmp_queue_lock;
     /* Input queue that holds all the parsed QMP requests */
@@ -207,11 +209,11 @@ struct Monitor {
     int flags;
     int suspend_cnt;            /* Needs to be accessed atomically */
     bool skip_flush;
-    bool use_io_thr;
+    bool use_io_thread;
 
     /*
      * State used only in the thread "owning" the monitor.
-     * If @use_io_thr, this is mon_global.mon_iothread.
+     * If @use_io_thread, this is @mon_iothread.
      * Else, it's the main thread.
      * These members can be safely accessed without locks.
      */
@@ -231,7 +233,7 @@ struct Monitor {
     QemuMutex mon_lock;
 
     /*
-     * Fields that are protected by the per-monitor lock.
+     * Members that are protected by the per-monitor lock
      */
     QLIST_HEAD(, mon_fd_t) fds;
     QString *outbuf;
@@ -240,22 +242,26 @@ struct Monitor {
     int mux_out;
 };
 
-/* Let's add monitor global variables to this struct. */
-static struct {
-    IOThread *mon_iothread;
-    /* Bottom half to dispatch the requests received from IO thread */
-    QEMUBH *qmp_dispatcher_bh;
-    /* Bottom half to deliver the responses back to clients */
-    QEMUBH *qmp_respond_bh;
-} mon_global;
+/* Shared monitor I/O thread */
+IOThread *mon_iothread;
+
+/* Bottom half to dispatch the requests received from I/O thread */
+QEMUBH *qmp_dispatcher_bh;
+
+/* Bottom half to deliver the responses back to clients */
+QEMUBH *qmp_respond_bh;
 
 struct QMPRequest {
     /* Owner of the request */
     Monitor *mon;
     /* "id" field of the request */
     QObject *id;
-    /* Request object to be handled */
+    /*
+     * Request object to be handled or Error to be reported
+     * (exactly one of them is non-null)
+     */
     QObject *req;
+    Error *err;
     /*
      * Whether we need to resume the monitor afterward.  This flag is
      * used to emulate the old QMP server behavior that the current
@@ -298,9 +304,9 @@ static inline bool monitor_is_qmp(const Monitor *mon)
 }
 
 /**
- * Whether @mon is using readline?  Note: not all HMP monitors use
- * readline, e.g., gdbserver has a non-interactive HMP monitor, so
- * readline is not used there.
+ * Is @mon is using readline?
+ * Note: not all HMP monitors use readline, e.g., gdbserver has a
+ * non-interactive HMP monitor, so readline is not used there.
  */
 static inline bool monitor_uses_readline(const Monitor *mon)
 {
@@ -314,14 +320,12 @@ static inline bool monitor_is_hmp_non_interactive(const Monitor *mon)
 
 /*
  * Return the clock to use for recording an event's time.
+ * It's QEMU_CLOCK_REALTIME, except for qtests it's
+ * QEMU_CLOCK_VIRTUAL, to support testing rate limits.
  * Beware: result is invalid before configure_accelerator().
  */
 static inline QEMUClockType monitor_get_event_clock(void)
 {
-    /*
-     * This allows us to perform tests on the monitor queues to verify
-     * that the rate limits are enforced.
-     */
     return qtest_enabled() ? QEMU_CLOCK_VIRTUAL : QEMU_CLOCK_REALTIME;
 }
 
@@ -360,10 +364,11 @@ static void qmp_request_free(QMPRequest *req)
 {
     qobject_unref(req->id);
     qobject_unref(req->req);
+    error_free(req->err);
     g_free(req);
 }
 
-/* Must with the mon->qmp.qmp_queue_lock held */
+/* Caller must hold mon->qmp.qmp_queue_lock */
 static void monitor_qmp_cleanup_req_queue_locked(Monitor *mon)
 {
     while (!g_queue_is_empty(mon->qmp.qmp_requests)) {
@@ -371,11 +376,11 @@ static void monitor_qmp_cleanup_req_queue_locked(Monitor *mon)
     }
 }
 
-/* Must with the mon->qmp.qmp_queue_lock held */
+/* Caller must hold the mon->qmp.qmp_queue_lock */
 static void monitor_qmp_cleanup_resp_queue_locked(Monitor *mon)
 {
     while (!g_queue_is_empty(mon->qmp.qmp_responses)) {
-        qobject_unref((QObject *)g_queue_pop_head(mon->qmp.qmp_responses));
+        qobject_unref((QDict *)g_queue_pop_head(mon->qmp.qmp_responses));
     }
 }
 
@@ -402,7 +407,7 @@ static gboolean monitor_unblocked(GIOChannel *chan, GIOCondition cond,
     return FALSE;
 }
 
-/* Called with mon->mon_lock held.  */
+/* Caller must hold mon->mon_lock */
 static void monitor_flush_locked(Monitor *mon)
 {
     int rc;
@@ -499,9 +504,9 @@ int monitor_fprintf(FILE *stream, const char *fmt, ...)
     return 0;
 }
 
-static void monitor_json_emitter_raw(Monitor *mon,
-                                     QObject *data)
+static void qmp_send_response(Monitor *mon, QDict *rsp)
 {
+    QObject *data = QOBJECT(rsp);
     QString *json;
 
     json = mon->flags & MONITOR_USE_PRETTY ? qobject_to_json_pretty(data) :
@@ -514,37 +519,35 @@ static void monitor_json_emitter_raw(Monitor *mon,
     qobject_unref(json);
 }
 
-static void monitor_json_emitter(Monitor *mon, QObject *data)
+static void qmp_queue_response(Monitor *mon, QDict *rsp)
 {
-    if (mon->use_io_thr) {
+    if (mon->use_io_thread) {
         /*
-         * If using IO thread, we need to queue the item so that IO
-         * thread will do the rest for us.  Take refcount so that
-         * caller won't free the data (which will be finally freed in
-         * responder thread).
+         * Push a reference to the response queue.  The I/O thread
+         * drains that queue and emits.
          */
         qemu_mutex_lock(&mon->qmp.qmp_queue_lock);
-        g_queue_push_tail(mon->qmp.qmp_responses, qobject_ref(data));
+        g_queue_push_tail(mon->qmp.qmp_responses, qobject_ref(rsp));
         qemu_mutex_unlock(&mon->qmp.qmp_queue_lock);
-        qemu_bh_schedule(mon_global.qmp_respond_bh);
+        qemu_bh_schedule(qmp_respond_bh);
     } else {
         /*
-         * If not using monitor IO thread, then we are in main thread.
-         * Do the emission right away.
+         * Not using monitor I/O thread, i.e. we are in the main thread.
+         * Emit right away.
          */
-        monitor_json_emitter_raw(mon, data);
+        qmp_send_response(mon, rsp);
     }
 }
 
 struct QMPResponse {
     Monitor *mon;
-    QObject *data;
+    QDict *data;
 };
 typedef struct QMPResponse QMPResponse;
 
-static QObject *monitor_qmp_response_pop_one(Monitor *mon)
+static QDict *monitor_qmp_response_pop_one(Monitor *mon)
 {
-    QObject *data;
+    QDict *data;
 
     qemu_mutex_lock(&mon->qmp.qmp_queue_lock);
     data = g_queue_pop_head(mon->qmp.qmp_responses);
@@ -555,10 +558,10 @@ static QObject *monitor_qmp_response_pop_one(Monitor *mon)
 
 static void monitor_qmp_response_flush(Monitor *mon)
 {
-    QObject *data;
+    QDict *data;
 
     while ((data = monitor_qmp_response_pop_one(mon))) {
-        monitor_json_emitter_raw(mon, data);
+        qmp_send_response(mon, data);
         qobject_unref(data);
     }
 }
@@ -570,7 +573,7 @@ static void monitor_qmp_response_flush(Monitor *mon)
 static bool monitor_qmp_response_pop_any(QMPResponse *response)
 {
     Monitor *mon;
-    QObject *data = NULL;
+    QDict *data = NULL;
 
     qemu_mutex_lock(&monitor_lock);
     QTAILQ_FOREACH(mon, &mon_list, entry) {
@@ -590,7 +593,7 @@ static void monitor_qmp_bh_responder(void *opaque)
     QMPResponse response;
 
     while (monitor_qmp_response_pop_any(&response)) {
-        monitor_json_emitter_raw(response.mon, response.data);
+        qmp_send_response(response.mon, response.data);
         qobject_unref(response.data);
     }
 }
@@ -606,8 +609,9 @@ static MonitorQAPIEventConf monitor_qapi_event_conf[QAPI_EVENT__MAX] = {
 };
 
 /*
- * Emits the event to every monitor instance, @event is only used for trace
- * Called with monitor_lock held.
+ * Broadcast an event to all monitors.
+ * @qdict is the event object.  Its member "event" must match @event.
+ * Caller must hold monitor_lock.
  */
 static void monitor_qapi_event_emit(QAPIEvent event, QDict *qdict)
 {
@@ -617,7 +621,7 @@ static void monitor_qapi_event_emit(QAPIEvent event, QDict *qdict)
     QTAILQ_FOREACH(mon, &mon_list, entry) {
         if (monitor_is_qmp(mon)
             && mon->qmp.commands != &qmp_cap_negotiation_commands) {
-            monitor_json_emitter(mon, QOBJECT(qdict));
+            qmp_queue_response(mon, qdict);
         }
     }
 }
@@ -762,7 +766,7 @@ static void monitor_qapi_event_init(void)
 static void handle_hmp_command(Monitor *mon, const char *cmdline);
 
 static void monitor_data_init(Monitor *mon, bool skip_flush,
-                              bool use_io_thr)
+                              bool use_io_thread)
 {
     memset(mon, 0, sizeof(Monitor));
     qemu_mutex_init(&mon->mon_lock);
@@ -771,7 +775,7 @@ static void monitor_data_init(Monitor *mon, bool skip_flush,
     /* Use *mon_cmds by default. */
     mon->cmd_table = mon_cmds;
     mon->skip_flush = skip_flush;
-    mon->use_io_thr = use_io_thr;
+    mon->use_io_thread = use_io_thread;
     mon->qmp.qmp_requests = g_queue_new();
     mon->qmp.qmp_responses = g_queue_new();
 }
@@ -976,8 +980,7 @@ static int parse_cmdline(const char *cmdline,
 }
 
 /*
- * Returns true if the command can be executed in preconfig mode
- * i.e. it has the 'p' flag.
+ * Can command @cmd be executed in preconfig state?
  */
 static bool cmd_can_preconfig(const mon_cmd_t *cmd)
 {
@@ -1246,96 +1249,56 @@ static void monitor_init_qmp_commands(void)
                          qmp_marshal_qmp_capabilities, QCO_ALLOW_PRECONFIG);
 }
 
-static bool qmp_cap_enabled(Monitor *mon, QMPCapability cap)
-{
-    return mon->qmp.qmp_caps[cap];
-}
-
 static bool qmp_oob_enabled(Monitor *mon)
 {
-    return qmp_cap_enabled(mon, QMP_CAPABILITY_OOB);
-}
-
-static void qmp_caps_check(Monitor *mon, QMPCapabilityList *list,
-                           Error **errp)
-{
-    for (; list; list = list->next) {
-        assert(list->value < QMP_CAPABILITY__MAX);
-        switch (list->value) {
-        case QMP_CAPABILITY_OOB:
-            if (!mon->use_io_thr) {
-                /*
-                 * Out-Of-Band only works with monitors that are
-                 * running on dedicated IOThread.
-                 */
-                error_setg(errp, "This monitor does not support "
-                           "Out-Of-Band (OOB)");
-                return;
-            }
-            break;
-        default:
-            break;
-        }
-    }
+    return mon->qmp.capab[QMP_CAPABILITY_OOB];
 }
 
-/* This function should only be called after capabilities are checked. */
-static void qmp_caps_apply(Monitor *mon, QMPCapabilityList *list)
+static void monitor_qmp_caps_reset(Monitor *mon)
 {
-    for (; list; list = list->next) {
-        mon->qmp.qmp_caps[list->value] = true;
-    }
+    memset(mon->qmp.capab_offered, 0, sizeof(mon->qmp.capab_offered));
+    memset(mon->qmp.capab, 0, sizeof(mon->qmp.capab));
+    mon->qmp.capab_offered[QMP_CAPABILITY_OOB] = mon->use_io_thread;
 }
 
 /*
- * Return true if check successful, or false otherwise.  When false is
- * returned, detailed error will be in errp if provided.
+ * Accept QMP capabilities in @list for @mon.
+ * On success, set mon->qmp.capab[], and return true.
+ * On error, set @errp, and return false.
  */
-static bool qmp_cmd_oob_check(Monitor *mon, QDict *req, Error **errp)
+static bool qmp_caps_accept(Monitor *mon, QMPCapabilityList *list,
+                            Error **errp)
 {
-    const char *command;
-    QmpCommand *cmd;
+    GString *unavailable = NULL;
+    bool capab[QMP_CAPABILITY__MAX];
 
-    command = qdict_get_try_str(req, "execute");
-    if (!command) {
-        error_setg(errp, "Command field 'execute' missing");
-        return false;
-    }
+    memset(capab, 0, sizeof(capab));
 
-    cmd = qmp_find_command(mon->qmp.commands, command);
-    if (!cmd) {
-        if (mon->qmp.commands == &qmp_cap_negotiation_commands) {
-            error_set(errp, ERROR_CLASS_COMMAND_NOT_FOUND,
-                      "Expecting capabilities negotiation "
-                      "with 'qmp_capabilities'");
-        } else {
-            error_set(errp, ERROR_CLASS_COMMAND_NOT_FOUND,
-                      "The command %s has not been found", command);
+    for (; list; list = list->next) {
+        if (!mon->qmp.capab_offered[list->value]) {
+            if (!unavailable) {
+                unavailable = g_string_new(QMPCapability_str(list->value));
+            } else {
+                g_string_append_printf(unavailable, ", %s",
+                                      QMPCapability_str(list->value));
+            }
         }
-        return false;
+        capab[list->value] = true;
     }
 
-    if (qmp_is_oob(req)) {
-        if (!qmp_oob_enabled(mon)) {
-            error_setg(errp, "Please enable Out-Of-Band first "
-                       "for the session during capabilities negotiation");
-            return false;
-        }
-        if (!(cmd->options & QCO_ALLOW_OOB)) {
-            error_setg(errp, "The command %s does not support OOB",
-                       command);
-            return false;
-        }
+    if (unavailable) {
+        error_setg(errp, "Capability %s not available", unavailable->str);
+        g_string_free(unavailable, true);
+        return false;
     }
 
+    memcpy(mon->qmp.capab, capab, sizeof(capab));
     return true;
 }
 
 void qmp_qmp_capabilities(bool has_enable, QMPCapabilityList *enable,
                           Error **errp)
 {
-    Error *local_err = NULL;
-
     if (cur_mon->qmp.commands == &qmp_commands) {
         error_set(errp, ERROR_CLASS_COMMAND_NOT_FOUND,
                   "Capabilities negotiation is already complete, command "
@@ -1343,19 +1306,8 @@ void qmp_qmp_capabilities(bool has_enable, QMPCapabilityList *enable,
         return;
     }
 
-    /* Enable QMP capabilities provided by the client if applicable. */
-    if (has_enable) {
-        qmp_caps_check(cur_mon, enable, &local_err);
-        if (local_err) {
-            /*
-             * Failed check on any of the capabilities will fail the
-             * entire command (and thus not apply any of the other
-             * capabilities that were also requested).
-             */
-            error_propagate(errp, local_err);
-            return;
-        }
-        qmp_caps_apply(cur_mon, enable);
+    if (!qmp_caps_accept(cur_mon, enable, errp)) {
+        return;
     }
 
     cur_mon->qmp.commands = &qmp_commands;
@@ -2264,7 +2216,7 @@ void qmp_getfd(const char *fdname, Error **errp)
         tmp_fd = monfd->fd;
         monfd->fd = fd;
         qemu_mutex_unlock(&cur_mon->mon_lock);
-        /* Make sure close() is out of critical section */
+        /* Make sure close() is outside critical section */
         close(tmp_fd);
         return;
     }
@@ -2293,7 +2245,7 @@ void qmp_closefd(const char *fdname, Error **errp)
         g_free(monfd->name);
         g_free(monfd);
         qemu_mutex_unlock(&cur_mon->mon_lock);
-        /* Make sure close() is out of critical section */
+        /* Make sure close() is outside critical section */
         close(tmp_fd);
         return;
     }
@@ -4137,78 +4089,53 @@ static int monitor_can_read(void *opaque)
 }
 
 /*
- * 1. This function takes ownership of rsp, err, and id.
- * 2. rsp, err, and id may be NULL.
- * 3. If err != NULL then rsp must be NULL.
+ * Emit QMP response @rsp with ID @id to @mon.
+ * Null @rsp can only happen for commands with QCO_NO_SUCCESS_RESP.
+ * Nothing is emitted then.
  */
-static void monitor_qmp_respond(Monitor *mon, QObject *rsp,
-                                Error *err, QObject *id)
+static void monitor_qmp_respond(Monitor *mon, QDict *rsp, QObject *id)
 {
-    QDict *qdict = NULL;
-
-    if (err) {
-        assert(!rsp);
-        qdict = qdict_new();
-        qdict_put_obj(qdict, "error", qmp_build_error_object(err));
-        error_free(err);
-        rsp = QOBJECT(qdict);
-    }
-
     if (rsp) {
         if (id) {
-            qdict_put_obj(qobject_to(QDict, rsp), "id", qobject_ref(id));
+            qdict_put_obj(rsp, "id", qobject_ref(id));
         }
 
-        monitor_json_emitter(mon, rsp);
+        qmp_queue_response(mon, rsp);
     }
-
-    qobject_unref(id);
-    qobject_unref(rsp);
 }
 
-/*
- * Dispatch one single QMP request. The function will free the req_obj
- * and objects inside it before return.
- */
-static void monitor_qmp_dispatch_one(QMPRequest *req_obj)
+static void monitor_qmp_dispatch(Monitor *mon, QObject *req, QObject *id)
 {
-    Monitor *mon, *old_mon;
-    QObject *req, *rsp = NULL, *id;
-    bool need_resume;
-
-    req = req_obj->req;
-    mon = req_obj->mon;
-    id = req_obj->id;
-    need_resume = req_obj->need_resume;
-
-    g_free(req_obj);
-
-    if (trace_event_get_state_backends(TRACE_HANDLE_QMP_COMMAND)) {
-        QString *req_json = qobject_to_json(req);
-        trace_handle_qmp_command(mon, qstring_get_str(req_json));
-        qobject_unref(req_json);
-    }
+    Monitor *old_mon;
+    QDict *rsp;
+    QDict *error;
 
     old_mon = cur_mon;
     cur_mon = mon;
 
-    rsp = qmp_dispatch(mon->qmp.commands, req);
+    rsp = qmp_dispatch(mon->qmp.commands, req, qmp_oob_enabled(mon));
 
     cur_mon = old_mon;
 
-    /* Respond if necessary */
-    monitor_qmp_respond(mon, rsp, NULL, id);
-
-    /* This pairs with the monitor_suspend() in handle_qmp_command(). */
-    if (need_resume) {
-        monitor_resume(mon);
+    if (mon->qmp.commands == &qmp_cap_negotiation_commands) {
+        error = qdict_get_qdict(rsp, "error");
+        if (error
+            && !g_strcmp0(qdict_get_try_str(error, "class"),
+                    QapiErrorClass_str(ERROR_CLASS_COMMAND_NOT_FOUND))) {
+            /* Provide a more useful error message */
+            qdict_del(error, "desc");
+            qdict_put_str(error, "desc", "Expecting capabilities negotiation"
+                          " with 'qmp_capabilities'");
+        }
     }
 
-    qobject_unref(req);
+    monitor_qmp_respond(mon, rsp, id);
+    qobject_unref(rsp);
 }
 
 /*
- * Pop one QMP request from monitor queues, return NULL if not found.
+ * Pop a QMP request from a monitor request queue.
+ * Return the request, or NULL all request queues are empty.
  * We are using round-robin fashion to pop the request, to avoid
  * processing commands only on a very busy monitor.  To achieve that,
  * when we process one request on a specific monitor, we put that
@@ -4247,13 +4174,30 @@ static QMPRequest *monitor_qmp_requests_pop_any(void)
 static void monitor_qmp_bh_dispatcher(void *data)
 {
     QMPRequest *req_obj = monitor_qmp_requests_pop_any();
+    QDict *rsp;
 
-    if (req_obj) {
+    if (!req_obj) {
+        return;
+    }
+
+    if (req_obj->req) {
         trace_monitor_qmp_cmd_in_band(qobject_get_try_str(req_obj->id) ?: "");
-        monitor_qmp_dispatch_one(req_obj);
-        /* Reschedule instead of looping so the main loop stays responsive */
-        qemu_bh_schedule(mon_global.qmp_dispatcher_bh);
+        monitor_qmp_dispatch(req_obj->mon, req_obj->req, req_obj->id);
+    } else {
+        assert(req_obj->err);
+        rsp = qmp_error_response(req_obj->err);
+        monitor_qmp_respond(req_obj->mon, rsp, NULL);
+        qobject_unref(rsp);
+    }
+
+    if (req_obj->need_resume) {
+        /* Pairs with the monitor_suspend() in handle_qmp_command() */
+        monitor_resume(req_obj->mon);
     }
+    qmp_request_free(req_obj);
+
+    /* Reschedule instead of looping so the main loop stays responsive */
+    qemu_bh_schedule(qmp_dispatcher_bh);
 }
 
 #define  QMP_REQ_QUEUE_LEN_MAX  (8)
@@ -4261,7 +4205,7 @@ static void monitor_qmp_bh_dispatcher(void *data)
 static void handle_qmp_command(JSONMessageParser *parser, GQueue *tokens)
 {
     QObject *req, *id = NULL;
-    QDict *qdict = NULL;
+    QDict *qdict;
     MonitorQMP *mon_qmp = container_of(parser, MonitorQMP, parser);
     Monitor *mon = container_of(mon_qmp, Monitor, qmp);
     Error *err = NULL;
@@ -4272,46 +4216,34 @@ static void handle_qmp_command(JSONMessageParser *parser, GQueue *tokens)
         /* json_parser_parse_err() sucks: can fail without setting @err */
         error_setg(&err, QERR_JSON_PARSING);
     }
-    if (err) {
-        goto err;
-    }
 
-    /* Check against the request in general layout */
-    qdict = qmp_dispatch_check_obj(req, &err);
-    if (!qdict) {
-        goto err;
-    }
+    qdict = qobject_to(QDict, req);
+    if (qdict) {
+        id = qobject_ref(qdict_get(qdict, "id"));
+        qdict_del(qdict, "id");
+    } /* else will fail qmp_dispatch() */
 
-    /* Check against OOB specific */
-    if (!qmp_cmd_oob_check(mon, qdict, &err)) {
-        goto err;
+    if (trace_event_get_state_backends(TRACE_HANDLE_QMP_COMMAND)) {
+        QString *req_json = qobject_to_json(req);
+        trace_handle_qmp_command(mon, qstring_get_str(req_json));
+        qobject_unref(req_json);
     }
 
-    id = qdict_get(qdict, "id");
-
-    /* When OOB is enabled, the "id" field is mandatory. */
-    if (qmp_oob_enabled(mon) && !id) {
-        error_setg(&err, "Out-Of-Band capability requires that "
-                   "every command contains an 'id' field");
-        goto err;
+    if (qdict && qmp_is_oob(qdict)) {
+        /* OOB commands are executed immediately */
+        trace_monitor_qmp_cmd_out_of_band(qobject_get_try_str(id)
+                                          ?: "");
+        monitor_qmp_dispatch(mon, req, id);
+        return;
     }
 
     req_obj = g_new0(QMPRequest, 1);
     req_obj->mon = mon;
-    req_obj->id = qobject_ref(id);
+    req_obj->id = id;
     req_obj->req = req;
+    req_obj->err = err;
     req_obj->need_resume = false;
 
-    qdict_del(qdict, "id");
-
-    if (qmp_is_oob(qdict)) {
-        /* Out-Of-Band (OOB) requests are executed directly in parser. */
-        trace_monitor_qmp_cmd_out_of_band(qobject_get_try_str(req_obj->id)
-                                          ?: "");
-        monitor_qmp_dispatch_one(req_obj);
-        return;
-    }
-
     /* Protect qmp_requests and fetching its length. */
     qemu_mutex_lock(&mon->qmp.qmp_queue_lock);
 
@@ -4328,6 +4260,12 @@ static void handle_qmp_command(JSONMessageParser *parser, GQueue *tokens)
         /* Drop the request if queue is full. */
         if (mon->qmp.qmp_requests->length >= QMP_REQ_QUEUE_LEN_MAX) {
             qemu_mutex_unlock(&mon->qmp.qmp_queue_lock);
+            /*
+             * FIXME @id's scope is just @mon, and broadcasting it is
+             * wrong.  If another monitor's client has a command with
+             * the same ID in flight, the event will incorrectly claim
+             * that command was dropped.
+             */
             qapi_event_send_command_dropped(id,
                                             COMMAND_DROP_REASON_QUEUE_FULL,
                                             &error_abort);
@@ -4345,12 +4283,7 @@ static void handle_qmp_command(JSONMessageParser *parser, GQueue *tokens)
     qemu_mutex_unlock(&mon->qmp.qmp_queue_lock);
 
     /* Kick the dispatcher routine */
-    qemu_bh_schedule(mon_global.qmp_dispatcher_bh);
-    return;
-
-err:
-    monitor_qmp_respond(mon, NULL, err, NULL);
-    qobject_unref(req);
+    qemu_bh_schedule(qmp_dispatcher_bh);
 }
 
 static void monitor_qmp_read(void *opaque, const uint8_t *buf, int size)
@@ -4400,10 +4333,10 @@ int monitor_suspend(Monitor *mon)
 
     if (monitor_is_qmp(mon)) {
         /*
-         * Kick iothread to make sure this takes effect.  It'll be
+         * Kick I/O thread to make sure this takes effect.  It'll be
          * evaluated again in prepare() of the watch object.
          */
-        aio_notify(iothread_get_aio_context(mon_global.mon_iothread));
+        aio_notify(iothread_get_aio_context(mon_iothread));
     }
 
     trace_monitor_suspend(mon, 1);
@@ -4419,11 +4352,11 @@ void monitor_resume(Monitor *mon)
     if (atomic_dec_fetch(&mon->suspend_cnt) == 0) {
         if (monitor_is_qmp(mon)) {
             /*
-             * For QMP monitors that are running in IOThread, let's
-             * kick the thread in case it's sleeping.
+             * For QMP monitors that are running in the I/O thread,
+             * let's kick the thread in case it's sleeping.
              */
-            if (mon->use_io_thr) {
-                aio_notify(iothread_get_aio_context(mon_global.mon_iothread));
+            if (mon->use_io_thread) {
+                aio_notify(iothread_get_aio_context(mon_iothread));
             }
         } else {
             assert(mon->rs);
@@ -4433,7 +4366,7 @@ void monitor_resume(Monitor *mon)
     trace_monitor_suspend(mon, -1);
 }
 
-static QObject *get_qmp_greeting(Monitor *mon)
+static QDict *qmp_greeting(Monitor *mon)
 {
     QList *cap_list = qlist_new();
     QObject *ver = NULL;
@@ -4442,33 +4375,27 @@ static QObject *get_qmp_greeting(Monitor *mon)
     qmp_marshal_query_version(NULL, &ver, NULL);
 
     for (cap = 0; cap < QMP_CAPABILITY__MAX; cap++) {
-        if (!mon->use_io_thr && cap == QMP_CAPABILITY_OOB) {
-            /* Monitors that are not using IOThread won't support OOB */
-            continue;
+        if (mon->qmp.capab_offered[cap]) {
+            qlist_append_str(cap_list, QMPCapability_str(cap));
         }
-        qlist_append_str(cap_list, QMPCapability_str(cap));
     }
 
-    return qobject_from_jsonf("{'QMP': {'version': %p, 'capabilities': %p}}",
-                              ver, cap_list);
-}
-
-static void monitor_qmp_caps_reset(Monitor *mon)
-{
-    memset(mon->qmp.qmp_caps, 0, sizeof(mon->qmp.qmp_caps));
+    return qdict_from_jsonf_nofail(
+        "{'QMP': {'version': %p, 'capabilities': %p}}",
+        ver, cap_list);
 }
 
 static void monitor_qmp_event(void *opaque, int event)
 {
-    QObject *data;
+    QDict *data;
     Monitor *mon = opaque;
 
     switch (event) {
     case CHR_EVENT_OPENED:
         mon->qmp.commands = &qmp_cap_negotiation_commands;
         monitor_qmp_caps_reset(mon);
-        data = get_qmp_greeting(mon);
-        monitor_json_emitter(mon, data);
+        data = qmp_greeting(mon);
+        qmp_queue_response(mon, data);
         qobject_unref(data);
         mon_refcount++;
         break;
@@ -4561,36 +4488,35 @@ static void sortcmdlist(void)
 
 static GMainContext *monitor_get_io_context(void)
 {
-    return iothread_get_g_main_context(mon_global.mon_iothread);
+    return iothread_get_g_main_context(mon_iothread);
 }
 
 static AioContext *monitor_get_aio_context(void)
 {
-    return iothread_get_aio_context(mon_global.mon_iothread);
+    return iothread_get_aio_context(mon_iothread);
 }
 
 static void monitor_iothread_init(void)
 {
-    mon_global.mon_iothread = iothread_create("mon_iothread",
-                                              &error_abort);
+    mon_iothread = iothread_create("mon_iothread", &error_abort);
 
     /*
-     * This MUST be on main loop thread since we have commands that
-     * have assumption to be run on main loop thread.  It would be
-     * nice that one day we can remove this assumption in the future.
+     * The dispatcher BH must run in the main loop thread, since we
+     * have commands assuming that context.  It would be nice to get
+     * rid of those assumptions.
      */
-    mon_global.qmp_dispatcher_bh = aio_bh_new(iohandler_get_aio_context(),
-                                              monitor_qmp_bh_dispatcher,
-                                              NULL);
+    qmp_dispatcher_bh = aio_bh_new(iohandler_get_aio_context(),
+                                   monitor_qmp_bh_dispatcher,
+                                   NULL);
 
     /*
-     * Unlike the dispatcher BH, this must be run on the monitor IO
-     * thread, so that monitors that are using IO thread will make
-     * sure read/write operations are all done on the IO thread.
+     * The responder BH must be run in the monitor I/O thread, so that
+     * monitors that are using the I/O thread have their output
+     * written by the I/O thread.
      */
-    mon_global.qmp_respond_bh = aio_bh_new(monitor_get_aio_context(),
-                                           monitor_qmp_bh_responder,
-                                           NULL);
+    qmp_respond_bh = aio_bh_new(monitor_get_aio_context(),
+                                monitor_qmp_bh_responder,
+                                NULL);
 }
 
 void monitor_init_globals(void)
@@ -4655,16 +4581,12 @@ static void monitor_qmp_setup_handlers_bh(void *opaque)
     Monitor *mon = opaque;
     GMainContext *context;
 
-    if (mon->use_io_thr) {
-        /*
-         * When use_io_thr is set, we use the global shared dedicated
-         * IO thread for this monitor to handle input/output.
-         */
+    if (mon->use_io_thread) {
+        /* Use @mon_iothread context */
         context = monitor_get_io_context();
-        /* We should have inited globals before reaching here. */
         assert(context);
     } else {
-        /* The default main loop, which is the main thread */
+        /* Use default main loop context */
         context = NULL;
     }
 
@@ -4681,12 +4603,12 @@ void monitor_init(Chardev *chr, int flags)
 
     if (use_oob) {
         if (CHARDEV_IS_MUX(chr)) {
-            error_report("Monitor Out-Of-Band is not supported with "
+            error_report("Monitor out-of-band is not supported with "
                          "MUX typed chardev backend");
             exit(1);
         }
         if (use_readline) {
-            error_report("Monitor Out-Of-band is only supported by QMP");
+            error_report("Monitor out-of-band is only supported by QMP");
             exit(1);
         }
     }
@@ -4706,7 +4628,7 @@ void monitor_init(Chardev *chr, int flags)
     if (monitor_is_qmp(mon)) {
         qemu_chr_fe_set_echo(&mon->chr, true);
         json_message_parser_init(&mon->qmp.parser, handle_qmp_command);
-        if (mon->use_io_thr) {
+        if (mon->use_io_thread) {
             /*
              * Make sure the old iowatch is gone.  It's possible when
              * e.g. the chardev is in client mode, with wait=on.
@@ -4714,15 +4636,12 @@ void monitor_init(Chardev *chr, int flags)
             remove_fd_in_watch(chr);
             /*
              * We can't call qemu_chr_fe_set_handlers() directly here
-             * since during the procedure the chardev will be active
-             * and running in monitor iothread, while we'll still do
-             * something before returning from it, which is a possible
-             * race too.  To avoid that, we just create a BH to setup
-             * the handlers.
+             * since chardev might be running in the monitor I/O
+             * thread.  Schedule a bottom half.
              */
             aio_bh_schedule_oneshot(monitor_get_aio_context(),
                                     monitor_qmp_setup_handlers_bh, mon);
-            /* We'll add this to mon_list in the BH when setup done */
+            /* The bottom half will add @mon to @mon_list */
             return;
         } else {
             qemu_chr_fe_set_handlers(&mon->chr, monitor_can_read,
@@ -4742,22 +4661,20 @@ void monitor_cleanup(void)
     Monitor *mon, *next;
 
     /*
-     * We need to explicitly stop the iothread (but not destroy it),
-     * cleanup the monitor resources, then destroy the iothread since
+     * We need to explicitly stop the I/O thread (but not destroy it),
+     * clean up the monitor resources, then destroy the I/O thread since
      * we need to unregister from chardev below in
      * monitor_data_destroy(), and chardev is not thread-safe yet
      */
-    iothread_stop(mon_global.mon_iothread);
+    iothread_stop(mon_iothread);
 
     /*
-     * After we have IOThread to send responses, it's possible that
-     * when we stop the IOThread there are still replies queued in the
-     * responder queue.  Flush all of them.  Note that even after this
-     * flush it's still possible that out buffer is not flushed.
-     * It'll be done in below monitor_flush() as the last resort.
+     * Flush all response queues.  Note that even after this flush,
+     * data may remain in output buffers.
      */
     monitor_qmp_bh_responder(NULL);
 
+    /* Flush output buffers and destroy monitors */
     qemu_mutex_lock(&monitor_lock);
     QTAILQ_FOREACH_SAFE(mon, &mon_list, entry, next) {
         QTAILQ_REMOVE(&mon_list, mon, entry);
@@ -4767,14 +4684,14 @@ void monitor_cleanup(void)
     }
     qemu_mutex_unlock(&monitor_lock);
 
-    /* QEMUBHs needs to be deleted before destroying the IOThread. */
-    qemu_bh_delete(mon_global.qmp_dispatcher_bh);
-    mon_global.qmp_dispatcher_bh = NULL;
-    qemu_bh_delete(mon_global.qmp_respond_bh);
-    mon_global.qmp_respond_bh = NULL;
+    /* QEMUBHs needs to be deleted before destroying the I/O thread */
+    qemu_bh_delete(qmp_dispatcher_bh);
+    qmp_dispatcher_bh = NULL;
+    qemu_bh_delete(qmp_respond_bh);
+    qmp_respond_bh = NULL;
 
-    iothread_destroy(mon_global.mon_iothread);
-    mon_global.mon_iothread = NULL;
+    iothread_destroy(mon_iothread);
+    mon_iothread = NULL;
 }
 
 QemuOptsList qemu_mon_opts = {