tools/libxl: Convert a legacy stream if needed

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_all_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>
Acked-by: Ian Jackson <ian.jackson@eu.citrix.com>

---
v4: Alter logic layout in check_all_finished()
v3: Simplified greatly from v2.  No practical change.

tools/libxl/libxl_internal.h

@@ -3258,12 +3258,14 @@ struct libxl__stream_read_state {
     libxl__ao *ao;
     libxl__domain_create_state *dcs;
     int fd;
+    bool legacy;
     void (*completion_callback)(libxl__egc *egc,
                                 libxl__stream_read_state *srs,
                                 int rc);
     /* Private */
     int rc;
     bool running;
+    libxl__conversion_helper_state chs;
 
     /* Main stream-reading data. */
     libxl__datacopier_state dc; /* Only used when reading a record */

tools/libxl/libxl_stream_read.c

@@ -86,6 +86,10 @@
  *    - stream_write_emulator()
  *    - stream_write_emulator_done()
  *    - stream_continue()
+ *
+ * Depending on the contents of the stream, there are likely to be several
+ * parallel tasks being managed.  check_all_finished() is used to join all
+ * tasks in both success and error cases.
  */
 
 /* Success/error/cleanup handling. */
@@ -93,6 +97,10 @@ static void stream_complete(libxl__egc *egc,
                             libxl__stream_read_state *stream, int rc);
 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_all_finished(libxl__egc *egc,
+                               libxl__stream_read_state *stream, int rc);
 
 /* Event chain for first iteration, from _start(). */
 static void stream_header_done(libxl__egc *egc,
@@ -151,6 +159,7 @@ void libxl__stream_read_init(libxl__stream_read_state *stream)
 {
     stream->rc = 0;
     stream->running = false;
+    libxl__conversion_helper_init(&stream->chs);
     FILLZERO(stream->dc);
     FILLZERO(stream->hdr);
     LIBXL_STAILQ_INIT(&stream->record_queue);
@@ -165,6 +174,7 @@ void libxl__stream_read_start(libxl__egc *egc,
                               libxl__stream_read_state *stream)
 {
     libxl__datacopier_state *dc = &stream->dc;
+    STATE_AO_GC(stream->ao);
     int rc = 0;
 
     libxl__stream_read_init(stream);
@@ -172,6 +182,29 @@ void libxl__stream_read_start(libxl__egc *egc,
     stream->running = true;
     stream->phase   = SRS_PHASE_NORMAL;
 
+    if (stream->legacy) {
+        /* Convert the legacy stream. */
+        libxl__conversion_helper_state *chs = &stream->chs;
+
+        chs->ao = stream->ao;
+        chs->legacy_fd = stream->fd;
+        chs->hvm =
+            (stream->dcs->guest_config->b_info.type == LIBXL_DOMAIN_TYPE_HVM);
+        chs->completion_callback = conversion_done;
+
+        rc = libxl__convert_legacy_stream(egc, &stream->chs);
+
+        if (rc) {
+            LOG(ERROR, "Failed to start the legacy stream conversion helper");
+            goto err;
+        }
+
+        assert(stream->chs.v2_carefd);
+        stream->fd = libxl__carefd_fd(stream->chs.v2_carefd);
+        stream->dcs->libxc_fd = stream->fd;
+    }
+    /* stream->fd is now a v2 stream. */
+
     dc->ao      = stream->ao;
     dc->readfd  = stream->fd;
     dc->writefd = -1;
@@ -545,6 +578,10 @@ static void stream_done(libxl__egc *egc,
     if (stream->emu_carefd)
         libxl__carefd_close(stream->emu_carefd);
 
+    /* If we started a conversion helper, we took ownership of its carefd. */
+    if (stream->chs.v2_carefd)
+        libxl__carefd_close(stream->chs.v2_carefd);
+
     /* The record queue had better be empty if the stream believes
      * itself to have been successful. */
     assert(LIBXL_STAILQ_EMPTY(&stream->record_queue) || stream->rc);
@@ -552,6 +589,35 @@ static void stream_done(libxl__egc *egc,
     LIBXL_STAILQ_FOREACH_SAFE(rec, &stream->record_queue, entry, trec)
         free_record(rec);
 
+    check_all_finished(egc, stream, stream->rc);
+}
+
+static void conversion_done(libxl__egc *egc,
+                            libxl__conversion_helper_state *chs, int rc)
+{
+    libxl__stream_read_state *stream = CONTAINER_OF(chs, *stream, chs);
+
+    check_all_finished(egc, stream, rc);
+}
+
+static void check_all_finished(libxl__egc *egc,
+                               libxl__stream_read_state *stream, int rc)
+{
+    STATE_AO_GC(stream->ao);
+
+    if (!stream->rc && rc) {
+        /* First reported failure. Tear everything down. */
+        stream->rc = rc;
+
+        libxl__stream_read_abort(egc, stream, rc);
+        libxl__conversion_helper_abort(egc, &stream->chs, rc);
+    }
+
+    /* Don't fire the callback until all our parallel tasks have stopped. */
+    if (libxl__stream_read_inuse(stream) ||
+        libxl__conversion_helper_inuse(&stream->chs))
+        return;
+
     stream->completion_callback(egc, stream, stream->rc);
 }