[PATCH 18/27] tools/libxl: Convert a legacy stream if needed

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

For backwards compatibility, a legacy stream needs converting before it can be
read by the v2 stream logic.

This causes the v2 stream logic to need to juggle two parallel tasks.
check_stream_finished() is introduced for the purpose of joining the tasks in
both success and error cases.

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>
---
 tools/libxl/libxl_internal.h    |    3 ++
 tools/libxl/libxl_stream_read.c |   85 ++++++++++++++++++++++++++++++++++++++-
 2 files changed, 87 insertions(+), 1 deletion(-)

diff --git a/tools/libxl/libxl_internal.h b/tools/libxl/libxl_internal.h
index 3e43cc6..5482950 100644
--- a/tools/libxl/libxl_internal.h
+++ b/tools/libxl/libxl_internal.h
@@ -3161,11 +3161,14 @@ struct libxl__stream_read_state {
     /* filled by the user */
     libxl__ao *ao;
     int fd;
+    bool legacy;
     void (*completion_callback)(libxl__egc *egc,
                                 libxl__domain_create_state *dcs,
                                 int rc);
     /* Private */
+    libxl__carefd *v2_carefd;
     int rc;
+    int joined_rc;
     bool running;
     libxl__datacopier_state dc;
     size_t expected_len;
diff --git a/tools/libxl/libxl_stream_read.c b/tools/libxl/libxl_stream_read.c
index 9cdaadf..87b9737 100644
--- a/tools/libxl/libxl_stream_read.c
+++ b/tools/libxl/libxl_stream_read.c
@@ -38,6 +38,10 @@
  * process_record() will choose the correct next action based upon the
  * record.  Upon completion of the action, the next record header will be read
  * from the stream.
+ *
+ * Depending on the contents of the stream, there are likely to be several
+ * parallel tasks being managed.  check_stream_finished() is used to join all
+ * tasks in both success and error cases.
  */
 
 static void stream_success(libxl__egc *egc,
@@ -47,6 +51,12 @@ static void stream_failed(libxl__egc *egc,
 static void stream_done(libxl__egc *egc,
                         libxl__stream_read_state *stream);
 
+static void conversion_done(libxl__egc *egc,
+                            libxl__conversion_helper_state *chs, int rc);
+static void check_stream_finished(libxl__egc *egc,
+                                  libxl__domain_create_state *dcs,
+                                  int rc, const char *what);
+
 /* Event callbacks for main reading loop. */
 static void stream_header_done(libxl__egc *egc,
                                libxl__datacopier_state *dc,
@@ -73,12 +83,33 @@ static void emulator_padding_done(libxl__egc *egc,
 void libxl__stream_read_start(libxl__egc *egc,
                               libxl__stream_read_state *stream)
 {
+    libxl__domain_create_state *dcs = CONTAINER_OF(stream, *dcs, srs);
     libxl__datacopier_state *dc = &stream->dc;
+    STATE_AO_GC(stream->ao);
     int ret = 0;
 
     /* State initialisation. */
     assert(!stream->running);
 
+    if (stream->legacy) {
+        /* Convert a legacy stream, if needed. */
+        dcs->chs.ao = stream->ao;
+        dcs->chs.legacy_fd = stream->fd;
+        dcs->chs.legacy_width = dcs->restore_params.legacy_width;
+        dcs->chs.hvm =
+            (dcs->guest_config->b_info.type == LIBXL_DOMAIN_TYPE_HVM);
+        dcs->chs.v2_carefd = NULL;
+        dcs->chs.completion_callback = conversion_done;
+
+        libxl__convert_legacy_stream(egc, &dcs->chs);
+
+        assert(dcs->chs.v2_carefd);
+        stream->v2_carefd = dcs->chs.v2_carefd;
+        stream->fd = libxl__carefd_fd(dcs->chs.v2_carefd);
+    }
+
+    /* stream->fd is now guarenteed to be a v2 stream. */
+
     memset(dc, 0, sizeof(*dc));
     dc->ao = stream->ao;
     dc->readfd = stream->fd;
@@ -164,7 +195,50 @@ static void stream_done(libxl__egc *egc,
 
     assert(!stream->running);
 
-    stream->completion_callback(egc, dcs, stream->rc);
+    if (stream->v2_carefd)
+        libxl__carefd_close(stream->v2_carefd);
+
+    check_stream_finished(egc, dcs, stream->rc, "stream");
+}
+
+static void check_stream_finished(libxl__egc *egc,
+                                  libxl__domain_create_state *dcs,
+                                  int rc, const char *what)
+{
+    libxl__stream_read_state *stream = &dcs->srs;
+    STATE_AO_GC(dcs->ao);
+
+    LOG(INFO, "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_read_inuse(&dcs->srs)) {
+            skip = true;
+            libxl__stream_read_abort(egc, &dcs->srs, rc);
+        }
+
+        if (libxl__convert_legacy_stream_inuse(&dcs->chs)) {
+            skip = true;
+            libxl__convert_legacy_stream_abort(egc, &dcs->chs, rc);
+        }
+
+        /* There is at least one more active task to join - wait for its
+           callback */
+        if ( skip )
+            return;
+    }
+
+    if (libxl__stream_read_inuse(&dcs->srs))
+        LOG(DEBUG, "stream still in use");
+    else if (libxl__convert_legacy_stream_inuse(&dcs->chs))
+        LOG(DEBUG, "conversion still in use");
+    else {
+        LOG(INFO, "Join complete: result %d", stream->joined_rc);
+        stream->completion_callback(egc, dcs, stream->joined_rc);
+    }
 }
 
 static void stream_header_done(libxl__egc *egc,
@@ -303,6 +377,15 @@ static void record_body_done(libxl__egc *egc,
     stream_failed(egc, stream, ret);
 }
 
+static void conversion_done(libxl__egc *egc,
+                            libxl__conversion_helper_state *chs, int rc)
+{
+    STATE_AO_GC(chs->ao);
+    libxl__domain_create_state *dcs = CONTAINER_OF(chs, *dcs, chs);
+
+    check_stream_finished(egc, dcs, rc, "conversion");
+}
+
 static void process_record(libxl__egc *egc,
                            libxl__stream_read_state *stream)
 {
-- 
1.7.10.4