[Qemu-devel] [RFC][PATCH v2 09/17] qmp proxy: core code for proxying qmp requests to guest

[Michael Roth]

This provides a QmpProxy class, 1 instance of which is shared by all QMP
servers/sessions to send/receive QMP requests/responses between QEMU and
the QEMU guest agent.

A single qmp_proxy_send_request() is the only interface currently needed
by a QMP session, QAPI/QMP's existing async support handles all the work
of doing callbacks and routing responses to the proper session.

Signed-off-by: Michael Roth <address@hidden>
---
 qapi-schema.json |    7 ++
 qmp-core.c       |    8 ++
 qmp-core.h       |    7 +-
 qmp-proxy-core.h |   21 ++++
 qmp-proxy.c      |  294 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 vl.c             |    1 +
 6 files changed, 332 insertions(+), 6 deletions(-)
 create mode 100644 qmp-proxy-core.h
 create mode 100644 qmp-proxy.c

diff --git a/qapi-schema.json b/qapi-schema.json
index de6c9a3..5292938 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -1498,3 +1498,10 @@
 # Since: 0.14.0
 ##
 { 'option': 'vnc', 'data': 'VncConfig', 'implicit': 'address' }
+
+## 0.15.0 Events ##
+{ 'event': 'GUEST_AGENT_UP' }
+{ 'command': 'get-guest-agent-up-event', 'returns': 'GUEST_AGENT_UP' }
+
+{ 'event': 'GUEST_AGENT_RESET' }
+{ 'command': 'get-guest-agent-reset-event', 'returns': 'GUEST_AGENT_RESET' }
diff --git a/qmp-core.c b/qmp-core.c
index 9f3d182..dab50a1 100644
--- a/qmp-core.c
+++ b/qmp-core.c
@@ -937,7 +937,15 @@ void qmp_async_complete_command(QmpCommandState *cmd, 
QObject *retval, Error *er
     qemu_free(cmd);
 }
 
+extern QmpProxy *qmp_proxy_default;
+
 void qmp_guest_dispatch(const char *name, const QDict *args, Error **errp,
                         QmpGuestCompletionFunc *cb, void *opaque)
 {
+    if (!qmp_proxy_default) {
+        /* TODO: should set errp here */
+        fprintf(stderr, "qmp proxy: no guest proxy found\n");
+        return;
+    }
+    qmp_proxy_send_request(qmp_proxy_default, name, args, errp, cb, opaque);
 }
diff --git a/qmp-core.h b/qmp-core.h
index b676020..114d290 100644
--- a/qmp-core.h
+++ b/qmp-core.h
@@ -4,6 +4,7 @@
 #include "monitor.h"
 #include "qmp-marshal-types.h"
 #include "error_int.h"
+#include "qmp-proxy-core.h"
 
 struct QmpCommandState
 {
@@ -85,11 +86,5 @@ int qmp_state_get_fd(QmpState *sess);
     }                                                        \
 } while(0)
 
-typedef void (QmpGuestCompletionFunc)(void *opaque, QObject *ret_data, Error 
*err);
-
-void qmp_guest_dispatch(const char *name, const QDict *args, Error **errp,
-                        QmpGuestCompletionFunc *cb, void *opaque);
-
-
 #endif
 
diff --git a/qmp-proxy-core.h b/qmp-proxy-core.h
new file mode 100644
index 0000000..6afdc23
--- /dev/null
+++ b/qmp-proxy-core.h
@@ -0,0 +1,21 @@
+#ifndef QMP_PROXY_CORE_H
+#define QMP_PROXY_CORE_H
+
+#define QMP_PROXY_PATH_DEFAULT "/tmp/qmp-proxy.sock"
+
+typedef void (QmpGuestCompletionFunc)(void *opaque, QObject *ret_data,
+                                      Error *err);
+
+void qmp_guest_dispatch(const char *name, const QDict *args, Error **errp,
+                        QmpGuestCompletionFunc *cb, void *opaque);
+
+typedef struct QmpProxy QmpProxy;
+
+void qmp_proxy_send_request(QmpProxy *p, const char *name,
+                            const QDict *args, Error **errp,
+                            QmpGuestCompletionFunc *cb, void *opaque);
+QmpProxy *qmp_proxy_new(CharDriverState *chr);
+void qmp_proxy_close(QmpProxy *p);
+int qmp_proxy_read(QmpProxy *p, const uint8_t *buf, int len);
+
+#endif
diff --git a/qmp-proxy.c b/qmp-proxy.c
new file mode 100644
index 0000000..a4c7eea
--- /dev/null
+++ b/qmp-proxy.c
@@ -0,0 +1,294 @@
+/*
+ * QMP definitions for communicating with guest agent
+ *
+ * Copyright IBM Corp. 2011
+ *
+ * Authors:
+ *  Michael Roth      <address@hidden>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2 or later.
+ * See the COPYING file in the top-level directory.
+ *
+ */
+
+#include <glib.h>
+#include "qmp.h"
+#include "qmp-core.h"
+#include "qemu-queue.h"
+#include "json-parser.h"
+#include "json-streamer.h"
+#include "qemu_socket.h"
+
+#define QMP_SENTINEL 0xFF
+
+typedef struct QmpProxyRequest {
+    const char *name;
+    const QDict *args;
+    QmpGuestCompletionFunc *cb;
+    void *opaque;
+    QString *json;
+    QTAILQ_ENTRY(QmpProxyRequest) entry;
+} QmpProxyRequest;
+
+struct QmpProxy {
+    int session_id;
+    JSONMessageParser parser;
+    GString *tx;
+    QEMUTimer *tx_timer;
+    int tx_timer_interval;
+    QTAILQ_HEAD(, QmpProxyRequest) requests;
+    CharDriverState *chr;
+};
+
+static GuestAgentUpEvent guest_agent_up_event;
+static GuestAgentResetEvent guest_agent_reset_event;
+static void qmp_proxy_write(QmpProxy *p);
+
+GuestAgentUpEvent *qmp_get_guest_agent_up_event(Error **errp)
+{
+    return &guest_agent_up_event;
+}
+
+GuestAgentResetEvent *qmp_get_guest_agent_reset_event(Error **errp)
+{
+    return &guest_agent_reset_event;
+}
+
+static int qmp_proxy_cancel_request(QmpProxy *p, QmpProxyRequest *r)
+{
+    if (r && r->cb) {
+        r->cb(r->opaque, NULL, NULL);
+    }
+
+    return 0;
+}
+
+static int qmp_proxy_cancel_all(QmpProxy *p)
+{
+    QmpProxyRequest *r, *tmp;
+    QTAILQ_FOREACH_SAFE(r, &p->requests, entry, tmp) {
+        qmp_proxy_cancel_request(p, r);
+        QTAILQ_REMOVE(&p->requests, r, entry);
+    }
+
+    return 0;
+}
+
+static void qmp_proxy_send_control_event(QmpProxy *p, QDict *evt)
+{
+    QString *json = qobject_to_json(QOBJECT(evt));;
+
+    /* currently there's no reason to send any queued data or control
+     * events ahead of the most recent event, so empty the buffer first
+     */
+    g_string_truncate(p->tx, p->tx->len);
+    g_string_append_c(p->tx, QMP_SENTINEL);
+    g_string_append(p->tx, qstring_get_str(json));
+    QDECREF(json);
+    qmp_proxy_write(p);
+}
+
+static void qmp_proxy_send_host_ack(QmpProxy *p, int guest_sid)
+{
+    QDict *evt = qdict_new();
+
+    qdict_put_obj(evt, "_control_event",
+                  QOBJECT(qstring_from_str("host_ack")));
+    qdict_put_obj(evt, "_control_arg_host_sid",
+                  QOBJECT(qint_from_int(p->session_id)));
+    qdict_put_obj(evt, "_control_arg_guest_sid",
+                  QOBJECT(qint_from_int(guest_sid)));
+
+    qmp_proxy_send_control_event(p, evt);
+}
+
+static void qmp_proxy_send_host_init(QmpProxy *p)
+{
+    QDict *evt = qdict_new();
+
+    p->session_id = g_random_int_range(1, G_MAXINT32);
+    qdict_put_obj(evt, "_control_event",
+                  QOBJECT(qstring_from_str("host_init")));
+    qdict_put_obj(evt, "_control_arg_host_sid",
+                  QOBJECT(qint_from_int(p->session_id)));
+
+    qmp_proxy_send_control_event(p, evt);
+}
+
+static void qmp_proxy_process_control_event(QmpProxy *p, const QDict *evt)
+{
+    const char *cmd;
+    int host_sid, guest_sid;
+
+    cmd = qdict_get_try_str(evt, "_control_event");
+    if (!cmd) {
+        fprintf(stderr, "received NULL control event\n");
+    } else if (strcmp(cmd, "guest_ack") == 0) {
+        host_sid = qdict_get_try_int(evt, "_control_arg_host_sid", 0);
+        if (!host_sid) {
+            fprintf(stderr, "invalid guest_ack: wrong host sid\n");
+            return;
+        }
+        /* guest responded to host_init, return a host_ack */
+        /* reset outstanding requests, then send an ack with the
+         * session id they passed us
+         */
+        guest_sid = qdict_get_try_int(evt, "_control_arg_guest_sid", 0);
+        if (!guest_sid) {
+            fprintf(stderr, "invalid guest_ack: missing guest sid\n");
+            return;
+        }
+        qmp_proxy_send_host_ack(p, guest_sid);
+    } else if (strcmp(cmd, "guest_up") == 0) {
+        /* guest agent [re-]started, cancel any outstanding request and
+         * start negotiation */
+        /* TODO: should generate a qmp event for this as well */
+        signal_notify(&guest_agent_up_event);
+        signal_notify(&guest_agent_reset_event);
+        qmp_proxy_cancel_all(p);
+        qmp_proxy_send_host_init(p);
+    } else if (strcmp(cmd, "guest_reset") == 0) {
+        /* guest agent reset it's state (likely due to a command timeout)
+         * cancel all outstanding requests */
+        signal_notify(&guest_agent_reset_event);
+        qmp_proxy_cancel_all(p);
+    } else {
+        fprintf(stderr, "received unknown control event\n");
+    }
+}
+
+static void qmp_proxy_process_event(JSONMessageParser *parser, QList *tokens)
+{
+    QmpProxy *p = container_of(parser, QmpProxy, parser);
+    QmpProxyRequest *r;
+    QObject *obj;
+    QDict *qdict;
+    Error *err = NULL;
+
+    fprintf(stderr, "qmp proxy: called\n");
+    obj = json_parser_parse_err(tokens, NULL, &err);
+    if (!obj) {
+        fprintf(stderr, "qmp proxy: failed to parse\n");
+        return;
+    } else {
+        fprintf(stderr, "qmp proxy: parse successful\n");
+        qdict = qobject_to_qdict(obj);
+    }
+
+    if (qdict_haskey(qdict, "_control_event")) {
+        /* handle transport-level control event */
+        qmp_proxy_process_control_event(p, qdict);
+    } else if (qdict_haskey(qdict, "return")) {
+        /* handle proxied qmp command response */
+        fprintf(stderr, "received return\n");
+        r = QTAILQ_FIRST(&p->requests);
+        if (!r) {
+            fprintf(stderr, "received return, but no request queued\n");
+            return;
+        }
+        /* XXX: can't assume type here */
+        fprintf(stderr, "recieved response for cmd: %s\nreturn: %s\n",
+                r->name, qstring_get_str(qobject_to_json(QOBJECT(qdict))));
+        r->cb(r->opaque, qdict_get(qdict, "return"), NULL);
+        QTAILQ_REMOVE(&p->requests, r, entry);
+        qemu_free(r);
+        fprintf(stderr, "done handling response\n");
+    } else {
+        fprintf(stderr, "received invalid payload format\n");
+    }
+
+    QDECREF(qdict);
+}
+
+static void qmp_proxy_write_handler(void *opaque)
+{
+    QmpProxy *p = opaque;
+    int max, len;
+
+    if (!p->tx->len) {
+        return;
+    }
+
+    max = qemu_chr_can_read(p->chr);
+    if (max) {
+        len = MIN(max, p->tx->len);
+        qemu_chr_read(p->chr, (uint8_t *)p->tx->str, len);
+        g_string_erase(p->tx, 0, len);
+    }
+
+    if (p->tx->len) {
+        /* defer transfer of remaining data */
+        qemu_mod_timer(p->tx_timer,
+                       qemu_get_clock(rt_clock) + p->tx_timer_interval);
+    }
+}
+
+int qmp_proxy_read(QmpProxy *p, const uint8_t *buf, int len)
+{
+    return json_message_parser_feed(&p->parser, (const char *)buf, len);
+}
+
+void qmp_proxy_write(QmpProxy *p)
+{
+    /* more stuff in our buffer, kick the write handler */
+    qmp_proxy_write_handler(p);
+}
+
+void qmp_proxy_send_request(QmpProxy *p, const char *name,
+                            const QDict *args, Error **errp,
+                            QmpGuestCompletionFunc *cb, void *opaque)
+{
+    QmpProxyRequest *r = qemu_mallocz(sizeof(QmpProxyRequest));
+    QDict *payload = qdict_new();
+    QString *json;
+
+    /* TODO: don't really need to hold on to name/args after encoding */
+    r->name = name;
+    r->args = args;
+    r->cb = cb;
+    r->opaque = opaque;
+
+    qdict_put_obj(payload, "execute", QOBJECT(qstring_from_str(r->name)));
+    /* TODO: casting a const so we can add it to our dictionary. bad. */
+    qdict_put_obj(payload, "arguments", QOBJECT((QDict *)args));
+
+    json = qobject_to_json(QOBJECT((QDict *)payload));
+    if (!json) {
+        goto out_bad;
+    }
+
+    QTAILQ_INSERT_TAIL(&p->requests, r, entry);
+    g_string_append(p->tx, qstring_get_str(json));
+    QDECREF(json);
+    qmp_proxy_write(p);
+    return;
+
+out_bad:
+    cb(opaque, NULL, NULL);
+    qemu_free(r);
+}
+
+QmpProxy *qmp_proxy_new(CharDriverState *chr)
+{
+    QmpProxy *p = qemu_mallocz(sizeof(QmpProxy));
+
+    signal_init(&guest_agent_up_event);
+    signal_init(&guest_agent_reset_event);
+
+    /* there's a reason for this madness */
+    p->tx_timer = qemu_new_timer(rt_clock, qmp_proxy_write_handler, p);
+    p->tx_timer_interval = 10;
+    p->tx = g_string_new("");
+    p->chr = chr;
+    json_message_parser_init(&p->parser, qmp_proxy_process_event);
+    QTAILQ_INIT(&p->requests);
+
+    return p;
+}
+
+void qmp_proxy_close(QmpProxy *p)
+{
+    qmp_proxy_cancel_all(p);
+    g_string_free(p->tx, TRUE);
+    qemu_free(p);
+}
diff --git a/vl.c b/vl.c
index 3fdc7cc..e8c49ef 100644
--- a/vl.c
+++ b/vl.c
@@ -231,6 +231,7 @@ int ctrl_grab = 0;
 unsigned int nb_prom_envs = 0;
 const char *prom_envs[MAX_PROM_ENVS];
 int boot_menu;
+QmpProxy *qmp_proxy_default;
 
 ShutdownEvent qemu_shutdown_event;
 ResetEvent qemu_reset_event;
-- 
1.7.0.4