[PATCH v2 20/27] tools/libxl: Infrastructure for writing a v2 stream

[Andrew Cooper <andrew.cooper3@citrix.com>]

From: Ross Lagerwall <ross.lagerwall@citrix.com>

This contains the event machinary and state machines to write non-checkpointed
migration v2 stream (with the exception of the xc_domain_save() handling which
is spliced later in a bisectable way).

Signed-off-by: Ross Lagerwall <ross.lagerwall@citrix.com>
Signed-off-by: Andrew Cooper <andrew.cooper3@citrix.com>
CC: Ian Campbell <Ian.Campbell@citrix.com>
CC: Ian Jackson <Ian.Jackson@eu.citrix.com>
CC: Wei Liu <wei.liu2@citrix.com>

---
As with the read side of things, this has undergone substantial changes in v2.
---
 tools/libxl/Makefile             |    2 +-
 tools/libxl/libxl_internal.h     |   47 ++++
 tools/libxl/libxl_stream_write.c |  451 ++++++++++++++++++++++++++++++++++++++
 3 files changed, 499 insertions(+), 1 deletion(-)
 create mode 100644 tools/libxl/libxl_stream_write.c

diff --git a/tools/libxl/Makefile b/tools/libxl/Makefile
index 0ebc35a..7d44483 100644
--- a/tools/libxl/Makefile
+++ b/tools/libxl/Makefile
@@ -95,7 +95,7 @@ LIBXL_OBJS = flexarray.o libxl.o libxl_create.o libxl_dm.o libxl_pci.o \
 			libxl_dom.o libxl_exec.o libxl_xshelp.o libxl_device.o \
 			libxl_internal.o libxl_utils.o libxl_uuid.o \
 			libxl_json.o libxl_aoutils.o libxl_numa.o libxl_vnuma.o \
-			libxl_stream_read.o \
+			libxl_stream_read.o libxl_stream_write.o \
 			libxl_save_callout.o _libxl_save_msgs_callout.o \
 			libxl_qmp.o libxl_event.o libxl_fork.o $(LIBXL_OBJS-y)
 LIBXL_OBJS += libxl_genid.o
diff --git a/tools/libxl/libxl_internal.h b/tools/libxl/libxl_internal.h
index 1cf1884..2beb534 100644
--- a/tools/libxl/libxl_internal.h
+++ b/tools/libxl/libxl_internal.h
@@ -2973,6 +2973,52 @@ typedef void libxl__domain_suspend_cb(libxl__egc*,
 typedef void libxl__save_device_model_cb(libxl__egc*,
                                          libxl__domain_suspend_state*, int rc);
 
+/* State for writing a libxl migration v2 stream */
+typedef struct libxl__stream_write_state libxl__stream_write_state;
+
+typedef void (*sws_record_done_cb)(libxl__egc *egc,
+                                   libxl__stream_write_state *sws);
+
+struct libxl__stream_write_state {
+    /* filled by the user */
+    libxl__ao *ao;
+    int fd;
+    uint32_t domid;
+    void (*completion_callback)(libxl__egc *egc,
+                                libxl__stream_write_state *sws,
+                                int rc);
+    /* Private */
+    int rc;
+    bool running;
+
+    /* Active-stuff handling */
+    int joined_rc;
+
+    /* Main stream-writing data */
+    size_t padding;
+    libxl__datacopier_state dc;
+    sws_record_done_cb record_done_callback;
+
+    /* Emulator blob handling */
+    libxl__datacopier_state emu_dc;
+    libxl__carefd *emu_carefd;
+    libxl__sr_rec_hdr emu_rec_hdr;
+    void *emu_body;
+};
+
+_hidden void libxl__stream_write_start(libxl__egc *egc,
+                                       libxl__stream_write_state *stream);
+
+_hidden void libxl__stream_write_abort(libxl__egc *egc,
+                                       libxl__stream_write_state *stream,
+                                       int rc);
+
+static inline bool libxl__stream_write_inuse(
+    const libxl__stream_write_state *stream)
+{
+    return stream->running;
+}
+
 typedef struct libxl__logdirty_switch {
     const char *cmd;
     const char *cmd_path;
@@ -3013,6 +3059,7 @@ struct libxl__domain_suspend_state {
     /* private for libxl__domain_save_device_model */
     libxl__save_device_model_cb *save_dm_callback;
     libxl__datacopier_state save_dm_datacopier;
+    libxl__stream_write_state sws;
 };
 
 
diff --git a/tools/libxl/libxl_stream_write.c b/tools/libxl/libxl_stream_write.c
new file mode 100644
index 0000000..bf568ad
--- /dev/null
+++ b/tools/libxl/libxl_stream_write.c
@@ -0,0 +1,451 @@
+/*
+ * Copyright (C) 2015      Citrix Ltd.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published
+ * by the Free Software Foundation; version 2.1 only. with the special
+ * exception on linking described in file LICENSE.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ */
+
+#include "libxl_osdeps.h" /* must come before any other headers */
+
+#include "libxl_internal.h"
+
+/*
+ * Infrastructure for writing a domain to a libxl migration v2 stream.
+ *
+ * Entry points from outside:
+ *  - libxl__stream_write_start()
+ *     - Start writing a stream from the start.
+ *
+ * In normal operation, there are two tasks running at once; this stream
+ * processing, and the libxl-save-helper.  check_stream_finished() is used to
+ * join all the tasks in both success and error cases.
+ *
+ * Nomenclature for event callbacks:
+ *  - $FOO_done(): Completion callback for $FOO
+ *  - write_$FOO(): Set up the datacopier to write a $FOO
+ *  - $BAR_header(): A $BAR record header only
+ *  - $BAR_record(): A complete $BAR record with header and content
+ *
+ * The main loop for a plain VM writes:
+ *  - Stream header
+ *  - Libxc record
+ *  - Toolstack record
+ *  - if (hvm), Qemu record
+ *  - End record
+ */
+
+static void stream_success(libxl__egc *egc,
+                           libxl__stream_write_state *stream);
+static void stream_failed(libxl__egc *egc,
+                          libxl__stream_write_state *stream, int ret);
+static void stream_done(libxl__egc *egc,
+                        libxl__stream_write_state *stream);
+
+static void check_stream_finished(libxl__egc *egc,
+                                  libxl__stream_write_state *stream,
+                                  int rc, const char *what);
+
+/* Event callbacks for plain VM. */
+static void stream_header_done(libxl__egc *egc,
+                               libxl__datacopier_state *dc,
+                               int rc, int onwrite, int errnoval);
+static void libxc_header_done(libxl__egc *egc,
+                              libxl__stream_write_state *stream);
+/* libxl__xc_domain_save_done() lives here, event-order wise. */
+static void write_toolstack_record(libxl__egc *egc,
+                                   libxl__stream_write_state *stream);
+static void toolstack_record_done(libxl__egc *egc,
+                                  libxl__stream_write_state *stream);
+static void write_emulator_record(libxl__egc *egc,
+                                  libxl__stream_write_state *stream);
+static void emulator_read_done(libxl__egc *egc,
+                               libxl__datacopier_state *dc,
+                               int rc, int onwrite, int errnoval);
+static void emulator_record_done(libxl__egc *egc,
+                                 libxl__stream_write_state *stream);
+static void write_end_record(libxl__egc *egc,
+                             libxl__stream_write_state *stream);
+
+/* Helper to set up reading some data from the stream. */
+static void write_done(libxl__egc *egc,
+                       libxl__datacopier_state *dc,
+                       int rc, int onwrite, int errnoval)
+{
+    libxl__stream_write_state *stream = CONTAINER_OF(dc, *stream, dc);
+    STATE_AO_GC(stream->ao);
+    sws_record_done_cb cb = stream->record_done_callback;
+
+    stream->record_done_callback = NULL;
+
+    if (onwrite || errnoval) {
+        LOG(ERROR, "rc %d, onwrite %d, errnoval %d", rc, onwrite, errnoval);
+        stream_failed(egc, stream, ERROR_FAIL);
+    }
+    else
+        cb(egc, stream);
+}
+
+static void setup_write(libxl__egc *egc,
+                        libxl__stream_write_state *stream,
+                        const char *what,
+                        libxl__sr_rec_hdr *hdr, void *body,
+                        sws_record_done_cb cb)
+{
+    static const uint8_t zero_padding[1U << REC_ALIGN_ORDER] = { 0 };
+
+    libxl__datacopier_state *dc = &stream->dc;
+    int ret;
+
+    assert(stream->record_done_callback == NULL);
+
+    dc->writewhat     = what;
+    dc->used          = 0;
+    dc->callback      = write_done;
+
+    ret = libxl__datacopier_start(dc);
+
+    if (ret) {
+        stream_failed(egc, stream, ret);
+        return;
+    }
+
+    size_t padsz = ROUNDUP(hdr->length, REC_ALIGN_ORDER) - hdr->length;
+
+    /* Insert header */
+    libxl__datacopier_prefixdata(egc, dc, hdr, sizeof(*hdr));
+
+    /* Optional body */
+    if (body)
+        libxl__datacopier_prefixdata(egc, dc, body, hdr->length);
+
+    /* Any required padding */
+    if (padsz > 0)
+        libxl__datacopier_prefixdata(egc, dc,
+                                     zero_padding, padsz);
+    stream->record_done_callback = cb;
+}
+
+void libxl__stream_write_start(libxl__egc *egc,
+                               libxl__stream_write_state *stream)
+{
+    libxl__datacopier_state *dc = &stream->dc;
+    STATE_AO_GC(stream->ao);
+    struct libxl__sr_hdr hdr = { 0 };
+    int ret = 0;
+
+    assert(!stream->running);
+    stream->running = true;
+
+    memset(dc, 0, sizeof(*dc));
+    dc->ao = ao;
+    dc->readfd = -1;
+
+    dc->writewhat = "save/migration stream";
+    dc->writefd = stream->fd;
+    dc->maxsz = -1;
+    dc->callback = stream_header_done;
+
+    ret = libxl__datacopier_start(dc);
+    if (ret)
+        goto err;
+
+    hdr.ident   = htobe64(RESTORE_STREAM_IDENT);
+    hdr.version = htobe32(RESTORE_STREAM_VERSION);
+    hdr.options = htobe32(0);
+
+    libxl__datacopier_prefixdata(egc, dc, &hdr, sizeof(hdr));
+    return;
+
+ err:
+    assert(ret);
+    stream_failed(egc, stream, ret);
+}
+
+void libxl__stream_write_abort(libxl__egc *egc,
+                               libxl__stream_write_state *stream, int rc)
+{
+    stream_failed(egc, stream, rc);
+}
+
+static void stream_success(libxl__egc *egc, libxl__stream_write_state *stream)
+{
+    stream->rc = 0;
+
+    stream_done(egc, stream);
+}
+
+static void stream_failed(libxl__egc *egc,
+                          libxl__stream_write_state *stream, int rc)
+{
+    assert(rc);
+    stream->rc = rc;
+
+    if (stream->running)
+        stream_done(egc, stream);
+}
+
+static void stream_done(libxl__egc *egc,
+                        libxl__stream_write_state *stream)
+{
+    assert(stream->running);
+    stream->running = false;
+
+    if (stream->emu_carefd)
+        libxl__carefd_close(stream->emu_carefd);
+    free(stream->emu_body);
+
+    check_stream_finished(egc, stream, stream->rc, "stream");
+}
+
+static void check_stream_finished(libxl__egc *egc,
+                                  libxl__stream_write_state *stream,
+                                  int rc, const char *what)
+{
+    libxl__domain_suspend_state *dss = CONTAINER_OF(stream, *dss, sws);
+    STATE_AO_GC(dss->ao);
+
+    LOG(DEBUG, "Task '%s' joining (rc %d)", what, rc);
+
+    if (rc && !stream->joined_rc) {
+        bool skip = false;
+        /* First reported failure from joining tasks.  Tear everything down */
+        stream->joined_rc = rc;
+
+        if (libxl__stream_write_inuse(stream)) {
+            skip = true;
+            libxl__stream_write_abort(egc, stream, rc);
+        }
+
+        if (libxl__save_helper_inuse(&dss->shs)) {
+            skip = true;
+            libxl__save_helper_abort(egc, &dss->shs);
+        }
+
+        /* There is at least one more active task to join - wait for its
+           callback */
+        if ( skip )
+            return;
+    }
+
+    if (libxl__stream_write_inuse(stream))
+        LOG(DEBUG, "stream still in use");
+    else if (libxl__save_helper_inuse(&dss->shs))
+        LOG(DEBUG, "save/restore still in use");
+    else {
+        LOG(DEBUG, "Join complete: result %d", stream->joined_rc);
+        stream->completion_callback(egc, stream, stream->joined_rc);
+    }
+}
+
+static void stream_header_done(libxl__egc *egc,
+                               libxl__datacopier_state *dc,
+                               int rc, int onwrite, int errnoval)
+{
+    libxl__stream_write_state *stream = CONTAINER_OF(dc, *stream, dc);
+    STATE_AO_GC(stream->ao);
+    struct libxl__sr_rec_hdr rec = { REC_TYPE_LIBXC_CONTEXT };
+    int ret;
+
+    if (onwrite || errnoval) {
+        ret = ERROR_FAIL;
+        LOG(ERROR, "rc %d, onwrite %d, errnoval %d", rc, onwrite, errnoval);
+        goto err;
+    }
+
+    setup_write(egc, stream, "libxc header",
+                &rec, NULL, libxc_header_done);
+    return;
+
+ err:
+    assert(ret);
+    stream_failed(egc, stream, ret);
+}
+
+static void libxc_header_done(libxl__egc *egc,
+                              libxl__stream_write_state *stream)
+{
+    libxl__domain_suspend_state *dss = CONTAINER_OF(stream, *dss, sws);
+
+    libxl__xc_domain_save(egc, dss);
+}
+
+static void __attribute__((unused))
+write_toolstack_record(libxl__egc *egc,
+                                   libxl__stream_write_state *stream)
+{
+    libxl__domain_suspend_state *dss = CONTAINER_OF(stream, *dss, sws);
+    STATE_AO_GC(stream->ao);
+    struct libxl__sr_rec_hdr rec = { REC_TYPE_XENSTORE_DATA };
+    int ret;
+    uint8_t *toolstack_buf = NULL; /* We must free this. */
+    uint32_t toolstack_len;
+
+    ret = libxl__toolstack_save(dss->domid, &toolstack_buf,
+                                &toolstack_len, dss);
+    if (ret)
+        goto err;
+
+    rec.length = toolstack_len;
+
+    setup_write(egc, stream, "toolstack record",
+                &rec, toolstack_buf,
+                toolstack_record_done);
+
+    free(toolstack_buf);
+    return;
+
+ err:
+    assert(ret);
+    free(toolstack_buf);
+    stream_failed(egc, stream, ret);
+}
+
+static void toolstack_record_done(libxl__egc *egc,
+                                  libxl__stream_write_state *stream)
+{
+    libxl__domain_suspend_state *dss = CONTAINER_OF(stream, *dss, sws);
+
+    if (dss->type == LIBXL_DOMAIN_TYPE_HVM)
+        write_emulator_record(egc, stream);
+    else
+        write_end_record(egc, stream);
+}
+
+static void write_emulator_record(libxl__egc *egc,
+                                  libxl__stream_write_state *stream)
+{
+    libxl__domain_suspend_state *dss = CONTAINER_OF(stream, *dss, sws);
+    libxl__datacopier_state *dc = &stream->emu_dc;
+    STATE_AO_GC(stream->ao);
+    struct libxl__sr_rec_hdr *rec = &stream->emu_rec_hdr;
+    struct libxl__sr_emulator_hdr *ehdr = NULL;
+    struct stat st;
+    int ret = 0;
+
+    assert(dss->type == LIBXL_DOMAIN_TYPE_HVM);
+
+    /* Convenience aliases */
+    const char *const filename = dss->dm_savefile;
+    const uint32_t domid = dss->domid;
+
+    libxl__carefd_begin();
+    int readfd = open(filename, O_RDONLY);
+    if (readfd == -1) {
+        ret = ERROR_FAIL;
+        LOGE(ERROR, "unable to open %s", filename);
+        libxl__carefd_unlock();
+        goto err;
+    }
+    stream->emu_carefd = libxl__carefd_record(CTX, readfd);
+    libxl__carefd_unlock();
+
+    if (fstat(readfd, &st)) {
+        ret = ERROR_FAIL;
+        LOGE(ERROR, "unable to fstat %s", filename);
+        goto err;
+    }
+
+    if (!S_ISREG(st.st_mode)) {
+        ret = ERROR_FAIL;
+        LOG(ERROR, "%s is not a plain file!", filename);
+        goto err;
+    }
+
+    rec->type = REC_TYPE_EMULATOR_CONTEXT;
+    rec->length = st.st_size + sizeof(*ehdr);
+    stream->emu_body = ehdr = libxl__malloc(NOGC, rec->length);
+
+    switch(libxl__device_model_version_running(gc, domid)) {
+    case LIBXL_DEVICE_MODEL_VERSION_QEMU_XEN_TRADITIONAL:
+        ehdr->id = EMULATOR_QEMU_TRADITIONAL;
+        break;
+
+    case LIBXL_DEVICE_MODEL_VERSION_QEMU_XEN:
+        ehdr->id = EMULATOR_QEMU_UPSTREAM;
+        break;
+
+    default:
+        ret = ERROR_FAIL;
+        goto err;
+    }
+    ehdr->index = 0;
+
+    memset(dc, 0, sizeof(*dc));
+    dc->ao = stream->ao;
+    dc->readwhat = "qemu save file";
+    dc->readfd = readfd;
+    dc->maxsz = -1;
+    dc->readbuf = stream->emu_body + sizeof(*ehdr);
+    dc->bytes_to_read = rec->length - sizeof(*ehdr);
+    dc->callback = emulator_read_done;
+
+    ret = libxl__datacopier_start(dc);
+    if (ret)
+        goto err;
+
+    return;
+
+ err:
+    assert(ret);
+    stream_failed(egc, stream, ret);
+}
+
+static void emulator_read_done(libxl__egc *egc,
+                               libxl__datacopier_state *dc,
+                               int rc, int onwrite, int errnoval)
+{
+    libxl__stream_write_state *stream = CONTAINER_OF(dc, *stream, emu_dc);
+    STATE_AO_GC(stream->ao);
+    int ret;
+
+    if (onwrite || onwrite || errnoval) {
+        ret = ERROR_FAIL;
+        LOG(ERROR, "rc %d, onwrite %d, errnoval %d", rc, onwrite, errnoval);
+        goto err;
+    }
+
+    libxl__carefd_close(stream->emu_carefd);
+    stream->emu_carefd = NULL;
+
+    setup_write(egc, stream, "emulator record",
+                &stream->emu_rec_hdr, stream->emu_body,
+                emulator_record_done);
+    return;
+
+ err:
+    assert(ret);
+    stream_failed(egc, stream, ret);
+}
+
+static void emulator_record_done(libxl__egc *egc,
+                                 libxl__stream_write_state *stream)
+{
+    free(stream->emu_body);
+    stream->emu_body = NULL;
+
+    write_end_record(egc, stream);
+}
+
+static void write_end_record(libxl__egc *egc,
+                             libxl__stream_write_state *stream)
+{
+    struct libxl__sr_rec_hdr rec = { REC_TYPE_END };
+
+    setup_write(egc, stream, "end record",
+                &rec, NULL, stream_success);
+}
+
+/*
+ * Local variables:
+ * mode: C
+ * c-basic-offset: 4
+ * indent-tabs-mode: nil
+ * End:
+ */
-- 
1.7.10.4