monitor/qmp.c

/*
 * QEMU monitor
 *
 * Copyright (c) 2003-2004 Fabrice Bellard
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
 * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */

#include "qemu/osdep.h"
#include "qemu/aio-wait.h"
#include "qemu/lockable.h"
#include "chardev/char-io.h"
#include "monitor-internal.h"
#include "qapi/error.h"
#include "qapi/qapi-commands-control.h"
#include "qapi/qapi-commands-char.h"
#include "qobject/qdict.h"
#include "qobject/qjson.h"
#include "qobject/qlist.h"
#include "qom/object_interfaces.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 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 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.
 *
 * Access must be atomic for thread safety.
 */
static bool qmp_dispatcher_co_busy = true;

struct QMPRequest {
    /* Owner of the request */
    MonitorQMP *mon;
    /*
     * Request object to be handled or Error to be reported
     * (exactly one of them is non-null)
     */
    QObject *req;
    Error *err;
};
typedef struct QMPRequest QMPRequest;

QmpCommandList qmp_commands, qmp_cap_negotiation_commands;

/* Monitor being serviced by the dispatcher.  Protected by BQL. */
static MonitorQMP *qmp_dispatcher_current_mon;

OBJECT_DEFINE_TYPE(MonitorQMP, monitor_qmp, MONITOR_QMP, MONITOR);

static void monitor_qmp_cleanup_req_queue_locked(MonitorQMP *mon);

static void monitor_qmp_finalize(Object *obj)
{
    MonitorQMP *mon = MONITOR_QMP(obj);

    json_message_parser_destroy(&mon->parser);
    qemu_mutex_destroy(&mon->qmp_queue_lock);
    monitor_qmp_cleanup_req_queue_locked(mon);
    g_queue_free(mon->qmp_requests);
}

static bool monitor_qmp_get_pretty(Object *obj, Error **errp)
{
    MonitorQMP *mon = MONITOR_QMP(obj);

    return mon->pretty;
}

static void monitor_qmp_set_pretty(Object *obj, bool val, Error **errp)
{
    MonitorQMP *mon = MONITOR_QMP(obj);

    mon->pretty = val;
}

static int monitor_qmp_get_close_action(Object *obj, Error **errp)
{
    MonitorQMP *mon = MONITOR_QMP(obj);

    return mon->close_action;
}

static void monitor_qmp_set_close_action(Object *obj, int val, Error **errp)
{
    MonitorQMP *mon = MONITOR_QMP(obj);

    mon->close_action = val;
}

static void monitor_qmp_emit_event(Monitor *mon, QAPIEvent event, QDict *qdict);
static bool monitor_qmp_requires_iothread(const Monitor *mon);
static void monitor_qmp_complete(UserCreatable *uc, Error **errp);
static bool monitor_qmp_prepare_delete(UserCreatable *uc, Error **errp);
static void monitor_qmp_accept_input(Monitor *mon);

static void monitor_qmp_class_init(ObjectClass *cls, const void *data)
{
    MonitorClass *moncls = MONITOR_CLASS(cls);
    UserCreatableClass *ucc = USER_CREATABLE_CLASS(cls);

    object_class_property_add_bool(cls, "pretty",
                                   monitor_qmp_get_pretty,
                                   monitor_qmp_set_pretty);
    object_class_property_add_enum(cls, "close-action",
                                   "MonitorQMPCloseAction",
                                   &MonitorQMPCloseAction_lookup,
                                   monitor_qmp_get_close_action,
                                   monitor_qmp_set_close_action);

    moncls->emit_event = monitor_qmp_emit_event;
    moncls->requires_iothread = monitor_qmp_requires_iothread;
    moncls->accept_input = monitor_qmp_accept_input;

    ucc->complete = monitor_qmp_complete;
    ucc->prepare_delete = monitor_qmp_prepare_delete;
}

static void handle_qmp_command(void *opaque, QObject *req, Error *err);
static void monitor_qmp_init(Object *obj)
{
    MonitorQMP *mon = MONITOR_QMP(obj);

    qemu_mutex_init(&mon->qmp_queue_lock);
    mon->qmp_requests = g_queue_new();

    json_message_parser_init(&mon->parser, handle_qmp_command, mon, NULL);
}

static void monitor_qmp_emit_event(Monitor *mon, QAPIEvent event, QDict *qdict)
{
    MonitorQMP *qmp = MONITOR_QMP(mon);

    WITH_QEMU_LOCK_GUARD(&mon->mon_lock) {
        if (qmp->commands == &qmp_cap_negotiation_commands) {
            return;
        }
    }

    qmp_send_response(qmp, qdict);
}

static bool monitor_qmp_requires_iothread(const Monitor *mon)
{
    return qemu_chr_has_feature(mon->chr.chr,
                                QEMU_CHAR_FEATURE_GCONTEXT);
}

static bool qmp_oob_enabled(MonitorQMP *mon)
{
    return mon->capab[QMP_CAPABILITY_OOB];
}

static void monitor_qmp_caps_reset(MonitorQMP *mon)
{
    memset(mon->capab_offered, 0, sizeof(mon->capab_offered));
    memset(mon->capab, 0, sizeof(mon->capab));
    mon->capab_offered[QMP_CAPABILITY_OOB] =
        monitor_requires_iothread(MONITOR(mon));
}

static void qmp_request_free(QMPRequest *req)
{
    qobject_unref(req->req);
    error_free(req->err);
    g_free(req);
}

/* Caller must hold mon->qmp.qmp_queue_lock */
static void monitor_qmp_cleanup_req_queue_locked(MonitorQMP *mon)
{
    while (!g_queue_is_empty(mon->qmp_requests)) {
        qmp_request_free(g_queue_pop_head(mon->qmp_requests));
    }
}

static void monitor_qmp_drain_queue(MonitorQMP *mon)
{
    QEMU_LOCK_GUARD(&mon->qmp_queue_lock);
    monitor_qmp_cleanup_req_queue_locked(mon);
}

static void monitor_qmp_cleanup_queue_and_resume(MonitorQMP *mon)
{
    QEMU_LOCK_GUARD(&mon->qmp_queue_lock);

    /*
     * Same condition as in monitor_qmp_dispatcher_co(), but before
     * removing an element from the queue (hence no `- 1`).
     * Also, the queue should not be empty either, otherwise the
     * monitor hasn't been suspended yet (or was already resumed).
     */
    bool need_resume = (!qmp_oob_enabled(mon) ||
        mon->qmp_requests->length == QMP_REQ_QUEUE_LEN_MAX)
        && !g_queue_is_empty(mon->qmp_requests);

    monitor_qmp_cleanup_req_queue_locked(mon);

    if (need_resume) {
        /*
         * handle_qmp_command() suspended the monitor because the
         * request queue filled up, to be resumed when the queue has
         * space again.  We just emptied it; resume the monitor.
         *
         * Without this, the monitor would remain suspended forever
         * when we get here while the monitor is suspended.  An
         * unfortunately timed CHR_EVENT_CLOSED can do the trick.
         */
        monitor_resume(&mon->parent_obj);
    }

}

void qmp_send_response(MonitorQMP *mon, const QDict *rsp)
{
    const QObject *data = QOBJECT(rsp);
    GString *json;

    json = qobject_to_json_pretty(data, mon->pretty);
    assert(json != NULL);
    trace_monitor_qmp_respond(mon, json->str);

    g_string_append_c(json, '\n');
    monitor_puts(&mon->parent_obj, json->str);

    g_string_free(json, true);
}

/*
 * Emit QMP response @rsp to @mon.
 * Null @rsp can only happen for commands with QCO_NO_SUCCESS_RESP.
 * Nothing is emitted then.
 */
static void monitor_qmp_respond(MonitorQMP *mon, QDict *rsp)
{
    if (rsp) {
        qmp_send_response(mon, rsp);
    }
}

/*
 * Runs outside of coroutine context for OOB commands, but in
 * coroutine context for everything else.
 */
static void monitor_qmp_dispatch(MonitorQMP *mon, QObject *req)
{
    QDict *rsp;
    QDict *error;

    rsp = qmp_dispatch(mon->commands, req, qmp_oob_enabled(mon),
                       &mon->parent_obj);

    if (mon->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'");
        }
    }

    monitor_qmp_respond(mon, rsp);
    qobject_unref(rsp);
}

/*
 * 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
 * monitor to the end of mon_list queue.
 *
 * Note: if the function returned with non-NULL, then the caller will
 * be with qmp_mon->qmp_queue_lock held, and the caller is responsible
 * to release it.
 */
static QMPRequest *monitor_qmp_requests_pop_any_with_lock(void)
{
    QMPRequest *req_obj = NULL;
    Monitor *mon;
    MonitorQMP *qmp_mon;

    QTAILQ_FOREACH(mon, &mon_list, entry) {
        qmp_mon = MONITOR_QMP(
            object_dynamic_cast(OBJECT(mon), TYPE_MONITOR_QMP));
        if (!qmp_mon) {
            continue;
        }

        qemu_mutex_lock(&qmp_mon->qmp_queue_lock);
        req_obj = g_queue_pop_head(qmp_mon->qmp_requests);
        if (req_obj) {
            /* With the lock of corresponding queue held */
            break;
        }
        qemu_mutex_unlock(&qmp_mon->qmp_queue_lock);
    }

    if (req_obj) {
        /*
         * We found one request on the monitor. Degrade this monitor's
         * priority to lowest by re-inserting it to end of queue.
         */
        QTAILQ_REMOVE(&mon_list, mon, entry);
        QTAILQ_INSERT_TAIL(&mon_list, mon, entry);
    }

    return req_obj;
}

static QMPRequest * coroutine_fn
monitor_qmp_dispatcher_pop_any(void)
{
    while (true) {
        /*
         * 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_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_set_mb(&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;
            }
        }

        /*
         * 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);

        /*
         * @req_obj has a request, we hold req_obj->mon->qmp_queue_lock
         */

        mon = req_obj->mon;
        qmp_dispatcher_current_mon = mon;

        /*
         * We need to resume the monitor if handle_qmp_command()
         * suspended it.  Two cases:
         * 1. OOB enabled: mon->qmp_requests has no more space
         *    Resume right away, so that OOB commands can get executed while
         *    this request is being processed.
         * 2. OOB disabled: always
         *    Resume only after we're done processing the request,
         * We need to save qmp_oob_enabled() for later, because
         * qmp_qmp_capabilities() can change it.
         */
        oob_enabled = qmp_oob_enabled(mon);
        if (oob_enabled
            && mon->qmp_requests->length == QMP_REQ_QUEUE_LEN_MAX - 1) {
            monitor_resume(&mon->parent_obj);
        }

        /*
         * Drop the queue mutex now, before yielding, otherwise we might
         * deadlock if the main thread tries to lock it.
         */
        qemu_mutex_unlock(&mon->qmp_queue_lock);

        if (qatomic_xchg(&qmp_dispatcher_co_busy, true) == true) {
            /*
             * Someone rescheduled us (probably because a new requests
             * came in), but we didn't actually yield. Do that now,
             * only to be immediately reentered and removed from the
             * list of scheduled coroutines.
             */
            qemu_coroutine_yield();
        }

        /* Process request */
        if (req_obj->req) {
            if (trace_event_get_state(TRACE_MONITOR_QMP_CMD_IN_BAND)) {
                QDict *qdict = qobject_to(QDict, req_obj->req);
                QObject *id = qdict ? qdict_get(qdict, "id") : NULL;
                GString *id_json;

                id_json = id ? qobject_to_json(id) : g_string_new(NULL);
                trace_monitor_qmp_cmd_in_band(id_json->str);
                g_string_free(id_json, true);
            }
            monitor_qmp_dispatch(mon, req_obj->req);
        } else {
            assert(req_obj->err);
            trace_monitor_qmp_err_in_band(error_get_pretty(req_obj->err));
            rsp = qmp_error_response(req_obj->err);
            req_obj->err = NULL;
            monitor_qmp_respond(mon, rsp);
            qobject_unref(rsp);
        }

        if (!oob_enabled) {
            monitor_resume(&mon->parent_obj);
        }

        qmp_request_free(req_obj);
        qmp_dispatcher_current_mon = NULL;
    }
    qatomic_set(&qmp_dispatcher_co, NULL);
}

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) &&
            qatomic_read(&qmp_dispatcher_co)) {
        aio_co_wake(qmp_dispatcher_co);
    }
}

static void handle_qmp_command(void *opaque, QObject *req, Error *err)
{
    MonitorQMP *mon = opaque;
    QDict *qdict = qobject_to(QDict, req);
    QMPRequest *req_obj;

    assert(!req != !err);

    if (req && trace_event_get_state_backends(TRACE_HANDLE_QMP_COMMAND)) {
        GString *req_json = qobject_to_json(req);
        trace_handle_qmp_command(mon, req_json->str);
        g_string_free(req_json, true);
    }

    if (qdict && qmp_is_oob(qdict)) {
        /* OOB commands are executed immediately */
        if (trace_event_get_state(TRACE_MONITOR_QMP_CMD_OUT_OF_BAND)) {
            QObject *id = qdict_get(qdict, "id");
            GString *id_json;

            id_json = id ? qobject_to_json(id) : g_string_new(NULL);
            trace_monitor_qmp_cmd_out_of_band(id_json->str);
            g_string_free(id_json, true);
        }
        monitor_qmp_dispatch(mon, req);
        qobject_unref(req);
        return;
    }

    req_obj = g_new0(QMPRequest, 1);
    req_obj->mon = mon;
    req_obj->req = req;
    req_obj->err = err;

    /* Protect qmp_requests and fetching its length. */
    WITH_QEMU_LOCK_GUARD(&mon->qmp_queue_lock) {

        /*
         * Suspend the monitor when we can't queue more requests after
         * this one.  Dequeuing in monitor_qmp_dispatcher_co() or
         * monitor_qmp_cleanup_queue_and_resume() will resume it.
         * Note that when OOB is disabled, we queue at most one command,
         * for backward compatibility.
         */
        if (!qmp_oob_enabled(mon) ||
            mon->qmp_requests->length == QMP_REQ_QUEUE_LEN_MAX - 1) {
            monitor_suspend(&mon->parent_obj);
        }

        /*
         * Put the request to the end of queue so that requests will be
         * handled in time order.  Ownership for req_obj, req,
         * etc. will be delivered to the handler side.
         */
        trace_monitor_qmp_in_band_enqueue(req_obj, mon,
                                          mon->qmp_requests->length);
        assert(mon->qmp_requests->length < QMP_REQ_QUEUE_LEN_MAX);
        g_queue_push_tail(mon->qmp_requests, req_obj);
    }

    /* Kick the dispatcher routine */
    qmp_dispatcher_co_wake();
}

static void monitor_qmp_read(void *opaque, const uint8_t *buf, int size)
{
    MonitorQMP *mon = opaque;

    json_message_parser_feed(&mon->parser, (const char *) buf, size);
}

static QDict *qmp_greeting(MonitorQMP *mon)
{
    QList *cap_list = qlist_new();
    QObject *ver = NULL;
    QDict *args;
    QMPCapability cap;

    args = qdict_new();
    qmp_marshal_query_version(args, &ver, NULL);
    qobject_unref(args);

    for (cap = 0; cap < QMP_CAPABILITY__MAX; cap++) {
        if (mon->capab_offered[cap]) {
            qlist_append_str(cap_list, QMPCapability_str(cap));
        }
    }

    return qdict_from_jsonf_nofail(
        "{'QMP': {'version': %p, 'capabilities': %p}}",
        ver, cap_list);
}

static void monitor_qmp_self_delete_bh(void *opaque)
{
    MonitorQMP *mon = opaque;
    const char *mon_id = object_get_canonical_path_component(
        OBJECT(mon));
    g_autofree char *chardev_id = g_strdup(mon->parent_obj.chardev_id);
    Error *local_error = NULL;

    if (!mon_id) {
        /*
         * Another monitor raced & ran 'object-del' on 'mon'
         * before this BH got scheduled, so we have a ref on
         * mon from monitor_qmp_event but it is already
         * unparented.
         */
        object_unref(mon);
        return;
    }

    user_creatable_del(mon_id, &local_error);
    /* Pairs with ref from monitor_qmp_event */
    object_unref(mon);
    if (local_error != NULL) {
        error_report_err(local_error);
    } else {
        qmp_chardev_remove(chardev_id, NULL);
    }
}

static void monitor_qmp_event(void *opaque, QEMUChrEvent event)
{
    QDict *data;
    MonitorQMP *mon = opaque;

    /*
     * Protect against race if a client drops & quickly
     * reconnects - we'll have the delete BH scheduled
     * so must not honour a new open request
     */
    if (mon->delete_pending) {
        return;
    }

    switch (event) {
    case CHR_EVENT_OPENED:
        WITH_QEMU_LOCK_GUARD(&mon->parent_obj.mon_lock) {
            mon->commands = &qmp_cap_negotiation_commands;
            monitor_qmp_caps_reset(mon);
        }
        data = qmp_greeting(mon);
        qmp_send_response(mon, data);
        qobject_unref(data);
        break;
    case CHR_EVENT_CLOSED:
        /*
         * Note: this is only useful when the output of the chardev
         * backend is still open.  For example, when the backend is
         * stdio, it's possible that stdout is still open when stdin
         * is closed.
         */
        monitor_qmp_cleanup_queue_and_resume(mon);
        json_message_parser_destroy(&mon->parser);
        json_message_parser_init(&mon->parser, handle_qmp_command,
                                 mon, NULL);
        monitor_fdsets_cleanup();
        switch (mon->close_action) {
        case MONITOR_QMP_CLOSE_ACTION_NONE:
            break;
        case MONITOR_QMP_CLOSE_ACTION_DELETE:
            mon->delete_pending = true;
            /*
             * Do NOT run in the AIO context associated with the
             * monitor. We need to run in the default AIO context
             * which is the same context in which 'qmp_object_del'
             * will execute
             *
             * Hold an extra ref in case a separate monitor races
             * with the BH by processing an explicit 'object-del'.
             * Will be released by monitor_qmp_self_delete_bh
             */
            object_ref(mon);
            aio_bh_schedule_oneshot(qemu_get_aio_context(),
                                    monitor_qmp_self_delete_bh, mon);
            break;
        default:
            g_assert_not_reached();
        }
        break;
    case CHR_EVENT_BREAK:
    case CHR_EVENT_MUX_IN:
    case CHR_EVENT_MUX_OUT:
    case CHR_EVENT_RESIZE:
        /* Ignore */
        break;
    }
}

static bool monitor_qmp_dispatcher_is_servicing(MonitorQMP *mon)
{
    return qmp_dispatcher_current_mon == mon;
}

static void monitor_qmp_setup_handlers_bh(void *opaque)
{
    MonitorQMP *mon = opaque;
    GMainContext *context;

    assert(monitor_requires_iothread(MONITOR(mon)));
    context = iothread_get_g_main_context(mon_iothread);
    assert(context);
    qemu_chr_fe_set_handlers(&mon->parent_obj.chr, monitor_can_read,
                             monitor_qmp_read, monitor_qmp_event,
                             NULL, &mon->parent_obj, context, true);
    monitor_list_append(&mon->parent_obj);
    qatomic_set(&mon->setup_pending, false);
}

void monitor_new_qmp(const char *id, const char *chardev_id,
                     bool pretty, Error **errp)
{
    g_autofree char *autoid = id ? NULL : monitor_compat_id();
    object_new_with_props(TYPE_MONITOR_QMP,
                          object_get_objects_root(),
                          id ? id : autoid,
                          errp,
                          "chardev", chardev_id,
                          "pretty", pretty ? "yes" : "no",
                          NULL);
}

static void monitor_qmp_complete(UserCreatable *uc, Error **errp)
{
    MonitorQMP *mon = MONITOR_QMP(uc);
    UserCreatableClass *ucc_parent =
        USER_CREATABLE_CLASS(
            object_class_get_parent(
                OBJECT_CLASS(MONITOR_QMP_GET_CLASS(mon))));
    ERRP_GUARD();

    ucc_parent->complete(uc, errp);
    if (*errp) {
        return;
    }

    qemu_chr_fe_set_echo(&mon->parent_obj.chr, true);

    if (monitor_requires_iothread(MONITOR(mon))) {
        /*
         * Make sure the old iowatch is gone.  It's possible when
         * e.g. the chardev is in client mode, with wait=on.
         */
        remove_fd_in_watch(mon->parent_obj.chr.chr);
        /*
         * Clean up listener IO sources early to prevent racy fd
         * handling between the main thread and the I/O thread.
         */
        remove_listener_fd_in_watch(mon->parent_obj.chr.chr);
        /*
         * We can't call qemu_chr_fe_set_handlers() directly here
         * since chardev might be running in the monitor I/O
         * thread.  Schedule a bottom half.
         */
        mon->setup_pending = true;
        aio_bh_schedule_oneshot(iothread_get_aio_context(mon_iothread),
                                monitor_qmp_setup_handlers_bh, mon);
        /* The bottom half will add @mon to @mon_list */
    } else {
        qemu_chr_fe_set_handlers(&mon->parent_obj.chr, monitor_can_read,
                                 monitor_qmp_read, monitor_qmp_event,
                                 NULL, &mon->parent_obj, NULL, true);
        monitor_list_append(&mon->parent_obj);
    }
}

static void monitor_qmp_iothread_quiesce(void *opaque)
{
    /* No-op: synchronization point only */
}

static bool monitor_qmp_prepare_delete(UserCreatable *uc, Error **errp)
{
    Monitor *mon = MONITOR(uc);
    MonitorQMP *qmp = MONITOR_QMP(uc);

    if (monitor_qmp_dispatcher_is_servicing(qmp)) {
        error_setg(errp, "Cannot delete the current QMP monitor");
        return false;
    }

    if (qatomic_read(&qmp->setup_pending)) {
        error_setg(errp, "monitor is still initializing");
        return false;
    }

    /* Remove from mon_list before chardev disconnect. */
    WITH_QEMU_LOCK_GUARD(&monitor_lock) {
        QTAILQ_REMOVE(&mon_list, mon, entry);
    }

    /* Cancel out_watch while gcontext still points to the right ctx. */
    WITH_QEMU_LOCK_GUARD(&mon->mon_lock) {
        monitor_cancel_out_watch(mon);
    }

    qemu_chr_fe_set_handlers(&mon->chr, NULL, NULL, NULL, NULL,
                             NULL, NULL, true);

    /* Drain requests from any in-flight monitor_qmp_read(). */
    monitor_qmp_drain_queue(qmp);

    WITH_QEMU_LOCK_GUARD(&mon->mon_lock) {
        /* Disable flushes before cancel -- gcontext is already wrong. */
        qemu_chr_fe_set_open(&mon->chr, false);
        monitor_cancel_out_watch(mon);
    }

    /* Synchronize with in-flight iothread callbacks. */
    if (monitor_requires_iothread(mon)) {
        aio_wait_bh_oneshot(iothread_get_aio_context(mon_iothread),
                            monitor_qmp_iothread_quiesce, NULL);
    }

    /* Catch requests from a racing monitor_qmp_read(). */
    monitor_qmp_drain_queue(qmp);
    monitor_fdsets_cleanup();

    return true;
}

static void monitor_qmp_accept_input(Monitor *mon)
{
    WITH_QEMU_LOCK_GUARD(&mon->mon_lock) {
        qemu_chr_fe_accept_input(&mon->chr);
    }
}