Re: [PATCH v6 5/8] Add dbus-vmstate object

[Daniel P . Berrangé]

On Wed, Dec 11, 2019 at 05:45:03PM +0400, Marc-André Lureau wrote:
> When instantiated, this object will connect to the given D-Bus bus
> "addr". During migration, it will take/restore the data from
> org.qemu.VMState1 instances. See documentation for details.
> 
> Signed-off-by: Marc-André Lureau <address@hidden>
> ---
>  MAINTAINERS                   |   2 +
>  backends/Makefile.objs        |   4 +
>  backends/dbus-vmstate.c       | 496 ++++++++++++++++++++++++++++++++++
>  docs/interop/dbus-vmstate.rst |  74 +++++
>  docs/interop/dbus.rst         |   5 +
>  docs/interop/index.rst        |   1 +
>  6 files changed, 582 insertions(+)
>  create mode 100644 backends/dbus-vmstate.c
>  create mode 100644 docs/interop/dbus-vmstate.rst
> 
> diff --git a/MAINTAINERS b/MAINTAINERS
> index f08fb4f24e..7af80d0c1d 100644
> --- a/MAINTAINERS
> +++ b/MAINTAINERS
> @@ -2202,9 +2202,11 @@ F: qapi/migration.json
>  D-Bus
>  M: Marc-André Lureau <address@hidden>
>  S: Maintained
> +F: backends/dbus-vmstate.c
>  F: util/dbus.c
>  F: include/qemu/dbus.h
>  F: docs/interop/dbus.rst
> +F: docs/interop/dbus-vmstate.rst
>  
>  Seccomp
>  M: Eduardo Otubo <address@hidden>
> diff --git a/backends/Makefile.objs b/backends/Makefile.objs
> index f0691116e8..28a847cd57 100644
> --- a/backends/Makefile.objs
> +++ b/backends/Makefile.objs
> @@ -17,3 +17,7 @@ endif
>  common-obj-$(call land,$(CONFIG_VHOST_USER),$(CONFIG_VIRTIO)) += vhost-user.o
>  
>  common-obj-$(CONFIG_LINUX) += hostmem-memfd.o
> +
> +common-obj-$(CONFIG_GIO) += dbus-vmstate.o
> +dbus-vmstate.o-cflags = $(GIO_CFLAGS)
> +dbus-vmstate.o-libs = $(GIO_LIBS)
> diff --git a/backends/dbus-vmstate.c b/backends/dbus-vmstate.c
> new file mode 100644
> index 0000000000..059dd420b8
> --- /dev/null
> +++ b/backends/dbus-vmstate.c
> @@ -0,0 +1,496 @@
> +/*
> + * QEMU dbus-vmstate
> + *
> + * Copyright (C) 2019 Red Hat Inc
> + *
> + * Authors:
> + *  Marc-André Lureau <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 "qemu/osdep.h"
> +#include "qemu/units.h"
> +#include "qemu/dbus.h"
> +#include "qemu/error-report.h"
> +#include "qapi/error.h"
> +#include "qom/object_interfaces.h"
> +#include "qapi/qmp/qerror.h"
> +#include "migration/vmstate.h"
> +
> +typedef struct DBusVMState DBusVMState;
> +typedef struct DBusVMStateClass DBusVMStateClass;
> +
> +#define TYPE_DBUS_VMSTATE "dbus-vmstate"
> +#define DBUS_VMSTATE(obj)                                \
> +    OBJECT_CHECK(DBusVMState, (obj), TYPE_DBUS_VMSTATE)
> +#define DBUS_VMSTATE_GET_CLASS(obj)                              \
> +    OBJECT_GET_CLASS(DBusVMStateClass, (obj), TYPE_DBUS_VMSTATE)
> +#define DBUS_VMSTATE_CLASS(klass)                                    \
> +    OBJECT_CLASS_CHECK(DBusVMStateClass, (klass), TYPE_DBUS_VMSTATE)
> +
> +struct DBusVMStateClass {
> +    ObjectClass parent_class;
> +};
> +

Not an objection to your patch here. This just reminds me that we
ought to follow GLib's lead and implement some helper macros to
automate all this tedious boilerplate. So we can just do something
simple like:

 QOM_DECLARE_FINAL_TYPE(DBusVMState, dbus_vmstate, DBUS_VMSATE, Object)

and an equiv to do the  TypeInfo declaration & registration.

> +struct DBusVMState {
> +    Object parent;
> +
> +    GDBusConnection *bus;
> +    char *dbus_addr;
> +    char *id_list;
> +
> +    uint32_t data_size;
> +    uint8_t *data;
> +};
> +
> +static const GDBusPropertyInfo vmstate_property_info[] = {
> +    { -1, (char *) "Id", (char *) "s",
> +      G_DBUS_PROPERTY_INFO_FLAGS_READABLE, NULL },
> +};
> +
> +static const GDBusPropertyInfo * const vmstate_property_info_pointers[] = {
> +    &vmstate_property_info[0],
> +    NULL
> +};
> +
> +static const GDBusInterfaceInfo vmstate1_interface_info = {
> +    -1,
> +    (char *) "org.qemu.VMState1",
> +    (GDBusMethodInfo **) NULL,
> +    (GDBusSignalInfo **) NULL,
> +    (GDBusPropertyInfo **) &vmstate_property_info_pointers,
> +    NULL,
> +};
> +
> +#define DBUS_VMSTATE_SIZE_LIMIT (1 * MiB)
> +
> +static GHashTable *
> +get_id_list_set(DBusVMState *self)
> +{
> +    g_auto(GStrv) ids = NULL;
> +    g_autoptr(GHashTable) set = NULL;
> +    int i;
> +
> +    if (!self->id_list) {
> +        return NULL;
> +    }
> +
> +    ids = g_strsplit(self->id_list, ",", -1);
> +    set = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, NULL);
> +    for (i = 0; ids[i]; i++) {
> +        g_hash_table_add(set, ids[i]);
> +        ids[i] = NULL;
> +    }
> +
> +    return g_steal_pointer(&set);
> +}
> +
> +static GHashTable *
> +dbus_get_proxies(DBusVMState *self, GError **err)
> +{
> +    g_autoptr(GHashTable) proxies = NULL;
> +    g_autoptr(GHashTable) ids = NULL;
> +    g_auto(GStrv) names = NULL;
> +    size_t i;
> +
> +    ids = get_id_list_set(self);
> +    proxies = g_hash_table_new_full(g_str_hash, g_str_equal,
> +                                    g_free, g_object_unref);
> +
> +    names = qemu_dbus_get_queued_owners(self->bus, "org.qemu.VMState1");
> +    if (!names) {
> +        return NULL;
> +    }
> +
> +    for (i = 0; names[i]; i++) {
> +        g_autoptr(GDBusProxy) proxy = NULL;
> +        g_autoptr(GVariant) result = NULL;
> +        g_autofree char *id = NULL;
> +        size_t size;
> +
> +        proxy = g_dbus_proxy_new_sync(self->bus, G_DBUS_PROXY_FLAGS_NONE,
> +                    (GDBusInterfaceInfo *) &vmstate1_interface_info,
> +                    names[i],
> +                    "/org/qemu/VMState1",
> +                    "org.qemu.VMState1",
> +                    NULL, err);
> +        if (!proxy) {
> +            return NULL;
> +        }
> +
> +        result = g_dbus_proxy_get_cached_property(proxy, "Id");
> +        if (!result) {
> +            g_set_error_literal(err, G_IO_ERROR, G_IO_ERROR_FAILED,
> +                                "VMState Id property is missing.");
> +            return NULL;
> +        }
> +
> +        id = g_variant_dup_string(result, &size);
> +        if (ids && !g_hash_table_remove(ids, id)) {
> +            g_clear_pointer(&id, g_free);
> +            g_clear_object(&proxy);
> +            continue;
> +        }
> +        if (size == 0 || size >= 256) {
> +            g_set_error(err, G_IO_ERROR, G_IO_ERROR_FAILED,
> +                        "VMState Id '%s' is invalid.", id);
> +            return NULL;
> +        }
> +
> +        if (!g_hash_table_insert(proxies, id, proxy)) {
> +            g_set_error(err, G_IO_ERROR, G_IO_ERROR_FAILED,
> +                        "Duplicated VMState Id '%s'", id);
> +            return NULL;
> +        }
> +        id = NULL;
> +        proxy = NULL;
> +
> +        g_clear_pointer(&result, g_variant_unref);
> +    }
> +
> +    if (ids) {
> +        g_autofree char **left = NULL;
> +
> +        left = (char **)g_hash_table_get_keys_as_array(ids, NULL);
> +        if (*left) {
> +            g_autofree char *leftids = g_strjoinv(",", left);
> +            g_set_error(err, G_IO_ERROR, G_IO_ERROR_FAILED,
> +                        "Required VMState Id are missing: %s", leftids);
> +            return NULL;
> +        }
> +    }
> +
> +    return g_steal_pointer(&proxies);
> +}
> +
> +static int
> +dbus_load_state_proxy(GDBusProxy *proxy, const uint8_t *data, size_t size)
> +{
> +    g_autoptr(GError) err = NULL;
> +    g_autoptr(GVariant) result = NULL;
> +    g_autoptr(GVariant) value = NULL;
> +
> +    value = g_variant_new_fixed_array(G_VARIANT_TYPE_BYTE,
> +                                      data, size, sizeof(char));
> +    result = g_dbus_proxy_call_sync(proxy, "Load",
> +                                    g_variant_new("(@ay)",
> +                                                  g_steal_pointer(&value)),
> +                                    G_DBUS_CALL_FLAGS_NO_AUTO_START,
> +                                    -1, NULL, &err);
> +    if (!result) {
> +        error_report("Failed to Load: %s", err->message);
> +        return -1;
> +    }
> +
> +    return 0;
> +}
> +
> +static int dbus_vmstate_post_load(void *opaque, int version_id)
> +{
> +    DBusVMState *self = DBUS_VMSTATE(opaque);
> +    g_autoptr(GInputStream) m = NULL;
> +    g_autoptr(GDataInputStream) s = NULL;
> +    g_autoptr(GError) err = NULL;
> +    g_autoptr(GHashTable) proxies = NULL;
> +    uint32_t nelem;
> +
> +    proxies = dbus_get_proxies(self, &err);
> +    if (!proxies) {
> +        error_report("Failed to get proxies: %s", err->message);
> +        return -1;
> +    }
> +
> +    m = g_memory_input_stream_new_from_data(self->data, self->data_size, 
> NULL);
> +    s = g_data_input_stream_new(m);
> +    g_data_input_stream_set_byte_order(s, 
> G_DATA_STREAM_BYTE_ORDER_BIG_ENDIAN);
> +
> +    nelem = g_data_input_stream_read_uint32(s, NULL, &err);
> +    if (err) {
> +        goto error;
> +    }
> +
> +    while (nelem > 0) {
> +        GDBusProxy *proxy = NULL;
> +        uint32_t len;
> +        gsize bytes_read, avail;
> +        char id[256];
> +
> +        len = g_data_input_stream_read_uint32(s, NULL, &err);
> +        if (err) {
> +            goto error;
> +        }
> +        if (len >= 256) {
> +            error_report("Invalid DBus vmstate proxy name %u", len);
> +            return -1;
> +        }
> +        if (!g_input_stream_read_all(G_INPUT_STREAM(s), id, len,
> +                                     &bytes_read, NULL, &err)) {
> +            goto error;
> +        }
> +        g_return_val_if_fail(bytes_read == len, -1);
> +        id[len] = 0;
> +
> +        proxy = g_hash_table_lookup(proxies, id);
> +        if (!proxy) {
> +            error_report("Failed to find proxy Id '%s'", id);
> +            return -1;
> +        }
> +
> +        len = g_data_input_stream_read_uint32(s, NULL, &err);
> +        avail = g_buffered_input_stream_get_available(
> +            G_BUFFERED_INPUT_STREAM(s));
> +
> +        if (len > DBUS_VMSTATE_SIZE_LIMIT || len > avail) {
> +            error_report("Invalid vmstate size: %u", len);
> +            return -1;
> +        }
> +
> +        if (dbus_load_state_proxy(proxy,
> +                
> g_buffered_input_stream_peek_buffer(G_BUFFERED_INPUT_STREAM(s),
> +                                                    NULL),
> +                len) < 0) {
> +            error_report("Failed to restore Id '%s'", id);
> +            return -1;
> +        }
> +
> +        if (!g_seekable_seek(G_SEEKABLE(s), len, G_SEEK_CUR, NULL, &err)) {
> +            goto error;
> +        }
> +
> +        nelem -= 1;
> +    }
> +
> +    return 0;
> +
> +error:
> +    error_report("Failed to read from stream: %s", err->message);
> +    return -1;
> +}
> +
> +static void
> +dbus_save_state_proxy(gpointer key,
> +                      gpointer value,
> +                      gpointer user_data)
> +{
> +    GDataOutputStream *s = user_data;
> +    const char *id = key;
> +    GDBusProxy *proxy = value;
> +    g_autoptr(GVariant) result = NULL;
> +    g_autoptr(GVariant) child = NULL;
> +    g_autoptr(GError) err = NULL;
> +    const uint8_t *data;
> +    gsize size;
> +
> +    result = g_dbus_proxy_call_sync(proxy, "Save",
> +                                    NULL, G_DBUS_CALL_FLAGS_NO_AUTO_START,
> +                                    -1, NULL, &err);
> +    if (!result) {
> +        error_report("Failed to Save: %s", err->message);
> +        return;
> +    }
> +
> +    child = g_variant_get_child_value(result, 0);
> +    data = g_variant_get_fixed_array(child, &size, sizeof(char));
> +    if (!data) {
> +        error_report("Failed to Save: not a byte array");
> +        return;
> +    }
> +    if (size > DBUS_VMSTATE_SIZE_LIMIT) {
> +        error_report("Too large vmstate data to save: %" G_GSIZE_FORMAT, 
> size);
> +        return;
> +    }
> +
> +    if (!g_data_output_stream_put_uint32(s, strlen(id), NULL, &err) ||
> +        !g_data_output_stream_put_string(s, id, NULL, &err) ||
> +        !g_data_output_stream_put_uint32(s, size, NULL, &err) ||
> +        !g_output_stream_write_all(G_OUTPUT_STREAM(s),
> +                                   data, size, NULL, NULL, &err)) {
> +        error_report("Failed to write to stream: %s", err->message);
> +    }

This is a bit of a bike-shed comment, but I'm curious if you
considered using GVariant for the serializing data vs the
data output stream ? I feel like GVariant is enforcing more
structure & safety on the data serialization process, which
could be appealing.

> +static int dbus_vmstate_pre_save(void *opaque)
> +{
> +    DBusVMState *self = DBUS_VMSTATE(opaque);
> +    g_autoptr(GOutputStream) m = NULL;
> +    g_autoptr(GDataOutputStream) s = NULL;
> +    g_autoptr(GHashTable) proxies = NULL;
> +    g_autoptr(GError) err = NULL;
> +
> +    proxies = dbus_get_proxies(self, &err);
> +    if (!proxies) {
> +        error_report("Failed to get proxies: %s", err->message);
> +        return -1;
> +    }
> +
> +    m = g_memory_output_stream_new_resizable();
> +    s = g_data_output_stream_new(m);
> +    g_data_output_stream_set_byte_order(s, 
> G_DATA_STREAM_BYTE_ORDER_BIG_ENDIAN);
> +
> +    if (!g_data_output_stream_put_uint32(s, g_hash_table_size(proxies),
> +                                         NULL, &err)) {
> +        error_report("Failed to write to stream: %s", err->message);
> +        return -1;
> +    }
> +
> +    g_hash_table_foreach(proxies, dbus_save_state_proxy, s);
> +
> +    if (g_memory_output_stream_get_size(G_MEMORY_OUTPUT_STREAM(m))
> +        > UINT32_MAX) {
> +        error_report("DBus vmstate buffer is too large");
> +        return -1;
> +    }
> +
> +    if (!g_output_stream_close(G_OUTPUT_STREAM(m), NULL, &err)) {
> +        error_report("Failed to close stream: %s", err->message);
> +        return -1;
> +    }
> +
> +    g_free(self->data);
> +    self->data_size =
> +        g_memory_output_stream_get_size(G_MEMORY_OUTPUT_STREAM(m));
> +    self->data =
> +        g_memory_output_stream_steal_data(G_MEMORY_OUTPUT_STREAM(m));
> +
> +    return 0;
> +}
> +
> +static const VMStateDescription dbus_vmstate = {
> +    .name = TYPE_DBUS_VMSTATE,
> +    .version_id = 0,
> +    .pre_save = dbus_vmstate_pre_save,
> +    .post_load = dbus_vmstate_post_load,
> +    .fields = (VMStateField[]) {
> +        VMSTATE_UINT32(data_size, DBusVMState),
> +        VMSTATE_VBUFFER_ALLOC_UINT32(data, DBusVMState, 0, 0, data_size),
> +        VMSTATE_END_OF_LIST()
> +    }
> +};
> +
> +static void
> +dbus_vmstate_complete(UserCreatable *uc, Error **errp)
> +{
> +    DBusVMState *self = DBUS_VMSTATE(uc);
> +    GError *err = NULL;

Can use g_autoptr for this

> +    GDBusConnection *bus;
> +
> +    if (!object_resolve_path_type("", TYPE_DBUS_VMSTATE, NULL)) {
> +        error_setg(errp, "There is already an instance of %s",
> +                   TYPE_DBUS_VMSTATE);
> +        return;
> +    }
> +
> +    if (!self->dbus_addr) {
> +        error_setg(errp, QERR_MISSING_PARAMETER, "addr");
> +        return;
> +    }
> +
> +    bus = g_dbus_connection_new_for_address_sync(self->dbus_addr,
> +             G_DBUS_CONNECTION_FLAGS_AUTHENTICATION_CLIENT |
> +             G_DBUS_CONNECTION_FLAGS_MESSAGE_BUS_CONNECTION,
> +             NULL, NULL, &err);

Why not just use  self->bus directly.

> +    if (err) {
> +        error_setg(errp, "failed to connect to DBus: '%s'", err->message);
> +        g_clear_error(&err);

Missing return here, as we don't want to register vmstate handler
if we fail.

> +    }
> +
> +    self->bus = bus;
> +
> +    if (vmstate_register(VMSTATE_IF(self), -1, &dbus_vmstate, self) < 0) {
> +        error_setg(errp, "Failed to register vmstate");
> +    }
> +}


> diff --git a/docs/interop/dbus-vmstate.rst b/docs/interop/dbus-vmstate.rst
> new file mode 100644
> index 0000000000..8693891640
> --- /dev/null
> +++ b/docs/interop/dbus-vmstate.rst
> @@ -0,0 +1,74 @@
> +=============
> +D-Bus VMState
> +=============
> +
> +Introduction
> +============
> +
> +The QEMU dbus-vmstate object's aim is to migrate helpers' data running
> +on a QEMU D-Bus bus. (refer to the :doc:`dbus` document for
> +recommendation on D-Bus usage)
> +
> +Upon migration, QEMU will go through the queue of
> +``org.qemu.VMState1`` D-Bus name owners and query their ``Id``. It
> +must be unique among the helpers.
> +
> +It will then save arbitrary data of each Id to be transferred in the
> +migration stream and restored/loaded at the corresponding destination
> +helper.
> +
> +The data amount to be transferred is limited to 1Mb. The state must be
> +saved quickly (a few seconds maximum). (D-Bus imposes a time limit on

A few seconds is too long IMHO. I think the expectation ought to
be a small fraction of a second.

Anything longer than that suggests there is some extra synchronization
work needing beyond serailizing state, which might suggest the need
for a separate DBus call.  eg a way to tell the backend to "quiesce"
itself perhaps

For now we can keep it simple and just say that this method should
not do anything except seriailize state in a fraction of a second.

> +reply anyway, and migration would fail if data isn't given quickly
> +enough.)
> +
> +dbus-vmstate object can be configured with the expected list of
> +helpers by setting its ``id-list`` property, with a comma-separated
> +``Id`` list.
> +
> +Interface
> +=========
> +
> +On object path ``/org/qemu/VMState1``, the following
> +``org.qemu.VMState1`` interface should be implemented:
> +
> +.. code:: xml
> +
> +  <interface name="org.qemu.VMState1">
> +    <property name="Id" type="s" access="read"/>
> +    <method name="Load">
> +      <arg type="ay" name="data" direction="in"/>
> +    </method>
> +    <method name="Save">
> +      <arg type="ay" name="data" direction="out"/>
> +    </method>
> +  </interface>
> +
> +"Id" property
> +-------------
> +
> +A string that identifies the helper uniquely. (maximum 256 bytes
> +including terminating NUL byte)
> +
> +.. note::
> +
> +   The helper ID namespace is a separate namespace. In particular, it is not
> +   related to QEMU "id" used in -object/-device objects.

Are there any expectations here on a scheme ? I feel leaving it
unspecified is probably a mistake. Should it follow a DBUs like
reverse.domain.name.style ?

> +
> +Load(in u8[] bytes) method
> +--------------------------
> +
> +The method called on destination with the state to restore.
> +
> +The helper may be initially started in a waiting state (with
> +an --incoming argument for example), and it may resume on success.
> +
> +An error may be returned to the caller.
> +
> +Save(out u8[] bytes) method
> +---------------------------
> +
> +The method called on the source to get the current state to be
> +migrated. The helper should continue to run normally.
> +
> +An error may be returned to the caller.

Regards,
Daniel
-- 
|: https://berrange.com      -o-    https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org         -o-            https://fstop138.berrange.com :|
|: https://entangle-photo.org    -o-    https://www.instagram.com/dberrange :|